From: Amit Kapila Date: Mon, 27 Jul 2026 03:36:05 +0000 (+0530) Subject: Fix issues in logical replication sequence synchronization. X-Git-Url: http://git.ipfire.org/gitweb/?a=commitdiff_plain;h=b8d9cf512c1259f97f9896593cc1c8352c1118ac;p=thirdparty%2Fpostgresql.git Fix issues in logical replication sequence synchronization. 1. Stop a running sequence synchronization worker when ALTER SUBSCRIPTION ... DISABLE is executed. The worker did not reread its subscription after starting a transaction, so it kept running with a stale copy and missed the disable. It now calls maybe_reread_subscription() after StartTransactionCommand(), matching the apply worker. 2. Restore the invariant that publisher-side synchronization slots are dropped last during ALTER SUBSCRIPTION ... REFRESH PUBLICATION. The slot-drop loop now runs after the sequence-removal loop, so the non-transactional slot drops happen only after all catalog changes that could still be rolled back on error. 3. Restore psql tab completion for ALTER SUBSCRIPTION ... REFRESH PUBLICATION WITH (. 4. Make pg_stat_subscription report NULL for the fields that do not apply to a sequence synchronization worker, which does not stream from a walsender, and update the documentation accordingly. 5. Update the pg_subscription_rel.srsublsn catalog documentation to describe its semantics for sequence rows. Reported-by: Noah Misch Author: vignesh C Reviewed-by: Hayato Kuroda Reviewed-by: Amit Kapila Backpatch-through: 19, where it was introduced Discussion: https://postgr.es/m/20260710045217.f0.noahmisch@microsoft.com --- diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 4b474c13917..6066c4784f4 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -8893,7 +8893,11 @@ SCRAM-SHA-256$<iteration count>:&l Remote LSN of the state change used for synchronization coordination when in s or r states, - otherwise null + otherwise null. For sequences, this instead holds the publisher + sequence's page LSN as of the last synchronization, which does not + track replication progress the way it does for tables; see + for how it is used to detect + out-of-sync sequences. diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 1ce0ef00799..a209e891b18 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -2473,8 +2473,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage Process ID of the leader apply worker if this process is a parallel - apply worker; NULL if this process is a leader apply worker or a table - synchronization worker + apply worker; NULL if this process is a leader apply worker, a table + synchronization worker or a sequence synchronization worker @@ -2484,7 +2484,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage OID of the relation that the worker is synchronizing; NULL for the - leader apply worker and parallel apply workers + leader apply worker, parallel apply workers and the sequence + synchronization worker @@ -2494,7 +2495,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage Last write-ahead log location received, the initial value of - this field being 0; NULL for parallel apply workers + this field being 0; NULL for parallel apply workers and the sequence + synchronization worker @@ -2504,7 +2506,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage Send time of last message received from origin WAL sender; NULL for - parallel apply workers + parallel apply workers and the sequence synchronization worker @@ -2514,7 +2516,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage Receipt time of last message received from origin WAL sender; NULL for - parallel apply workers + parallel apply workers and the sequence synchronization worker @@ -2524,7 +2526,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage Last write-ahead log location reported to origin WAL sender; NULL for - parallel apply workers + parallel apply workers and the sequence synchronization worker @@ -2534,7 +2536,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage Time of last write-ahead log location reported to origin WAL - sender; NULL for parallel apply workers + sender; NULL for parallel apply workers and the sequence synchronization + worker diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index d4504b4a0c6..013ac46db07 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1288,34 +1288,6 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, } } - /* - * Drop the tablesync slots associated with removed tables. This has - * to be at the end because otherwise if there is an error while doing - * the database operations we won't be able to rollback dropped slots. - */ - foreach_ptr(SubRemoveRels, sub_remove_rel, sub_remove_rels) - { - if (sub_remove_rel->state != SUBREL_STATE_READY && - sub_remove_rel->state != SUBREL_STATE_SYNCDONE) - { - char syncslotname[NAMEDATALEN] = {0}; - - /* - * For READY/SYNCDONE states we know the tablesync slot has - * already been dropped by the tablesync worker. - * - * For other states, there is no certainty, maybe the slot - * does not exist yet. Also, if we fail after removing some of - * the slots, next time, it will again try to drop already - * dropped slots and fail. For these reasons, we allow - * missing_ok = true for the drop. - */ - ReplicationSlotNameForTablesync(sub->oid, sub_remove_rel->relid, - syncslotname, sizeof(syncslotname)); - ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); - } - } - /* * Next remove state for sequences we should not care about anymore * using the data we collected above @@ -1343,6 +1315,34 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, sub->name)); } } + + /* + * Drop the tablesync slots associated with removed tables. This has + * to be at the end because otherwise if there is an error while doing + * the database operations we won't be able to rollback dropped slots. + */ + foreach_ptr(SubRemoveRels, sub_remove_rel, sub_remove_rels) + { + if (sub_remove_rel->state != SUBREL_STATE_READY && + sub_remove_rel->state != SUBREL_STATE_SYNCDONE) + { + char syncslotname[NAMEDATALEN] = {0}; + + /* + * For READY/SYNCDONE states we know the tablesync slot has + * already been dropped by the tablesync worker. + * + * For other states, there is no certainty, maybe the slot + * does not exist yet. Also, if we fail after removing some of + * the slots, next time, it will again try to drop already + * dropped slots and fail. For these reasons, we allow + * missing_ok = true for the drop. + */ + ReplicationSlotNameForTablesync(sub->oid, sub_remove_rel->relid, + syncslotname, sizeof(syncslotname)); + ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); + } + } } PG_FINALLY(); { diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c index d0370056de3..fe506a98c20 100644 --- a/src/backend/replication/logical/sequencesync.c +++ b/src/backend/replication/logical/sequencesync.c @@ -495,6 +495,7 @@ copy_sequences(WalReceiverConn *conn) TupleTableSlot *slot; StartTransactionCommand(); + maybe_reread_subscription(); for (int idx = cur_batch_base_index; idx < n_seqinfos; idx++) { @@ -724,6 +725,7 @@ LogicalRepSyncSequences(void) StringInfoData app_name; StartTransactionCommand(); + maybe_reread_subscription(); rel = table_open(SubscriptionRelRelationId, AccessShareLock); diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 0ff5cef63cd..0bd19074010 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -5985,8 +5985,18 @@ SetupApplyOrSyncWorker(int worker_slot) */ /* Initialise stats to a sanish value */ - MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time = - MyLogicalRepWorker->reply_time = GetCurrentTimestamp(); + if (am_sequencesync_worker()) + { + MyLogicalRepWorker->last_send_time = + MyLogicalRepWorker->last_recv_time = + MyLogicalRepWorker->reply_time = 0; + } + else + { + MyLogicalRepWorker->last_send_time = + MyLogicalRepWorker->last_recv_time = + MyLogicalRepWorker->reply_time = GetCurrentTimestamp(); + } /* Load the libpq-specific functions */ load_file("libpqwalreceiver", false); diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index 1cacc8c3ea2..17dcabe755f 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -2354,6 +2354,9 @@ match_previous_words(int pattern_id, /* ALTER SUBSCRIPTION REFRESH */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH")) COMPLETE_WITH("PUBLICATION", "SEQUENCES"); + /* ALTER SUBSCRIPTION REFRESH PUBLICATION */ + else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH", "PUBLICATION")) + COMPLETE_WITH("WITH ("); /* ALTER SUBSCRIPTION REFRESH PUBLICATION WITH ( */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH", "PUBLICATION", "WITH", "(")) COMPLETE_WITH("copy_data");