From 29751e39b37a904725ce7086ef7e451255398ccf Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 11 Oct 2026 09:33:41 +0000 Subject: [PATCH 1/3] dir-sim: the history as a log of verified append-only batches log (pure, no I/O) is what D-DSP-2's storage crate writes and reads. LogState::batch holds only what an image has beyond the log: values, roots and versions since the last save, tags and verdicts that changed. A batch carries any number of logical versions (the history's own clock, never the storage version), and committed() advances the state only after the store write succeeded; a batch is committed once, in order. Every batch starts with a header: number, first version, version count, record count, FNV-1a checksum. replay refuses the whole log at the first batch with a numbering gap, a missing header, records out of position, a count or checksum mismatch, or versions that do not follow. Nothing is repaired; replay_prefix is the separate call that reads up to the damage and names it. The encoding destructures every struct exhaustively, so a new field is a compile error rather than a field dropped from the log. Decoding rebuilds through validating constructors (RemoteMailbox::new, Dn128::new) and refuses trailing bytes. Tests: every Change, Violation and Recipient variant and a full ObservedNode round-trip, every strict prefix refused; two-batch log restores the same reads; one batch holds five versions; four kinds of torn save refused; the explicit prefix; forward-only commits. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G22yT6htkcdyXsihxxXdrg --- crates/lance-graph-dir-sim/src/lib.rs | 1 + crates/lance-graph-dir-sim/src/log.rs | 1346 +++++++++++++++++++ crates/lance-graph-dir-sim/tests/persist.rs | 137 ++ 3 files changed, 1484 insertions(+) create mode 100644 crates/lance-graph-dir-sim/src/log.rs diff --git a/crates/lance-graph-dir-sim/src/lib.rs b/crates/lance-graph-dir-sim/src/lib.rs index 964b34f7b..a263ac316 100644 --- a/crates/lance-graph-dir-sim/src/lib.rs +++ b/crates/lance-graph-dir-sim/src/lib.rs @@ -43,6 +43,7 @@ mod exec; pub mod groups; pub mod image; pub mod ldap; +pub mod log; pub mod observe; pub mod proxy; pub mod rule; diff --git a/crates/lance-graph-dir-sim/src/log.rs b/crates/lance-graph-dir-sim/src/log.rs new file mode 100644 index 000000000..0cb7611c8 --- /dev/null +++ b/crates/lance-graph-dir-sim/src/log.rs @@ -0,0 +1,1346 @@ +//! A history as an append-only log of verified batches (W-3, decision 3(b)). +//! +//! Persisting a [`StoredImage`] writes only what is new since the last save, +//! as one **batch** of records; a storage crate puts one batch in one commit. +//! The logical [`VersionId`]s inside a batch are the history's own clock and +//! never the storage version: one batch may carry many logical versions, or +//! none (a tag moved, a verdict recorded). +//! +//! Every batch starts with a header record giving its number, the logical +//! versions it holds, its record count and a checksum over its records. +//! [`replay`] reads batches in number order and refuses the whole log at the +//! first batch that fails a check: a gap in the numbering, a missing or +//! misplaced header, a record count or checksum that does not match, or a +//! version that does not follow the previous batch. Nothing is repaired and +//! no prefix is returned by [`replay`]; [`replay_prefix`] is the separate, +//! explicit call that reads up to the last good batch and names the damage. +//! +//! Pure: no I/O. A storage crate turns a [`Record`] into a row and back. +//! +//! The encoding is little-endian with fixed widths: `Option` is a `0`/`1` +//! tag, a sequence a `u32` count, a string a `u32` byte length and UTF-8. +//! Enum tags are fixed numbers, never reused. Every struct is encoded by +//! exhaustive destructuring, so a new field fails to compile here instead +//! of being dropped from the log. + +use crate::image::{StoredImage, StoredOrigin, StoredRule, StoredVersion}; +use crate::snapshot::{Observation, ObservedNode, ObservedRecipient}; +use ogar_dir_core::{DirectoryScope, Dn128, Guid128}; +use ogar_dir_sim::{ + AddressRole, ArchiveState, Attribute, Change, Endpoint, EvidenceRef, KeyId, MailboxState, + NodeKind, NodeState, Recipient, RecipientAttributes, RemoteKind, RemoteMailbox, ValueId, + VersionId, Violation, +}; +use std::collections::BTreeMap; + +// ---------------------------------------------------------------- bytes + +/// Why a record body did not decode. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DecodeError { + /// Byte offset in the body. + pub at: usize, + /// What was expected there. + pub what: &'static str, +} + +#[derive(Default)] +struct W(Vec); + +impl W { + fn u8(&mut self, v: u8) { + self.0.push(v); + } + fn u16(&mut self, v: u16) { + self.0.extend_from_slice(&v.to_le_bytes()); + } + fn u32(&mut self, v: u32) { + self.0.extend_from_slice(&v.to_le_bytes()); + } + fn i32(&mut self, v: i32) { + self.0.extend_from_slice(&v.to_le_bytes()); + } + fn u64(&mut self, v: u64) { + self.0.extend_from_slice(&v.to_le_bytes()); + } + fn i64(&mut self, v: i64) { + self.0.extend_from_slice(&v.to_le_bytes()); + } + fn bool(&mut self, v: bool) { + self.u8(u8::from(v)); + } + fn len(&mut self, n: usize) { + self.u32(u32::try_from(n).expect("a record part holds fewer than 2^32 items")); + } + fn str(&mut self, s: &str) { + self.len(s.len()); + self.0.extend_from_slice(s.as_bytes()); + } + fn guid(&mut self, g: Guid128) { + self.0.extend_from_slice(&g.0); + } + fn opt(&mut self, v: Option, f: impl FnOnce(&mut Self, T)) { + match v { + None => self.u8(0), + Some(x) => { + self.u8(1); + f(self, x); + } + } + } + fn seq(&mut self, items: &[T], mut f: impl FnMut(&mut Self, &T)) { + self.len(items.len()); + for x in items { + f(self, x); + } + } +} + +struct R<'a> { + b: &'a [u8], + at: usize, +} + +impl<'a> R<'a> { + fn new(b: &'a [u8]) -> Self { + Self { b, at: 0 } + } + fn err(&self, what: &'static str) -> Result { + Err(DecodeError { at: self.at, what }) + } + fn take(&mut self, n: usize, what: &'static str) -> Result<&'a [u8], DecodeError> { + match self.b.get(self.at..self.at.saturating_add(n)) { + Some(s) if s.len() == n => { + self.at += n; + Ok(s) + } + _ => self.err(what), + } + } + fn u8(&mut self) -> Result { + Ok(self.take(1, "u8")?[0]) + } + fn u16(&mut self) -> Result { + Ok(u16::from_le_bytes( + self.take(2, "u16")?.try_into().expect("2"), + )) + } + fn u32(&mut self) -> Result { + Ok(u32::from_le_bytes( + self.take(4, "u32")?.try_into().expect("4"), + )) + } + fn i32(&mut self) -> Result { + Ok(i32::from_le_bytes( + self.take(4, "i32")?.try_into().expect("4"), + )) + } + fn u64(&mut self) -> Result { + Ok(u64::from_le_bytes( + self.take(8, "u64")?.try_into().expect("8"), + )) + } + fn i64(&mut self) -> Result { + Ok(i64::from_le_bytes( + self.take(8, "i64")?.try_into().expect("8"), + )) + } + fn bool(&mut self) -> Result { + match self.u8()? { + 0 => Ok(false), + 1 => Ok(true), + _ => self.err("bool"), + } + } + fn len(&mut self) -> Result { + let n = self.u32()? as usize; + // A count can never exceed the bytes left: every item takes one. + if n > self.b.len() - self.at { + return self.err("count"); + } + Ok(n) + } + fn string(&mut self) -> Result { + let n = self.len()?; + let at = self.at; + let s = self.take(n, "string")?; + String::from_utf8(s.to_vec()).map_err(|_| DecodeError { at, what: "utf-8" }) + } + fn guid(&mut self) -> Result { + Ok(Guid128(self.take(16, "guid")?.try_into().expect("16"))) + } + fn opt( + &mut self, + f: impl FnOnce(&mut Self) -> Result, + ) -> Result, DecodeError> { + match self.u8()? { + 0 => Ok(None), + 1 => f(self).map(Some), + _ => self.err("option tag"), + } + } + fn seq( + &mut self, + mut f: impl FnMut(&mut Self) -> Result, + ) -> Result, DecodeError> { + let n = self.len()?; + (0..n).map(|_| f(self)).collect() + } + fn end(&self) -> Result<(), DecodeError> { + if self.at == self.b.len() { + Ok(()) + } else { + self.err("end of record") + } + } +} + +// ---------------------------------------------------------------- types + +fn w_node_kind(w: &mut W, k: NodeKind) { + w.u8(match k { + NodeKind::User => 0, + NodeKind::Group => 1, + }); +} +fn r_node_kind(r: &mut R<'_>) -> Result { + match r.u8()? { + 0 => Ok(NodeKind::User), + 1 => Ok(NodeKind::Group), + _ => r.err("node kind"), + } +} + +fn w_dn(w: &mut W, d: Dn128) { + w.seq(d.codes(), |w, c| w.u8(*c)); +} +fn r_dn(r: &mut R<'_>) -> Result { + let at = r.at; + let codes = r.seq(R::u8)?; + Dn128::new(&codes).map_err(|_| DecodeError { at, what: "dn" }) +} + +fn w_archive(w: &mut W, a: ArchiveState) { + w.u8(match a { + ArchiveState::None => 0, + ArchiveState::Provisioned => 1, + ArchiveState::Deprovisioned => 2, + }); +} +fn r_archive(r: &mut R<'_>) -> Result { + match r.u8()? { + 0 => Ok(ArchiveState::None), + 1 => Ok(ArchiveState::Provisioned), + 2 => Ok(ArchiveState::Deprovisioned), + _ => r.err("archive state"), + } +} + +fn w_recipient(w: &mut W, rc: Recipient) { + match rc { + Recipient::NotMailEnabled => w.u8(0), + Recipient::OnPremisesMailbox { archive } => { + w.u8(1); + w_archive(w, archive); + } + Recipient::RemoteMailbox(m) => { + w.u8(2); + w.u8(match m.kind() { + RemoteKind::User => 0, + RemoteKind::Room => 1, + RemoteKind::Equipment => 2, + RemoteKind::Shared => 3, + }); + w.u8(match m.mailbox() { + MailboxState::Provisioned => 0, + MailboxState::Migrated => 1, + MailboxState::Deprovisioned => 2, + }); + w_archive(w, m.archive()); + w.opt(m.routing(), |w, v| w.u32(v.0)); + } + Recipient::Other(a) => { + w.u8(3); + let RecipientAttributes { + remote_recipient_type, + display_type, + type_details, + target_address, + } = a; + w.opt(remote_recipient_type, W::u32); + w.opt(display_type, W::i32); + w.opt(type_details, W::u64); + w.opt(target_address, |w, v| w.u32(v.0)); + } + } +} +fn r_recipient(r: &mut R<'_>) -> Result { + match r.u8()? { + 0 => Ok(Recipient::NotMailEnabled), + 1 => Ok(Recipient::OnPremisesMailbox { + archive: r_archive(r)?, + }), + 2 => { + let at = r.at; + let kind = match r.u8()? { + 0 => RemoteKind::User, + 1 => RemoteKind::Room, + 2 => RemoteKind::Equipment, + 3 => RemoteKind::Shared, + _ => return r.err("remote kind"), + }; + let mailbox = match r.u8()? { + 0 => MailboxState::Provisioned, + 1 => MailboxState::Migrated, + 2 => MailboxState::Deprovisioned, + _ => return r.err("mailbox state"), + }; + let archive = r_archive(r)?; + let routing = r.opt(|r| r.u32().map(ValueId))?; + RemoteMailbox::new(kind, mailbox, archive, routing) + .map(Recipient::RemoteMailbox) + .ok_or(DecodeError { + at, + what: "remote mailbox", + }) + } + 3 => Ok(Recipient::Other(RecipientAttributes { + remote_recipient_type: r.opt(R::u32)?, + display_type: r.opt(R::i32)?, + type_details: r.opt(R::u64)?, + target_address: r.opt(|r| r.u32().map(ValueId))?, + })), + _ => r.err("recipient"), + } +} + +fn w_state(w: &mut W, s: &NodeState) { + let NodeState { + kind, + active, + upn, + primary_smtp, + dn, + recipient, + } = *s; + w_node_kind(w, kind); + w.opt(active, W::bool); + w.opt(upn, |w, v| w.u32(v.0)); + w.opt(primary_smtp, |w, v| w.u32(v.0)); + w.opt(dn, w_dn); + w.opt(recipient, w_recipient); +} +fn r_state(r: &mut R<'_>) -> Result { + Ok(NodeState { + kind: r_node_kind(r)?, + active: r.opt(R::bool)?, + upn: r.opt(|r| r.u32().map(ValueId))?, + primary_smtp: r.opt(|r| r.u32().map(ValueId))?, + dn: r.opt(r_dn)?, + recipient: r.opt(r_recipient)?, + }) +} + +fn w_change(w: &mut W, c: &Change) { + match *c { + Change::RemoveMembership { user, group } => { + w.u8(0); + w.guid(user); + w.guid(group); + } + Change::DeleteNode { node, ref state } => { + w.u8(1); + w.guid(node); + w_state(w, state); + } + Change::SetAttribute { + node, + attribute, + from, + to, + } => { + w.u8(2); + w.guid(node); + w.u8(match attribute { + Attribute::Upn => 0, + Attribute::PrimarySmtp => 1, + }); + w.opt(from, |w, v| w.u32(v.0)); + w.opt(to, |w, v| w.u32(v.0)); + } + Change::SetActive { node, from, to } => { + w.u8(3); + w.guid(node); + w.opt(from, W::bool); + w.opt(to, W::bool); + } + Change::SetLocation { node, from, to } => { + w.u8(4); + w.guid(node); + w.opt(from, w_dn); + w.opt(to, w_dn); + } + Change::SetRecipient { node, from, to } => { + w.u8(5); + w.guid(node); + w.opt(from, w_recipient); + w.opt(to, w_recipient); + } + Change::CreateNode { node, ref state } => { + w.u8(6); + w.guid(node); + w_state(w, state); + } + Change::AddMembership { user, group } => { + w.u8(7); + w.guid(user); + w.guid(group); + } + } +} +fn r_change(r: &mut R<'_>) -> Result { + Ok(match r.u8()? { + 0 => Change::RemoveMembership { + user: r.guid()?, + group: r.guid()?, + }, + 1 => Change::DeleteNode { + node: r.guid()?, + state: r_state(r)?, + }, + 2 => Change::SetAttribute { + node: r.guid()?, + attribute: match r.u8()? { + 0 => Attribute::Upn, + 1 => Attribute::PrimarySmtp, + _ => return r.err("attribute"), + }, + from: r.opt(|r| r.u32().map(ValueId))?, + to: r.opt(|r| r.u32().map(ValueId))?, + }, + 3 => Change::SetActive { + node: r.guid()?, + from: r.opt(R::bool)?, + to: r.opt(R::bool)?, + }, + 4 => Change::SetLocation { + node: r.guid()?, + from: r.opt(r_dn)?, + to: r.opt(r_dn)?, + }, + 5 => Change::SetRecipient { + node: r.guid()?, + from: r.opt(r_recipient)?, + to: r.opt(r_recipient)?, + }, + 6 => Change::CreateNode { + node: r.guid()?, + state: r_state(r)?, + }, + 7 => Change::AddMembership { + user: r.guid()?, + group: r.guid()?, + }, + _ => return r.err("change"), + }) +} + +fn w_role(w: &mut W, a: AddressRole) { + w.u8(match a { + AddressRole::Upn => 0, + AddressRole::PrimarySmtp => 1, + AddressRole::SecondarySmtp => 2, + AddressRole::Mail => 3, + AddressRole::Routing => 4, + }); +} +fn r_role(r: &mut R<'_>) -> Result { + match r.u8()? { + 0 => Ok(AddressRole::Upn), + 1 => Ok(AddressRole::PrimarySmtp), + 2 => Ok(AddressRole::SecondarySmtp), + 3 => Ok(AddressRole::Mail), + 4 => Ok(AddressRole::Routing), + _ => r.err("address role"), + } +} + +fn w_violation(w: &mut W, v: &Violation) { + match v { + Violation::DuplicateSmtp { key, owners } => { + w.u8(0); + w.u32(key.0); + w.seq(owners, |w, g| w.guid(*g)); + } + Violation::DuplicateUpn { key, owners } => { + w.u8(1); + w.u32(key.0); + w.seq(owners, |w, g| w.guid(*g)); + } + Violation::AddressConflict { key, holders } => { + w.u8(2); + w.u32(key.0); + w.seq(holders, |w, (g, a)| { + w.guid(*g); + w_role(w, *a); + }); + } + Violation::RoutingNotInProxies { node } => { + w.u8(3); + w.guid(*node); + } + Violation::RoutingMismatch { node } => { + w.u8(4); + w.guid(*node); + } + Violation::DanglingMembership { + user, + group, + missing, + } => { + w.u8(5); + w.guid(*user); + w.guid(*group); + w.u8(match missing { + Endpoint::User => 0, + Endpoint::Group => 1, + }); + } + } +} +fn r_violation(r: &mut R<'_>) -> Result { + Ok(match r.u8()? { + 0 => Violation::DuplicateSmtp { + key: KeyId(r.u32()?), + owners: r.seq(R::guid)?, + }, + 1 => Violation::DuplicateUpn { + key: KeyId(r.u32()?), + owners: r.seq(R::guid)?, + }, + 2 => Violation::AddressConflict { + key: KeyId(r.u32()?), + holders: r.seq(|r| Ok((r.guid()?, r_role(r)?)))?, + }, + 3 => Violation::RoutingNotInProxies { node: r.guid()? }, + 4 => Violation::RoutingMismatch { node: r.guid()? }, + 5 => Violation::DanglingMembership { + user: r.guid()?, + group: r.guid()?, + missing: match r.u8()? { + 0 => Endpoint::User, + 1 => Endpoint::Group, + _ => return r.err("endpoint"), + }, + }, + _ => return r.err("violation"), + }) +} + +fn w_version(w: &mut W, v: &StoredVersion) { + let StoredVersion { + id, + parent, + origin, + delta, + } = v; + w.u64(id.0); + w.opt(*parent, |w, p| w.u64(p.0)); + match origin { + StoredOrigin::Observed { + source, + observed_at_ms, + } => { + w.u8(0); + w.str(source); + w.i64(*observed_at_ms); + } + StoredOrigin::Simulated { rule, evidence } => { + let StoredRule { name, version } = rule; + w.u8(1); + w.str(name); + w.u16(*version); + w.seq(evidence, |w, e| w.str(&e.0)); + } + } + w.seq(delta, w_change); +} +fn r_version(r: &mut R<'_>) -> Result { + let id = VersionId(r.u64()?); + let parent = r.opt(|r| r.u64().map(VersionId))?; + let origin = match r.u8()? { + 0 => StoredOrigin::Observed { + source: r.string()?, + observed_at_ms: r.i64()?, + }, + 1 => StoredOrigin::Simulated { + rule: StoredRule { + name: r.string()?, + version: r.u16()?, + }, + evidence: r.seq(|r| r.string().map(EvidenceRef))?, + }, + _ => return r.err("origin"), + }; + Ok(StoredVersion { + id, + parent, + origin, + delta: r.seq(r_change)?, + }) +} + +fn w_observed(w: &mut W, n: &ObservedNode) { + let ObservedNode { + kind, + active, + upn, + primary_smtp, + proxies, + dn, + recipient, + mail, + alias, + exchange_guid, + security, + } = n; + w_node_kind(w, *kind); + w.opt(*active, W::bool); + w.opt(upn.as_deref(), W::str); + w.opt(primary_smtp.as_deref(), W::str); + w.seq(proxies, |w, p| w.str(p)); + w.opt(*dn, w_dn); + w.opt(recipient.as_ref(), |w, rc| { + let ObservedRecipient { + remote_recipient_type, + display_type, + type_details, + target_address, + } = rc; + w.opt(*remote_recipient_type, W::u32); + w.opt(*display_type, W::i32); + w.opt(*type_details, W::u64); + w.opt(target_address.as_deref(), W::str); + }); + w.opt(mail.as_deref(), W::str); + w.opt(alias.as_deref(), W::str); + w.opt(*exchange_guid, W::guid); + w.opt(*security, W::bool); +} +fn r_observed(r: &mut R<'_>) -> Result { + Ok(ObservedNode { + kind: r_node_kind(r)?, + active: r.opt(R::bool)?, + upn: r.opt(R::string)?, + primary_smtp: r.opt(R::string)?, + proxies: r.seq(R::string)?, + dn: r.opt(r_dn)?, + recipient: r.opt(|r| { + Ok(ObservedRecipient { + remote_recipient_type: r.opt(R::u32)?, + display_type: r.opt(R::i32)?, + type_details: r.opt(R::u64)?, + target_address: r.opt(R::string)?, + }) + })?, + mail: r.opt(R::string)?, + alias: r.opt(R::string)?, + exchange_guid: r.opt(R::guid)?, + security: r.opt(R::bool)?, + }) +} + +fn w_observation(w: &mut W, o: &Observation) { + let Observation { + scope, + nodes, + members, + } = o; + w.guid(scope.0); + w.seq(nodes, |w, (g, n)| { + w.guid(*g); + w_observed(w, n); + }); + w.seq(members, |w, (u, g)| { + w.guid(*u); + w.guid(*g); + }); +} +fn r_observation(r: &mut R<'_>) -> Result { + Ok(Observation { + scope: DirectoryScope(r.guid()?), + nodes: r.seq(|r| Ok((r.guid()?, r_observed(r)?)))?, + members: r.seq(|r| Ok((r.guid()?, r.guid()?)))?, + }) +} + +// ---------------------------------------------------------------- records + +/// What a record holds. The numbers are stored and never reused. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +#[repr(u8)] +pub enum RecordKind { + /// The batch header: first record of every batch. + Header = 0, + /// A dictionary value, by index. + Value = 1, + /// A root's observation, by version. + Root = 2, + /// A version. + Version = 3, + /// A tag set to a version. + Tag = 4, + /// A verdict recorded for a version. + Verdict = 5, +} + +impl RecordKind { + fn from_u8(k: u8) -> Option { + Some(match k { + 0 => Self::Header, + 1 => Self::Value, + 2 => Self::Root, + 3 => Self::Version, + 4 => Self::Tag, + 5 => Self::Verdict, + _ => return None, + }) + } +} + +/// One record as a storage crate stores it: a row. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Record { + /// The batch it belongs to. + pub batch: u64, + /// Its position in the batch; the header is `0`. + pub seq: u32, + /// What it holds ([`RecordKind`] as a number). + pub kind: u8, + /// The encoded body. + pub body: Vec, +} + +/// A batch header. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Header { + /// The batch number; batches are numbered from 0, densely. + pub batch: u64, + /// The first logical version the batch holds. + pub first_version: u64, + /// How many logical versions it holds. + pub versions: u64, + /// How many records follow the header. + pub records: u32, + /// FNV-1a 64 over every record after the header, in order. + pub checksum: u64, +} + +fn fnv1a(records: &[Record]) -> u64 { + let mut h: u64 = 0xcbf2_9ce4_8422_2325; + let mut eat = |b: u8| { + h ^= u64::from(b); + h = h.wrapping_mul(0x0000_0100_0000_01b3); + }; + for r in records { + eat(r.kind); + for b in u32::try_from(r.body.len()) + .expect("record body under 4 GiB") + .to_le_bytes() + { + eat(b); + } + for &b in &r.body { + eat(b); + } + } + h +} + +/// A batch ready to store: its records, header first. +#[derive(Clone, Debug)] +pub struct Batch { + /// The records, in `seq` order. + pub records: Vec, + next: LogState, +} + +impl Batch { + /// The batch number. + pub fn number(&self) -> u64 { + self.records[0].batch + } +} + +// ---------------------------------------------------------------- writer + +/// What the log already holds: everything [`LogState::batch`] leaves out. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct LogState { + next_batch: u64, + values: usize, + versions: u64, + tags: BTreeMap, + verdicts: BTreeMap>, +} + +/// Why an image cannot be appended to the log. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum LogError { + /// The image has fewer values or versions than the log already holds: + /// it is not a continuation of the logged history. + NotAContinuation, + /// A tag the log holds is gone from the image; tags are never removed. + TagRemoved(String), + /// A verdict the log holds is gone from the image. + VerdictRemoved(VersionId), + /// The batch was built against another log state (a batch is committed + /// once, in number order). + StaleBatch { + /// The batch the log expects next. + expected: u64, + /// The batch offered. + found: u64, + }, +} + +impl LogState { + /// An empty log. + pub fn new() -> Self { + Self::default() + } + + /// The number the next batch gets. + pub fn next_batch(&self) -> u64 { + self.next_batch + } + + /// The batch holding what `image` has and the log does not, or `None` + /// when there is nothing new. The log state is unchanged until the + /// batch is [`committed`](Self::committed). + /// + /// # Errors + /// + /// [`LogError`] when `image` does not extend the logged history. + pub fn batch(&self, image: &StoredImage) -> Result, LogError> { + let versions = image.versions.len() as u64; + if image.values.len() < self.values || versions < self.versions { + return Err(LogError::NotAContinuation); + } + let tags: BTreeMap = image.tags.iter().cloned().collect(); + let verdicts: BTreeMap> = + image.verdicts.iter().cloned().collect(); + if let Some(t) = self.tags.keys().find(|t| !tags.contains_key(*t)) { + return Err(LogError::TagRemoved(t.clone())); + } + if let Some(v) = self.verdicts.keys().find(|v| !verdicts.contains_key(v)) { + return Err(LogError::VerdictRemoved(*v)); + } + + let mut body: Vec<(RecordKind, Vec)> = Vec::new(); + for (i, value) in image.values.iter().enumerate().skip(self.values) { + let mut w = W::default(); + w.u32(u32::try_from(i).expect("fewer than 2^32 values")); + w.str(value); + body.push((RecordKind::Value, w.0)); + } + for (v, obs) in &image.roots { + if v.0 >= self.versions { + let mut w = W::default(); + w.u64(v.0); + w_observation(&mut w, obs); + body.push((RecordKind::Root, w.0)); + } + } + for v in &image.versions[self.versions as usize..] { + let mut w = W::default(); + w_version(&mut w, v); + body.push((RecordKind::Version, w.0)); + } + for (name, v) in &tags { + if self.tags.get(name) != Some(v) { + let mut w = W::default(); + w.str(name); + w.u64(v.0); + body.push((RecordKind::Tag, w.0)); + } + } + for (v, x) in &verdicts { + if self.verdicts.get(v) != Some(x) { + let mut w = W::default(); + w.u64(v.0); + w.seq(x, w_violation); + body.push((RecordKind::Verdict, w.0)); + } + } + if body.is_empty() { + return Ok(None); + } + + let batch = self.next_batch; + let records: Vec = body + .into_iter() + .enumerate() + .map(|(i, (kind, body))| Record { + batch, + seq: u32::try_from(i + 1).expect("fewer than 2^32 records in a batch"), + kind: kind as u8, + body, + }) + .collect(); + let header = Header { + batch, + first_version: self.versions, + versions: versions - self.versions, + records: u32::try_from(records.len()).expect("fewer than 2^32 records"), + checksum: fnv1a(&records), + }; + let mut all = Vec::with_capacity(records.len() + 1); + all.push(Record { + batch, + seq: 0, + kind: RecordKind::Header as u8, + body: encode_header(&header), + }); + all.extend(records); + Ok(Some(Batch { + records: all, + next: LogState { + next_batch: batch + 1, + values: image.values.len(), + versions, + tags, + verdicts, + }, + })) + } + + /// Record that `batch` is durably stored. Call only after the storage + /// commit succeeded. + /// + /// # Errors + /// + /// [`LogError::StaleBatch`] if `batch` is not the next one. + pub fn committed(&mut self, batch: &Batch) -> Result<(), LogError> { + if batch.number() != self.next_batch { + return Err(LogError::StaleBatch { + expected: self.next_batch, + found: batch.number(), + }); + } + *self = batch.next.clone(); + Ok(()) + } +} + +fn encode_header(h: &Header) -> Vec { + let Header { + batch, + first_version, + versions, + records, + checksum, + } = *h; + let mut w = W::default(); + w.u64(batch); + w.u64(first_version); + w.u64(versions); + w.u32(records); + w.u64(checksum); + w.0 +} + +fn decode_header(b: &[u8]) -> Result { + let mut r = R::new(b); + let h = Header { + batch: r.u64()?, + first_version: r.u64()?, + versions: r.u64()?, + records: r.u32()?, + checksum: r.u64()?, + }; + r.end()?; + Ok(h) +} + +// ---------------------------------------------------------------- reader + +/// Why a log was refused. Every variant names the batch it found damaged. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ReplayError { + /// A batch number is missing (or a batch appears twice). + BatchGap { + /// The number expected. + expected: u64, + /// The number found. + found: u64, + }, + /// A batch whose first record is not a header for that batch. + MissingHeader(u64), + /// Record positions are not `0..=count` in order. + RecordOrder(u64), + /// The batch holds another number of records than its header says. + RecordCount { + /// The batch. + batch: u64, + /// The header's count. + expected: u32, + /// The records present. + found: u32, + }, + /// The records do not hash to the header's checksum. + Checksum(u64), + /// A record of an unknown kind. + UnknownKind(u64, u8), + /// A record body that does not decode. + Decode(u64, DecodeError), + /// A value at another index than the next one. + ValueOrder(u64), + /// A version that does not follow the previous one, or a count that + /// disagrees with the header. + VersionOrder(u64), +} + +/// The image a log holds, and the log state to append after it. +/// +/// # Errors +/// +/// [`ReplayError`] at the first damaged batch; nothing is returned for the +/// batches before it. +pub fn replay( + records: impl IntoIterator, +) -> Result<(StoredImage, LogState), ReplayError> { + let (image, state, err) = replay_prefix(records); + match err { + None => Ok((image, state)), + Some(e) => Err(e), + } +} + +/// [`replay`] up to the last good batch: the image and log state of every +/// batch before the first damaged one, and the damage, if any. +/// +/// For inspecting a damaged log. Appending after a prefix would overwrite +/// the damaged batch's number, so a caller decides that explicitly. +pub fn replay_prefix( + records: impl IntoIterator, +) -> (StoredImage, LogState, Option) { + let mut records: Vec = records.into_iter().collect(); + records.sort_by_key(|r| (r.batch, r.seq)); + let mut image = StoredImage::default(); + let mut state = LogState::new(); + let mut tags: BTreeMap = BTreeMap::new(); + let mut verdicts: BTreeMap> = BTreeMap::new(); + let mut err = None; + for run in records.chunk_by(|a, b| a.batch == b.batch) { + let b = run[0].batch; + let mut next = image.clone(); + let (mut ntags, mut nverdicts) = (tags.clone(), verdicts.clone()); + match apply_batch( + b, + state.next_batch, + run, + &mut next, + &mut ntags, + &mut nverdicts, + ) { + Ok(()) => { + image = next; + tags = ntags; + verdicts = nverdicts; + state.next_batch = b + 1; + } + Err(e) => { + err = Some(e); + break; + } + } + } + image.tags = tags.clone().into_iter().collect(); + image.verdicts = verdicts.clone().into_iter().collect(); + state.values = image.values.len(); + state.versions = image.versions.len() as u64; + state.tags = tags; + state.verdicts = verdicts; + (image, state, err) +} + +fn apply_batch( + b: u64, + expected: u64, + run: &[Record], + image: &mut StoredImage, + tags: &mut BTreeMap, + verdicts: &mut BTreeMap>, +) -> Result<(), ReplayError> { + if b != expected { + return Err(ReplayError::BatchGap { expected, found: b }); + } + let head = &run[0]; + if head.seq != 0 || head.kind != RecordKind::Header as u8 { + return Err(ReplayError::MissingHeader(b)); + } + let header = decode_header(&head.body).map_err(|e| ReplayError::Decode(b, e))?; + if header.batch != b { + return Err(ReplayError::MissingHeader(b)); + } + let rest = &run[1..]; + if rest + .iter() + .enumerate() + .any(|(i, r)| r.seq as usize != i + 1 || r.kind == RecordKind::Header as u8) + { + return Err(ReplayError::RecordOrder(b)); + } + let found = u32::try_from(rest.len()).unwrap_or(u32::MAX); + if found != header.records { + return Err(ReplayError::RecordCount { + batch: b, + expected: header.records, + found, + }); + } + if fnv1a(rest) != header.checksum { + return Err(ReplayError::Checksum(b)); + } + if header.first_version != image.versions.len() as u64 { + return Err(ReplayError::VersionOrder(b)); + } + let decode = |e| ReplayError::Decode(b, e); + let mut added = 0u64; + for rec in rest { + let kind = RecordKind::from_u8(rec.kind).ok_or(ReplayError::UnknownKind(b, rec.kind))?; + let mut r = R::new(&rec.body); + match kind { + RecordKind::Header => unreachable!("rejected above"), + RecordKind::Value => { + let i = r.u32().map_err(decode)? as usize; + let s = r.string().map_err(decode)?; + if i != image.values.len() { + return Err(ReplayError::ValueOrder(b)); + } + image.values.push(s); + } + RecordKind::Root => { + let v = VersionId(r.u64().map_err(decode)?); + image + .roots + .push((v, r_observation(&mut r).map_err(decode)?)); + } + RecordKind::Version => { + let v = r_version(&mut r).map_err(decode)?; + if v.id.0 != image.versions.len() as u64 { + return Err(ReplayError::VersionOrder(b)); + } + image.versions.push(v); + added += 1; + } + RecordKind::Tag => { + let name = r.string().map_err(decode)?; + tags.insert(name, VersionId(r.u64().map_err(decode)?)); + } + RecordKind::Verdict => { + let v = VersionId(r.u64().map_err(decode)?); + verdicts.insert(v, r.seq(r_violation).map_err(decode)?); + } + } + r.end().map_err(decode)?; + } + if added != header.versions { + return Err(ReplayError::VersionOrder(b)); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn g(n: u8) -> Guid128 { + Guid128([n; 16]) + } + + fn round( + x: &T, + w: impl Fn(&mut W, &T), + r: impl Fn(&mut R<'_>) -> Result, + ) { + let mut out = W::default(); + w(&mut out, x); + let mut rd = R::new(&out.0); + assert_eq!(&r(&mut rd).unwrap(), x); + rd.end().unwrap(); + // Every strict prefix is refused, never read as something shorter. + for n in 0..out.0.len() { + let mut rd = R::new(&out.0[..n]); + assert!( + r(&mut rd).and_then(|_| rd.end()).is_err(), + "prefix {n} of {x:?}" + ); + } + } + + fn recipients() -> Vec { + vec![ + Recipient::NotMailEnabled, + Recipient::OnPremisesMailbox { + archive: ArchiveState::Provisioned, + }, + Recipient::RemoteMailbox( + RemoteMailbox::new( + RemoteKind::Shared, + MailboxState::Migrated, + ArchiveState::Deprovisioned, + Some(ValueId(9)), + ) + .unwrap(), + ), + Recipient::Other(RecipientAttributes { + remote_recipient_type: Some(4), + display_type: Some(-2_147_483_642), + type_details: Some(2_147_483_648), + target_address: Some(ValueId(3)), + }), + ] + } + + fn state(rc: Option) -> NodeState { + NodeState { + kind: NodeKind::User, + active: Some(false), + upn: Some(ValueId(1)), + primary_smtp: None, + dn: Some(Dn128::new(&[0, 7, 255]).unwrap()), + recipient: rc, + } + } + + #[test] + fn every_change_round_trips() { + let mut changes = vec![ + Change::RemoveMembership { + user: g(1), + group: g(2), + }, + Change::DeleteNode { + node: g(3), + state: state(None), + }, + Change::SetAttribute { + node: g(4), + attribute: Attribute::PrimarySmtp, + from: None, + to: Some(ValueId(5)), + }, + Change::SetActive { + node: g(5), + from: Some(true), + to: None, + }, + Change::SetLocation { + node: g(6), + from: None, + to: Some(Dn128::new(&[1]).unwrap()), + }, + Change::CreateNode { + node: g(7), + state: state(Some(Recipient::NotMailEnabled)), + }, + Change::AddMembership { + user: g(8), + group: g(9), + }, + ]; + for rc in recipients() { + changes.push(Change::SetRecipient { + node: g(10), + from: Some(rc), + to: None, + }); + } + for c in &changes { + round(c, w_change, r_change); + } + } + + #[test] + fn every_violation_round_trips() { + for v in [ + Violation::DuplicateSmtp { + key: KeyId(1), + owners: vec![g(1), g(2)], + }, + Violation::DuplicateUpn { + key: KeyId(2), + owners: vec![], + }, + Violation::AddressConflict { + key: KeyId(3), + holders: vec![ + (g(1), AddressRole::Upn), + (g(2), AddressRole::Routing), + (g(3), AddressRole::Mail), + ], + }, + Violation::RoutingNotInProxies { node: g(4) }, + Violation::RoutingMismatch { node: g(5) }, + Violation::DanglingMembership { + user: g(6), + group: g(7), + missing: Endpoint::Group, + }, + ] { + round(&v, w_violation, r_violation); + } + } + + #[test] + fn an_observation_round_trips_with_every_field() { + let node = ObservedNode { + kind: NodeKind::User, + active: None, + upn: Some("a@x.test".into()), + primary_smtp: Some("A@x.test".into()), + proxies: vec!["smtp:b@x.test".into(), "X500:/o=ä".into()], + dn: Some(Dn128::new(&[2, 3]).unwrap()), + recipient: Some(ObservedRecipient { + remote_recipient_type: Some(4), + display_type: Some(-1), + type_details: None, + target_address: Some("a@t.mail.onmicrosoft.com".into()), + }), + mail: Some("card@x.test".into()), + alias: Some("a".into()), + exchange_guid: Some(g(0xEE)), + security: Some(true), + }; + round(&node, w_observed, r_observed); + let obs = Observation { + scope: DirectoryScope(g(0x5C)), + nodes: vec![(g(1), node), (g(2), ObservedNode::group())], + members: vec![(g(1), g(2))], + }; + let mut w = W::default(); + w_observation(&mut w, &obs); + let mut r = R::new(&w.0); + let back = r_observation(&mut r).unwrap(); + r.end().unwrap(); + assert_eq!(back.scope, obs.scope); + assert_eq!(back.nodes, obs.nodes); + assert_eq!(back.members, obs.members); + } + + // A remote mailbox the lifecycle cannot produce is refused on read, + // not built (a deprovisioned mailbox with a routing address). + #[test] + fn an_impossible_remote_mailbox_is_refused() { + let mut w = W::default(); + w.u8(2); + w.u8(0); + w.u8(2); + w_archive(&mut w, ArchiveState::None); + w.opt(Some(ValueId(1)), |w, v| w.u32(v.0)); + let err = r_recipient(&mut R::new(&w.0)).unwrap_err(); + assert_eq!(err.what, "remote mailbox"); + } +} diff --git a/crates/lance-graph-dir-sim/tests/persist.rs b/crates/lance-graph-dir-sim/tests/persist.rs index 63139491f..636738959 100644 --- a/crates/lance-graph-dir-sim/tests/persist.rs +++ b/crates/lance-graph-dir-sim/tests/persist.rs @@ -307,3 +307,140 @@ fn a_revert_resolves_without_registering_it() { let back = VersionStore::restore_stored(stored, &all_rules()).unwrap(); assert_eq!(back.diff(vs[0], r), st.diff(vs[0], r)); } + +// ---- The log: verified append-only batches (decision 3(b)). + +use lance_graph_dir_sim::log::{replay, replay_prefix, LogError, LogState, Record, ReplayError}; + +/// Save `st` as one batch on top of `log`, returning the batch's records. +fn save(st: &VersionStore, log: &mut LogState) -> Vec { + let batch = log + .batch(&StoredImage::from(st.image())) + .unwrap() + .expect("something new"); + log.committed(&batch).unwrap(); + batch.records +} + +fn restored(records: Vec) -> VersionStore { + let (image, _) = replay(records).unwrap(); + VersionStore::restore_stored(image, &all_rules()).unwrap() +} + +// Two saves, the second holding only what the first did not: the replayed +// log restores a store that reads the same as the original. +#[test] +fn a_log_of_two_batches_restores_the_same_reads() { + let (mut st, mut vs, _) = history(); + let mut log = LogState::new(); + let mut records = save(&st, &mut log); + let first_len = records.len(); + let r = st.revert(vs[2], vs[0], &ev("UNDO-1")).unwrap(); + st.validate(r).unwrap(); + vs.push(r); + let second = save(&st, &mut log); + assert!( + second.len() < first_len, + "the second batch is only the new part" + ); + assert!(second.iter().all(|x| x.batch == 1)); + records.extend(second); + let back = restored(records); + assert_eq!(reads(&back, &vs), reads(&st, &vs)); + // Nothing new: no batch. + assert!(log.batch(&StoredImage::from(st.image())).unwrap().is_none()); +} + +// The logical clock is the history's own: one batch holds five versions. +#[test] +fn one_batch_carries_many_logical_versions() { + let (st, vs, _) = history(); + let records = save(&st, &mut LogState::new()); + assert!(records.iter().all(|r| r.batch == 0)); + let (image, state) = replay(records).unwrap(); + assert_eq!(image.versions.len(), vs.len()); + assert_eq!(state.next_batch(), 1); +} + +fn two_batches() -> (Vec, Vec) { + let (mut st, vs, _) = history(); + let mut log = LogState::new(); + let a = save(&st, &mut log); + st.revert(vs[2], vs[0], &ev("UNDO-1")).unwrap(); + let b = save(&st, &mut log); + (a, b) +} + +// Each way a save can be torn is refused, naming the batch. +#[test] +fn a_torn_save_is_refused() { + let (a, b) = two_batches(); + let all = || a.iter().chain(&b).cloned().collect::>(); + + let mut lost_tail = all(); + lost_tail.pop(); + assert!(matches!( + replay(lost_tail), + Err(ReplayError::RecordCount { batch: 1, .. }) + )); + + let headless: Vec = all() + .into_iter() + .filter(|r| !(r.batch == 1 && r.seq == 0)) + .collect(); + assert_eq!(replay(headless).unwrap_err(), ReplayError::MissingHeader(1)); + + let mut flipped = all(); + let last = flipped.len() - 1; + *flipped[last].body.last_mut().unwrap() ^= 1; + assert_eq!(replay(flipped).unwrap_err(), ReplayError::Checksum(1)); + + assert_eq!( + replay(b.clone()).unwrap_err(), + ReplayError::BatchGap { + expected: 0, + found: 1 + } + ); +} + +// Reading up to the damage is its own call: it returns the good batches +// and names the bad one; `replay` returns nothing. +#[test] +fn the_prefix_up_to_the_damage_is_explicit() { + let (a, mut b) = two_batches(); + b.pop(); + let (good, _) = replay(a.clone()).unwrap(); + let (image, state, err) = replay_prefix(a.into_iter().chain(b)); + assert!(matches!( + err, + Some(ReplayError::RecordCount { batch: 1, .. }) + )); + assert_eq!(image.versions, good.versions); + assert_eq!(state.next_batch(), 1); +} + +// A batch is committed once, in order; an image that does not extend the +// log is refused. +#[test] +fn the_log_moves_forward_only() { + let (st, _, _) = history(); + let mut log = LogState::new(); + let image = StoredImage::from(st.image()); + let first = log.batch(&image).unwrap().unwrap(); + let again = log.batch(&image).unwrap().unwrap(); + log.committed(&first).unwrap(); + assert_eq!( + log.committed(&again), + Err(LogError::StaleBatch { + expected: 1, + found: 0 + }) + ); + let mut shorter = VersionStore::new(); + shorter.observe("lab", 1, observed()).unwrap(); + assert_eq!( + log.batch(&StoredImage::from(shorter.image())).unwrap_err(), + LogError::NotAContinuation + ); +} From 251ec01460dc4da717da4da1399d52318504497e Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 11 Oct 2026 09:34:25 +0000 Subject: [PATCH 2/3] plan + board: D-DSP-2a log (verified batches), D-DSP-2b Lance I/O queued Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G22yT6htkcdyXsihxxXdrg --- .claude/board/STATUS_BOARD.md | 3 ++- .claude/plans/dir-sim-persist-v1.md | 13 +++++++++++-- 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index 62a3010d2..f3ee70c57 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -43,7 +43,8 @@ Plan: `.claude/plans/dir-sim-persist-v1.md`. |---|---|---|---| | **D-DSP-0** | `StoreImage` + `VersionStore::image` / `restore` (dir-sim, no I/O) | Shipped (#1473) | P-W3-0: round trip answers every read the same; damaged image refused | | **D-DSP-1** | `VersionStore::revert(current, earlier, evidence)` — undo as a version | Shipped (#1473) | the revert's view equals `earlier`'s | -| **D-DSP-2** | Lance writer/reader crate for `StoredImage`: logical versions in verifiable append-only batches (one batch = one Lance commit), torn saves refused | Queued | write → read → restore equals the in-memory image; a truncated or reordered batch is refused | +| **D-DSP-2a** | `log`: verified append-only batches of a `StoredImage` (pure, no I/O); torn saves refused; explicit prefix read | In PR | two-batch restore reads the same; four torn-save shapes refused; five disable runs red | +| **D-DSP-2** | Lance writer/reader crate for `StoredImage`: logical versions in verifiable append-only batches (one batch = one Lance commit), torn saves refused | Queued (2b: Lance I/O over `log`) | write → read → restore equals the in-memory image; a truncated or reordered batch is refused | | **D-DSP-4** | `StoredImage` + `RuleRegistry` (exact `(name, version)`) + `restore_stored` + read-only `History` (decision 2(a)) | In PR | unknown rule refused, never substituted; history still reads; disable runs red | | **D-DSP-3** | `ViewActors` (W-2) + `from_graph` groups (wishlist #19) in dir-sim | Shipped (#1473) | disable runs red | diff --git a/.claude/plans/dir-sim-persist-v1.md b/.claude/plans/dir-sim-persist-v1.md index be8bf7d46..56bdf7249 100644 --- a/.claude/plans/dir-sim-persist-v1.md +++ b/.claude/plans/dir-sim-persist-v1.md @@ -78,8 +78,17 @@ as a simulated version whose origin names the restored version. leaves the log refused until reconciled. In-band idempotency stays OPEN, as in the reference implementation. - Records: `Observation`, `Change`, `Origin` and `Violation` need a - stable encoding; it lives in the Lance crate next to the reader that - must decode it, with a round-trip test per variant. + stable encoding. **Amended 2026-10-11:** it lives in + `lance-graph-dir-sim::log` (pure, no I/O), next to the types, so the + storage crate only maps a `Record` to a row and back. + - [x] **D-DSP-2a** `log`: `LogState::batch` / `committed`, `replay`, + `replay_prefix`; per-variant round trips with every strict prefix + refused; two-batch restore; torn saves (lost tail, lost header, + flipped byte, missing batch) refused. Disable runs red: checksum, + record count, batch gap, the sparse skip, stale commit. + - [ ] **D-DSP-2b** the Lance crate: one batch = one Lance append; the + dataset version checked before and after (stale writer refused, + race detected); read once at restore. Needs disk for lancedb. - [x] **Decision 4(b), 2026-10-11:** group-to-role grants are the authority's versioned policy, not directory state: OGAR `ogar_rbac::group_policy::GroupRolePolicy` (append-only, own From 4a2b86a2e2317b505cf102761c0b5cc10705f285 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 11 Oct 2026 10:37:13 +0000 Subject: [PATCH 3/3] dir-sim log: refuse a rewritten history; replay without per-batch copies LogState::batch only compared counts, so an image with the logged number of values and versions but different content was accepted: the sparse loops skipped the changed prefix and could return None, reporting a save while replay kept the old history. LogState now keeps a digest of the logged values, roots (in version order) and versions, and batch refuses an image whose prefix digests differently (NotAContinuation). LogState's Default holds the digest of the empty history, so the first save is checked like the rest. replay_prefix cloned the accumulated image before every batch, making a restore quadratic in the number of saves. A batch is now applied in place and undone on failure by truncating values, roots and versions to their previous lengths; its tags and verdicts collect apart and merge only when the batch is whole. Tests: a_divergent_history_is_not_a_continuation (changed value, version and root each refused; red before: the changed value saved as nothing new); a_batch_failing_after_it_applied_is_undone (a header claiming one version more fails only after the batch's value and tag were applied; the image and state are those of the batches before it). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G22yT6htkcdyXsihxxXdrg --- crates/lance-graph-dir-sim/src/log.rs | 119 ++++++++++++++++++-- crates/lance-graph-dir-sim/tests/persist.rs | 44 ++++++++ 2 files changed, 153 insertions(+), 10 deletions(-) diff --git a/crates/lance-graph-dir-sim/src/log.rs b/crates/lance-graph-dir-sim/src/log.rs index 0cb7611c8..8bca8c322 100644 --- a/crates/lance-graph-dir-sim/src/log.rs +++ b/crates/lance-graph-dir-sim/src/log.rs @@ -775,11 +775,14 @@ impl Batch { // ---------------------------------------------------------------- writer /// What the log already holds: everything [`LogState::batch`] leaves out. -#[derive(Clone, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq)] pub struct LogState { next_batch: u64, values: usize, versions: u64, + /// A digest of the logged values, roots and versions (`prefix_digest`): + /// what an image must still hold to continue the log. + prefix: u64, tags: BTreeMap, verdicts: BTreeMap>, } @@ -787,8 +790,9 @@ pub struct LogState { /// Why an image cannot be appended to the log. #[derive(Clone, Debug, PartialEq, Eq)] pub enum LogError { - /// The image has fewer values or versions than the log already holds: - /// it is not a continuation of the logged history. + /// The image does not start with the logged history: it has fewer + /// values or versions than the log holds, or the logged values, roots + /// or versions differ in it. NotAContinuation, /// A tag the log holds is gone from the image; tags are never removed. TagRemoved(String), @@ -804,6 +808,21 @@ pub enum LogError { }, } +impl Default for LogState { + /// An empty log. Its prefix is the digest of the empty history, not 0, + /// so the first image is checked like every later one. + fn default() -> Self { + Self { + next_batch: 0, + values: 0, + versions: 0, + prefix: prefix_digest(&StoredImage::default(), 0, 0), + tags: BTreeMap::new(), + verdicts: BTreeMap::new(), + } + } +} + impl LogState { /// An empty log. pub fn new() -> Self { @@ -824,7 +843,10 @@ impl LogState { /// [`LogError`] when `image` does not extend the logged history. pub fn batch(&self, image: &StoredImage) -> Result, LogError> { let versions = image.versions.len() as u64; - if image.values.len() < self.values || versions < self.versions { + if image.values.len() < self.values + || versions < self.versions + || prefix_digest(image, self.values, self.versions) != self.prefix + { return Err(LogError::NotAContinuation); } let tags: BTreeMap = image.tags.iter().cloned().collect(); @@ -909,6 +931,7 @@ impl LogState { next_batch: batch + 1, values: image.values.len(), versions, + prefix: prefix_digest(image, image.values.len(), versions), tags, verdicts, }, @@ -933,6 +956,45 @@ impl LogState { } } +/// FNV-1a over the first `values` values, the roots of versions below +/// `versions` (in version order) and the first `versions` versions, each +/// item length-prefixed. Two images agree on it exactly when they agree on +/// that prefix (up to hash collision), which is what lets +/// [`LogState::batch`] refuse an image that rewrote logged history. +/// +/// It reads the whole prefix, so a save costs time linear in the history; +/// `batch` already walks every root. +fn prefix_digest(image: &StoredImage, values: usize, versions: u64) -> u64 { + let mut h: u64 = 0xcbf2_9ce4_8422_2325; + let mut eat = |w: W| { + let len = u32::try_from(w.0.len()).expect("item under 4 GiB"); + for b in len.to_le_bytes().into_iter().chain(w.0) { + h ^= u64::from(b); + h = h.wrapping_mul(0x0000_0100_0000_01b3); + } + }; + for value in image.values.iter().take(values) { + let mut w = W::default(); + w.str(value); + eat(w); + } + let mut roots: Vec<&(VersionId, Observation)> = + image.roots.iter().filter(|(v, _)| v.0 < versions).collect(); + roots.sort_by_key(|(v, _)| *v); + for (v, obs) in roots { + let mut w = W::default(); + w.u64(v.0); + w_observation(&mut w, obs); + eat(w); + } + for v in image.versions.iter().take(versions as usize) { + let mut w = W::default(); + w_version(&mut w, v); + eat(w); + } + h +} + fn encode_header(h: &Header) -> Vec { let Header { batch, @@ -1034,23 +1096,29 @@ pub fn replay_prefix( let mut err = None; for run in records.chunk_by(|a, b| a.batch == b.batch) { let b = run[0].batch; - let mut next = image.clone(); - let (mut ntags, mut nverdicts) = (tags.clone(), verdicts.clone()); + // A batch is applied to the image in place and undone on failure by + // truncating to these lengths (values, roots and versions only ever + // grow), so replay never copies the accumulated image. Its tags and + // verdicts collect apart and merge only once the batch is whole. + let lens = (image.values.len(), image.roots.len(), image.versions.len()); + let (mut ntags, mut nverdicts) = (BTreeMap::new(), BTreeMap::new()); match apply_batch( b, state.next_batch, run, - &mut next, + &mut image, &mut ntags, &mut nverdicts, ) { Ok(()) => { - image = next; - tags = ntags; - verdicts = nverdicts; + tags.extend(ntags); + verdicts.extend(nverdicts); state.next_batch = b + 1; } Err(e) => { + image.values.truncate(lens.0); + image.roots.truncate(lens.1); + image.versions.truncate(lens.2); err = Some(e); break; } @@ -1060,6 +1128,7 @@ pub fn replay_prefix( image.verdicts = verdicts.clone().into_iter().collect(); state.values = image.values.len(); state.versions = image.versions.len() as u64; + state.prefix = prefix_digest(&image, state.values, state.versions); state.tags = tags; state.verdicts = verdicts; (image, state, err) @@ -1160,6 +1229,36 @@ mod tests { Guid128([n; 16]) } + // A batch that fails after it already changed the image (here: its + // header claims a version it does not carry, found only at the end) + // leaves the image exactly as the batches before it built it. + #[test] + fn a_batch_failing_after_it_applied_is_undone() { + let ab = StoredImage { + values: vec!["a".into(), "b".into()], + ..StoredImage::default() + }; + let mut abc = ab.clone(); + abc.values.push("c".into()); + abc.tags.push(("t".into(), VersionId(0))); + + let mut log = LogState::new(); + let first = log.batch(&ab).unwrap().unwrap(); + log.committed(&first).unwrap(); + let mut second = log.batch(&abc).unwrap().unwrap(); + let mut header = decode_header(&second.records[0].body).unwrap(); + header.versions = 1; + second.records[0].body = encode_header(&header); + + let (image, state, err) = replay_prefix(first.records.into_iter().chain(second.records)); + assert_eq!(err, Some(ReplayError::VersionOrder(1))); + assert_eq!(image.values, ab.values, "the failed batch's value is gone"); + assert!(image.tags.is_empty(), "the failed batch's tag is gone"); + assert_eq!(state.next_batch(), 1); + // The state replay returns continues the log it read. + assert!(state.batch(&ab).unwrap().is_none()); + } + fn round( x: &T, w: impl Fn(&mut W, &T), diff --git a/crates/lance-graph-dir-sim/tests/persist.rs b/crates/lance-graph-dir-sim/tests/persist.rs index 636738959..e2597bd40 100644 --- a/crates/lance-graph-dir-sim/tests/persist.rs +++ b/crates/lance-graph-dir-sim/tests/persist.rs @@ -444,3 +444,47 @@ fn the_log_moves_forward_only() { LogError::NotAContinuation ); } + +// An image with the logged counts but a different history is not a +// continuation: the log must not "save" it by appending nothing (or only +// the tail) while replay keeps returning the old history. +#[test] +fn a_divergent_history_is_not_a_continuation() { + let (st, _, _) = history(); + let mut log = LogState::new(); + save(&st, &mut log); + let logged = StoredImage::from(st.image()); + + let mut value = logged.clone(); + value.values[0].push('x'); + assert_eq!( + log.batch(&value).unwrap_err(), + LogError::NotAContinuation, + "a changed value" + ); + + let mut version = logged.clone(); + version + .versions + .iter_mut() + .find(|v| !v.delta.is_empty()) + .expect("a version with changes") + .delta + .pop(); + assert_eq!( + log.batch(&version).unwrap_err(), + LogError::NotAContinuation, + "a changed version" + ); + + let mut root = logged.clone(); + root.roots[0].1.scope = DirectoryScope(g(0x77)); + assert_eq!( + log.batch(&root).unwrap_err(), + LogError::NotAContinuation, + "a changed root" + ); + + // The unchanged image still has nothing new. + assert!(log.batch(&logged).unwrap().is_none()); +}