Report the target server state after a lost target connection

.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

src/bin/pgcopydb/copydb.h

@@ -269,6 +269,8 @@ typedef struct CopyDataSpec
 	bool hasDBCreatePrivilege;
 	bool hasDBTempPrivilege;
 
+	char targetStartTime[BUFSIZE];
+
 	Catalogs catalogs;
 } CopyDataSpec;
 

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

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

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");
+	}
+}

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 */

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)

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

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

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

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