Complete dropForeignOrigin flag implementation and tests

The databases we listen to at Gadget see really huge transactions when we do shard moves. Shard moves use logical replication to move data from one database to another, and logical replication does a bigass COPY statement for each table at the start of replication to move the initial state of the table from the source database to the destination.

These create huge transactions in the WAL which the wal-listener needs to process. We don't really care about this data downstream -- we in fact don't want it at all, as it is just a replay of data on a new shard that we've already seen on the old shard. Luckily, postgres has a property in the replication protocol that allows us to tell postgres that we don't actually want these transactions streamed to us in the wal-listener: the `ORIGIN` option of a subscription!

When you create a subscription, you can specify that you only want messages with no origin, which means only messages that originate from the instance the subscription is on, instead of any upstream instances:

```
primary1=# CREATE SUBSCRIPTION sub_pri1_pri3
primary1-#   CONNECTION 'dbname=foo host=primary3 user=repuser'
primary1-#   PUBLICATION pub_pri3 WITH (origin = NONE);
```

This matches what we want when using the wal-listener at Gadget: we want transactions from a data plane database for envs that currently live on that database. But, when an env is moving to that database, the transactions will have an origin of the upstream shard move source, and we don't want that data, as we're still actually serving the data from that source, and listening to it too.

Sadly, this `ORIGIN` option is only supported on PG 16 and newer, and we were on PG 15 at the time, so we couldn't use it. This PR switches that around, as now we're on PG 17! Horray

At the time though, we were persistent, and tried to make due with a different, lower level feature of the postgres logical replication protocol, where each incoming message includes an `origin` property that describes where it came from. We hoped we could inspect this property and quickly drop messages that had origins we didn't care about as well. We added the `dropForeignOrigin` configuration flag to the wal-listener to set this flag in #31. Even more sadly, we learned the hard way that while the `origin` property of was present in the older postgres versions, and valid in the protocol, it was just never set at all. Sad.

So, this PR drops support for the old, busted, message-inspection based way of origin filtering, in favour of the better peforming, server-side filtering using the ORIGIN property of the subscription.

Author: airhorns

.github/workflows/test.yml

@@ -59,32 +59,58 @@ jobs:
     - name: Download dependencies
       run: go mod download
 
-    - name: Start PostgreSQL with logical replication
+    - name: Create Docker network
+      run: docker network create wal-test-network
+
+    - name: Start primary PostgreSQL (upstream)
       run: |
         docker run -d \
-          --name postgres-test \
+          --name postgres-primary \
+          --network wal-test-network \
           -e POSTGRES_USER=postgres \
           -e POSTGRES_PASSWORD=postgres \
           -e POSTGRES_DB=postgres \
           -p 5432:5432 \
           postgres:17 \
           -c wal_level=logical \
-          -c max_wal_senders=10 \
-          -c max_replication_slots=10
+          -c max_wal_senders=20 \
+          -c max_replication_slots=20
+
+    - name: Start downstream PostgreSQL (for foreign origin testing)
+      run: |
+        docker run -d \
+          --name postgres-downstream \
+          --network wal-test-network \
+          -e POSTGRES_USER=postgres \
+          -e POSTGRES_PASSWORD=postgres \
+          -e POSTGRES_DB=postgres \
+          -p 5433:5432 \
+          postgres:17 \
+          -c wal_level=logical \
+          -c max_wal_senders=20 \
+          -c max_replication_slots=20
 
-    - name: Wait for PostgreSQL to be ready
+    - name: Wait for PostgreSQL instances to be ready
       run: |
         for i in {1..30}; do
-          docker exec postgres-test pg_isready -U postgres && break
-          echo "Waiting for PostgreSQL..."
+          docker exec postgres-primary pg_isready -U postgres && break
+          echo "Waiting for primary PostgreSQL..."
+          sleep 2
+        done
+        for i in {1..30}; do
+          docker exec postgres-downstream pg_isready -U postgres && break
+          echo "Waiting for downstream PostgreSQL..."
           sleep 2
         done
 
     - name: Verify PostgreSQL configuration
       run: |
-        docker exec postgres-test psql -U postgres -c "SHOW wal_level;"
-        docker exec postgres-test psql -U postgres -c "SHOW max_replication_slots;"
-        docker exec postgres-test psql -U postgres -c "SHOW max_wal_senders;"
+        echo "=== Primary PostgreSQL ==="
+        docker exec postgres-primary psql -U postgres -c "SHOW wal_level;"
+        docker exec postgres-primary psql -U postgres -c "SHOW max_replication_slots;"
+        echo "=== Downstream PostgreSQL ==="
+        docker exec postgres-downstream psql -U postgres -c "SHOW wal_level;"
+        docker exec postgres-downstream psql -U postgres -c "SHOW max_replication_slots;"
     
     - name: Run integration tests
       env:
@@ -93,4 +119,9 @@ jobs:
         POSTGRES_USER: postgres
         POSTGRES_PASSWORD: postgres
         POSTGRES_DB: postgres
+        POSTGRES_DOWNSTREAM_HOST: localhost
+        POSTGRES_DOWNSTREAM_PORT: 5433
+        # Internal hostnames for container-to-container communication (used by subscriptions)
+        POSTGRES_PRIMARY_HOST_INTERNAL: postgres-primary
+        POSTGRES_PRIMARY_PORT_INTERNAL: 5432
       run: go test -v -tags=integration -timeout=300s ./listener/...

docker/docker-compose.yml

@@ -9,12 +9,14 @@ x-postgres-common:
   user: postgres
   restart: always
   healthcheck:
-    test: 'pg_isready -U postgres --dbname=my_db'
+    test: 'pg_isready -U postgres --dbname=postgres'
     interval: 10s
     timeout: 5s
     retries: 5
 
 services:
+  # Primary PostgreSQL - the "upstream" database
+  # This is where we make changes that get replicated to the downstream
   postgres_primary:
     <<: *postgres-common
     ports:
@@ -29,30 +31,31 @@ services:
       postgres 
       -c wal_level=logical
       -c hot_standby=on 
-      -c max_wal_senders=10 
-      -c max_replication_slots=10 
+      -c max_wal_senders=20 
+      -c max_replication_slots=20 
       -c hot_standby_feedback=on
     volumes:
       - ./scripts:/docker-entrypoint-initdb.d
 
-  postgres_replica:
+  # Downstream PostgreSQL - receives logical replication from primary
+  # wal-listener connects here to test foreign origin filtering
+  postgres_downstream:
     <<: *postgres-common
     ports:
       - 5433:5432
     environment:
-      PGUSER: replicator
-      PGPASSWORD: replicator_password
+      POSTGRES_USER: postgres
+      POSTGRES_DB: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_HOST_AUTH_METHOD: "scram-sha-256\nhost replication all 0.0.0.0/0 md5"
+      POSTGRES_INITDB_ARGS: "--auth-host=scram-sha-256"
     command: |
-      bash -c "
-      until pg_basebackup --pgdata=/var/lib/postgresql/data -R --slot=replication_slot --host=postgres_primary --port=5432
-      do
-      echo 'Waiting for primary to connect...'
-      sleep 1s
-      done
-      echo 'Backup done, starting replica...'
-      chmod 0700 /var/lib/postgresql/data
-      postgres
-      "
+      postgres 
+      -c wal_level=logical
+      -c hot_standby=on 
+      -c max_wal_senders=20 
+      -c max_replication_slots=20 
+      -c hot_standby_feedback=on
     depends_on:
       - postgres_primary