From 56a272d311af88605bed91e056e0261b70f1d006 Mon Sep 17 00:00:00 2001 From: Fujii Masao Date: Thu, 12 Nov 2026 02:03:13 +0900 Subject: [PATCH v1] Fix lost wakeups in logical WAL senders Logical WAL senders could sleep even when WAL was available for decoding, delaying logical replication until another event woke them. When synchronous logical replication is configured, this could also delay commits waiting for acknowledgment from the subscriber. Commit bc971f4025c changed WAL sender wakeups to use condition variables, which notify only senders already on the waitlist. However, WalSndWaitForWal() checks the flush position on a primary, or the replay position on a standby, before WalSndWait() joins the waitlist. If the required WAL becomes available between these two steps, the wakeup can be missed, leaving the sender asleep despite having enough WAL to continue decoding. Fix this by passing the required WAL position to WalSndWait() and rechecking the flush or replay position after joining the waitlist. If enough WAL is available, return to WalSndWaitForWal() to reevaluate its conditions. This allows decoding to proceed without waiting for another wakeup when the required WAL became available before waitlist registration, avoiding the resulting replication and synchronous commit delays. Backpatch to v16, where commit bc971f4025c introduced the condition-variable wakeup mechanism. Author: Fujii Masao Backpatch-through: 16 --- src/backend/replication/walsender.c | 42 +++++++++++++++++++++++++---- 1 file changed, 37 insertions(+), 5 deletions(-) diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 37b45d8d286..70ebb7a07f5 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -264,7 +264,8 @@ static void WalSndKeepalive(bool requestReply, XLogRecPtr writePtr); static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); static long WalSndComputeSleeptime(TimestampTz now); -static void WalSndWait(uint32 socket_events, long timeout, uint32 wait_event); +static void WalSndWait(uint32 socket_events, long timeout, uint32 wait_event, + XLogRecPtr target_lsn); static void WalSndPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write); static void WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write); static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, @@ -1483,7 +1484,7 @@ ProcessPendingWrites(void) /* Sleep until something happens or we time out */ WalSndWait(WL_SOCKET_WRITEABLE | WL_SOCKET_READABLE, sleeptime, - WAIT_EVENT_WAL_SENDER_WRITE_DATA); + WAIT_EVENT_WAL_SENDER_WRITE_DATA, InvalidXLogRecPtr); /* Clear any already-pending wakeups */ ResetLatch(MyLatch); @@ -1719,7 +1720,8 @@ WalSndWaitForWal(XLogRecPtr loc) last_flush = now; } - WalSndWait(wakeEvents, sleeptime, WAIT_EVENT_WAL_SENDER_WAIT_WAL); + WalSndWait(wakeEvents, sleeptime, WAIT_EVENT_WAL_SENDER_WAIT_WAL, + loc); } /* reactivate latch so WalSndLoop knows to continue */ @@ -2651,7 +2653,8 @@ WalSndLoop(WalSndSendDataCallback send_data) } /* Sleep until something happens or we time out */ - WalSndWait(wakeEvents, sleeptime, WAIT_EVENT_WAL_SENDER_MAIN); + WalSndWait(wakeEvents, sleeptime, WAIT_EVENT_WAL_SENDER_MAIN, + InvalidXLogRecPtr); } } } @@ -3446,9 +3449,13 @@ WalSndWakeup(bool physical, bool logical) * Wait for readiness on the FeBe socket, or a timeout. The mask should be * composed of optional WL_SOCKET_WRITEABLE and WL_SOCKET_READABLE flags. Exit * on postmaster death. + * + * target_lsn is the required WAL position for + * WAIT_EVENT_WAL_SENDER_WAIT_WAL and is unused for other wait events. */ static void -WalSndWait(uint32 socket_events, long timeout, uint32 wait_event) +WalSndWait(uint32 socket_events, long timeout, uint32 wait_event, + XLogRecPtr target_lsn) { WaitEvent event; @@ -3521,8 +3528,33 @@ WalSndWait(uint32 socket_events, long timeout, uint32 wait_event) } } else if (MyWalSnd->kind == REPLICATION_KIND_LOGICAL) + { ConditionVariablePrepareToSleep(&WalSndCtl->wal_replay_cv); + /* + * Recheck the WAL position tested by WalSndWaitForWal() now that + * we're on the waitlist. Otherwise, we could miss a wakeup for a WAL + * flush or replay that occurred before we joined it. + */ + if (wait_event == WAIT_EVENT_WAL_SENDER_WAIT_WAL) + { + XLogRecPtr flushPtr; + + Assert(XLogRecPtrIsValid(target_lsn)); + + if (!RecoveryInProgress()) + flushPtr = GetFlushRecPtr(NULL); + else + flushPtr = GetXLogReplayRecPtr(NULL); + + if (flushPtr >= target_lsn) + { + ConditionVariableCancelSleep(); + return; + } + } + } + if (WaitEventSetWait(FeBeWaitSet, timeout, &event, 1, wait_event) == 1 && (event.events & WL_POSTMASTER_DEATH)) { -- 2.56.0