feat: Implement separate, potentially pooled ELECTRIC_QUERY_DATABASE_URL (#2438)

Closes #1885

- Introduce separate env var `ELECTRIC_QUERY_DATABASE_URL`
- Add another `connection_opts` field to `replication_opts`, so there
are two of them
- Fallback to using the same ones, but still keep them separate and
maintain them separately
- Main disadvantage of this approach is that the fallbacks for SSL and
IPv6 might need to happen twice if a separate connection string is not
being used - this can be improved by only populating the
`replication_opts`'s `connection_opts` right before starting the
replication client with the single `connection_opts` available if a
separate one was not provided.
- Make all tests (e.g. router tests) use a `pgBouncer` connection string
- Had to ensure connections are terminated explicitly at the end of the
test to avoid hanging connections from the pooler making the tests
really slow.

## TODO
- [x] Write integration test with `pgBouncer` as well that simply syncs
a shape with a pooled connection URL, although the unit tests already
cover that
- [ ] Avoid duplicate SSL and IPv6 fallbacks when separate query
connection string is not provided

Author: msfstef

.changeset/flat-hornets-ring.md

@@ -0,0 +1,6 @@
+---
+"@core/sync-service": patch
+"@electric-sql/docs": patch
+---
+
+Implement `ELECTRIC_QUERY_DATABASE_URL` optional env var to perform queries with separate, potentially pooled connection string.

.github/workflows/elixir_tests.yml

@@ -39,6 +39,19 @@ jobs:
           --health-retries 5
         ports:
           - 54321:5432
+
+      pgbouncer:
+        image: bitnami/pgbouncer:latest
+        env:
+          PGBOUNCER_AUTH_TYPE: trust
+          PGBOUNCER_DATABASE: "*"
+          PGBOUNCER_POOL_MODE: transaction
+          POSTGRESQL_HOST: postgres
+          POSTGRESQL_DATABASE: electric
+          POSTGRESQL_USERNAME: postgres
+          POSTGRESQL_PASSWORD: password
+        ports:
+          - 64321:6432
     steps:
       - uses: actions/checkout@v4
 

integration-tests/tests/_macros.luxinc

@@ -3,14 +3,20 @@
 
 [global pg_container_name=]
 [global pg_host_port=54331]
+[global pg_pooler_host_port=64331]
 [global database_url=postgresql://postgres:password@localhost:$pg_host_port/postgres?sslmode=disable]
+[global pooled_database_url=postgresql://postgres:password@localhost:$pg_pooler_host_port/postgres?sslmode=disable]
 
 [macro setup_pg_with_name shell_name initdb_args config_opts]
   [shell $shell_name]
     -$fail_pattern
+
+    # Create network for given contianer to be able to attach pooler
+    !docker network ls | grep ${pg_container_name}-network || docker network create ${pg_container_name}-network
 
     !docker run \
       --name $pg_container_name \
+      --network ${pg_container_name}-network \
       -e POSTGRES_DB=electric \
       -e POSTGRES_USER=postgres \
       -e POSTGRES_PASSWORD=password \
@@ -31,6 +37,27 @@
   [invoke setup_pg_with_name "pg" $initdb_args $config_opts]
 [endmacro]
 
+[macro setup_pg_with_pooler initdb_args config_opts]
+  [invoke setup_pg_with_name "pg" $initdb_args $config_opts]
+  [shell "pg_pooler"]
+    -$fail_pattern
+
+    !docker run \
+      --name "${pg_container_name}-pooler" \
+      --network ${pg_container_name}-network \
+      -e PGBOUNCER_AUTH_TYPE=trust \
+      -e PGBOUNCER_DATABASE=* \
+      -e PGBOUNCER_POOL_MODE=transaction \
+      -e POSTGRESQL_HOST=$pg_container_name \
+      -e POSTGRESQL_DATABASE=electric \
+      -e POSTGRESQL_USERNAME=postgres \
+      -e POSTGRESQL_PASSWORD=password \
+      -p $pg_pooler_host_port:6432 \
+      bitnami/pgbouncer:latest
+
+    ??LOG process up: PgBouncer
+[endmacro]
+
 [macro stop_pg]
   [shell pg_lifecycle]
     # This timeout is needed until https://github.com/electric-sql/electric/issues/1632 is fixed.
@@ -87,6 +114,10 @@
   [invoke setup_electric_with_env "DATABASE_URL=$database_url"]
 [endmacro]
 
+[macro setup_electric_with_pooler]
+  [invoke setup_electric_with_env "DATABASE_URL=$database_url ELECTRIC_QUERY_DATABASE_URL=$pooled_database_url"]
+[endmacro]
+
 [macro setup_multi_tenant_electric]
   [invoke setup_electric_with_env ""]
 [endmacro]
@@ -125,11 +156,12 @@
 
 [macro teardown_container container_name]
   -$fail_pattern
-  !docker rm -f -v $container_name
+  !docker rm -f -v $container_name 2>/dev/null || true
   ?$PS1
 [endmacro]
 
 [macro teardown]
+  [invoke teardown_container "${pg_container_name}-pooler"]
   [invoke teardown_container $pg_container_name]
   !../scripts/clean_up.sh
   ?$PS1

integration-tests/tests/pooled-connections.lux

@@ -0,0 +1,71 @@
+[doc Verify Electric handles data replication with pooled connection]
+
+[include _macros.luxinc]
+
+[global pg_container_name=poold-connections__pg]
+
+###
+
+## Start a new Postgres cluster
+[invoke setup_pg_with_pooler "" ""]
+
+## Add some data
+[invoke start_psql]
+[shell psql]
+  """!
+  CREATE TABLE items (
+    id UUID PRIMARY KEY,
+    val TEXT
+  );
+  """
+  ??CREATE TABLE
+
+  """!
+  INSERT INTO
+    items (id, val)
+  SELECT
+    gen_random_uuid(),
+    '#' || generate_series || ' initial val'
+  FROM
+    generate_series(1, 10);
+  """
+  ??INSERT 0 10
+
+## Start the sync service.
+[invoke setup_electric_with_pooler]
+
+[shell electric]
+  ??[info] Starting replication from postgres
+  
+# Initialize a shape and collect the offset
+[shell client]
+  # strip ANSI codes from response for easier matching
+  !curl -v -X GET "http://localhost:3000/v1/shape?table=items&offset=-1"
+  ?electric-handle: ([\d-]+)
+  [local handle=$1]
+  ?electric-offset: ([\w\d_]+)
+  [local offset=$1]
+  ??"val":"#10 initial val"
+
+## Add some more data
+[shell psql]
+  """!
+  INSERT INTO
+    items (id, val)
+  SELECT
+    gen_random_uuid(),
+    '#' || generate_series || ' new val'
+  FROM
+    generate_series(1, 10);
+  """
+  ??INSERT 0 10
+
+# Client should be able to continue same shape
+[shell client]
+  [sleep 2]
+  !curl -v -X GET "http://localhost:3000/v1/shape?table=items&handle=$handle&offset=$offset"
+  ??HTTP/1.1 200 OK
+  ??"val":"#10 new val"
+
+[cleanup]
+  [invoke teardown]

packages/sync-service/.env.test

@@ -1,2 +1,3 @@
 ELECTRIC_LOG_LEVEL=info
 DATABASE_URL=postgresql://postgres:password@localhost:54321/postgres?sslmode=disable
+ELECTRIC_QUERY_DATABASE_URL=postgresql://postgres:password@localhost:64321/postgres?sslmode=disable

packages/sync-service/README.md

@@ -64,7 +64,7 @@ reuse that configuration if you want:
     config :my_app, Repo, database_config
 
     config :electric,
-      connection_opts: Electric.Utils.obfuscate_password(database_config)
+      replication_connection_opts: Electric.Utils.obfuscate_password(database_config)
 
 Or if you're getting your db connection from an environment variable, then you
 can use
@@ -74,7 +74,7 @@ can use
     {:ok, database_config} = Electric.Config.parse_postgresql_uri(System.fetch_env!("DATABASE_URL"))
 
     config :electric,
-      connection_opts: Electric.Utils.obfuscate_password(database_config)
+      replication_connection_opts: Electric.Utils.obfuscate_password(database_config)
 
 The Electric app will startup along with the rest of your Elixir app.
 

packages/sync-service/config/runtime.exs

@@ -123,14 +123,23 @@ config :opentelemetry,
        local_parent_not_sampled: :always_off
      }}
 
-database_url_config = env!("DATABASE_URL", &Electric.Config.parse_postgresql_uri!/1)
+replication_database_url_config = env!("DATABASE_URL", &Electric.Config.parse_postgresql_uri!/1)
 
-database_ipv6_config =
-  env!("ELECTRIC_DATABASE_USE_IPV6", :boolean, false)
+query_database_url_config =
+  env!(
+    "ELECTRIC_QUERY_DATABASE_URL",
+    &Electric.Config.parse_postgresql_uri!/1,
+    replication_database_url_config
+  )
 
-connection_opts = database_url_config ++ [ipv6: database_ipv6_config]
+database_ipv6_config = env!("ELECTRIC_DATABASE_USE_IPV6", :boolean, false)
 
-config :electric, connection_opts: Electric.Utils.obfuscate_password(connection_opts)
+replication_connection_opts = replication_database_url_config ++ [ipv6: database_ipv6_config]
+query_connection_opts = query_database_url_config ++ [ipv6: database_ipv6_config]
+
+config :electric,
+  replication_connection_opts: Electric.Utils.obfuscate_password(replication_connection_opts),
+  query_connection_opts: Electric.Utils.obfuscate_password(query_connection_opts)
 
 enable_integration_testing? = env!("ELECTRIC_ENABLE_INTEGRATION_TESTING", :boolean, nil)
 cache_max_age = env!("ELECTRIC_CACHE_MAX_AGE", :integer, nil)

packages/sync-service/dev/docker-compose.yml

@@ -38,6 +38,20 @@ services:
       - docker-entrypoint.sh
       - -c
       - config_file=/etc/postgresql.conf
+  pgbouncer:
+    image: bitnami/pgbouncer:latest
+    environment:
+      PGBOUNCER_AUTH_TYPE: trust
+      PGBOUNCER_DATABASE: "*"
+      PGBOUNCER_POOL_MODE: transaction
+      POSTGRESQL_HOST: postgres
+      POSTGRESQL_DATABASE: electric
+      POSTGRESQL_USERNAME: postgres
+      POSTGRESQL_PASSWORD: password
+    ports:
+      - "64321:6432"
+    depends_on:
+      - postgres
   nginx:
     image: nginx:latest
     ports:

packages/sync-service/lib/electric.ex

@@ -68,7 +68,9 @@ defmodule Electric do
 
   ### Database
 
-  - `connection_opts` - **Required**
+  - `replication_connection_opts` - **Required**
+     #{NimbleOptions.docs(opts_schema, nest_level: 1)}.
+  - `query_connection_opts` - Optional separate connection string that can use a pooler for non-replication queries (default: nil)
      #{NimbleOptions.docs(opts_schema, nest_level: 1)}.
   - `db_pool_size` - How many connections Electric opens as a pool for handling shape queries (default: `#{default.(:db_pool_size)}`)
   - `replication_stream_id` - Suffix for the logical replication publication and slot name (default: `#{default.(:replication_stream_id)}`)

packages/sync-service/lib/electric/application.ex

@@ -84,10 +84,14 @@ defmodule Electric.Application do
 
     slot_name = Keyword.get(opts, :slot_name, "electric_slot_#{replication_stream_id}")
 
+    replication_connection_opts = get_env!(opts, :replication_connection_opts)
+
     Keyword.merge(
       core_config,
-      connection_opts: get_env!(opts, :connection_opts),
+      connection_opts:
+        get_env_with_default(opts, :query_connection_opts, replication_connection_opts),
       replication_opts: [
+        connection_opts: replication_connection_opts,
         publication_name: publication_name,
         slot_name: slot_name,
         slot_temporary?: get_env(opts, :replication_slot_temporary?)