feat(persistence): MqPop v2 payload — PEL delivery/seen timestamps survive kill-9 and promotion (task #47)

CHANGELOG.md

@@ -6,6 +6,42 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed — MqPop WAL/replication payload v2 carries real PEL idle metadata (task #47)
+
+`apply_mq_pop` (boot-time WAL replay AND live replica apply — same function,
+`src/shard/shared_databases.rs`) hardcoded a claimed PEL entry's
+`delivery_time`/`seen_time` to `0` regardless of when the original `MQ.POP`
+actually claimed it. Since `XPENDING`'s idle time is `now - delivery_time`,
+a kill-9 restart or a promoted replica reported "idle since the Unix epoch"
+(an astronomically large idle time) instead of the real elapsed duration —
+`XPENDING`/`XPENDING ... IDLE` consumers on a recovered node saw wrong data
+for every pre-crash claim.
+
+Fixed by bumping the `MqPop` WAL record to its own v2 payload
+(`MQ_POP_WAL_V1`/`MQ_POP_WAL_V2` in `src/mq/wal.rs` — `MqPop` now versions
+independently of the shared `MQ_WAL_VERSION` used by every other MQ record
+kind) that appends `[delivery_time_ms:u64][seen_time_ms:u64]` captured once
+per POP batch (every entry one `MQ.POP` claims shares a single `now`, per
+`Stream::read_group_new`). `src/shard/mq_exec.rs`'s POP handler now reads
+these back out of the PEL/consumer state it just wrote instead of
+discarding them. Decode is version-led and fail-closed: v1 payloads (no
+trailing timing fields) still decode, filling an explicit `0` sentinel that
+`apply_mq_pop` resolves to `current_time_ms()` at apply time — a strict
+improvement over the old hardcoded-`0` behavior, not a behavior match to it.
+Any version byte other than 1 or 2 is rejected (`None`), same fail-closed
+posture as every other MQ WAL decoder.
+
+Producer (`src/shard/mq_exec.rs`), boot-time replay
+(`src/shard/shared_databases.rs::apply_mq_pop`), and live replica apply
+(`src/replication/apply.rs::apply_mq`) all updated; the MQ WAL fuzz target
+(`fuzz/fuzz_targets/mq_wal_record.rs`) needed no changes since it fuzzes
+`decode_mq_pop` generically. New coverage: codec round-trip tests for v1
+(sentinel-filled) and v2 (full round-trip) in `src/mq/wal.rs`, a kill-9
+integration test asserting `XPENDING` idle survives restart
+(`tests/crash_recovery_mq_effects.rs::xpending_idle_metadata_survives_kill9`),
+and a replica-promotion test asserting the promoted node's `XPENDING` idle
+reflects the master's original claim time, not the replica's own apply time
+(`tests/replication_mq.rs::promoted_replica_reports_real_pending_idle_time`).
 ### Fixed — unified replica poison-record policy across graph/MQ/WS/temporal planes (task #48)
 
 Each replica-apply plane (graph WAL replay, MQ effect records, workspace

src/mq/wal.rs

@@ -223,30 +223,50 @@ pub type ClaimedEntry = (u64, u64, u64);
 /// One DLQ routing decision: (source_ms, source_seq, dlq_ms, dlq_seq).
 pub type DlqRouting = (u64, u64, u64, u64);
 
+/// `MqPop` diverges from the shared `MQ_WAL_VERSION` scheme: it carries its
+/// OWN version byte because task #47 needed to add fields to this one
+/// record kind without bumping every other MQ record's format in lockstep.
+/// v1 is the pre-task-#47 layout (no PEL timing metadata); v2 (current)
+/// adds `delivery_time_ms`/`seen_time_ms` so a promoted replica or a
+/// kill-9 restart can reconstruct real `XPENDING` idle times instead of
+/// resetting the PEL entry's clock to 0 (which reported "maximally idle"
+/// until the entry was next touched).
+pub const MQ_POP_WAL_V1: u8 = 1;
+/// Current `MqPop` payload version. See [`MQ_POP_WAL_V1`].
+pub const MQ_POP_WAL_V2: u8 = 2;
+
 /// Encode an MqPop WAL payload. Captures the full outcome of a POP: which
 /// ids were claimed (and at what delivery_count), the consumer group's
-/// resulting `last_delivered_id`, and which claimed ids were immediately
-/// routed to the DLQ (source id -> assigned DLQ-stream id). The fixed
-/// group/consumer names (`__mq_consumers` / `__mq_default`) used by every
-/// MQ.POP are NOT carried -- baked in at replay, matching the single-group-
-/// per-queue design `src/shard/mq_exec.rs` implements today.
+/// resulting `last_delivered_id`, which claimed ids were immediately
+/// routed to the DLQ (source id -> assigned DLQ-stream id), and (v2+) the
+/// PEL timing metadata (`delivery_time_ms`, `seen_time_ms`) captured at
+/// claim time. The fixed group/consumer names (`__mq_consumers` /
+/// `__mq_default`) used by every MQ.POP are NOT carried -- baked in at
+/// replay, matching the single-group-per-queue design
+/// `src/shard/mq_exec.rs` implements today. Every entry claimed by one
+/// MQ.POP shares a single `delivery_time_ms`/`seen_time_ms` (the call's
+/// `now`, per `Stream::read_group_new`), so one pair of `u64`s covers the
+/// whole batch losslessly -- no need for a per-entry timestamp.
 ///
-/// Layout:
-/// `[version:u8=1][db_index:u32][key_len:u32][key:N]`
+/// Layout (v2):
+/// `[version:u8=2][db_index:u32][key_len:u32][key:N]`
 /// `[last_delivered_ms:u64][last_delivered_seq:u64]`
 /// `[claimed_count:u32]` then per entry `[ms:u64][seq:u64][delivery_count:u64]`
 /// `[dlq_count:u32]` then per entry `[src_ms:u64][src_seq:u64][dlq_ms:u64][dlq_seq:u64]`
+/// `[delivery_time_ms:u64][seen_time_ms:u64]`
 pub fn encode_mq_pop(
     db_index: u32,
     queue_key: &[u8],
     last_delivered: (u64, u64),
     claimed: &[ClaimedEntry],
     dlq: &[DlqRouting],
+    delivery_time_ms: u64,
+    seen_time_ms: u64,
 ) -> Vec<u8> {
     let mut payload = Vec::with_capacity(
-        1 + 4 + 4 + queue_key.len() + 16 + 4 + claimed.len() * 24 + 4 + dlq.len() * 32,
+        1 + 4 + 4 + queue_key.len() + 16 + 4 + claimed.len() * 24 + 4 + dlq.len() * 32 + 16,
     );
-    payload.push(MQ_WAL_VERSION);
+    payload.push(MQ_POP_WAL_V2);
     payload.extend_from_slice(&db_index.to_le_bytes());
     payload.extend_from_slice(&(queue_key.len() as u32).to_le_bytes());
     payload.extend_from_slice(queue_key);
@@ -265,18 +285,42 @@ pub fn encode_mq_pop(
         payload.extend_from_slice(&dlq_ms.to_le_bytes());
         payload.extend_from_slice(&dlq_seq.to_le_bytes());
     }
+    payload.extend_from_slice(&delivery_time_ms.to_le_bytes());
+    payload.extend_from_slice(&seen_time_ms.to_le_bytes());
     payload
 }
 
 /// Decode an MqPop WAL payload.
 ///
-/// Returns `(db_index, queue_key, last_delivered, claimed, dlq)` or `None`
-/// if malformed or an unsupported version.
+/// Returns `(db_index, queue_key, last_delivered, claimed, dlq,
+/// delivery_time_ms, seen_time_ms)` or `None` if malformed or an
+/// unsupported version.
+///
+/// Version handling (own scheme, see [`MQ_POP_WAL_V2`]):
+/// - v1: legacy layout with no trailing timing fields. Decodes fine;
+///   `delivery_time_ms`/`seen_time_ms` are returned as sentinel `0`,
+///   EXPLICITLY meaning "not recorded -- the apply-time caller must
+///   substitute its own current-time reading" (this is a STRICT
+///   IMPROVEMENT over the pre-task-#47 apply behavior, which hardcoded the
+///   PEL entry's clock to 0 forever; callers now resolve the sentinel to
+///   "now" instead).
+/// - v2: full layout: real captured timestamps round-trip exactly.
+/// - anything else: rejected (`None`), same fail-closed posture as every
+///   other MQ WAL decoder in this module.
 #[allow(clippy::type_complexity)]
 pub fn decode_mq_pop(
     payload: &[u8],
-) -> Option<(u32, Vec<u8>, (u64, u64), Vec<ClaimedEntry>, Vec<DlqRouting>)> {
-    if payload.is_empty() || payload[0] != MQ_WAL_VERSION {
+) -> Option<(
+    u32,
+    Vec<u8>,
+    (u64, u64),
+    Vec<ClaimedEntry>,
+    Vec<DlqRouting>,
+    u64,
+    u64,
+)> {
+    let version = *payload.first()?;
+    if version != MQ_POP_WAL_V1 && version != MQ_POP_WAL_V2 {
         return None;
     }
     let p = &payload[1..];
@@ -331,7 +375,27 @@ pub fn decode_mq_pop(
         dlq.push((src_ms, src_seq, dlq_ms, dlq_seq));
     }
 
-    Some((db_index, key, (last_ms, last_seq), claimed, dlq))
+    let (delivery_time_ms, seen_time_ms) = if version >= MQ_POP_WAL_V2 {
+        if p.len() < off + 16 {
+            return None;
+        }
+        let delivery_time_ms = u64::from_le_bytes(p[off..off + 8].try_into().ok()?);
+        off += 8;
+        let seen_time_ms = u64::from_le_bytes(p[off..off + 8].try_into().ok()?);
+        (delivery_time_ms, seen_time_ms)
+    } else {
+        (0u64, 0u64)
+    };
+
+    Some((
+        db_index,
+        key,
+        (last_ms, last_seq),
+        claimed,
+        dlq,
+        delivery_time_ms,
+        seen_time_ms,
+    ))
 }
 
 // ── MqTrigger (0x74) ──────────────────────────────────────────────────────
@@ -703,27 +767,33 @@ mod tests {
     fn test_mq_pop_roundtrip() {
         let claimed: Vec<ClaimedEntry> = vec![(1, 0, 1), (1, 1, 1), (1, 2, 2)];
         let dlq: Vec<DlqRouting> = vec![(1, 2, 2, 0)];
-        let payload = encode_mq_pop(1, b"orders", (1, 2), &claimed, &dlq);
-        let (db, key, last, decoded_claimed, decoded_dlq) = decode_mq_pop(&payload).unwrap();
+        let payload = encode_mq_pop(1, b"orders", (1, 2), &claimed, &dlq, 12345, 12300);
+        let (db, key, last, decoded_claimed, decoded_dlq, delivery_time_ms, seen_time_ms) =
+            decode_mq_pop(&payload).unwrap();
         assert_eq!(db, 1);
         assert_eq!(key, b"orders");
         assert_eq!(last, (1, 2));
         assert_eq!(decoded_claimed, claimed);
         assert_eq!(decoded_dlq, dlq);
+        assert_eq!(delivery_time_ms, 12345);
+        assert_eq!(seen_time_ms, 12300);
     }
 
     #[test]
     fn test_mq_pop_roundtrip_empty() {
-        let payload = encode_mq_pop(0, b"q", (0, 0), &[], &[]);
-        let (_, _, last, claimed, dlq) = decode_mq_pop(&payload).unwrap();
+        let payload = encode_mq_pop(0, b"q", (0, 0), &[], &[], 0, 0);
+        let (_, _, last, claimed, dlq, delivery_time_ms, seen_time_ms) =
+            decode_mq_pop(&payload).unwrap();
         assert_eq!(last, (0, 0));
         assert!(claimed.is_empty());
         assert!(dlq.is_empty());
+        assert_eq!(delivery_time_ms, 0);
+        assert_eq!(seen_time_ms, 0);
     }
 
     #[test]
     fn test_mq_pop_malformed_truncated_claimed() {
-        let mut bad = vec![MQ_WAL_VERSION];
+        let mut bad = vec![MQ_POP_WAL_V2];
         bad.extend_from_slice(&0u32.to_le_bytes());
         bad.extend_from_slice(&1u32.to_le_bytes());
         bad.push(b'q');
@@ -735,11 +805,56 @@ mod tests {
 
     #[test]
     fn test_mq_pop_unknown_version_rejected() {
-        let mut payload = encode_mq_pop(0, b"q", (0, 0), &[], &[]);
+        let mut payload = encode_mq_pop(0, b"q", (0, 0), &[], &[], 0, 0);
         payload[0] = 42;
         assert!(decode_mq_pop(&payload).is_none());
     }
 
+    /// v1 payloads (pre-task-#47, no trailing timing fields) must still
+    /// decode -- and must fill the sentinel `0` for both new fields,
+    /// documented in `decode_mq_pop` as "not recorded; apply-time caller
+    /// substitutes current time" rather than silently misreporting a real
+    /// timestamp.
+    #[test]
+    fn test_mq_pop_v1_decodes_with_sentinel_timing() {
+        let claimed: Vec<ClaimedEntry> = vec![(1, 0, 1)];
+        let dlq: Vec<DlqRouting> = vec![];
+        let mut v1 = vec![MQ_POP_WAL_V1];
+        v1.extend_from_slice(&0u32.to_le_bytes()); // db_index
+        v1.extend_from_slice(&(b"q".len() as u32).to_le_bytes());
+        v1.extend_from_slice(b"q");
+        v1.extend_from_slice(&1u64.to_le_bytes()); // last_delivered ms
+        v1.extend_from_slice(&0u64.to_le_bytes()); // last_delivered seq
+        v1.extend_from_slice(&(claimed.len() as u32).to_le_bytes());
+        for (ms, seq, dc) in &claimed {
+            v1.extend_from_slice(&ms.to_le_bytes());
+            v1.extend_from_slice(&seq.to_le_bytes());
+            v1.extend_from_slice(&dc.to_le_bytes());
+        }
+        v1.extend_from_slice(&(dlq.len() as u32).to_le_bytes());
+        // NOTE: no trailing delivery_time_ms/seen_time_ms -- that's the v1/v2 delta.
+
+        let (db, key, last, decoded_claimed, decoded_dlq, delivery_time_ms, seen_time_ms) =
+            decode_mq_pop(&v1).unwrap();
+        assert_eq!(db, 0);
+        assert_eq!(key, b"q");
+        assert_eq!(last, (1, 0));
+        assert_eq!(decoded_claimed, claimed);
+        assert!(decoded_dlq.is_empty());
+        assert_eq!(delivery_time_ms, 0);
+        assert_eq!(seen_time_ms, 0);
+    }
+
+    #[test]
+    fn test_mq_pop_v2_truncated_timing_rejected() {
+        // A v2-tagged payload that's missing the trailing 16 timing bytes
+        // must fail closed, not silently degrade to sentinel-0 (that would
+        // hide real truncation/corruption behind the v1 fallback path).
+        let payload = encode_mq_pop(0, b"q", (0, 0), &[], &[], 999, 888);
+        let truncated = &payload[..payload.len() - 16];
+        assert!(decode_mq_pop(truncated).is_none());
+    }
+
     // --- MqDrop roundtrip ---
 
     #[test]

src/replication/apply.rs

@@ -612,38 +612,42 @@ fn apply_mq(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame])
             );
         })
     } else if cmd.eq_ignore_ascii_case(MQ_REPL_POP) {
-        decode_mq_pop(payload).map(|(db_index, key, last_delivered, claimed, dlq)| {
-            let claimed: Vec<(StreamId, u64)> = claimed
-                .into_iter()
-                .map(|(ms, seq, dc)| (StreamId { ms, seq }, dc))
-                .collect();
-            let dlq: Vec<(StreamId, StreamId)> = dlq
-                .into_iter()
-                .map(|(src_ms, src_seq, dlq_ms, dlq_seq)| {
-                    (
-                        StreamId {
-                            ms: src_ms,
-                            seq: src_seq,
-                        },
-                        StreamId {
-                            ms: dlq_ms,
-                            seq: dlq_seq,
-                        },
-                    )
-                })
-                .collect();
-            apply_mq_pop(
-                s,
-                clamp_mq_db(db_count, db_index),
-                &key,
-                StreamId {
-                    ms: last_delivered.0,
-                    seq: last_delivered.1,
-                },
-                claimed,
-                dlq,
-            );
-        })
+        decode_mq_pop(payload).map(
+            |(db_index, key, last_delivered, claimed, dlq, delivery_time_ms, seen_time_ms)| {
+                let claimed: Vec<(StreamId, u64)> = claimed
+                    .into_iter()
+                    .map(|(ms, seq, dc)| (StreamId { ms, seq }, dc))
+                    .collect();
+                let dlq: Vec<(StreamId, StreamId)> = dlq
+                    .into_iter()
+                    .map(|(src_ms, src_seq, dlq_ms, dlq_seq)| {
+                        (
+                            StreamId {
+                                ms: src_ms,
+                                seq: src_seq,
+                            },
+                            StreamId {
+                                ms: dlq_ms,
+                                seq: dlq_seq,
+                            },
+                        )
+                    })
+                    .collect();
+                apply_mq_pop(
+                    s,
+                    clamp_mq_db(db_count, db_index),
+                    &key,
+                    StreamId {
+                        ms: last_delivered.0,
+                        seq: last_delivered.1,
+                    },
+                    claimed,
+                    dlq,
+                    delivery_time_ms,
+                    seen_time_ms,
+                );
+            },
+        )
     } else if cmd.eq_ignore_ascii_case(MQ_REPL_ACK) {
         decode_mq_ack(payload).map(|(db_index, key, ms, seq)| {
             apply_mq_ack(