From ead21d00cbec48df82662287fcd3f9db3f6bba12 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, including any need to wait for standby confirmation. 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 | 43 +++++++++++++++++++++++++---- 1 file changed, 37 insertions(+), 6 deletions(-) diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index d0e335629e7..54686c8c3e4 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -311,7 +311,8 @@ static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); static void WalSndCheckShutdownTimeout(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, @@ -1754,7 +1755,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); @@ -2095,7 +2096,7 @@ WalSndWaitForWal(XLogRecPtr loc) last_flush = now; } - WalSndWait(wakeEvents, sleeptime, wait_event); + WalSndWait(wakeEvents, sleeptime, wait_event, loc); } /* reactivate latch so WalSndLoop knows to continue */ @@ -3215,7 +3216,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); } } } @@ -3901,7 +3903,7 @@ WalSndDone(WalSndSendDataCallback send_data) /* Sleep until something happens or we time out */ WalSndWait(WL_SOCKET_WRITEABLE, sleeptime, - WAIT_EVENT_WAL_SENDER_WRITE_DATA); + WAIT_EVENT_WAL_SENDER_WRITE_DATA, InvalidXLogRecPtr); /* Clear any already-pending wakeups */ ResetLatch(MyLatch); @@ -4101,9 +4103,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_FOR_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; @@ -4182,8 +4188,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_FOR_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