chore(electric): Use a PG connection pool instead of opening a new connection for every query #500
Conversation
VAX-1035 Implement a connection pool for Postgres connections to use in the WebsocketServer module
We're currently spawning new connections as needed when a new Satellite client connects to the server and needs its LSN and/or initial subscription data fetched from Postgres. A connection pool will allows us to manage this mess by setting the upper bound on the number of concurrent connections used across all Satellite clients. |
alco
force-pushed
the
alco/vax-1035-pg-connection-pool
branch
2 times, most recently
from
43d300f to
30ebc03
Compare
|
|
||
| :error -> | ||
| conn_config | ||
| |> Connectors.get_connection_opts(replication: false) |
There was a problem hiding this comment.
I think this replication: false may be causing problems. previously the default connection params would include create a connection in replica mode whereas with this pool approach you're always connecting in non-replica mode.
| end | ||
| ) do | ||
| defp start_subscription(%State{origin: origin, repl_config: rep_conf}) do | ||
| ConnectionPool.checkout!(origin, fn conn -> |
There was a problem hiding this comment.
so I think this is where the replication: false bites you. might be an argument that you don't need the connection pool here as this is a long-running connection. it's the transient connections, which are all replication: false, which need the pool
There was a problem hiding this comment.
It's good that you mention this, the replication option has slipped my mind while I was working on the pool. Unfortunately, I don't think that's the problem here. Both tyimes PostgresManager needs to connect to the DB, it runs regular SQL statements: ALTER SUBSCRIPTION, SELECT. The long-running replication connection is set up in LogicalReplicationProducer.init(), I haven't touched that code.
I took a closer look at the failing E2E tests. There are two types of recurring issues there: (1) missing output to match on (but I think it's the test code that's wrong there) and (2) "Received subscribe response for unknown subscription" error.
The former must have been caused by different scheduling of concurrent processes due to the introduction of the connnection pool, I can fix that in the test code. The latter is an actual new problem I need to look more closely at. Probably caused by the same connection being used to run different queries.
| @@ -181,14 +185,14 @@ defmodule Electric.Replication.PostgresConnectorMng do | |||
| "Attempting to initialize #{origin}: #{conn_config.username}@#{conn_config.host}:#{conn_config.port}" | |||
| ) | |||
|
|
|||
| Client.with_conn(conn_config, fn conn -> | |||
There was a problem hiding this comment.
this is replication: true but I'm not sure it needs to be... there's things we weren't able to do in the extension migration stuff because the connection was replica: false but not sure if turning off replication causes other problems
Two reasons for this:
1. WebSocketServer, migrations plug, and anything else useful that
Electric provides requires a running PostgresConnector. So it makes
sense to start it early as part of the main supervision tree.
2. When I was doing some refactoring elsewhere that broke
PostgresConnector's startup, the error didn't bubble up because the
return value of Connectors.start_connector() was ignored.
alco
force-pushed
the
alco/vax-1035-pg-connection-pool
branch
from
83f9876 to
c205c07
Compare
Closes VAX-1035.
CachedWal.Api.current_position()returningnil.