diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md index bf5d852c7..c9da6b702 100644 --- a/.claude/board/STATUS_BOARD.md +++ b/.claude/board/STATUS_BOARD.md @@ -44,9 +44,9 @@ 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-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-2a** | `log`: verified append-only batches of a `StoredImage` (pure, no I/O); torn saves refused; explicit prefix read | Shipped (#1482) | 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 | In PR (2b: `lance-graph-dir-sim-lance`, one batch = one Lance append) | two saves restore to the same reads; missing dataset = empty log; stale writer writes nothing (create and append); torn batch refused at restore; out-of-order batch refused | +| **D-DSP-4** | `StoredImage` + `RuleRegistry` (exact `(name, version)`) + `restore_stored` + read-only `History` (decision 2(a)) | Shipped (#1481) | 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 | ## D-LBL — labels of a concept, pulled through the plug (2026-10-11) diff --git a/.claude/plans/dir-sim-persist-v1.md b/.claude/plans/dir-sim-persist-v1.md index 56bdf7249..370c6af34 100644 --- a/.claude/plans/dir-sim-persist-v1.md +++ b/.claude/plans/dir-sim-persist-v1.md @@ -86,9 +86,13 @@ as a simulated version whose origin names the restored version. 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] **D-DSP-2b** the Lance crate `lance-graph-dir-sim-lance` (own + workspace, lance 13 without cloud stores; dir-sim itself stays + free of lance): one batch = one Lance append (`Create`, then + `Append`); the dataset version checked before (stale writer + refused, nothing written) and after (`Race` when the commit is not + the next version); read once at restore. Logical versions stay in + the records; the Lance version is only the commit counter. - [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 diff --git a/.github/workflows/dir-sim-persist.yml b/.github/workflows/dir-sim-persist.yml new file mode 100644 index 000000000..77300aed4 --- /dev/null +++ b/.github/workflows/dir-sim-persist.yml @@ -0,0 +1,71 @@ +name: dir-sim persist +# lance-graph-dir-sim-lance stores the dir-sim history log in Lance. It is +# workspace-excluded (own [workspace] root, like lance-graph-dir-sim, which +# it depends on), so no other job builds it. Without this one a change to +# `lance_graph_dir_sim::log` or a lance bump could break it unnoticed. +# +# lance's build compiles .proto files: `protoc` alone is not enough, the +# well-known includes (libprotobuf-dev) are needed too. +on: + push: + branches: + - main + pull_request: + paths: + - crates/lance-graph-dir-sim/** + - crates/lance-graph-dir-sim-lance/** + - crates/lance-graph-quack/** + - crates/lance-graph-mask-risc/** + - .github/workflows/dir-sim-persist.yml + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +env: + CARGO_TERM_COLOR: always + CARGO_INCREMENTAL: "0" + CARGO_PROFILE_DEV_DEBUG: "0" + CARGO_PROFILE_TEST_DEBUG: "0" + RUST_BACKTRACE: "1" + +jobs: + lance-log: + name: dir-sim log in Lance + runs-on: ubuntu-24.04 + timeout-minutes: 45 + defaults: + run: + working-directory: lance-graph + steps: + - uses: actions/checkout@v4 + with: + path: lance-graph + persist-credentials: false + - name: Checkout AdaWorldAPI/ndarray (sibling dependency) + uses: actions/checkout@v4 + with: + repository: AdaWorldAPI/ndarray + path: ndarray + persist-credentials: false + - name: Checkout AdaWorldAPI/OGAR (sibling dependency) + uses: actions/checkout@v4 + with: + repository: AdaWorldAPI/OGAR + path: OGAR + persist-credentials: false + - name: Install protoc and the protobuf includes (for lance) + run: sudo apt-get update && sudo apt-get install -y protobuf-compiler libprotobuf-dev + - name: Setup rust toolchain (pinned by rust-toolchain.toml) + run: rustup show + - uses: Swatinem/rust-cache@v2 + with: + workspaces: lance-graph/crates/lance-graph-dir-sim-lance + - name: clippy + run: cargo clippy --manifest-path crates/lance-graph-dir-sim-lance/Cargo.toml --all-targets -- -D warnings + - name: test + run: cargo test --manifest-path crates/lance-graph-dir-sim-lance/Cargo.toml diff --git a/crates/lance-graph-dir-sim-lance/Cargo.toml b/crates/lance-graph-dir-sim-lance/Cargo.toml new file mode 100644 index 000000000..f2565ec06 --- /dev/null +++ b/crates/lance-graph-dir-sim-lance/Cargo.toml @@ -0,0 +1,46 @@ +# lance-graph-dir-sim-lance — the dir-sim history log stored in Lance +# (plan dir-sim-persist-v1, D-DSP-2b). EXCLUDED from the lance-graph +# workspace (own [workspace] root), like lance-graph-dir-sim, which it +# depends on: dir-sim stays free of lance, arrow and datafusion. +# +# One log batch is one Lance append. The history's logical versions live +# in the records; the Lance dataset version is only the physical commit +# counter, used to refuse a stale writer and detect a race. +# +# Build/verify: +# cargo test --manifest-path crates/lance-graph-dir-sim-lance/Cargo.toml + +[package] +name = "lance-graph-dir-sim-lance" +version = "0.1.0" +edition = "2021" +publish = false +license = "Apache-2.0" +description = "Lance storage for the dir-sim history log: one verified batch per append, dataset version checked before and after, read once at restore." + +[workspace] + +[dependencies] +lance-graph-dir-sim = { path = "../lance-graph-dir-sim" } +# The lance family moves in lockstep with the lance-graph root (13 / arrow +# 58); no cloud object stores, a local or file URI is enough here. +lance = { version = "13.*", default-features = false } +arrow-array = "58" +arrow-schema = "58" + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt-multi-thread"] } +tempfile = "3" +ogar-dir-core = { git = "https://github.com/AdaWorldAPI/OGAR", branch = "main" } +ogar-dir-sim = { git = "https://github.com/AdaWorldAPI/OGAR", branch = "main" } + +# The same patches as lance-graph-dir-sim: [patch] applies only at the +# workspace root, so a crate that depends on dir-sim repeats them. +[patch."https://github.com/AdaWorldAPI/ndarray"] +ndarray = { path = "../../../ndarray" } + +[patch."https://github.com/AdaWorldAPI/OGAR"] +ogar-dir-core = { path = "../../../OGAR/crates/ogar-dir-core" } +ogar-dir-sim = { path = "../../../OGAR/crates/ogar-dir-sim" } +ogar-ad = { path = "../../../OGAR/crates/ogar-ad" } +ogar-az = { path = "../../../OGAR/crates/ogar-az" } diff --git a/crates/lance-graph-dir-sim-lance/src/lib.rs b/crates/lance-graph-dir-sim-lance/src/lib.rs new file mode 100644 index 000000000..71721de90 --- /dev/null +++ b/crates/lance-graph-dir-sim-lance/src/lib.rs @@ -0,0 +1,294 @@ +//! The dir-sim history log stored in a Lance dataset (plan +//! `dir-sim-persist-v1`, D-DSP-2b). +//! +//! [`lance_graph_dir_sim::log`] decides what a save writes: a batch of +//! records holding what an image has beyond the log, with a checksummed +//! header. This crate only stores those records and reads them back. +//! +//! - **One batch is one Lance append.** Every record of a batch lands in one +//! commit, so a batch is in the dataset whole or not at all. The log state +//! advances ([`LogState::committed`]) only after that commit succeeded. +//! - **Two clocks, never mixed.** The history's logical versions travel in +//! the records. The Lance dataset version is only the physical commit +//! counter, used here to refuse a stale writer and to detect a race; it is +//! never written into, or read as, a logical version. +//! - **Read once at restore.** [`LanceLog::restore`] reads every record in +//! one scan and hands them to [`replay`], which refuses the whole log at +//! the first damaged batch. Nothing is repaired. +//! +//! A writer holds the dataset version it last saw. [`LanceLog::append`] +//! refuses to write when the dataset has moved since +//! ([`StoreError::StaleWriter`]): Lance would rebase an append onto a newer +//! version, which would interleave two writers' batches. A writer that +//! slips in between that check and the commit makes the new version +//! something other than the next one; that is reported as +//! [`StoreError::Race`] after the fact, since the commit has landed. + +use std::sync::Arc; + +use arrow_array::{ + Array, BinaryArray, RecordBatch, RecordBatchIterator, UInt32Array, UInt64Array, UInt8Array, +}; +use arrow_schema::{ArrowError, DataType, Field, Schema}; +use lance::dataset::{WriteMode, WriteParams}; +use lance::Dataset; +use lance_graph_dir_sim::image::StoredImage; +use lance_graph_dir_sim::log::{replay, Batch, LogError, LogState, Record, ReplayError}; + +/// Why a save or a restore failed. +#[derive(Debug)] +pub enum StoreError { + /// Lance refused an operation. + Lance(lance::Error), + /// The stored records do not form a valid log. + Replay(ReplayError), + /// The batch does not follow the log state it was built against. + Log(LogError), + /// The dataset is not at the version this writer last saw (`None`: no + /// dataset). Nothing was written. + StaleWriter { + /// The version this writer last saw. + expected: Option, + /// The version the dataset is at. + found: Option, + }, + /// The commit landed, but not as the next version: another writer + /// committed between this writer's check and its commit. The batch is + /// stored; the log state was not advanced. + Race { + /// The version the commit should have made. + expected: u64, + /// The version it made. + found: u64, + }, + /// A stored column is missing or has another type. + Schema(String), +} + +impl From for StoreError { + fn from(e: lance::Error) -> Self { + StoreError::Lance(e) + } +} + +/// The record columns: `batch`, `seq`, `kind`, `body`. +fn schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("batch", DataType::UInt64, false), + Field::new("seq", DataType::UInt32, false), + Field::new("kind", DataType::UInt8, false), + Field::new("body", DataType::Binary, false), + ])) +} + +fn to_batch(records: &[Record]) -> Result { + RecordBatch::try_new( + schema(), + vec![ + Arc::new(UInt64Array::from_iter_values( + records.iter().map(|r| r.batch), + )), + Arc::new(UInt32Array::from_iter_values(records.iter().map(|r| r.seq))), + Arc::new(UInt8Array::from_iter_values(records.iter().map(|r| r.kind))), + Arc::new(BinaryArray::from_iter_values( + records.iter().map(|r| r.body.as_slice()), + )), + ], + ) +} + +fn column<'a, T: 'static>(b: &'a RecordBatch, name: &str) -> Result<&'a T, StoreError> { + b.column_by_name(name) + .and_then(|c| c.as_any().downcast_ref::()) + .ok_or_else(|| StoreError::Schema(format!("column `{name}`"))) +} + +fn from_batch(b: &RecordBatch) -> Result, StoreError> { + let batch: &UInt64Array = column(b, "batch")?; + let seq: &UInt32Array = column(b, "seq")?; + let kind: &UInt8Array = column(b, "kind")?; + let body: &BinaryArray = column(b, "body")?; + if [ + batch.null_count(), + seq.null_count(), + kind.null_count(), + body.null_count(), + ] + .iter() + .any(|&n| n != 0) + { + return Err(StoreError::Schema("null in a record column".into())); + } + Ok((0..b.num_rows()) + .map(|i| Record { + batch: batch.value(i), + seq: seq.value(i), + kind: kind.value(i), + body: body.value(i).to_vec(), + }) + .collect()) +} + +fn is_not_found(e: &lance::Error) -> bool { + matches!(e, lance::Error::DatasetNotFound { .. }) +} + +/// A writer's handle on the log dataset at `uri`. +#[derive(Debug, Clone)] +pub struct LanceLog { + uri: String, + /// The dataset version this writer last saw; `None` while there is no + /// dataset. + version: Option, +} + +impl LanceLog { + /// A writer for a log that does not exist yet. The first + /// [`append`](Self::append) creates the dataset and refuses if one is + /// already there. + pub fn new(uri: impl Into) -> Self { + Self { + uri: uri.into(), + version: None, + } + } + + /// The dataset version this writer last saw (the physical commit + /// counter, not a logical version). + pub fn version(&self) -> Option { + self.version + } + + /// Read the whole log at `uri` once and replay it: the stored image, + /// the log state to append after it, and a writer at the dataset's + /// current version. A missing dataset is an empty log. + /// + /// # Errors + /// + /// [`StoreError::Replay`] at the first damaged batch; + /// [`StoreError::Lance`] / [`StoreError::Schema`] when the dataset + /// cannot be read as records. + pub async fn restore( + uri: impl Into, + ) -> Result<(StoredImage, LogState, LanceLog), StoreError> { + let uri = uri.into(); + let ds = match Dataset::open(&uri).await { + Ok(ds) => ds, + Err(e) if is_not_found(&e) => { + let (image, state) = replay(Vec::new()).map_err(StoreError::Replay)?; + return Ok((image, state, LanceLog { uri, version: None })); + } + Err(e) => return Err(e.into()), + }; + let all = ds.scan().try_into_batch().await?; + let records = from_batch(&all)?; + let (image, state) = replay(records).map_err(StoreError::Replay)?; + let version = Some(ds.version().version); + Ok((image, state, LanceLog { uri, version })) + } + + /// Store `batch` as one Lance commit, then advance `log` past it. + /// + /// Returns the new dataset version. + /// + /// # Errors + /// + /// - [`StoreError::Log`] if `batch` is not the next batch of `log`; + /// nothing is written. + /// - [`StoreError::StaleWriter`] if the dataset is not at the version + /// this writer last saw; nothing is written. + /// - [`StoreError::Race`] if the commit landed as another version than + /// the next one; the batch is stored, `log` is not advanced. + pub async fn append(&mut self, log: &mut LogState, batch: &Batch) -> Result { + if batch.number() != log.next_batch() { + return Err(StoreError::Log(LogError::StaleBatch { + expected: log.next_batch(), + found: batch.number(), + })); + } + let found = match Dataset::open(&self.uri).await { + Ok(ds) => Some(ds.latest_version_id().await?), + Err(e) if is_not_found(&e) => None, + Err(e) => return Err(e.into()), + }; + if found != self.version { + return Err(StoreError::StaleWriter { + expected: self.version, + found, + }); + } + let rb = to_batch(&batch.records).map_err(|e| StoreError::Schema(e.to_string()))?; + let reader = RecordBatchIterator::new([Ok(rb)], schema()); + let mode = if found.is_some() { + WriteMode::Append + } else { + WriteMode::Create + }; + let params = WriteParams { + mode, + ..WriteParams::default() + }; + let ds = match Dataset::write(reader, self.uri.as_str(), Some(params)).await { + Ok(ds) => ds, + // Another writer created the dataset after the check above. + Err(lance::Error::DatasetAlreadyExists { .. }) => { + return Err(StoreError::StaleWriter { + expected: None, + found: Dataset::open(&self.uri) + .await + .ok() + .map(|d| d.version().version), + }) + } + Err(e) => return Err(e.into()), + }; + let made = ds.version().version; + let next = next_version(found); + if made != next { + return Err(StoreError::Race { + expected: next, + found: made, + }); + } + log.committed(batch).map_err(StoreError::Log)?; + self.version = Some(made); + Ok(made) + } +} + +/// The dataset version a commit on top of `seen` makes: Lance numbers a +/// new dataset 1 and each later commit one higher. +fn next_version(seen: Option) -> u64 { + seen.map_or(1, |v| v + 1) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn records_round_trip_through_a_record_batch() { + let records = vec![ + Record { + batch: 3, + seq: 0, + kind: 9, + body: vec![1, 2, 3], + }, + Record { + batch: 3, + seq: 1, + kind: 2, + body: vec![], + }, + ]; + let rb = to_batch(&records).unwrap(); + assert_eq!(from_batch(&rb).unwrap(), records); + } + + #[test] + fn the_next_version_is_one_more_or_one_for_a_new_dataset() { + assert_eq!(next_version(None), 1); + assert_eq!(next_version(Some(4)), 5); + } +} diff --git a/crates/lance-graph-dir-sim-lance/tests/lance_log.rs b/crates/lance-graph-dir-sim-lance/tests/lance_log.rs new file mode 100644 index 000000000..624481613 --- /dev/null +++ b/crates/lance-graph-dir-sim-lance/tests/lance_log.rs @@ -0,0 +1,175 @@ +//! The log stored in Lance: one batch per append, restored in one read, +//! a stale writer refused, a torn batch refused at restore. + +use lance_graph_dir_sim::image::{RuleRegistry, StoredImage}; +use lance_graph_dir_sim::log::{LogState, ReplayError}; +use lance_graph_dir_sim::{Observation, ObservedNode, VersionStore}; +use lance_graph_dir_sim_lance::{LanceLog, StoreError}; +use ogar_dir_core::{DirectoryScope, Guid128}; +use ogar_dir_sim::VersionId; + +const SCOPE: DirectoryScope = DirectoryScope(Guid128([0x5C; 16])); + +fn g(n: u8) -> Guid128 { + Guid128([n; 16]) +} + +fn observation(users: &[(u8, &str)]) -> Observation { + Observation { + scope: SCOPE, + nodes: users + .iter() + .map(|&(n, a)| (g(n), ObservedNode::user(a, a))) + .collect(), + members: vec![], + } +} + +/// A store with one observation, and the versions it holds. +fn first() -> (VersionStore, Vec) { + let mut st = VersionStore::new(); + let v = st + .observe("lab", 1, observation(&[(0xA1, "alice@x.test")])) + .unwrap(); + (st, vec![v]) +} + +fn second(st: &mut VersionStore, vs: &mut Vec) { + let v = st + .observe( + "lab", + 2, + observation(&[(0xA1, "alice@x.test"), (0xB0, "bob@x.test")]), + ) + .unwrap(); + vs.push(v); +} + +fn image(st: &VersionStore) -> StoredImage { + StoredImage::from(st.image()) +} + +fn uri(dir: &tempfile::TempDir) -> String { + dir.path().join("log.lance").to_str().unwrap().to_string() +} + +// Two saves, each one append; the dataset restores to a store that reads +// the same, and the physical version is the commit count, not a logical +// version (the store holds two logical versions after the second save, +// but one save could carry any number). +#[tokio::test] +async fn two_saves_restore_to_the_same_reads() { + let dir = tempfile::tempdir().unwrap(); + let (mut st, mut vs) = first(); + let mut log = LogState::new(); + let mut w = LanceLog::new(uri(&dir)); + + let b = log.batch(&image(&st)).unwrap().unwrap(); + assert_eq!(w.append(&mut log, &b).await.unwrap(), 1); + second(&mut st, &mut vs); + let b = log.batch(&image(&st)).unwrap().unwrap(); + assert_eq!(w.append(&mut log, &b).await.unwrap(), 2); + + let (stored, state, w2) = LanceLog::restore(uri(&dir)).await.unwrap(); + assert_eq!(w2.version(), Some(2)); + assert_eq!(state, log, "the restored log continues where the writer is"); + let back = VersionStore::restore_stored(stored, &RuleRegistry::new()).unwrap(); + assert_eq!(back.diff(vs[0], vs[1]), st.diff(vs[0], vs[1])); + assert_eq!(back.version(vs[1]), st.version(vs[1])); + // Nothing new: no batch, no append. + assert!(log.batch(&image(&st)).unwrap().is_none()); +} + +// A missing dataset is an empty log. +#[tokio::test] +async fn a_missing_dataset_is_an_empty_log() { + let dir = tempfile::tempdir().unwrap(); + let (stored, state, w) = LanceLog::restore(uri(&dir)).await.unwrap(); + assert!(stored.versions.is_empty()); + assert_eq!(state, LogState::new()); + assert_eq!(w.version(), None); +} + +// A writer whose view of the dataset is out of date writes nothing: not on +// create (the dataset exists) and not on append (it moved on). +#[tokio::test] +async fn a_stale_writer_writes_nothing() { + let dir = tempfile::tempdir().unwrap(); + let (mut st, mut vs) = first(); + let mut log = LogState::new(); + let mut a = LanceLog::new(uri(&dir)); + let b0 = log.batch(&image(&st)).unwrap().unwrap(); + + // A second writer that also believes the log is empty. + let mut late = LanceLog::new(uri(&dir)); + let mut late_log = LogState::new(); + a.append(&mut log, &b0).await.unwrap(); + assert!(matches!( + late.append(&mut late_log, &b0).await, + Err(StoreError::StaleWriter { + expected: None, + found: Some(1) + }) + )); + assert_eq!( + late_log, + LogState::new(), + "the refused writer's log did not move" + ); + + // A writer restored at version 1, overtaken by `a` at version 2. + let (_, mut old_log, mut old) = LanceLog::restore(uri(&dir)).await.unwrap(); + second(&mut st, &mut vs); + let b1 = log.batch(&image(&st)).unwrap().unwrap(); + let old_b1 = old_log.batch(&image(&st)).unwrap().unwrap(); + a.append(&mut log, &b1).await.unwrap(); + assert!(matches!( + old.append(&mut old_log, &old_b1).await, + Err(StoreError::StaleWriter { + expected: Some(1), + found: Some(2) + }) + )); + + // Only `a`'s two batches are stored. + let (stored, _, w) = LanceLog::restore(uri(&dir)).await.unwrap(); + assert_eq!(w.version(), Some(2)); + assert_eq!(stored.versions.len(), 2); +} + +// A batch stored with a record missing is refused at restore, naming the +// batch; nothing is returned. +#[tokio::test] +async fn a_torn_batch_is_refused_at_restore() { + let dir = tempfile::tempdir().unwrap(); + let (st, _) = first(); + let mut log = LogState::new(); + let mut w = LanceLog::new(uri(&dir)); + let mut b = log.batch(&image(&st)).unwrap().unwrap(); + b.records.pop(); + w.append(&mut log, &b).await.unwrap(); + assert!(matches!( + LanceLog::restore(uri(&dir)).await, + Err(StoreError::Replay(ReplayError::RecordCount { + batch: 0, + .. + })) + )); +} + +// A batch built against another log state is refused before anything is +// written. +#[tokio::test] +async fn a_batch_out_of_order_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let (st, _) = first(); + let mut log = LogState::new(); + let mut w = LanceLog::new(uri(&dir)); + let b0 = log.batch(&image(&st)).unwrap().unwrap(); + w.append(&mut log, &b0).await.unwrap(); + assert!(matches!( + w.append(&mut log, &b0).await, + Err(StoreError::Log(_)) + )); + assert_eq!(w.version(), Some(1)); +}