End-to-end integration test in CI — and the startup bug it caught

.github/workflows/ci.yml

@@ -56,6 +56,43 @@ jobs:
       - name: Test
         run: go test ./...
 
+  integration:
+    name: End-to-end integration test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+          cache: true
+
+      # Plain `docker run` instead of job services: services can't override
+      # the container command, and Postgres needs -c wal_level=logical.
+      - name: Start PostgreSQL (wal_level=logical) and Redis
+        run: |
+          docker run -d --name pg -p 5432:5432 \
+            -e POSTGRES_PASSWORD=postgres \
+            postgres:16 -c wal_level=logical
+          docker run -d --name redis -p 6379:6379 redis:7
+          for i in $(seq 1 30); do
+            if docker exec pg pg_isready -U postgres -q; then break; fi
+            sleep 1
+          done
+          docker exec pg pg_isready -U postgres
+
+      - name: Run integration tests
+        env:
+          GTC_TEST_DATABASE_URL: postgres://postgres:postgres@localhost:5432/postgres
+          GTC_TEST_REDIS_URL: redis://localhost:6379
+        run: go test -tags integration -v -timeout 5m ./test/integration
+
+      - name: Dump service logs on failure
+        if: failure()
+        run: |
+          docker logs pg || true
+          docker logs redis || true
+
   cross-compile:
     name: Cross-compile all targets
     runs-on: ubuntu-latest

CLAUDE.md

@@ -13,6 +13,10 @@ go build -o gateway ./cmd/gateway
 go test ./...
 go run ./cmd/gateway
 
+# End-to-end test (real PostgreSQL wal_level=logical + Redis; also run in CI):
+GTC_TEST_DATABASE_URL=postgres://... GTC_TEST_REDIS_URL=redis://... \
+  go test -tags integration -timeout 5m ./test/integration
+
 # Dashboard (ui/): the built dist/ is committed and embedded via go:embed,
 # so plain `go build` works without Node. After changing ui/src, rebuild:
 cd ui && npm ci && npm run build

CONTRIBUTING.md

@@ -66,6 +66,20 @@ go vet ./...
 go test ./...
 ```
 
+There is also an end-to-end test that runs the compiled gateway against a
+real PostgreSQL and Redis (CI runs it on every PR). To run it locally you
+need PostgreSQL with `wal_level=logical` and a Redis, e.g.:
+
+```bash
+docker run -d --name pg -p 5432:5432 -e POSTGRES_PASSWORD=postgres \
+  postgres:16 -c wal_level=logical
+docker run -d --name redis -p 6379:6379 redis:7
+
+GTC_TEST_DATABASE_URL=postgres://postgres:postgres@localhost:5432/postgres \
+GTC_TEST_REDIS_URL=redis://localhost:6379 \
+go test -tags integration -v -timeout 5m ./test/integration
+```
+
 - Follow the [Uber Go Style Guide](https://github.com/uber-go/guide/blob/master/style.md).
 - Add or update tests for behavior you change. Correctness-critical areas
   (the pgoutput decoder, LSN acknowledgment, TOAST handling, key templates)

internal/adapters/primary/wal/reader.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -199,17 +200,35 @@ func (r *Reader) setupReplication(ctx context.Context) error {
 	return nil
 }
 
+// quoteLiteral makes a string safe to embed in a simple-protocol query.
+// Replication connections reject the extended query protocol (parameterized
+// queries), so catalog checks on r.conn must be plain SQL.
+func quoteLiteral(s string) string {
+	return "'" + strings.ReplaceAll(s, "'", "''") + "'"
+}
+
+// readSimpleQuery runs one simple-protocol query and returns its rows.
+func readSimpleQuery(ctx context.Context, conn *pgconn.PgConn, sql string) ([][][]byte, error) {
+	results, err := conn.Exec(ctx, sql).ReadAll()
+	if err != nil {
+		return nil, err
+	}
+	if len(results) == 0 {
+		return nil, nil
+	}
+	return results[0].Rows, nil
+}
+
 func (r *Reader) ensurePublication(ctx context.Context) error {
-	checkResult := r.conn.ExecParams(ctx,
-		"SELECT 1 FROM pg_publication WHERE pubname = $1",
-		[][]byte{[]byte(r.config.PublicationName)},
-		nil, nil, nil,
-	).Read()
-	if checkResult.Err != nil {
-		return fmt.Errorf("check publication failed: %w", checkResult.Err)
+	rows, err := readSimpleQuery(ctx, r.conn, fmt.Sprintf(
+		"SELECT 1 FROM pg_publication WHERE pubname = %s",
+		quoteLiteral(r.config.PublicationName),
+	))
+	if err != nil {
+		return fmt.Errorf("check publication failed: %w", err)
 	}
 
-	if len(checkResult.Rows) > 0 {
+	if len(rows) > 0 {
 		r.logger.Info("publication exists", slog.String("publication", r.config.PublicationName))
 		return nil
 	}
@@ -235,17 +254,16 @@ func (r *Reader) ensurePublication(ctx context.Context) error {
 }
 
 func (r *Reader) ensureReplicationSlot(ctx context.Context) (pglogrepl.LSN, error) {
-	checkResult := r.conn.ExecParams(ctx,
-		"SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = $1",
-		[][]byte{[]byte(r.config.SlotName)},
-		nil, nil, nil,
-	).Read()
-	if checkResult.Err != nil {
-		return 0, fmt.Errorf("check replication slot failed: %w", checkResult.Err)
+	rows, err := readSimpleQuery(ctx, r.conn, fmt.Sprintf(
+		"SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = %s",
+		quoteLiteral(r.config.SlotName),
+	))
+	if err != nil {
+		return 0, fmt.Errorf("check replication slot failed: %w", err)
 	}
 
-	if len(checkResult.Rows) > 0 && len(checkResult.Rows[0]) > 0 {
-		lsnStr := string(checkResult.Rows[0][0])
+	if len(rows) > 0 && len(rows[0]) > 0 {
+		lsnStr := string(rows[0][0])
 		startLSN, parseErr := pglogrepl.ParseLSN(lsnStr)
 		if parseErr != nil {
 			return 0, fmt.Errorf("parse confirmed_flush_lsn failed: %w", parseErr)