cubelinux/cubelinux-2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

feat(cubesys): Task 5 transactions (BEGIN/COMMIT/ROLLBACK) + HELLO identity wiring

Browse Source 
- Session gains txn: Option<Txn> (BEGIN snapshot + buffered ops) and
  identity: Option<TenantIdentity> (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.
This commit is contained in:
CUBELinux-2
2026-08-11 12:50:54 -04:00
parent 6209955b46
commit 308e20852c
3 changed files with 450 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cubesys/src/bin/cube-server.rs
+21 -15
View File
@@ -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(&registry, 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}"),
cubesys/src/commands.rs
+284 -4
View File
@@ -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<u8>, 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<HashBackend>,
ops: Vec<TxnOp>,
}
/// 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<String, CmdStat>,
/// 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<TenantIdentity>,
/// 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<Txn>,
}
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<TenantIdentity> {
self.identity.clone()
}
/// Shared handle to the underlying concurrent store.
pub fn store(&self) -> Arc<ConcurrentStore> {
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<crate::store::TxnEntry> = 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 <path>".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 <path>".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 <dir>".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 <path>".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 <path>"))?;
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<HashBackend> {
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<Czyx, String> {
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<u8> {
if let Ok(v) = t.parse::<u8>() {
@@ -373,3 +554,102 @@ pub fn parse_transform(s: &str) -> Option<TransformId> {
_ => 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);
}
}
cubesys/src/store.rs
+145 -1
View File
@@ -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<u8>,
/// Only set for `op:"txn"`; empty otherwise.
batch: Vec<TxnEntry>,
}
#[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<Vec<u8>>,
}
/// 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<u8> {
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<Vec<TxnEntry>, 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<WalEntry> {
let line = line.trim();
if line.is_empty() || !line.starts_with('{') {
@@ -785,6 +915,7 @@ fn decode_wal(line: &str) -> Option<WalEntry> {
op: WalOp::Put,
coord,
value,
batch: Vec::new(),
})
}
"del" => Some(WalEntry {
@@ -792,7 +923,20 @@ fn decode_wal(line: &str) -> Option<WalEntry> {
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,
}
}