From 451efe8a21621f9785c44730ab568bb3a1291643 Mon Sep 17 00:00:00 2001 From: Amit Kapila Date: Mon, 3 Aug 2026 11:03:25 +0530 Subject: [PATCH v3] Validate publisher for retain_dead_tuples in the apply worker. Enabling retain_dead_tuples requires the publisher to run PostgreSQL 19 or later and to not be in recovery. Previously this was checked only at DDL time. That forced ALTER SUBSCRIPTION ... ENABLE to connect to the publisher, so pg_upgrade (which re-enables subscriptions during restore) failed if the publisher was unreachable. It was also not authoritative, since the publisher's version or recovery status can change afterwards, for example after a failover. Perform the check authoritatively in the apply worker when it connects, and stop running it when enabling a subscription. ENABLE is the only command issued during restore that triggered it, so this also fixes the pg_upgrade failure. The DDL-time check is kept as a convenience for the other paths, none of which are issued during restore. --- src/backend/commands/subscriptioncmds.c | 22 +++++++++------------- src/backend/replication/logical/worker.c | 15 +++++++++++++++ src/include/commands/subscriptioncmds.h | 4 ++++ 3 files changed, 28 insertions(+), 13 deletions(-) diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 67f5699b2c7..d3eef2a3efa 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -139,7 +139,6 @@ static void check_publications_origin_sequences(WalReceiverConn *wrconn, Oid *subrel_local_oids, int subrel_count, char *subname); -static void check_pub_dead_tuple_retention(WalReceiverConn *wrconn); static void check_duplicates_in_publist(List *publist, Datum *datums); static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname); static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err); @@ -977,7 +976,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, NULL, 0, stmt->subname); if (opts.retaindeadtuples) - check_pub_dead_tuple_retention(wrconn); + CheckPubDeadTupleRetention(wrconn); /* * Set sync state based on if we were asked to do data copy or @@ -2084,14 +2083,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, ApplyLauncherWakeupAtCommit(); update_tuple = true; - - /* - * The subscription might be initially created with - * connect=false and retain_dead_tuples=true, meaning the - * remote server's status may not be checked. Ensure this - * check is conducted now. - */ - check_pub_rdt = sub->retaindeadtuples && opts.enabled; break; } @@ -2428,7 +2419,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, PG_TRY(); { if (retain_dead_tuples) - check_pub_dead_tuple_retention(wrconn); + CheckPubDeadTupleRetention(wrconn); check_publications_origin_tables(wrconn, sub->publications, false, retain_dead_tuples, origin, NULL, 0, @@ -3355,10 +3346,15 @@ check_publications_origin_sequences(WalReceiverConn *wrconn, List *publications, * than the PG19, or if the publisher is in recovery (i.e., it is a standby * server). * + * This is used both at DDL time (as a convenience, when a connection to the + * publisher is already being made) and by the apply worker when it connects, + * which is the authoritative check because the publisher's version and + * recovery status can change after the DDL command. + * * See comments atop worker.c for a detailed explanation. */ -static void -check_pub_dead_tuple_retention(WalReceiverConn *wrconn) +void +CheckPubDeadTupleRetention(WalReceiverConn *wrconn) { WalRcvExecResult *res; Oid RecoveryRow[1] = {BOOLOID}; diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 1ca19c1a7a8..79ec6fae59e 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -5734,6 +5734,21 @@ run_apply_worker(void) */ (void) walrcv_identify_system(LogRepWorkerWalRcvConn, &startpointTLI, NULL); + /* + * If retain_dead_tuples is enabled, verify that the publisher is suitable, + * that is, it runs a version that supports the feature and is not in + * recovery. This is the authoritative check. Although the same validation + * is performed opportunistically at DDL time, the publisher's version or + * recovery status may have changed since then, for example after a + * failover. + */ + if (MySubscription->retaindeadtuples) + { + StartTransactionCommand(); + CheckPubDeadTupleRetention(LogRepWorkerWalRcvConn); + CommitTransactionCommand(); + } + set_apply_error_context_origin(originname); set_stream_options(&options, slotname, &origin_startpos); diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h index 63504232a14..c735db60020 100644 --- a/src/include/commands/subscriptioncmds.h +++ b/src/include/commands/subscriptioncmds.h @@ -18,6 +18,8 @@ #include "catalog/objectaddress.h" #include "parser/parse_node.h" +struct WalReceiverConn; /* avoid pulling in walreceiver.h here */ + extern ObjectAddress CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, bool isTopLevel); extern ObjectAddress AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, bool isTopLevel); @@ -36,4 +38,6 @@ extern void CheckSubDeadTupleRetention(bool check_guc, bool sub_disabled, bool retention_active, bool max_retention_set); +extern void CheckPubDeadTupleRetention(struct WalReceiverConn *wrconn); + #endif /* SUBSCRIPTIONCMDS_H */ -- 2.54.0