From 7f6fb5c71de3da513bea8e693985841e4e62d8ff Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Thu, 28 May 2026 11:48:38 +0100 Subject: [PATCH 1/2] postgres_cdc: add signalling support postgres_cdc: update tests to verify singal event postgres_cdc: wait for messages to be acked before restarting --- .../components/pages/inputs/postgres_cdc.adoc | 56 ++ internal/impl/postgresql/bench/Taskfile.yaml | 12 + .../postgresql/bench/benchmark_config.yaml | 18 +- internal/impl/postgresql/bench/create.sql | 5 + internal/impl/postgresql/control_signal.go | 150 ++++ .../control_signal_integration_test.go | 639 ++++++++++++++++++ internal/impl/postgresql/input_pg_stream.go | 177 ++++- internal/impl/postgresql/integration_test.go | 17 +- .../impl/postgresql/pglogicalstream/config.go | 8 + .../pglogicalstream/logical_stream.go | 285 ++++++-- .../postgresql/pglogicalstream/monitor.go | 8 +- .../postgresql/pglogicalstream/snapshotter.go | 14 +- internal/replication/signalling.go | 66 ++ 13 files changed, 1386 insertions(+), 69 deletions(-) create mode 100644 internal/impl/postgresql/control_signal.go create mode 100644 internal/impl/postgresql/control_signal_integration_test.go create mode 100644 internal/replication/signalling.go diff --git a/docs/modules/components/pages/inputs/postgres_cdc.adoc b/docs/modules/components/pages/inputs/postgres_cdc.adoc index a7d64a843b..fc0278a885 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. Leave empty (the default) to disable signalling entirely. 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** + +**`execute-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 +`data-collections` key listing the fully-qualified tables (`schema.table`) to snapshot. +`data-collections` must be non-empty; a signal with an empty or absent `data-collections` +is ignored and streaming continues uninterrupted. + +```sql +INSERT INTO dbo.rpcn_signal_table (type, data) +VALUES ('execute-snapshot', '{"data-collections": ["dbo.events", "dbo.products"]}'); +``` + + +*Type*: `string` + +*Default*: `""` +Requires version 4.102.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..d2b30aa4dd 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 ('execute-snapshot', '{\"data-collections\": [\"public.Cart\"]}');" + # ── CDC / Snapshot benchmarks ──────────────────────────────────────────────── bench:run: diff --git a/internal/impl/postgresql/bench/benchmark_config.yaml b/internal/impl/postgresql/bench/benchmark_config.yaml index a673fc8503..2371b2b0e1 100644 --- a/internal/impl/postgresql/bench/benchmark_config.yaml +++ b/internal/impl/postgresql/bench/benchmark_config.yaml @@ -11,8 +11,8 @@ input: - products - cart slot_name: bench_slot + signal_table_name: rpcn_signal_table batching: - count: 1000 period: 1s # pipeline: @@ -22,14 +22,14 @@ input: # root.metadata = @ output: - processors: - - benchmark: - interval: 1s - count_bytes: true - # file: - # path: "./benchmark_results.json" - # codec: lines - drop: {} + # processors: + # - benchmark: + # interval: 1s + # count_bytes: true + file: + path: "./benchmark_results.json" + codec: lines + # drop: {} logger: level: INFO 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/control_signal.go b/internal/impl/postgresql/control_signal.go new file mode 100644 index 0000000000..6e7456cf07 --- /dev/null +++ b/internal/impl/postgresql/control_signal.go @@ -0,0 +1,150 @@ +// 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/replication" +) + +var _ replication.Signaller = (*pgSignaller)(nil) + +type pgSignaller struct { + *replication.ControlSignaller + + schema string + signalTableName string +} + +// newPGSignaller creates a replication.Signaller that detects signal INSERTs on the given schema.tableName. +func newPGSignaller(schema, tableName string, log *service.Logger) *pgSignaller { + s := replication.NewControlSignaller(log) + return &pgSignaller{ControlSignaller: s, schema: schema, signalTableName: tableName} +} + +// Listen returns any actionable signal found. Signal rows are always +// forwarded downstream as normal messages regardless of the outcome here. +// +// An empty tableName means signalling is disabled: every message is ignored. +// +// Only validated execute-snapshot signals return non-nil - everything else +// (an unsupported type, or a validated no-op) returns (nil, nil) and is +// acked immediately like any other message. +func (s *pgSignaller) Listen(_ context.Context, signal any) (*replication.ControlSignal, error) { + if s.signalTableName == "" { + return nil, nil + } + msg, ok := signal.(pglogicalstream.StreamMessage) + if !ok { + return nil, nil + } + if msg.Schema != s.schema || msg.Table != s.signalTableName { + return nil, nil + } + if msg.Operation != pglogicalstream.InsertOpType { + return nil, nil + } + + // deserialise control signal + row, ok := msg.Data.(map[string]any) + if !ok { + return nil, fmt.Errorf("expected map for %s message data, got %T", s.signalTableName, msg.Data) + } + dataStr, ok := row["data"].(string) + if !ok { + return nil, fmt.Errorf("expected string for %s.data column, got %T", s.signalTableName, row["data"]) + } + var sig replication.ControlSignal + if err := json.Unmarshal([]byte(dataStr), &sig); err != nil { + return nil, fmt.Errorf("unmarshaling signal %s.data: %w", s.signalTableName, err) + } + sig.ID = fmt.Sprintf("%v", row["id"]) + if sig.Type, ok = row["type"].(string); !ok { + return nil, errors.New("parsing 'type' data") + } + + log := s.Log.With("id", sig.ID, "type", sig.Type) + + if !sig.IsSnapshot() { + log.Infof("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. + if len(sig.DataCollections) == 0 { + log.Warnf("Signal %q received but data-collections is empty — ignoring, streaming continues uninterrupted", sig.Type) + return nil, nil + } + if len(tableNamesFromSchema(sig.DataCollections, s.schema)) == 0 { + log.Warnf("Signal %q received but data-collections %v matched no tables for schema %q — ignoring, streaming continues uninterrupted", sig.Type, sig.DataCollections, s.schema) + return nil, nil + } + + log.Infof("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, cp *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 := cp.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/control_signal_integration_test.go b/internal/impl/postgresql/control_signal_integration_test.go new file mode 100644 index 0000000000..4af0f08d05 --- /dev/null +++ b/internal/impl/postgresql/control_signal_integration_test.go @@ -0,0 +1,639 @@ +// 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()) + + 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 an execute-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 ('execute-snapshot', '{"data-collections": ["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 ('execute-snapshot', '{"data-collections": ["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 ('execute-snapshot', '{"data-collections": ["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 ('execute-snapshot', '{"data-collections": ["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 ('execute-snapshot', '{"data-collections": ["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 data-collections is empty", func(t *testing.T) { + mu.Lock() + received = nil + mu.Unlock() + + // A signal with an empty data-collections 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 ('execute-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 ('execute-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 ('execute-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 ('{"data-collections": ["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', '{"data-collections": ["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 TestControlSignalTableNames(t *testing.T) { + tests := []struct { + name string + dataCollections []string + schema string + want []string + }{ + { + name: "empty data-collections returns nil", + dataCollections: nil, + schema: "dbo", + want: nil, + }, + { + name: "matching schema.table extracts table name", + dataCollections: []string{"dbo.events"}, + schema: "dbo", + want: []string{"events"}, + }, + { + name: "multiple entries same schema", + dataCollections: []string{"dbo.events", "dbo.products"}, + schema: "dbo", + want: []string{"events", "products"}, + }, + { + name: "cross-schema entry is excluded", + dataCollections: []string{"other.events"}, + schema: "dbo", + want: []string{}, + }, + { + name: "mixed: matching and non-matching schemas", + dataCollections: []string{"dbo.events", "other.products"}, + schema: "dbo", + want: []string{"events"}, + }, + { + name: "schema match is case-insensitive", + dataCollections: []string{"DBO.events"}, + schema: "dbo", + want: []string{"events"}, + }, + { + name: "entry without schema prefix is included", + dataCollections: []string{"events"}, + schema: "dbo", + want: []string{"events"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := tableNamesFromSchema(tt.dataCollections, tt.schema) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 0e38b08ea0..cbb1c5f788 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -27,6 +27,7 @@ import ( "github.com/redpanda-data/connect/v4/internal/asyncroutine" "github.com/redpanda-data/connect/v4/internal/impl/postgresql/pglogicalstream" "github.com/redpanda-data/connect/v4/internal/license" + "github.com/redpanda-data/connect/v4/internal/replication" ) const ( @@ -46,6 +47,7 @@ const ( fieldMaxParallelSnapshotTables = "max_parallel_snapshot_tables" fieldUnchangedToastValue = "unchanged_toast_value" fieldHeartbeatInterval = "heartbeat_interval" + fieldSignalTableName = "signal_table_name" fieldAWSIAMAuth = "aws" // FieldAWSIAMAuthEnabled enabled field. FieldAWSIAMAuthEnabled = "enabled" @@ -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** + +**` + "`execute-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 +` + "`data-collections`" + ` key listing the fully-qualified tables (` + "`schema.table`" + `) to snapshot. +` + "`data-collections`" + ` must be non-empty; a signal with an empty or absent ` + "`data-collections`" + ` +is ignored and streaming continues uninterrupted. + +` + "```sql" + ` +INSERT INTO dbo.rpcn_signal_table (type, data) +VALUES ('execute-snapshot', '{"data-collections": ["dbo.events", "dbo.products"]}'); +` + "```"). + Example("rpcn_signal_table"). + Default(""). + Advanced(). + Version("4.102.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,20 +394,25 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser Logger: logger, UnchangedToastValue: unchangedToastValue, HeartbeatInterval: heartbeatInterval, + SignalTableName: signalTableName, }, batching: batching, checkpointLimit: checkpointLimit, msgChan: make(chan asyncMessage), + signalQueue: make(chan *replication.ControlSignal, 1), mgr: mgr, logger: mgr.Logger(), snapshotMetrics: snapshotMetrics, replicationLag: replicationLag, stopSig: shutdown.NewSignaller(), + streamSnapshot: streamSnapshot, iamAuthEnabled: iamAuthEnabled, } + i.controlSig = newPGSignaller(schema, signalTableName, logger) + // Has stopped is how we notify that we're not connected. This will get reset at connection time. i.stopSig.TriggerHasStopped() @@ -395,8 +454,15 @@ type pgStreamInput struct { snapshotMetrics *service.MetricGauge replicationLag *service.MetricGauge + controlSig *pgSignaller stopSig *shutdown.Signaller + // signalQueue hands a detected signal off to runSignalWorker. Capacity 1 + // is enough: processStream pauses WAL forwarding before enqueueing (see + // PauseWALStreaming), so it can't detect another signal until the + // current one's re-snapshot completes and resumes it. + signalQueue chan *replication.ControlSignal + // snapshotAckWG tracks in-flight snapshot batches: incremented when a // snapshot batch (nil LSN) is enqueued and decremented when it is // acknowledged. The snapshot->stream handoff blocks until it drains so the @@ -405,6 +471,7 @@ type pgStreamInput struct { // IAM authentication fields iamAuthEnabled bool + streamSnapshot bool // original value of streamConfig.StreamOldData, preserved for reconnects } func (p *pgStreamInput) Connect(ctx context.Context) error { @@ -415,6 +482,8 @@ func (p *pgStreamInput) Connect(ctx context.Context) error { } } + 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) @@ -426,10 +495,15 @@ func (p *pgStreamInput) Connect(ctx context.Context) error { // Reset our stop signal p.stopSig = shutdown.NewSignaller() go p.processStream(pgStream, batcher) + + if p.streamConfig.SignallingEnabled() { + go p.runSignalWorker(pgStream) + } return err } 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() @@ -468,7 +542,7 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher p.logger.Debugf("timed flush batch error: %s", err) break } - if err := p.flushBatch(ctx, pgStream, cp, flushedBatch); err != nil { + if err := p.flushBatch(ctx, pgStream, cp, flushedBatch, false); err != nil { p.logger.Debugf("failed to flush batch: %s", err) break } @@ -488,7 +562,7 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher p.stopSig.TriggerSoftStop() break } - if err := p.flushBatch(ctx, pgStream, cp, flushedBatch); err != nil { + if err := p.flushBatch(ctx, pgStream, cp, flushedBatch, false); err != nil { p.logger.Debugf("failed to flush snapshot completion batch: %s", err) p.stopSig.TriggerSoftStop() break @@ -510,9 +584,14 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher var ( flush bool mb []byte - err error ) for _, msg := range batch { + sig, err := p.controlSig.Listen(ctx, msg) + if err != nil { + p.logger.Errorf("failed to detect snapshot 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,6 +611,12 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher if msg.BeforeData != nil { batchMsg.MetaSetImmut("before", service.ImmutableAny{V: msg.BeforeData}) } + + if sig != nil { + p.handleControlSignal(ctx, pgStream, cp, batcher, sig, batchMsg) + continue + } + if batcher.Add(batchMsg) { flush = true } @@ -543,7 +628,7 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher p.logger.Debugf("error flushing batch: %s", err) break } - if err := p.flushBatch(ctx, pgStream, cp, flushedBatch); err != nil { + if err := p.flushBatch(ctx, pgStream, cp, flushedBatch, false); err != nil { p.logger.Debugf("failed to flush batch: %s", err) break } @@ -568,6 +653,7 @@ func (p *pgStreamInput) flushBatch( pgStream *pglogicalstream.Stream, checkpointer *checkpoint.Capped[*string], batch service.MessageBatch, + holdAck bool, ) error { if len(batch) == 0 { return nil @@ -601,6 +687,11 @@ func (p *pgStreamInput) flushBatch( if maxLSN == nil { return nil } + if holdAck { + // Held back deliberately: this is the signal's own flush, and its + // LSN must stay unconfirmed until its re-snapshot durably completes. + return nil + } if err = pgStream.AckLSN(ctx, *maxLSN); err != nil { return fmt.Errorf("unable to ack LSN to postgres: %w", err) } @@ -647,3 +738,79 @@ func (p *pgStreamInput) Close(ctx context.Context) error { } return nil } + +// handleControlSignal flushes what preceded the signal in batch normally, +// then flushes the signal alone with its ack held back so only its own +// position stays unconfirmed, waits for that flush to be visible downstream, +// and hands the signal off to runSignalWorker to run its re-snapshot. +func (p *pgStreamInput) handleControlSignal( + ctx context.Context, + pgStream *pglogicalstream.Stream, + cp *checkpoint.Capped[*string], + batcher *service.Batcher, + sig *replication.ControlSignal, + batchMsg *service.Message, +) { + // flush batch preceding signal + p.logger.Infof("snapshot signal received (lsn=%s), queueing re-snapshot", sig.LSN) + if precedingBatch, ferr := batcher.Flush(ctx); ferr == nil { + if ferr := p.flushBatch(ctx, pgStream, cp, precedingBatch, false); ferr != nil { + p.logger.Debugf("failed to flush batch preceding signal: %s", ferr) + } + } + + // flush signal, holding back the PG ack until snapshot is done, + // this way if snapshot fails we try again. + 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) + return + } + if ferr := p.flushBatch(ctx, pgStream, cp, signalBatch, true); ferr != nil { + p.logger.Warnf("failed to flush signal message, signal will be retried on redelivery: %s", ferr) + return + } + + // Wait for downstream delivery (not a Postgres ack, which stays held + // back) so the signal is visible before any re-snapshot rows it triggers. + const waitInterval = 100 * time.Millisecond + if werr := awaitCheckpointLSN(ctx, cp, sig.LSN, waitInterval); werr != nil { + p.logger.Warnf("gave up waiting to acknowledge signal LSN, streaming continues uninterrupted: %s", werr) + return + } + + // synchronously pause wal streaming before snapshotting + pgStream.PauseStreaming() + + // hand off control signal (snapshot in this case) + select { + case p.signalQueue <- sig: + case <-ctx.Done(): + pgStream.ResumeStreaming() + } +} + +// runSignalWorker runs each detected signal's re-snapshot on pgStream's +// live connection. Must run on its own goroutine: RunForcedSnapshot sends +// scanned rows and its completion sentinel on the same channel processStream +// drains, so processStream must stay free to keep receiving from it. +func (p *pgStreamInput) runSignalWorker(pgStream *pglogicalstream.Stream) { + ctx, cancel := p.stopSig.SoftStopCtx(context.Background()) + defer cancel() + for { + select { + case sig := <-p.signalQueue: + tables := tableNamesFromSchema(sig.DataCollections, p.streamConfig.DBSchema) + if err := pgStream.RunSnapshot(ctx, tables); err != nil { + p.logger.Errorf("forced re-snapshot for signal '%s' failed, signal will be retried on redelivery: %s", sig.Type, err) + continue + } + if err := pgStream.AckLSN(ctx, string(sig.LSN)); err != nil { + p.logger.Warnf("failed to acknowledge signal '%s' after re-snapshot: %s", sig.Type, err) + } + case <-ctx.Done(): + return + } + } +} diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index 52d3b1e71c..2126ca9bb9 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..0a382186ea 100644 --- a/internal/impl/postgresql/pglogicalstream/config.go +++ b/internal/impl/postgresql/pglogicalstream/config.go @@ -40,6 +40,9 @@ type Config struct { BatchSize int // If true, include BEGIN and COMMIT messages in the stream IncludeTxnMarkers bool + // Signal table is treated as control signals rather than data so along with being + // excluded from snapshot scans, is appended to list of DBTables. + SignalTableName string Logger *service.Logger @@ -51,3 +54,8 @@ type Config struct { // The interval to send logical messages HeartbeatInterval time.Duration } + +// SignallingEnabled checks to see if signalling is enabled. +func (c *Config) SignallingEnabled() bool { + return c.SignalTableName != "" +} diff --git a/internal/impl/postgresql/pglogicalstream/logical_stream.go b/internal/impl/postgresql/pglogicalstream/logical_stream.go index 13267f3083..104f8accdf 100644 --- a/internal/impl/postgresql/pglogicalstream/logical_stream.go +++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go @@ -36,6 +36,8 @@ const decodingPlugin = "pgoutput" type Stream struct { pgConn *pgconn.PgConn + config *Config + schema string shutSig *shutdown.Signaller ackedLSNMu sync.Mutex @@ -46,15 +48,19 @@ type Stream struct { messages chan []StreamMessage errors chan error - // snapshotAcked is closed (once) by the input layer via - // MarkSnapshotAcknowledged once every snapshot message has been acknowledged - // downstream, unblocking promotion of the replication slot. - snapshotAcked chan struct{} - snapshotAckedOnce sync.Once + // snapshotAcked is the completion signal for the snapshot round in + // flight, if any. beginSnapshotAck creates it before that round's + // sentinel is sent; MarkSnapshotAcknowledged closes it once acked + // downstream. A fresh channel per round lets RunForcedSnapshot run + // repeatedly over the Stream's life. + snapshotAckMu sync.Mutex + snapshotAcked chan struct{} + + // walPause tracks WAL forwarding pause state - see PauseWALStreaming. + walPause walPause includeTxnMarkers bool slotName string - tables []TableFQN snapshotBatchSize int decodingPluginArguments []string logger *service.Logger @@ -111,18 +117,19 @@ 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 } stream := &Stream{ pgConn: dbConn, + config: config, + schema: schema, messages: make(chan []StreamMessage), errors: make(chan error, 1), - snapshotAcked: make(chan struct{}), slotName: config.ReplicationSlotName, snapshotBatchSize: batchSize, - tables: tables, maxSnapshotWorkers: config.MaxSnapshotWorkers, logger: config.Logger, shutSig: shutdown.NewSignaller(), @@ -176,9 +183,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() { @@ -264,7 +280,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 } @@ -278,13 +294,14 @@ func NewPgStream(ctx context.Context, config *Config) (*Stream, error) { // slot. This guarantees a crash during the handoff re-runs the // snapshot on restart instead of losing the un-acked rows: the // permanent slot is only created past this barrier. + acked := stream.beginSnapshotAck() select { case stream.messages <- []StreamMessage{{Operation: SnapshotCompleteOpType}}: case <-ctx.Done(): return } select { - case <-stream.snapshotAcked: + case <-acked: case <-ctx.Done(): return } @@ -449,6 +466,9 @@ func (s *Stream) streamMessages(currentLSN LSN) error { ctx, done := s.shutSig.SoftStopCtx(context.Background()) defer done() for !s.shutSig.IsSoftStopSignalled() { + if err := s.waitIfWALPaused(ctx); err != nil { + return err + } if committed, err := commitLSN(time.Now().After(nextStandbyMessageDeadline)); err != nil { return err } else if committed { @@ -497,7 +517,7 @@ func (s *Stream) streamMessages(currentLSN LSN) error { return fmt.Errorf("parsing XLogData: %w", err) } msgLSN := xld.WALStart + LSN(len(xld.WALData)) - result, err := s.processChange(ctx, msgLSN, xld, relations, typeMap, schemaCache, ¤tTxnCommitTime) + result, message, err := s.processChange(msgLSN, xld, relations, typeMap, schemaCache, ¤tTxnCommitTime) if err != nil { return fmt.Errorf("decoding postgres changes failed: %w", err) } @@ -511,6 +531,13 @@ func (s *Stream) streamMessages(currentLSN LSN) error { lastEmittedLSN = msgLSN lastEmittedCommitLSN = msgLSN } + if message != nil { + select { + case s.messages <- []StreamMessage{*message}: + case <-ctx.Done(): + return ctx.Err() + } + } default: return fmt.Errorf("unknown message type: %c", msg.Data[0]) } @@ -527,11 +554,13 @@ const ( changeResultEmittedMessage processChangeResult = 2 ) -// Handle handles the pgoutput output. -func (s *Stream) processChange(ctx context.Context, msgLSN LSN, xld XLogData, relations map[uint32]*RelationMessage, typeMap *pgtype.Map, schemaCache map[uint32]any, currentTxnCommitTime *time.Time) (processChangeResult, error) { +// processChange decodes a single WAL change and returns the StreamMessage to +// emit for it, if any. Does not send to s.messages itself - the caller does +// that. +func (s *Stream) processChange(msgLSN LSN, xld XLogData, relations map[uint32]*RelationMessage, typeMap *pgtype.Map, schemaCache map[uint32]any, currentTxnCommitTime *time.Time) (processChangeResult, *StreamMessage, error) { logicalMsg, err := Parse(xld.WALData) if err != nil { - return changeResultNoMessage, err + return changeResultNoMessage, nil, err } // Invalidate the schema cache when a RelationMessage arrives — PostgreSQL sends one @@ -551,23 +580,23 @@ func (s *Stream) processChange(ctx context.Context, msgLSN LSN, xld XLogData, re // parse changes inside the transaction message, err := toStreamMessage(logicalMsg, relations, typeMap, s.unchangedToastValue) if err != nil { - return changeResultNoMessage, err + return changeResultNoMessage, nil, err } if message == nil { // In the case of heartbeats we can treat that the same as suppressed commit messages and advance the LSN that way. // this is only needed for low frequency tables to continue to progress the LSN. if logicalMsg, ok := logicalMsg.(*LogicalDecodingMessage); ok && logicalMsg.Prefix == "redpanda_connect_"+s.slotName { - return changeResultSuppressedCommitMessage, nil + return changeResultSuppressedCommitMessage, nil, nil } - return changeResultNoMessage, nil + return changeResultNoMessage, nil, nil } if !s.includeTxnMarkers { switch message.Operation { case CommitOpType: - return changeResultSuppressedCommitMessage, nil + return changeResultSuppressedCommitMessage, nil, nil case BeginOpType: - return changeResultNoMessage, nil + return changeResultNoMessage, nil, nil } } @@ -596,15 +625,10 @@ func (s *Stream) processChange(ctx context.Context, msgLSN LSN, xld XLogData, re message.CommitTime = *currentTxnCommitTime lsn := msgLSN.String() message.LSN = &lsn - select { - case s.messages <- []StreamMessage{*message}: - return changeResultEmittedMessage, nil - case <-ctx.Done(): - return changeResultNoMessage, ctx.Err() - } + return changeResultEmittedMessage, message, nil } -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,10 +640,10 @@ 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) + primaryKeyColumns, err := s.getPrimaryKeyColumn(ctx, snapshotter, table) if err != nil { return fmt.Errorf("getting primary key column for table %v: %w", table, err) } @@ -808,11 +832,174 @@ func (s *Stream) Messages() chan []StreamMessage { return s.messages } -// MarkSnapshotAcknowledged is called by the input layer once every snapshot -// message has been acknowledged downstream, unblocking promotion of the -// replication slot. Safe to call multiple times. +// beginSnapshotAck creates and registers the completion signal for a new +// snapshot round, returning it for the caller to wait on. Must be called +// before that round's completion sentinel is sent, so MarkSnapshotAcknowledged +// has something to close once the input layer confirms delivery. +func (s *Stream) beginSnapshotAck() <-chan struct{} { + s.snapshotAckMu.Lock() + defer s.snapshotAckMu.Unlock() + ch := make(chan struct{}) + s.snapshotAcked = ch + return ch +} + +// MarkSnapshotAcknowledged is called by the input layer once every message +// from the most recently started snapshot round has been acknowledged +// downstream, unblocking whatever is waiting on that round's completion. +// Safe to call even when no round is currently pending. func (s *Stream) MarkSnapshotAcknowledged() { - s.snapshotAckedOnce.Do(func() { close(s.snapshotAcked) }) + s.snapshotAckMu.Lock() + ch := s.snapshotAcked + s.snapshotAcked = nil + s.snapshotAckMu.Unlock() + if ch != nil { + close(ch) + } +} + +// walPause guards whether WAL forwarding is paused and the satellite +// keepalive goroutine running for the duration - see PauseWALStreaming. +type walPause struct { + sync.Mutex + ch chan struct{} + cancel context.CancelFunc + done chan struct{} +} + +// begin records a new pause's cancel (stops the satellite goroutine) and +// done (closed once it exits). +func (p *walPause) begin(cancel context.CancelFunc, done chan struct{}) { + p.Lock() + defer p.Unlock() + p.ch = make(chan struct{}) + p.cancel = cancel + p.done = done +} + +// end clears the pause and returns what begin recorded. +func (p *walPause) end() (ch chan struct{}, cancel context.CancelFunc, done chan struct{}) { + p.Lock() + defer p.Unlock() + ch, cancel, done = p.ch, p.cancel, p.done + p.ch, p.cancel, p.done = nil, nil, nil + return +} + +// waitChan returns the channel to wait on if currently paused, or nil if not. +func (p *walPause) waitChan() chan struct{} { + p.Lock() + defer p.Unlock() + return p.ch +} + +// PauseStreaming holds back forwarding of new WAL messages until +// ResumeWALStreaming is called: streamMessages stops receiving from Postgres +// entirely. A satellite goroutine sends standby status updates in the +// meantime, so Postgres doesn't hit wal_sender_timeout. Call synchronously +// before handing a signal off to run its re-snapshot. +func (s *Stream) PauseStreaming() { + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + s.walPause.begin(cancel, done) + go func() { + defer close(done) + ticker := time.NewTicker(s.standbyMessageTimeout) + defer ticker.Stop() + for { + select { + case <-ticker.C: + if err := s.commitAckedLSN(ctx, s.getAckedLSN()); err != nil { + s.logger.Warnf("failed to send keepalive while WAL forwarding is paused: %s", err) + } + case <-ctx.Done(): + return + } + } + }() +} + +// ResumeStreaming releases a pause started by PauseWALStreaming, waiting +// for the satellite keepalive goroutine to exit first so it never writes to +// pgConn concurrently with streamMessages resuming. +func (s *Stream) ResumeStreaming() { + ch, cancel, done := s.walPause.end() + if cancel != nil { + cancel() + } + if done != nil { + <-done + } + if ch != nil { + close(ch) + } +} + +// waitIfWALPaused blocks until any pause started by PauseWALStreaming +// resumes, or ctx is done. Returns immediately if nothing is paused. +func (s *Stream) waitIfWALPaused(ctx context.Context) error { + ch := s.walPause.waitChan() + if ch == nil { + return nil + } + select { + case <-ch: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// RunSnapshot re-scans the given tables (by bare name, resolved against +// this stream's configured schema) against current DB state on a separate +// connection, without restarting replication. Rows are emitted on Messages() +// like ordinary WAL changes, terminated by the SnapshotCompleteOpType +// sentinel. +// +// Callers must call PauseWALStreaming themselves before handing off to this +// method (see its doc for why) - this method resumes it via defer. +// +// Callers must keep draining Messages() from a different goroutine while +// this call is in flight, or it deadlocks: scanned rows and the completion +// sentinel are sent on that same unbuffered channel. +// +// Callers must serialize calls to this method: only one round's completion +// can be tracked at a time (see snapshotAcked). +func (s *Stream) RunSnapshot(ctx context.Context, tableNames []string) error { + defer s.ResumeStreaming() + + tables := make([]TableFQN, 0, len(tableNames)) + for _, table := range tableNames { + normalized, err := sanitize.NormalizePostgresIdentifier(table) + if err != nil { + return fmt.Errorf("invalid snapshot table name %q: %w", table, err) + } + tables = append(tables, TableFQN{Schema: s.schema, Table: normalized}) + } + + snapshotter, err := newSnapshotter(s.config, s.config.DBRawDSN, s.logger, "", s.config.MaxSnapshotWorkers) + if err != nil { + return fmt.Errorf("creating snapshotter for re-snapshot: %w", err) + } + if err := s.processSnapshot(ctx, tables, snapshotter); err != nil { + return fmt.Errorf("processing snapshot: %w", err) + } + for _, table := range tables { + s.monitor.MarkSnapshotComplete(table) + } + + acked := s.beginSnapshotAck() + select { + case s.messages <- []StreamMessage{{Operation: SnapshotCompleteOpType}}: + case <-ctx.Done(): + return ctx.Err() + } + select { + case <-acked: + case <-ctx.Done(): + return ctx.Err() + } + return nil } // Errors is a channel that can be used to see if and error has occurred internally and the stream should be restarted. @@ -820,7 +1007,10 @@ func (s *Stream) Errors() chan error { return s.errors } -func (s *Stream) getPrimaryKeyColumn(ctx context.Context, table TableFQN) ([]string, error) { +// getPrimaryKeyColumn queries through snapshotter's own connection pool, not +// s.pgConn: RunForcedSnapshot runs alongside an active streamMessages loop +// that also uses s.pgConn, so this can't safely share it. +func (*Stream) getPrimaryKeyColumn(ctx context.Context, snapshotter *snapshotter, table TableFQN) ([]string, error) { /// Query to get all primary key columns in their correct order q, err := sanitize.SQLQuery(` SELECT a.attname @@ -835,21 +1025,26 @@ func (s *Stream) getPrimaryKeyColumn(ctx context.Context, table TableFQN) ([]str return nil, fmt.Errorf("sanitizing query: %w", err) } - reader := s.pgConn.Exec(ctx, q) - data, err := reader.ReadAll() + rows, err := snapshotter.connPool.QueryContext(ctx, q) if err != nil { return nil, fmt.Errorf("reading query results: %w", err) } + defer rows.Close() - if len(data) == 0 || len(data[0].Rows) == 0 { - return nil, fmt.Errorf("no primary key found for table %s", table) - } - - // Extract all primary key column names - pkColumns := make([]string, len(data[0].Rows)) - for i, row := range data[0].Rows { + var pkColumns []string + for rows.Next() { + var col string + if err := rows.Scan(&col); err != nil { + return nil, fmt.Errorf("scanning primary key column: %w", err) + } // Postgres gives us back normalized identifiers here - we need to quote them. - pkColumns[i] = sanitize.QuotePostgresIdentifier(string(row[0])) + pkColumns = append(pkColumns, sanitize.QuotePostgresIdentifier(col)) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("reading query results: %w", err) + } + if len(pkColumns) == 0 { + return nil, fmt.Errorf("no primary key found for table %s", table) } return pkColumns, nil 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/replication/signalling.go b/internal/replication/signalling.go new file mode 100644 index 0000000000..f3d62d6989 --- /dev/null +++ b/internal/replication/signalling.go @@ -0,0 +1,66 @@ +// 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" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +// ControlSignal represents a insert into the signal table. +type ControlSignal struct { + ID string + Type string + DataCollections []string `json:"data-collections"` + + // LSN is the position the signal was observed at, in whatever raw form + // the connector represents positions. Populated by Listen, not part of + // the signal's own 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 == "execute-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 so the caller can flush and hold back +// acking exactly that message before running the signal's action. 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. +// +// ControlSignaller does not implement Listen itself, since it has no way to +// know the shape of a connector's replication messages - connectors must +// embed it and provide their own Listen (see +// internal/impl/postgresql/signaller.go for an example). +type Signaller interface { + Listen(ctx context.Context, event any) (*ControlSignal, error) +} + +// ControlSignaller can be used to handle and process control signals. +// Currently just a thin holder for the logger; embed it in a +// connector-specific type that implements Listen to obtain a full Signaller. +type ControlSignaller struct { + Log *service.Logger +} + +// NewControlSignaller creates a ControlSignaller. 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, + } +} From d8246697207c92842f49637e405eb1474f5f49a0 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Fri, 24 Jul 2026 14:35:47 +0100 Subject: [PATCH 2/2] clean up comments --- internal/impl/postgresql/input_pg_stream.go | 5 +++-- .../postgresql/pglogicalstream/logical_stream.go | 16 ++++++++-------- .../postgresql/pglogicalstream/stream_message.go | 2 ++ internal/replication/signalling.go | 2 +- 4 files changed, 14 insertions(+), 11 deletions(-) diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index cbb1c5f788..b968ec5b91 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -459,7 +459,7 @@ type pgStreamInput struct { // signalQueue hands a detected signal off to runSignalWorker. Capacity 1 // is enough: processStream pauses WAL forwarding before enqueueing (see - // PauseWALStreaming), so it can't detect another signal until the + // PauseStreaming), so it can't detect another signal until the // current one's re-snapshot completes and resumes it. signalQueue chan *replication.ControlSignal @@ -613,6 +613,7 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher } if sig != nil { + batchMsg.MetaSet("operation", string(pglogicalstream.ControlSignalOpType)) p.handleControlSignal(ctx, pgStream, cp, batcher, sig, batchMsg) continue } @@ -792,7 +793,7 @@ func (p *pgStreamInput) handleControlSignal( } // runSignalWorker runs each detected signal's re-snapshot on pgStream's -// live connection. Must run on its own goroutine: RunForcedSnapshot sends +// live connection. Must run on its own goroutine: RunSnapshot sends // scanned rows and its completion sentinel on the same channel processStream // drains, so processStream must stay free to keep receiving from it. func (p *pgStreamInput) runSignalWorker(pgStream *pglogicalstream.Stream) { diff --git a/internal/impl/postgresql/pglogicalstream/logical_stream.go b/internal/impl/postgresql/pglogicalstream/logical_stream.go index 104f8accdf..15e449277f 100644 --- a/internal/impl/postgresql/pglogicalstream/logical_stream.go +++ b/internal/impl/postgresql/pglogicalstream/logical_stream.go @@ -51,12 +51,12 @@ type Stream struct { // snapshotAcked is the completion signal for the snapshot round in // flight, if any. beginSnapshotAck creates it before that round's // sentinel is sent; MarkSnapshotAcknowledged closes it once acked - // downstream. A fresh channel per round lets RunForcedSnapshot run + // downstream. A fresh channel per round lets RunSnapshot run // repeatedly over the Stream's life. snapshotAckMu sync.Mutex snapshotAcked chan struct{} - // walPause tracks WAL forwarding pause state - see PauseWALStreaming. + // walPause tracks WAL forwarding pause state - see PauseStreaming. walPause walPause includeTxnMarkers bool @@ -859,7 +859,7 @@ func (s *Stream) MarkSnapshotAcknowledged() { } // walPause guards whether WAL forwarding is paused and the satellite -// keepalive goroutine running for the duration - see PauseWALStreaming. +// keepalive goroutine running for the duration - see PauseStreaming. type walPause struct { sync.Mutex ch chan struct{} @@ -894,7 +894,7 @@ func (p *walPause) waitChan() chan struct{} { } // PauseStreaming holds back forwarding of new WAL messages until -// ResumeWALStreaming is called: streamMessages stops receiving from Postgres +// ResumeStreaming is called: streamMessages stops receiving from Postgres // entirely. A satellite goroutine sends standby status updates in the // meantime, so Postgres doesn't hit wal_sender_timeout. Call synchronously // before handing a signal off to run its re-snapshot. @@ -919,7 +919,7 @@ func (s *Stream) PauseStreaming() { }() } -// ResumeStreaming releases a pause started by PauseWALStreaming, waiting +// ResumeStreaming releases a pause started by PauseStreaming, waiting // for the satellite keepalive goroutine to exit first so it never writes to // pgConn concurrently with streamMessages resuming. func (s *Stream) ResumeStreaming() { @@ -935,7 +935,7 @@ func (s *Stream) ResumeStreaming() { } } -// waitIfWALPaused blocks until any pause started by PauseWALStreaming +// waitIfWALPaused blocks until any pause started by PauseStreaming // resumes, or ctx is done. Returns immediately if nothing is paused. func (s *Stream) waitIfWALPaused(ctx context.Context) error { ch := s.walPause.waitChan() @@ -956,7 +956,7 @@ func (s *Stream) waitIfWALPaused(ctx context.Context) error { // like ordinary WAL changes, terminated by the SnapshotCompleteOpType // sentinel. // -// Callers must call PauseWALStreaming themselves before handing off to this +// Callers must call PauseStreaming themselves before handing off to this // method (see its doc for why) - this method resumes it via defer. // // Callers must keep draining Messages() from a different goroutine while @@ -1008,7 +1008,7 @@ func (s *Stream) Errors() chan error { } // getPrimaryKeyColumn queries through snapshotter's own connection pool, not -// s.pgConn: RunForcedSnapshot runs alongside an active streamMessages loop +// s.pgConn: RunSnapshot runs alongside an active streamMessages loop // that also uses s.pgConn, so this can't safely share it. func (*Stream) getPrimaryKeyColumn(ctx context.Context, snapshotter *snapshotter, table TableFQN) ([]string, error) { /// Query to get all primary key columns in their correct order diff --git a/internal/impl/postgresql/pglogicalstream/stream_message.go b/internal/impl/postgresql/pglogicalstream/stream_message.go index 6ec126f9fd..eb49fe1b4b 100644 --- a/internal/impl/postgresql/pglogicalstream/stream_message.go +++ b/internal/impl/postgresql/pglogicalstream/stream_message.go @@ -41,6 +41,8 @@ const ( // replication slot until every snapshot message has been acknowledged // downstream; it is never forwarded downstream. SnapshotCompleteOpType OpType = "snapshot_complete" + // ControlSignalOpType indicates the event is a control signal from the configured signalling table. + ControlSignalOpType OpType = "control_signal" ) // StreamMessage represents a single change from the database diff --git a/internal/replication/signalling.go b/internal/replication/signalling.go index f3d62d6989..fd0093c651 100644 --- a/internal/replication/signalling.go +++ b/internal/replication/signalling.go @@ -45,7 +45,7 @@ func (s *ControlSignal) IsSnapshot() bool { // ControlSignaller does not implement Listen itself, since it has no way to // know the shape of a connector's replication messages - connectors must // embed it and provide their own Listen (see -// internal/impl/postgresql/signaller.go for an example). +// internal/impl/postgresql/control_signal.go for an example). type Signaller interface { Listen(ctx context.Context, event any) (*ControlSignal, error) }