Merge master into next (FSP rekey overlapping-epoch, drain-erase fix)

src/node/session.rs

@@ -40,6 +40,20 @@ pub(crate) enum EndToEndState {
     Established(NoiseSession),
 }
 
+/// Which key epoch a frame decrypted against in the trial-decrypt
+/// cascade. Reported by [`SessionEntry::fsp_trial_decrypt`] so the
+/// receive path can react to a non-`Current` epoch.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum EpochSlot {
+    /// The current (active) session — steady-state traffic.
+    Current,
+    /// The pending (new, not yet promoted) session — the peer cut over
+    /// before this endpoint did.
+    Pending,
+    /// The previous (draining) session — old-epoch stragglers.
+    Previous,
+}
+
 impl EndToEndState {
     /// Check if the session is established and ready for data.
     pub(crate) fn is_established(&self) -> bool {
@@ -117,6 +131,20 @@ pub(crate) struct SessionEntry {
     previous_noise_session: Option<NoiseSession>,
     /// When drain window started (Unix ms). 0 = no drain.
     drain_started_ms: u64,
+    /// Last time an inbound frame authenticated against the `previous`
+    /// slot (Unix ms). 0 = the `previous` slot has not been used since
+    /// the drain began.
+    ///
+    /// Drives peer-progress-aware retirement of the old epoch: the drain
+    /// window does not expire until `DRAIN_WINDOW_SECS` have elapsed
+    /// since the LATER of the cutover and the last `previous`-slot
+    /// decrypt. The old epoch therefore lives as long as the peer is
+    /// still transmitting on it — closing the rep-003 gap where the
+    /// initiator erased the `previous` slot on a fixed wall-clock timer
+    /// while the peer (having never received msg3) was still sealing
+    /// every frame in the old epoch, leaving the trial-decrypt cascade
+    /// with no slot that could decrypt them.
+    previous_last_used_ms: u64,
     /// In-progress rekey state (runs alongside Established session).
     rekey_state: Option<HandshakeState>,
     /// Pending completed session awaiting K-bit cutover.
@@ -126,8 +154,27 @@ pub(crate) struct SessionEntry {
     /// Dampening: last time peer sent us a rekey msg1 (Unix ms).
     last_peer_rekey_ms: u64,
     /// When the FSP rekey handshake completed (initiator sent msg3, Unix ms).
-    /// Used to defer cutover until msg3 has time to reach the responder.
+    /// Drives the initiator's liveness-bound cutover timer. Cleared on
+    /// cutover. The timer is no longer safety-critical: overlapping-epoch
+    /// trial-decrypt covers any cutover skew. It only bounds how long the
+    /// initiator advertises the old K-bit.
     rekey_completed_ms: u64,
+    /// Encoded SessionMsg3 payload retained for retransmission (initiator).
+    /// Set when the rekey initiator sends msg3; cleared once the responder
+    /// is confirmed on the new epoch (an authenticated peer frame against
+    /// `pending` or new `current`) or the rekey cycle is abandoned.
+    /// Retransmission lifetime is tied to responder reception of msg3,
+    /// decoupled from the initiator's own cutover.
+    rekey_msg3_payload: Option<Vec<u8>>,
+    /// When the next rekey msg3 retransmission should fire (Unix ms).
+    /// 0 = no retransmission scheduled.
+    rekey_msg3_next_resend_ms: u64,
+    /// Number of rekey msg3 retransmissions performed so far.
+    rekey_msg3_resend_count: u32,
+    /// Whether the rekey peer has been observed on the new epoch — set
+    /// when an inbound frame authenticates against `pending` or against
+    /// the post-cutover `current`. Stops msg3 retransmission.
+    peer_new_epoch_confirmed: bool,
     /// Per-session symmetric jitter applied to the rekey timer trigger.
     /// Drawn once at construction (and at each cutover) uniformly from
     /// `[-REKEY_JITTER_SECS, +REKEY_JITTER_SECS]`. Desynchronizes
@@ -165,11 +212,16 @@ impl SessionEntry {
             current_k_bit: false,
             previous_noise_session: None,
             drain_started_ms: 0,
+            previous_last_used_ms: 0,
             rekey_state: None,
             pending_new_session: None,
             rekey_initiator: false,
             last_peer_rekey_ms: 0,
             rekey_completed_ms: 0,
+            rekey_msg3_payload: None,
+            rekey_msg3_next_resend_ms: 0,
+            rekey_msg3_resend_count: 0,
+            peer_new_epoch_confirmed: false,
             rekey_jitter_secs: draw_rekey_jitter(),
         }
     }
@@ -371,11 +423,6 @@ impl SessionEntry {
         self.pending_new_session.as_ref()
     }
 
-    /// Get the previous session for decryption fallback during drain.
-    pub(crate) fn previous_noise_session_mut(&mut self) -> Option<&mut NoiseSession> {
-        self.previous_noise_session.as_mut()
-    }
-
     /// Whether we initiated the current rekey.
     pub(crate) fn is_rekey_initiator(&self) -> bool {
         self.rekey_initiator
@@ -427,6 +474,147 @@ impl SessionEntry {
         self.rekey_completed_ms = ms;
     }
 
+    // === Rekey msg3 Retransmission (Initiator, liveness-only) ===
+
+    /// Retain the encoded SessionMsg3 payload for retransmission and
+    /// schedule the first resend. Called by the rekey initiator after
+    /// sending msg3.
+    ///
+    /// Retransmission is a liveness mechanism: the responder must
+    /// eventually receive msg3 to derive the new session. It runs until
+    /// the peer is confirmed on the new epoch, independent of the
+    /// initiator's own cutover.
+    pub(crate) fn set_rekey_msg3_payload(&mut self, payload: Vec<u8>, next_resend_at_ms: u64) {
+        self.rekey_msg3_payload = Some(payload);
+        self.rekey_msg3_next_resend_ms = next_resend_at_ms;
+        self.rekey_msg3_resend_count = 0;
+        self.peer_new_epoch_confirmed = false;
+    }
+
+    /// Get the retained rekey msg3 payload for retransmission.
+    pub(crate) fn rekey_msg3_payload(&self) -> Option<&[u8]> {
+        self.rekey_msg3_payload.as_deref()
+    }
+
+    /// When the next rekey msg3 retransmission should fire (Unix ms).
+    pub(crate) fn rekey_msg3_next_resend_ms(&self) -> u64 {
+        self.rekey_msg3_next_resend_ms
+    }
+
+    /// Number of rekey msg3 retransmissions performed so far.
+    pub(crate) fn rekey_msg3_resend_count(&self) -> u32 {
+        self.rekey_msg3_resend_count
+    }
+
+    /// Record a rekey msg3 retransmission and schedule the next one.
+    pub(crate) fn record_rekey_msg3_resend(&mut self, next_resend_at_ms: u64) {
+        self.rekey_msg3_resend_count += 1;
+        self.rekey_msg3_next_resend_ms = next_resend_at_ms;
+    }
+
+    /// Clear the retained rekey msg3 payload (responder confirmed on the
+    /// new epoch, or the cycle abandoned).
+    pub(crate) fn clear_rekey_msg3_payload(&mut self) {
+        self.rekey_msg3_payload = None;
+        self.rekey_msg3_next_resend_ms = 0;
+        self.rekey_msg3_resend_count = 0;
+    }
+
+    /// Whether the rekey peer has been observed on the new epoch.
+    #[cfg(test)]
+    pub(crate) fn peer_new_epoch_confirmed(&self) -> bool {
+        self.peer_new_epoch_confirmed
+    }
+
+    /// Mark the rekey peer as confirmed on the new epoch and stop msg3
+    /// retransmission. Called when an inbound frame authenticates against
+    /// the `pending` or post-cutover `current` session.
+    pub(crate) fn confirm_peer_new_epoch(&mut self) {
+        self.peer_new_epoch_confirmed = true;
+        self.clear_rekey_msg3_payload();
+    }
+
+    // === Overlapping-epoch trial-decrypt slots ===
+
+    /// Mutable access to the current (active) `NoiseSession`, if established.
+    pub(crate) fn current_noise_session_mut(&mut self) -> Option<&mut NoiseSession> {
+        match self.state.as_mut() {
+            Some(EndToEndState::Established(s)) => Some(s),
+            _ => None,
+        }
+    }
+
+    /// Trial-decrypt an encrypted FSP frame against every live key epoch.
+    ///
+    /// During a rekey window an endpoint may legitimately receive frames
+    /// sealed in up to three epochs: `current` (steady state), `pending`
+    /// (the peer cut over before this endpoint did), and `previous`
+    /// (drain-window stragglers sealed in the old epoch). This cascade
+    /// makes the receive path able to decrypt any of them, so no cutover
+    /// ordering and no packet reordering can cause a decrypt failure.
+    ///
+    /// The received K-bit is an ordering hint only — it selects which
+    /// slot to try first to save one cheap `check()` rejection on the
+    /// common cut-over-in-progress packet. Correctness never depends on
+    /// it.
+    ///
+    /// Continues the cascade on **any** error, including
+    /// `ReplayDetected`: a replay rejection from a non-matching slot is
+    /// just "wrong key, try the next one." This is replay-safe by
+    /// construction — `decrypt_with_replay_check_and_aad` only mutates a
+    /// slot's `ReplayWindow` after a successful decrypt, so a failed
+    /// trial against the wrong slot leaves that slot untouched. Only the
+    /// slot that authenticates the packet advances its window.
+    ///
+    /// A successful decrypt against the `previous` slot refreshes the
+    /// drain deadline (`refresh_previous_use`): the old epoch must stay
+    /// retained as long as the peer is still sealing frames in it. See
+    /// [`SessionEntry::drain_expired`] for why.
+    ///
+    /// Returns the plaintext plus the slot it decrypted against, or
+    /// `None` if every live slot failed (a genuine drop).
+    pub(crate) fn fsp_trial_decrypt(
+        &mut self,
+        ciphertext: &[u8],
+        counter: u64,
+        aad: &[u8],
+        received_k_bit: bool,
+        now_ms: u64,
+    ) -> Option<(Vec<u8>, EpochSlot)> {
+        // Hint: if the received K-bit differs from our current epoch and
+        // we hold a pending session, the peer has likely cut over — try
+        // `pending` first. Pure optimisation; the cascade still tries
+        // every slot regardless.
+        let pending_first =
+            received_k_bit != self.current_k_bit && self.pending_new_session.is_some();
+
+        let order: [EpochSlot; 3] = if pending_first {
+            [EpochSlot::Pending, EpochSlot::Current, EpochSlot::Previous]
+        } else {
+            [EpochSlot::Current, EpochSlot::Pending, EpochSlot::Previous]
+        };
+
+        for slot in order {
+            let session = match slot {
+                EpochSlot::Current => self.current_noise_session_mut(),
+                EpochSlot::Pending => self.pending_new_session.as_mut(),
+                EpochSlot::Previous => self.previous_noise_session.as_mut(),
+            };
+            if let Some(session) = session
+                && let Ok(pt) = session.decrypt_with_replay_check_and_aad(ciphertext, counter, aad)
+            {
+                // The peer is still transmitting on the old epoch:
+                // push the drain deadline out so the `previous` slot
+                // is not retired out from under a peer still using it.
+                if slot == EpochSlot::Previous {
+                    self.refresh_previous_use(now_ms);
+                }
+                return Some((pt, slot));
+            }
+        }
+        None
+    }
+
     /// Store a completed rekey session.
     pub(crate) fn set_pending_session(&mut self, session: NoiseSession) {
         self.pending_new_session = Some(session);
@@ -444,11 +632,14 @@ impl SessionEntry {
         self.rekey_state.take()
     }
 
-    /// Cut over to the pending new session (initiator side).
+    /// Promote the pending session to current (the cutover).
     ///
-    /// Moves current session to previous (for drain), promotes pending to current,
-    /// flips the K-bit.
-    pub(crate) fn cutover_to_new_session(&mut self, now_ms: u64) -> bool {
+    /// Moves the current session to `previous` (for drain), promotes
+    /// `pending` to `current`, flips the K-bit, stamps the drain window.
+    /// Shared by both cutover triggers: the initiator's liveness timer
+    /// (`cutover_to_new_session`) and a successful `pending` trial-decrypt
+    /// (`handle_peer_kbit_flip`).
+    fn promote_pending(&mut self, now_ms: u64) -> bool {
         let new_session = match self.pending_new_session.take() {
             Some(s) => s,
             None => return false,
@@ -459,6 +650,10 @@ impl SessionEntry {
             self.previous_noise_session = Some(old);
         }
         self.drain_started_ms = now_ms;
+        // Fresh drain: no `previous`-slot use observed yet. The drain
+        // deadline starts from the cutover and is pushed out by any
+        // subsequent `previous`-slot decrypt.
+        self.previous_last_used_ms = 0;
 
         // Promote pending to current
         self.state = Some(EndToEndState::Established(new_session));
@@ -477,38 +672,53 @@ impl SessionEntry {
         true
     }
 
-    /// Handle receiving a K-bit flip from the peer (responder side).
+    /// Cut over to the pending new session on the initiator's liveness
+    /// timer.
+    ///
+    /// This is the unconditional timer-driven cutover (option A). It is
+    /// safe on any schedule: overlapping-epoch trial-decrypt covers the
+    /// skew between the two endpoints' cutovers. It does NOT confirm the
+    /// peer is on the new epoch — the responder may still be awaiting
+    /// msg3 — so msg3 retransmission must continue past this cutover.
+    pub(crate) fn cutover_to_new_session(&mut self, now_ms: u64) -> bool {
+        self.promote_pending(now_ms)
+    }
+
+    /// Promote the pending session because a peer frame authenticated
+    /// against it.
+    ///
+    /// A frame that decrypts against `pending` is itself the cutover
+    /// signal — proof the peer derived the new session and moved to it.
+    /// This unifies the trigger: the header K-bit is only a hint, the
+    /// authenticated decrypt is the actual proof. Used by the
+    /// trial-decrypt cascade on a successful `pending` decrypt.
     pub(crate) fn handle_peer_kbit_flip(&mut self, now_ms: u64) -> bool {
-        let new_session = match self.pending_new_session.take() {
-            Some(s) => s,
-            None => return false,
-        };
-
-        // Demote current to previous for drain
-        if let Some(EndToEndState::Established(old)) = self.state.take() {
-            self.previous_noise_session = Some(old);
-        }
-        self.drain_started_ms = now_ms;
-
-        // Promote pending to current
-        self.state = Some(EndToEndState::Established(new_session));
-        self.current_k_bit = !self.current_k_bit;
-        self.session_start_ms = now_ms;
-        self.rekey_state = None;
-        self.rekey_initiator = false;
-        self.rekey_jitter_secs = draw_rekey_jitter();
-
-        // Reset MMP counters to avoid metric discontinuity
-        let now = Instant::now();
-        if let Some(mmp) = &mut self.mmp {
-            mmp.reset_for_rekey(now);
-        }
-        true
+        self.promote_pending(now_ms)
     }
 
     /// Check if the drain window has expired.
+    ///
+    /// Peer-progress-aware: the deadline is `drain_ms` after the LATER
+    /// of the cutover (`drain_started_ms`) and the last inbound frame
+    /// that authenticated against the `previous` slot
+    /// (`previous_last_used_ms`). The old epoch is therefore retired
+    /// only once the peer has been silent on it for a full drain
+    /// window — never while the peer is still transmitting on it.
+    ///
+    /// This closes the rep-003 gap: a fixed wall-clock drain timer,
+    /// started unilaterally at the initiator's Option-A cutover, would
+    /// erase the `previous` slot 10 s later even if the peer (having
+    /// lost msg3) was still sealing every frame in the old epoch — the
+    /// trial-decrypt cascade then had no slot to decrypt them, a
+    /// permanent silent decrypt failure. A peer that never catches up
+    /// is instead handled by the FSP session liveness path (fresh
+    /// handshake / teardown of a genuinely dead link).
     pub(crate) fn drain_expired(&self, now_ms: u64, drain_ms: u64) -> bool {
-        self.drain_started_ms > 0 && now_ms.saturating_sub(self.drain_started_ms) >= drain_ms
+        if self.drain_started_ms == 0 {
+            return false;
+        }
+        let deadline_anchor = self.drain_started_ms.max(self.previous_last_used_ms);
+        now_ms.saturating_sub(deadline_anchor) >= drain_ms
     }
 
     /// Whether a drain is in progress.
@@ -516,16 +726,540 @@ impl SessionEntry {
         self.drain_started_ms > 0
     }
 
+    /// Refresh the drain deadline because an inbound frame authenticated
+    /// against the `previous` slot — the peer is still using the old
+    /// epoch, so it must stay retained. No-op if no drain is in
+    /// progress (a `previous` slot installed only for test purposes).
+    pub(crate) fn refresh_previous_use(&mut self, now_ms: u64) {
+        if self.drain_started_ms > 0 {
+            self.previous_last_used_ms = now_ms;
+        }
+    }
+
     /// Complete the drain: drop previous session.
     pub(crate) fn complete_drain(&mut self) {
         self.previous_noise_session = None;
         self.drain_started_ms = 0;
+        self.previous_last_used_ms = 0;
     }
 
     /// Abandon an in-progress rekey.
+    ///
+    /// Drops the in-flight handshake state, the pending session, and any
+    /// retained msg3 retransmission payload, returning the entry to a
+    /// clean `Established` state. Safe under overlapping-epoch decrypt:
+    /// an abandoned cycle never leaves a divergent unsafe state — the
+    /// endpoints simply stay on `current` until a later cycle completes.
     pub(crate) fn abandon_rekey(&mut self) {
         self.rekey_state = None;
         self.pending_new_session = None;
         self.rekey_initiator = false;
+        self.rekey_completed_ms = 0;
+        self.clear_rekey_msg3_payload();
+        self.peer_new_epoch_confirmed = false;
+    }
+
+    // === Test-only helpers ===
+
+    /// Install a session directly in the `previous` (draining) slot.
+    #[cfg(test)]
+    pub(crate) fn set_previous_session_for_test(&mut self, session: NoiseSession, now_ms: u64) {
+        self.previous_noise_session = Some(session);
+        self.drain_started_ms = now_ms;
+    }
+
+    /// Read the highest received counter of the `previous` slot, if any.
+    #[cfg(test)]
+    pub(crate) fn previous_highest_counter(&self) -> Option<u64> {
+        self.previous_noise_session
+            .as_ref()
+            .map(|s| s.highest_received_counter())
+    }
+
+    /// Read the highest received counter of the `pending` slot, if any.
+    #[cfg(test)]
+    pub(crate) fn pending_highest_counter(&self) -> Option<u64> {
+        self.pending_new_session
+            .as_ref()
+            .map(|s| s.highest_received_counter())
+    }
+
+    /// Read the highest received counter of the `current` slot, if
+    /// established.
+    #[cfg(test)]
+    pub(crate) fn current_highest_counter(&self) -> Option<u64> {
+        match self.state.as_ref() {
+            Some(EndToEndState::Established(s)) => Some(s.highest_received_counter()),
+            _ => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod overlapping_epoch_tests {
+    use super::*;
+    use crate::node::session_wire::{FSP_FLAG_K, build_fsp_header};
+    use crate::noise::HandshakeState;
+    use secp256k1::{Keypair, Secp256k1, SecretKey};
+
+    /// Deterministic keypair from a single seed byte.
+    fn keypair(seed: u8) -> Keypair {
+        let secp = Secp256k1::new();
+        let mut bytes = [1u8; 32];
+        bytes[0] = seed;
+        let sk = SecretKey::from_slice(&bytes).expect("valid secret key");
+        Keypair::from_secret_key(&secp, &sk)
+    }
+
+    /// Run a full XX handshake and return `(initiator_session,
+    /// responder_session)` — a paired sender/receiver.
+    fn xk_pair(init_seed: u8, resp_seed: u8) -> (NoiseSession, NoiseSession) {
+        let init_kp = keypair(init_seed);
+        let resp_kp = keypair(resp_seed);
+        let mut initiator = HandshakeState::new_initiator(init_kp);
+        initiator.set_local_epoch([0xA1, 0xB2, 0xC3, 0xD4, 0x11, 0x22, 0x33, 0x44]);
+        let mut responder = HandshakeState::new_responder(resp_kp);
+        responder.set_local_epoch([0xD4, 0xC3, 0xB2, 0xA1, 0x44, 0x33, 0x22, 0x11]);
+
+        let msg1 = initiator.write_message_1().unwrap();
+        responder.read_message_1(&msg1).unwrap();
+        let msg2 = responder.write_message_2().unwrap();
+        initiator.read_message_2(&msg2).unwrap();
+        let msg3 = initiator.write_message_3().unwrap();
+        responder.read_message_3(&msg3).unwrap();
+
+        (
+            initiator.into_session().unwrap(),
+            responder.into_session().unwrap(),
+        )
+    }
+
+    /// Seal an FSP frame: returns `(ciphertext, counter, header_bytes,
+    /// k_bit)` exactly as the send path produces them.
+    fn seal(sender: &mut NoiseSession, plaintext: &[u8], k_bit: bool) -> (Vec<u8>, u64, [u8; 12]) {
+        let counter = sender.current_send_counter();
+        let flags = if k_bit { FSP_FLAG_K } else { 0 };
+        let header = build_fsp_header(counter, flags, plaintext.len() as u16);
+        let ciphertext = sender.encrypt_with_aad(plaintext, &header).unwrap();
+        (ciphertext, counter, header)
+    }
+
+    /// Build a `SessionEntry` whose `current` slot is `session`.
+    fn entry_with_current(session: NoiseSession) -> SessionEntry {
+        let addr = NodeAddr::from_bytes([7u8; 16]);
+        let pubkey = keypair(99).public_key();
+        let mut entry = SessionEntry::new(
+            addr,
+            pubkey,
+            EndToEndState::Established(session),
+            1_000,
+            true,
+        );
+        entry.mark_established(1_000);
+        entry
+    }
+
+    // 1. A frame sealed in `current` decrypts against `current`; the
+    //    `pending` and `previous` windows are left untouched.
+    #[test]
+    fn trial_decrypt_picks_current() {
+        let (mut cur_send, cur_recv) = xk_pair(1, 2);
+        let (_p_send, p_recv) = xk_pair(3, 4);
+        let (_o_send, o_recv) = xk_pair(5, 6);
+
+        let mut entry = entry_with_current(cur_recv);
+        entry.set_pending_session(p_recv);
+        entry.set_previous_session_for_test(o_recv, 1_000);
+
+        let (ct, counter, hdr) = seal(&mut cur_send, b"steady-state", false);
+        let (pt, slot) = entry
+            .fsp_trial_decrypt(&ct, counter, &hdr, false, 2_000)
+            .expect("current frame must decrypt");
+
+        assert_eq!(pt, b"steady-state");
+        assert_eq!(slot, EpochSlot::Current);
+        assert_eq!(entry.pending_highest_counter(), Some(0));
+        assert_eq!(entry.previous_highest_counter(), Some(0));
+    }
+
+    // 2. A frame sealed in `pending` decrypts via the cascade and the
+    //    entry promotes pending -> current, current -> previous, K-bit
+    //    flips.
+    #[test]
+    fn trial_decrypt_picks_pending_and_promotes() {
+        let (_cur_send, cur_recv) = xk_pair(1, 2);
+        let (mut p_send, p_recv) = xk_pair(3, 4);
+
+        let mut entry = entry_with_current(cur_recv);
+        let k_before = entry.current_k_bit();
+        entry.set_pending_session(p_recv);
+
+        // Peer sealed in the new epoch with the flipped K-bit.
+        let (ct, counter, hdr) = seal(&mut p_send, b"new-epoch", !k_before);
+        let (pt, slot) = entry
+            .fsp_trial_decrypt(&ct, counter, &hdr, !k_before, 2_000)
+            .expect("pending frame must decrypt");
+        assert_eq!(pt, b"new-epoch");
+        assert_eq!(slot, EpochSlot::Pending);
+
+        // Receive path promotes on a pending hit.
+        entry.handle_peer_kbit_flip(2_000);
+        assert!(entry.pending_new_session().is_none());
+        assert!(entry.previous_highest_counter().is_some());
+        assert_ne!(entry.current_k_bit(), k_before);
+    }
+
+    // 3. After cutover, an old-epoch frame decrypts against `previous`
+    //    with no state change.
+    #[test]
+    fn trial_decrypt_picks_previous_during_drain() {
+        let (mut old_send, old_recv) = xk_pair(1, 2);
+        let (_new_send, new_recv) = xk_pair(3, 4);
+
+        // Start with the old session, install the new as pending, cut over.
+        let mut entry = entry_with_current(new_recv);
+        entry.set_previous_session_for_test(old_recv, 1_500);
+        let k_after = entry.current_k_bit();
+
+        // Old-epoch straggler still in flight after our cutover.
+        let (ct, counter, hdr) = seal(&mut old_send, b"old-straggler", !k_after);
+        let (pt, slot) = entry
+            .fsp_trial_decrypt(&ct, counter, &hdr, !k_after, 3_000)
+            .expect("previous frame must decrypt");
+
+        assert_eq!(pt, b"old-straggler");
+        assert_eq!(slot, EpochSlot::Previous);
+        // No promotion, no K-bit change.
+        assert_eq!(entry.current_k_bit(), k_after);
+        assert!(entry.is_draining());
+    }
+
+    // 4. After a pending-driven promotion, an old-epoch straggler still
+    //    decrypts (against the now-`previous` slot) — the reordering
+    //    case that broke attempt 1.
+    #[test]
+    fn trial_decrypt_reordered_old_after_cutover() {
+        let (mut cur_send, cur_recv) = xk_pair(1, 2);
+        let (mut p_send, p_recv) = xk_pair(3, 4);
+
+        let mut entry = entry_with_current(cur_recv);
+        let k_before = entry.current_k_bit();
+        entry.set_pending_session(p_recv);
+
+        // New-epoch frame promotes the entry.
+        let (ct_new, c_new, hdr_new) = seal(&mut p_send, b"after-cutover", !k_before);
+        let (_pt, slot) = entry
+            .fsp_trial_decrypt(&ct_new, c_new, &hdr_new, !k_before, 2_000)
+            .unwrap();
+        assert_eq!(slot, EpochSlot::Pending);
+        entry.handle_peer_kbit_flip(2_000);
+
+        // Now an OLD-epoch straggler arrives reordered after cutover.
+        let (ct_old, c_old, hdr_old) = seal(&mut cur_send, b"reordered-old", k_before);
+        let (pt, slot) = entry
+            .fsp_trial_decrypt(&ct_old, c_old, &hdr_old, k_before, 2_500)
+            .expect("reordered old-epoch frame must still decrypt");
+        assert_eq!(pt, b"reordered-old");
+        assert_eq!(slot, EpochSlot::Previous);
+    }
+
+    // 5. A genuine replay of a `current`-epoch counter is rejected (all
+    //    slots fail) and does not corrupt the other windows; a `current`
+    //    replay while `pending` holds the real key still authenticates
+    //    against `pending`.
+    #[test]
+    fn trial_decrypt_replay_is_per_slot() {
+        let (mut cur_send, cur_recv) = xk_pair(1, 2);
+        let (mut p_send, p_recv) = xk_pair(3, 4);
+
+        let mut entry = entry_with_current(cur_recv);
+        let k_before = entry.current_k_bit();
+        entry.set_pending_session(p_recv);
+
+        // First delivery on `current`.
+        let (ct, counter, hdr) = seal(&mut cur_send, b"first", k_before);
+        let (_pt, slot) = entry
+            .fsp_trial_decrypt(&ct, counter, &hdr, k_before, 2_000)
+            .unwrap();
+        assert_eq!(slot, EpochSlot::Current);
+
+        // Replaying the exact same `current` frame: all slots fail.
+        assert!(
+            entry
+                .fsp_trial_decrypt(&ct, counter, &hdr, k_before, 2_100)
+                .is_none(),
+            "a genuine replay must be rejected by every slot"
+        );
+        // The pending window must not have been corrupted by the failed
+        // trials.
+        assert_eq!(entry.pending_highest_counter(), Some(0));
+
+        // A pending-epoch frame sharing counter 0 with `current` (whose
+        // counter 0 is already consumed) still authenticates against
+        // `pending`.
+        let (ct_p, c_p, hdr_p) = seal(&mut p_send, b"pending-c0", !k_before);
+        assert_eq!(c_p, 0);
+        let (pt, slot) = entry
+            .fsp_trial_decrypt(&ct_p, c_p, &hdr_p, !k_before, 2_200)
+            .expect("pending frame must decrypt despite current replay overlap");
+        assert_eq!(pt, b"pending-c0");
+        assert_eq!(slot, EpochSlot::Pending);
+    }
+
+    // 6. A failed trial against a non-winning slot leaves that slot's
+    //    ReplayWindow::highest() unchanged (the section-4 invariant).
+    #[test]
+    fn trial_decrypt_failed_slot_leaves_replay_window_intact() {
+        let (_cur_send, cur_recv) = xk_pair(1, 2);
+        let (mut p_send, p_recv) = xk_pair(3, 4);
+        let (_o_send, o_recv) = xk_pair(5, 6);
+
+        let mut entry = entry_with_current(cur_recv);
+        let k_before = entry.current_k_bit();
+        entry.set_pending_session(p_recv);
+        entry.set_previous_session_for_test(o_recv, 1_000);
+
+        // Advance the pending sender so its frame carries a non-zero
+        // counter — the cascade will still try (and fail) `current` and
+        // `previous` first.
+        for _ in 0..4 {
+            let _ = seal(&mut p_send, b"warmup", !k_before);
+        }
+        let (ct, counter, hdr) = seal(&mut p_send, b"pending-hit", !k_before);
+        assert_eq!(counter, 4);
+
+        let (_pt, slot) = entry
+            .fsp_trial_decrypt(&ct, counter, &hdr, false, 2_000) // hint says "current first"
+            .expect("pending frame must decrypt");
+        assert_eq!(slot, EpochSlot::Pending);
+
+        // current and previous were tried and failed: their windows must
+        // be untouched (highest still 0).
+        assert_eq!(entry.current_highest_counter(), Some(0));
+        assert_eq!(entry.previous_highest_counter(), Some(0));
+        // Only the winning slot advanced.
+        assert_eq!(entry.pending_highest_counter(), Some(4));
+    }
+
+    // 7. The retained msg3 payload is cleared once a peer frame
+    //    authenticates against `pending`/new-`current`, not at the
+    //    initiator's own cutover.
+    #[test]
+    fn msg3_retransmit_stops_on_peer_new_epoch_confirmed() {
+        let (_cur_send, cur_recv) = xk_pair(1, 2);
+        let (mut p_send, p_recv) = xk_pair(3, 4);
+
+        let mut entry = entry_with_current(cur_recv);
+        entry.set_pending_session(p_recv);
+        entry.set_rekey_completed_ms(1_000);
+        entry.set_rekey_msg3_payload(vec![0xAB; 73], 1_500);
+
+        // The initiator's own liveness-timer cutover must NOT clear the
+        // retained msg3 payload — the responder may not have it yet.
+        assert!(entry.cutover_to_new_session(2_000));
+        assert!(
+            entry.rekey_msg3_payload().is_some(),
+            "cutover alone must not stop msg3 retransmission"
+        );
+        assert!(!entry.peer_new_epoch_confirmed());
+
+        // A peer frame authenticated against the post-cutover `current`
+        // (new epoch) confirms the responder and clears the payload.
+        let (ct, counter, hdr) = seal(&mut p_send, b"peer-on-new-epoch", entry.current_k_bit());
+        let k_now = entry.current_k_bit();
+        let (_pt, slot) = entry
+            .fsp_trial_decrypt(&ct, counter, &hdr, k_now, 2_500)
+            .unwrap();
+        assert_eq!(slot, EpochSlot::Current);
+        // Receive-path logic: current hit + retained payload + no pending
+        // => responder confirmed.
+        assert!(entry.rekey_msg3_payload().is_some() && entry.pending_new_session().is_none());
+        entry.confirm_peer_new_epoch();
+        assert!(entry.peer_new_epoch_confirmed());
+        assert!(entry.rekey_msg3_payload().is_none());
+    }
+
+    // 8. After exhausting the retransmission budget, abandon_rekey runs
+    //    and the entry is a clean Established with no pending.
+    #[test]
+    fn msg3_retransmit_budget_exhaustion_abandons_cleanly() {
+        let (_cur_send, cur_recv) = xk_pair(1, 2);
+        let (_p_send, p_recv) = xk_pair(3, 4);
+
+        let mut entry = entry_with_current(cur_recv);
+        entry.set_pending_session(p_recv);
+        entry.set_rekey_completed_ms(1_000);
+        entry.set_rekey_msg3_payload(vec![0xCD; 73], 1_500);
+
+        // Simulate the resend driver exhausting its budget.
+        let max_resends = 8;
+        for i in 0..max_resends {
+            entry.record_rekey_msg3_resend(2_000 + i as u64 * 100);
+        }
+        assert_eq!(entry.rekey_msg3_resend_count(), max_resends);
+
+        // Budget exhausted -> abandon.
+        entry.abandon_rekey();
+        assert!(entry.rekey_msg3_payload().is_none());
+        assert!(entry.pending_new_session().is_none());
+        assert!(!entry.has_rekey_in_progress());
+        assert!(entry.is_established());
+        assert!(!entry.peer_new_epoch_confirmed());
+    }
+
+    // 9. The initiator cuts over on its timer while the responder has
+    //    not; both directions still decrypt (overlapping epochs).
+    //
+    // Naming: in each `xk_pair` the `.0` session is node A's view and
+    // `.1` is node B's view; they seal frames the other one decrypts. A
+    // is the rekey initiator, B the responder.
+    #[test]
+    fn initiator_cutover_safe_before_responder() {
+        // Pre-rekey ("old") epoch and post-rekey ("new") epoch pairs.
+        let (old_a, old_b) = xk_pair(1, 2);
+        let (new_a, mut new_b) = xk_pair(3, 4);
+
+        // Node A (initiator): current = old_a, pending = new_a, retains a
+        // msg3 payload, then cuts over on its own liveness timer before
+        // the responder has received msg3.
+        let mut a = entry_with_current(old_a);
+        a.set_rekey_completed_ms(1_000);
+        a.set_rekey_msg3_payload(vec![0xEE; 73], 1_500);
+        a.set_pending_session(new_a);
+        assert!(a.cutover_to_new_session(2_000));
+        // A now: current = new_a, previous = old_a, K-bit flipped, msg3
+        // payload still retained (responder not yet confirmed).
+        assert!(a.rekey_msg3_payload().is_some());
+
+        // Node B (responder): still entirely on the old epoch — no
+        // pending, msg3 not yet received. B's `current` slot is `old_b`.
+        let mut b = entry_with_current(old_b);
+
+        // A -> B sealed in the NEW epoch (B has no pending slot yet).
+        // This is the one residual liveness-bounded drop, closed by msg3
+        // retransmission. Confirm it is a clean drop, not a panic.
+        let (ct_new, c_new, hdr_new) = seal(&mut new_b, b"new-from-a", true);
+        assert!(
+            b.fsp_trial_decrypt(&ct_new, c_new, &hdr_new, true, 2_100)
+                .is_none(),
+            "responder without msg3 drops the new-epoch frame cleanly"
+        );
+
+        // B -> A still sealed in the OLD epoch (B seals from its own
+        // `current` slot). A kept the old session as `previous`, so it
+        // decrypts fine despite the cutover skew.
+        let (ct_old, c_old, hdr_old) = {
+            let b_old = b.current_noise_session_mut().unwrap();
+            seal(b_old, b"old-from-b", false)
+        };
+        let (pt, slot) = a
+            .fsp_trial_decrypt(&ct_old, c_old, &hdr_old, false, 2_200)
+            .expect("initiator must still decrypt the responder's old-epoch frame");
+        assert_eq!(pt, b"old-from-b");
+        assert_eq!(slot, EpochSlot::Previous);
+
+        // Once B receives msg3 it derives the new session as pending; A's
+        // new-epoch frames then decrypt and drive B's promotion.
+        let (new_a2, mut new_b2) = xk_pair(3, 4);
+        b.set_pending_session(new_a2);
+        let (ct_new2, c_new2, hdr_new2) = seal(&mut new_b2, b"new-from-a-2", true);
+        let (pt, slot) = b
+            .fsp_trial_decrypt(&ct_new2, c_new2, &hdr_new2, true, 2_300)
+            .expect("responder must decrypt new-epoch frame once pending is installed");
+        assert_eq!(pt, b"new-from-a-2");
+        assert_eq!(slot, EpochSlot::Pending);
+    }
+
+    // 10. Drain-window expiry is peer-progress-aware (the rep-003
+    //     scenario). After the initiator cuts over, a responder that
+    //     never received msg3 keeps transmitting on the OLD epoch. The
+    //     `previous` slot must NOT be retired while those old-epoch
+    //     frames keep decrypting — even well past the 10 s fixed
+    //     window — and must be retired once the peer finally goes
+    //     silent on the old epoch.
+    #[test]
+    fn drain_expiry_is_peer_progress_aware() {
+        const DRAIN_MS: u64 = 10_000;
+        let cutover_ms = 1_000;
+
+        // Build the post-cutover state via the production cutover path:
+        // start with the old session as `current`, the new session as
+        // `pending`, then cut over so `current` = new epoch, `previous`
+        // = old epoch, and the drain clock is stamped at the cutover.
+        let (mut old_send, old_recv) = xk_pair(1, 2);
+        let (_new_send, new_recv) = xk_pair(3, 4);
+        let mut entry = entry_with_current(old_recv);
+        entry.set_pending_session(new_recv);
+        assert!(entry.cutover_to_new_session(cutover_ms));
+        assert!(entry.is_draining());
+
+        // The peer (responder) is still on the OLD epoch — it never
+        // received msg3. Each old-epoch frame must decrypt against
+        // `previous` and push the drain deadline out. Deliver frames at
+        // t = 5s, 15s, 25s — the latter two are well past the fixed
+        // 10 s window measured from the cutover at t=1s.
+        let k_old = !entry.current_k_bit();
+        for &t in &[5_000u64, 15_000, 25_000] {
+            let (ct, counter, hdr) = seal(&mut old_send, b"still-old-epoch", k_old);
+            let (_pt, slot) = entry
+                .fsp_trial_decrypt(&ct, counter, &hdr, k_old, t)
+                .expect("old-epoch frame must still decrypt while peer uses it");
+            assert_eq!(slot, EpochSlot::Previous);
+            // Even though `now - drain_started_ms` exceeds DRAIN_MS, the
+            // window is NOT expired: the peer just used `previous`.
+            assert!(
+                !entry.drain_expired(t, DRAIN_MS),
+                "previous slot must not be retired while peer keeps using it (t={t})"
+            );
+            assert!(
+                entry.previous_highest_counter().is_some(),
+                "previous slot must remain live (t={t})"
+            );
+        }
+
+        // The peer cuts over (or dies): no more old-epoch frames. The
+        // last `previous`-slot use was at t=25_000; the window now
+        // elapses DRAIN_MS after that, NOT DRAIN_MS after the cutover.
+        assert!(
+            !entry.drain_expired(34_999, DRAIN_MS),
+            "window must not expire before DRAIN_MS past the last previous use"
+        );
+        assert!(
+            entry.drain_expired(35_000, DRAIN_MS),
+            "window must expire DRAIN_MS after the last previous-slot decrypt"
+        );
+
+        // Once expired, complete_drain retires the previous slot.
+        entry.complete_drain();
+        assert!(entry.previous_highest_counter().is_none());
+        assert!(!entry.is_draining());
+    }
+
+    // 11. Absent any peer traffic on the old epoch, drain expiry still
+    //     fires on the plain wall-clock window measured from the
+    //     cutover — the peer-progress refinement must not delay
+    //     retirement when the peer cut over cleanly (no `previous`-slot
+    //     use at all).
+    #[test]
+    fn drain_expiry_unaffected_when_peer_off_old_epoch() {
+        const DRAIN_MS: u64 = 10_000;
+        let cutover_ms = 1_000;
+
+        let (_old_send, old_recv) = xk_pair(1, 2);
+        let (_new_send, new_recv) = xk_pair(3, 4);
+        let mut entry = entry_with_current(old_recv);
+        entry.set_pending_session(new_recv);
+        assert!(entry.cutover_to_new_session(cutover_ms));
+
+        // No old-epoch frames ever arrive: `previous_last_used_ms` stays
+        // 0, the deadline anchor is the cutover time.
+        assert!(
+            !entry.drain_expired(cutover_ms + DRAIN_MS - 1, DRAIN_MS),
+            "window must not expire early"
+        );
+        assert!(
+            entry.drain_expired(cutover_ms + DRAIN_MS, DRAIN_MS),
+            "window must expire on the plain wall-clock timer when peer is off the old epoch"
+        );
     }
 }