Retry post-copy VACUUM ANALYZE on transient target connection loss

src/bin/pgcopydb/pgsql.c

@@ -2296,39 +2296,52 @@ pgsql_execute_log_error(PGSQL *pgsql,
 		pgsql->connectionType == PGSQL_CONN_SOURCE ? "SOURCE" : "TARGET";
 
 	/*
-	 * PostgreSQL Error message might contain several lines. Log each of
-	 * them as a separate ERROR line here.
+	 * When the connection is already gone (e.g. an async teardown in
+	 * pgsql_fetch_results already logged the real error and called
+	 * pgsql_finish), PQerrorMessage(NULL) returns the unhelpful literal
+	 * "connection pointer is NULL". Skip the libpq message/PID block in that
+	 * case and use a backend PID of 0, while still logging the useful SQL
+	 * query/params context below. Mirrors the guard in log_connection_error.
 	 */
-	char *message = PQerrorMessage(pgsql->connection);
+	int backendPID = pgsql->connection ? PQbackendPID(pgsql->connection) : 0;
 
-	LinesBuffer lbuf = { 0 };
-
-	/*
-	 * PostgreSQL error message could be a static memory area in the
-	 * code, in which case we are not allowed to edit the text and inject
-	 * newlines. Duplicate the memory area before manipulating it.
-	 *
-	 * Also, because we use the Boehm GC lib, refrain from manually calling
-	 * free() on the newly allocated memory area.
-	 */
-	if (message != NULL)
+	if (pgsql->connection != NULL)
 	{
-		/* make sure message is writable by splitLines */
-		message = strdup(message);
-	}
+		/*
+		 * PostgreSQL Error message might contain several lines. Log each of
+		 * them as a separate ERROR line here.
+		 */
+		char *message = PQerrorMessage(pgsql->connection);
 
-	if (!splitLines(&lbuf, message))
-	{
-		/* errors have already been logged */
-		return;
-	}
+		LinesBuffer lbuf = { 0 };
 
-	for (uint64_t lineNumber = 0; lineNumber < lbuf.count; lineNumber++)
-	{
-		log_error("[%s %d] %s",
-				  endpoint,
-				  PQbackendPID(pgsql->connection),
-				  lbuf.lines[lineNumber]);
+		/*
+		 * PostgreSQL error message could be a static memory area in the
+		 * code, in which case we are not allowed to edit the text and inject
+		 * newlines. Duplicate the memory area before manipulating it.
+		 *
+		 * Also, because we use the Boehm GC lib, refrain from manually calling
+		 * free() on the newly allocated memory area.
+		 */
+		if (message != NULL)
+		{
+			/* make sure message is writable by splitLines */
+			message = strdup(message);
+		}
+
+		if (!splitLines(&lbuf, message))
+		{
+			/* errors have already been logged */
+			return;
+		}
+
+		for (uint64_t lineNumber = 0; lineNumber < lbuf.count; lineNumber++)
+		{
+			log_error("[%s %d] %s",
+					  endpoint,
+					  backendPID,
+					  lbuf.lines[lineNumber]);
+		}
 	}
 
 	if (pgsql->logSQL)
@@ -2338,15 +2351,15 @@ pgsql_execute_log_error(PGSQL *pgsql,
 		{
 			log_error("[%s %d] SQL query: %s",
 					  endpoint,
-					  PQbackendPID(pgsql->connection),
+					  backendPID,
 					  sql);
 		}
 
 		if (debugParameters != NULL)
 		{
 			log_error("[%s %d] SQL params: %s",
 					  endpoint,
-					  PQbackendPID(pgsql->connection),
+					  backendPID,
 					  debugParameters->data);
 		}
 	}
@@ -4645,6 +4658,13 @@ pgsql_stream_log_error(PGSQL *pgsql, PGresult *res, const char *message)
 	}
 
 	clear_results(pgsql);
+
+	/*
+	 * This function always tears the connection down. Mark it bad first so
+	 * callers can distinguish a lost/broken connection (worth retrying) from a
+	 * genuine SQL error via pgsql->status.
+	 */
+	pgsql->status = PG_CONNECTION_BAD;
 	pgsql_finish(pgsql);
 }
 

src/bin/pgcopydb/vacuum.c

@@ -17,6 +17,9 @@
 #include "signals.h"
 #include "summary.h"
 
+/* number of times to attempt VACUUM ANALYZE when the target connection drops */
+#define VACUUM_ANALYZE_MAX_ATTEMPTS 3
+
 /*
  * vacuum_start_supervisor starts a VACUUM supervisor process.
  */
@@ -324,15 +327,6 @@ vacuum_analyze_table_by_oid(CopyDataSpec *specs, uint32_t oid)
 		return false;
 	}
 
-	PGSQL dst = { 0 };
-
-	/* initialize our connection to the target database */
-	if (!pgsql_init(&dst, specs->connStrings.target_pguri, PGSQL_CONN_TARGET))
-	{
-		/* errors have already been logged */
-		return false;
-	}
-
 	/* finally, vacuum analyze the table and its indexes */
 	char vacuum[BUFSIZE] = { 0 };
 
@@ -369,14 +363,53 @@ vacuum_analyze_table_by_oid(CopyDataSpec *specs, uint32_t oid)
 		return false;
 	}
 
-	if (!pgsql_execute(&dst, vacuum))
+	/*
+	 * VACUUM ANALYZE is an optional post-copy optimization. A momentary
+	 * connection loss to the target (e.g. an SSL EOF) should not fail the
+	 * worker, so retry on a lost/broken connection. Each attempt uses a fresh
+	 * connection; pgsql_execute opens it, and the interactive open-retry
+	 * policy absorbs the reconnect wait. Genuine SQL errors leave the
+	 * connection status OK and are not retried.
+	 */
+	bool vacuumed = false;
+
+	for (int attempt = 1; attempt <= VACUUM_ANALYZE_MAX_ATTEMPTS; attempt++)
+	{
+		PGSQL dst = { 0 };
+
+		if (!pgsql_init(&dst, specs->connStrings.target_pguri, PGSQL_CONN_TARGET))
+		{
+			/* errors have already been logged */
+			return false;
+		}
+
+		if (pgsql_execute(&dst, vacuum))
+		{
+			(void) pgsql_finish(&dst);
+			vacuumed = true;
+			break;
+		}
+
+		/* only a lost/broken connection is worth retrying */
+		bool connectionLost = (dst.status == PG_CONNECTION_BAD);
+		(void) pgsql_finish(&dst);
+
+		if (!connectionLost || attempt == VACUUM_ANALYZE_MAX_ATTEMPTS)
+		{
+			break;
+		}
+
+		log_warn("Retrying VACUUM ANALYZE for %s after target connection loss "
+				 "(attempt %d/%d)",
+				 table.qname, attempt + 1, VACUUM_ANALYZE_MAX_ATTEMPTS);
+	}
+
+	if (!vacuumed)
 	{
 		log_error("Failed to run command, see above for details: %s", vacuum);
 		return false;
 	}
 
-	(void) pgsql_finish(&dst);
-
 	if (!summary_finish_vacuum(sourceDB, &tableSpecs))
 	{
 		/* errors have already been logged */