Fix CDC endpos on an idle source, retry COPY on source loss, move to Debian 12

.github/workflows/run-tests-tiered.yml

@@ -76,6 +76,7 @@ jobs:
     name: Smoke tests (PG ${{ matrix.PGVERSION }})
     runs-on: ubuntu-latest
     needs: [style_checker, banned_api_checker, docs]
+    if: github.event_name == 'push' || github.event.pull_request.head.repo.full_name == github.repository
     strategy:
       fail-fast: true
       matrix:
@@ -165,6 +166,7 @@ jobs:
           - fk-partition-clone
           - follow-defer-validate-fks
           - follow-sequence-reset
+          - copy-source-reconnect
     steps:
       - name: Checkout repository
         uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6

Dockerfile

@@ -1,6 +1,6 @@
 # syntax=docker/dockerfile:latest
 # Define a base image with all our build dependencies.
-FROM debian:11-slim AS build
+FROM debian:12-slim AS build
 
 # multi-arch
 ARG TARGETPLATFORM
@@ -18,7 +18,7 @@ RUN dpkg --add-architecture ${TARGETARCH:-arm64} && apt update \
 	gnupg
 
 RUN curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add -
-RUN echo "deb http://apt.postgresql.org/pub/repos/apt bullseye-pgdg main ${PGVERSION}" > /etc/apt/sources.list.d/pgdg.list
+RUN echo "deb http://apt.postgresql.org/pub/repos/apt bookworm-pgdg main ${PGVERSION}" > /etc/apt/sources.list.d/pgdg.list
 
 RUN dpkg --add-architecture ${TARGETARCH:-arm64} && apt update \
   && apt install -qqy --no-install-recommends \
@@ -89,7 +89,7 @@ RUN make -s clean && make -s -j$(nproc) install
 COPY tests tests
 
 # Now the "run" image, as small as possible
-FROM debian:11-slim AS run
+FROM debian:12-slim AS run
 
 # multi-arch
 ARG TARGETPLATFORM
@@ -110,7 +110,7 @@ RUN dpkg --add-architecture ${TARGETARCH:-arm64} && apt update \
 	gnupg
 
 RUN curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add -
-RUN echo "deb http://apt.postgresql.org/pub/repos/apt bullseye-pgdg main ${PGVERSION}" > /etc/apt/sources.list.d/pgdg.list
+RUN echo "deb http://apt.postgresql.org/pub/repos/apt bookworm-pgdg main ${PGVERSION}" > /etc/apt/sources.list.d/pgdg.list
 
 RUN dpkg --add-architecture ${TARGETARCH:-arm64} && apt update \
   && apt install -qqy --no-install-suggests --no-install-recommends \

src/bin/pgcopydb/pgsql.c

@@ -2484,12 +2484,24 @@ is_response_ok(PGresult *result)
 #define SQLSTATE_IS_CONNECTION_EXCEPTION(pgsql) \
 	(pgsql->sqlstate[0] == '0' && pgsql->sqlstate[1] == '8')
 
+/* never widen to Class 57: 57014 query_canceled must not be retried */
+#define SQLSTATE_IS_SESSION_TERMINATION(pgsql) \
+	(strncmp(pgsql->sqlstate, "57P01", 5) == 0 || \
+	 strncmp(pgsql->sqlstate, "57P02", 5) == 0 || \
+	 strncmp(pgsql->sqlstate, "57P03", 5) == 0)
+
 bool
 pgsql_state_is_connection_error(PGSQL *pgsql)
 {
-	return pgsql->connection != NULL &&
-		   (PQstatus(pgsql->connection) == CONNECTION_BAD ||
-			SQLSTATE_IS_CONNECTION_EXCEPTION(pgsql));
+	if (pgsql->connection != NULL &&
+		PQstatus(pgsql->connection) == CONNECTION_BAD)
+	{
+		return true;
+	}
+
+	return pgsql->status == PG_CONNECTION_BAD ||
+		   SQLSTATE_IS_CONNECTION_EXCEPTION(pgsql) ||
+		   SQLSTATE_IS_SESSION_TERMINATION(pgsql);
 }
 
 
@@ -3288,6 +3300,10 @@ pgcopy_log_error(PGSQL *pgsql, PGresult *res, const char *context)
 			strlcpy(pgsql->sqlstate, sqlstate, sizeof(pgsql->sqlstate));
 		}
 	}
+	else
+	{
+		pgsql->sqlstate[0] = '\0';
+	}
 
 	char *endpoint =
 		pgsql->connectionType == PGSQL_CONN_SOURCE ? "SOURCE" : "TARGET";
@@ -3325,6 +3341,11 @@ pgcopy_log_error(PGSQL *pgsql, PGresult *res, const char *context)
 		PQclear(res);
 	}
 
+	if (PQstatus(pgsql->connection) == CONNECTION_BAD)
+	{
+		pgsql->status = PG_CONNECTION_BAD;
+	}
+
 	clear_results(pgsql);
 	pgsql_finish(pgsql);
 }
@@ -4268,7 +4289,7 @@ pgsql_stream_logical(LogicalStreamClient *client, LogicalStreamContext *context)
 			 * happens, stop streaming — we have all the WAL we need.
 			 */
 			if (context->endpos != InvalidXLogRecPtr &&
-				context->endpos <= cur_record_lsn)
+				context->endpos <= client->current.written_lsn)
 			{
 				log_info("Streamed up to write_lsn %X/%X, "
 						 "flush_lsn %X/%X, stopping: "

src/bin/pgcopydb/snapshot.c

@@ -258,7 +258,8 @@ copydb_close_snapshot(CopyDataSpec *copySpecs)
 		{
 			(void) pgsql_finish(&(snapshot->stream.pgsql));
 		}
-		else if (snapshot->kind == SNAPSHOT_KIND_SQL)
+		else if (snapshot->kind == SNAPSHOT_KIND_SQL &&
+				 pgsql->connection != NULL)
 		{
 			/* only COMMIT sql snapshot kinds, no need for logical rep ones */
 			if (!pgsql_commit(pgsql))

src/bin/pgcopydb/table-data.c

@@ -1383,6 +1383,42 @@ typedef struct UpdateCopyStatsContext
 } UpdateCopyStatsContext;
 
 
+static bool
+copydb_reset_copy_connections(CopyDataSpec *specs, PGSQL *src, PGSQL *dst)
+{
+	bool srcIsSnapshot = src == &(specs->sourceSnapshot.pgsql);
+
+	(void) pgsql_finish(src);
+	(void) pgsql_finish(dst);
+
+	if (srcIsSnapshot)
+	{
+		if (!copydb_set_snapshot(specs))
+		{
+			log_error("Failed to re-open the source snapshot \"%s\", "
+					  "see above for details",
+					  specs->sourceSnapshot.snapshot);
+			return false;
+		}
+
+		pgsql_set_copy_retry_policy(&(src->retryPolicy));
+	}
+	else if (pgsql_open_connection(src) == NULL)
+	{
+		return false;
+	}
+
+	if (!pgsql_set_gucs(dst, dstSettings))
+	{
+		log_error("Failed to set our GUC settings on the target connection, "
+				  "see above for details");
+		return false;
+	}
+
+	return true;
+}
+
+
 /*
  * copydb_copy_table implements the sub-process activity to pg_dump |
  * pg_restore the table's data and then create the indexes and the constraints
@@ -1434,33 +1470,40 @@ copydb_copy_table(CopyDataSpec *specs, PGSQL *src, PGSQL *dst,
 	{
 		++attempts;
 
-		/* re-init stats between attempts */
-		CopyStats empty = { 0 };
-		stats = empty;
+		bool connected =
+			attempts == 1 ||
+			copydb_reset_copy_connections(specs, src, dst);
 
-		UpdateCopyStatsContext context = {
-			.specs = specs,
-			.tableSpecs = tableSpecs
-		};
+		if (connected)
+		{
+			/* re-init stats between attempts */
+			CopyStats empty = { 0 };
+			stats = empty;
 
-		/* ignore previous attempts, we need only one success here */
-		success = pg_copy(src, dst,
-						  &(tableSpecs->copyArgs), &stats,
-						  &context, &copydb_update_copy_stats_hook);
+			UpdateCopyStatsContext context = {
+				.specs = specs,
+				.tableSpecs = tableSpecs
+			};
 
-		if (success)
-		{
-			/* success, get out of the retry loop */
-			if (attempts > 1)
+			/* ignore previous attempts, we need only one success here */
+			success = pg_copy(src, dst,
+							  &(tableSpecs->copyArgs), &stats,
+							  &context, &copydb_update_copy_stats_hook);
+
+			if (success)
 			{
-				log_info("Table %s COPY succeeded after %d attempts",
-						 tableSpecs->sourceTable->qname,
-						 attempts);
+				/* success, get out of the retry loop */
+				if (attempts > 1)
+				{
+					log_info("Table %s COPY succeeded after %d attempts",
+							 tableSpecs->sourceTable->qname,
+							 attempts);
+				}
+				break;
 			}
-			break;
 		}
 
-		/* only retry on Connection Exception errors (SQLSTATE class 08) */
+		/* only retry when the source or the target connection is lost */
 		bool isConnectionError =
 			pgsql_state_is_connection_error(src) ||
 			pgsql_state_is_connection_error(dst);

tests/Makefile

@@ -18,7 +18,8 @@ all: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions
 	 follow-pgoutput 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 \
-	 fk-partition-clone follow-defer-validate-fks follow-sequence-reset;
+	 fk-partition-clone follow-defer-validate-fks follow-sequence-reset \
+	 copy-source-reconnect;
 
 pagila: build
 	$(MAKE) -C $@
@@ -113,6 +114,9 @@ follow-defer-validate-fks: build
 follow-sequence-reset: build
 	$(MAKE) -C $@
 
+copy-source-reconnect: build
+	$(MAKE) -C $@
+
 timescaledb: build
 	$(MAKE) -C $@
 
@@ -128,3 +132,4 @@ build:
 .PHONY: endpos-in-multi-wal-txn exclude-extension
 .PHONY: blob-snapshot-release follow-defer-indexes fk-not-valid defer-validate-fks
 .PHONY: fk-partition-clone follow-defer-validate-fks follow-sequence-reset
+.PHONY: copy-source-reconnect

tests/copy-source-reconnect/Dockerfile

@@ -0,0 +1,9 @@
+FROM pagila
+
+COPY --from=pgcopydb /usr/local/bin/pgcopydb /usr/local/bin
+
+WORKDIR /usr/src/pgcopydb
+COPY copydb.sh copydb.sh
+
+USER docker
+CMD ["/usr/src/pgcopydb/copydb.sh"]

tests/copy-source-reconnect/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: down build test

tests/copy-source-reconnect/compose.yaml

@@ -0,0 +1,42 @@
+services:
+  source:
+    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
+  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
+    cap_add:
+      - SYS_ADMIN
+      - SYS_PTRACE
+    environment:
+      PGSSLMODE: "require"
+      PGCOPYDB_SOURCE_PGURI: postgres://postgres:h4ckm3@source/postgres
+      PGCOPYDB_TARGET_PGURI: postgres://postgres:h4ckm3@target/postgres
+      PGCOPYDB_TABLE_JOBS: 1
+      PGCOPYDB_INDEX_JOBS: 1
+      PGCOPYDB_FAIL_FAST: "true"
+    depends_on:
+      - source
+      - target

tests/copy-source-reconnect/copydb.sh

@@ -0,0 +1,61 @@
+#! /bin/bash
+
+set -x
+set -e
+set -o pipefail
+
+export PAGER=cat
+
+pgcopydb ping
+
+psql -d ${PGCOPYDB_SOURCE_PGURI} <<EOF
+create table copytest(id bigint primary key, payload text);
+
+insert into copytest(id, payload)
+     select i, repeat('a', 400) from generate_series(1, 400000) as i;
+EOF
+
+psql -v ON_ERROR_STOP=1 -d ${PGCOPYDB_SOURCE_PGURI} <<EOF &
+set statement_timeout to '120s';
+
+do \$\$
+declare
+    victim int;
+begin
+    loop
+        perform pg_stat_clear_snapshot();
+
+        select pid into victim
+          from pg_stat_activity
+         where query ilike '%copytest%to stdout%'
+           and pid <> pg_backend_pid()
+         limit 1;
+
+        if victim is not null
+        then
+            perform pg_terminate_backend(victim);
+            exit;
+        end if;
+
+        perform pg_sleep(0.02);
+    end loop;
+end
+\$\$;
+EOF
+killer=$!
+
+pgcopydb clone --notice 2>&1 | tee /tmp/clone.log
+
+wait ${killer}
+
+grep "COPY succeeded after" /tmp/clone.log
+
+count=$(psql -At -d ${PGCOPYDB_TARGET_PGURI} -c 'select count(*) from copytest')
+
+if [ "${count}" != "400000" ]
+then
+    echo "target table copytest has ${count} rows, expected 400000"
+    exit 1
+fi
+
+pgcopydb compare data

tests/exclude-extension/compose.yaml

@@ -28,6 +28,8 @@ services:
     build:
       context: .
       dockerfile: Dockerfile
+      args:
+        PGCOPYDB_IMAGE: pgcopydb:pg${PGVERSION:-16}
     environment:
       PGCOPYDB_SOURCE_PGURI: postgres://postgres:h4ckm3@source/postgres
       PGCOPYDB_TARGET_PGURI: postgres://postgres:h4ckm3@target/postgres