Handle endpos pipeline shutdown cleanly and reset sequences on follow

.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

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(&copySpecs, &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

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

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)
 			{

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

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

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"]

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

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

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

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