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 doing 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.

Reported-by: Noah Misch <noah@leadboat.com>
Analyzed-by: Jeff Davis <pgsql@j-davis.com>
Author: Amit Kapila <amit.kapila16@gmail.com>
Reviewed-by: Jeff Davis <pgsql@j-davis.com>
Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Backpatch-through: 19, where it was introduced
Discussion: https://postgr.es/m/20260710195902.4f.noahmisch@microsoft.com

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e330fb13e42e3f3524ef0025abd7340cadd9eec4..9671541caf9394fc73cacd2b026be7342a674bd2 100644 (file)
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -135,7 +135,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);
@@ -918,7 +917,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
@@ -1931,14 +1930,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;
                        }
 
@@ -2275,7 +2266,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,
@@ -3147,10 +3138,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 24ce03be0673df59a91eaa72be11baee92b07fd3..2dd421412d65a9bcf86e2a63a5d2fad34256c79a 100644 (file)
--- 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 63504232a14813a862c46a1171bfac3674901be6..c735db6002077df7512412abd67806a3abe3fe1b 100644 (file)
--- 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 */