From 5ed0aff8c3d44547004d740acb2917f20658c39c Mon Sep 17 00:00:00 2001 From: Chris Munns Date: Mon, 22 Jun 2026 10:16:42 -0400 Subject: [PATCH 1/2] Handle endpos pipeline shutdown cleanly and reset sequences on follow In follow mode the receive, transform, and apply processes are connected by Unix pipes. When the apply process reaches endpos it exits and closes its read end of the pipe. Upstream processes that are still writing trailing messages then hit EPIPE and exit non-zero, and the supervisor ANDs every child's status, so a migration that completed correctly at endpos was reported as a failure. This adds two layers of handling: - Child side: transform and receive treat an EPIPE on the downstream pipe as a clean shutdown only when endpos has been durably reached for the last message they processed. In every other case (endpos unset, or not yet reached) a broken pipe is still a failure. - Supervisor backstop: follow_wait_subprocesses declares overall success when the apply process exited cleanly and endpos has been durably applied (endpos <= replay_lsn), regardless of upstream teardown noise. The apply process is authoritative, so this cannot mask a genuine pre-endpos failure: if apply crashed or endpos was not reached, the failure still propagates. The false failure also skipped the end-of-migration sequence reset. follow_reset_sequences is now also run by the standalone "pgcopydb follow" command once endpos is durably reached, so a resumed CDC run that catches up to endpos updates target sequences to current source values. Previously only "clone --follow" reset sequences, leaving resume-after-crash with stale sequences. --- src/bin/pgcopydb/cli_clone_follow.c | 32 +++++++++++++++++++ src/bin/pgcopydb/follow.c | 48 +++++++++++++++++++++++++++++ src/bin/pgcopydb/ld_stream.c | 31 ++++++++++++++++--- src/bin/pgcopydb/ld_transform.c | 44 ++++++++++++++++++++++++-- 4 files changed, 148 insertions(+), 7 deletions(-) diff --git a/src/bin/pgcopydb/cli_clone_follow.c b/src/bin/pgcopydb/cli_clone_follow.c index 5d904daad..0fb5e0ef6 100644 --- a/src/bin/pgcopydb/cli_clone_follow.c +++ b/src/bin/pgcopydb/cli_clone_follow.c @@ -643,6 +643,38 @@ cli_follow(int argc, char **argv) exit(EXIT_CODE_INTERNAL_ERROR); } + /* + * When CDC has durably reached endpos (cutover), reset the sequences on the + * target database to their current values on the source. Postgres logical + * decoding does not replicate sequences, so without this final step the + * target sequences are left at the values captured during the initial base + * copy. This mirrors what "clone --follow" does at the end of its run, and + * makes a resumed follow that catches up to endpos safe to cut over from. + * + * We only do this once endpos is reached: an interrupted continuous follow + * (no endpos, or stopped early by a signal) must not advance sequences ahead + * of the data that was actually applied to the target. + */ + bool reachedEndpos = false; + + if (!follow_reached_endpos(&specs, &reachedEndpos)) + { + /* errors have already been logged */ + exit(EXIT_CODE_INTERNAL_ERROR); + } + + if (reachedEndpos) + { + log_info("Resetting sequences on the target database to match the " + "current values on the source database"); + + if (!follow_reset_sequences(©Specs, &specs)) + { + /* errors have already been logged */ + exit(EXIT_CODE_TARGET); + } + } + /* * CDC has ended (endpos reached / cutover). If FKs were created NOT VALID * via --defer-validate-fks, remind the operator at this final, visible diff --git a/src/bin/pgcopydb/follow.c b/src/bin/pgcopydb/follow.c index 7e47d824a..eb9aa17bb 100644 --- a/src/bin/pgcopydb/follow.c +++ b/src/bin/pgcopydb/follow.c @@ -1122,6 +1122,54 @@ follow_wait_subprocesses(StreamSpecs *specs) pg_usleep(150 * 1000); } + /* + * Data-safe backstop for the endpos shutdown race. + * + * The apply (catchup) process is authoritative: it exits successfully only + * once it has durably applied changes through endpos, syncing replay_lsn in + * the sentinel. When that has happened, the migration's CDC goal is met and + * the data is on the target through endpos. Any non-zero exit from the + * upstream prefetch/transform processes at that point is teardown noise + * (typically EPIPE as the apply process closes the pipe) and must not turn a + * completed migration into a reported failure. + * + * This is strictly gated: we only override when the apply process itself + * exited successfully AND endpos has been reached (endpos <= replay_lsn). If + * apply crashed, or endpos was not reached, we leave success as-is so the + * failure propagates and the operator can resume. + */ + if (!success) + { + FollowSubProcess *catchup = &(specs->catchup); + + bool applyExitedOk = + catchup->pid > 0 && + catchup->exited && + catchup->returnCode == 0 && + signal_is_handled(catchup->sig); + + if (applyExitedOk) + { + /* refresh sentinel so replay_lsn reflects the final apply state */ + if (!follow_get_sentinel(specs, &(specs->sentinel), false)) + { + log_warn("Failed to get sentinel values"); + } + + if (specs->sentinel.endpos != InvalidXLogRecPtr && + specs->sentinel.endpos <= specs->sentinel.replay_lsn) + { + log_info("Apply process durably reached endpos %X/%X " + "(replay_lsn %X/%X); treating upstream pipeline " + "teardown as a clean shutdown", + LSN_FORMAT_ARGS(specs->sentinel.endpos), + LSN_FORMAT_ARGS(specs->sentinel.replay_lsn)); + + success = true; + } + } + } + return success; } diff --git a/src/bin/pgcopydb/ld_stream.c b/src/bin/pgcopydb/ld_stream.c index d173a2a30..c9ce934bf 100644 --- a/src/bin/pgcopydb/ld_stream.c +++ b/src/bin/pgcopydb/ld_stream.c @@ -545,11 +545,34 @@ startLogicalStreaming(StreamSpecs *specs) } else if (privateContext->pipelineBroken) { - log_error("Downstream pipeline process has exited, " - "stopping at %X/%X", - LSN_FORMAT_ARGS(context.tracking->written_lsn)); + /* + * The downstream pipeline closed its read end. When endpos has + * been reached (we have streamed at or past it), that's the + * expected clean shutdown: the transform/apply processes are + * done on purpose. Otherwise the downstream died unexpectedly + * and we must report failure. Either way, a genuine downstream + * failure is also caught independently by the apply process's + * own non-zero exit at the follow supervisor. + */ + if (context.endpos != InvalidXLogRecPtr && + context.tracking != NULL && + context.endpos <= context.tracking->written_lsn) + { + log_info("Downstream pipeline reached endpos %X/%X and " + "closed the pipe; receive stopping cleanly at %X/%X", + LSN_FORMAT_ARGS(context.endpos), + LSN_FORMAT_ARGS(context.tracking->written_lsn)); - return false; + retry = false; + } + else + { + log_error("Downstream pipeline process has exited, " + "stopping at %X/%X", + LSN_FORMAT_ARGS(context.tracking->written_lsn)); + + return false; + } } else if (!retry) { diff --git a/src/bin/pgcopydb/ld_transform.c b/src/bin/pgcopydb/ld_transform.c index fbebdfdbd..8bf0cebb9 100644 --- a/src/bin/pgcopydb/ld_transform.c +++ b/src/bin/pgcopydb/ld_transform.c @@ -198,9 +198,35 @@ stream_transform_stream_internal(StreamSpecs *specs) if (!read_from_stream(privateContext->in, &context)) { - log_error("Failed to transform JSON messages from input stream, " - "see above for details"); - return false; + /* + * When the apply process reaches endpos it exits and closes the read + * end of our output pipe. If we were still writing trailing messages + * we get EPIPE (pipelineBroken). That is an expected, clean shutdown + * when endpos has been reached for the last message we processed: the + * downstream is done on purpose, not crashed. + * + * We only treat EPIPE as clean when endpos is set AND the last message + * we handled is at or past it. In every other case (endpos unset, or + * not yet reached) a broken pipe means the downstream died + * unexpectedly and we must report failure. A genuine downstream crash + * is also caught independently by the apply process's own non-zero + * exit at the follow supervisor. + */ + if (privateContext->pipelineBroken && + privateContext->endpos != InvalidXLogRecPtr && + privateContext->endpos <= privateContext->metadata.lsn) + { + log_info("Apply process reached endpos %X/%X and closed the pipe; " + "transform stopping cleanly at %X/%X", + LSN_FORMAT_ARGS(privateContext->endpos), + LSN_FORMAT_ARGS(privateContext->metadata.lsn)); + } + else + { + log_error("Failed to transform JSON messages from input stream, " + "see above for details"); + return false; + } } /* we might have stopped reading mid-file, let's close it. */ @@ -503,6 +529,18 @@ stream_transform_write_message(StreamContext *privateContext, { if (!stream_write_message(privateContext->out, currentMsg)) { + /* + * A broken downstream pipe (EPIPE) is expected when the apply + * process has reached endpos and exited: it closes its read end + * while we might still be writing trailing messages. Record that + * so the caller can decide whether this is a clean endpos + * shutdown or a genuine failure. See stream_transform_stream(). + */ + if (errno == EPIPE) + { + privateContext->pipelineBroken = true; + } + /* errors have already been logged */ return false; } From 78980aae6bf2ebc575740609d875e08a93d0b3ab Mon Sep 17 00:00:00 2001 From: Chris Munns Date: Mon, 22 Jun 2026 17:06:35 -0400 Subject: [PATCH 2/2] Add follow-sequence-reset regression test Adds a deterministic regression test for the sequence reset performed by the standalone `pgcopydb follow` command when it reaches endpos (the path used by resume-cdc helpers, which previously did not reset sequences). The test clones pagila, advances rental_rental_id_seq on the source by inserting rows, sets endpos, and runs `pgcopydb follow --resume`. Because CDC replays the inserts with OVERRIDING SYSTEM VALUE (explicit ids that do not advance the target sequence), the target sequence only catches up to the source if follow_reset_sequences runs at endpos. The test asserts the target sequence advanced from the snapshot value to match the source. Verified the test fails (target stuck at the snapshot value) when the reset is removed, and passes with it in place. --- .github/workflows/run-tests-tiered.yml | 1 + tests/Makefile | 7 +- tests/follow-sequence-reset/Dockerfile | 8 ++ tests/follow-sequence-reset/Dockerfile.pg | 9 +++ tests/follow-sequence-reset/Makefile | 15 ++++ tests/follow-sequence-reset/compose.yaml | 42 ++++++++++ tests/follow-sequence-reset/copydb.sh | 96 +++++++++++++++++++++++ tests/follow-sequence-reset/dml.sql | 11 +++ 8 files changed, 187 insertions(+), 2 deletions(-) create mode 100644 tests/follow-sequence-reset/Dockerfile create mode 100644 tests/follow-sequence-reset/Dockerfile.pg create mode 100644 tests/follow-sequence-reset/Makefile create mode 100644 tests/follow-sequence-reset/compose.yaml create mode 100755 tests/follow-sequence-reset/copydb.sh create mode 100644 tests/follow-sequence-reset/dml.sql diff --git a/.github/workflows/run-tests-tiered.yml b/.github/workflows/run-tests-tiered.yml index fccb834b6..932253975 100644 --- a/.github/workflows/run-tests-tiered.yml +++ b/.github/workflows/run-tests-tiered.yml @@ -159,6 +159,7 @@ jobs: - fk-not-valid - defer-validate-fks - follow-defer-validate-fks + - follow-sequence-reset steps: - name: Checkout repository uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 diff --git a/tests/Makefile b/tests/Makefile index f4147c0f5..06983c7d4 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -15,7 +15,7 @@ all: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions follow-wal2json follow-standby follow-9.6 follow-data-only follow-target-reconnect \ endpos-in-multi-wal-txn exclude-extension \ blob-snapshot-release follow-defer-indexes fk-not-valid defer-validate-fks \ - follow-defer-validate-fks; + follow-defer-validate-fks follow-sequence-reset; pagila: build $(MAKE) -C $@ @@ -89,6 +89,9 @@ defer-validate-fks: build follow-defer-validate-fks: build $(MAKE) -C $@ +follow-sequence-reset: build + $(MAKE) -C $@ + timescaledb: build $(MAKE) -C $@ @@ -102,4 +105,4 @@ build: .PHONY: follow-wal2json follow-standby follow-9.6 follow-target-reconnect .PHONY: endpos-in-multi-wal-txn exclude-extension .PHONY: blob-snapshot-release follow-defer-indexes fk-not-valid defer-validate-fks -.PHONY: follow-defer-validate-fks +.PHONY: follow-defer-validate-fks follow-sequence-reset diff --git a/tests/follow-sequence-reset/Dockerfile b/tests/follow-sequence-reset/Dockerfile new file mode 100644 index 000000000..b3e1a80c6 --- /dev/null +++ b/tests/follow-sequence-reset/Dockerfile @@ -0,0 +1,8 @@ +FROM pagila + +WORKDIR /usr/src/pgcopydb +COPY ./copydb.sh copydb.sh +COPY ./dml.sql dml.sql + +USER docker +CMD ["/usr/src/pgcopydb/copydb.sh"] diff --git a/tests/follow-sequence-reset/Dockerfile.pg b/tests/follow-sequence-reset/Dockerfile.pg new file mode 100644 index 000000000..52bcaa3a2 --- /dev/null +++ b/tests/follow-sequence-reset/Dockerfile.pg @@ -0,0 +1,9 @@ +ARG PGVERSION=16 +FROM postgres:${PGVERSION} + +ARG PGVERSION=16 +USER root +RUN apt-get update \ + && apt-get install -y --no-install-recommends postgresql-${PGVERSION}-wal2json \ + && rm -rf /var/lib/apt/lists/* +USER postgres diff --git a/tests/follow-sequence-reset/Makefile b/tests/follow-sequence-reset/Makefile new file mode 100644 index 000000000..bc07ec272 --- /dev/null +++ b/tests/follow-sequence-reset/Makefile @@ -0,0 +1,15 @@ +# Copyright (c) 2021 The PostgreSQL Global Development Group. +# Licensed under the PostgreSQL License. + +test: down run down ; + +run: build + $(DOCKER) compose run test + +down: + $(DOCKER) compose down + +build: + $(DOCKER) compose build + +.PHONY: run down build test diff --git a/tests/follow-sequence-reset/compose.yaml b/tests/follow-sequence-reset/compose.yaml new file mode 100644 index 000000000..8d949535e --- /dev/null +++ b/tests/follow-sequence-reset/compose.yaml @@ -0,0 +1,42 @@ +services: + source: + build: + context: . + dockerfile: Dockerfile.pg + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + command: > + -c wal_level=logical + -c ssl=on + -c ssl_cert_file=/etc/ssl/certs/ssl-cert-snakeoil.pem + -c ssl_key_file=/etc/ssl/private/ssl-cert-snakeoil.key + target: + image: postgres:${PGVERSION:-16} + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + command: > + -c ssl=on + -c ssl_cert_file=/etc/ssl/certs/ssl-cert-snakeoil.pem + -c ssl_key_file=/etc/ssl/private/ssl-cert-snakeoil.key + test: + build: + context: . + dockerfile: Dockerfile + environment: + PGSSLMODE: "require" + PGCOPYDB_SOURCE_PGURI: postgres://postgres:h4ckm3@source/postgres + PGCOPYDB_TARGET_PGURI: postgres://postgres:h4ckm3@target/postgres + PGCOPYDB_TABLE_JOBS: 4 + PGCOPYDB_INDEX_JOBS: 2 + PGCOPYDB_OUTPUT_PLUGIN: wal2json + depends_on: + - source + - target diff --git a/tests/follow-sequence-reset/copydb.sh b/tests/follow-sequence-reset/copydb.sh new file mode 100755 index 000000000..fbda0e86b --- /dev/null +++ b/tests/follow-sequence-reset/copydb.sh @@ -0,0 +1,96 @@ +#! /bin/bash + +set -x +set -e + +# Disable pager for psql to avoid hanging in non-interactive environments +export PAGER=cat + +# This script expects the following environment variables to be set: +# +# - PGCOPYDB_SOURCE_PGURI +# - PGCOPYDB_TARGET_PGURI +# - PGCOPYDB_TABLE_JOBS +# - PGCOPYDB_INDEX_JOBS +# - PGCOPYDB_OUTPUT_PLUGIN +# +# Regression test for sequence reset after a standalone `pgcopydb follow` +# reaches endpos -- the path used by resume-cdc helpers. +# +# Postgres logical decoding does not replicate sequences, and CDC replays +# INSERTs with OVERRIDING SYSTEM VALUE (explicit ids that do NOT advance the +# target sequence). So after the base copy the target sequence is stuck at the +# snapshot value, and only catches up to the source if follow_reset_sequences +# runs when follow reaches endpos. Without that reset the target sequence would +# hand out already-used ids after cutover. + +# make sure source and target databases are ready +pgcopydb ping + +psql -o /tmp/s.out -d ${PGCOPYDB_SOURCE_PGURI} -1 -f /usr/src/pagila/pagila-schema.sql +psql -o /tmp/d.out -d ${PGCOPYDB_SOURCE_PGURI} -1 -f /usr/src/pagila/pagila-data.sql + +slot=pgcopydb +seq=rental_rental_id_seq + +# create the replication slot that captures all the changes +# PGCOPYDB_OUTPUT_PLUGIN is set to wal2json in compose.yaml +coproc ( pgcopydb snapshot --follow --slot-name ${slot} ) + +sleep 1 + +# now setup the replication origin (target) and the pgcopydb.sentinel (source) +pgcopydb stream setup + +# initial base copy: clones table data AND sequences at the snapshot value +pgcopydb clone + +# sequence value on the target right after the base copy (the snapshot value) +tgt_after_clone=`psql -AtqX -d ${PGCOPYDB_TARGET_PGURI} -c "select last_value from ${seq}"` +echo "target ${seq} after clone: ${tgt_after_clone}" + +# advance the source sequence well beyond the snapshot value +psql -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/dml.sql + +# allow replaying changes, and set the end position to the current WAL location +pgcopydb stream sentinel set apply +pgcopydb stream sentinel set endpos --current + +# standalone follow: replays the inserts and, on reaching endpos, resets the +# sequences on the target. This is the path used by resume-cdc helpers. +pgcopydb follow --resume + +src_seq=`psql -AtqX -d ${PGCOPYDB_SOURCE_PGURI} -c "select last_value from ${seq}"` +tgt_seq=`psql -AtqX -d ${PGCOPYDB_TARGET_PGURI} -c "select last_value from ${seq}"` + +echo "source ${seq}: ${src_seq}" +echo "target ${seq}: ${tgt_seq}" + +# the sequence must have advanced during CDC, otherwise the test proves nothing +if [ "${src_seq}" -le "${tgt_after_clone}" ] +then + echo "ERROR: source sequence did not advance during CDC" + echo " after clone: ${tgt_after_clone}, source now: ${src_seq}" + echo " the test is not exercising the sequence reset" + exit 1 +fi + +# the target sequence must have been reset to match the source at endpos +if [ "${src_seq}" != "${tgt_seq}" ] +then + echo "ERROR: target sequence ${seq} was not reset to match the source" + echo " source: ${src_seq}, target: ${tgt_seq}" + echo " (target stuck at snapshot value ${tgt_after_clone} means the reset" + echo " did not run after follow reached endpos)" + exit 1 +fi + +echo "" +echo "follow-sequence-reset test: PASSED" +echo " target ${seq} advanced from ${tgt_after_clone} to ${tgt_seq}, matching source" + +kill -TERM ${COPROC_PID} +wait ${COPROC_PID} + +# cleanup +pgcopydb stream cleanup diff --git a/tests/follow-sequence-reset/dml.sql b/tests/follow-sequence-reset/dml.sql new file mode 100644 index 000000000..b2ce9bbd6 --- /dev/null +++ b/tests/follow-sequence-reset/dml.sql @@ -0,0 +1,11 @@ +--- +--- pgcopydb test/follow-sequence-reset/dml.sql +--- +--- Advance rental_rental_id_seq on the source by inserting rows without an +--- explicit rental_id, so nextval() fires. rental_date is varied to satisfy +--- the (rental_date, inventory_id, customer_id) unique constraint. +--- +insert into rental(rental_date, inventory_id, customer_id, staff_id, last_update) +select '2022-06-01'::timestamp + (x || ' seconds')::interval, + 371, 291, 1, '2022-06-01' +from generate_series(1, 10) as t(x);