feat(datastore,postgres): add logical replication support to Postgres WatchAPI

CHANGELOG.md

@@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 ## [Unreleased]
 ### Added
 - New metric `check_permissionship_total` for CheckPermission and CheckBulkPermissions that counts the number of requests that returned HAS_PERMISSION. Also, `write_relationships_updates` also includes BulkImport calls (https://github.com/authzed/spicedb/pull/3240)
+- Postgres: opt-in logical-replication-based Watch API implementation (`--datastore-watch-logical-replication-enabled`) that streams changes from the WAL in exact commit order through a temporary replication slot, instead of polling the transactions table. Watch revisions gain a byte-sortable commit-LSN component (embedded alongside the usual snapshot, so ZedTokens remain fully usable across the rest of the API), watch no longer requires `track_commit_timestamp=on` in this mode, and `EmitImmediatelyStrategy` is now supported on Postgres. Requires `wal_level=logical` and the `REPLICATION` privilege; the polling watch remains the default (https://github.com/authzed/spicedb/pull/3245)
 
 ### Changed
 - Schema: reads inside write transactions now use a cheap hash-only lookup (`schema_revision`) to check the cache before loading the full schema blob, reducing DB round-trips on cache hits (https://github.com/authzed/spicedb/pull/3160)

go.mod

@@ -56,6 +56,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0
 	github.com/hashicorp/go-memdb v1.3.5
 	github.com/jackc/pgio v1.0.0
+	github.com/jackc/pglogrepl v0.0.0-20260401131349-e37c41485510
 	github.com/jackc/pgx-zerolog v0.0.0-20230315001418-f978528409eb
 	github.com/jackc/pgx/v5 v5.9.2
 	github.com/jeroenrinzema/psql-wire v0.17.0

go.sum

@@ -303,6 +303,8 @@ github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE=
 github.com/jackc/pgio v1.0.0/go.mod h1:oP+2QK2wFfUWgr+gxjoBH9KGBb31Eio69xUb0w5bYf8=
+github.com/jackc/pglogrepl v0.0.0-20260401131349-e37c41485510 h1:+PJCokZ2BhyDKlncScmiNzBwqOx+yH1i8xRlWN/wn6A=
+github.com/jackc/pglogrepl v0.0.0-20260401131349-e37c41485510/go.mod h1:UzTJ5Jjuf4O9hYWW+HYVwVldYz9J7CaePW0iuNJkrPQ=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=

internal/datastore/postgres/gc.go

@@ -12,6 +12,7 @@
 
 	"github.com/authzed/spicedb/internal/datastore/common"
 	"github.com/authzed/spicedb/internal/datastore/postgres/schema"
+	log "github.com/authzed/spicedb/internal/logging"
 	"github.com/authzed/spicedb/pkg/datastore"
 	"github.com/authzed/spicedb/pkg/spiceerrors"
 )
@@ -183,6 +184,37 @@
 		return removed, fmt.Errorf("failed to GC transactions table: %w", err)
 	}
 
+	// Then the commit positions of those same transactions, if the commit LSN
+	// ledger is in use.
+	//
+	// This must run *after* the transactions above, and the order is not
+	// cosmetic. These are separate statements, so there is always a moment
+	// between them, and a crash can make that moment permanent. Deleting
+	// positions first would leave transaction rows that appear to have no
+	// recorded position, which is exactly how a transaction lost to a
+	// replication slot recreation appears — a watch reading them would refuse to
+	// deliver and tell its consumer to re-bootstrap. Deleting them second leaves
+	// the opposite: positions whose transaction is gone, which every reader
+	// joins away.
+	//
+	// ledger_gap is deliberately not pruned. It gains one row per slot
+	// invalidation, which should be rare enough to alert on, and a gap row stops
+	// matching watches on its own once every consumer's cursor is above it.
+	positionsRemoved, err := pgg.batchDelete(
+		ctx,
+		conn,
+		schema.TableLedgerXidLSN,
+		gcPKCols,
+		sq.Lt{schema.ColXID: minTxAlive},
+		nil,
+	)
+	if err != nil {
+		return removed, fmt.Errorf("failed to GC commit positions table: %w", err)
+	}
+	if positionsRemoved > 0 {
+		log.Ctx(ctx).Trace().Int64("commit-positions", positionsRemoved).Msg("deleted recorded commit positions")
+	}
+
 	// Delete any namespace rows with deleted_transaction <= the transaction ID.
 	removed.Namespaces, err = pgg.batchDelete(
 		ctx,