From 1dab11ff038011274e48eebfa91b19a977570999 Mon Sep 17 00:00:00 2001 From: Chris Munns Date: Fri, 16 Oct 2026 15:31:57 -0400 Subject: [PATCH] Report the target server state after a lost target connection At the start of the copy, log the target server start time and the CREATE INDEX memory budget (index jobs times maintenance_work_mem). When a COPY, CREATE INDEX, or VACUUM worker loses its target connection, connect again and log whether the target restarted, terminated the session, or cannot be reached. --- .github/workflows/run-tests-tiered.yml | 1 + src/bin/pgcopydb/copydb.h | 2 + src/bin/pgcopydb/indexes.c | 5 + src/bin/pgcopydb/table-data.c | 5 +- src/bin/pgcopydb/target_state.c | 158 ++++++++++++++++++++++ src/bin/pgcopydb/target_state.h | 10 ++ src/bin/pgcopydb/vacuum.c | 7 + tests/Makefile | 7 +- tests/index-target-terminate/Dockerfile | 10 ++ tests/index-target-terminate/Makefile | 15 ++ tests/index-target-terminate/compose.yaml | 29 ++++ tests/index-target-terminate/copydb.sh | 66 +++++++++ tests/index-target-terminate/ddl.sql | 14 ++ 13 files changed, 326 insertions(+), 3 deletions(-) create mode 100644 src/bin/pgcopydb/target_state.c create mode 100644 src/bin/pgcopydb/target_state.h create mode 100644 tests/index-target-terminate/Dockerfile create mode 100644 tests/index-target-terminate/Makefile create mode 100644 tests/index-target-terminate/compose.yaml create mode 100755 tests/index-target-terminate/copydb.sh create mode 100644 tests/index-target-terminate/ddl.sql diff --git a/.github/workflows/run-tests-tiered.yml b/.github/workflows/run-tests-tiered.yml index bfc1e0614..1cb3b98b4 100644 --- a/.github/workflows/run-tests-tiered.yml +++ b/.github/workflows/run-tests-tiered.yml @@ -166,6 +166,7 @@ jobs: - follow-defer-validate-fks - follow-sequence-reset - copy-source-reconnect + - index-target-terminate steps: - name: Checkout repository uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 diff --git a/src/bin/pgcopydb/copydb.h b/src/bin/pgcopydb/copydb.h index 8e934ad34..01973618f 100644 --- a/src/bin/pgcopydb/copydb.h +++ b/src/bin/pgcopydb/copydb.h @@ -269,6 +269,8 @@ typedef struct CopyDataSpec bool hasDBCreatePrivilege; bool hasDBTempPrivilege; + char targetStartTime[BUFSIZE]; + Catalogs catalogs; } CopyDataSpec; diff --git a/src/bin/pgcopydb/indexes.c b/src/bin/pgcopydb/indexes.c index 270370bfc..f1bf5f62f 100644 --- a/src/bin/pgcopydb/indexes.c +++ b/src/bin/pgcopydb/indexes.c @@ -20,6 +20,7 @@ #include "signals.h" #include "string_utils.h" #include "summary.h" +#include "target_state.h" static bool copydb_add_table_indexes_hook(void *context, SourceIndex *index); @@ -190,6 +191,8 @@ copydb_start_index_workers(CopyDataSpec *specs) log_info("STEP 6: starting %d CREATE INDEX processes", specs->indexJobs); log_info("STEP 7: constraints are built by the CREATE INDEX processes"); + (void) copydb_target_inspect(specs); + for (int i = 0; i < specs->indexJobs; i++) { /* @@ -306,6 +309,8 @@ copydb_index_worker(CopyDataSpec *specs) { ++errors; + (void) copydb_target_report(specs, &dst); + log_error("Failed to create index with oid %u, " "see above for details", mesg.data.oid); diff --git a/src/bin/pgcopydb/table-data.c b/src/bin/pgcopydb/table-data.c index 0729eed0d..56327c841 100644 --- a/src/bin/pgcopydb/table-data.c +++ b/src/bin/pgcopydb/table-data.c @@ -24,6 +24,7 @@ #include "signals.h" #include "string_utils.h" #include "summary.h" +#include "target_state.h" static bool copydb_copy_supervisor_add_table_hook(void *ctx, SourceTable *table); @@ -66,6 +67,8 @@ copydb_copy_all_table_data(CopyDataSpec *specs) return false; } + (void) copydb_target_inspect(specs); + if (!copydb_process_table_data(specs)) { log_fatal("Failed to COPY the data, see above for details"); @@ -1034,7 +1037,7 @@ copydb_copy_data_by_oid(CopyDataSpec *specs, PGSQL *src, PGSQL *dst, { if (!copydb_copy_table(specs, src, dst, tableSpecs)) { - /* errors have already been logged */ + (void) copydb_target_report(specs, dst); return false; } } diff --git a/src/bin/pgcopydb/target_state.c b/src/bin/pgcopydb/target_state.c new file mode 100644 index 000000000..a16fc054c --- /dev/null +++ b/src/bin/pgcopydb/target_state.c @@ -0,0 +1,158 @@ +#include "copydb.h" +#include "log.h" +#include "pgsql.h" +#include "string_utils.h" +#include "target_state.h" + + +static bool +target_start_time(PGSQL *dst, char *startTime, size_t size) +{ + SingleValueResultContext context = { { 0 }, PGSQL_RESULT_STRING, false }; + + if (!pgsql_execute_with_params(dst, + "select pg_postmaster_start_time()::text", + 0, NULL, NULL, + &context, &parseSingleValueResult) || + !context.parsedOk || + context.strVal == NULL) + { + return false; + } + + strlcpy(startTime, context.strVal, size); + + return true; +} + + +static void +target_log_index_memory(CopyDataSpec *specs, PGSQL *dst) +{ + SingleValueResultContext budget = { { 0 }, PGSQL_RESULT_STRING, false }; + + char jobs[BUFSIZE] = { 0 }; + sformat(jobs, sizeof(jobs), "%d", specs->indexJobs); + + const char *paramValues[1] = { jobs }; + + const char *sql = + "select format('CREATE INDEX memory on the target: %s processes, " + "maintenance_work_mem %s, max_parallel_maintenance_workers %s, " + "total up to %s', " + "$1::bigint, " + "current_setting('maintenance_work_mem'), " + "current_setting('max_parallel_maintenance_workers'), " + "pg_size_pretty(pg_size_bytes(current_setting('maintenance_work_mem')) " + "* $1::bigint))"; + + if (pgsql_execute_with_params(dst, sql, 1, NULL, paramValues, + &budget, &parseSingleValueResult) && + budget.parsedOk && + budget.strVal != NULL) + { + log_info("%s", budget.strVal); + } + else + { + log_warn("Failed to compute the CREATE INDEX memory on the target"); + } +} + + +void +copydb_target_inspect(CopyDataSpec *specs) +{ + if (!IS_EMPTY_STRING_BUFFER(specs->targetStartTime)) + { + return; + } + + PGSQL dst = { 0 }; + + if (!pgsql_init(&dst, specs->connStrings.target_pguri, PGSQL_CONN_TARGET) || + !pgsql_set_gucs(&dst, dstSettings)) + { + log_warn("Failed to read the target server state"); + (void) pgsql_finish(&dst); + return; + } + + if (target_start_time(&dst, + specs->targetStartTime, + sizeof(specs->targetStartTime))) + { + log_info("Target server is up since %s", specs->targetStartTime); + } + else + { + log_warn("Failed to read the target server start time"); + } + + (void) target_log_index_memory(specs, &dst); + (void) pgsql_finish(&dst); +} + + +void +copydb_target_report(CopyDataSpec *specs, PGSQL *failed) +{ + if (!pgsql_state_is_connection_error(failed)) + { + return; + } + + log_error("Lost the target connection (SQLSTATE %s), " + "checking the target server state", + IS_EMPTY_STRING_BUFFER(failed->sqlstate) ? + "unknown" : failed->sqlstate); + + PGSQL dst = { 0 }; + char startTime[BUFSIZE] = { 0 }; + SingleValueResultContext recovery = { { 0 }, PGSQL_RESULT_BOOL, false }; + + if (!pgsql_init(&dst, specs->connStrings.target_pguri, PGSQL_CONN_TARGET)) + { + return; + } + + dst.connectionStatementType = PGSQL_CONNECTION_MULTI_STATEMENT; + + if (!target_start_time(&dst, startTime, sizeof(startTime)) || + !pgsql_execute_with_params(&dst, "select pg_is_in_recovery()", + 0, NULL, NULL, + &recovery, &parseSingleValueResult) || + !recovery.parsedOk) + { + log_error("Failed to connect to the target server again: " + "the server is down, restarting, or failing over"); + (void) pgsql_finish(&dst); + return; + } + + (void) pgsql_finish(&dst); + + if (IS_EMPTY_STRING_BUFFER(specs->targetStartTime)) + { + log_error("Target server is up since %s", startTime); + } + else if (!streq(startTime, specs->targetStartTime)) + { + log_error("Target server restarted at %s, it was up since %s " + "before this step started", + startTime, + specs->targetStartTime); + } + else + { + log_error("Target server did not restart, it is up since %s: " + "a process on the target terminated this session, " + "for example pg_terminate_backend()", + startTime); + } + + if (recovery.boolVal) + { + log_error("Target server is in recovery: a failover can be in progress"); + } +} diff --git a/src/bin/pgcopydb/target_state.h b/src/bin/pgcopydb/target_state.h new file mode 100644 index 000000000..8443a2675 --- /dev/null +++ b/src/bin/pgcopydb/target_state.h @@ -0,0 +1,10 @@ +#ifndef TARGET_STATE_H +#define TARGET_STATE_H + +#include "copydb.h" +#include "pgsql.h" + +void copydb_target_inspect(CopyDataSpec *specs); +void copydb_target_report(CopyDataSpec *specs, PGSQL *failed); + +#endif /* TARGET_STATE_H */ diff --git a/src/bin/pgcopydb/vacuum.c b/src/bin/pgcopydb/vacuum.c index 350163418..e4570a59b 100644 --- a/src/bin/pgcopydb/vacuum.c +++ b/src/bin/pgcopydb/vacuum.c @@ -16,6 +16,7 @@ #include "log.h" #include "signals.h" #include "summary.h" +#include "target_state.h" /* number of times to attempt VACUUM ANALYZE when the target connection drops */ #define VACUUM_ANALYZE_MAX_ATTEMPTS 3 @@ -392,6 +393,12 @@ vacuum_analyze_table_by_oid(CopyDataSpec *specs, uint32_t oid) /* only a lost/broken connection is worth retrying */ bool connectionLost = (dst.status == PG_CONNECTION_BAD); + + if (connectionLost && attempt == VACUUM_ANALYZE_MAX_ATTEMPTS) + { + (void) copydb_target_report(specs, &dst); + } + (void) pgsql_finish(&dst); if (!connectionLost || attempt == VACUUM_ANALYZE_MAX_ATTEMPTS) diff --git a/tests/Makefile b/tests/Makefile index 4610abbd2..506efa051 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -19,7 +19,7 @@ all: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions 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 \ - copy-source-reconnect; + copy-source-reconnect index-target-terminate; pagila: build $(MAKE) -C $@ @@ -120,6 +120,9 @@ follow-sequence-reset: build copy-source-reconnect: build $(MAKE) -C $@ +index-target-terminate: build + $(MAKE) -C $@ + timescaledb: build $(MAKE) -C $@ @@ -135,4 +138,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 +.PHONY: copy-source-reconnect index-target-terminate diff --git a/tests/index-target-terminate/Dockerfile b/tests/index-target-terminate/Dockerfile new file mode 100644 index 000000000..16875d9fd --- /dev/null +++ b/tests/index-target-terminate/Dockerfile @@ -0,0 +1,10 @@ +FROM pagila + +COPY --from=pgcopydb /usr/local/bin/pgcopydb /usr/local/bin + +WORKDIR /usr/src/pgcopydb +COPY copydb.sh copydb.sh +COPY ddl.sql ddl.sql + +USER docker +CMD ["/usr/src/pgcopydb/copydb.sh"] diff --git a/tests/index-target-terminate/Makefile b/tests/index-target-terminate/Makefile new file mode 100644 index 000000000..d405ef0e4 --- /dev/null +++ b/tests/index-target-terminate/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 diff --git a/tests/index-target-terminate/compose.yaml b/tests/index-target-terminate/compose.yaml new file mode 100644 index 000000000..5e9fdd783 --- /dev/null +++ b/tests/index-target-terminate/compose.yaml @@ -0,0 +1,29 @@ +services: + source: + image: postgres:${PGVERSION:-18} + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + target: + image: postgres:${PGVERSION:-18} + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + test: + build: + context: . + dockerfile: Dockerfile + environment: + PGCOPYDB_SOURCE_PGURI: postgres://postgres:h4ckm3@source/postgres + PGCOPYDB_TARGET_PGURI: postgres://postgres:h4ckm3@target/postgres + PGCOPYDB_TABLE_JOBS: 2 + PGCOPYDB_INDEX_JOBS: 2 + depends_on: + - source + - target diff --git a/tests/index-target-terminate/copydb.sh b/tests/index-target-terminate/copydb.sh new file mode 100755 index 000000000..9e6db6a98 --- /dev/null +++ b/tests/index-target-terminate/copydb.sh @@ -0,0 +1,66 @@ +#!/bin/bash + +set -x +set -e + +export PAGER=cat + +DIR=/tmp/index-target-terminate + +pgcopydb ping + +psql -v ON_ERROR_STOP=1 -d "${PGCOPYDB_SOURCE_PGURI}" -f /usr/src/pgcopydb/ddl.sql + +psql -v ON_ERROR_STOP=1 -d "${PGCOPYDB_TARGET_PGURI}" <<'EOF' +CREATE FUNCTION slow_idx() RETURNS event_trigger LANGUAGE plpgsql AS $$ +BEGIN + IF current_query() ~ 'idx_slow_val' + THEN + PERFORM pg_sleep(60); + END IF; +END +$$; + +CREATE EVENT TRIGGER slow_idx ON ddl_command_start + WHEN TAG IN ('CREATE INDEX') + EXECUTE FUNCTION slow_idx(); +EOF + +rm -rf "${DIR}" + +pgcopydb clone --dir "${DIR}" --notice > /tmp/clone.log 2>&1 & +CLONE_PID=$! + +TERMINATED=0 + +for i in $(seq 1 60) +do + TERMINATED=$(psql -tA -d "${PGCOPYDB_TARGET_PGURI}" -c " + SELECT count(pg_terminate_backend(pid)) + FROM pg_stat_activity + WHERE pid <> pg_backend_pid() + AND query LIKE 'CREATE INDEX idx_slow_val%'") + + if [ "${TERMINATED}" -ge 1 ] + then + break + fi + + sleep 1 +done + +if wait ${CLONE_PID} +then + cat /tmp/clone.log + echo "clone was expected to fail" + exit 1 +fi + +cat /tmp/clone.log + +test "${TERMINATED}" -ge 1 + +grep -q "Target server is up since" /tmp/clone.log +grep -q "CREATE INDEX memory on the target: 2 processes" /tmp/clone.log +grep -q "Lost the target connection" /tmp/clone.log +grep -q "Target server did not restart" /tmp/clone.log diff --git a/tests/index-target-terminate/ddl.sql b/tests/index-target-terminate/ddl.sql new file mode 100644 index 000000000..904cc9b68 --- /dev/null +++ b/tests/index-target-terminate/ddl.sql @@ -0,0 +1,14 @@ +CREATE TABLE t_ok ( + id integer PRIMARY KEY, + val text +); + +CREATE TABLE t_slow ( + id integer PRIMARY KEY, + val text +); + +CREATE INDEX idx_slow_val ON t_slow(val); + +INSERT INTO t_ok SELECT i, 'ok' || i FROM generate_series(1, 1000) AS i; +INSERT INTO t_slow SELECT i, 'slow' || i FROM generate_series(1, 1000) AS i;