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/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; } 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);