feat(wal): typed wal_append channel end-to-end + WS.CREATE created_at (Wave B stage 1 / kernel M1)

CHANGELOG.md

@@ -6,6 +6,83 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed — WAL v3 `wal_append` channel now preserves the caller's REAL record type end-to-end (K1a, storage-kernel M1 stage 1)
+
+`ShardDatabases::wal_append` / `try_wal_append_required` and
+`mq_exec::wal_append_on_slice` took a plain `Bytes` payload; the shard
+event-loop's 1ms-tick drain (`event_loop.rs`, two sites) unconditionally
+re-wrapped every message as an outer `WalRecordType::Command` record. Every
+non-Command producer (`XactCommit`, `WorkspaceCreate`/`WorkspaceDrop`,
+`MqCreate`/`MqAck`, `GraphTemporal`) worked around this by pre-framing its
+own record with `write_wal_v3_record` and sending the ALREADY-FRAMED bytes
+through — a second, nested WAL frame inside the outer `Command` frame,
+recovered on replay via a bespoke `read_wal_v3_record`-on-payload unwrap per
+record type. One nested-framing call (`handler_monoio/txn.rs` /
+`handler_sharded/txn.rs` XactCommit) additionally passed the transaction's
+`txn_id` into the inner frame's `lsn` field, mislabeling it.
+
+Fixed structurally: the channel item is now `(WalRecordType, Bytes)` — the
+producer's real type plus the UNFRAMED payload — threaded through
+`ShardSlice::wal_append_tx` / `ShardDatabases::wal_append_txs` end-to-end.
+The drain calls `wal.append(record_type, &payload)` directly, so the WAL
+writer assigns the real LSN and does the single framing. Every producer's
+pre-framing (`write_wal_v3_record` call) was deleted:
+`handler_monoio`/`handler_sharded` `write.rs` (WS.CREATE/DROP) and `txn.rs`
+(XactCommit — the `txn_id`-as-`lsn` mislabel is gone with it),
+`shard/uring_handler.rs`'s WS batch path, `shard/mq_exec.rs` (MqCreate/MqAck),
+and `command::temporal::apply_invalidate` (GraphTemporal — PR #286 had just
+added this record's pre-framing; it is deleted in favor of the typed
+channel, and the function now RETURNS the raw payload instead of pushing it
+into `GraphStore::wal_pending`, which stays exclusively `Command`-typed RESP
+bytes). The one caller that cannot change its `Frame`-only return signature
+without a ~90-call-site test ripple (`command::graph::dispatch_graph_command`,
+the cross-shard `GraphCommand` entry point) stashes the returned payload in
+a new `GraphStore::temporal_wal_pending` side-channel instead, which its two
+real callers (`shard/spsc_handler.rs`) `.take()` right after dispatch.
+
+Replay is **fully backward compatible, no format bump**: every existing
+direct-type match arm (`replay_workspace_wal`, `replay_mq_wal`,
+`replay_temporal_wal` in `shared_databases.rs`) already had a nested-Command
+unwrap fallback (added when the records were still nested) plus, for
+`GraphTemporal`, a legacy-raw fallback for even older un-nested records —
+both are UNTOUCHED, so segments written before this fix keep replaying
+exactly as before. New records simply hit the direct-type arm for the first
+time instead of falling through the unwrap.
+
+`persistence::recovery.rs` Phase 4's `on_command` closure had no
+`XactCommit` arm at all (silent `_ => {}`) — now reachable for the first
+time because the outer type used to always be `Command`. Decision: an
+EXPLICIT documented no-op (counts toward `commands_replayed`, never
+dispatches). The forward-image KV payload (`encode_xact_commit_payload`) is
+redundant in every reachable config — the transaction's individual SET/DEL
+ops already ride either this same Phase-4 WAL replay as ordinary `Command`
+records (`--wal-kv-log` on) or the AOF, the KV recovery authority in every
+config (Phase 4b falls back to it whenever `kv_commands_replayed == 0`).
+Decoding it would be a no-op at best and risks double-applying a
+non-idempotent op at worst — the same overlap risk the adjacent Phase 4b
+comment already flags. `wal_v3::replay::replay_wal_v3_dir_commands` (the
+legacy last-resort-fallback path, used only when no AOF exists) already had
+a correct `replay_xact_commit` call for the real outer type — untouched,
+and now reachable for the first time too since it stops receiving records
+nested inside `Command`.
+
+### Fixed — WS.CREATE's `created_at` was silently dropped by the WAL, restored as 0 after every restart (K1b, storage-kernel M1 stage 1)
+
+`encode_workspace_create`/`decode_workspace_create` (`src/workspace/wal.rs`)
+only ever serialized `[ws_id][name_len][name]` — the `created_at` computed
+at WS.CREATE time never reached the payload, so `replay_workspace_wal`
+(`shared_databases.rs`) had no choice but to hardcode `created_at: 0` on
+every restart, silently losing the real creation time. Fixed with a
+versioned, backward-compatible layout: new records append a trailing
+`created_at_ms: i64 LE`; the decoder accepts BOTH the old
+(`20 + name_len`-byte) and new (`20 + name_len + 8`-byte) lengths, returning
+`created_at = 0` for old records (matching prior restart behavior exactly —
+no format bump, mixed-segment compatible). All three producers
+(`handler_monoio`/`handler_sharded` `write.rs`, `shard/uring_handler.rs`'s
+batch path) now pass the already-computed `created_at` through; the replay
+decoder threads the real value into `WorkspaceMetadata` instead of the
+hardcoded `0`.
+
 ### Fixed — autovacuum Pass C recycled the sole durable copy of graph/WS/MQ/temporal WAL history in legacy mode (task #43, P1)
 
 `AutovacuumDaemon::run_tick`'s Pass C (`src/shard/autovacuum.rs`) called

src/command/graph/mod.rs

@@ -153,7 +153,18 @@ pub fn dispatch_graph_command(store: &mut GraphStore, command: &Frame) -> Frame
                     &graph_name,
                     wall_ms,
                 ) {
-                    Ok(()) => Frame::SimpleString(Bytes::from_static(b"OK")),
+                    // `dispatch_graph_command` only returns a `Frame` (its ~90
+                    // test call sites and the two real cross-shard callers in
+                    // `spsc_handler.rs` all assume that), so the GraphTemporal
+                    // payload can't ride the return value. Stash it in the
+                    // dedicated `temporal_wal_pending` side-channel — distinct
+                    // from `wal_pending` (always `Command`-typed RESP bytes) —
+                    // for the caller to `.take()` and send through the typed
+                    // `wal_append` channel right after this call returns.
+                    Ok(payload) => {
+                        store.temporal_wal_pending = Some(payload);
+                        Frame::SimpleString(Bytes::from_static(b"OK"))
+                    }
                     Err(e) => Frame::Error(Bytes::from_static(e)),
                 }
             }

src/command/temporal.rs

@@ -149,19 +149,35 @@ pub fn parse_invalidate_at(args: &[Frame]) -> Option<(Bytes, bool, u64, i64)> {
 
 /// Apply a TEMPORAL.INVALIDATE mutation to a graph store.
 ///
-/// Sets `valid_to = wall_ms` on the entity and pushes the WAL payload into
-/// `gs.wal_pending`. The CALLER drains the WAL and appends on its own shard —
-/// this keeps the function usable from both the connection-local path and the
+/// Sets `valid_to = wall_ms` on the entity and returns the UNFRAMED
+/// `GraphTemporal` WAL payload (`encode_graph_temporal` bytes, no
+/// `write_wal_v3_record` wrapping). The CALLER sends it through the typed
+/// `wal_append` channel tagged `WalRecordType::GraphTemporal` — this keeps
+/// the function usable from both the connection-local path and the
 /// shard-side `ShardMessage::GraphCommand` handler (multi-shard routing sends
-/// the command to the shard that owns the graph name).
+/// the command to the shard that owns the graph name; that caller instead
+/// stashes the payload in `gs.temporal_wal_pending` since it can only return
+/// a `Frame`, see `command::graph::dispatch_graph_command`).
+///
+/// K1a: earlier versions of this function pre-framed the payload with
+/// `write_wal_v3_record(WalRecordType::GraphTemporal)` and pushed it into
+/// `gs.wal_pending` (which is otherwise always `Command`-typed RESP bytes)
+/// because the cross-thread `wal_append` channel used to force EVERY record
+/// to an outer `WalRecordType::Command`, and nesting was the only way replay
+/// could recover the real type. Now that the channel carries the real type
+/// end-to-end, returning the raw payload directly is both simpler and
+/// removes the double-framing overhead. Replay still tolerates the OLD
+/// nested-Command form for segments written before this fix — see
+/// `ShardDatabases::replay_temporal_wal`'s nested-Command-unwrap arm and its
+/// legacy-raw-fallback for segments from even before THAT fix.
 #[cfg(feature = "graph")]
 pub fn apply_invalidate(
     gs: &mut crate::graph::store::GraphStore,
     entity_id: u64,
     is_node: bool,
     graph_name: &Bytes,
     wall_ms: i64,
-) -> Result<(), &'static [u8]> {
+) -> Result<Vec<u8>, &'static [u8]> {
     let Some(named_graph) = gs.get_graph_mut(graph_name) else {
         return Err(ERR_GRAPH_NOT_FOUND);
     };
@@ -185,39 +201,21 @@ pub fn apply_invalidate(
     if !mutated {
         return Err(ERR_ENTITY_NOT_FOUND);
     }
-    // Pre-frame with `write_wal_v3_record` (WalRecordType::GraphTemporal)
-    // exactly like the MQ producers (`mq_exec.rs::handle_create`/`handle_ack`)
-    // do for MqCreate/MqAck -- the shard event-loop drain re-wraps whatever
-    // arrives on `wal_append_tx` as an OUTER `WalRecordType::Command` record,
-    // so pre-framing here is what lets `replay_temporal_wal`'s nested-Command
-    // unwrap (`read_wal_v3_record`) recover this record UNAMBIGUOUSLY by
-    // type tag + CRC, instead of guessing from the raw byte layout. Before
-    // this fix the raw (un-framed) 25-byte payload was pushed directly,
-    // which forced replay to discriminate it from a RESP command payload by
-    // a `payload[0] != b'*'` heuristic on the entity_id's low byte -- WRONG
-    // for any entity_id whose low byte happened to be 0x2A, silently
-    // dropping the invalidation on replay (see `replay_temporal_wal`'s
-    // legacy-raw-fallback comment for the mixed-segment compatibility path
-    // this leaves in place for WAL segments written before this fix).
-    let inner_payload = crate::persistence::wal_v3::record::encode_graph_temporal(
+    // K1a: return the UNFRAMED payload — no `write_wal_v3_record` pre-framing
+    // and no `gs.wal_pending` push. The typed `wal_append` channel carries
+    // `WalRecordType::GraphTemporal` end-to-end now, so the caller can send
+    // this straight through without a second nested WAL frame.
+    let payload = crate::persistence::wal_v3::record::encode_graph_temporal(
         entity_id, is_node, wall_ms, wall_ms,
     );
-    let mut framed = Vec::with_capacity(20 + inner_payload.len());
-    crate::persistence::wal_v3::record::write_wal_v3_record(
-        &mut framed,
-        0,
-        crate::persistence::wal_v3::record::WalRecordType::GraphTemporal,
-        &inner_payload,
-    );
-    gs.wal_pending.push(framed);
     // Task #32: TEMPORAL.INVALIDATE mutates valid_to on write_buf directly,
     // changing what a subsequent read sees -- invalidate the graph's cached
     // query results. Re-fetch rather than reuse `named_graph` above: the
     // borrow was released at the `let Some(named_graph) = ...` match end.
     if let Some(named_graph) = gs.get_graph_mut(graph_name) {
         named_graph.touch();
     }
-    Ok(())
+    Ok(payload)
 }
 
 /// Capture the current wall-clock time as i64 Unix milliseconds.

src/graph/store.rs

@@ -252,7 +252,18 @@ pub struct GraphStore {
     next_lsn: u64,
     /// Pending WAL records produced by write handlers. Connection handlers
     /// drain this after dispatch and send bytes via `shard_databases.wal_append()`.
+    /// Always `WalRecordType::Command` (RESP-encoded GRAPH.* commands) — see
+    /// `temporal_wal_pending` for the one non-Command graph WAL record type.
     pub(crate) wal_pending: Vec<Vec<u8>>,
+    /// One pending `GraphTemporal` WAL payload (unframed, raw
+    /// `encode_graph_temporal` bytes), set by `TEMPORAL.INVALIDATE`'s
+    /// `command::temporal::apply_invalidate`. Kept separate from
+    /// `wal_pending` (which is always `Command`-typed RESP bytes) so the
+    /// cross-thread `wal_append` channel can carry each record's REAL
+    /// `WalRecordType` end-to-end instead of nesting a second WAL frame
+    /// inside a `Command` payload. Callers `.take()` this immediately after
+    /// dispatch, same drain discipline as `drain_wal()`.
+    pub(crate) temporal_wal_pending: Option<Vec<u8>>,
     /// Monotonic freshness counter for the GRAPH engine on this shard.
     ///
     /// Bumped (Release) after every successful mutating operation: `create_graph`,
@@ -286,6 +297,7 @@ impl GraphStore {
             graphs: None,
             next_lsn: 0,
             wal_pending: Vec::new(),
+            temporal_wal_pending: None,
             version_token: AtomicU64::new(0),
             dirty: false,
             snapshot_lsn: 0,

src/persistence/recovery.rs

@@ -408,6 +408,46 @@ pub fn recover_shard_v3_pitr(
                     // File lifecycle events -- verify against manifest (future)
                     result.commands_replayed += 1;
                 }
+                WalRecordType::XactCommit => {
+                    // K1a decision (storage-audit-2026-07-12-wal.md §5.1):
+                    // intentional documented no-op, NOT a gap. Before K1a this
+                    // record arrived here nested inside an outer `Command`
+                    // frame, so `read_wal_v3_record`-parsing the payload as
+                    // RESP silently produced nothing (the audit's "XactCommit
+                    // replay is functionally dead"); now that the typed
+                    // channel preserves the real outer type, this arm is
+                    // reachable for the first time and needs an explicit
+                    // decision rather than silently falling into `_ => {}`.
+                    //
+                    // The forward-image KV payload
+                    // (`encode_xact_commit_payload`) is REDUNDANT here in
+                    // every reachable config: the transaction's individual
+                    // SET/DEL ops already ride EITHER this same Phase-4 WAL
+                    // replay as ordinary `Command` records (when
+                    // `--wal-kv-log` is on — ordinary KV write dispatch does
+                    // not distinguish "inside a cross-store txn" from any
+                    // other write) OR the AOF, which is the KV recovery
+                    // authority in every reachable config (Phase 4b below
+                    // falls back to it whenever `kv_commands_replayed == 0`).
+                    // Decoding + re-applying the forward image would at best
+                    // be a no-op (same final value) and at worst double-apply
+                    // a non-idempotent op if that invariant ever drifts — the
+                    // exact risk the Phase 4b comment above already flags for
+                    // WAL+AOF overlap. Counted (like Vector*/File* above) so
+                    // `commands_replayed` reflects what was actually on disk;
+                    // never dispatched.
+                    //
+                    // ASYMMETRY (intentional): the last-resort legacy fallback
+                    // `wal_v3::replay::replay_wal_v3_dir_commands` DOES apply
+                    // XactCommit via `replay_xact_commit`. That path runs only
+                    // when this Phase 4 replayed ZERO KV commands AND no
+                    // `appendonly.aof` exists — i.e. exactly when neither of
+                    // the redundant coverage sources argued above is present,
+                    // so applying the forward image there is the only way the
+                    // txn's writes survive at all. Same reasoning, opposite
+                    // conclusion, because the preconditions are complementary.
+                    result.commands_replayed += 1;
+                }
                 _ => {}
             }
         };

src/persistence/wal_v3/segment.rs

@@ -876,6 +876,56 @@ mod tests {
         assert_eq!(count, 10);
     }
 
+    /// K1a RED/GREEN: a WS.CREATE record sent through the SAME
+    /// `(WalRecordType, Bytes)` typed channel + drain pattern the shard event
+    /// loop uses (`event_loop.rs`'s `wal_append_rx.try_recv()` loop) must land
+    /// on disk with its REAL outer type (`WorkspaceCreate`) — not re-wrapped
+    /// as `WalRecordType::Command`, the nested-Command workaround this task
+    /// retires. Before K1a the channel only carried `Bytes` and the drain
+    /// forced `WalRecordType::Command` unconditionally
+    /// (`wal.append(WalRecordType::Command, &data)`), so this assertion would
+    /// have failed (`record.record_type == Command`, not `WorkspaceCreate`)
+    /// and the payload the RESP-parser tried to read would have been the
+    /// WorkspaceCreate bytes themselves — not even valid RESP.
+    #[test]
+    fn test_typed_channel_preserves_workspace_create_outer_type() {
+        let tmp = tempfile::tempdir().unwrap();
+        let wal_dir = tmp.path().join("wal");
+        let mut writer = WalWriterV3::new(0, &wal_dir, DEFAULT_SEGMENT_SIZE).unwrap();
+
+        // Producer side: exactly what handler_monoio/write.rs's WS.CREATE arm
+        // does post-K1a — build the UNFRAMED payload and send it with its
+        // real type over the typed channel. No `write_wal_v3_record`
+        // pre-framing.
+        let (tx, rx) = crate::runtime::channel::mpsc_bounded::<(WalRecordType, bytes::Bytes)>(4096);
+        let ws_id = [7u8; 16];
+        let payload =
+            crate::workspace::wal::encode_workspace_create(&ws_id, b"acme", 1_752_000_000_000);
+        tx.try_send((WalRecordType::WorkspaceCreate, bytes::Bytes::from(payload)))
+            .unwrap();
+
+        // Drain side: the exact loop body from `event_loop.rs`'s 1ms tick.
+        while let Ok((record_type, data)) = rx.try_recv() {
+            writer.append(record_type, &data);
+        }
+        writer.flush_sync().unwrap();
+
+        // Read back the raw segment bytes and parse the FIRST (only) record.
+        let data = fs::read(WalSegment::segment_path(&wal_dir, 1)).unwrap();
+        let record = read_wal_v3_record(&data[WAL_V3_HEADER_SIZE..]).expect("record parses");
+        assert_eq!(
+            record.record_type,
+            WalRecordType::WorkspaceCreate,
+            "K1a: outer type must be the producer's REAL type, not re-wrapped Command"
+        );
+        // And the payload decodes directly — no nested-Command unwrap needed.
+        let (decoded_id, decoded_name, created_at) =
+            crate::workspace::wal::decode_workspace_create(&record.payload).unwrap();
+        assert_eq!(decoded_id, ws_id);
+        assert_eq!(decoded_name, b"acme");
+        assert_eq!(created_at, 1_752_000_000_000);
+    }
+
     #[test]
     fn test_wait_durable_zero_and_already_durable_are_noops() {
         let tmp = tempfile::tempdir().unwrap();

src/replication/apply.rs

@@ -325,9 +325,12 @@ fn apply_graph(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Fram
 
 /// Apply a replicated `TEMPORAL.INVALIDATE-AT` record: same mutation the
 /// master ran (`apply_invalidate`) with the master's pinned `wall_ms`. The
-/// drained `GraphTemporal` WAL payload is dropped, matching `apply_graph`'s
-/// no-local-persistence model (a restarted replica resyncs from the master;
-/// leaving it in `wal_pending` would leak into an unrelated later drain).
+/// returned `GraphTemporal` WAL payload is dropped (`Ok(_)` is ignored),
+/// matching `apply_graph`'s no-local-persistence model (a restarted replica
+/// resyncs from the master).
+///
+/// K1a: `apply_invalidate` now returns the payload directly instead of
+/// pushing it into `gs.wal_pending` — there is nothing left to drain here.
 #[cfg(feature = "graph")]
 fn apply_temporal_invalidate(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame]) {
     let Some((graph_name, is_node, entity_id, wall_ms)) =
@@ -354,7 +357,6 @@ fn apply_temporal_invalidate(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8]
             String::from_utf8_lossy(e)
         );
     }
-    let _ = s.graph_store.drain_wal();
 }
 
 /// Mirror of the master's connection-layer index-parity block