diff --git a/src/bin/pgcopydb/pgsql.c b/src/bin/pgcopydb/pgsql.c index f241b5a6..32d7d03d 100644 --- a/src/bin/pgcopydb/pgsql.c +++ b/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,7 +2351,7 @@ pgsql_execute_log_error(PGSQL *pgsql, { log_error("[%s %d] SQL query: %s", endpoint, - PQbackendPID(pgsql->connection), + backendPID, sql); } @@ -2346,7 +2359,7 @@ pgsql_execute_log_error(PGSQL *pgsql, { 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); } diff --git a/src/bin/pgcopydb/vacuum.c b/src/bin/pgcopydb/vacuum.c index c7c42875..35016341 100644 --- a/src/bin/pgcopydb/vacuum.c +++ b/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 */