diff --git a/.github/workflows/ci-checks.yml b/.github/workflows/ci-checks.yml
index 5c54ab3..f8bf11b 100644
--- a/.github/workflows/ci-checks.yml
+++ b/.github/workflows/ci-checks.yml
@@ -108,6 +108,44 @@ jobs:
     - name: Unit tests (libpq / pq-sys)
       run: cargo test --lib --no-default-features --features libpq --verbose
 
+  msrv:
+    name: MSRV (Rust 1.87)
+    runs-on: ubuntu-latest
+    # The library must build on the declared rust-version. Tests are not built here:
+    # a dev-dependency (proptest 1.12) may need a newer compiler than the library does.
+    # Keep the toolchain in step with `rust-version` in Cargo.toml.
+    steps:
+    - uses: actions/checkout@v5
+
+    - name: Install Rust (MSRV)
+      uses: dtolnay/rust-toolchain@master
+      with:
+        toolchain: "1.87"
+
+    - name: Install dependencies
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y libpq-dev libssl-dev pkg-config cmake
+
+    - name: Cache cargo registry
+      uses: actions/cache@v5
+      with:
+        path: |
+          ~/.cargo/registry
+          ~/.cargo/git
+        key: ${{ runner.os }}-cargo-msrv-${{ hashFiles('**/Cargo.lock') }}
+        restore-keys: |
+          ${{ runner.os }}-cargo-msrv-
+
+    - name: Check (default features)
+      run: cargo check --locked --lib
+
+    - name: Check (no_std, no default features)
+      run: cargo check --locked --lib --no-default-features
+
+    - name: Check (libpq / pq-sys)
+      run: cargo check --locked --lib --no-default-features --features libpq
+
   examples:
     name: Examples build
     runs-on: ubuntu-latest
@@ -137,6 +175,8 @@ jobs:
         restore-keys: |
           ${{ runner.os }}-cargo-examples-
 
+    # --locked: each example has its own lock, which a root dependency-floor bump leaves
+    # stale; without the flag cargo regenerates it silently and the drift goes unseen.
     - name: Check every example
       run: |
         set -uo pipefail
@@ -146,7 +186,7 @@ jobs:
           name=$(basename "$(dirname "$manifest")")
           ran=$((ran + 1))
           echo "::group::$name"
-          cargo check --manifest-path "$manifest" --all-targets || failed+=("$name")
+          cargo check --locked --manifest-path "$manifest" --all-targets || failed+=("$name")
           echo "::endgroup::"
         done
 
diff --git a/Cargo.lock b/Cargo.lock
index cdb2747..a8f7a45 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -461,9 +461,9 @@ checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34"
 
 [[package]]
 name = "equivalent"
-version = "1.0.2"
+version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826"
 
 [[package]]
 name = "errno"
@@ -967,9 +967,9 @@ dependencies = [
 
 [[package]]
 name = "pq-sys"
-version = "0.7.6"
+version = "0.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9152654059e4fb5b66a4052acf3ee43a7a7132332e44cdb72a81ee93bd038950"
+checksum = "297a95d6f2378891f949d9556723e15140e50d29eac07458dabcee8688995b1d"
 dependencies = [
  "libc",
  "pkg-config",
diff --git a/Cargo.toml b/Cargo.toml
index 5eb0a68..9e2e5bf 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -42,7 +42,7 @@ smallvec = { version = "1.16.4", features = ["serde"] }
 pg-walstream-macros = { workspace = true, optional = true }
 
 # libpq backend (opt-in) — pq-sys: raw FFI to the system libpq via pkg-config, with pre-generated bindings (no libclang/bindgen at build time).
-pq-sys = { version = "0.7.6", optional = true }
+pq-sys = { version = "0.7.8", optional = true }
 
 # rustls-tls
 tokio-rustls = { version = "0.26.6", optional = true }
diff --git a/README.md b/README.md
index c9d9cdc..9626027 100644
--- a/README.md
+++ b/README.md
@@ -17,7 +17,7 @@ A high-performance Rust library for PostgreSQL logical and physical replication
 - **Pure-Rust Backend (default)**: The default `rustls-tls` backend needs no libpq and no OpenSSL, using `aws-lc-rs` for hardware-accelerated TLS (AES-NI, AVX2, SHA-NI). 
 - **TLS/SSL Support**: All PostgreSQL SSL modes (`disable`, `allow`, `prefer`, `require`, `verify-ca`, `verify-full`)
 - **Authentication**: Cleartext, MD5, and SCRAM-SHA-256 authentication methods
-- **Streaming Transactions**: Support for streaming large transactions (protocol v2+). Each streamed change carries the xid of its subtransaction in `ChangeEvent::xid`, so a `StreamAbort` for a rolled-back savepoint names exactly the changes to drop
+- **Streaming Transactions**: Support for streaming large transactions (protocol v2+). Each streamed change carries the xid of its subtransaction in `ChangeEvent::xid`, so a `StreamAbort` for a rolled-back savepoint names exactly the changes to drop, except a `Message`, which always carries the top-level xid
 - **Two-Phase Commit**: Prepared transaction support (protocol v3+)
 - **Parallel Streaming**: Multi-stream parallel replication (protocol v4+)
 - **Zero-Copy Operations**: Efficient buffer management using the `bytes` crate with drain-loop batch queue optimization. The libpq backend copies each COPY message into a reusable `BytesMut` buffer, then reference-counts it downstream as `Bytes` (no per-message heap allocation after warmup)
@@ -39,7 +39,7 @@ Add this to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-pg_walstream = "0.8"
+pg_walstream = "0.9"
 ```
 
 By default, this uses the pure-Rust `rustls-tls` backend — no libpq and no OpenSSL, only `cmake` + a C compiler at build time (for `aws-lc-rs`).
@@ -48,11 +48,25 @@ To use the C **libpq** backend instead (bound by `pq-sys`; requires system libpq
 
 ```toml
 [dependencies]
-pg_walstream = { version = "0.8", default-features = false, features = ["libpq"] }
+pg_walstream = { version = "0.9", default-features = false, features = ["libpq"] }
 ```
 
 If both backends are enabled, `rustls-tls` takes priority automatically.
 
+## Unreleased: changes since 0.9.0
+<!-- Rename to "Upgrading from 0.9" when the release version is chosen. -->
+
+The next release changes the public API, so code written for 0.9 may not compile unchanged.
+
+- **`ChangeEvent` has a new public field `xid: Option<u32>`.** Build events with `xid: None` and add `..` to struct patterns (`let ChangeEvent { event_type, lsn, .. } = event`). It is `Some` only for a `Relation`, `Type`, `Insert`, `Update`, `Delete`, `Truncate` or `Message` inside a streamed chunk.
+- **`ChangeEvent::encode` and `ChangeEvent::decode` (deprecated since 0.9.0) are removed.** Serialize events with `serde_json`; JSON written by 0.9 (no `xid` key) still deserializes, with `xid` as `None`. Bytes written by `encode` are not JSON and cannot be read after the upgrade, so convert them on 0.9 first. bincode and postcard cannot read a `ColumnValue`.
+- **`StreamingReplicationMessage::xid`, returned by `LogicalReplicationParser::parse_wal_message`, changes meaning inside a streamed chunk (protocol v2+).** For `Relation`, `Type`, `Insert`, `Update`, `Delete`, `Truncate` and `Message` it is now the xid pgoutput wrote before that change, which is a subtransaction's xid after a savepoint; it used to be the chunk's xid. A consumer that buffers rows by `xid` and flushes on `StreamCommit` must take the top-level xid from `StreamStart`, or it strands the savepoint's rows. A `Message` always carries the top-level xid.
+- **The `libpq` feature now also builds `socket2`,** which only `rustls-tls` needed before.
+- **`LogicalReplicationStream::new` now runs `IDENTIFY_SYSTEM`, which `start` ran before:** one more round trip at construction. A reply with no usable system identifier and timeline now fails `new` with `ReplicationError::PermanentConnection`, where `start` used to continue without it. `server_identity()` returns what it read.
+- **`send_feedback()` before `start()` returns `ReplicationError::Protocol`;** it returned `Ok(())`. After `start()` and before the first message it sends an all-zero status update instead of returning early, and `stop()` of a stream that was never started logs a warning.
+- **A cancelled command poisons the connection.** If the future of `start()`, `slot_confirmed_flush_lsn()` or `timeline_history()` is dropped before the reply arrives, every later command on that connection fails with `ReplicationError::PermanentConnection`; create a new stream.
+- **`EventStream::next_event` carries a reconnect across a dropped future.** A `timeout` or `select!` that cancels `next_event()` while the stream reconnects no longer restarts the reconnect from its first attempt; the next call resumes it, so a timeout shorter than a reconnect cannot starve it. While such a reconnect is parked, `inner()`, `inner_mut()` and `current_lsn()` panic until the next `next_event()` returns (as they already do after an abandoned `poll_next`); the feedback methods keep working.
+
 ## Feature Flags
 
 pg-walstream provides two connection backends plus a `std` toggle, all selected at compile time. `rustls-tls` is the default; `libpq` is opt-in. When both are enabled, `rustls-tls` takes priority:
@@ -97,7 +111,7 @@ sudo apt-get install cmake gcc
 Then add to `Cargo.toml`:
 
 ```toml
-pg_walstream = { version = "0.8", features = ["rustls-tls"] }
+pg_walstream = { version = "0.9", features = ["rustls-tls"] }
 ```
 
 #### TLS trust store
@@ -313,7 +327,11 @@ let mut copy = loop {
 };
 let c = copy.consistent_point();
 while let Some(event) = stream.next_event_before(c, &cancel).await? {
-    sink.buffer(event); // per transaction, decided at its commit event
+    let lsn = event.lsn;
+    // per transaction, decided at its commit event; true once that transaction is applied
+    if sink.buffer(event).await? {
+        stream.shared_lsn_feedback.update_applied_lsn(lsn);
+    }
 }
 let mut fed = std::time::Instant::now();
 while let Some(row) = copy.next_row().await? {
@@ -327,9 +345,13 @@ copy.finish()?;
 // stream.next_event(&cancel) continues at the first transaction committed at or past c.
 ```
 
-`next_event_before(C)` returns `None` once every transaction committed below `C` has been delivered. Decide each transaction at its commit event. Below `C`, apply it without the copied tables' rows, which the copy already holds. At or past `C`, apply it in full, after the copy. A transaction that ends in `RollbackPrepared` or in a `StreamAbort` of the whole transaction is dropped. A `StreamAbort` of a subtransaction drops the buffered changes whose `ChangeEvent::xid` equals its `subtransaction_xid`. Streamed chunks, `BeginPrepare` through `Prepare`, and `StreamPrepare` can arrive before the `None` for a transaction that commits at or past `C`, so buffer each transaction until its commit event. `EventStream::next_event_before` does the same with retries.
+`next_event_before(C)` returns `None` once every transaction committed below `C` has been delivered. Decide each transaction at its commit event. Below `C`, apply it without the copied tables' rows, which the copy already holds. At or past `C`, apply it in full, after the copy. A transaction that ends in `RollbackPrepared` or in a `StreamAbort` of the whole transaction is dropped. A `StreamAbort` of a subtransaction drops the buffered changes whose `ChangeEvent::xid` equals its `subtransaction_xid`, except a `Message`, which always carries the top-level xid. Streamed chunks, `BeginPrepare` through `Prepare`, and `StreamPrepare` can arrive before the `None` for a transaction that commits at or past `C`, so buffer each transaction until its commit event. `EventStream::next_event_before` does the same with retries.
+
+Acknowledge each transaction you decided below `C` (`update_applied_lsn` with its commit event's `lsn`, or `EventStream::update_applied_lsn`) before the copy loads. A reconnect replays from the acknowledged position, so without it every transaction delivered before the `None` arrives again after the copy and is applied twice: 50 of 50 transactions in a measured idle-database run, none with the acknowledgement. `EventStream::next_event_before` reconnects without telling you, so also start a buffered transaction over when a `Begin`, a `BeginPrepare` or a `StreamStart` with `first_segment` arrives for one you already hold. That replay of a streamed transaction was not reproduced, so treat it as a precaution.
+
+The capture holds a second replication connection and a temporary slot until slot creation ends. `capture_aligned` fails with a permanent error, before creating the slot, if that connection reaches another cluster or timeline than the stream's, which a connection string with several hosts can cause. Slot creation waits for the transactions open when it starts and can also wait for ones that start during the wait, a prepared transaction left open included. `capture` has no deadline of its own: end the blocker, or cancel the walsender from a second connection with `SELECT pg_cancel_backend(active_pid) FROM pg_replication_slots WHERE slot_name LIKE 'pg_walstream_aligned_%'`. Dropping the capture future closes its connections, but a slot creation already sent keeps its walsender waiting on the server, so abandoned captures behind a long transaction can exhaust `max_wal_senders`. `with_cancellation` covers only the copy loop, and `next_row()` is not cancel-safe where a table's `COPY` starts, so do not race it in a `select!` that can drop it.
 
-The capture holds a second replication connection and a temporary slot until slot creation ends. Slot creation waits for the transactions open when it starts and can also wait for ones that start during the wait. Dropping the capture future closes its connections, but a slot creation already sent keeps its walsender waiting on the server, so abandoned captures behind a long transaction can exhaust `max_wal_senders`. `with_cancellation` covers only the copy loop. The copy pins `xmin` until `finish()`.
+The copy pins `xmin` until `finish()`, and it holds an `ACCESS SHARE` lock on each copied table from `capture` on. Reads and writes go on, but a `TRUNCATE`, `ALTER TABLE`, `DROP`, `VACUUM FULL` or `CLUSTER` on those tables waits until `finish()` or drop, and writes issued behind a waiting one wait with it (an `ALTER TABLE` waited 2.4 s, an `INSERT` behind it failed with `55P03` at its 2 s `lock_timeout`). Give that DDL a `lock_timeout`, or run it before the capture or after the copy. The lock is what keeps the copy exact: a `COPY` reads the table's current file, so a `TRUNCATE` or rewriting `ALTER TABLE` committed after `C` would otherwise return it empty. If such a command is in flight when `capture` asks for its lock, or committed after `C`, `capture` fails and you capture again.
 
 ## Connection Timeouts
 
diff --git a/examples/arbitrary-fuzzing/Cargo.lock b/examples/arbitrary-fuzzing/Cargo.lock
index 7c6f11e..dd89a76 100644
--- a/examples/arbitrary-fuzzing/Cargo.lock
+++ b/examples/arbitrary-fuzzing/Cargo.lock
@@ -254,9 +254,9 @@ checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
diff --git a/examples/basic-streaming/Cargo.lock b/examples/basic-streaming/Cargo.lock
index 52dad4e..aa38e7b 100644
--- a/examples/basic-streaming/Cargo.lock
+++ b/examples/basic-streaming/Cargo.lock
@@ -28,9 +28,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -39,9 +39,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -789,9 +789,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -1096,9 +1096,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/binary-column-access/Cargo.lock b/examples/binary-column-access/Cargo.lock
index 70b9ea0..dfe967e 100644
--- a/examples/binary-column-access/Cargo.lock
+++ b/examples/binary-column-access/Cargo.lock
@@ -19,9 +19,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -30,9 +30,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -588,9 +588,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -825,9 +825,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/derive-router/Cargo.lock b/examples/derive-router/Cargo.lock
index f99e011..279a6ee 100644
--- a/examples/derive-router/Cargo.lock
+++ b/examples/derive-router/Cargo.lock
@@ -30,9 +30,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -41,9 +41,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -667,9 +667,9 @@ dependencies = [
 
 [[package]]
 name = "pq-sys"
-version = "0.7.6"
+version = "0.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9152654059e4fb5b66a4052acf3ee43a7a7132332e44cdb72a81ee93bd038950"
+checksum = "297a95d6f2378891f949d9556723e15140e50d29eac07458dabcee8688995b1d"
 dependencies = [
  "libc",
  "pkg-config",
@@ -865,9 +865,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -1185,9 +1185,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/initial-snapshot/Cargo.lock b/examples/initial-snapshot/Cargo.lock
index e731744..a03d189 100644
--- a/examples/initial-snapshot/Cargo.lock
+++ b/examples/initial-snapshot/Cargo.lock
@@ -28,9 +28,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -39,9 +39,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -734,9 +734,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -1021,9 +1021,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/pg-basebackup/Cargo.lock b/examples/pg-basebackup/Cargo.lock
index 7bd1359..7a9f70a 100644
--- a/examples/pg-basebackup/Cargo.lock
+++ b/examples/pg-basebackup/Cargo.lock
@@ -34,9 +34,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -45,9 +45,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -747,9 +747,9 @@ checksum = "e320a6c5ad31d271ad523dcf3ad13e2767ad8b1cb8f047f75a8aeaf8da139da2"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -1044,9 +1044,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/polling/Cargo.lock b/examples/polling/Cargo.lock
index 46e509c..352eb62 100644
--- a/examples/polling/Cargo.lock
+++ b/examples/polling/Cargo.lock
@@ -28,9 +28,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -39,9 +39,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -705,9 +705,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -1002,9 +1002,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/rate-limited-streaming/Cargo.lock b/examples/rate-limited-streaming/Cargo.lock
index e63eaf9..a34c45d 100644
--- a/examples/rate-limited-streaming/Cargo.lock
+++ b/examples/rate-limited-streaming/Cargo.lock
@@ -28,9 +28,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -39,9 +39,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -789,9 +789,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -1096,9 +1096,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/raw-xlogdata/Cargo.lock b/examples/raw-xlogdata/Cargo.lock
index 77d02e5..199abcf 100644
--- a/examples/raw-xlogdata/Cargo.lock
+++ b/examples/raw-xlogdata/Cargo.lock
@@ -28,9 +28,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -39,9 +39,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -730,9 +730,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -1027,9 +1027,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/safe-transaction-consumer/Cargo.lock b/examples/safe-transaction-consumer/Cargo.lock
index f438010..419919c 100644
--- a/examples/safe-transaction-consumer/Cargo.lock
+++ b/examples/safe-transaction-consumer/Cargo.lock
@@ -28,9 +28,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -39,9 +39,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -698,9 +698,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -994,9 +994,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/tokio-spawn-streaming/Cargo.lock b/examples/tokio-spawn-streaming/Cargo.lock
index a805e92..e64b270 100644
--- a/examples/tokio-spawn-streaming/Cargo.lock
+++ b/examples/tokio-spawn-streaming/Cargo.lock
@@ -28,9 +28,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -39,9 +39,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -688,9 +688,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -994,9 +994,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/examples/typed-deserialization/Cargo.lock b/examples/typed-deserialization/Cargo.lock
index a0cd9ec..3d15251 100644
--- a/examples/typed-deserialization/Cargo.lock
+++ b/examples/typed-deserialization/Cargo.lock
@@ -36,9 +36,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -47,9 +47,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -712,9 +712,9 @@ dependencies = [
 
 [[package]]
 name = "pq-sys"
-version = "0.7.6"
+version = "0.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9152654059e4fb5b66a4052acf3ee43a7a7132332e44cdb72a81ee93bd038950"
+checksum = "297a95d6f2378891f949d9556723e15140e50d29eac07458dabcee8688995b1d"
 dependencies = [
  "libc",
  "pkg-config",
@@ -933,9 +933,9 @@ checksum = "b2aa850e253778c88a04c3d7323b043aeda9d3e30d5971937c1855769763678e"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
@@ -1316,9 +1316,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.10"
+version = "1.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1c8e20c8a38fc7d65abfe74eb7280e4aa667e8eec8bc84996d2874810748d9"
+checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171"
 dependencies = [
  "rustls-pki-types",
 ]
diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock
index 070dd99..33d7c05 100644
--- a/fuzz/Cargo.lock
+++ b/fuzz/Cargo.lock
@@ -330,9 +330,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
diff --git a/integration-tests/aligned_snapshot.rs b/integration-tests/aligned_snapshot.rs
index d323ef4..c9bbc42 100644
--- a/integration-tests/aligned_snapshot.rs
+++ b/integration-tests/aligned_snapshot.rs
@@ -736,6 +736,19 @@ impl Recording {
         }
     }
 
+    /// Rows inserted by an event delivered before the `None` that the copy holds too.
+    fn overlap(&self) -> Vec<i64> {
+        let copied: std::collections::BTreeSet<i64> =
+            self.copy.iter().map(|(row, _)| *row).collect();
+        self.before
+            .iter()
+            .filter_map(|e| match step(e.event_type.clone(), self.prefix) {
+                Step::Change(row, Some(_), false) if copied.contains(&row) => Some(row),
+                _ => None,
+            })
+            .collect()
+    }
+
     fn changes_before(&self) -> usize {
         self.before
             .iter()
@@ -885,14 +898,27 @@ async fn a_table_added_to_a_running_stream_merges_exactly_by_the_documented_flow
 }
 
 /// Applying the changes delivered before the `None` and then the copy feeds a
-/// strict sink rows it already holds, or changes to rows it lacks.
+/// strict sink a row twice: a row the writer inserted below `C` comes from the
+/// stream and again from the copy. The updates to the 20 rows from before the
+/// writer started, and the deletes of rows inserted before the table joined the
+/// publication, are out of step for want of a baseline whatever the overlap, so
+/// only a row in both says the overlap was exercised.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 #[ignore = "requires a live PostgreSQL with wal_level = logical"]
 async fn a_strict_sink_fed_the_overlap_and_then_the_copy_goes_out_of_step() {
     tokio::time::timeout(TEST_TIMEOUT, async {
         let (_guard, _admin, recording) =
             table_added_to_a_running_stream("aligned_naive", "aligned_naive_main").await;
-        assert!(!recording.merge_naively().out_of_step.is_empty());
+        let overlap = recording.overlap();
+        assert!(
+            !overlap.is_empty(),
+            "the stream and the copy never held the same insert"
+        );
+        let out_of_step = recording.merge_naively().out_of_step;
+        assert!(
+            overlap.iter().any(|row| out_of_step.contains(row)),
+            "{out_of_step:?}"
+        );
     })
     .await
     .expect("test timed out");
@@ -1056,20 +1082,19 @@ async fn a_dropped_capture_leaves_no_walsender_or_slot_once_the_blocker_ends() {
         exec(&mut blocker, "BEGIN");
         let blocking = xid(&mut blocker);
 
-        let abandoned = tokio::time::timeout(
-            Duration::from_millis(500),
-            AlignedSnapshot::capture(
-                &replication_conn_string(),
-                &["aligned_dropped_pub"],
-                &[("public", "aligned_dropped")],
-            ),
-        )
-        .await;
-        assert!(abandoned.is_err(), "the capture did not wait for the open transaction");
-        eventually("the abandoned setup waits for the blocker", || {
-            walsender_waits_for(&mut admin, blocking)
-        })
-        .await;
+        let conninfo = replication_conn_string();
+        let mut capture = Box::pin(AlignedSnapshot::capture(
+            &conninfo,
+            &["aligned_dropped_pub"],
+            &[("public", "aligned_dropped")],
+        ));
+        tokio::select! {
+            _ = &mut capture => panic!("the capture did not wait for the open transaction"),
+            () = eventually("slot creation waits for the blocker", || {
+                walsender_waits_for(&mut admin, blocking)
+            }) => {}
+        }
+        drop(capture);
         let walsender = count(&exec(
             &mut admin,
             "SELECT active_pid FROM pg_replication_slots WHERE slot_name LIKE 'pg_walstream_aligned_%'",
@@ -1091,3 +1116,214 @@ async fn a_dropped_capture_leaves_no_walsender_or_slot_once_the_blocker_ends() {
     .await
     .expect("test timed out");
 }
+
+/// A table whose `COPY` would read an empty new file instead of the snapshot's rows
+/// (`TRUNCATE`, a rewriting `ALTER TABLE`, a drop) cannot change while the copy is open.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[ignore = "requires a live PostgreSQL with wal_level = logical"]
+async fn ddl_that_would_empty_the_copy_waits_for_it_instead() {
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let mut admin = regular_conn();
+        exec(&mut admin, "SET lock_timeout = '500ms'");
+        for ddl in [
+            "TRUNCATE aligned_ddl",
+            "ALTER TABLE aligned_ddl ADD COLUMN extra double precision DEFAULT random()",
+            "ALTER TABLE aligned_ddl ALTER COLUMN v TYPE int USING length(v)",
+            "DROP TABLE aligned_ddl",
+        ] {
+            for sql in [
+                "DROP PUBLICATION IF EXISTS aligned_ddl_pub",
+                "DROP TABLE IF EXISTS aligned_ddl",
+                "CREATE TABLE aligned_ddl (id bigint PRIMARY KEY, v text NOT NULL)",
+                "INSERT INTO aligned_ddl SELECT g, 'orig' FROM generate_series(1, 100) g",
+                "CREATE PUBLICATION aligned_ddl_pub FOR TABLE aligned_ddl",
+            ] {
+                exec(&mut admin, sql);
+            }
+            let mut snapshot = AlignedSnapshot::capture(
+                &replication_conn_string(),
+                &["aligned_ddl_pub"],
+                &[("public", "aligned_ddl")],
+            )
+            .await
+            .expect("capture");
+
+            let blocked = admin.exec(ddl).err().unwrap_or_else(|| {
+                panic!("`{ddl}` committed while the copy was open, so the copy reads the new table")
+            });
+            assert!(
+                blocked.to_string().contains("lock timeout"),
+                "{ddl}: {blocked}"
+            );
+
+            let mut rows = 0;
+            while snapshot.next_row().await.expect("copy").is_some() {
+                rows += 1;
+            }
+            assert_eq!(rows, 100, "after `{ddl}`");
+            snapshot.finish().expect("finish");
+            exec(&mut admin, ddl);
+        }
+    })
+    .await
+    .expect("test timed out");
+}
+
+/// A writer that truncates and refills the table in one transaction, over and over, never leaves it
+/// empty for a snapshot, so a copy of 0 rows is the silent loss.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[ignore = "requires a live PostgreSQL with wal_level = logical"]
+async fn a_table_truncated_around_the_capture_is_never_copied_silently_empty() {
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let mut admin = regular_conn();
+        for sql in [
+            "DROP PUBLICATION IF EXISTS aligned_storm_pub",
+            "DROP TABLE IF EXISTS aligned_storm",
+            "CREATE TABLE aligned_storm (id bigint PRIMARY KEY, v text NOT NULL)",
+            "INSERT INTO aligned_storm SELECT g, 'orig' FROM generate_series(1, 100) g",
+            "CREATE PUBLICATION aligned_storm_pub FOR TABLE aligned_storm",
+        ] {
+            exec(&mut admin, sql);
+        }
+        let stop = Arc::new(AtomicBool::new(false));
+        let writer = {
+            let stop = Arc::clone(&stop);
+            std::thread::spawn(move || {
+                let mut w = regular_conn();
+                let _ = w.exec("SET lock_timeout = '2s'");
+                while !stop.load(Ordering::Relaxed) {
+                    // A refused lock rolls the transaction back and the next one retries.
+                    if w.exec(
+                        "BEGIN; TRUNCATE aligned_storm; \
+                         INSERT INTO aligned_storm SELECT g, 'n' FROM generate_series(1, 100) g; COMMIT",
+                    )
+                    .is_err()
+                    {
+                        let _ = w.exec("ROLLBACK");
+                    }
+                    std::thread::sleep(Duration::from_millis(20));
+                }
+            })
+        };
+
+        let (mut copied, mut refused) = (0, 0);
+        for _ in 0..15 {
+            match AlignedSnapshot::capture(
+                &replication_conn_string(),
+                &["aligned_storm_pub"],
+                &[("public", "aligned_storm")],
+            )
+            .await
+            {
+                Ok(mut snapshot) => {
+                    let mut rows = 0;
+                    while snapshot.next_row().await.expect("copy").is_some() {
+                        rows += 1;
+                    }
+                    assert_eq!(rows, 100, "a copy that is neither exact nor an error");
+                    snapshot.finish().expect("finish");
+                    copied += 1;
+                }
+                Err(e) => {
+                    let message = e.to_string();
+                    assert!(
+                        message.contains("changed after the snapshot")
+                            || message.contains("could not obtain lock"),
+                        "{message}"
+                    );
+                    refused += 1;
+                }
+            }
+        }
+        stop.store(true, Ordering::Relaxed);
+        writer.join().unwrap();
+        println!("captures: {copied} exact, {refused} refused");
+
+        let mut snapshot = AlignedSnapshot::capture(
+            &replication_conn_string(),
+            &["aligned_storm_pub"],
+            &[("public", "aligned_storm")],
+        )
+        .await
+        .expect("a quiet table captures");
+        let mut rows = 0;
+        while snapshot.next_row().await.expect("copy").is_some() {
+            rows += 1;
+        }
+        assert_eq!(rows, 100);
+        snapshot.finish().expect("finish");
+    })
+    .await
+    .expect("test timed out");
+}
+
+/// The same race on a partitioned table published through its root. The root has no file of its own,
+/// so only a check of its partitions sees a leaf that was truncated after the snapshot.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[ignore = "requires a live PostgreSQL with wal_level = logical"]
+async fn a_leaf_truncated_around_the_capture_is_never_copied_silently_short() {
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let mut admin = regular_conn();
+        for sql in [
+            "DROP PUBLICATION IF EXISTS aligned_leaf_pub",
+            "DROP TABLE IF EXISTS aligned_leaf CASCADE",
+            "CREATE TABLE aligned_leaf (id bigint PRIMARY KEY, v text NOT NULL) PARTITION BY RANGE (id)",
+            "CREATE TABLE aligned_leaf_a PARTITION OF aligned_leaf FOR VALUES FROM (0) TO (100)",
+            "CREATE TABLE aligned_leaf_b PARTITION OF aligned_leaf FOR VALUES FROM (100) TO (200)",
+            "INSERT INTO aligned_leaf SELECT g, 'orig' FROM generate_series(0, 199) g",
+            "CREATE PUBLICATION aligned_leaf_pub FOR TABLE aligned_leaf WITH (publish_via_partition_root = true)",
+        ] {
+            exec(&mut admin, sql);
+        }
+        let stop = Arc::new(AtomicBool::new(false));
+        let writer = {
+            let stop = Arc::clone(&stop);
+            std::thread::spawn(move || {
+                let mut w = regular_conn();
+                let _ = w.exec("SET lock_timeout = '2s'");
+                while !stop.load(Ordering::Relaxed) {
+                    if w.exec(
+                        "BEGIN; TRUNCATE aligned_leaf_a; \
+                         INSERT INTO aligned_leaf_a SELECT g, 'n' FROM generate_series(0, 99) g; COMMIT",
+                    )
+                    .is_err()
+                    {
+                        let _ = w.exec("ROLLBACK");
+                    }
+                    std::thread::sleep(Duration::from_millis(20));
+                }
+            })
+        };
+
+        for _ in 0..15 {
+            match AlignedSnapshot::capture(
+                &replication_conn_string(),
+                &["aligned_leaf_pub"],
+                &[("public", "aligned_leaf")],
+            )
+            .await
+            {
+                Ok(mut snapshot) => {
+                    let mut rows = 0;
+                    while snapshot.next_row().await.expect("copy").is_some() {
+                        rows += 1;
+                    }
+                    assert_eq!(rows, 200, "a copy that is neither exact nor an error");
+                    snapshot.finish().expect("finish");
+                }
+                Err(e) => {
+                    let message = e.to_string();
+                    assert!(
+                        message.contains("changed after the snapshot")
+                            || message.contains("could not obtain lock"),
+                        "{message}"
+                    );
+                }
+            }
+        }
+        stop.store(true, Ordering::Relaxed);
+        writer.join().unwrap();
+    })
+    .await
+    .expect("test timed out");
+}
diff --git a/integration-tests/initial_snapshot.rs b/integration-tests/initial_snapshot.rs
index 2319f87..a883a47 100644
--- a/integration-tests/initial_snapshot.rs
+++ b/integration-tests/initial_snapshot.rs
@@ -689,11 +689,6 @@ async fn snapshot_cancelled_during_slot_creation_leaves_no_slot() {
     )
     .await
     .expect("connect");
-    let cancelled = tokio::time::timeout(Duration::from_millis(500), stream.snapshot()).await;
-    assert!(
-        cancelled.is_err(),
-        "slot creation must still be waiting on the open transaction"
-    );
     let mut probe = regular_conn();
     let count = |probe: &mut PgReplicationConnection| {
         probe_count(
@@ -704,11 +699,20 @@ async fn snapshot_cancelled_during_slot_creation_leaves_no_slot() {
                 .expect("count slots"),
         )
     };
-    assert_eq!(
-        count(&mut probe),
-        1,
-        "the server holds the half-created slot while it waits"
-    );
+    // Cancel only once the server holds the half-created slot, however long the connect takes.
+    let mut snapshot = Box::pin(stream.snapshot());
+    let deadline = std::time::Instant::now() + Duration::from_secs(30);
+    while count(&mut probe) != 1 {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "the server never held the half-created slot while it waited"
+        );
+        tokio::select! {
+            _ = &mut snapshot => panic!("slot creation must still be waiting on the open transaction"),
+            () = tokio::time::sleep(Duration::from_millis(50)) => {}
+        }
+    }
+    drop(snapshot);
     blocker.exec("COMMIT").expect("commit");
 
     let deadline = std::time::Instant::now() + Duration::from_secs(30);
diff --git a/integration-tests/reconnect_recovery.rs b/integration-tests/reconnect_recovery.rs
index 979f8d5..c801498 100644
--- a/integration-tests/reconnect_recovery.rs
+++ b/integration-tests/reconnect_recovery.rs
@@ -5,7 +5,8 @@
 //!
 //! `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.
+//! and `EventStream::next_event` under `tokio::time::timeout`, the way a consumer drives it,
+//! including a timeout shorter than the reconnect itself.
 //!
 //! Requires PostgreSQL with `wal_level=logical`, `DATABASE_URL` in replication mode and,
 //! optionally, `DATABASE_URL_REGULAR` for the same database.
@@ -16,6 +17,8 @@ use pg_walstream::{
     ReplicationStreamConfig,
 };
 use std::time::Instant;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
 use tokio_util::sync::CancellationToken;
 
 /// Generous: a reconnect sits out a 1 s backoff, and a loaded CI runner is slower still.
@@ -85,6 +88,28 @@ impl Fixture {
         ));
     }
 
+    /// Wait until the slot exists on the server: a command the client cancelled still runs there.
+    fn wait_for_slot(&mut self) {
+        let sql = format!(
+            "SELECT 1 FROM pg_replication_slots WHERE slot_name = '{}'",
+            self.slot
+        );
+        let started = Instant::now();
+        while started.elapsed() < Duration::from_secs(10) {
+            if self
+                .regular
+                .exec(&sql)
+                .ok()
+                .and_then(|r| r.get_value(0, 0))
+                .is_some()
+            {
+                return;
+            }
+            std::thread::sleep(Duration::from_millis(20));
+        }
+        panic!("the slot was never created");
+    }
+
     fn insert(&mut self, id: i32) {
         self.regular
             .exec(&format!("INSERT INTO {} VALUES ({id})", self.slot))
@@ -170,17 +195,21 @@ async fn read_until(stream: &mut LogicalReplicationStream, until: i32) -> Vec<i3
     .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<i32> {
+/// [`read_until`] through `EventStream::next_event`, each read cut off after `slice`, 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, slice: Duration) -> Vec<i32> {
     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");
+        assert!(
+            !left.is_zero(),
+            "row {until} never arrived after the reconnect"
+        );
+        let Ok(event) = tokio::time::timeout(left.min(slice), events.next_event()).await else {
+            continue;
+        };
+        let event = event.expect("the stream must recover, not give up");
         match event.event_type {
             EventType::Insert { data, .. } => ids.push(inserted_id(&data)),
             EventType::Commit { end_lsn, .. } => {
@@ -194,6 +223,88 @@ async fn read_until_timed(events: &mut EventStream, until: i32) -> Vec<i32> {
     }
 }
 
+/// A TCP proxy to PostgreSQL that holds every reply for `latency`, as a slow link would. A
+/// connect, `IDENTIFY_SYSTEM` and `START_REPLICATION` are several round trips, so a
+/// reconnect through it takes a good part of a second.
+struct SlowLink {
+    port: u16,
+    accept_loop: tokio::task::JoinHandle<()>,
+}
+
+impl SlowLink {
+    async fn to(server: String, latency: Duration) -> Self {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind proxy");
+        let port = listener.local_addr().expect("proxy address").port();
+        let accept_loop = tokio::spawn(async move {
+            while let Ok((client, _)) = listener.accept().await {
+                let server = server.clone();
+                tokio::spawn(async move {
+                    let Ok(upstream) = TcpStream::connect(server).await else {
+                        return;
+                    };
+                    let (mut from_client, mut to_client) = client.into_split();
+                    let (mut from_server, mut to_server) = upstream.into_split();
+                    let forward = tokio::spawn(async move {
+                        let _ = tokio::io::copy(&mut from_client, &mut to_server).await;
+                    });
+                    // Stamp each reply on arrival and release it `latency` later, in order.
+                    let (queue, mut delayed) = tokio::sync::mpsc::unbounded_channel();
+                    let read = async move {
+                        let mut buf = [0u8; 8192];
+                        while let Ok(n) = from_server.read(&mut buf).await {
+                            if n == 0
+                                || queue
+                                    .send((
+                                        tokio::time::Instant::now() + latency,
+                                        buf[..n].to_vec(),
+                                    ))
+                                    .is_err()
+                            {
+                                break;
+                            }
+                        }
+                    };
+                    let write = async move {
+                        while let Some((due, chunk)) = delayed.recv().await {
+                            tokio::time::sleep_until(due).await;
+                            if to_client.write_all(&chunk).await.is_err() {
+                                break;
+                            }
+                        }
+                        let _ = to_client.shutdown().await;
+                    };
+                    tokio::join!(read, write);
+                    forward.abort();
+                });
+            }
+        });
+        Self { port, accept_loop }
+    }
+}
+
+impl Drop for SlowLink {
+    fn drop(&mut self) {
+        self.accept_loop.abort();
+    }
+}
+
+/// The `host[:port]` of a connection URL, as written.
+fn authority_of(url: &str) -> &str {
+    let after_scheme = url.split_once("://").map_or(url, |(_, rest)| rest);
+    let authority = after_scheme.split('/').next().unwrap_or(after_scheme);
+    authority.rsplit('@').next().unwrap_or(authority)
+}
+
+/// `host:port` to dial for a connection URL: its default port when it names none.
+fn server_of(url: &str) -> String {
+    let server = authority_of(url);
+    if server.contains(':') {
+        server.to_string()
+    } else {
+        format!("{server}:5432")
+    }
+}
+
 /// 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]
@@ -237,14 +348,14 @@ async fn an_event_stream_under_timeout_survives_repeated_terminations() {
     let mut events = stream.into_stream(CancellationToken::new());
 
     fixture.insert(1);
-    assert_eq!(read_until_timed(&mut events, 1).await, [1]);
+    assert_eq!(read_until_timed(&mut events, 1, READ_TIMEOUT).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,
+            read_until_timed(&mut events, id, READ_TIMEOUT).await,
             [id],
             "row {id} exactly once"
         );
@@ -253,3 +364,128 @@ async fn an_event_stream_under_timeout_survives_repeated_terminations() {
         previous = current;
     }
 }
+
+/// A reconnect that takes longer than the consumer's read timeout still completes. The link
+/// holds every reply for 120 ms, so a reconnect takes several hundred milliseconds and each
+/// read is cut off at 200 ms. A reconnect that every cut-off read restarts from its first
+/// connect attempt never gets there, and row 2 never arrives.
+#[tokio::test]
+#[ignore = "requires live PostgreSQL with wal_level=logical"]
+async fn a_reconnect_slower_than_the_read_timeout_still_completes() {
+    let mut fixture = Fixture::new("it_reconnect_slow_link");
+    let url = replication_conn_string();
+    let link = SlowLink::to(server_of(&url), Duration::from_millis(120)).await;
+    let url = url.replacen(authority_of(&url), &format!("127.0.0.1:{}", link.port), 1);
+    let mut stream = LogicalReplicationStream::new(&url, fixture.config())
+        .await
+        .expect("stream");
+    stream.start(None).await.expect("start");
+    let mut events = stream.into_stream(CancellationToken::new());
+    let slice = Duration::from_millis(200);
+
+    fixture.insert(1);
+    assert_eq!(read_until_timed(&mut events, 1, slice).await, [1]);
+    let before = fixture.walsender_pid();
+
+    fixture.terminate_walsender();
+    fixture.insert(2);
+
+    assert_eq!(
+        read_until_timed(&mut events, 2, slice).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"
+    );
+}
+
+/// A `start` that a timeout cancelled leaves the connection refusing every command, with an
+/// error that reads as retryable. The stream must say it is to be rebuilt (a permanent error,
+/// so a caller's retry loop stops), while a read still reconnects it and delivers the row.
+#[tokio::test]
+#[ignore = "requires live PostgreSQL with wal_level=logical"]
+async fn a_cancelled_start_is_permanent_but_a_read_still_reconnects() {
+    let mut fixture = Fixture::new("it_reconnect_cancelled_start");
+    let url = replication_conn_string();
+    let link = SlowLink::to(server_of(&url), Duration::from_millis(120)).await;
+    let url = url.replacen(authority_of(&url), &format!("127.0.0.1:{}", link.port), 1);
+    let mut stream = LogicalReplicationStream::new(&url, fixture.config())
+        .await
+        .expect("stream");
+
+    // A start takes several round trips of 120 ms each.
+    let started = tokio::time::timeout(Duration::from_millis(50), stream.start(None)).await;
+    assert!(started.is_err(), "start must outlast the 50 ms timeout");
+
+    let err = stream.start(None).await.unwrap_err();
+    assert!(
+        err.is_permanent(),
+        "a retry loop must learn to rebuild the stream, got {err:?}"
+    );
+
+    fixture.wait_for_slot();
+    fixture.insert(1);
+    assert_eq!(read_until(&mut stream, 1).await, [1]);
+}
+
+/// A reconnect that a timeout cut off, resumed by the next call, still lets a keepalive at or
+/// past the boundary end `next_event_before`. The parked read must stay bounded: unbounded, it
+/// ignores the keepalive and waits for a row, which an idle database never sends.
+#[tokio::test]
+#[ignore = "requires live PostgreSQL with wal_level=logical"]
+async fn a_parked_reconnect_still_ends_a_bounded_read_on_a_keepalive() {
+    let mut fixture = Fixture::new("it_reconnect_bounded");
+    let url = replication_conn_string();
+    let server = server_of(&url);
+    let link = SlowLink::to(server.clone(), Duration::from_millis(120)).await;
+    let url = url.replacen(&server, &format!("127.0.0.1:{}", link.port), 1);
+    let mut stream = LogicalReplicationStream::new(&url, 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, READ_TIMEOUT).await, [1]);
+
+    fixture.terminate_walsender();
+    // Written after the kill: only the new walsender can report a position at or past it.
+    fixture
+        .regular
+        .exec(&format!(
+            "DROP TABLE IF EXISTS {0}_other; \
+             CREATE TABLE {0}_other (id integer); INSERT INTO {0}_other VALUES (1)",
+            fixture.slot
+        ))
+        .expect("unpublished write");
+    let boundary = fixture
+        .regular
+        .exec("SELECT (pg_current_wal_lsn() - '0/0'::pg_lsn)::bigint")
+        .expect("wal position")
+        .get_value(0, 0)
+        .expect("a position")
+        .parse::<u64>()
+        .expect("integer position");
+    let boundary = pg_walstream::Lsn::new(boundary);
+
+    // The reconnect takes several 120 ms round trips: the first call is cut off inside it,
+    // so the second resumes it from the parked read.
+    let cut = tokio::time::timeout(
+        Duration::from_millis(500),
+        events.next_event_before(boundary),
+    )
+    .await;
+    assert!(cut.is_err(), "the reconnect must outlast the timeout");
+
+    let ended = tokio::time::timeout(READ_TIMEOUT, events.next_event_before(boundary))
+        .await
+        .expect("a keepalive past the boundary must end the read; it waited for a row instead")
+        .expect("read");
+    assert!(ended.is_none(), "nothing is committed past the boundary");
+    drop(events);
+    let _ = fixture
+        .regular
+        .exec(&format!("DROP TABLE IF EXISTS {}_other", fixture.slot));
+}
diff --git a/load-tests/Cargo.lock b/load-tests/Cargo.lock
index b692708..33fbd58 100644
--- a/load-tests/Cargo.lock
+++ b/load-tests/Cargo.lock
@@ -42,9 +42,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.18.2"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13"
+checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -53,9 +53,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.46.0"
+version = "0.47.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc"
+checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52"
 dependencies = [
  "cc",
  "cmake",
@@ -896,9 +896,9 @@ dependencies = [
 
 [[package]]
 name = "pq-sys"
-version = "0.7.6"
+version = "0.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9152654059e4fb5b66a4052acf3ee43a7a7132332e44cdb72a81ee93bd038950"
+checksum = "297a95d6f2378891f949d9556723e15140e50d29eac07458dabcee8688995b1d"
 dependencies = [
  "libc",
  "pkg-config",
@@ -1159,9 +1159,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
 
 [[package]]
 name = "smallvec"
-version = "1.16.3"
+version = "1.16.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b"
+checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8"
 dependencies = [
  "serde",
 ]
diff --git a/src/connection/libpq.rs b/src/connection/libpq.rs
index 71924d5..5f1a681 100644
--- a/src/connection/libpq.rs
+++ b/src/connection/libpq.rs
@@ -1047,6 +1047,8 @@ impl PgReplicationConnection {
         self.reject_in_flight()?;
         let c_query = CString::new(query)
             .map_err(|e| ReplicationError::protocol(format!("Invalid query string: {e}")))?;
+        // The socket is registered for this command and released after it unless a COPY holds a registration: a connection built under one runtime may be driven by another, and an `AsyncFd` stays bound to the reactor it was registered with.
+        let registered_here = self.async_fd.is_none();
         self.initialize_async_socket()?;
 
         // Raised before the first await and lowered on every exit path; a dropped
@@ -1054,6 +1056,9 @@ impl PgReplicationConnection {
         self.in_flight = true;
         let result = self.exec_in_flight(&c_query, query).await;
         self.in_flight = false;
+        if registered_here {
+            self.async_fd = None;
+        }
         result
     }
 
@@ -3155,6 +3160,9 @@ mod tests {
 
     /// A silent server must neither block the runtime nor keep the socket open once the caller gives up,
     /// and the startup packet must have reached it before the connect gave up.
+    ///
+    /// The caller gives up when the server reports the startup packet, not after a fixed time: a time limit
+    /// that is too short for a slow or stalled thread cancels the connect before it has sent anything.
     #[tokio::test(flavor = "current_thread")]
     async fn connect_async_yields_to_a_timeout_and_releases_its_socket() {
         use std::sync::atomic::{AtomicUsize, Ordering};
@@ -3166,30 +3174,37 @@ mod tests {
         let conninfo = format!(
             "host=127.0.0.1 port={port} dbname=async user=async sslmode=disable gssencmode=disable connect_timeout=10"
         );
+        let (startup_seen, startup_packet) = tokio::sync::oneshot::channel();
         let silent_server = tokio::spawn(async move {
             let (mut sock, _) = listener.accept().await.unwrap();
-            let mut startup = Vec::new();
-            sock.read_to_end(&mut startup).await.unwrap();
+            let mut len = [0u8; 4];
+            sock.read_exact(&mut len).await.unwrap();
+            let mut startup = vec![0; (u32::from_be_bytes(len) as usize).saturating_sub(4)];
+            sock.read_exact(&mut startup).await.unwrap();
+            let _ = startup_seen.send(());
+            sock.read_to_end(&mut Vec::new()).await.unwrap();
             startup
         });
+        // Scheduled, not timed: a timer tick can lose a race that a stalled thread decides.
         let ticks = Arc::new(AtomicUsize::new(0));
         let counter = Arc::clone(&ticks);
         let heartbeat = tokio::spawn(async move {
             loop {
-                tokio::time::sleep(Duration::from_millis(5)).await;
                 counter.fetch_add(1, Ordering::Relaxed);
+                tokio::task::yield_now().await;
             }
         });
 
         let started = std::time::Instant::now();
-        let result = tokio::time::timeout(
-            Duration::from_millis(50),
-            PgReplicationConnection::connect_async(&conninfo),
-        )
-        .await;
-        assert!(result.is_err(), "the outer timeout must fire");
+        tokio::select! {
+            result = PgReplicationConnection::connect_async(&conninfo) => {
+                panic!("connect_async must stay pending against a silent server: {:?}", result.err());
+            }
+            seen = startup_packet => seen.expect("the server task ended before the startup packet"),
+        }
+        // A blocking connect would hold the thread for the whole `connect_timeout` of 10 s.
         assert!(
-            started.elapsed() < Duration::from_millis(500),
+            started.elapsed() < Duration::from_secs(5),
             "{:?}",
             started.elapsed()
         );
@@ -3199,13 +3214,13 @@ mod tests {
         );
         heartbeat.abort();
 
-        let startup = tokio::time::timeout(Duration::from_secs(2), silent_server)
+        let startup = tokio::time::timeout(Duration::from_secs(10), silent_server)
             .await
             .expect("dropping the connect future must close its socket")
             .unwrap();
         assert!(
             !startup.is_empty(),
-            "the startup packet never reached the server before the timeout"
+            "the startup packet never reached the server before the connect was dropped"
         );
     }
 
@@ -3728,6 +3743,265 @@ mod tests {
         }
     }
 
+    // === Stand-in server ===
+
+    /// What a stand-in server answers to a startup packet.
+    #[derive(Clone, Copy)]
+    enum Greeting {
+        /// Accept the login; the session reports `in_hot_standby` as `standby`.
+        Accept { standby: bool },
+        /// Refuse it with an ErrorResponse of `(sqlstate, message)`.
+        Reject(&'static str, &'static str),
+    }
+
+    /// A PostgreSQL stand-in on plain threads, so it outlives any runtime a test builds. It counts startup packets, answers each with its greeting and, once a login is accepted, answers every simple query with one row. Dropping it stops the listener and waits for the sessions, which end when their clients hang up.
+    struct StandIn {
+        port: u16,
+        startups: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+        stopping: std::sync::Arc<std::sync::atomic::AtomicBool>,
+        listener: Option<std::thread::JoinHandle<()>>,
+    }
+
+    impl StandIn {
+        fn new(greeting: Greeting) -> Self {
+            use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+            use std::sync::Arc;
+
+            let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+            let port = listener.local_addr().unwrap().port();
+            let startups = Arc::new(AtomicUsize::new(0));
+            let stopping = Arc::new(AtomicBool::new(false));
+            let (counter, stop) = (Arc::clone(&startups), Arc::clone(&stopping));
+            let listener = std::thread::spawn(move || {
+                let mut sessions = Vec::new();
+                for stream in listener.incoming() {
+                    let (stream, counter) = (stream.unwrap(), Arc::clone(&counter));
+                    if stop.load(Ordering::SeqCst) {
+                        break;
+                    }
+                    sessions.push(std::thread::spawn(move || {
+                        let _ = Self::serve(stream, greeting, &counter);
+                    }));
+                }
+                // A session ends when its client hangs up, so the test must drop its connections first.
+                for session in sessions {
+                    let _ = session.join();
+                }
+            });
+            Self {
+                port,
+                startups,
+                stopping,
+                listener: Some(listener),
+            }
+        }
+
+        fn startups(&self) -> usize {
+            self.startups.load(std::sync::atomic::Ordering::SeqCst)
+        }
+
+        /// One backend message: its type byte, a length that counts itself, and the body.
+        fn message(kind: u8, body: &[u8]) -> Vec<u8> {
+            let mut out = vec![kind];
+            out.extend_from_slice(&(body.len() as u32 + 4).to_be_bytes());
+            out.extend_from_slice(body);
+            out
+        }
+
+        fn serve(
+            mut sock: std::net::TcpStream,
+            greeting: Greeting,
+            startups: &std::sync::atomic::AtomicUsize,
+        ) -> std::io::Result<()> {
+            use std::io::{Read, Write};
+
+            let read_frame = |sock: &mut std::net::TcpStream| -> std::io::Result<Vec<u8>> {
+                let mut len = [0u8; 4];
+                sock.read_exact(&mut len)?;
+                let mut body = vec![0; (u32::from_be_bytes(len) as usize).saturating_sub(4)];
+                sock.read_exact(&mut body)?;
+                Ok(body)
+            };
+            read_frame(&mut sock)?;
+            startups.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            let standby = match greeting {
+                Greeting::Reject(sqlstate, text) => {
+                    let fields = format!("SFATAL\0VFATAL\0C{sqlstate}\0M{text}\0\0");
+                    return sock.write_all(&Self::message(b'E', fields.as_bytes()));
+                }
+                Greeting::Accept { standby } => standby,
+            };
+            let mut hello = Self::message(b'R', &0u32.to_be_bytes());
+            for (name, value) in [
+                ("client_encoding", "UTF8"),
+                ("server_version", "17.0"),
+                ("default_transaction_read_only", "off"),
+                ("in_hot_standby", if standby { "on" } else { "off" }),
+            ] {
+                hello.extend(Self::message(b'S', format!("{name}\0{value}\0").as_bytes()));
+            }
+            hello.extend(Self::message(b'K', &[0; 8]));
+            hello.extend(Self::message(b'Z', b"I"));
+            sock.write_all(&hello)?;
+            loop {
+                let mut kind = [0u8; 1];
+                sock.read_exact(&mut kind)?;
+                read_frame(&mut sock)?;
+                if kind[0] == b'X' {
+                    return Ok(());
+                }
+                // One text column "x" holding "1".
+                let mut row_description = 1u16.to_be_bytes().to_vec();
+                row_description.extend_from_slice(b"x\0");
+                row_description.extend_from_slice(&[0, 0, 0, 0, 0, 0, 0, 0, 0, 25, 255, 255]);
+                row_description.extend_from_slice(&[255, 255, 255, 255, 0, 0]);
+                let mut reply = Self::message(b'T', &row_description);
+                reply.extend(Self::message(b'D', &[0, 1, 0, 0, 0, 1, b'1']));
+                reply.extend(Self::message(b'C', b"SELECT 1\0"));
+                reply.extend(Self::message(b'Z', b"I"));
+                sock.write_all(&reply)?;
+            }
+        }
+    }
+
+    impl Drop for StandIn {
+        fn drop(&mut self) {
+            self.stopping
+                .store(true, std::sync::atomic::Ordering::SeqCst);
+            // A connection wakes the listener out of `accept` to see the flag.
+            let _ = std::net::TcpStream::connect(("127.0.0.1", self.port));
+            if let Some(listener) = self.listener.take() {
+                let _ = listener.join();
+            }
+        }
+    }
+
+    /// A conninfo that lists `servers` in order, each address once.
+    fn stand_in_conninfo(servers: &[&StandIn], extra: &str) -> String {
+        let hosts = vec!["127.0.0.1"; servers.len()].join(",");
+        let ports: Vec<String> = servers.iter().map(|s| s.port.to_string()).collect();
+        format!(
+            "host={hosts} port={} dbname=async user=async sslmode=disable gssencmode=disable connect_timeout=5 {extra}",
+            ports.join(",")
+        )
+    }
+
+    /// A server that cannot accept connections yet is passed over for the next host, in whatever language it words the error, and `connect_async` words the failure as `PQconnectdb` does.
+    #[test]
+    fn connect_async_walks_past_every_server_that_cannot_accept_connections() {
+        let servers =
+            |greeting| -> Vec<StandIn> { (0..3).map(|_| StandIn::new(greeting)).collect() };
+        // The text of an error with each server's port replaced by its position in the list.
+        let text = |err: &ReplicationError, servers: &[StandIn]| {
+            servers
+                .iter()
+                .enumerate()
+                .fold(err.to_string(), |text, (i, s)| {
+                    text.replace(&s.port.to_string(), &format!("<{i}>"))
+                })
+        };
+        for message in [
+            "the database system is starting up",
+            "das Datenbanksystem nimmt keine Verbindungen an",
+        ] {
+            let greeting = Greeting::Reject("57P03", message);
+            let (for_sync, for_async) = (servers(greeting), servers(greeting));
+            let list =
+                |servers: &[StandIn]| stand_in_conninfo(&servers.iter().collect::<Vec<_>>(), "");
+            let sync = PgReplicationConnection::connect(&list(&for_sync))
+                .err()
+                .expect("every server refuses");
+            let async_ = block_on(PgReplicationConnection::connect_async(&list(&for_async)))
+                .err()
+                .expect("every server refuses");
+            let count = |servers: &[StandIn]| servers.iter().map(StandIn::startups).sum::<usize>();
+            assert_eq!(count(&for_sync), 3, "{message}: connect");
+            assert_eq!(count(&for_async), 3, "{message}: connect_async");
+            assert_eq!(
+                text(&async_, &for_async),
+                text(&sync, &for_sync),
+                "{message}"
+            );
+        }
+    }
+
+    /// A bad password is still an `Authentication` error, which the retry loop treats differently from a transient one.
+    #[test]
+    fn connect_async_reports_a_bad_password_as_an_authentication_error() {
+        let server = StandIn::new(Greeting::Reject(
+            "28P01",
+            "password authentication failed for user \"async\"",
+        ));
+        let err = block_on(PgReplicationConnection::connect_async(&stand_in_conninfo(
+            &[&server, &server],
+            "",
+        )))
+        .err()
+        .expect("the server refuses every login");
+        assert!(matches!(err, ReplicationError::Authentication(_)), "{err}");
+    }
+
+    /// A server that cannot accept connections yet, such as a standby still starting up, is passed over for the next host.
+    #[test]
+    fn connect_async_walks_past_a_server_that_is_not_accepting_connections() {
+        let starting = StandIn::new(Greeting::Reject(
+            "57P03",
+            "the database system is starting up",
+        ));
+        let ready = StandIn::new(Greeting::Accept { standby: false });
+        block_on(PgReplicationConnection::connect_async(&stand_in_conninfo(
+            &[&starting, &ready],
+            "",
+        )))
+        .map(drop)
+        .expect("the second host accepts");
+        assert_eq!((starting.startups(), ready.startups()), (1, 1));
+    }
+
+    /// A session that does not match `target_session_attrs` is judged after startup, and the walk goes on to the next host.
+    #[test]
+    fn connect_async_walks_past_a_session_of_the_wrong_kind() {
+        let standby = StandIn::new(Greeting::Accept { standby: true });
+        let primary = StandIn::new(Greeting::Accept { standby: false });
+        block_on(PgReplicationConnection::connect_async(&stand_in_conninfo(
+            &[&standby, &primary],
+            "target_session_attrs=read-write",
+        )))
+        .map(drop)
+        .expect("the primary matches");
+        assert_eq!((standby.startups(), primary.startups()), (1, 1));
+    }
+
+    /// A connection built under one runtime must run its commands under another, whether the first was dropped (its reactor is gone) or only kept without being driven (its reactor never fires).
+    #[test]
+    fn a_connection_runs_commands_under_a_runtime_other_than_the_one_that_built_it() {
+        let server = StandIn::new(Greeting::Accept { standby: false });
+        let conninfo = stand_in_conninfo(&[&server], "");
+        for drop_first in [true, false] {
+            let first = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            let mut conn = first
+                .block_on(PgReplicationConnection::connect_async(&conninfo))
+                .unwrap();
+            first.block_on(conn.exec_async("IDENTIFY_SYSTEM")).unwrap();
+            let kept = if drop_first {
+                drop(first);
+                None
+            } else {
+                Some(first)
+            };
+            let reply = block_on(async {
+                tokio::time::timeout(Duration::from_secs(5), conn.exec_async("IDENTIFY_SYSTEM"))
+                    .await
+                    .unwrap_or_else(|_| panic!("hung (first runtime dropped: {drop_first})"))
+            });
+            assert!(reply.is_ok(), "dropped {drop_first}: {:?}", reply.err());
+            drop(kept);
+        }
+    }
+
     /// `connect_timeout` follows libpq's `strtol` rules, and only libpq 16 and older raise 1 s to 2 s.
     #[test]
     fn connect_timeout_follows_libpq_integer_rules() {
diff --git a/src/protocol.rs b/src/protocol.rs
index 72a3723..49fb0eb 100644
--- a/src/protocol.rs
+++ b/src/protocol.rs
@@ -529,7 +529,16 @@ impl RelationInfo {
 #[derive(Debug, Clone)]
 pub struct StreamingReplicationMessage {
     pub message: LogicalReplicationMessage,
+    /// `true` for a message read inside a streamed chunk, which is exactly when `xid` is `Some`.
     pub is_streaming: bool,
+    /// The xid of a message read inside a streamed chunk (protocol v2+); `None` outside one.
+    ///
+    /// For `Relation`, `Type`, `Insert`, `Update`, `Delete`, `Truncate` and `Message` it is the
+    /// xid pgoutput wrote just before that change: a subtransaction's xid after a savepoint,
+    /// the top-level transaction's otherwise. A `Message` is the exception: pgoutput always
+    /// stamps it with the top-level xid. `StreamStart` carries the chunk's top-level xid and
+    /// `StreamStop` none, so take the top-level xid from `StreamStart`; do not assume every
+    /// `xid` inside the chunk equals it.
     pub xid: Option<Xid>,
 }
 
@@ -3561,6 +3570,23 @@ mod tests {
         }
     }
 
+    /// Within one chunk `xid` follows the stamp on each change: the chunk's own xid, then a subtransaction's, then back.
+    #[test]
+    fn xid_follows_each_change_in_a_chunk_not_the_chunk() {
+        let mut parser = LogicalReplicationParser::with_protocol_version(2);
+        let started = parser.parse_wal_message(&stream_start(745)).unwrap();
+        assert_eq!(started.xid, Some(745));
+
+        let insert = streamed_changes().swap_remove(2);
+        let mut xids = Vec::new();
+        for stamp in [745, 746, 745] {
+            let sent = StreamingReplicationMessage::new_streaming(insert.clone(), stamp);
+            let bytes = crate::pgoutput_encode::encode_streaming_message_to_bytes(&sent, 2);
+            xids.push(parser.parse_wal_message(&bytes).unwrap().xid);
+        }
+        assert_eq!(xids, [Some(745), Some(746), Some(745)]);
+    }
+
     /// Outside a chunk no change carries an xid, `StreamStart` carries its own and `StreamStop` none.
     #[test]
     fn only_changes_inside_a_chunk_carry_a_change_xid() {
diff --git a/src/retry.rs b/src/retry.rs
index 223149e..50d4abc 100644
--- a/src/retry.rs
+++ b/src/retry.rs
@@ -256,12 +256,13 @@ impl ReplicationConnectionRetry {
 
     /// [`connect_with_retry`](Self::connect_with_retry) that stops once `budget` has passed, checked only between attempts.
     ///
-    /// An attempt in flight is never cut short, because its own per-address `connect_timeout` bounds it and a multi-host attempt needs that time to reach its next host. A backoff that would end past `budget` ends at `budget` with [`connect_budget_exceeded`].
+    /// An attempt in flight is never cut short, because its own per-address `connect_timeout` bounds it and a multi-host attempt needs that time to reach its next host. A backoff that would end past `budget` ends at `budget` with [`connect_budget_exceeded`]. A zero `budget` means none, as for `connect_timeout`.
     pub(crate) async fn connect_with_retry_within(
         &self,
         budget: Duration,
     ) -> Result<PgReplicationConnection> {
-        self.connect_until(Some(budget)).await
+        self.connect_until((!budget.is_zero()).then_some(budget))
+            .await
     }
 
     async fn connect_until(&self, budget: Option<Duration>) -> Result<PgReplicationConnection> {
@@ -394,6 +395,48 @@ mod tests {
         silent.abort();
     }
 
+    /// A server that accepts every connection and hangs up at once, so each attempt fails
+    /// without a timeout of its own. Returns its port and the number of connections seen.
+    async fn server_that_hangs_up() -> (u16, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let attempts = std::sync::Arc::new(AtomicUsize::new(0));
+        let count = std::sync::Arc::clone(&attempts);
+        tokio::spawn(async move {
+            while let Ok((socket, _)) = listener.accept().await {
+                count.fetch_add(1, Ordering::SeqCst);
+                drop(socket);
+            }
+        });
+        (port, attempts)
+    }
+
+    /// A zero budget is "no budget", not a budget already spent: it made one attempt and
+    /// returned `Timeout("... 0ns")` whatever `max_attempts` said.
+    #[tokio::test]
+    async fn a_zero_connect_budget_does_not_cut_the_retries_short() {
+        let (port, attempts) = server_that_hangs_up().await;
+        let retry = ReplicationConnectionRetry::new(
+            fixed_backoff(3, Duration::from_millis(5)),
+            format!(
+                "host=127.0.0.1 port={port} user=u dbname=db sslmode=disable gssencmode=disable"
+            ),
+        );
+
+        let err = match retry.connect_with_retry_within(Duration::ZERO).await {
+            Ok(_) => panic!("the server hangs up on every connection"),
+            Err(e) => e,
+        };
+
+        assert!(!matches!(err, ReplicationError::Timeout(_)), "{err:?}");
+        assert_eq!(
+            attempts.load(std::sync::atomic::Ordering::SeqCst),
+            3,
+            "every attempt `max_attempts` allows must be made"
+        );
+    }
+
     /// A backoff that would end past the budget ends at the budget with a timeout instead.
     #[tokio::test]
     async fn a_backoff_past_the_connect_budget_times_out_at_the_budget() {
diff --git a/src/snapshot/aligned.rs b/src/snapshot/aligned.rs
index 15d548e..642c5d4 100644
--- a/src/snapshot/aligned.rs
+++ b/src/snapshot/aligned.rs
@@ -8,7 +8,8 @@ use super::table_copy::{poisoned_err, TableCopy};
 use crate::connection::PgReplicationConnection;
 use crate::error::{ReplicationError, Result};
 use crate::prelude::*;
-use crate::stream::LogicalReplicationStream;
+use crate::sql_builder::quote_ident;
+use crate::stream::{LogicalReplicationStream, ServerIdentity};
 use crate::types::{Lsn, ReplicationSlotOptions, SlotType};
 use core::sync::atomic::{AtomicU32, Ordering};
 use tokio_util::sync::CancellationToken;
@@ -16,10 +17,13 @@ use tokio_util::sync::CancellationToken;
 /// A copy of chosen tables exact at a temporary side slot's consistent point
 /// `C`, to be merged with a stream that is already running.
 ///
-/// The copy contains every transaction whose commit record lies below `C` and
+/// The snapshot contains every transaction whose commit record lies below `C` and
 /// none at or past it, which is what
 /// [`EXPORT_SNAPSHOT`](https://www.postgresql.org/docs/current/logicaldecoding-explanation.html#LOGICALDECODING-EXPLANATION-EXPORTED-SNAPSHOTS)
-/// guarantees.
+/// guarantees. A `COPY` still reads the table's current file, so a `TRUNCATE` or
+/// rewriting `ALTER TABLE` committed after `C` would make it return nothing.
+/// [`capture`](Self::capture) therefore locks the tables against such commands
+/// (see Cost) and fails for a table one has already changed.
 ///
 /// # Merging with the running stream
 ///
@@ -35,7 +39,33 @@ use tokio_util::sync::CancellationToken;
 /// transaction that ends in `RollbackPrepared` or in a `StreamAbort` of the
 /// whole transaction is dropped. A `StreamAbort` of a subtransaction drops the
 /// buffered changes whose [`ChangeEvent::xid`](crate::ChangeEvent::xid) equals
-/// its `subtransaction_xid`.
+/// its `subtransaction_xid`, except a `Message`, which always carries the
+/// top-level xid.
+///
+/// Acknowledge each transaction decided below `C` before loading the copy. A
+/// reconnect replays from the acknowledged position, so without it every
+/// transaction delivered before the `None` arrives again after the copy and is
+/// applied twice (50 of 50 in a measured idle-database run, none with the
+/// acknowledgement). [`EventStream::next_event_before`](crate::EventStream::next_event_before)
+/// reconnects without telling the caller, so a buffer keyed by transaction id
+/// should also start over when a `Begin`, a `BeginPrepare` or a `StreamStart`
+/// with `first_segment` arrives for a transaction it already holds. That replay
+/// of a streamed transaction was not reproduced; this is a precaution.
+///
+/// ```no_run
+/// # use pg_walstream::{AlignedSnapshot, EventType, LogicalReplicationStream, Result};
+/// # use tokio_util::sync::CancellationToken;
+/// # async fn merge(mut stream: LogicalReplicationStream, copy: AlignedSnapshot, cancel: CancellationToken) -> Result<()> {
+/// let c = copy.consistent_point();
+/// while let Some(event) = stream.next_event_before(c, &cancel).await? {
+///     // Buffer it, and apply its transaction at the commit event.
+///     if let EventType::Commit { .. } | EventType::StreamCommit { .. } = event.event_type {
+///         stream.shared_lsn_feedback.update_applied_lsn(event.lsn);
+///     }
+/// }
+/// # Ok(())
+/// # }
+/// ```
 ///
 /// [`next_event_before`](LogicalReplicationStream::next_event_before) returns
 /// `None` once every commit below `C` has been delivered, which is where the
@@ -64,6 +94,17 @@ use tokio_util::sync::CancellationToken;
 /// database's `xmin` until [`finish`](Self::finish) or drop, exactly as a
 /// [`Snapshot`](super::Snapshot) does.
 ///
+/// The same transaction holds an `ACCESS SHARE` lock on every requested table
+/// from [`capture`](Self::capture) on. Reads and writes are not blocked, but a
+/// `TRUNCATE`, `ALTER TABLE`, `DROP`, `VACUUM FULL` or `CLUSTER` on those tables
+/// waits until [`finish`](Self::finish) or drop, and every write issued behind
+/// such a waiting command waits with it, up to its `lock_timeout` if it has one
+/// (an `ALTER TABLE` waited 2.4 s, and an `INSERT` behind it failed with
+/// `55P03` at its 2 s timeout). Give that DDL a `lock_timeout`, or run it before
+/// the capture or after the copy. One that already holds its lock when
+/// `capture` asks for its own makes `capture` fail at once, and so does a table
+/// changed since `C`: capture again.
+///
 /// # Cancellation
 ///
 /// Dropping the future that [`capture`](Self::capture) returns closes its
@@ -73,6 +114,15 @@ use tokio_util::sync::CancellationToken;
 /// stream's own reconnect. [`with_cancellation`](Self::with_cancellation)
 /// covers only the copy loop.
 ///
+/// `capture` has no deadline of its own, and a prepared transaction left open
+/// blocks the slot creation like any other. To end the wait, end the blocker or
+/// cancel the walsender from a second connection, for example with
+/// `SELECT pg_cancel_backend(active_pid) FROM pg_replication_slots WHERE slot_name LIKE 'pg_walstream_aligned_%'`.
+///
+/// [`next_row`](Self::next_row) is not cancel-safe where a table's `COPY`
+/// starts. Do not race it in a `select!` that can drop it, because a dropped
+/// start leaves the reader unusable; race the feedback ticker instead.
+///
 /// # Failure
 ///
 /// After any error the handle is poisoned. A new capture has a new `C`, so a
@@ -101,10 +151,19 @@ impl AlignedSnapshot {
         conninfo: &str,
         publications: &[&str],
         tables: &[(&str, &str)],
+    ) -> Result<Self> {
+        Self::capture_on(conninfo, publications, tables, None).await
+    }
+
+    async fn capture_on(
+        conninfo: &str,
+        publications: &[&str],
+        tables: &[(&str, &str)],
+        expected: Option<ServerIdentity>,
     ) -> Result<Self> {
         let publications = validate_request(publications, tables)?;
         let (reader, tables, consistent_point) =
-            capture_tables(conninfo, &publications, tables).await?;
+            capture_tables(conninfo, &publications, tables, expected).await?;
 
         tracing::info!(
             "Aligned snapshot ready: {} table(s) at {}",
@@ -131,6 +190,8 @@ impl AlignedSnapshot {
     }
 
     /// The next row, or `Ok(None)` once every table has been copied.
+    ///
+    /// Not cancel-safe where a table's `COPY` starts: a future dropped there leaves the reader unusable and every later call fails.
     pub async fn next_row(&mut self) -> Result<Option<SnapshotRow>> {
         self.copy.next_row().await
     }
@@ -172,6 +233,11 @@ impl LogicalReplicationStream {
     /// [`AlignedSnapshot::capture`] with this stream's connection string and
     /// publications, leaving this stream's own connection untouched.
     ///
+    /// The side connection must reach the stream's cluster and timeline
+    /// ([`server_identity`](Self::server_identity)). A connection string with
+    /// several hosts can land it elsewhere, which fails with a permanent error
+    /// before any slot is created.
+    ///
     /// The returned future borrows nothing from the stream, so it can be
     /// spawned, or raced against the
     /// [`send_feedback`](Self::send_feedback) ticker that keeps the stream alive
@@ -183,6 +249,7 @@ impl LogicalReplicationStream {
         let logical = self.config().slot_type == SlotType::Logical;
         let conninfo = self.conninfo().to_string();
         let publications = plan::split_publication_names(&self.config().publication_name);
+        let identity = *self.server_identity();
         let tables: Vec<(String, String)> = tables
             .iter()
             .map(|(schema, table)| (schema.to_string(), table.to_string()))
@@ -198,7 +265,7 @@ impl LogicalReplicationStream {
                 .iter()
                 .map(|(schema, table)| (schema.as_str(), table.as_str()))
                 .collect();
-            AlignedSnapshot::capture(&conninfo, &publications, &tables).await
+            AlignedSnapshot::capture_on(&conninfo, &publications, &tables, Some(identity)).await
         }
     }
 }
@@ -208,9 +275,13 @@ async fn capture_tables(
     conninfo: &str,
     publications: &[String],
     requested: &[(&str, &str)],
+    expected: Option<ServerIdentity>,
 ) -> Result<(PgReplicationConnection, Vec<SnapshotTable>, Lsn)> {
     let mut side =
         PgReplicationConnection::connect_async(&plan::derive_side_conninfo(conninfo)).await?;
+    if let Some(expected) = expected {
+        expected.ensure_same_server(&ServerIdentity::identify(&mut side).await?)?;
+    }
     let created = side
         .create_replication_slot_with_options_async(
             &side_slot_name(),
@@ -240,9 +311,67 @@ async fn capture_tables(
     drop(side);
 
     let tables = select_tables(published, requested, publications)?;
+    pin_tables(&mut reader, &tables).await?;
     Ok((reader, tables, consistent_point))
 }
 
+/// Lock the requested tables, then fail if one changed since the snapshot.
+///
+/// A `COPY` reads the table's current file with the snapshot's visibility, so a `TRUNCATE` or rewriting `ALTER TABLE` that committed after `C` makes it return nothing, silently. `ACCESS SHARE` blocks those commands until the reader's transaction ends; `NOWAIT` fails fast instead of waiting without a deadline behind one in flight. One that committed before the lock changes the table's file or name, which the second statement reports.
+async fn pin_tables(reader: &mut PgReplicationConnection, tables: &[SnapshotTable]) -> Result<()> {
+    let (lock, changed) = pin_sql(tables)?;
+    reader.exec_async(&lock).await?;
+    let result = reader.exec_async(&changed).await?;
+    let moved: Vec<String> = (0..result.ntuples())
+        .filter_map(|row| result.get_value(row, 0))
+        .collect();
+    if moved.is_empty() {
+        return Ok(());
+    }
+    Err(ReplicationError::protocol(format!(
+        "table(s) {} changed after the snapshot, so a copy would not match it: truncated, \
+         rewritten, renamed or dropped. Capture again",
+        moved.join(", ")
+    )))
+}
+
+/// The `LOCK TABLE` statement and the query that lists the tables whose file or name moved on since the snapshot.
+///
+/// The second query's `pg_relation_filenode` reads the current catalog while `pg_class` reads the snapshot's, so a difference is a change made after `C`. A partitioned table has no file of its own and its `COPY` reads every partition, so the query walks the partitions the snapshot sees, and `pg_partition_tree` (current catalog) catches one detached since. The locks come first because only then can nothing change further.
+fn pin_sql(tables: &[SnapshotTable]) -> Result<(String, String)> {
+    let names = tables
+        .iter()
+        .map(|t| {
+            Ok(format!(
+                "{}.{}",
+                quote_ident(&t.relation.namespace)?,
+                quote_ident(&t.relation.relation_name)?
+            ))
+        })
+        .collect::<Result<Vec<_>>>()?
+        .join(", ");
+    let oids = tables
+        .iter()
+        .map(|t| t.relation.relation_id.to_string())
+        .collect::<Vec<_>>()
+        .join(", ");
+    Ok((
+        format!("LOCK TABLE {names} IN ACCESS SHARE MODE NOWAIT"),
+        format!(
+            "WITH RECURSIVE tree(root, oid) AS (SELECT c.oid, c.oid FROM pg_class c \
+             WHERE c.oid IN ({oids}) UNION ALL SELECT t.root, i.inhrelid FROM tree t \
+             JOIN pg_class p ON p.oid = t.oid AND p.relkind = 'p' \
+             JOIN pg_inherits i ON i.inhparent = t.oid) \
+             SELECT format('%I.%I', n.nspname, c.relname) FROM tree t \
+             JOIN pg_class c ON c.oid = t.oid JOIN pg_namespace n ON n.oid = c.relnamespace \
+             WHERE c.relfilenode IS DISTINCT FROM coalesce(pg_relation_filenode(c.oid), 0) \
+             OR to_regclass(format('%I.%I', n.nspname, c.relname))::oid IS DISTINCT FROM c.oid \
+             OR (t.oid <> t.root AND NOT EXISTS \
+             (SELECT 1 FROM pg_partition_tree(t.root) pt WHERE pt.relid = t.oid))"
+        ),
+    ))
+}
+
 /// Reject requests that cannot name a well-defined copy, before any connection opens.
 fn validate_request(publications: &[&str], tables: &[(&str, &str)]) -> Result<Vec<String>> {
     let publications: Vec<String> = publications
@@ -395,6 +524,23 @@ mod tests {
         assert_eq!(validate_request(&[" p ", ""], &t).unwrap(), ["p"]);
     }
 
+    #[test]
+    fn pinning_quotes_every_name_and_lists_every_table() {
+        let tables = published(&[("My Schema", "Order \"x\""), ("public", "select")]);
+        let (lock, changed) = pin_sql(&tables).unwrap();
+        assert_eq!(
+            lock,
+            "LOCK TABLE \"My Schema\".\"Order \"\"x\"\"\", \"public\".\"select\" IN ACCESS SHARE MODE NOWAIT"
+        );
+        assert!(changed.contains("c.oid IN (0, 1)"), "{changed}");
+        // A partitioned table has no file of its own: its partitions are what a `COPY` reads.
+        assert!(
+            changed.contains("pg_inherits") && changed.contains("pg_partition_tree"),
+            "{changed}"
+        );
+        assert!(pin_sql(&published(&[("public", "a\0b")])).is_err());
+    }
+
     #[test]
     fn slot_creation_result_needs_both_columns() {
         let (lsn, name) = slot_creation_result(Some("0/16B3748"), Some("00000003-1")).unwrap();
diff --git a/src/snapshot/mod.rs b/src/snapshot/mod.rs
index 5633918..20b2ef5 100644
--- a/src/snapshot/mod.rs
+++ b/src/snapshot/mod.rs
@@ -480,8 +480,9 @@ async fn resolve_tables(
         ))
         .await?;
 
-    // Run inside the snapshot transaction so the catalog we read is the one the
-    // COPYs will see, and a concurrent ALTER PUBLICATION cannot race us.
+    // The table list is the one the snapshot sees, so a concurrent ALTER PUBLICATION
+    // cannot race us. The COPYs resolve each table in the current catalog, though, and
+    // read an empty new file after a TRUNCATE or a rewriting ALTER that committed later.
     let sql = plan::build_catalog_sql(server_version, publications)?;
     let result = reader.exec_async(&sql).await?;
     let rows = catalog_rows_from_result(&result)?;
diff --git a/src/stream.rs b/src/stream.rs
index a9ae69a..db2597b 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -53,6 +53,8 @@ pub struct LogicalReplicationStream {
     held: Option<HeldFrame>,
     /// The last keepalive's `wal_end` on this connection: every commit below it has been delivered.
     keepalive_wal_end: Lsn,
+    /// Raised when a reconnect starts and lowered when one succeeds, so a reconnect a caller dropped mid-way leaves it raised.
+    recovering: bool,
 }
 
 /// A `'w'` frame held back at a boundary, with the commit position that stopped it.
@@ -250,9 +252,23 @@ impl ServerIdentity {
     }
 
     /// Identify the server on the far end of `connection`.
-    async fn identify(connection: &mut PgReplicationConnection) -> Result<Self> {
+    pub(crate) async fn identify(connection: &mut PgReplicationConnection) -> Result<Self> {
         Self::from_identify_system(&connection.identify_system_async().await?)
     }
+
+    /// Fail permanently unless `other`, read on a second connection opened for this stream, names the same cluster and timeline.
+    ///
+    /// A conninfo with several hosts can land the second connection on another cluster, whose WAL positions mean nothing to this stream.
+    pub(crate) fn ensure_same_server(&self, other: &Self) -> Result<()> {
+        if self == other {
+            return Ok(());
+        }
+        Err(ReplicationError::permanent_connection(format!(
+            "a side connection reached system {} on timeline {}, but the stream runs on system {} \
+             on timeline {}; its positions would not match the stream's",
+            other.system_id, other.timeline, self.system_id, self.timeline
+        )))
+    }
 }
 
 /// One line of a timeline history file, naming an ancestor timeline and the WAL position where its successor branched off.
@@ -449,7 +465,7 @@ impl ReplicationStreamConfig {
     /// * `streaming_mode` - Streaming mode for logical decoding (off/on/parallel).
     ///   `on` requires protocol version >= 2; `parallel` requires protocol version >= 4.
     /// * `feedback_interval` - How often to send status feedback to PostgreSQL
-    /// * `connection_timeout` - Maximum time to wait for connection establishment
+    /// * `connection_timeout` - Time budget for connecting: the per-address `connect_timeout`, and the limit on retries between attempts. See [`with_connection_timeout`](Self::with_connection_timeout)
     /// * `health_check_interval` - How often to check connection health
     /// * `retry_config` - Configuration for connection retry behavior
     ///
@@ -674,7 +690,9 @@ impl ReplicationStreamConfig {
         self
     }
 
-    /// Set the connection establishment timeout.
+    /// Set the connection time budget.
+    ///
+    /// It is each address's `connect_timeout` (unless the connection string sets one) and it stops the retries between attempts once spent, but an attempt in flight is never cut short, so a host name with several addresses can take up to that long for each. It also bounds the `IDENTIFY_SYSTEM` that follows in [`new`](LogicalReplicationStream::new). `Duration::ZERO` means no timeout.
     #[inline]
     pub fn with_connection_timeout(mut self, d: Duration) -> Self {
         self.connection_timeout = d;
@@ -842,6 +860,7 @@ impl LogicalReplicationStream {
             replay_stop: ReplayStop::Running,
             held: None,
             keepalive_wal_end: Lsn::new(0),
+            recovering: false,
         })
     }
 
@@ -853,16 +872,30 @@ impl LogicalReplicationStream {
         &self.identity
     }
 
+    /// A command that a caller cancelled before its reply leaves the connection refusing every later one with a `Protocol` error that reads as retryable, and nothing before [`start`](Self::start) replaces the connection. The refusal is the only `Protocol` a command meets on a connection that reports dead, so it is one a retry cannot cure; a read reconnects instead, and keeps the transient classification.
+    fn rebuild_if_refused(&self, error: ReplicationError) -> ReplicationError {
+        match error {
+            ReplicationError::Protocol(refusal) if !self.connection.is_alive() => {
+                ReplicationError::permanent_connection(format!("{refusal}; create a new stream"))
+            }
+            other => other,
+        }
+    }
+
     /// The replication slot's `confirmed_flush_lsn`, read on the stream's own connection, or `None` when the slot does not exist.
     ///
     /// A point-in-time read that must run before [`start`](Self::start), while the connection still accepts queries. The stream never feeds the value into its feedback. Requires a `replication=database` connection.
     ///
     /// # Errors
     ///
-    /// Returns an error when the stream is already streaming, when the query fails, or when the slot is not a logical slot.
+    /// Returns an error when the stream is already streaming, when the query fails, or when the slot is not a logical slot. After a cancelled startup command the error is [`ReplicationError::PermanentConnection`], as for [`start`](Self::start).
     pub async fn slot_confirmed_flush_lsn(&mut self) -> Result<Option<Lsn>> {
         let sql = crate::sql_builder::build_slot_confirmed_flush_sql(&self.config.slot_name)?;
-        let result = self.connection.exec_async(&sql).await?;
+        let result = self
+            .connection
+            .exec_async(&sql)
+            .await
+            .map_err(|e| self.rebuild_if_refused(e))?;
         confirmed_flush_lsn_from(&self.config.slot_name, &result)
     }
 
@@ -872,7 +905,7 @@ impl LogicalReplicationStream {
     ///
     /// # Errors
     ///
-    /// Returns an error when `TIMELINE_HISTORY` fails, which it does once the stream is streaming, or when its reply is not a well-formed history of that timeline.
+    /// Returns an error when `TIMELINE_HISTORY` fails, which it does once the stream is streaming, or when its reply is not a well-formed history of that timeline. After a cancelled startup command the error is [`ReplicationError::PermanentConnection`], as for [`start`](Self::start).
     pub async fn timeline_history(&mut self) -> Result<Vec<TimelineSwitch>> {
         let timeline = self.identity.timeline;
         if timeline == 1 {
@@ -881,7 +914,8 @@ impl LogicalReplicationStream {
         let result = self
             .connection
             .exec_async(&format!("TIMELINE_HISTORY {timeline}"))
-            .await?;
+            .await
+            .map_err(|e| self.rebuild_if_refused(e))?;
         timeline_history_from(timeline, &result)
     }
 
@@ -1007,6 +1041,7 @@ impl LogicalReplicationStream {
     /// - Replication slot creation fails (only when automatic creation is enabled)
     /// - Starting replication command fails
     /// - The slot is missing and [`existing_slot_required`](ReplicationStreamConfig::existing_slot_required) is enabled, yielding [`ReplicationError::ReplicationSlot`]
+    /// - A startup command of this stream (`start`, [`slot_confirmed_flush_lsn`](Self::slot_confirmed_flush_lsn), [`timeline_history`](Self::timeline_history)) was cancelled before its reply, for instance by a `timeout`: the connection refuses every later command and the error is [`ReplicationError::PermanentConnection`], so create a new stream
     ///
     /// # Example
     ///
@@ -1038,7 +1073,9 @@ impl LogicalReplicationStream {
     pub async fn start(&mut self, start_lsn: Option<XLogRecPtr>) -> Result<()> {
         info!("Starting logical replication stream");
 
-        self.initialize().await?;
+        self.initialize()
+            .await
+            .map_err(|e| self.rebuild_if_refused(e))?;
         let start_lsn = start_lsn.unwrap_or(INVALID_XLOG_REC_PTR);
 
         // Seed `applied` with the caller's startpoint: `start(Some(X))` asserts
@@ -1057,7 +1094,8 @@ impl LogicalReplicationStream {
             .collect::<Vec<_>>();
         self.connection
             .start_replication_async(&self.config.slot_name, start_lsn, &options_ref)
-            .await?;
+            .await
+            .map_err(|e| self.rebuild_if_refused(e))?;
 
         self.reset_barrier();
         if !self.slot.is_present() {
@@ -1343,7 +1381,7 @@ impl LogicalReplicationStream {
             // 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 {
+            match self.recover_connection_boxed().await {
                 Ok(_) => {
                     info!("Connection recovered successfully");
                 }
@@ -1390,9 +1428,19 @@ impl LogicalReplicationStream {
         Ok(())
     }
 
+    /// [`recover_connection`](Self::recover_connection) boxed in a frame of its own. Boxing it where it is awaited builds the several-KB connect state machine on that caller's stack first, and the callers are the per-event read futures, which would then reserve it on every poll.
+    #[cold]
+    #[inline(never)]
+    fn recover_connection_boxed(
+        &mut self,
+    ) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send + '_>> {
+        Box::pin(self.recover_connection())
+    }
+
     /// Recover connection after a failure
     async fn recover_connection(&mut self) -> Result<()> {
         info!("Attempting to recover replication connection");
+        self.recovering = true;
         // The new connection resends everything from the applied position, held frame included.
         self.reset_barrier();
 
@@ -1475,6 +1523,7 @@ impl LogicalReplicationStream {
         // before the disconnect has no matching StreamStop coming.
         self.parser.reset();
 
+        self.recovering = false;
         info!("Replication connection recovered and restarted");
         Ok(())
     }
@@ -1490,6 +1539,10 @@ impl LogicalReplicationStream {
     /// For an even more ergonomic API, consider using `into_stream()` which provides
     /// an async iterator interface that automatically calls this method.
     ///
+    /// Not cancel-safe across a reconnect: dropping the future (a `timeout`) mid-reconnect
+    /// restarts the reconnect on the next call, so a timeout shorter than a reconnect never
+    /// lets it finish. [`EventStream::next_event`] keeps the reconnect across such drops.
+    ///
     /// # Arguments
     /// * `cancellation_token` - Token to signal cancellation/shutdown
     ///
@@ -1568,7 +1621,16 @@ impl LogicalReplicationStream {
         //
         // The check itself is cheap when nothing is wrong (it just compares `Instant::now()` against `last_health_check`), but `Instant::now()` is a vDSO syscall on Linux. At 100k+ rps that adds up, so we gate the health check on the per-event feedback counter and let it amortize across `HEALTH_CHECK_EVENT_INTERVAL` events. The actual time-based interval inside `check_connection_health()` still applies, this just skips the syscall on the inner-loop hot path.
         if self.feedback_check_counter & (HEALTH_CHECK_EVENT_INTERVAL - 1) == 0 {
-            if let Err(e) = self.check_connection_health().await {
+            let health = tokio::select! {
+                biased;
+                health = self.check_connection_health() => health,
+                _ = cancellation_token.cancelled() => {
+                    return Err(ReplicationError::cancelled(
+                        "Operation cancelled during reconnect"
+                    ));
+                }
+            };
+            if let Err(e) = health {
                 warn!("Health check failed: {}", e);
                 // Don't fail immediately, try to continue
             }
@@ -1649,7 +1711,17 @@ impl LogicalReplicationStream {
 
             // Attempt connection recovery if connection is dead
             if !self.connection.is_alive() {
-                if let Err(recovery_err) = Box::pin(self.recover_connection()).await {
+                // The token ends a reconnect too: a parked one is awaited by `shutdown()`.
+                let recovery = tokio::select! {
+                    biased;
+                    recovery = self.recover_connection_boxed() => recovery,
+                    _ = cancellation_token.cancelled() => {
+                        return Err(ReplicationError::cancelled(
+                            "Operation cancelled during reconnect"
+                        ));
+                    }
+                };
+                if let Err(recovery_err) = recovery {
                     // 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
@@ -2348,8 +2420,9 @@ pub struct EventStream {
 /// It **owns** the stream for the duration of the poll and hands it back
 /// alongside the result — which is why "polling" and "holding the stream" are
 /// mutually exclusive, and why dropping this future drops the connection.
-type PollFuture =
-    std::pin::Pin<Box<dyn Future<Output = (LogicalReplicationStream, Result<ChangeEvent>)> + Send>>;
+type PollFuture = std::pin::Pin<
+    Box<dyn Future<Output = (LogicalReplicationStream, Result<Option<ChangeEvent>>)> + Send>,
+>;
 
 /// Where an [`EventStream`]'s replication stream is, and whether it is done.
 ///
@@ -2417,11 +2490,17 @@ impl StreamState {
     /// needing a real server to stall on.
     #[cfg(test)]
     fn park_with_result(&mut self, result: Result<ChangeEvent>) {
+        self.park_resolved(result.map(Some));
+    }
+
+    /// [`park_with_result`](Self::park_with_result) for a read that may have stopped at its boundary (`Ok(None)`).
+    #[cfg(test)]
+    fn park_resolved(&mut self, result: Result<Option<ChangeEvent>>) {
         let placeholder: PollFuture = Box::pin(core::future::pending());
         let StreamState::Ready { stream, .. } =
             core::mem::replace(self, StreamState::Polling(placeholder))
         else {
-            panic!("park_with_result expects a stream in hand");
+            panic!("park_resolved expects a stream in hand");
         };
         let stream = *stream;
         *self = StreamState::Polling(Box::pin(async move { (stream, result) }));
@@ -2436,8 +2515,8 @@ impl StreamState {
         }
     }
 
-    /// Park the stream in a fresh poll future. A no-op unless idle.
-    fn start_poll(&mut self, cancellation_token: &CancellationToken) {
+    /// Park the stream in a fresh poll future that reads up to `boundary`. A no-op unless idle.
+    fn start_poll(&mut self, cancellation_token: &CancellationToken, boundary: Option<Lsn>) {
         if !matches!(
             self,
             StreamState::Ready {
@@ -2459,7 +2538,9 @@ impl StreamState {
             let mut stream = *stream;
             let cancellation_token = cancellation_token.clone();
             *self = StreamState::Polling(Box::pin(async move {
-                let result = stream.next_event_with_retry(&cancellation_token).await;
+                let result = stream
+                    .next_event_with_retry_until(&cancellation_token, boundary)
+                    .await;
                 (stream, result)
             }));
         }
@@ -2540,7 +2621,8 @@ impl EventStream {
     /// includes the ordinary case of a poll that was *abandoned*, such as
     /// `tokio::time::timeout(d, stream.next())` expiring or a `select!` arm
     /// losing. Call [`next_event`](Self::next_event) first: it reclaims the
-    /// parked stream. These accessors are synchronous and cannot.
+    /// parked stream. These accessors are synchronous and cannot. A
+    /// `next_event` dropped while it was reconnecting parks the stream the same way.
     pub fn inner(&self) -> &LogicalReplicationStream {
         self.state
             .stream()
@@ -2632,17 +2714,25 @@ impl EventStream {
     /// **Note**: If you also use this type as a `futures::Stream`, prefer calling
     /// `StreamExt::next()` instead to avoid confusion. This method is named `next_event()`
     /// to avoid conflict with `StreamExt::next()`.
+    ///
+    /// A reconnect that a dropped future (a `timeout`, a `select!` arm) interrupted carries on
+    /// in the next call instead of restarting, so a timeout shorter than a reconnect does not
+    /// starve it. Only the first call after the connection dies reconnects inline, and loses
+    /// its partial reconnect if dropped.
     pub async fn next_event(&mut self) -> Result<ChangeEvent> {
         if let Some(event) = self.held.take() {
             return Ok(event);
         }
+        self.park_if_recovering(None);
         // A `poll_next` that returned `Pending` parked the stream inside
         // `inflight`, and dropping the `Next`/`select!` future that drove it does
         // not clear that field. Resume the parked future rather than asserting it
         // cannot exist: abandoning a poll is ordinary use (`tokio::time::timeout`
         // around `StreamExt::next`), and its event is still owed to the caller.
         if let Some(result) = self.resume_parked_poll().await {
-            return result;
+            if let Some(event) = result? {
+                return Ok(event);
+            }
         }
 
         let inner = self
@@ -2660,8 +2750,9 @@ impl EventStream {
         if let Some(event) = self.held.take() {
             return Ok(self.deliver_unless_past(event, boundary));
         }
+        self.park_if_recovering(Some(boundary));
         if let Some(result) = self.resume_parked_poll().await {
-            return Ok(self.deliver_unless_past(result?, boundary));
+            return Ok(result?.and_then(|event| self.deliver_unless_past(event, boundary)));
         }
         let inner = self
             .state
@@ -2682,8 +2773,15 @@ impl EventStream {
         Some(event)
     }
 
+    /// A reconnect dropped mid-way (a caller's `timeout`) restarts from its first attempt, so one longer than the timeout never finishes; parked in a poll future it survives the drop.
+    fn park_if_recovering(&mut self, boundary: Option<Lsn>) {
+        if self.state.stream().is_some_and(|stream| stream.recovering) {
+            self.state.start_poll(&self.cancellation_token, boundary);
+        }
+    }
+
     /// Await a poll future that `poll_next` parked and its caller abandoned, taking the stream back.
-    async fn resume_parked_poll(&mut self) -> Option<Result<ChangeEvent>> {
+    async fn resume_parked_poll(&mut self) -> Option<Result<Option<ChangeEvent>>> {
         let StreamState::Polling(fut) = &mut self.state else {
             return None;
         };
@@ -2772,7 +2870,7 @@ impl futures_core::Stream for EventStream {
             cancellation_token,
             ..
         } = &mut *self;
-        state.start_poll(cancellation_token);
+        state.start_poll(cancellation_token, None);
 
         let StreamState::Polling(fut) = &mut self.state else {
             // Unreachable: `start_poll` parks anything that is Ready and not
@@ -2787,6 +2885,19 @@ impl futures_core::Stream for EventStream {
                 // was the last event decides `terminated`, so the state
                 // transition is part of each arm rather than a separate step.
                 let stream = Box::new(stream);
+                let result = match result {
+                    Ok(Some(event)) => Ok(event),
+                    // A dropped `next_event_before` parked a read that stopped at its boundary.
+                    Ok(None) => {
+                        self.state = StreamState::Ready {
+                            stream,
+                            terminated: false,
+                        };
+                        cx.waker().wake_by_ref();
+                        return std::task::Poll::Pending;
+                    }
+                    Err(e) => Err(e),
+                };
                 match result {
                     Ok(event) => {
                         self.state = StreamState::Ready {
@@ -5018,6 +5129,7 @@ pub(crate) mod tests {
             replay_stop: ReplayStop::Running,
             held: None,
             keepalive_wal_end: Lsn::new(0),
+            recovering: false,
         }
     }
 
@@ -5318,6 +5430,27 @@ pub(crate) mod tests {
         assert!(stream.check_server_identity(&identity(7682, 1)).is_ok());
     }
 
+    /// A second connection that lands on another cluster or timeline must not be used for the stream's positions.
+    #[test]
+    fn a_side_connection_must_report_the_streams_identity() {
+        let stream = identity(7682, 1);
+        assert!(stream.ensure_same_server(&identity(7682, 1)).is_ok());
+
+        for other in [identity(9999, 1), identity(7682, 2)] {
+            let err = stream.ensure_same_server(&other).unwrap_err();
+            assert!(
+                matches!(err, ReplicationError::PermanentConnection(_)),
+                "{err:?}"
+            );
+            assert!(err.is_permanent(), "{err}");
+            let msg = err.to_string();
+            assert!(
+                msg.contains("7682") && msg.contains(&other.system_id.to_string()),
+                "{msg}"
+            );
+        }
+    }
+
     /// A promoted standby keeps the primary's systemid, so only the timeline
     /// moves. This is the failover case, and it must be permanent: retrying
     /// would stream the post-fork history under LSNs the consumer already used.
@@ -5504,6 +5637,225 @@ pub(crate) mod tests {
         );
     }
 
+    /// A server that counts the connections it accepts and either holds them without ever
+    /// answering or, with `hang_up`, closes them at once. Returns its port and the count.
+    async fn counting_server(hang_up: bool) -> (u16, Arc<std::sync::atomic::AtomicUsize>) {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let accepted = Arc::new(AtomicUsize::new(0));
+        let count = Arc::clone(&accepted);
+        tokio::spawn(async move {
+            let mut held = Vec::new();
+            while let Ok((socket, _)) = listener.accept().await {
+                count.fetch_add(1, Ordering::SeqCst);
+                if !hang_up {
+                    held.push(socket);
+                }
+            }
+        });
+        (port, accepted)
+    }
+
+    fn conninfo_for(port: u16) -> String {
+        format!(
+            "postgresql://u@127.0.0.1:{port}/db?replication=database&sslmode=disable&connect_timeout=60"
+        )
+    }
+
+    /// A stream whose reads fail and whose reconnect hangs on a server that never answers.
+    /// Each connection the server counts is one connect attempt.
+    async fn stream_reconnecting_to_a_silent_server(
+    ) -> (EventStream, Arc<std::sync::atomic::AtomicUsize>) {
+        let (port, dialed) = counting_server(false).await;
+        let config = ReplicationStreamConfig::builder("slot", "pub")
+            .with_connection_timeout(Duration::from_secs(60));
+        let mut stream = create_test_stream(config);
+        stream.retry_handler =
+            ReplicationConnectionRetry::new(RetryConfig::default(), conninfo_for(port));
+        (stream.into_stream(CancellationToken::new()), dialed)
+    }
+
+    /// A caller's `timeout` shorter than the reconnect must not restart it: a dropped
+    /// read used to take the connect attempt with it, so the next read dialled again and
+    /// no reconnect could finish. The first read runs the reconnect inline and the second
+    /// parks it, so at most two dials, however many reads time out after that.
+    #[tokio::test]
+    async fn a_caller_timeout_does_not_restart_the_reconnect() {
+        let (mut events, dialed) = stream_reconnecting_to_a_silent_server().await;
+        let lsn = Lsn::new(0x100);
+
+        for round in 0..8 {
+            let read = Duration::from_millis(40);
+            // `next_event_before` shares the parking with `next_event`.
+            let timed_out = if round % 2 == 0 {
+                tokio::time::timeout(read, events.next_event())
+                    .await
+                    .is_err()
+            } else {
+                tokio::time::timeout(read, events.next_event_before(lsn))
+                    .await
+                    .is_err()
+            };
+            assert!(timed_out, "the reconnect hangs, so each read must time out");
+        }
+
+        let dials = dialed.load(std::sync::atomic::Ordering::SeqCst);
+        assert!(
+            (1..=2).contains(&dials),
+            "8 timed-out reads dialled {dials} times; the reconnect must survive them"
+        );
+    }
+
+    /// Cancelling ends a parked reconnect at once, so `shutdown()`, which awaits the
+    /// parked read, does not sit out the connect timeout.
+    #[tokio::test]
+    async fn shutdown_does_not_wait_for_a_parked_reconnect() {
+        let (mut events, _) = stream_reconnecting_to_a_silent_server().await;
+        for _ in 0..3 {
+            tokio::time::timeout(Duration::from_millis(40), events.next_event())
+                .await
+                .unwrap_err();
+        }
+
+        let shutdown = tokio::time::timeout(Duration::from_secs(10), events.shutdown()).await;
+
+        assert!(
+            shutdown.is_ok(),
+            "shutdown waited on the reconnect's connect timeout (60 s)"
+        );
+    }
+
+    /// A reconnect that the periodic health check starts is ended by the token too: with a
+    /// zero interval the first read runs the check, which finds the connection dead.
+    #[tokio::test]
+    async fn shutdown_does_not_wait_for_a_health_check_reconnect() {
+        let (port, _) = counting_server(false).await;
+        let config = ReplicationStreamConfig::builder("slot", "pub")
+            .with_connection_timeout(Duration::from_secs(60))
+            .with_health_check_interval(Duration::ZERO);
+        let mut stream = create_test_stream(config);
+        stream.retry_handler =
+            ReplicationConnectionRetry::new(RetryConfig::default(), conninfo_for(port));
+        let mut events = stream.into_stream(CancellationToken::new());
+        for _ in 0..3 {
+            tokio::time::timeout(Duration::from_millis(40), events.next_event())
+                .await
+                .unwrap_err();
+        }
+
+        let shutdown = tokio::time::timeout(Duration::from_secs(10), events.shutdown()).await;
+
+        assert!(
+            shutdown.is_ok(),
+            "shutdown waited on a health-check reconnect's connect timeout (60 s)"
+        );
+    }
+
+    /// A stream whose startup command was cancelled before the silent peer answered, which
+    /// leaves its connection refusing every later command.
+    #[cfg(feature = "rustls-tls")]
+    async fn stream_with_a_cancelled_startup_command(
+    ) -> (LogicalReplicationStream, std::net::TcpStream) {
+        let (connection, peer) = PgReplicationConnection::silent_for_testing();
+        let mut stream = stream_that_cannot_read_or_reconnect();
+        stream.connection = connection;
+        stream.identity = identity(7682, 2);
+        tokio::time::timeout(Duration::from_millis(50), stream.slot_confirmed_flush_lsn())
+            .await
+            .expect_err("the peer never answers");
+        (stream, peer)
+    }
+
+    /// The refusal a cancelled command leaves reads as retryable, yet nothing before `start`
+    /// replaces the connection: the pre-start commands must say the stream is to be rebuilt,
+    /// or a caller's retry loop spins on it forever.
+    #[cfg(feature = "rustls-tls")]
+    #[tokio::test(start_paused = true)]
+    async fn a_cancelled_startup_command_makes_the_next_one_a_permanent_error() {
+        let (mut stream, _peer) = stream_with_a_cancelled_startup_command().await;
+
+        let errors = [
+            stream.slot_confirmed_flush_lsn().await.unwrap_err(),
+            stream.timeline_history().await.unwrap_err(),
+            stream.start(None).await.unwrap_err(),
+        ];
+
+        for error in errors {
+            assert!(
+                matches!(&error, ReplicationError::PermanentConnection(m) if m.contains("cancelled") && m.contains("new stream")),
+                "{error:?}"
+            );
+            assert!(error.is_permanent());
+        }
+    }
+
+    /// Only the pre-start commands say "rebuild": a read must keep reconnecting a poisoned
+    /// connection, which a permanent error from the connection itself would have turned
+    /// into an immediate give-up.
+    #[cfg(feature = "rustls-tls")]
+    #[tokio::test(start_paused = true)]
+    async fn a_read_still_reconnects_after_a_cancelled_startup_command() {
+        let (mut stream, _peer) = stream_with_a_cancelled_startup_command().await;
+        let started = tokio::time::Instant::now();
+
+        let err = stream
+            .next_event_with_retry(&CancellationToken::new())
+            .await
+            .unwrap_err();
+
+        assert!(!err.is_permanent(), "{err:?}");
+        assert!(
+            started.elapsed() >= Duration::from_secs(3),
+            "gave up after {:?} instead of retrying through the backoff",
+            started.elapsed()
+        );
+    }
+
+    fn retries_three_times_quickly() -> RetryConfig {
+        RetryConfig {
+            max_attempts: 3,
+            initial_delay: Duration::from_millis(5),
+            ..RetryConfig::default()
+        }
+    }
+
+    /// `connection_timeout = Duration::ZERO` is "no timeout", so connecting still makes
+    /// every attempt `max_attempts` allows; it used to give up after the first with a
+    /// `Timeout` of `0ns`.
+    #[tokio::test]
+    async fn new_with_a_zero_connection_timeout_makes_every_connect_attempt() {
+        let (port, dialed) = counting_server(true).await;
+        let config = ReplicationStreamConfig::builder("slot", "pub")
+            .with_connection_timeout(Duration::ZERO)
+            .with_retry_config(retries_three_times_quickly());
+
+        let err = match LogicalReplicationStream::new(&conninfo_for(port), config).await {
+            Ok(_) => panic!("the server hangs up on every connection"),
+            Err(e) => e,
+        };
+
+        assert!(!matches!(err, ReplicationError::Timeout(_)), "{err:?}");
+        assert_eq!(dialed.load(std::sync::atomic::Ordering::SeqCst), 3);
+    }
+
+    /// The same for a reconnect.
+    #[tokio::test]
+    async fn recovery_with_a_zero_connection_timeout_makes_every_connect_attempt() {
+        let (port, dialed) = counting_server(true).await;
+        let config = ReplicationStreamConfig::builder("slot", "pub")
+            .with_connection_timeout(Duration::ZERO)
+            .with_retry_config(retries_three_times_quickly());
+        let mut stream = create_test_stream(config.clone());
+        stream.retry_handler =
+            ReplicationConnectionRetry::new(config.retry_config, conninfo_for(port));
+
+        let err = stream.recover_connection().await.unwrap_err();
+
+        assert!(!matches!(err, ReplicationError::Timeout(_)), "{err:?}");
+        assert_eq!(dialed.load(std::sync::atomic::Ordering::SeqCst), 3);
+    }
+
     /// 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<u8> {
@@ -9335,6 +9687,31 @@ pub(crate) mod tests {
         assert_eq!(held.event_type.commit_lsn(), Some(boundary));
     }
 
+    /// A read a dropped `next_event_before` parked can end at its boundary, which is no event: `next_event_before` reports the stop again,
+    /// and `next_event` and `poll_next` take the stream back and read on instead of returning one.
+    #[tokio::test]
+    async fn a_parked_read_that_stopped_at_its_boundary_is_not_an_event() {
+        let boundary = Lsn::new(0x200);
+        let mut events =
+            create_test_stream(create_test_config()).into_stream(CancellationToken::new());
+
+        events.state.park_resolved(Ok(None));
+        assert!(events.next_event_before(boundary).await.unwrap().is_none());
+        assert!(!events.state.is_polling(), "the stream must be taken back");
+
+        events.state.park_resolved(Ok(None));
+        let mut cx = std::task::Context::from_waker(std::task::Waker::noop());
+        assert!(
+            futures_core::Stream::poll_next(std::pin::Pin::new(&mut events), &mut cx).is_pending()
+        );
+        assert!(!events.state.is_polling(), "the stream must be taken back");
+
+        events.state.park_resolved(Ok(None));
+        // The read that follows has no server to reach; only that it began is checked.
+        let _ = tokio::time::timeout(Duration::from_millis(50), events.next_event()).await;
+        assert!(!events.state.is_polling(), "the stream must be taken back");
+    }
+
     /// 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]
@@ -10897,14 +11274,17 @@ mod startup_tests {
     use super::*;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
-    /// What happened when an operation raced a 50 ms outer timeout.
+    /// What happened when an operation raced the outer timeout.
     struct Raced {
         timed_out: bool,
         elapsed: Duration,
         heartbeat_ticks: usize,
     }
 
-    /// Race `op` against a 50 ms timeout beside a 5 ms heartbeat on a fresh runtime thread, panicking if that thread stays blocked for five seconds.
+    /// How long `op` is raced for. A loaded machine can stall the runtime thread for tens of milliseconds, and a stall longer than this wakes the timeout before the heartbeat has run once.
+    const OUTER_TIMEOUT: Duration = Duration::from_millis(250);
+
+    /// Race `op` against `OUTER_TIMEOUT` beside a 5 ms heartbeat on a fresh runtime thread, panicking if that thread stays blocked for five seconds.
     fn race_outer_timeout<F, Fut>(multi_thread: bool, op: F) -> Raced
     where
         F: FnOnce() -> Fut + Send + 'static,
@@ -10934,7 +11314,7 @@ mod startup_tests {
                 });
                 tokio::task::yield_now().await;
                 let started = Instant::now();
-                let result = tokio::time::timeout(Duration::from_millis(50), op()).await;
+                let result = tokio::time::timeout(OUTER_TIMEOUT, op()).await;
                 let elapsed = started.elapsed();
                 heartbeat.abort();
                 (result.is_err(), elapsed)
@@ -10956,8 +11336,9 @@ mod startup_tests {
             "the operation finished before the outer timeout, after {:?}",
             raced.elapsed
         );
+        // A blocked thread would have waited out the 2 s `connect_timeout` first.
         assert!(
-            raced.elapsed < Duration::from_millis(500),
+            raced.elapsed < OUTER_TIMEOUT + Duration::from_millis(1000),
             "the outer timeout fired late, after {:?}",
             raced.elapsed
         );
diff --git a/src/types.rs b/src/types.rs
index d5f9cc8..fb8c06e 100644
--- a/src/types.rs
+++ b/src/types.rs
@@ -647,8 +647,9 @@ pub enum EventType {
     ///
     /// With `subtransaction_xid == transaction_id` the whole streamed transaction
     /// is rolled back. Otherwise only that subtransaction is, and its changes
-    /// are the ones whose [`ChangeEvent::xid`] equals `subtransaction_xid`. Rolling
-    /// back nested savepoints sends one `StreamAbort` per subtransaction.
+    /// are the ones whose [`ChangeEvent::xid`] equals `subtransaction_xid`, except
+    /// a `Message`, which always carries the top-level xid. Rolling back nested
+    /// savepoints sends one `StreamAbort` per subtransaction.
     StreamAbort {
         transaction_id: u32,
         subtransaction_xid: u32,
@@ -802,10 +803,11 @@ pub struct ChangeEvent {
     /// The xid pgoutput wrote before this change inside a streamed chunk.
     ///
     /// It is the subtransaction's xid for a change made after a savepoint, and
-    /// the top-level transaction's otherwise. A `StreamAbort` whose
+    /// the top-level transaction's otherwise, except for a `Message`, which
+    /// always carries the top-level xid. A `StreamAbort` whose
     /// `subtransaction_xid` differs from its `transaction_id` rolls back the
-    /// changes with that `xid`. `None` outside a streamed chunk and for events
-    /// that are not changes.
+    /// changes with that `xid`, so it never names a `Message`. `None` outside a
+    /// streamed chunk and for events that are not changes.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub xid: Option<u32>,
 }
@@ -2124,6 +2126,107 @@ mod tests {
         assert_eq!(decoded.metadata, event.metadata);
     }
 
+    /// One event per payload shape serde has to carry: row data, `Option`s, `Bytes`, timestamps and unit variants.
+    fn serde_samples() -> Vec<EventType> {
+        let ts = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap();
+        let data = RowData::from_pairs(vec![
+            ("id", ColumnValue::text("1")),
+            ("note", ColumnValue::Null),
+        ]);
+        vec![
+            EventType::Insert {
+                schema: Arc::from("public"),
+                table: Arc::from("t"),
+                relation_oid: 7,
+                data: data.clone(),
+            },
+            EventType::Update {
+                schema: Arc::from("public"),
+                table: Arc::from("t"),
+                relation_oid: 7,
+                old_data: None,
+                new_data: data,
+                replica_identity: ReplicaIdentity::Default,
+                key_columns: vec![Arc::from("id")],
+            },
+            EventType::Message {
+                flags: 1,
+                message_lsn: Lsn::new(u64::MAX),
+                prefix: Arc::from("p"),
+                content: Bytes::from_static(b"c"),
+            },
+            EventType::Commit {
+                commit_timestamp: ts,
+                commit_lsn: Lsn::new(0x100),
+                end_lsn: Lsn::new(0x200),
+            },
+            EventType::StreamAbort {
+                transaction_id: 7,
+                subtransaction_xid: 8,
+                abort_lsn: Some(Lsn::new(0x100)),
+                abort_timestamp: None,
+            },
+            EventType::StreamStop,
+        ]
+    }
+
+    fn event_with_xid(event_type: EventType, xid: Option<u32>) -> ChangeEvent {
+        ChangeEvent {
+            event_type,
+            lsn: Lsn::new(0x10),
+            metadata: None,
+            xid,
+        }
+    }
+
+    /// `xid` survives a `serde_json` round trip as `None`, a subtransaction's xid and `u32::MAX`, on every payload shape.
+    #[test]
+    fn change_event_xid_round_trips_through_json() {
+        for event_type in serde_samples() {
+            for xid in [None, Some(746), Some(u32::MAX)] {
+                let event = event_with_xid(event_type.clone(), xid);
+                let json = serde_json::to_vec(&event).unwrap();
+                let decoded: ChangeEvent = serde_json::from_slice(&json).unwrap();
+                assert_eq!(decoded.xid, xid, "{event_type:?}");
+                assert_eq!(decoded.event_type, event_type);
+                assert_eq!(decoded.lsn, event.lsn);
+            }
+        }
+    }
+
+    /// JSON written before `xid` existed has no such key and must still deserialize, with `xid` as `None`.
+    #[test]
+    fn json_without_an_xid_key_deserializes_with_no_xid() {
+        let v090 = r#"{"event_type":"StreamStop","lsn":16,"metadata":null}"#;
+        let decoded: ChangeEvent = serde_json::from_str(v090).unwrap();
+        assert_eq!(decoded.event_type, EventType::StreamStop);
+        assert_eq!(decoded.xid, None);
+
+        for event_type in serde_samples() {
+            let mut json =
+                serde_json::to_value(event_with_xid(event_type.clone(), Some(746))).unwrap();
+            assert_eq!(json["xid"], 746, "{event_type:?}");
+            json.as_object_mut().unwrap().remove("xid");
+            let decoded: ChangeEvent = serde_json::from_value(json).unwrap();
+            assert_eq!(decoded.xid, None, "{event_type:?}");
+            assert_eq!(decoded.event_type, event_type);
+        }
+    }
+
+    /// An event without an `xid` serializes with no `xid` key, so it is byte-identical to the JSON 0.9 wrote.
+    #[test]
+    fn a_missing_xid_is_not_serialized() {
+        let v090 = r#"{"event_type":"StreamStop","lsn":16,"metadata":null}"#;
+        assert_eq!(
+            serde_json::to_string(&event_with_xid(EventType::StreamStop, None)).unwrap(),
+            v090
+        );
+        for event_type in serde_samples() {
+            let json = serde_json::to_value(event_with_xid(event_type.clone(), None)).unwrap();
+            assert!(json.get("xid").is_none(), "{event_type:?}: {json}");
+        }
+    }
+
     #[test]
     fn test_lsn_zero() {
         let lsn = Lsn::new(0);