From b372e9f48d60d195fffd1c459675651b69280987 Mon Sep 17 00:00:00 2001 From: danielshih Date: Fri, 30 Oct 2026 13:35:59 +0000 Subject: [PATCH] Keep reconnect state out of the next_event future a memcpy of it each time: +7% CPU per event on the rustls backend, +4% on libpq. Box the two recover_connection awaits, since reconnecting is cold. The retry loop also held its whole match scrutinee, a Result>, across the awaits in its error arm, a 1 KB slot in the future; bind only the error out of the match. Keep the next_event_before future small and guard it EventStream::next_event_before matched `self.held.take()` and `resume_parked_poll().await` as nested scrutinees, so both Option temporaries lived across the final read: its future was 2,800 B against 728 B for next_event. It runs only while an aligned snapshot catches up, so the cost is small, but it is the same pattern the previous commit removes from the retry loop, and the guard could not cover it. Take each early read in its own `if let` that returns, so its event ends before the next await. The future is now 720 B. Behaviour is unchanged. Extend read_path_futures_stay_small to both next_event_before futures. Add the two cases no test reached: a parked poll that ended in an error hands it to the next read, and a commit exactly at the boundary is held while one LSN below is delivered (the boundary is a copy's consistent point, which contains everything before it and nothing at or after it). Test the retry loop's permanent-error and live-connection arms Two arms of next_event_with_retry_until had no test: a permanent read error ends the loop at once, and a decode error on a connection that is still up is surfaced, not retried, because the bad frame is already consumed and a retry would silently skip a change. A null test connection never reports alive and never carries a server error, so neither was reachable. Add two test-only seams on the native connection: one in COPY mode whose server end the test holds, so an ErrorResponse written to it comes out of get_copy_data_async through the real worker; and a setter for the liveness flag the worker normally keeps. The permanent case runs on a real worker and socket with a hard timeout; the live-connection case uses the paused clock, which shows that no backoff started. Both run on the rustls backend; the lines they cover are shared. --- .github/workflows/ci-checks.yml | 5 +- Cargo.toml | 6 +- integration-tests/reconnect_recovery.rs | 255 ++++++++++++++ src/connection/native/connection.rs | 16 + src/stream.rs | 443 ++++++++++++++++++------ 5 files changed, 613 insertions(+), 112 deletions(-) create mode 100644 integration-tests/reconnect_recovery.rs diff --git a/.github/workflows/ci-checks.yml b/.github/workflows/ci-checks.yml index 1d447ed..5c54ab3 100644 --- a/.github/workflows/ci-checks.yml +++ b/.github/workflows/ci-checks.yml @@ -390,8 +390,9 @@ jobs: # `PQgetCopyData` on libpq, with nothing shared between the two. Each # backend sends `unread_stream_feedback`'s status update through its # own writer. `startup_metadata` drives each backend's own non-blocking - # startup path. - error_classification|upload_manifest|initial_snapshot|require_existing_slot|unread_stream_feedback|aligned_snapshot|startup_metadata) + # startup path. `reconnect_recovery` reconnects through each backend's own + # connect, identity check and START_REPLICATION. + error_classification|upload_manifest|initial_snapshot|require_existing_slot|unread_stream_feedback|aligned_snapshot|startup_metadata|reconnect_recovery) variants=("" "--no-default-features --features libpq") ;; *) variants=("") ;; esac diff --git a/Cargo.toml b/Cargo.toml index ed80e2d..5eb0a68 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -85,7 +85,7 @@ rustls-tls = [ ] [dev-dependencies] -tokio = { version = "1.54.1", features = ["full"] } +tokio = { version = "1.54.1", features = ["full", "test-util"] } criterion = { package = "codspeed-criterion-compat", version = "5.0.3", features = ["html_reports"] } futures = "0.3.34" serde_bytes = "0.11.19" @@ -209,6 +209,10 @@ path = "integration-tests/libpq_async_connect.rs" name = "libpq_async_commands" path = "integration-tests/libpq_async_commands.rs" +[[test]] +name = "reconnect_recovery" +path = "integration-tests/reconnect_recovery.rs" + [[bench]] name = "wal_pipeline" harness = false diff --git a/integration-tests/reconnect_recovery.rs b/integration-tests/reconnect_recovery.rs new file mode 100644 index 0000000..979f8d5 --- /dev/null +++ b/integration-tests/reconnect_recovery.rs @@ -0,0 +1,255 @@ +#![cfg(any(feature = "libpq", feature = "rustls-tls"))] + +//! A stream whose walsender is terminated reconnects and carries on, with no row lost +//! and none repeated. +//! +//! `require_existing_slot` covers recovery that fails. This is the only test of one that +//! succeeds, through both read entry points: `LogicalReplicationStream::next_event_with_retry` +//! and `EventStream::next_event` under `tokio::time::timeout`, the way a consumer drives it. +//! +//! Requires PostgreSQL with `wal_level=logical`, `DATABASE_URL` in replication mode and, +//! optionally, `DATABASE_URL_REGULAR` for the same database. + +use core::time::Duration; +use pg_walstream::{ + EventStream, EventType, LogicalReplicationStream, PgReplicationConnection, + ReplicationStreamConfig, +}; +use std::time::Instant; +use tokio_util::sync::CancellationToken; + +/// Generous: a reconnect sits out a 1 s backoff, and a loaded CI runner is slower still. +const READ_TIMEOUT: Duration = Duration::from_secs(30); + +fn replication_conn_string() -> String { + std::env::var("DATABASE_URL").unwrap_or_else(|_| { + "postgresql://postgres:postgres@localhost:5432/test_walstream?replication=database" + .to_string() + }) +} + +struct Fixture { + regular: PgReplicationConnection, + slot: &'static str, +} + +impl Fixture { + fn new(slot: &'static str) -> Self { + let regular_url = std::env::var("DATABASE_URL_REGULAR").unwrap_or_else(|_| { + replication_conn_string() + .replace("?replication=database", "") + .replace("&replication=database", "") + }); + let regular = PgReplicationConnection::connect(®ular_url).expect("regular connection"); + let mut fixture = Self { regular, slot }; + fixture.drop_slot(); + fixture + .regular + .exec(&format!( + "DROP PUBLICATION IF EXISTS {slot}_pub; \ + DROP TABLE IF EXISTS {slot}; \ + CREATE TABLE {slot} (id integer PRIMARY KEY); \ + CREATE PUBLICATION {slot}_pub FOR TABLE {slot}" + )) + .expect("create published table"); + fixture + } + + /// A persistent slot: a temporary one is dropped with its connection, and the + /// reconnect would recreate it at a later point and lose the rows under test. + fn config(&self) -> ReplicationStreamConfig { + ReplicationStreamConfig::builder(self.slot, format!("{}_pub", self.slot)) + } + + fn drop_slot(&mut self) { + let started = Instant::now(); + let active_sql = format!( + "SELECT active FROM pg_replication_slots WHERE slot_name = '{}'", + self.slot + ); + while started.elapsed() < Duration::from_secs(10) + && self + .regular + .exec(&active_sql) + .ok() + .and_then(|r| r.get_value(0, 0)) + .as_deref() + == Some("t") + { + std::thread::sleep(Duration::from_millis(20)); + } + let _ = self.regular.exec(&format!( + "SELECT pg_drop_replication_slot('{0}') WHERE EXISTS \ + (SELECT 1 FROM pg_replication_slots WHERE slot_name = '{0}')", + self.slot + )); + } + + fn insert(&mut self, id: i32) { + self.regular + .exec(&format!("INSERT INTO {} VALUES ({id})", self.slot)) + .expect("insert row"); + } + + fn walsender_pid(&mut self) -> String { + self.regular + .exec(&format!( + "SELECT active_pid FROM pg_replication_slots WHERE slot_name = '{}'", + self.slot + )) + .expect("query the slot") + .get_value(0, 0) + .expect("the slot has a walsender attached") + } + + /// Kill the walsender, and insist it was there to kill: a `NULL` `active_pid` would + /// terminate nothing and let the test pass without any recovery. + fn terminate_walsender(&mut self) { + let terminated = self + .regular + .exec(&format!( + "SELECT pg_terminate_backend(active_pid) FROM pg_replication_slots \ + WHERE slot_name = '{}'", + self.slot + )) + .expect("terminate the walsender") + .get_value(0, 0); + assert_eq!( + terminated.as_deref(), + Some("t"), + "no live walsender was terminated" + ); + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.drop_slot(); + let _ = self.regular.exec(&format!( + "DROP PUBLICATION IF EXISTS {0}_pub; DROP TABLE IF EXISTS {0}", + self.slot + )); + } +} + +fn inserted_id(data: &pg_walstream::RowData) -> i32 { + data.get("id") + .and_then(|value| value.as_str()) + .expect("id column") + .parse() + .expect("integer id") +} + +/// Read to the commit that carries `until`, acknowledging each commit, and return the ids +/// seen. Acknowledging matters: a reconnect resumes from the applied position, so an +/// unacknowledged row would be delivered again, which is correct but not what this asserts. +async fn read_until(stream: &mut LogicalReplicationStream, until: i32) -> Vec { + tokio::time::timeout(READ_TIMEOUT, async { + let cancel = CancellationToken::new(); + let mut ids = Vec::new(); + loop { + let event = stream + .next_event_with_retry(&cancel) + .await + .expect("the stream must recover, not give up"); + match event.event_type { + EventType::Insert { data, .. } => ids.push(inserted_id(&data)), + EventType::Commit { end_lsn, .. } => { + stream + .shared_lsn_feedback + .update_applied_lsn(end_lsn.value()); + if ids.last() == Some(&until) { + return ids; + } + } + _ => {} + } + } + }) + .await + .unwrap_or_else(|_| panic!("row {until} never arrived after the reconnect")) +} + +/// [`read_until`] through `EventStream::next_event` under a timeout, the shape a consumer +/// uses and the one whose future is moved on every event. +async fn read_until_timed(events: &mut EventStream, until: i32) -> Vec { + let mut ids = Vec::new(); + let deadline = Instant::now() + READ_TIMEOUT; + loop { + let left = deadline.saturating_duration_since(Instant::now()); + let event = tokio::time::timeout(left, events.next_event()) + .await + .unwrap_or_else(|_| panic!("row {until} never arrived after the reconnect")) + .expect("the stream must recover, not give up"); + match event.event_type { + EventType::Insert { data, .. } => ids.push(inserted_id(&data)), + EventType::Commit { end_lsn, .. } => { + events.update_applied_lsn(end_lsn.value()); + if ids.last() == Some(&until) { + return ids; + } + } + _ => {} + } + } +} + +/// One terminated walsender: the row inserted around it arrives exactly once, from a new +/// walsender, and the earlier row is not delivered again. +#[tokio::test] +#[ignore = "requires live PostgreSQL with wal_level=logical"] +async fn a_stream_resumes_after_its_walsender_is_terminated() { + let mut fixture = Fixture::new("it_reconnect_basic"); + let mut stream = LogicalReplicationStream::new(&replication_conn_string(), fixture.config()) + .await + .expect("stream"); + stream.start(None).await.expect("start"); + + fixture.insert(1); + assert_eq!(read_until(&mut stream, 1).await, [1]); + let before = fixture.walsender_pid(); + + fixture.terminate_walsender(); + fixture.insert(2); + + assert_eq!( + read_until(&mut stream, 2).await, + [2], + "row 2 exactly once, and row 1 not again" + ); + assert_ne!( + fixture.walsender_pid(), + before, + "the rows must come from a new walsender" + ); +} + +/// Two terminations in a row through `EventStream::next_event` under `timeout`: the loop +/// recovers more than once, and each recovery leaves the next read working. +#[tokio::test] +#[ignore = "requires live PostgreSQL with wal_level=logical"] +async fn an_event_stream_under_timeout_survives_repeated_terminations() { + let mut fixture = Fixture::new("it_reconnect_timed"); + let mut stream = LogicalReplicationStream::new(&replication_conn_string(), fixture.config()) + .await + .expect("stream"); + stream.start(None).await.expect("start"); + let mut events = stream.into_stream(CancellationToken::new()); + + fixture.insert(1); + assert_eq!(read_until_timed(&mut events, 1).await, [1]); + let mut previous = fixture.walsender_pid(); + + for id in [2, 3] { + fixture.terminate_walsender(); + fixture.insert(id); + assert_eq!( + read_until_timed(&mut events, id).await, + [id], + "row {id} exactly once" + ); + let current = fixture.walsender_pid(); + assert_ne!(current, previous, "row {id} must come from a new walsender"); + previous = current; + } +} diff --git a/src/connection/native/connection.rs b/src/connection/native/connection.rs index 8dede8c..0a98dbe 100644 --- a/src/connection/native/connection.rs +++ b/src/connection/native/connection.rs @@ -1929,6 +1929,22 @@ impl NativeConnection { conn } + /// Test-only: a [`silent_for_testing`](Self::silent_for_testing) connection already in + /// COPY mode. Its reads go to the real worker and socket, so whatever the caller writes + /// to the returned server end (an `ErrorResponse`, say) comes out of + /// `get_copy_data_async` exactly as it would from a live server. + pub(crate) fn silent_replication_for_testing() -> (Self, std::net::TcpStream) { + let (mut conn, peer) = Self::silent_for_testing(); + conn.copy = CopyMode::Replication; + (conn, peer) + } + + /// Test-only: force the liveness flag the worker normally maintains, for paths that + /// branch on `is_alive()` (a null connection never reports alive). + pub(crate) fn set_alive_for_testing(&self, alive: bool) { + self.alive.store(alive, Ordering::Relaxed); + } + /// Test-only: whether this connection uses the inline driver. fn driver_is_inline(&self) -> bool { matches!(self.driver, Driver::Inline { .. }) diff --git a/src/stream.rs b/src/stream.rs index dbaaad3..a9ae69a 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -1339,7 +1339,11 @@ impl LogicalReplicationStream { if !self.connection.is_alive() { warn!("Connection health check failed, attempting recovery"); - match self.recover_connection().await { + // Boxed: reconnecting is cold, and awaiting it inline would make every + // retrying read future (`next_event_with_retry`, `EventStream::next_event`) + // carry the whole connect state machine, which callers then memcpy once + // per event. + match Box::pin(self.recover_connection()).await { Ok(_) => { info!("Connection recovered successfully"); } @@ -1577,104 +1581,107 @@ impl LogicalReplicationStream { loop { attempt += 1; - match self.next_event_until(cancellation_token, boundary).await { + // Only the error leaves this statement. Matching with the arms' awaits inside + // would keep the whole `Result>` temporary alive across them, + // and callers memcpy that slot once per event. + let e = match self.next_event_until(cancellation_token, boundary).await { Ok(event) => return Ok(event), - Err(e) => { - // Cancellation and bounded-replay stop are terminal — never retry. - if matches!( - e, - ReplicationError::Cancelled(_) | ReplicationError::StreamStopped(_) - ) { - debug!("Terminal signal, ending stream: {}", e); - return Err(e); - } + Err(e) => e, + }; - if e.is_permanent() { - error!("Permanent error in event processing: {}", e); - return Err(e); - } + // Cancellation and bounded-replay stop are terminal — never retry. + if matches!( + e, + ReplicationError::Cancelled(_) | ReplicationError::StreamStopped(_) + ) { + debug!("Terminal signal, ending stream: {}", e); + return Err(e); + } - // A decode failure on a *live* connection is not recoverable by - // retrying: `next_event` has already consumed the offending frame, - // so looping just reads the NEXT frame and returns that — silently - // dropping a change the consumer never saw, while the received LSN - // has already advanced past it. Surface it instead. - // - // Server-reported conditions do NOT reach this guard as - // `Protocol`: `from_sqlstate` maps the transient ones (57P01 - // admin shutdown, 55006 slot busy, …) to `TransientConnection`, - // which is not in the match below. And any error out of the COPY - // read path clears the backend's liveness first — the native - // worker stores `alive = false` for every non-cancelled error, - // and the libpq backend reports `is_alive() == false` once - // `copy_end` is latched — so `is_alive()` is already false by the - // time a server error arrives here. What this guard actually - // catches is parse-stage failures on a connection that is still - // healthy. Unclassified SQLSTATEs still fall back to `Protocol`; - // those genuinely are terminal here for the reason above. - if self.connection.is_alive() - && matches!( - e, - ReplicationError::Protocol(_) - | ReplicationError::Buffer(_) - | ReplicationError::Deserialize(_) - ) - { - error!("Unrecoverable decode error on a live connection: {}", e); - return Err(e); - } + if e.is_permanent() { + error!("Permanent error in event processing: {}", e); + return Err(e); + } - // Check if we've exhausted retry attempts - if attempt >= MAX_ATTEMPTS { - error!( - "Exhausted retry attempts ({}) for event processing: {}", - MAX_ATTEMPTS, e - ); - return Err(e); - } + // A decode failure on a *live* connection is not recoverable by + // retrying: `next_event` has already consumed the offending frame, + // so looping just reads the NEXT frame and returns that — silently + // dropping a change the consumer never saw, while the received LSN + // has already advanced past it. Surface it instead. + // + // Server-reported conditions do NOT reach this guard as + // `Protocol`: `from_sqlstate` maps the transient ones (57P01 + // admin shutdown, 55006 slot busy, …) to `TransientConnection`, + // which is not in the match below. And any error out of the COPY + // read path clears the backend's liveness first — the native + // worker stores `alive = false` for every non-cancelled error, + // and the libpq backend reports `is_alive() == false` once + // `copy_end` is latched — so `is_alive()` is already false by the + // time a server error arrives here. What this guard actually + // catches is parse-stage failures on a connection that is still + // healthy. Unclassified SQLSTATEs still fall back to `Protocol`; + // those genuinely are terminal here for the reason above. + if self.connection.is_alive() + && matches!( + e, + ReplicationError::Protocol(_) + | ReplicationError::Buffer(_) + | ReplicationError::Deserialize(_) + ) + { + error!("Unrecoverable decode error on a live connection: {}", e); + return Err(e); + } - warn!( - "Transient error in event processing (attempt {}/{}): {}", - attempt, MAX_ATTEMPTS, e - ); + // Check if we've exhausted retry attempts + if attempt >= MAX_ATTEMPTS { + error!( + "Exhausted retry attempts ({}) for event processing: {}", + MAX_ATTEMPTS, e + ); + return Err(e); + } - // Attempt connection recovery if connection is dead - if !self.connection.is_alive() { - if let Err(recovery_err) = self.recover_connection().await { - // Only a *permanent* recovery failure ends the stream. - // Returning on a transient one skipped the backoff - // below and handed the consumer an error it would - // immediately retry, at zero delay. The connection is - // marked dead on this path, so the next attempt - // re-enters recovery rather than reading from a socket - // that never entered COPY mode. - if recovery_err.is_permanent() || attempt >= MAX_ATTEMPTS { - error!("Failed to recover connection: {}", recovery_err); - return Err(recovery_err); - } - warn!( - "Connection recovery failed (attempt {}/{}), backing off: {}", - attempt, MAX_ATTEMPTS, recovery_err - ); - } else { - info!("Connection recovered successfully"); - } - } + warn!( + "Transient error in event processing (attempt {}/{}): {}", + attempt, MAX_ATTEMPTS, e + ); - // Exponential backoff before retry - let delay = Duration::from_millis(1000 * (1 << (attempt - 1))); - debug!("Waiting {:?} before retry attempt {}", delay, attempt + 1); - - // Use tokio::select! to allow cancellation during backoff - tokio::select! { - biased; - _ = tokio::time::sleep(delay) => {}, - _ = cancellation_token.cancelled() => { - return Err(ReplicationError::cancelled( - "Operation cancelled during retry backoff" - )); - } + // Attempt connection recovery if connection is dead + if !self.connection.is_alive() { + if let Err(recovery_err) = Box::pin(self.recover_connection()).await { + // Only a *permanent* recovery failure ends the stream. + // Returning on a transient one skipped the backoff + // below and handed the consumer an error it would + // immediately retry, at zero delay. The connection is + // marked dead on this path, so the next attempt + // re-enters recovery rather than reading from a socket + // that never entered COPY mode. + if recovery_err.is_permanent() || attempt >= MAX_ATTEMPTS { + error!("Failed to recover connection: {}", recovery_err); + return Err(recovery_err); } + warn!( + "Connection recovery failed (attempt {}/{}), backing off: {}", + attempt, MAX_ATTEMPTS, recovery_err + ); + } else { + info!("Connection recovered successfully"); + } + } + + // Exponential backoff before retry + let delay = Duration::from_millis(1000 * (1 << (attempt - 1))); + debug!("Waiting {:?} before retry attempt {}", delay, attempt + 1); + + // Use tokio::select! to allow cancellation during backoff + tokio::select! { + biased; + _ = tokio::time::sleep(delay) => {}, + _ = cancellation_token.cancelled() => { + return Err(ReplicationError::cancelled( + "Operation cancelled during retry backoff" + )); } } } @@ -2647,27 +2654,32 @@ impl EventStream { /// [`LogicalReplicationStream::next_event_before`] with this stream's retry and recovery. pub async fn next_event_before(&mut self, boundary: Lsn) -> Result> { - let event = match self.held.take() { - Some(event) => event, - None => match self.resume_parked_poll().await { - Some(result) => result?, - None => { - let inner = self - .state - .stream_mut() - .expect("Polling was handled above, so the stream is in hand"); - return inner - .next_event_with_retry_until(&self.cancellation_token, Some(boundary)) - .await; - } - }, - }; - // A held event or one a parked poll read without the boundary. + // Each early read is its own statement that returns, so its event-sized + // temporary is gone before the next `.await`. Matching them as one expression + // kept both alive across the final read: ~2 KB in a future callers move. + if let Some(event) = self.held.take() { + return Ok(self.deliver_unless_past(event, boundary)); + } + if let Some(result) = self.resume_parked_poll().await { + return Ok(self.deliver_unless_past(result?, boundary)); + } + let inner = self + .state + .stream_mut() + .expect("Polling was handled above, so the stream is in hand"); + inner + .next_event_with_retry_until(&self.cancellation_token, Some(boundary)) + .await + } + + /// An event read without the boundary (held, or from an abandoned poll): hand it over + /// unless its commit is at or past `boundary`, in which case keep it for the next read. + fn deliver_unless_past(&mut self, event: ChangeEvent, boundary: Lsn) -> Option { if event.event_type.commit_lsn().is_some_and(|c| c >= boundary) { self.held = Some(event); - return Ok(None); + return None; } - Ok(Some(event)) + Some(event) } /// Await a poll future that `poll_next` parked and its caller abandoned, taking the stream back. @@ -5435,6 +5447,145 @@ pub(crate) mod tests { ); } + /// A stream whose reads fail (the test connection is never in COPY mode) and whose + /// recovery fails at once and transiently, with no network: an exhausted + /// `max_duration` makes the reconnect give up before its first attempt. + fn stream_that_cannot_read_or_reconnect() -> LogicalReplicationStream { + let retry = RetryConfig { + max_duration: Duration::ZERO, + ..RetryConfig::default() + }; + create_test_stream(ReplicationStreamConfig::builder("slot", "pub").with_retry_config(retry)) + } + + /// The loop makes three attempts with a 1 s then a 2 s backoff between them, retries + /// through a failed recovery, and hands the last read error back. + #[tokio::test(start_paused = true)] + async fn next_event_with_retry_backs_off_and_gives_up_after_three_attempts() { + let mut stream = stream_that_cannot_read_or_reconnect(); + let token = CancellationToken::new(); + let started = tokio::time::Instant::now(); + + let err = stream.next_event_with_retry(&token).await.unwrap_err(); + + assert!( + matches!(err, ReplicationError::Protocol(_)), + "the last read error comes back, not the recovery error or a cancellation: {err:?}" + ); + let waited = started.elapsed(); + assert!( + waited >= Duration::from_secs(3) && waited < Duration::from_millis(3100), + "1 s + 2 s of backoff between three attempts, waited {waited:?}" + ); + } + + /// Cancelling while the loop sleeps between attempts ends it at once, not after the delay. + #[tokio::test(start_paused = true)] + async fn next_event_with_retry_backoff_ends_on_cancellation() { + let mut stream = stream_that_cannot_read_or_reconnect(); + let token = CancellationToken::new(); + let started = tokio::time::Instant::now(); + let cancel_midway = async { + tokio::time::sleep(Duration::from_millis(500)).await; + token.cancel(); + }; + + let (result, ()) = tokio::join!(stream.next_event_with_retry(&token), cancel_midway); + + let err = result.unwrap_err(); + assert!( + matches!(&err, ReplicationError::Cancelled(m) if m.contains("retry backoff")), + "expected the cancellation raised from the backoff, got {err:?}" + ); + let waited = started.elapsed(); + assert!( + waited < Duration::from_secs(1), + "cancelled at 500 ms, yet waited {waited:?}" + ); + } + + /// The server-side `ErrorResponse` a COPY stream ends with, as the wire carries it. + #[cfg(feature = "rustls-tls")] + fn error_response(sqlstate: &str, message: &str) -> Vec { + let fields = format!("SERROR\0C{sqlstate}\0M{message}\0\0"); + let mut frame = vec![b'E']; + frame.extend_from_slice(&((4 + fields.len()) as i32).to_be_bytes()); + frame.extend_from_slice(fields.as_bytes()); + frame + } + + /// A permanent error out of the COPY stream (`42704`: the slot is gone) ends the loop + /// on the first attempt. Retrying cannot help, and recovery would only reconnect into + /// the same failure. A real worker and socket carry the error, so this is the path a + /// live server takes. Real time, because a paused clock would jump ahead while the + /// worker thread is still reading; a retry would sit out a 1 s and a 2 s backoff. + #[cfg(feature = "rustls-tls")] + #[tokio::test] + async fn next_event_with_retry_returns_a_permanent_read_error_without_retrying() { + use std::io::Write; + let (connection, mut server) = PgReplicationConnection::silent_replication_for_testing(); + let mut stream = stream_that_cannot_read_or_reconnect(); + stream.connection = connection; + server + .write_all(&error_response( + "42704", + "replication slot \"slot\" does not exist", + )) + .unwrap(); + let token = CancellationToken::new(); + let started = std::time::Instant::now(); + + let err = tokio::time::timeout( + Duration::from_secs(20), + stream.next_event_with_retry(&token), + ) + .await + .expect("the error must reach the loop, not leave it waiting") + .unwrap_err(); + + assert!( + matches!(err, ReplicationError::ReplicationSlot(_)), + "the server's error comes back as it was classified: {err:?}" + ); + assert!(err.is_permanent()); + let waited = started.elapsed(); + assert!( + waited < Duration::from_secs(2), + "a permanent error must not wait out a backoff, waited {waited:?}" + ); + } + + /// A frame that fails to decode on a connection that is still up is surfaced, not + /// retried: the bad frame is already consumed, so a retry would read the next one and + /// silently skip a change. The paused clock shows no backoff was started. + #[cfg(feature = "rustls-tls")] + #[tokio::test(start_paused = true)] + async fn next_event_with_retry_surfaces_a_decode_error_on_a_live_connection() { + let retry = RetryConfig { + max_duration: Duration::ZERO, + ..RetryConfig::default() + }; + let config = ReplicationStreamConfig::builder("slot", "pub").with_retry_config(retry); + // One byte is too short to be an XLogData header. + let mut stream = + create_test_stream_with_frames(config, vec![bytes::Bytes::from_static(b"w")]); + stream.connection.set_alive_for_testing(true); + let token = CancellationToken::new(); + let started = tokio::time::Instant::now(); + + let err = stream.next_event_with_retry(&token).await.unwrap_err(); + + assert!( + matches!(err, ReplicationError::Protocol(_)), + "the decode error itself comes back: {err:?}" + ); + assert_eq!( + started.elapsed(), + Duration::ZERO, + "a decode error on a live connection must not be retried" + ); + } + #[tokio::test] async fn event_stream_terminates_none_on_stream_stopped() { use futures::StreamExt; @@ -5449,6 +5600,35 @@ pub(crate) mod tests { assert!(es.is_terminated(), "stream should mark itself terminated"); } + /// Callers move the `next_event` future once per event (`timeout`, `select!`, a + /// `Stream` adapter), so its size is a per-event cost, and `stream.rs` is outside + /// every benchmark. Two things have inflated it: reconnect state awaited inline + /// (several KB) and a `ChangeEvent`-sized value held across an await in the retry + /// loop (about 1 KB). The limit sits below the second, so either one fails this. + #[test] + fn read_path_futures_stay_small() { + const LIMIT: usize = 1024; + let mut stream = create_test_stream(create_test_config()); + let token = CancellationToken::new(); + let boundary = Lsn::new(0x100); + let plain = std::mem::size_of_val(&stream.next_event(&token)); + let retry = std::mem::size_of_val(&stream.next_event_with_retry(&token)); + let before = std::mem::size_of_val(&stream.next_event_before(boundary, &token)); + let mut es = stream.into_stream(token); + let event_stream = std::mem::size_of_val(&es.next_event()); + let event_stream_before = std::mem::size_of_val(&es.next_event_before(boundary)); + + for (name, size) in [ + ("LogicalReplicationStream::next_event", plain), + ("LogicalReplicationStream::next_event_with_retry", retry), + ("LogicalReplicationStream::next_event_before", before), + ("EventStream::next_event", event_stream), + ("EventStream::next_event_before", event_stream_before), + ] { + assert!(size <= LIMIT, "{name} future grew to {size} bytes"); + } + } + #[test] fn test_validate_replication_options_valid_v1() { let config = ReplicationStreamConfig::new( @@ -9129,6 +9309,51 @@ pub(crate) mod tests { ); } + /// A commit exactly at the boundary belongs after it: the boundary is the consistent + /// point of a copy, which already contains everything committed before it and nothing + /// at or after it. One LSN below is delivered, the boundary itself is held. + #[tokio::test] + async fn event_stream_holds_a_parked_commit_exactly_at_the_boundary() { + let boundary = Lsn::new(0x200); + let mut events = + create_test_stream(create_test_config()).into_stream(CancellationToken::new()); + + events.state.park_with_result(Ok(begin_event(0x1ff))); + let below = events.next_event_before(boundary).await.unwrap(); + assert_eq!( + below.unwrap().event_type.commit_lsn(), + Some(Lsn::new(0x1ff)), + "one below the boundary is still before it" + ); + + events.state.park_with_result(Ok(begin_event(0x200))); + assert!( + events.next_event_before(boundary).await.unwrap().is_none(), + "a commit at the boundary must be held back" + ); + let held = events.next_event().await.unwrap(); + assert_eq!(held.event_type.commit_lsn(), Some(boundary)); + } + + /// An abandoned poll that ended in an error hands that error to the next read, and the + /// stream is back in hand for the one after. + #[tokio::test] + async fn event_stream_surfaces_the_error_of_a_parked_poll() { + let mut events = + create_test_stream(create_test_config()).into_stream(CancellationToken::new()); + events + .state + .park_with_result(Err(ReplicationError::protocol("parked failure"))); + + let err = events.next_event_before(Lsn::new(0x200)).await.unwrap_err(); + + assert!( + matches!(&err, ReplicationError::Protocol(m) if m == "parked failure"), + "{err:?}" + ); + assert!(!events.state.is_polling(), "the stream must be taken back"); + } + #[tokio::test] async fn next_raw_event_returns_seeded_wal_payload() { let payload = build_begin_payload(0x2000, 7);