From 6e24fbbbe8c37a3fa1bcc4d259fdef42812a8216 Mon Sep 17 00:00:00 2001 From: Amit Kapila Date: Fri, 31 Jul 2026 11:54:50 +0530 Subject: [PATCH v1] Validate publisher for retain_dead_tuples at apply worker connect time. Enabling retain_dead_tuples checked the publisher (version >= 19 and not in recovery) only at DDL time. That forced a connection to the publisher while re-enabling subscriptions during pg_upgrade, so the upgrade failed if the publisher was unreachable. It was also insufficient: the publisher's version and recovery status can change after the DDL command (for example, after a failover), leaving the check stale. Move the authoritative check to the apply worker, which runs it when it connects to the publisher. The DDL-time check is retained as a convenience but is now skipped during binary upgrade, since that path only recreates catalog state and must not connect to the publisher. --- src/backend/commands/subscriptioncmds.c | 23 ++++++++++++++++++----- src/backend/replication/logical/worker.c | 16 ++++++++++++++++ src/include/commands/subscriptioncmds.h | 4 ++++ 3 files changed, 38 insertions(+), 5 deletions(-) diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 67f5699b2c7..c2a62198d45 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 @@ -2391,6 +2390,15 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, heap_freetuple(tup); } + /* + * During binary upgrade, we only recreate the catalog state and must not + * connect to the publisher. The publisher's suitability for + * retain_dead_tuples is validated authoritatively by the apply worker + * when it connects, so skip the opportunistic DDL-time check here. + */ + if (IsBinaryUpgrade) + check_pub_rdt = false; + /* * Try to acquire the connection necessary either for modifying the slot * or for checking if the remote server permits enabling @@ -2428,7 +2436,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 +3363,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..80ce11d7c0c 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -5734,6 +5734,22 @@ 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), and DDL-time validation is skipped entirely during + * binary upgrade. + */ + 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