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);