From 661a913cc5fe34364a1db9a4a782c44bb790b9d0 Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Mon, 19 Oct 2026 22:26:33 +0000 Subject: [PATCH] Report durable advances to the source without waiting Standby replies went out only every status interval (10 s by default) or when a keepalive asked. Apply and flush, bounded as before by shadow replay, the ClickHouse-acknowledged resume-safe position and the persisted floor, could therefore sit a whole interval behind what the daemon had made durable: the source kept that WAL longer, and on the write-heavy repro (interval 1 s) it was most of a second of the drain's acknowledgement tail. Reply as soon as apply or flush advances past what the source was last told, as PostgreSQL's walreceiver does after each flush, but at most ten times per interval. The values sent are unchanged, so nothing is acknowledged earlier than it is durable; only the delay shrinks. A zero --status-interval is now refused: PostgreSQL reads a zero wal_receiver_status_interval as no replies, and the source keeps WAL until a reply releases it. Co-Authored-By: Claude Opus 5.5 --- src/bin/stream/args.rs | 4 +- src/source/source_feed.rs | 79 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 81 insertions(+), 2 deletions(-) diff --git a/src/bin/stream/args.rs b/src/bin/stream/args.rs index 7ed2ebb0..7cbbe423 100644 --- a/src/bin/stream/args.rs +++ b/src/bin/stream/args.rs @@ -137,7 +137,9 @@ pub(crate) struct Args { /// segment boundary. #[arg(long)] pub(crate) start_lsn: Option, - #[arg(long, default_value_t = 10)] + /// Seconds between standby status replies and resume manifest writes. + /// Must be positive. PostgreSQL retains WAL until status replies allow removal. + #[arg(long, default_value_t = 10, value_parser = clap::value_parser!(u64).range(1..))] pub(crate) status_interval: u64, /// Concurrent archive fetches, each holding at most one WAL segment. diff --git a/src/source/source_feed.rs b/src/source/source_feed.rs index 3250e46d..38185b15 100644 --- a/src/source/source_feed.rs +++ b/src/source/source_feed.rs @@ -334,7 +334,11 @@ impl SourceFeed { "source replication receive timeout" ); let status_interval = self.status_interval.min(RECEIVE_TIMEOUT / 2); - if self.last_status.elapsed() >= status_interval { + // PostgreSQL's walreceiver reports WAL progress without waiting. + let advanced = status.write_lsn > self.floors.write + || status.flush_lsn > self.floors.flush + || status.apply_lsn > self.floors.apply; + if advanced || self.last_status.elapsed() >= status_interval { self.send_status(status).await?; } let status_due = self.last_status + status_interval; @@ -647,6 +651,79 @@ mod tests { } } + #[tokio::test(start_paused = true)] + async fn advances_are_reported_without_waiting_for_status_interval() { + let (feed, mut server) = mock_feed().await; + let mut feed = feed.with_status_interval(Duration::from_secs(10)); + let replies = tokio::spawn(async move { + let mut reported = Vec::new(); + while let Ok(tag) = server.read_u8().await { + assert_eq!(tag, b'd'); + let len = server.read_u32().await.unwrap() as usize; + let mut body = vec![0; len - 4]; + server.read_exact(&mut body).await.unwrap(); + reported.push(( + u64::from_be_bytes(body[1..9].try_into().unwrap()), + u64::from_be_bytes(body[9..17].try_into().unwrap()), + u64::from_be_bytes(body[17..25].try_into().unwrap()), + )); + } + reported + }); + let mut buf = Vec::new(); + for (write, flush, apply) in [ + (100, 0, 0), + (100, 0, 100), + (100, 100, 100), + (200, 100, 100), + (200, 100, 200), + (200, 200, 200), + ] { + let status = StandbyStatus { + write_lsn: Pos::new(write), + flush_lsn: Pos::new(flush), + apply_lsn: Pos::new(apply), + }; + for _ in 0..2 { + let poll = feed.next_event(status, &mut buf); + let timed_out = tokio::time::timeout(Duration::from_millis(100), poll).await; + assert!(timed_out.is_err()); + } + } + drop(feed); + assert_eq!( + replies.await.unwrap(), + [ + (100, 0, 0), + (100, 0, 100), + (100, 100, 100), + (200, 0, 100), + (200, 0, 200), + (200, 200, 200), + ] + ); + } + + /// Read buffered WAL even when status replies are due on every loop. + #[tokio::test(start_paused = true)] + async fn zero_interval_reads_buffered_wal() { + let (feed, mut server) = mock_feed().await; + let mut feed = feed.with_status_interval(Duration::ZERO); + let mut frame = vec![b'w']; + for field in [16u64, 19, 0] { + frame.extend_from_slice(&field.to_be_bytes()); + } + frame.extend_from_slice(b"wal"); + backend(&mut server, b'd', &frame).await; + let mut buf = Vec::new(); + let poll = feed.next_event(StandbyStatus::default(), &mut buf); + let event = tokio::time::timeout(Duration::from_secs(1), poll).await; + assert!( + matches!(event, Ok(Ok(SourceEvent::Wal(w))) if w.start_lsn == 16 && w.data == b"wal"), + "{event:?}" + ); + } + fn triple(s: StandbyStatus) -> (u64, u64, u64) { (s.write_lsn.get(), s.flush_lsn.get(), s.apply_lsn.get()) }