chore: e2e test for multi tenancy

.github/workflows/ts_test.yml

@@ -70,6 +70,8 @@ jobs:
     defaults:
       run:
         working-directory: ${{ matrix.package_dir }}
+    env:
+      DATABASE_ID: ci_test_tenant
     steps:
       - uses: actions/checkout@v4
       - uses: erlef/setup-beam@v1
@@ -111,7 +113,7 @@ jobs:
             mix run --no-halt &
 
           wait-on: |
-            http-get://localhost:3000/v1/health
+            http-get://localhost:3000/v1/health?database_id=${{ env.DATABASE_ID }}
 
           tail: true
           log-output-resume: stderr

integration-tests/tests/_macros.luxinc

@@ -5,8 +5,8 @@
 [global pg_host_port=54331]
 [global database_url=postgresql://postgres:password@localhost:$pg_host_port/postgres?sslmode=disable]
 
-[macro setup_pg initdb_args config_opts]
-  [shell pg]
+[macro setup_pg_with_name shell_name initdb_args config_opts]
+  [shell $shell_name]
     -$fail_pattern
 
     !docker run \
@@ -27,6 +27,10 @@
     -
 [endmacro]
 
+[macro setup_pg initdb_args config_opts]
+  [invoke setup_pg "pg" $initdb_args $config_opts]
+[endmacro]
+
 [macro stop_pg]
   [shell pg_lifecycle]
     # This timeout is needed until https://github.com/electric-sql/electric/issues/1632 is fixed.
@@ -48,11 +52,15 @@
     ??database system is ready to accept connections
 [endmacro]
 
-[macro start_psql]
-  [shell psql]
+[macro start_psql_shell shell_name pg_container_name]
+  [shell $shell_name]
     !docker exec -u postgres -it $pg_container_name psql
 [endmacro]
 
+[macro start_psql]
+  [invoke start_psql_shell psql $pg_container_name]
+[endmacro]
+
 [macro seed_pg]
   [shell psql]
     !docker exec -u postgres -it $pg_container_name psql
@@ -80,6 +88,10 @@
 [endmacro]
 
 [macro setup_electric]
+  [invoke setup_electric_with_env "DATABASE_ID=integration_test_tenant DATABASE_URL=$database_url"]
+[endmacro]
+
+[macro setup_multi_tenant_electric]
   [invoke setup_electric_with_env ""]
 [endmacro]
 
@@ -91,7 +103,22 @@
   [shell $shell_name]
     -$fail_pattern
 
-    !DATABASE_URL=$database_url PORT=$port $env ../scripts/electric_dev.sh
+    !PORT=$port $env ../scripts/electric_dev.sh
+[endmacro]
+
+[macro add_tenant tenant_id electric_port]
+  [shell $tenant_id]
+    !curl -X POST http://localhost:$electric_port/v1/admin/database \
+      -H "Content-Type: application/json" \
+      -d "{\"database_id\":\"$tenant_id\",\"DATABASE_URL\":\"$database_url\"}"
+    ??"$tenant_id"
+[endmacro]
+
+[macro check_tenant_status tenant_id expected_status electric_port]
+  [shell $tenant_id]
+    [invoke wait-for "curl -X GET http://localhost:$electric_port/v1/health?database_id=$tenant_id" "\{\"status\":\"$expected_status\"\}" 10 $PS1]
+    #!curl -X GET http://localhost:$electric_port/v1/health?database_id=$tenant_id
+    #??{"status":"$expected_status"}
 [endmacro]
 
 [macro teardown]
@@ -101,3 +128,21 @@
   !../scripts/clean_up.sh
   ?$PS1
 [endmacro]
+
+[macro wait-for command match max_time prompt]
+    [loop i 1..$max_time]
+        @$match
+        !$command
+        ??$command
+        ?$prompt
+        [sleep 1]
+    [endloop]
+    # The last prompt won't match since the loop pattern will
+    # match before it, so match it here instead.
+    ?$prompt
+
+    # Sync up after the loop.
+    !$command
+    ??$command
+    ?$prompt
+[endmacro]

integration-tests/tests/invalidated-replication-slot.lux

@@ -6,7 +6,7 @@
 
 [my invalidated_slot_error=
   """
-  [error] GenServer Electric.Connection.Manager terminating
+  [error] :gen_statem {Electric.Registry.Processes, {Electric.Postgres.ReplicationClient, :default, "integration_test_tenant"}} terminating
   ** (Postgrex.Error) ERROR 55000 (object_not_in_prerequisite_state) cannot read from logical replication slot "electric_slot_integration"
 
   This slot has been invalidated because it exceeded the maximum reserved size.

integration-tests/tests/multi-tenancy.lux

@@ -0,0 +1,221 @@
+[doc Verify support for multi tenancy]
+
+[include _macros.luxinc]
+
+[global tenant1_pg_container_name=multi-tenancy-tenant1__pg]
+[global tenant1_pg_host_port=54331]
+[global tenant1_database_url=postgresql://postgres:password@localhost:$tenant1_pg_host_port/postgres?sslmode=disable]
+
+[global tenant2_pg_container_name=multi-tenancy-tenant2__pg]
+[global tenant2_pg_host_port=54332]
+[global tenant2_database_url=postgresql://postgres:password@localhost:$tenant2_pg_host_port/postgres?sslmode=disable]
+
+###
+
+## Start a new Postgres DB
+[global pg_container_name=$tenant1_pg_container_name]
+[global pg_host_port=$tenant1_pg_host_port]
+[global database_url=$tenant1_database_url]
+[invoke setup_pg_with_name "tenant1_pg" "" ""]
+
+## Start Electric in multi tenancy mode
+[invoke setup_multi_tenant_electric]
+
+[shell electric]
+  ??[info] Running Electric.Plug.Router with Bandit 1.5.5 at 0.0.0.0:3000 (http)
+
+## Create tenant 1
+[invoke add_tenant "tenant1" 3000]
+[invoke check_tenant_status "tenant1" "active" 3000]
+
+## Setup a second Postgres DB
+[global pg_container_name=$tenant2_pg_container_name]
+[global pg_host_port=$tenant2_pg_host_port]
+[global database_url=$tenant2_database_url]
+[invoke setup_pg_with_name "tenant2_pg" "" ""]
+
+## Create tenant 2
+[invoke add_tenant "tenant2" 3000]
+[invoke check_tenant_status "tenant2" "active" 3000]
+
+## Insert some data in both DBs
+[invoke start_psql_shell "tenant1_psql" $tenant1_pg_container_name]
+[invoke start_psql_shell "tenant2_psql" $tenant2_pg_container_name]
+
+[shell tenant1_psql]
+  !CREATE TABLE items (id INT PRIMARY KEY, val TEXT);
+  ??CREATE TABLE
+  !INSERT INTO items (id, val) VALUES (1, 'tenant1');
+  ??INSERT 0 1
+
+[shell tenant2_psql]
+  !CREATE TABLE items (id INT PRIMARY KEY, val TEXT);
+  ??CREATE TABLE
+  !INSERT INTO items (id, val) VALUES (1, 'tenant2');
+  ??INSERT 0 1
+
+## Check that both tenants can query their data
+[shell tenant1]
+  # Chech tenant 1 data
+  !curl -i -X GET "http://localhost:3000/v1/shape/items?offset=-1&database_id=tenant1"
+  ?\e\[1melectric-shape-id\e\[0m: ([\d-]+)
+  [local shape_id=$1]
+  ?\e\[1melectric-chunk-last-offset\e\[0m: ([\d_]+)
+  [local offset=$1]
+  """??
+  [{"key":"\"public\".\"items\"/\"1\"","value":{"id":"1","val":"tenant1"},"headers":{"operation":"insert","relation":["public","items"]},"offset":"$offset"}
+  ]
+  """
+
+# Check tenant 2 data
+[shell tenant2]
+  !curl -i -X GET "http://localhost:3000/v1/shape/items?offset=-1&database_id=tenant2"
+  ?\e\[1melectric-shape-id\e\[0m: ([\d-]+)
+  [local shape_id=$1]
+  ?\e\[1melectric-chunk-last-offset\e\[0m: ([\d_]+)
+  [local offset=$1]
+  """??
+  [{"key":"\"public\".\"items\"/\"1\"","value":{"id":"1","val":"tenant2"},"headers":{"operation":"insert","relation":["public","items"]},"offset":"$offset"}
+  ]
+  """
+
+## Now do a live query on tenant 1
+[shell tenant1]
+  ??$PS1
+  !curl -i -X GET "localhost:3000/v1/shape/items?offset=$offset&shape_id="$shape_id"&database_id=tenant1&live"
+
+## And a live query on tenant 2
+[shell tenant2]
+  ??$PS1
+  !curl -i -X GET "localhost:3000/v1/shape/items?offset=$offset&shape_id="$shape_id"&database_id=tenant2&live"
+
+## Insert some data in tenant 1
+[shell tenant1_psql]
+  !INSERT INTO items (id, val) VALUES (2, 'tenant1');
+  ??INSERT 0 1
+
+## Insert some data in tenant 2
+[shell tenant2_psql]
+  !INSERT INTO items (id, val) VALUES (2, 'tenant2');
+  ??INSERT 0 1
+
+## Check that tenant 1 sees the new data
+[shell tenant1]
+  # give some time for the data to sync
+  [sleep 1]
+  ?\e\[1melectric-chunk-last-offset\e\[0m: ([\d_]+)
+  [local offset=$1]
+  ??[{"offset":"$offset","value":{"id":"2","val":"tenant1"},"key":"\"public\".\"items\"/\"2\"","headers":{"relation":["public","items"],"operation":"insert","txid":
+  ?[\d+]
+  ??}},{"headers":{"control":"up-to-date"}}]$PS1
+
+## Check that tenant 2 sees the new data
+[shell tenant2]
+  [sleep 1]
+  ?\e\[1melectric-chunk-last-offset\e\[0m: ([\d_]+)
+  [local offset=$1]
+  ??[{"offset":"$offset","value":{"id":"2","val":"tenant2"},"key":"\"public\".\"items\"/\"2\"","headers":{"relation":["public","items"],"operation":"insert","txid":
+  ?[\d+]
+  ??}},{"headers":{"control":"up-to-date"}}]$PS1
+
+# Disable fail pattern for Electric as we are going to kill it
+[shell electric]
+  -
+
+## kill Electric
+[shell orchestrator]
+  !kill $(lsof -ti:3000)
+  ??$PS1
+
+## restart Electric
+[shell electric]
+  ??$PS1
+  # Re-enable fail pattern for Electric
+  -$fail_pattern
+  [invoke setup_multi_tenant_electric]
+  ?? Reloading tenant tenant1 from storage
+  ?? Reloading tenant tenant2 from storage
+
+## Make a query to check that they still see their data
+[shell tenant1]
+  # wait for Electric to start
+  [invoke wait-for "lsof -Pi :3000 -sTCP:LISTEN" "TCP \*:3000" 10 $PS1]
+  # Query the shape
+  !curl -i -X GET "http://localhost:3000/v1/shape/items?offset=${offset}&shape_id=${shape_id}&database_id=tenant1"
+  ???[{"headers":{"control":"up-to-date"}}]
+  ??$PS1
+
+[shell tenant2]
+  # wait for Electric to start
+  [invoke wait-for "lsof -Pi :3000 -sTCP:LISTEN" "TCP \*:3000" 10 $PS1]
+  # Query the shape
+  !curl -i -X GET "http://localhost:3000/v1/shape/items?offset=${offset}&shape_id=${shape_id}&database_id=tenant2"
+  ???[{"headers":{"control":"up-to-date"}}]
+  ??$PS1
+
+## Make a live query on both and check that it still works
+[shell tenant1]
+  !curl -i -X GET "localhost:3000/v1/shape/items?offset=$offset&shape_id="$shape_id"&database_id=tenant1&live"
+
+[shell tenant2]
+  !curl -i -X GET "localhost:3000/v1/shape/items?offset=$offset&shape_id="$shape_id"&database_id=tenant2&live"
+
+## Insert some data in tenant 1
+[shell tenant1_psql]
+  !INSERT INTO items (id, val) VALUES (3, 'tenant 1');
+  ??INSERT 0 1
+
+## Insert some data in tenant 2
+[shell tenant2_psql]
+  !INSERT INTO items (id, val) VALUES (3, 'tenant 2');
+  ??INSERT 0 1
+
+## Check that tenant 1 sees the new data
+[shell tenant1]
+  # give some time for the data to sync
+  [sleep 1]
+  ?\e\[1melectric-chunk-last-offset\e\[0m: ([\d_]+)
+  [local offset=$1]
+  ??[{"offset":"$offset","value":{"id":"3","val":"tenant 1"},"key":"\"public\".\"items\"/\"3\"","headers":{"relation":["public","items"],"operation":"insert","txid":
+  ?[\d+]
+  ??}},{"headers":{"control":"up-to-date"}}]$PS1
+
+## Check that tenant 2 sees the new data
+[shell tenant2]
+  [sleep 1]
+  ?\e\[1melectric-chunk-last-offset\e\[0m: ([\d_]+)
+  [local offset=$1]
+  ??[{"offset":"$offset","value":{"id":"3","val":"tenant 2"},"key":"\"public\".\"items\"/\"3\"","headers":{"relation":["public","items"],"operation":"insert","txid":
+  ?[\d+]
+  ??}},{"headers":{"control":"up-to-date"}}]$PS1
+
+[shell electric]
+  # disable fail pattern because deleting a tenant will stop the tenant processes
+  # which will output some error messages because of the shutdown
+  -
+
+## delete one of the tenants
+[shell orchestrator]
+  !curl -X DELETE "http://localhost:3000/v1/admin/database?database_id=tenant2"
+  ???"tenant2"
+  ??$PS1
+  # Verify that tenant 2 is deleted
+  !curl -X GET http://localhost:3000/v1/health?database_id=tenant2
+  ???Database not found.
+  ??$PS1
+  # Verify that tenant 1 still exists
+  [invoke check_tenant_status "tenant1" "active" 3000]
+
+## kill Electric
+[shell orchestrator]
+  !kill $(lsof -ti:3000)
+  ??$PS1
+
+## restart Electric and check that only tenant 1 is reloaded and not tenant 2
+[shell electric]
+  ??$PS1
+  # Set fail pattern to fail if tenant 2 is reloaded
+  -Reloading tenant tenant2 from storage
+  !PORT=3000 ../scripts/electric_dev.sh
+  ?? Reloading tenant tenant1 from storage
+  ??[info] Running Electric.Plug.Router with Bandit 1.5.5 at 0.0.0.0:3000 (http)

integration-tests/tests/resuming-replication-at-consistent-point.lux

@@ -75,15 +75,15 @@
   ?Txn received in Shapes.Consumer: %Electric.Replication.Changes.Transaction{xid: $xid
 
   # Both consumers hit their call limit and exit with simulated storage failures.
-  ?\[error\] GenServer {Electric\.Registry\.Processes, {Electric\.Shapes\.Consumer, :default, "[0-9-]+"}} terminating
+  ?\[error\] GenServer {Electric\.Registry\.Processes, {Electric\.Shapes\.Consumer, :default, "integration_test_tenant", "[0-9-]+"}} terminating
   ??Simulated storage failure
-  ?\[error\] GenServer {Electric\.Registry\.Processes, {Electric\.Shapes\.Consumer, :default, "[0-9-]+"}} terminating
+  ?\[error\] GenServer {Electric\.Registry\.Processes, {Electric\.Shapes\.Consumer, :default, "integration_test_tenant", "[0-9-]+"}} terminating
   ??Simulated storage failure
 
   # The log collector process and the replication client both exit, as their lifetimes are tied
   # together by the supervision tree design.
-  ??[error] GenServer {Electric.Registry.Processes, {Electric.Replication.ShapeLogCollector, :default}} terminating
-  ??[error] :gen_statem {Electric.Registry.Processes, {Electric.Postgres.ReplicationClient, :default}} terminating
+  ??[error] GenServer {Electric.Registry.Processes, {Electric.Replication.ShapeLogCollector, :default, "integration_test_tenant"}} terminating
+  ??[error] :gen_statem {Electric.Registry.Processes, {Electric.Postgres.ReplicationClient, :default, "integration_test_tenant"}} terminating
 
   # Observe that both shape consumers and the replication client have restarted.
   ??[debug] Found existing replication slot