From 308e20852c7fa18a225417417ff0de6670562ed5 Mon Sep 17 00:00:00 2001 From: CUBELinux-2 Date: Tue, 11 Aug 2026 12:50:54 -0400 Subject: [PATCH] feat(cubesys): Task 5 transactions (BEGIN/COMMIT/ROLLBACK) + HELLO identity wiring - Session gains txn: Option (BEGIN snapshot + buffered ops) and identity: Option (owner enforcement hook for Tasks 6+). - prog/write/del buffer into the open txn; commit_txn applies the whole batch under one write lock + a SINGLE WalOp::Txn entry (atomic + durable replay). ROLLBACK discards. Reads consult the BEGIN snapshot (isolation). - cube-server: HELLO-resolved tenant yields Some(ts); default-tenant path consistent; Session built once per connection so txns span frames. - WalEntry gains batch field; decode_wal + encode_wal handle Txn replay. - 4 new T5 tests: buffer/commit, rollback, snapshot isolation, durable reopen via WAL replay. ./check quick green. --- cubesys/src/bin/cube-server.rs | 36 +++-- cubesys/src/commands.rs | 288 ++++++++++++++++++++++++++++++++- cubesys/src/store.rs | 146 ++++++++++++++++- 3 files changed, 450 insertions(+), 20 deletions(-) diff --git a/cubesys/src/bin/cube-server.rs b/cubesys/src/bin/cube-server.rs index 7c88ac7..1c8cb30 100644 --- a/cubesys/src/bin/cube-server.rs +++ b/cubesys/src/bin/cube-server.rs @@ -36,6 +36,7 @@ use std::str::FromStr; use std::sync::Arc; use std::thread; +use cubesys::commands::Session; use cubesys::net::{read_stream_frame, write_frame}; use cubesys::store::DurabilityConfig; use cubesys::tenant::{TenantConfig, TenantId, TenantRegistry, TenantSession}; @@ -72,7 +73,7 @@ impl Server { // that sends HELLO first is routed to (and stamps) its // declared tenant; one that does not falls back to the // shared "default" tenant. - let session = match read_stream_frame(&mut stream) { + let ts = match read_stream_frame(&mut stream) { Ok(req) => { let line = req.trim(); if line.is_empty() { @@ -113,7 +114,7 @@ impl Server { { return; } - ts + Some(ts) } Err(e) => { let _ = @@ -126,8 +127,11 @@ impl Server { // run the line as its first command. match resolve_default(®istry, per_tenant) { Ok(ts) => { - handle_command(&ts, &mut stream, line); - return; + let mut sess = Session::for_tenant(&ts); + handle_command(&mut sess, &mut stream, line); + // Hold the session for the rest of + // the connection (so txns span lines). + Some(ts) } Err(e) => { let _ = @@ -143,6 +147,13 @@ impl Server { } }; + let ts = ts.expect("tenant session resolved"); + + // Build the per-connection interpreter session once; it + // carries any in-flight transaction across the frames + // that follow (Task 5: BEGIN..COMMIT spanning lines). + let mut session = Session::for_tenant(&ts); + // HELLO already consumed; serve subsequent command // frames on this connection until the peer closes. while let Ok(req) = read_stream_frame(&mut stream) { @@ -150,7 +161,7 @@ impl Server { if line.is_empty() { break; } - handle_command(&session, &mut stream, line); + handle_command(&mut session, &mut stream, line); } }); } @@ -200,16 +211,11 @@ fn resolve_default( } } -/// Execute one command line against the tenant's session store and write the -/// reply frame. This is the single shared interpreter path (mirrors the old -/// `Session::exec`). -fn handle_command( - session: &TenantSession, - stream: &mut std::os::unix::net::UnixStream, - line: &str, -) { - let store = session.store.clone(); - let response = cubesys::commands::exec_on_store(&store, line); +/// Execute one command line against the connection's [`Session`] and write the +/// reply frame. The session is held across frames so `BEGIN`/`COMMIT` keep +/// their transaction state (Task 5). +fn handle_command(session: &mut Session, stream: &mut std::os::unix::net::UnixStream, line: &str) { + let response = session.exec(line); let reply = match response { Ok(out) => out, Err(e) => format!("error: {e}"), diff --git a/cubesys/src/commands.rs b/cubesys/src/commands.rs index 2c6ea81..aef3e1b 100644 --- a/cubesys/src/commands.rs +++ b/cubesys/src/commands.rs @@ -12,13 +12,32 @@ //! so the server can hand a cloned handle to each worker thread. use crate::store::ConcurrentStore; +use crate::tenant::{TenantIdentity, TenantSession}; use cubecode::{CodeCell, Kind, Op, Vm}; -use cubecoords::CubeHeader; +use cubecoords::{CubeHeader, Czyx}; use cubecrypt::{CubeEnv, KeySlot, Selector, TransformId}; +use cubestore::{CubeStore, HashBackend}; use std::collections::BTreeMap; use std::sync::Arc; use std::time::Instant; +/// A buffered write or delete awaiting `COMMIT`. `put: Some((bytes, header))` +/// is a put of a full record (the exact backend bytes `put_record` would +/// write); `put: None` is a delete. +struct TxnOp { + coord: Czyx, + put: Option<(Vec, CubeHeader)>, +} + +/// An in-flight transaction (Task 5). `snapshot` is a consistent clone taken at +/// `BEGIN` so reads during the txn are isolated from concurrent external +/// writers; `ops` is the buffered set of writes/deletes applied atomically on +/// `COMMIT`. +struct Txn { + snapshot: CubeStore, + ops: Vec, +} + /// Per-command latency accumulator (cumulative; the daemon reports these via /// the `stats` command). Counts and sums are exact; mean/max are derived. #[derive(Default, Clone)] @@ -37,6 +56,14 @@ pub struct Session { calls: u64, /// Per top-level command latency histogram (command name -> stats). per_cmd: BTreeMap, + /// The client's declared identity (Task 3): tenant + owner. Stamped by the + /// daemon after a `HELLO`, then threaded into mutating ops for owner + /// enforcement (Tasks 6+). + identity: Option, + /// In-flight transaction (Task 5). `Some` between `BEGIN` and the matching + /// `COMMIT`/`ROLLBACK`; mutating commands buffer into it instead of + /// touching the store live, and reads consult the `BEGIN` snapshot. + txn: Option, } impl Default for Session { @@ -52,6 +79,8 @@ impl Session { store: Arc::new(ConcurrentStore::memory()), calls: 0, per_cmd: BTreeMap::new(), + identity: None, + txn: None, } } @@ -61,9 +90,35 @@ impl Session { store, calls: 0, per_cmd: BTreeMap::new(), + identity: None, + txn: None, } } + /// Build a session bound to a tenant's store + identity (used by the + /// daemon per connection, so a `BEGIN`/`COMMIT` spanning multiple command + /// frames keeps its txn state on the same `Session`). + pub fn for_tenant(ts: &TenantSession) -> Self { + Session { + store: ts.store.clone(), + calls: 0, + per_cmd: BTreeMap::new(), + identity: ts.identity(), + txn: None, + } + } + + /// Stamp the client identity (called by the daemon after a successful + /// `HELLO`, or via [`Session::for_tenant`] from the tenant session). + pub fn set_identity(&mut self, id: TenantIdentity) { + self.identity = Some(id); + } + + /// The currently-stamped identity, if any. + pub fn identity(&self) -> Option { + self.identity.clone() + } + /// Shared handle to the underlying concurrent store. pub fn store(&self) -> Arc { self.store.clone() @@ -133,6 +188,45 @@ impl Session { let cmd = it.next().ok_or_else(|| "empty line".to_string())?; let store = &self.store; match cmd { + "begin" => { + if self.txn.is_some() { + return Err("begin: already in a transaction".to_string()); + } + // Take a consistent snapshot now; reads during the txn consult + // it (isolation from concurrent external writers). + let snap = store.read_snapshot(); + self.txn = Some(Txn { + snapshot: snap, + ops: Vec::new(), + }); + return Ok("ok: transaction begun".to_string()); + } + "commit" => { + let txn = self + .txn + .take() + .ok_or_else(|| "commit: no transaction is open".to_string())?; + let n = txn.ops.len(); + // Apply every buffered op atomically under one store write lock + // and append a SINGLE WAL entry (WalOp::Txn) so the whole batch + // is durable as one unit and replays idempotently. + let batch: Vec = txn + .ops + .iter() + .map(|op| crate::store::TxnEntry { + coord: op.coord, + value: op.put.as_ref().map(|(v, _)| v.clone()), + }) + .collect(); + store.commit_txn(&batch); + return Ok(format!("ok: committed {n} operation(s)")); + } + "rollback" => { + if self.txn.take().is_none() { + return Err("rollback: no transaction is open".to_string()); + } + return Ok("ok: transaction rolled back".to_string()); + } "stats" => Ok(self.stats()), "query" => { let dt = it @@ -168,6 +262,27 @@ impl Session { return Err("prog: no ops given".to_string()); } let name = path.rsplit('/').next().unwrap_or(path); + // Compute the exact record bytes `put_record` would write, using + // a throwaway store so we can buffer (or apply) them without + // duplicating the record codec. + let coord = scratch_code_coord(path, Kind::Fn, name, &ops)?; + let value = { + let mut scratch = CubeStore::new(HashBackend::new()); + crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &ops) + .map_err(|e| e.to_string())?; + scratch.get_raw(&coord).unwrap_or_default() + }; + let header = header_for_code(Kind::Fn, name, &ops); + if let Some(txn) = self.txn.as_mut() { + txn.ops.push(TxnOp { + coord, + put: Some((value, header)), + }); + return Ok(format!( + "buffered prog {path} ({} ops) — commit to apply", + ops.len() + )); + } let coord = store .put_code_cell(path, Kind::Fn, name, &[], &ops) .map_err(|e| e.to_string())?; @@ -186,15 +301,43 @@ impl Session { let code = cubecode::decode(&bytes) .map_err(|e| format!("bytecode decode error: {e:?}"))?; let name = path.rsplit('/').next().unwrap_or(path); + let coord = scratch_code_coord(path, Kind::Fn, name, &code)?; + let value = { + let mut scratch = CubeStore::new(HashBackend::new()); + crate::store_code_cell(&mut scratch, path, Kind::Fn, name, &[], &code) + .map_err(|e| e.to_string())?; + scratch.get_raw(&coord).unwrap_or_default() + }; + let header = header_for_code(Kind::Fn, name, &code); + if let Some(txn) = self.txn.as_mut() { + txn.ops.push(TxnOp { + coord, + put: Some((value, header)), + }); + return Ok(format!( + "buffered write {path} ({} bytes) — commit to apply", + bytes.len() + )); + } let coord = store .put_code_cell(path, Kind::Fn, name, &[], &code) .map_err(|e| e.to_string())?; Ok(format!("wrote {path} -> coord {}", coord.pack_u32())) } + "del" => { + let path = it.next().ok_or_else(|| "del needs ".to_string())?; + let coord = crate::path_to_czyx(path).map_err(|e| e.to_string())?; + if let Some(txn) = self.txn.as_mut() { + txn.ops.push(TxnOp { coord, put: None }); + return Ok(format!("buffered del {path} — commit to apply")); + } + store.delete_raw(&coord); + Ok(format!("deleted {path}")) + } "run" => { let path = it.next().ok_or_else(|| "run needs ".to_string())?; let _coord = crate::path_to_czyx(path).map_err(|e| e.to_string())?; - let sn = store.read_snapshot(); + let sn = txn_snapshot(self); let cell = crate::load_code_cell(&sn, path).map_err(|e| e.to_string())?; let mut vm = Vm::new(sn); let res = vm.run(cell.label); @@ -209,7 +352,8 @@ impl Session { } "ls" => { let dir = it.next().ok_or_else(|| "ls needs ".to_string())?; - let fs = cubefs::CubeFs::new(store.read_snapshot()); + let sn = txn_snapshot(self); + let fs = cubefs::CubeFs::new(sn); let entries = fs.readdir(dir).map_err(|e| format!("ls {dir}: {e:?}"))?; if entries.is_empty() { Ok(format!("ls {dir} -> (empty)")) @@ -220,7 +364,8 @@ impl Session { } "stat" => { let path = it.next().ok_or_else(|| "stat needs ".to_string())?; - let fs = cubefs::CubeFs::new(store.read_snapshot()); + let sn = txn_snapshot(self); + let fs = cubefs::CubeFs::new(sn); let a = fs .getattr(path) .map_err(|e| format!("stat {path}: {e:?}"))?; @@ -230,6 +375,11 @@ impl Session { )) } "seal" | "open" => { + if self.txn.is_some() { + return Err(format!( + "{cmd} inside a transaction is not supported; commit or rollback first" + )); + } let path = it.next().ok_or_else(|| format!("{cmd} needs "))?; let keyc = it .next() @@ -291,6 +441,37 @@ impl Session { } } +/// The store view for read commands: the live store normally, or the +/// transaction's `BEGIN` snapshot while a transaction is open (so reads are +/// isolated from concurrent external writers — [`Txn::snapshot`]). +pub fn txn_snapshot(s: &Session) -> CubeStore { + match &s.txn { + Some(t) => t.snapshot.clone(), + None => s.store.read_snapshot(), + } +} + +/// Compute the coordinate a `store_code_cell` call would target, without +/// writing — used to buffer `prog`/`write` mutations during a transaction. +fn scratch_code_coord(path: &str, kind: Kind, name: &str, code: &[Op]) -> Result { + let mut scratch = CubeStore::new(HashBackend::new()); + crate::store_code_cell(&mut scratch, path, kind, name, &[], code).map_err(|e| e.to_string()) +} + +/// Build the `CubeHeader` a `store_code_cell` call would attach (mirrors +/// `crate::store_code_cell`), so a buffered txn put carries the same header. +fn header_for_code(kind: Kind, name: &str, code: &[Op]) -> CubeHeader { + let mut h = CubeHeader::new(); + h.title = Some(name.to_string()); + h.doc_type = Some(kind.as_str().to_string()); + h.linked_records = Vec::new(); + if h.doc_type.as_deref() == Some("fn") { + h.size_bytes = Some(cubecode::encode(code).len() as u64); + } + h.refresh_flags(); + h +} + /// Parse a byte token: decimal (`42`) or hex (`0x2a`). fn parse_byte(t: &str) -> Option { if let Ok(v) = t.parse::() { @@ -373,3 +554,102 @@ pub fn parse_transform(s: &str) -> Option { _ => None, } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::store::DurabilityConfig; + use std::sync::Arc; + + fn session() -> Session { + Session::with_store(Arc::new(ConcurrentStore::memory())) + } + + #[test] + fn begin_commit_applies_buffered_writes() { + let mut s = session(); + s.exec("begin").unwrap(); + // buffered, not yet visible + assert!(s.exec("prog /c001/z001/y001/x001 const 2 const 3 add halt").is_ok()); + assert!(s.store.get_raw(&Czyx::new(1, 1, 1, 1)).is_none()); + // COMMIT makes all ops durable+visible at once + s.exec("commit").unwrap(); + let v = s.store.get_raw(&Czyx::new(1, 1, 1, 1)); + assert!(v.is_some(), "prog buffered during txn must appear after commit"); + } + + #[test] + fn rollback_discards_buffered_writes() { + let mut s = session(); + s.exec("begin").unwrap(); + let _ = s.exec("write /c002/z001/y001/x001 2a 2b 3c"); + s.exec("rollback").unwrap(); + assert!(s.store.get_raw(&Czyx::new(2, 1, 1, 1)).is_none()); + // begin/commit nested errors + assert!(s.exec("begin").is_ok()); + assert!(s.exec("begin").is_err()); // already in txn + assert!(s.exec("commit").is_ok()); + } + + #[test] + fn txn_isolation_begin_snapshot_hides_live_writer() { + // Stream A opens a txn and snapshots; stream B mutates the live store. + // A's reads (run) must NOT see B's write until A commits. + let mut a = session(); + let mut b = session(); + a.store = b.store.clone(); // shared backend, two sessions + b.exec("prog /c003/z001/y001/x001 const 7 halt").unwrap(); + assert!(b.store.get_raw(&Czyx::new(3, 1, 1, 1)).is_some()); + // A begins AFTER B's write -> snapshot already has it + a.exec("begin").unwrap(); + // B deletes it live; A's snapshot must still see it via its txn view + b.exec("del /c003/z001/y001/x001").unwrap(); + let snap = txn_snapshot(&a); + assert!( + snap.get_raw(&Czyx::new(3, 1, 1, 1)).is_some(), + "txn snapshot must isolate A from B's concurrent delete" + ); + // once A commits (no own writes) the live store reflects B's delete + a.exec("commit").unwrap(); + assert!(b.store.get_raw(&Czyx::new(3, 1, 1, 1)).is_none()); + } + + #[test] + fn commit_is_durable_across_reopen() { + let dir = std::env::temp_dir().join(format!("cube2-txn-{}", std::process::id())); + let _ = std::fs::create_dir_all(&dir); + let db = dir.join("db.cubedb"); + let wal = dir.join("wal.ndjson"); + let rec = dir.join("recovery.jsonl"); + let _ = std::fs::remove_file(&db); + let _ = std::fs::remove_file(&wal); + let cfg = DurabilityConfig::default(); + let cs = ConcurrentStore::open( + db.to_str().unwrap(), + wal.to_str().unwrap(), + rec.to_str().unwrap(), + cfg, + ) + .unwrap(); + let mut s = Session::with_store(Arc::new(cs)); + s.exec("begin").unwrap(); + s.exec("prog /c004/z001/y001/x001 const 9 halt").unwrap(); + s.exec("commit").unwrap(); + s.store.checkpoint(); + drop(s); + + // Reopen: the committed txn must replay from the WAL. + let cs2 = ConcurrentStore::open( + db.to_str().unwrap(), + wal.to_str().unwrap(), + rec.to_str().unwrap(), + DurabilityConfig::default(), + ) + .unwrap(); + assert!( + cs2.get_raw(&Czyx::new(4, 1, 1, 1)).is_some(), + "committed txn must survive reopen via WAL replay" + ); + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/cubesys/src/store.rs b/cubesys/src/store.rs index b85d878..ac8165f 100644 --- a/cubesys/src/store.rs +++ b/cubesys/src/store.rs @@ -66,18 +66,36 @@ const DELTA_COMPACT_BYTES: u64 = 1_048_576; /// `seq` is monotonic so replay can skip already-checkpointed entries. For /// `op:"put"`, `v` is the raw backend value bytes (header_len||hdr||body) /// hex-encoded; replay does `put_raw(coord, v)` — identical to a normal write, -/// so replays are idempotent. +/// so replays are idempotent. For `op:"txn"`, `batch` carries the committed +/// batch (puts/deletes) as one durable unit. struct WalEntry { seq: u64, op: WalOp, coord: Czyx, value: Vec, + /// Only set for `op:"txn"`; empty otherwise. + batch: Vec, } #[derive(Clone, Copy, PartialEq, Eq)] enum WalOp { Put, Delete, + /// A transaction commit: a batch of puts/deletes applied atomically. + /// `Some(value)` is a put (value is the raw backend bytes), `None` is a + /// delete. Replayed idempotently (each entry is a put_raw/delete_raw). + Txn, +} + +/// One entry in a [`WalOp::Txn`] batch: `value: Some(bytes)` is a put, +/// `value: None` is a delete. Carries exactly what the store needs to apply +/// or replay the op — no header parsing required. +#[derive(Clone)] +pub struct TxnEntry { + /// Coordinate to write or delete. + pub coord: Czyx, + /// `Some(raw_backend_bytes)` => put; `None` => delete. + pub value: Option>, } /// The write-ahead log. Append-only NDJSON, group-fsync'd on a timer. @@ -206,6 +224,31 @@ impl Wal { op, coord, value, + batch: Vec::new(), + }); + let mut p = self.pending.lock().unwrap(); + p.push_str(&line); + p.push('\n'); + } + + /// Append a transaction commit. In memory mode this is a no-op. The whole + /// batch lands as ONE WAL line (`op:"txn"`), so a crash mid-txn replays as + /// a single all-or-nothing unit. + fn append_txn(&self, batch: &[TxnEntry]) { + let has_file = self.file.lock().unwrap().is_some(); + if !has_file { + return; + } + let s = self.seq.fetch_add(1, Ordering::SeqCst) + 1; + for e in batch { + self.mark_dirty(e.coord); + } + let line = encode_wal(&WalEntry { + seq: s, + op: WalOp::Txn, + coord: Czyx::new(0, 0, 0, 0), + value: Vec::new(), + batch: batch.to_vec(), }); let mut p = self.pending.lock().unwrap(); p.push_str(&line); @@ -245,6 +288,18 @@ impl Wal { match e.op { WalOp::Put => store.put_raw(e.coord, e.value), WalOp::Delete => store.delete_raw(&e.coord), + WalOp::Txn => { + // A committed batch replays as its puts/deletes. + // Idempotent (each is put_raw/delete_raw), so a + // crash that left the batch half-applied before the + // checkpoint still converges on replay. + for te in &e.batch { + match &te.value { + Some(v) => store.put_raw(te.coord, v.clone()), + None => store.delete_raw(&te.coord), + } + } + } } applied += 1; } @@ -504,6 +559,21 @@ impl ConcurrentStore { self.wal.append(WalOp::Put, key, value); } + /// Apply a transaction batch atomically: under ONE store write lock, apply + /// every put/delete, then append a SINGLE `WalOp::Txn` WAL entry so the + /// whole batch is durable as one unit and replays idempotently. + pub fn commit_txn(&self, batch: &[TxnEntry]) { + let mut g = self.inner.write().unwrap(); + for e in batch { + match &e.value { + Some(v) => g.put_raw(e.coord, v.clone()), + None => g.delete_raw(&e.coord), + } + } + drop(g); // release the write lock before touching the WAL + self.wal.append_txn(batch); + } + /// Store a code cell at `path` (the path->code bridge), durability-logged. pub fn put_code_cell( &self, @@ -761,9 +831,69 @@ fn encode_wal(e: &WalEntry) -> String { "{{\"seq\":{},\"op\":\"del\",\"c\":{},\"z\":{},\"y\":{},\"x\":{}}}", e.seq, e.coord.c, e.coord.z, e.coord.y, e.coord.x ), + WalOp::Txn => format!( + "{{\"seq\":{},\"op\":\"txn\",\"batch\":\"{}\"}}", + e.seq, + to_hex(&pack_txn(&e.batch)) + ), } } +/// Serialize a txn batch into a flat byte buffer (then hex'd for the WAL): +/// for each entry: 1 flag byte (1=put, 0=del) + c,z,y,x (4 bytes) + +/// if put: 4-byte BE length + value bytes. +fn pack_txn(batch: &[TxnEntry]) -> Vec { + let mut out = Vec::new(); + for e in batch { + match &e.value { + Some(v) => { + out.push(1); + out.extend_from_slice(&[e.coord.c, e.coord.z, e.coord.y, e.coord.x]); + out.extend_from_slice(&(v.len() as u32).to_be_bytes()); + out.extend_from_slice(v); + } + None => { + out.push(0); + out.extend_from_slice(&[e.coord.c, e.coord.z, e.coord.y, e.coord.x]); + } + } + } + out +} + +/// Reverse of [`pack_txn`]. +fn unpack_txn(mut buf: &[u8]) -> Result, String> { + let mut out = Vec::new(); + while !buf.is_empty() { + let flag = buf[0]; + buf = &buf[1..]; + if buf.len() < 4 { + return Err("txn entry truncated (coord)".into()); + } + let coord = Czyx::new(buf[0], buf[1], buf[2], buf[3]); + buf = &buf[4..]; + if flag == 1 { + if buf.len() < 4 { + return Err("txn entry truncated (len)".into()); + } + let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize; + buf = &buf[4..]; + if buf.len() < len { + return Err("txn entry truncated (value)".into()); + } + let value = buf[..len].to_vec(); + buf = &buf[len..]; + out.push(TxnEntry { + coord, + value: Some(value), + }); + } else { + out.push(TxnEntry { coord, value: None }); + } + } + Ok(out) +} + fn decode_wal(line: &str) -> Option { let line = line.trim(); if line.is_empty() || !line.starts_with('{') { @@ -785,6 +915,7 @@ fn decode_wal(line: &str) -> Option { op: WalOp::Put, coord, value, + batch: Vec::new(), }) } "del" => Some(WalEntry { @@ -792,7 +923,20 @@ fn decode_wal(line: &str) -> Option { op: WalOp::Delete, coord, value: Vec::new(), + batch: Vec::new(), }), + "txn" => { + let raw = field_str(line, "batch"); + let bytes = from_hex(raw).ok()?; + let batch = unpack_txn(&bytes).ok()?; + Some(WalEntry { + seq, + op: WalOp::Txn, + coord, + value: Vec::new(), + batch, + }) + } _ => None, } }