postgres_cdc: add signalling support by josephwoodward · Pull Request #4576 · redpanda-data/connect

josephwoodward · 2026-07-02T00:25:28Z

This change features a new signal channel (backed by PostgreSQL) that allows you to signal the connector to perform a snapshot on an adhoc basis.

This process will:

Pauses streaming and performs snapshotting of the table(s) specified in the signal payload (it can be all tables, or just new tables).
Logs generated by the signal and subsequent snapshot will feature a structured logging column of the ID specified in the signal event raised by the user.
Once snapshotting is completed the connector resume streaming from its original cached checkpoint.

claude · 2026-07-02T00:31:05Z

+					if msg.LSN != nil {
+						if resolveFn, err := cp.Track(ctx, msg.LSN, 0); err == nil {
+							if maxLSN := resolveFn(); maxLSN != nil && *maxLSN != nil {
+								if err := pgStream.AckLSN(ctx, **maxLSN); err != nil {
+									p.logger.Warnf("failed to ack signal LSN: %s", err)
+								}
+							}
+						}
+					}


This inline ack can advance Postgres's confirmed flush LSN past data that hasn't been delivered downstream yet, breaking at-least-once delivery (CONTRIBUTING §5.4.2).

Data messages processed earlier in the loop are only added to the batcher (batcher.Add) and are not tracked in the checkpointer until flushBatch runs after the loop. Here the signal is tracked with cp.Track(ctx, msg.LSN, 0) and immediately resolved via resolveFn(). When there are no other in-flight (tracked-but-unresolved) batches, resolveFn() returns the signal's LSN and AckLSN confirms it to Postgres — even though lower-LSN rows are still sitting unflushed in the batcher (and rows for tables not in data-collections won't be recovered by the re-snapshot).

Failure scenario: batcher holds data1, data2 (batch policy not yet triggered, all prior flushed batches already acked), then a signal row is processed. AckLSN(signalLSN) advances confirmed_flush_lsn past data1/data2. A crash before those are delivered+acked downstream loses them permanently, since Postgres won't resend anything <= signalLSN.

The signal LSN should be gated behind the same downstream-ack path as the buffered data (e.g. flush the batcher first so buffered rows are tracked ahead of the signal), rather than acked immediately.

claude · 2026-07-02T00:31:13Z

+	if pending, signal := p.controlSig.IsPending(); pending && signal.IsSnapshot() {
+		p.logger.Infof("%q signal pending, triggering re-snapshots", signal.Type)
+
+		p.streamConfig.StreamOldData = true
+		p.streamConfig.ForceSnapshot = true
+		defer func() { p.streamConfig.ForceSnapshot = false }()
+
+		p.streamConfig.SnapshotTables = signal.TableNames(p.streamConfig.DBSchema)
+		defer func() { p.streamConfig.SnapshotTables = nil }()
+
+		p.controlSig.Reset()


p.controlSig.Reset() clears the pending signal before the stream that consumes it is successfully created. If NewPgStream (line 479) fails — a routine, expected condition for a network-backed CDC connector on reconnect — Connect returns an error and the framework retries, but IsPending() is now false, so the else-branch runs and no re-snapshot is performed.

If the signal row's LSN was already acked in the prior stream (so streaming resumes past it and the row is never re-read), the user's execute-snapshot request is silently dropped with no warning logged. Reset() should only be called once the stream that will act on the signal has been successfully constructed.

claude · 2026-07-02T00:31:25Z

Commits
LGTM — single commit postgres_cdc: add signalling support follows the system: message format (valid system, lowercase, imperative mood) and its message accurately describes the change. Generated docs live alongside the code change as expected for a single-commit PR.

Review
The PR adds execute-snapshot signal support to postgres_cdc (a signal table whose INSERTs trigger a re-snapshot without dropping the replication slot). The overall shape is consistent with the existing connector, the signal table is correctly excluded from snapshot output while included in the publication, the re-snapshot uses a REPEATABLE READ transaction, and the monitor.go nil guards and empty-snapshotName handling are sound. Two correctness concerns around delivery guarantees:

Premature signal-LSN ack can lose buffered data (at-least-once violation). In the signal branch of processStream, the signal LSN is tracked and immediately resolved/acked, which can advance Postgres's confirmed flush LSN past lower-LSN rows still buffered in the batcher (not yet flushed/tracked). A crash before those rows are delivered downstream loses them. See inline comment on input_pg_stream.go L549-L557. (CONTRIBUTING §5.4.2)
Signal reset before stream creation. controlSig.Reset() is called before NewPgStream succeeds; a NewPgStream failure on the re-snapshot reconnect silently drops the pending signal. See inline comment on input_pg_stream.go L464-L474.

claude · 2026-07-05T05:55:46Z

      - task: cdb:setup
      - cmd: echo "Inserting test data, please hold..."
-      - task: sqlcl:pdb:data:users
+      # - task: sqlcl:pdb:data:users


This comments out the test-data insertion step in the Oracle benchmark setup. Two problems:

It is unrelated to this PR's postgres_cdc signalling feature — changes should stay within the scope agreed for the PR (CONTRIBUTING.md §3.1.1).

Disabling the data load means the Oracle benchmark no longer inserts any rows.

This looks like an accidental local-debugging leftover (correlates with the temp commit) — please revert.

claude · 2026-07-05T05:55:50Z

    n % 100,
    ((n % 1000) + (n % 100) / 100.0)::decimal(10,2)
-FROM generate_series(1, 150000) AS n;
+FROM generate_series(1, 1) AS n;


The benchmark seed is reduced from generate_series(1, 150000) to generate_series(1, 1) here (and products.sql from 150000 to 100). This shrinks the dataset to a trivial size, which defeats throughput benchmarking — certification requires benchmarks run across throughput levels to establish CPU/memory trendlines (CONTRIBUTING.md §1.3.4–§1.3.5).

This appears to be an accidental local-debug leftover — please restore the original row counts.

claude · 2026-07-05T05:55:53Z

+  # processors:
+  #   - benchmark:
+  #       interval: 1s
+  #       count_bytes: true
+  file:
+    path: "./benchmark_results.json"
+    codec: lines
+  # drop: {}

 logger:
-  level: INFO
+  level: DEBUG


These edits look like committed local-debugging state rather than an intentional benchmark config: the benchmark processor + drop output are commented out and replaced with a file sink, the batching count: 1000 was removed, and the logger is switched to DEBUG. Writing every record to a file and emitting debug logs will skew the benchmark measurements this harness exists to produce (CONTRIBUTING.md §1.3.4).

Please restore the benchmarking output (benchmark processor / drop) and INFO logging.

claude · 2026-07-05T05:56:05Z

Commits

Commit temp (4e8d038) — non-descriptive WIP-style message. It fails the message-quality bar (vague/incomprehensible) and looks like an unsquashed work-in-progress commit that should be squashed into the feature commit.
Commit benchmark tweaks (2fbf30e) — message is vague and not in imperative mood, and lacks a system: scope. Expected form is e.g. bench: reduce postgres benchmark seed data (imperative, scoped). It also mixes an unrelated Oracle benchmark change with the Postgres benchmark changes.

The first commit postgres_cdc: add signalling support follows the policy and is well-formed.

Review

The core signalling implementation (signaller.go, internal/replication/signalling.go, the ForceSnapshot/re-snapshot path, and the monitor nil guards) is carefully written, well-commented, and covered by unit + integration tests. The issues found are all accidentally-committed local-debug leftovers in the benchmark harness (correlating with the temp commit):

internal/impl/oracledb/bench/Taskfile.yaml — comments out the Oracle test-data insertion step. Unrelated to this PR (scope, §3.1.1) and disables the benchmark data load.
internal/impl/postgresql/bench/users.sql / products.sql — benchmark seed reduced from 150000 rows to 1 / 100, defeating throughput benchmarking (§1.3.4–§1.3.5).
internal/impl/postgresql/bench/benchmark_config.yaml — benchmark output/processor replaced with a file sink, batching count removed, and logging switched to DEBUG; these skew benchmark measurements.

postgres_cdc: add signalling support

10c5cc2

github-actions Bot added the run-integration-tests label Jul 2, 2026

claude Bot reviewed Jul 2, 2026

View reviewed changes

josephwoodward added 2 commits July 5, 2026 06:50

temp

4e8d038

benchmark tweaks

2fbf30e

claude Bot reviewed Jul 5, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

postgres_cdc: add signalling support#4576

postgres_cdc: add signalling support#4576
josephwoodward wants to merge 3 commits into
mainfrom
jw/postgrescdc_add_signal_support

josephwoodward commented Jul 2, 2026

Uh oh!

claude Bot Jul 2, 2026

Uh oh!

claude Bot Jul 2, 2026

Uh oh!

claude Bot commented Jul 2, 2026

Uh oh!

claude Bot Jul 5, 2026

Uh oh!

claude Bot Jul 5, 2026

Uh oh!

claude Bot Jul 5, 2026

Uh oh!

claude Bot commented Jul 5, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

josephwoodward commented Jul 2, 2026

Uh oh!

claude Bot Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

claude Bot Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

claude Bot commented Jul 2, 2026

Uh oh!

claude Bot Jul 5, 2026

Choose a reason for hiding this comment

Uh oh!

claude Bot Jul 5, 2026

Choose a reason for hiding this comment

Uh oh!

claude Bot Jul 5, 2026

Choose a reason for hiding this comment

Uh oh!

claude Bot commented Jul 5, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant