diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc index a7d64a843b..33182e9b02 100644 --- a/docs/modules/components/pages/inputs/postgres_cdc.adoc +++ b/docs/modules/components/pages/inputs/postgres_cdc.adoc @@ -99,6 +99,7 @@ input: role: "" # No default (optional) role_external_id: "" # No default (optional) roles: [] # No default (optional) + signal_table_name: "" auto_replay_nacks: true batching: count: 0 @@ -561,6 +562,61 @@ Optional external ID for the role assumption. *Default*: `""` +=== `signal_table_name` + +The name of the table used to send control signals to the connector, excluding the schema. The table must +exist in the schema configured via the `schema` field and must have exactly these columns: + +- **id** — any type representable as a string (e.g. `SERIAL`, `BIGSERIAL`, `UUID`, `VARCHAR`) +- **type** — `VARCHAR` — the signal type (see supported signals below) +- **data** — `TEXT` — a JSON object containing signal parameters + +Create the table with: + +```sql +CREATE TABLE . ( + id SERIAL PRIMARY KEY, + type VARCHAR(32), + data TEXT +); +``` + +Signal rows are published as regular output messages (`operation=insert`, `table=`). +To exclude them from downstream processing, filter on the `table` metadata field using a +`mapping` processor: + +```yaml +pipeline: + processors: + - mapping: | + root = if @table == "rpcn_signal_table" { deleted() } else { this } +``` + +**Supported signals** + +**`trigger-snapshot`** — triggers a re-snapshot of one or more tables without dropping the +replication slot. The `data` column must contain a JSON object with a +`dataset` key listing the fully-qualified tables (`schema.table`) to snapshot. +`dataset` must be non-empty; a signal with an empty or absent `dataset` +is ignored and streaming continues uninterrupted. + +```sql +INSERT INTO dbo.rpcn_signal_table (type, data) +VALUES ('trigger-snapshot', '{"dataset": ["dbo.events", "dbo.products"]}'); +``` + + +*Type*: `string` + +*Default*: `""` +Requires version 4.103.0 or newer + +```yml +# Examples + +signal_table_name: rpcn_signal_table +``` + === `auto_replay_nacks` Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to `false` these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. diff --git a/internal/impl/postgresql/bench/Taskfile.yaml b/internal/impl/postgresql/bench/Taskfile.yaml index 69d5119dea..a5cb9b66e7 100644 --- a/internal/impl/postgresql/bench/Taskfile.yaml +++ b/internal/impl/postgresql/bench/Taskfile.yaml @@ -64,6 +64,18 @@ tasks: cmds: - psql {{.PG_DSN}} -c "SELECT pg_drop_replication_slot('bench_slot') FROM pg_replication_slots WHERE slot_name = 'bench_slot';" + psql:create:signal-table: + desc: Create rpcn_signal_table in the public schema used for snapshot signalling + cmds: + - psql {{.PG_DSN}} -c "CREATE TABLE IF NOT EXISTS public.rpcn_signal_table (id SERIAL PRIMARY KEY, type VARCHAR(32), data TEXT);" + + psql:signal:snapshot: + desc: Insert a snapshot signal into public.rpcn_signal_table to trigger a re-snapshot + deps: [psql:create:signal-table] + cmds: + - | + psql {{.PG_DSN}} -c "INSERT INTO public.rpcn_signal_table (type, data) VALUES ('trigger-snapshot', '{\"dataset\": [\"public.Users\"]}');" + # ── CDC / Snapshot benchmarks ──────────────────────────────────────────────── bench:run: diff --git a/internal/impl/postgresql/bench/create.sql b/internal/impl/postgresql/bench/create.sql index 4cf74d14f8..d4b8d14d77 100644 --- a/internal/impl/postgresql/bench/create.sql +++ b/internal/impl/postgresql/bench/create.sql @@ -1,4 +1,9 @@ -- PostgreSQL Benchmark Setup Script +CREATE TABLE IF NOT EXISTS public.rpcn_signal_table ( + id SERIAL PRIMARY KEY, + type VARCHAR(32), + data TEXT +); CREATE TABLE IF NOT EXISTS public.users ( id SERIAL PRIMARY KEY, diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 0e38b08ea0..79d8872a0f 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -46,10 +46,12 @@ const ( fieldMaxParallelSnapshotTables = "max_parallel_snapshot_tables" fieldUnchangedToastValue = "unchanged_toast_value" fieldHeartbeatInterval = "heartbeat_interval" + fieldSignalTableName = "signal_table_name" fieldAWSIAMAuth = "aws" // FieldAWSIAMAuthEnabled enabled field. FieldAWSIAMAuthEnabled = "enabled" - shutdownTimeout = 5 * time.Second + + shutdownTimeout = 5 * time.Second ) func notImportedAWSOptFn(_ context.Context, awsConf *service.ParsedConfig, _ *pgconn.Config, _ *service.Logger) (TokenBuilder, error) { @@ -191,7 +193,54 @@ This connector uses the naming pattern ` + "`pglog_stream_. ( + id SERIAL PRIMARY KEY, + type VARCHAR(32), + data TEXT +); +` + "```" + ` + +Signal rows are published as regular output messages (` + "`operation=insert`" + `, ` + "`table=`" + `). +To exclude them from downstream processing, filter on the ` + "`table`" + ` metadata field using a +` + "`mapping`" + ` processor: + +` + "```yaml" + ` +pipeline: + processors: + - mapping: | + root = if @table == "rpcn_signal_table" { deleted() } else { this } +` + "```" + ` + +**Supported signals** + +**` + "`trigger-snapshot`" + `** — triggers a re-snapshot of one or more tables without dropping the +replication slot. The ` + "`data`" + ` column must contain a JSON object with a +` + "`dataset`" + ` key listing the fully-qualified tables (` + "`schema.table`" + `) to snapshot. +` + "`dataset`" + ` must be non-empty; a signal with an empty or absent ` + "`dataset`" + ` +is ignored and streaming continues uninterrupted. + +` + "```sql" + ` +INSERT INTO dbo.rpcn_signal_table (type, data) +VALUES ('trigger-snapshot', '{"dataset": ["dbo.events", "dbo.products"]}'); +` + "```"). + Example("rpcn_signal_table"). + Default(""). + Advanced(). + Version("4.103.0"), + ). Field(service.NewAutoRetryNacksToggleField()). Field(service.NewBatchPolicyField(fieldBatching)) } @@ -215,6 +264,7 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser heartbeatInterval time.Duration iamAuthEnabled bool iamAuthTokenBuilder TokenBuilder + signalTableName string ) if err := license.CheckRunningEnterprise(mgr); err != nil { @@ -289,6 +339,10 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser return nil, err } + if signalTableName, err = conf.FieldString(fieldSignalTableName); err != nil { + return nil, err + } + awsConf := conf.Namespace(fieldAWSIAMAuth) iamAuthEnabled, _ = awsConf.FieldBool(FieldAWSIAMAuthEnabled) @@ -340,6 +394,7 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser Logger: logger, UnchangedToastValue: unchangedToastValue, HeartbeatInterval: heartbeatInterval, + SignalTableName: signalTableName, }, batching: batching, checkpointLimit: checkpointLimit, @@ -350,10 +405,16 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser snapshotMetrics: snapshotMetrics, replicationLag: replicationLag, stopSig: shutdown.NewSignaller(), + streamSnapshot: streamSnapshot, iamAuthEnabled: iamAuthEnabled, } + // Initialise signaller eagerly so IsPending() is safe to call before the first Connect(). + if i.controlSig, err = NewControlSignaller(schema, signalTableName, logger); err != nil { + return nil, err + } + // Has stopped is how we notify that we're not connected. This will get reset at connection time. i.stopSig.TriggerHasStopped() @@ -395,6 +456,7 @@ type pgStreamInput struct { snapshotMetrics *service.MetricGauge replicationLag *service.MetricGauge + controlSig *postgresSignaller stopSig *shutdown.Signaller // snapshotAckWG tracks in-flight snapshot batches: incremented when a @@ -403,8 +465,8 @@ type pgStreamInput struct { // replication slot is not promoted before snapshot rows are durable. snapshotAckWG sync.WaitGroup - // IAM authentication fields iamAuthEnabled bool + streamSnapshot bool } func (p *pgStreamInput) Connect(ctx context.Context) error { @@ -415,10 +477,30 @@ func (p *pgStreamInput) Connect(ctx context.Context) error { } } + // determine if we have a pending signal and handle it + pending, signal := p.controlSig.IsPending() + if pending && signal.IsSnapshot() { + p.logger.Infof("%q signal pending, triggering re-snapshot", signal.Type) + + p.streamConfig.StreamOldData = true + p.streamConfig.ForceSnapshot = true + defer func() { p.streamConfig.ForceSnapshot = false }() + + p.streamConfig.SnapshotTables = tableNamesFromSchema(signal.Dataset, p.streamConfig.DBSchema) + defer func() { p.streamConfig.SnapshotTables = nil }() + } else { + p.streamConfig.StreamOldData = p.streamSnapshot + } + pgStream, err := pglogicalstream.NewPgStream(ctx, p.streamConfig) if err != nil { return fmt.Errorf("unable to create replication stream: %w", err) } + if pending { + // signal has already been acked, re-shapshot is a best effort + // as we don't want to get in a snapshot loop. + p.controlSig.Reset() + } batcher, err := p.batching.NewBatcher(p.mgr) if err != nil { return err @@ -430,6 +512,7 @@ func (p *pgStreamInput) Connect(ctx context.Context) error { } func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher *service.Batcher) { + p.logger.Debug("Launched processing stream") monitorLoop := asyncroutine.NewPeriodic(p.streamConfig.WalMonitorInterval, func() { // Periodically collect stats report := pgStream.GetProgress() @@ -508,11 +591,17 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher break } var ( - flush bool - mb []byte - err error + flush bool + mb []byte + signalHandled bool ) for _, msg := range batch { + sig, err := p.controlSig.Listen(ctx, msg) + if err != nil { + p.logger.Errorf("failed to detect control signal in change event, skipping message: %s", err) + continue + } + if mb, err = json.Marshal(msg.Data); err != nil { p.logger.Errorf("failure to marshal message: %s", err) break @@ -532,9 +621,47 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher if msg.BeforeData != nil { batchMsg.MetaSetImmut("before", service.ImmutableAny{V: msg.BeforeData}) } - if batcher.Add(batchMsg) { - flush = true + + if sig == nil { + if batcher.Add(batchMsg) { + flush = true + } + continue + } + + // Flush signal and messages preceeding it before switching to re-snapshot. + p.logger.Infof("snapshot signal received (lsn=%s), pausing stream to re-run snapshot", sig.LSN) + batcher.Add(batchMsg) + signalBatch, ferr := batcher.Flush(ctx) + if ferr != nil { + p.logger.Warnf("failed to flush signal message, signal will be retried on redelivery: %s", ferr) + continue + } + if ferr := p.flushBatch(ctx, pgStream, cp, signalBatch); ferr != nil { + p.logger.Warnf("failed to flush signal message, signal will be retried on redelivery: %s", ferr) + continue + } + + // Watch for batch to be checkpointed before re-snapshotting, logging then waiting indefinitely + // matching the snapshot ack barrier above. + p.logger.Infof("waiting to confirm signal LSN %s is acknowledged downstream before restarting to re-run its snapshot", sig.LSN) + const waitInterval = 100 * time.Millisecond + if werr := awaitCheckpointLSN(ctx, cp, sig.LSN, waitInterval); werr != nil { + p.logger.Warnf("stream shutting down while still waiting to confirm the signal's ack; it will be retried once replication resumes: %s", werr) + continue } + + // Store so Connect re-runs the snapshot on reconnect. + p.controlSig.StoreSignal(sig) + p.stopSig.TriggerSoftStop() + signalHandled = true + // Abandon the rest of this batch: already acked up to and + // including the signal, so nothing here is lost - it + // redelivers once replication resumes after the snapshot. + break + } + if signalHandled { + break } if flush { nextTimedBatchChan = nil diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index 674bb8c1ee..ad073bd2c9 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -53,7 +53,19 @@ func GetFakeFlightRecord() FakeFlightRecord { return flightRecord } -func ResourceWithPostgreSQLVersion(t *testing.T, version string) (string, *sql.DB, error) { +// TestDB wraps sql.DB with testing utilities for database integration tests. +type TestDB struct { + *sql.DB + T *testing.T +} + +// MustExec executes a SQL query and fails the test if an error occurs. +func (db *TestDB) MustExec(query string, args ...any) { + _, err := db.Exec(query, args...) + require.NoError(db.T, err) +} + +func ResourceWithPostgreSQLVersion(t *testing.T, version string) (string, *TestDB, error) { ctr, err := testcontainers.Run(t.Context(), "postgres:"+version, testcontainers.WithExposedPorts("5432/tcp"), testcontainers.WithEnv(map[string]string{ @@ -158,7 +170,8 @@ func ResourceWithPostgreSQLVersion(t *testing.T, version string) (string, *sql.D } }) - return databaseURL, db, nil + testDB := &TestDB{db, t} + return databaseURL, testDB, nil } func TestIntegrationPostgresNoTxnMarkers(t *testing.T) { diff --git a/internal/impl/postgresql/pglogicalstream/config.go b/internal/impl/postgresql/pglogicalstream/config.go index 73ff4c1c4d..e08b2669a8 100644 --- a/internal/impl/postgresql/pglogicalstream/config.go +++ b/internal/impl/postgresql/pglogicalstream/config.go @@ -40,6 +40,18 @@ type Config struct { BatchSize int // If true, include BEGIN and COMMIT messages in the stream IncludeTxnMarkers bool + // SnapshotTables overrides DBTables for snapshot scans when non-empty. + // The publication always tracks DBTables regardless of this field. + SnapshotTables []string + // ForceSnapshot forces a snapshot to run even when the replication slot already exists. + // When true and the slot exists, a snapshot is taken against current DB state using a + // REPEATABLE READ transaction rather than an EXPORT_SNAPSHOT. Used for signal-triggered + // re-snapshots that do not need to drop and recreate the slot. + ForceSnapshot bool + // SignalTableName is the name of the signal table. Rows inserted into this + // table are treated as control signals rather than data, and the table is + // excluded from snapshot scans. + SignalTableName string Logger *service.Logger diff --git a/internal/impl/postgresql/pglogicalstream/logical_stream.go b/internal/impl/postgresql/pglogicalstream/logical_stream.go index 13267f3083..1e19d26342 100644 --- a/internal/impl/postgresql/pglogicalstream/logical_stream.go +++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go @@ -54,7 +54,6 @@ type Stream struct { includeTxnMarkers bool slotName string - tables []TableFQN snapshotBatchSize int decodingPluginArguments []string logger *service.Logger @@ -111,6 +110,7 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { } tables = append(tables, TableFQN{Schema: schema, Table: normalized}) } + batchSize := 1000 if config.BatchSize > 0 { batchSize = config.BatchSize @@ -122,7 +122,6 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { snapshotAcked: make(chan struct{}), slotName: config.ReplicationSlotName, snapshotBatchSize: batchSize, - tables: tables, maxSnapshotWorkers: config.MaxSnapshotWorkers, logger: config.Logger, shutSig: shutdown.NewSignaller(), @@ -176,9 +175,18 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { stream.decodingPluginArguments = pluginArguments + tablesForPublication := tables + if config.SignalTableName != "" { + normalizedSignalTable, err := sanitize.NormalizePostgresIdentifier(config.SignalTableName) + if err != nil { + return nil, fmt.Errorf("invalid signal table name %q: %w", config.SignalTableName, err) + } + tablesForPublication = append(slices.Clone(tables), TableFQN{Schema: schema, Table: normalizedSignalTable}) + } + pubName := "pglog_stream_" + config.ReplicationSlotName - stream.logger.Infof("Creating publication %s for tables: %s", pubName, tables) - if err = CreatePublication(ctx, stream.pgConn, pubName, tables); err != nil { + stream.logger.Infof("Creating publication %s for tables: %s", pubName, tablesForPublication) + if err = CreatePublication(ctx, stream.pgConn, pubName, tablesForPublication); err != nil { return nil, err } cleanups = append(cleanups, func() { @@ -210,21 +218,84 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { stream.ackedLSN = confirmedLSNFromDB stream.ackedLSNMu.Unlock() } - if config.StreamOldData { - for _, table := range tables { - stream.monitor.MarkSnapshotComplete(table) + if config.ForceSnapshot && config.StreamOldData { + // SnapshotTables narrows a forced re-snapshot to the tables a + // signal targeted. It must never affect the fresh-slot initial + // snapshot below, which always scans every configured table. + forceSnapshotTables := tables + if len(config.SnapshotTables) > 0 { + forceSnapshotTables = make([]TableFQN, 0, len(config.SnapshotTables)) + for _, table := range config.SnapshotTables { + normalized, err := sanitize.NormalizePostgresIdentifier(table) + if err != nil { + return nil, fmt.Errorf("invalid snapshot table name %q: %w", table, err) + } + forceSnapshotTables = append(forceSnapshotTables, TableFQN{Schema: schema, Table: normalized}) + } } - } - stream.logger.Debugf("starting stream from LSN %s", confirmedLSNFromDB.String()) - if err = stream.startLr(ctx, confirmedLSNFromDB); err != nil { - return nil, err - } - go func() { - defer stream.shutSig.TriggerHasStopped() - if err := stream.streamMessages(confirmedLSNFromDB); err != nil { - stream.errors <- fmt.Errorf("logical replication stream error: %w", err) + + snapshotter, err := newSnapshotter(config, config.DBRawDSN, config.Logger, "", config.MaxSnapshotWorkers) + if err != nil { + return nil, fmt.Errorf("creating snapshotter for re-snapshot: %w", err) + } + go func() { + defer stream.shutSig.TriggerHasStopped() + + ctx, done := stream.shutSig.SoftStopCtx(context.Background()) + defer done() + + if err := stream.processSnapshot(ctx, forceSnapshotTables, snapshotter); err != nil { + stream.errors <- fmt.Errorf("processing snapshot: %w", err) + return + } + + for _, table := range forceSnapshotTables { + stream.monitor.MarkSnapshotComplete(table) + } + + // Emit a sentinel so the input layer knows the re-snapshot is fully + // emitted, then block until it confirms every re-snapshot message + // has been acknowledged downstream before resuming replication - + // avoiding a race between the snapshot's own rows and newly + // streamed WAL changes. Connect already cleared the pending + // signal before this attempt started, so a crash here does not + // retry the re-snapshot. + select { + case stream.messages <- []StreamMessage{{Operation: SnapshotCompleteOpType}}: + case <-ctx.Done(): + return + } + select { + case <-stream.snapshotAcked: + case <-ctx.Done(): + return + } + + if err := stream.startLr(ctx, confirmedLSNFromDB); err != nil { + stream.errors <- fmt.Errorf("starting logical replication: %w", err) + return + } + if err := stream.streamMessages(confirmedLSNFromDB); err != nil { + stream.errors <- fmt.Errorf("logical replication stream error: %w", err) + } + }() + } else { + if config.StreamOldData { + for _, table := range tables { + stream.monitor.MarkSnapshotComplete(table) + } } - }() + stream.logger.Debugf("starting stream from LSN %s", confirmedLSNFromDB.String()) + if err = stream.startLr(ctx, confirmedLSNFromDB); err != nil { + return nil, err + } + go func() { + defer stream.shutSig.TriggerHasStopped() + if err := stream.streamMessages(confirmedLSNFromDB); err != nil { + stream.errors <- fmt.Errorf("logical replication stream error: %w", err) + } + }() + } cleanups = nil return stream, nil } @@ -264,7 +335,7 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { defer done() var startLSN LSN if snapshotter != nil { - if err = stream.processSnapshot(ctx, snapshotter); err != nil { + if err = stream.processSnapshot(ctx, tables, snapshotter); err != nil { stream.errors <- fmt.Errorf("processing snapshot: %w", err) return } @@ -604,7 +675,7 @@ func (s *Stream) processChange(ctx context.Context, msgLSN LSN, xld XLogData, re } } -func (s *Stream) processSnapshot(ctx context.Context, snapshotter *snapshotter) error { +func (s *Stream) processSnapshot(ctx context.Context, tables []TableFQN, snapshotter *snapshotter) error { if err := snapshotter.Prepare(ctx); err != nil { return fmt.Errorf("unable to prepare snapshot: %w", err) } @@ -616,7 +687,7 @@ func (s *Stream) processSnapshot(ctx context.Context, snapshotter *snapshotter) snapshotTasks := []func(context.Context) error{} - for _, table := range s.tables { + for _, table := range tables { s.logger.Infof("Planning snapshot scan for table: %v", table) planStartTime := time.Now() primaryKeyColumns, err := s.getPrimaryKeyColumn(ctx, table) diff --git a/internal/impl/postgresql/pglogicalstream/monitor.go b/internal/impl/postgresql/pglogicalstream/monitor.go index e0644dac4a..e6cef3955a 100644 --- a/internal/impl/postgresql/pglogicalstream/monitor.go +++ b/internal/impl/postgresql/pglogicalstream/monitor.go @@ -81,12 +81,16 @@ func NewMonitor( // UpdateSnapshotProgressForTable updates the snapshot ingestion progress for a given table. func (m *Monitor) UpdateSnapshotProgressForTable(table TableFQN, read int) { - m.snapshotProgress[table].Add(int64(read)) + if p := m.snapshotProgress[table]; p != nil { + p.Add(int64(read)) + } } // MarkSnapshotComplete means that we finished snapshotting. func (m *Monitor) MarkSnapshotComplete(table TableFQN) { - m.snapshotProgress[table].Store(int64(m.tableStat[table])) + if p := m.snapshotProgress[table]; p != nil { + p.Store(int64(m.tableStat[table])) + } } // we need to read the tables stat to calculate the snapshot ingestion progress. diff --git a/internal/impl/postgresql/pglogicalstream/snapshotter.go b/internal/impl/postgresql/pglogicalstream/snapshotter.go index 313a86835e..22bd0e7f3e 100644 --- a/internal/impl/postgresql/pglogicalstream/snapshotter.go +++ b/internal/impl/postgresql/pglogicalstream/snapshotter.go @@ -68,12 +68,14 @@ func (s *snapshotter) openTxn(ctx context.Context, _ int) (*sql.Tx, error) { if err != nil { return nil, fmt.Errorf("unable to start reader txn: %w", err) } - sq, err := sanitize.SQLQuery("SET TRANSACTION SNAPSHOT $1", s.snapshotName) - if err != nil { - return nil, err - } - if _, err := tx.ExecContext(ctx, sq); err != nil { - return nil, fmt.Errorf("unable to set txn snapshot to %s: %w", s.snapshotName, err) + if s.snapshotName != "" { + sq, err := sanitize.SQLQuery("SET TRANSACTION SNAPSHOT $1", s.snapshotName) + if err != nil { + return nil, err + } + if _, err := tx.ExecContext(ctx, sq); err != nil { + return nil, fmt.Errorf("unable to set txn snapshot to %s: %w", s.snapshotName, err) + } } // Oh postgres, pg hackers will tell you the statistics/analyzer just aren't tuned right or up to date, // and they are probably right, but this is the easiest way to tell postgres that we actually want to diff --git a/internal/impl/postgresql/signaller.go b/internal/impl/postgresql/signaller.go new file mode 100644 index 0000000000..ef075f9725 --- /dev/null +++ b/internal/impl/postgresql/signaller.go @@ -0,0 +1,166 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package pgstream + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/Jeffail/checkpoint" + + "github.com/redpanda-data/benthos/v4/public/service" + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream" + "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream/sanitize" + "github.com/redpanda-data/connect/v4/internal/replication" +) + +var _ replication.Signaller = (*postgresSignaller)(nil) + +type postgresSignaller struct { + *replication.ControlSignaller + + schema string + tableName string +} + +// NewControlSignaller creates a replication.Signaller that detects signal +// INSERTs on the given schema.tableName. +func NewControlSignaller(schema, tableName string, log *service.Logger) (*postgresSignaller, error) { + normalizedSchema, err := wireFormPostgresIdentifier(schema) + if err != nil { + return nil, fmt.Errorf("invalid schema %q: %w", schema, err) + } + normalizedTableName := tableName + if tableName != "" { + if normalizedTableName, err = wireFormPostgresIdentifier(tableName); err != nil { + return nil, fmt.Errorf("invalid signal table name %q: %w", tableName, err) + } + } + return &postgresSignaller{ControlSignaller: replication.NewControlSignaller(log), schema: normalizedSchema, tableName: normalizedTableName}, nil +} + +func wireFormPostgresIdentifier(name string) (string, error) { + normalized, err := sanitize.NormalizePostgresIdentifier(name) + if err != nil { + return "", err + } + return sanitize.UnquotePostgresIdentifier(normalized) +} + +// Listen returns any actionable signal found; it does not call StoreSignal - +// that's the caller's job once delivery is confirmed. Signal rows should always be +// forwarded downstream as normal messages regardless of the outcome here. +func (s *postgresSignaller) Listen(_ context.Context, signal any) (*replication.ControlSignal, error) { + msg, ok := signal.(pglogicalstream.StreamMessage) + if !ok { + return nil, nil + } + if msg.Operation != pglogicalstream.InsertOpType { + return nil, nil + } + if msg.Schema != s.schema || msg.Table != s.tableName { + return nil, nil + } + + row, ok := msg.Data.(map[string]any) + if !ok { + return nil, fmt.Errorf("expected map for %s message data, got %T", s.tableName, msg.Data) + } + dataStr, ok := row["data"].(string) + if !ok { + return nil, fmt.Errorf("expected string for %s.data column, got %T", s.tableName, row["data"]) + } + + var sig replication.ControlSignal + if err := json.Unmarshal([]byte(dataStr), &sig); err != nil { + return nil, fmt.Errorf("unmarshaling control signal %s.data: %w", s.tableName, err) + } + + sig.ID = fmt.Sprintf("%v", row["id"]) + + evType, ok := row["type"].(string) + if !ok { + return nil, errors.New("parsing control signals's 'type' data") + } + sig.Type = evType + + log := s.Log.With("id", sig.ID, "type", sig.Type) + + if !sig.IsSnapshot() { + log.Warnf("Control signal %q received but not a recognized action, forwarding as a regular message", sig.Type) + return nil, nil + } + + // Invalid or no-op signals are not returned as actionable, so streaming + // continues uninterrupted. + if len(sig.Dataset) == 0 { + log.Warnf("Control signal %q received but dataset is empty — ignoring, streaming continues uninterrupted", sig.Type) + return nil, nil + } + if len(tableNamesFromSchema(sig.Dataset, s.schema)) == 0 { + log.Warnf("Control signal %q received but dataset %v matched no tables for schema %q — ignoring, streaming continues uninterrupted", sig.Type, sig.Dataset, s.schema) + return nil, nil + } + + log.Infof("Control signal %q received: operation=%s lsn=%v", sig.Type, msg.Operation, msg.LSN) + + if msg.LSN != nil { + sig.LSN = []byte(*msg.LSN) + } + return &sig, nil +} + +// awaitCheckpointLSN blocks until checkpointer's highest resolved offset has reached or passed target, +// meaning every message up to and including it has been acknowledged downstream. An empty target is a no-op. +func awaitCheckpointLSN(ctx context.Context, checkpointer *checkpoint.Capped[*string], target []byte, waitInterval time.Duration) error { + if len(target) == 0 { + return nil + } + targetLSN, err := pglogicalstream.ParseLSN(string(target)) + if err != nil { + return fmt.Errorf("unable to parse target LSN: %w", err) + } + + ticker := time.NewTicker(waitInterval) + defer ticker.Stop() + for { + if highest := checkpointer.Highest(); highest != nil && *highest != nil { + if gotLSN, err := pglogicalstream.ParseLSN(**highest); err == nil && gotLSN >= targetLSN { + return nil + } + } + select { + case <-ticker.C: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +func tableNamesFromSchema(collections []string, schema string) []string { + if len(collections) == 0 { + return nil + } + tables := make([]string, 0, len(collections)) + for _, dc := range collections { + table := dc + if idx := strings.LastIndex(dc, "."); idx >= 0 { + if !strings.EqualFold(dc[:idx], schema) { + continue + } + table = dc[idx+1:] + } + tables = append(tables, table) + } + return tables +} diff --git a/internal/impl/postgresql/signalling_test.go b/internal/impl/postgresql/signalling_test.go new file mode 100644 index 0000000000..2a42e51212 --- /dev/null +++ b/internal/impl/postgresql/signalling_test.go @@ -0,0 +1,694 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package pgstream + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" + "github.com/redpanda-data/benthos/v4/public/service/integration" + "github.com/redpanda-data/connect/v4/internal/license" +) + +func TestIntegrationSignallingConfiguration(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + db.MustExec(`CREATE SCHEMA IF NOT EXISTS dbo`) + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.custom_signal_table (id VARCHAR(32), type VARCHAR(32), data TEXT)`) + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.events (id SERIAL PRIMARY KEY, name TEXT)`) + + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('initial')`) + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('initial')`) + + template := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: test_slot_signalling + stream_snapshot: true + signal_table_name: custom_signal_table + schema: dbo + tables: + - events +`, databaseURL) + + streamOutBuilder := service.NewStreamBuilder() + require.NoError(t, streamOutBuilder.SetLoggerYAML(`level: DEBUG`)) + require.NoError(t, streamOutBuilder.AddInputYAML(template)) + require.NoError(t, streamOutBuilder.AddProcessorYAML(`mapping: 'root = @'`)) + + var ( + received []any + mu sync.Mutex + ) + require.NoError(t, streamOutBuilder.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + data, err := msg.AsStructured() + if err != nil { + return err + } + m := data.(map[string]any) + if _, ok := m["lsn"]; ok { + m["lsn"] = "XXX/XXX" + } + delete(m, "schema") + delete(m, "commit_ts_ms") + received = append(received, m) + } + return nil + })) + + streamOut, err := streamOutBuilder.Build() + require.NoError(t, err) + license.InjectTestService(streamOut.Resources()) + t.Cleanup(func() { + require.NoError(t, streamOut.StopWithin(5*time.Second)) + }) + + go func() { + if err := streamOut.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + + // Wait for the initial snapshot to complete before inserting streaming records. + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 2) + }, 25*time.Second, 100*time.Millisecond) + + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('stream')`) + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('stream')`) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 4) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + }) + mu.Unlock() +} + +func TestIntegrationSignallingDisabledWhenTableNameEmpty(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + db.MustExec(`CREATE SCHEMA IF NOT EXISTS dbo`) + // Shaped exactly like a signal table, and replicated as an ordinary data + // table below - with signal_table_name left unset, it must never be + // treated as one. + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.rpcn_signal_table (id SERIAL PRIMARY KEY, type VARCHAR(32), data TEXT)`) + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.events (id SERIAL PRIMARY KEY, name TEXT)`) + + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('initial')`) + + template := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: test_slot_signalling_disabled + stream_snapshot: true + schema: dbo + tables: + - events + - rpcn_signal_table +`, databaseURL) + + streamOutBuilder := service.NewStreamBuilder() + require.NoError(t, streamOutBuilder.SetLoggerYAML(`level: DEBUG`)) + require.NoError(t, streamOutBuilder.AddInputYAML(template)) + require.NoError(t, streamOutBuilder.AddProcessorYAML(`mapping: 'root = @'`)) + + var ( + received []any + mu sync.Mutex + ) + require.NoError(t, streamOutBuilder.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + data, err := msg.AsStructured() + if err != nil { + return err + } + m := data.(map[string]any) + if _, ok := m["lsn"]; ok { + m["lsn"] = "XXX/XXX" + } + delete(m, "schema") + delete(m, "commit_ts_ms") + received = append(received, m) + } + return nil + })) + + streamOut, err := streamOutBuilder.Build() + require.NoError(t, err) + license.InjectTestService(streamOut.Resources()) + + go func() { + if err := streamOut.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + + // Wait for the initial snapshot row from dbo.events. + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 1) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + received = nil + mu.Unlock() + + // This row is shaped exactly like a trigger-snapshot signal targeting + // dbo.events. With signal_table_name unset, it must be forwarded as an + // ordinary message rather than detected as a signal, and must not + // trigger a re-snapshot of dbo.events. + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('trigger-snapshot', '{"dataset": ["dbo.events"]}')`) + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('stream')`) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 2) + }, 25*time.Second, 100*time.Millisecond) + + // A re-snapshot, if incorrectly triggered, would add further "read" + // messages for dbo.events shortly after. Give that a moment, then confirm + // nothing beyond the two ordinary inserts above ever arrives. + time.Sleep(500 * time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"}, + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + }) + mu.Unlock() + + require.NoError(t, streamOut.StopWithin(10*time.Second)) +} + +func TestIntegrationSignalling(t *testing.T) { + integration.CheckSkip(t) + databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16") + require.NoError(t, err) + + db.MustExec(`CREATE SCHEMA IF NOT EXISTS dbo`) + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.rpcn_signal_table (id SERIAL PRIMARY KEY, type VARCHAR(32), data TEXT)`) + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.events (id SERIAL PRIMARY KEY, name TEXT)`) + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.products (id SERIAL PRIMARY KEY, name TEXT)`) + // dbo.newtable table doesn't exist in replication slot + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.newtable (id SERIAL PRIMARY KEY, name TEXT)`) + + // Pre-insert an events row so the snapshot phase produces a message. + // The signal table must NOT appear in snapshot output. + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('initial')`) + db.MustExec(`INSERT INTO dbo.products (name) VALUES ('initial')`) + + // elements accumulates the expected items across subtests; each subtest + // appends its contribution before asserting with ElementsMatch. + var elements []any + elements = append(elements, + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "read", "table": "products"}, + ) + + template := fmt.Sprintf(` +postgres_cdc: + dsn: %s + slot_name: test_slot_signalling + stream_snapshot: true + signal_table_name: rpcn_signal_table + schema: dbo + tables: + - events + - products +`, databaseURL) + + streamOutBuilder := service.NewStreamBuilder() + require.NoError(t, streamOutBuilder.SetLoggerYAML(`level: DEBUG`)) + require.NoError(t, streamOutBuilder.AddInputYAML(template)) + require.NoError(t, streamOutBuilder.AddProcessorYAML(`mapping: 'root = @'`)) + + var ( + received []any + mu sync.Mutex + ) + require.NoError(t, streamOutBuilder.AddBatchConsumerFunc(func(_ context.Context, batch service.MessageBatch) error { + mu.Lock() + defer mu.Unlock() + for _, msg := range batch { + data, err := msg.AsStructured() + if err != nil { + return err + } + m := data.(map[string]any) + if _, ok := m["lsn"]; ok { + m["lsn"] = "XXX/XXX" + } + delete(m, "schema") + delete(m, "commit_ts_ms") + received = append(received, m) + } + return nil + })) + + streamOut, err := streamOutBuilder.Build() + require.NoError(t, err) + license.InjectTestService(streamOut.Resources()) + + go func() { + if err := streamOut.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) { + t.Error(err) + } + }() + + t.Run("Captures initial snapshot and streaming on start up", func(t *testing.T) { + // Wait for the initial snapshot row from dbo.events and dbo.products. + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 2) + }, 25*time.Second, 100*time.Millisecond) + + // insert streaming records + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('stream')`) + db.MustExec(`INSERT INTO dbo.products (name) VALUES ('stream')`) + + elements = append(elements, + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "read", "table": "products"}, + ) + + // Wait for streaming records of dbo.events and dbo.products + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 4) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "read", "table": "products"}, + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + map[string]any{"operation": "insert", "table": "products", "lsn": "XXX/XXX"}, + }) + mu.Unlock() + }) + + t.Run("Can signal snapshot of one table more than once in a row", func(t *testing.T) { + mu.Lock() + received = nil // reset to assert for this test + mu.Unlock() + + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('trigger-snapshot', '{"dataset": ["dbo.events"]}')`) + + // Wait for the re-snapshot to complete: the signal row is published as a + // normal message plus two snapshot reads of the events table. + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 3) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"}, + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "read", "table": "events"}, + }) + mu.Unlock() + + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('trigger-snapshot', '{"dataset": ["dbo.events"]}')`) + + // Wait for the second re-snapshot: another signal row plus two snapshot reads, + // accumulated on top of the previous three. + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 6) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"}, + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"}, + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "read", "table": "events"}, + }) + mu.Unlock() + }) + + t.Run("Can signal snapshot of multiple tables", func(t *testing.T) { + mu.Lock() + received = nil + mu.Unlock() + + // Insert streaming records and let them arrive via WAL before firing the + // signal, so they don't get counted as snapshot reads. + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('evt1')`) + db.MustExec(`INSERT INTO dbo.products (name) VALUES ('evt1')`) + + elements = append(elements, + map[string]any{"operation": "read", "table": "events"}, + map[string]any{"operation": "read", "table": "products"}, + ) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 2) + }, 25*time.Second, 100*time.Millisecond) + + // Reset once WAL inserts are consumed so the snapshot reads are isolated. + mu.Lock() + received = nil + mu.Unlock() + + // The signal row itself is published as a normal message alongside the + // snapshot reads, so include it in the expected set. + elements = append(elements, map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"}) + expected := len(elements) + + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('trigger-snapshot', '{"dataset": ["dbo.events", "dbo.products"]}')`) + + // Wait for the re-snapshot reads: one per row across both tables. + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, expected) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, elements) + mu.Unlock() + }) + + t.Run("Resumes streaming all configured tables after snapshot", func(t *testing.T) { + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('evt2')`) + db.MustExec(`INSERT INTO dbo.products (name) VALUES ('new2')`) + + elements = append(elements, + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + map[string]any{"operation": "insert", "table": "products", "lsn": "XXX/XXX"}, + ) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, len(elements)) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, elements) + mu.Unlock() + }) + + t.Run("Can resnapshot new tables but not append to the replication slot", func(t *testing.T) { + mu.Lock() + received = nil + mu.Unlock() + + db.MustExec(`CREATE TABLE IF NOT EXISTS dbo.temptable (id SERIAL PRIMARY KEY, name TEXT)`) + + // Insert streaming records and let them arrive via WAL before firing the + // signal, so they don't get counted as snapshot reads. + db.MustExec(`INSERT INTO dbo.temptable (name) VALUES ('evt1')`) + db.MustExec(`INSERT INTO dbo.temptable (name) VALUES ('evt2')`) + + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('trigger-snapshot', '{"dataset": ["dbo.temptable"]}')`) + + // Wait for the signal row plus one snapshot read per temptable row. + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 3) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"}, + map[string]any{"operation": "read", "table": "temptable"}, + map[string]any{"operation": "read", "table": "temptable"}, + }) + mu.Unlock() + }) + + t.Run("Ignores signal and continues streaming when dataset is empty", func(t *testing.T) { + mu.Lock() + received = nil + mu.Unlock() + + // A signal with an empty dataset is a no-op: no snapshot runs and + // WAL streaming continues. The signal row is still published as a normal + // message; only the subsequent streaming insert and the signal row arrive. + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('trigger-snapshot', '{}')`) + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('post-noop')`) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 2) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"}, + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + }) + mu.Unlock() + }) + + t.Run("Drops signal row with malformed JSON data and continues streaming", func(t *testing.T) { + mu.Lock() + received = nil + mu.Unlock() + + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('trigger-snapshot', 'not-valid-json{')`) + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('after-bad-json')`) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 1) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + }) + mu.Unlock() + }) + + t.Run("Drops signal row with NULL data column and continues streaming", func(t *testing.T) { + mu.Lock() + received = nil + mu.Unlock() + + // Omitting data leaves it NULL, so row["data"].(string) fails, Listen + // returns an error, and the row is skipped entirely by the caller. + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type) VALUES ('trigger-snapshot')`) + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('after-null-data')`) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 1) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + }) + mu.Unlock() + }) + + t.Run("Drops signal row with NULL type column and continues streaming", func(t *testing.T) { + mu.Lock() + received = nil + mu.Unlock() + + // Omitting type leaves it NULL, so row["type"].(string) fails, Listen + // returns an error, and the row is skipped entirely by the caller. + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (data) VALUES ('{"dataset": ["dbo.events"]}')`) + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('after-null-type')`) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 1) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + }) + mu.Unlock() + }) + + t.Run("Forwards signal row with unrecognized type without triggering a re-snapshot", func(t *testing.T) { + mu.Lock() + received = nil + mu.Unlock() + + db.MustExec(`INSERT INTO dbo.rpcn_signal_table (type, data) VALUES ('unsupported-signal', '{"dataset": ["dbo.events"]}')`) + db.MustExec(`INSERT INTO dbo.events (name) VALUES ('after-unsupported-type')`) + + assert.EventuallyWithT(t, func(c *assert.CollectT) { + mu.Lock() + defer mu.Unlock() + assert.Len(c, received, 2) + }, 25*time.Second, 100*time.Millisecond) + + mu.Lock() + require.ElementsMatch(t, received, []any{ + map[string]any{"operation": "insert", "table": "rpcn_signal_table", "lsn": "XXX/XXX"}, + map[string]any{"operation": "insert", "table": "events", "lsn": "XXX/XXX"}, + }) + mu.Unlock() + }) + + require.NoError(t, streamOut.StopWithin(10*time.Second)) +} + +func TestNewControlSignallerNormalizesIdentifiers(t *testing.T) { + tests := []struct { + name string + schema string + tableName string + wantSchema string + wantTableName string + wantErr bool + }{ + { + name: "unquoted mixed-case is folded to lowercase, matching what Postgres reports on the wire", + schema: "dbo", + tableName: "RpcnSignalTable", + wantSchema: "dbo", + wantTableName: "rpcnsignaltable", + }, + { + name: "quoted mixed-case preserves case", + schema: "dbo", + tableName: `"RpcnSignalTable"`, + wantSchema: "dbo", + wantTableName: "RpcnSignalTable", + }, + { + name: "empty table name disables signalling and is left untouched", + schema: "dbo", + tableName: "", + wantSchema: "dbo", + wantTableName: "", + }, + { + name: "invalid schema identifier is rejected", + schema: "", + tableName: "rpcn_signal_table", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s, err := NewControlSignaller(tt.schema, tt.tableName, nil) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.wantSchema, s.schema) + assert.Equal(t, tt.wantTableName, s.tableName) + }) + } +} + +func TestControlSignalTableNames(t *testing.T) { + tests := []struct { + name string + dataset []string + schema string + want []string + }{ + { + name: "empty dataset returns nil", + dataset: nil, + schema: "dbo", + want: nil, + }, + { + name: "matching schema.table extracts table name", + dataset: []string{"dbo.events"}, + schema: "dbo", + want: []string{"events"}, + }, + { + name: "multiple entries same schema", + dataset: []string{"dbo.events", "dbo.products"}, + schema: "dbo", + want: []string{"events", "products"}, + }, + { + name: "cross-schema entry is excluded", + dataset: []string{"other.events"}, + schema: "dbo", + want: []string{}, + }, + { + name: "mixed: matching and non-matching schemas", + dataset: []string{"dbo.events", "other.products"}, + schema: "dbo", + want: []string{"events"}, + }, + { + name: "schema match is case-insensitive", + dataset: []string{"DBO.events"}, + schema: "dbo", + want: []string{"events"}, + }, + { + name: "entry without schema prefix is included", + dataset: []string{"events"}, + schema: "dbo", + want: []string{"events"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := tableNamesFromSchema(tt.dataset, tt.schema) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/internal/replication/signalling.go b/internal/replication/signalling.go new file mode 100644 index 0000000000..9b879ec5ca --- /dev/null +++ b/internal/replication/signalling.go @@ -0,0 +1,108 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package replication + +import ( + "context" + "sync/atomic" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +// ControlSignal represents a insert into the signal table. +type ControlSignal struct { + ID string + Type string + Dataset []string `json:"dataset"` + + // LSN is the log sequence number/offset the signal was observed at, in + // whatever raw form the connector's replication stream represents + // positions (e.g. a decimal/hex string, or raw binary bytes for + // connectors like Oracle whose SCNs aren't naturally textual). It is + // populated by the connector's Listen implementation, not part of the + // signal's own encoded payload. + LSN []byte `json:"-"` +} + +// IsSnapshot returns true if the ControlSignal is a snapshot signal. +func (s *ControlSignal) IsSnapshot() bool { + if s == nil { + return false + } + + return s.Type == "trigger-snapshot" +} + +// Signaller is implemented by connector-specific control signal handlers. +// Listen inspects a decoded replication message and, if it recognizes an +// actionable signal, returns it directly - in the same call that detected it +// - so the caller can flush exactly that batch before acting on it, rather +// than reacting to a separately-scheduled notification a differently-timed +// flush could race ahead of. Only signals whose action could be interrupted +// by a crash need this: anything else should return (nil, nil) and be +// acknowledged like an ordinary message. +// +// Whether the signal's own position is acknowledged before or after its +// action completes is a connector-specific choice: acking before (once +// delivery is confirmed) is simpler and never redelivers, at the cost of +// losing the action if a crash lands between the ack and the action +// finishing; acking after avoids that window but means a source that +// redelivers unacked events will keep resending the signal until Listen +// recognizes and drops the redelivery itself. +// +// ControlSignaller intentionally does not implement Listen itself: it has no +// way to know the shape of a connector's replication messages, so embedding +// it alone does not satisfy this interface. Connectors must embed +// ControlSignaller and provide their own Listen (see +// internal/impl/postgresql/signaller.go for an example) — this way, a +// connector that forgets to do so fails to compile rather than panicking at +// runtime. +type Signaller interface { + Listen(ctx context.Context, event any) (*ControlSignal, error) + IsPending() (bool, *ControlSignal) + Reset() +} + +// ControlSignaller can be used to handle and process control signals. +// +// Listen implementations must not call StoreSignal until the signal (and +// anything batched ahead of it) has actually been acknowledged downstream — +// otherwise a caller relying on IsPending to decide whether to act on the +// signal (e.g. re-running a snapshot) could act on it before it's durably +// delivered. +type ControlSignaller struct { + Log *service.Logger + + signalPending atomic.Pointer[ControlSignal] +} + +// NewControlSignaller creates a ControlSignaller for detecting signal INSERTs +// on the given schema.tableName. Embed the result in a connector-specific type +// that implements Listen to obtain a full Signaller. +func NewControlSignaller(log *service.Logger) *ControlSignaller { + return &ControlSignaller{ + Log: log, + } +} + +// IsPending determines if there's a signal current pending. +func (o *ControlSignaller) IsPending() (bool, *ControlSignal) { + sig := o.signalPending.Load() + return sig != nil, sig +} + +// Reset resets the current captured signal. +func (o *ControlSignaller) Reset() { + o.signalPending.Store(nil) +} + +// StoreSignal stores the published signal ready for reading when needed. +func (o *ControlSignaller) StoreSignal(sig *ControlSignal) { + o.signalPending.Store(sig) +}