Store: add with_immediate_tx / with_deferred_read transaction combinators

[George-RD]

Parent

#208 (Spec: deepened Store interface)

Blocked by

None.

Deliverables

Add the private Store transaction combinators (expand step — no mass migration yet).

• Store::with_immediate_tx<T>(&self, f: impl FnOnce(&rusqlite::Transaction) -> Result<T, StoreError>) -> Result<T, StoreError>: locks conn, opens TransactionBehavior::Immediate, runs f, commits on Ok, rolls back on Err.
• Store::with_deferred_read<T>(...) for the sole Deferred read path (store/audit_support.rs::verify_and_replay_aggregate), kept explicitly separate from the write combinator.
• Migrate ONE pilot write site to with_immediate_tx to prove the shape (leave the rest for the T6 sweep).
• Tests: commit on Ok; rollback on Err; two racing writers serialize (D-050 write-serialization / TOCTOU closure, cf. D-167/D-169).

Rationale and the full seam are in spec #208 (decision D-001). Files: crates/openspine-kernel/src/store/mod.rs, store/audit_support.rs.

Verification

./scripts/check.sh green (builds openspine-shell bin first, per AGENTS.md). No public API removed yet; conn stays pub(crate) until T6.

🤖 Generated from spec #208 (wayfinder #189, map #182).

[coderabbitai]

Coding Plan

Summary

• Add two pub(crate) combinators to Store in store/mod.rs: with_immediate_tx for writes and a separate with_deferred_read for the single read path.
• Migrate one pilot write site (claim_pending_draft_write) and the sole deferred read path (verify_and_replay_aggregate) to prove the shape. Leave the remaining sites for the T6 sweep.
• Keep conn as pub(crate). Remove no public API, per the ticket.
• Add commit, rollback, and two-writer serialization tests. Then verify with ./scripts/check.sh, which builds the openspine-shell binary first.

Follow-up Notes for Implementing Agent

• The ticket cites decision "D-001" as the Store interface rationale, but the in-repo decision log records D-001 as the "Lyra is a substrate" framing. The load-bearing precedents for the BEGIN IMMEDIATE serialization are D-050 (atomic upsert / TOCTOU closure), D-167, and D-169. Reference D-050 in the serialization test comment.
• store/mod.rs already carries the // openspine:allow-large-module escape, so adding the combinators there does not break the 500-line file-size check.
• Do not migrate additional write sites in this ticket. The ticket scopes exactly one pilot write site plus the one deferred read path.

Design Choices

Design Choice 1: Select the single pilot write site for migration

Options Considered:

1. pending_draft::claim_pending_draft_write — a self-contained check-then-insert transaction that already has an in-file racing-writer test.
2. boot_clock::check_boot_clock — the smallest transaction body, but it has no existing concurrency test.
3. A larger multi-table site (for example worker_dispatch or standing_rules_*) — higher review risk and more branching.

Chosen Option: 1

Rationale: claim_pending_draft_write is small and self-contained. It calls no other transaction helpers. It already has a racing-writer test. The new serialization test can extend this existing test. This makes the migration low-risk and easy to review.

Design Choice 2: Define how `with_deferred_read` finishes the read-only transaction

Options Considered:

1. Run the closure, then drop the transaction (rollback on Drop) on both Ok and Err. This matches current behavior.
2. Commit on Ok to mirror with_immediate_tx exactly.

Chosen Option: 1

Rationale: The read path writes nothing. Dropping the transaction preserves current behavior. This also keeps the read combinator explicitly distinct from the write combinator, as the ticket requests.

💡 User Tips

Regenerate the plan with different choices with @coderabbitai <feedback>.

Implementation Steps

Phase 1: Add the transaction combinators

This phase adds two private combinators to Store. It factors the shared lock-open-run-commit sequence into one place.

Add `with_immediate_tx` and `with_deferred_read` to `Store`

Add both combinators as methods on impl Store.

• Update crates/openspine-kernel/src/store/mod.rs to add Transaction and TransactionBehavior to the existing use rusqlite::{...} line.
• Add Store::with_immediate_tx<T>(&self, f: impl FnOnce(&rusqlite::Transaction) -> Result<T, StoreError>) -> Result<T, StoreError> near open/open_in_memory.
• with_immediate_tx locks self.conn, opens conn.transaction_with_behavior(TransactionBehavior::Immediate), runs f, calls tx.commit() on Ok, and returns the closure error on Err so the transaction rolls back on Drop.
• Add Store::with_deferred_read<T>(&self, f: impl FnOnce(&rusqlite::Transaction) -> Result<T, StoreError>) -> Result<T, StoreError> as a separate method.
• with_deferred_read locks self.conn, opens TransactionBehavior::Deferred, runs f, and drops the transaction without committing on both Ok and Err.
• Keep both methods pub(crate). Do not change the visibility of conn; it stays pub(crate) per the ticket.
• Mirror the existing error style: do not call .unwrap() on self.conn.lock() because parking_lot::Mutex is not poisonable. Propagate rusqlite errors through ? into StoreError::Sqlite.

🤖 Prompt for AI agents

Implement two new private (`pub(crate)`) combinator methods on `Store` in
`crates/openspine-kernel/src/store/mod.rs`, and centralize the shared
lock-open-run-commit logic.

- First, update the existing `use rusqlite::{...}` import line in `store/mod.rs`
to include `Transaction` and `TransactionBehavior`.
- Add `Store::with_immediate_tx<T>(&self, f: impl FnOnce(&rusqlite::Transaction)
-> Result<T, StoreError>) -> Result<T, StoreError>` near
`open`/`open_in_memory`. It must: lock `self.conn`, open a transaction with
`conn.transaction_with_behavior(TransactionBehavior::Immediate)`, run `f` on
that transaction, call `tx.commit()` when `f` returns `Ok`, and return the
closure's error when `f` returns `Err` (letting the transaction roll back via
`Drop`).
- Add a separate method `Store::with_deferred_read<T>(&self, f: impl
FnOnce(&rusqlite::Transaction) -> Result<T, StoreError>) -> Result<T,
StoreError>`. It must: lock `self.conn`, open a transaction with
`TransactionBehavior::Deferred`, run `f`, and drop the transaction without
committing, regardless of whether `f` returns `Ok` or `Err`.
- Keep both methods `pub(crate)`. Do not change the visibility of `conn`, which
stays `pub(crate)`.
- Do not call `.unwrap()` on `self.conn.lock()`, since `parking_lot::Mutex`
cannot be poisoned. Propagate rusqlite errors through `?` so they convert into
`StoreError::Sqlite`, matching existing error-handling style in the file.

Phase 2: Migrate the pilot write site and the deferred read path

This phase proves the combinator shape at two real call sites: one write path and the single deferred read path.

Task 1: Migrate `claim_pending_draft_write` to `with_immediate_tx`

Replace the inline Immediate transaction with the new combinator.

• Update crates/openspine-kernel/src/store/pending_draft.rs.
• Move the current body of claim_pending_draft_write (the SELECT EXISTS check, the early Ok(false) return, and the INSERT) into a with_immediate_tx closure.
• Preserve the existing return contract: Ok(true) when the row is inserted, Ok(false) when a pending row already exists.
• Remove the now-unused direct conn.lock() and transaction_with_behavior calls in this function.
• Adjust the use rusqlite::{...} line in this file if TransactionBehavior is no longer referenced after the migration.

Task 2: Migrate `verify_and_replay_aggregate` to `with_deferred_read`

Replace the inline Deferred transaction with the new read combinator.

• Update crates/openspine-kernel/src/store/audit_support.rs.
• Move the hash-chain verification and the Store::replay_audit_conn call into a with_deferred_read closure that receives the &Transaction.
• Keep the same single-snapshot guarantee: the whole-ledger verification and the aggregate replay must run on the same transaction inside one closure.
• Preserve all current error outcomes: StoreError::LedgerCorrupted on chain or metadata mismatch, and propagated StoreError::Sqlite and replay_audit_conn errors.
• Remove the now-unused direct conn.lock() and transaction_with_behavior(TransactionBehavior::Deferred) calls, and remove TransactionBehavior from this file's imports if it becomes unused.

🤖 Prompt for AI agents

Migrate two existing call sites onto the new combinators from Phase 1: one write
path and the single deferred read path. Do not migrate any other write sites in
this ticket.

For `claim_pending_draft_write` in
`crates/openspine-kernel/src/store/pending_draft.rs`:
- Move the current function body (the `SELECT EXISTS` check, the early
`Ok(false)` return, and the `INSERT`) into a closure passed to
`with_immediate_tx`.
- Preserve the exact return contract: `Ok(true)` when the row is inserted,
`Ok(false)` when a pending row already exists.
- Remove the now-unused direct `conn.lock()` and `transaction_with_behavior`
calls.
- If `TransactionBehavior` becomes unused in this file's imports after the
migration, remove it from the `use rusqlite::{...}` line.

For `verify_and_replay_aggregate` in
`crates/openspine-kernel/src/store/audit_support.rs`:
- Move the hash-chain verification logic and the `Store::replay_audit_conn` call
into a `with_deferred_read` closure that receives the `&Transaction`.
- Ensure the whole-ledger verification and the aggregate replay run on the same
transaction, inside one closure, to keep the single-snapshot guarantee.
- Preserve all current error outcomes exactly: `StoreError::LedgerCorrupted` on
chain or metadata mismatch, and propagated `StoreError::Sqlite` and
`replay_audit_conn` errors.
- Remove the now-unused direct `conn.lock()` and
`transaction_with_behavior(TransactionBehavior::Deferred)` calls, and remove
`TransactionBehavior` from this file's imports if it becomes unused.

Phase 3: Tests

This phase adds isolated behavior tests for the combinators and a real serialization test for the pilot write site.

Task 1: Add commit and rollback unit tests

Add minimal unit tests that use the kv_state table.

• Add tests in crates/openspine-kernel/src/store/tests.rs (wired via the existing #[cfg(test)] mod tests;).
• Add a test that calls with_immediate_tx, inserts a kv_state row in the closure, returns Ok, and asserts the row is present with get_kv.
• Add a test that calls with_immediate_tx, inserts a kv_state row, then returns Err, and asserts the row is absent with get_kv, proving rollback.
• Add a test that calls with_deferred_read, reads a kv_state value inside the closure, and asserts the returned value matches a row set beforehand.
• Use Store::open_in_memory() and follow the assertion style of the existing kv_state_round_trips_and_upserts test.

Task 2: Add a two-writer serialization test for the pilot site

Add a cross-connection racing-writer test that exercises BEGIN IMMEDIATE serialization.

• Update the racing test for claim_pending_draft_write (currently in pending_draft.rs) to follow the nerve_concurrency_tests.rs template.
• Open two independent Store::open(&path) connections against one tempfile::tempdir() path instead of one shared Arc<Store>.
• Set busy_timeout explicitly on both connections' locked conn before the race.
• Use a Barrier::new(2) and one std::thread::spawn per Store, moving each Store into its thread by value.
• Assert that exactly one thread returns Ok(true) and that count_pending_draft_writes() equals 1.
• Handle each thread's result as a Result rather than unconditionally calling .unwrap(), since the loser may block until the winner commits and then return Ok(false).

🤖 Prompt for AI agents

Add tests that verify combinator behavior in isolation and verify real
two-writer serialization for the pilot write site.

For commit and rollback unit tests, add tests in
`crates/openspine-kernel/src/store/tests.rs` (already wired via `#[cfg(test)]
mod tests;`), using the `kv_state` table as a minimal fixture and
`Store::open_in_memory()`:
- Add a test that calls `with_immediate_tx`, inserts a `kv_state` row inside the
closure, returns `Ok`, and then asserts the row is present using `get_kv`.
- Add a test that calls `with_immediate_tx`, inserts a `kv_state` row inside the
closure, but returns `Err`, and then asserts the row is absent using `get_kv`,
to prove rollback.
- Add a test that calls `with_deferred_read`, reads a `kv_state` value inside
the closure, and asserts the returned value matches a row set beforehand.
- Follow the assertion style of the existing `kv_state_round_trips_and_upserts`
test.

For the two-writer serialization test, update the existing racing test for
`claim_pending_draft_write` (currently in `pending_draft.rs`) to follow the
`nerve_concurrency_tests.rs` template:
- Open two independent `Store::open(&path)` connections against one
`tempfile::tempdir()` path, instead of one shared `Arc<Store>`.
- Set `busy_timeout` explicitly on both connections' locked `conn` before
starting the race.
- Use a `Barrier::new(2)` together with one `std::thread::spawn` per `Store`,
moving each `Store` into its thread by value.
- Assert that exactly one thread returns `Ok(true)`, and assert that
`count_pending_draft_writes()` equals `1`.
- Handle each thread's result as a `Result` instead of calling `.unwrap()`
unconditionally, because the losing thread may block until the winner commits
and then return `Ok(false)`.

Research

The kernel store lives in crates/openspine-kernel/src/store/. Store holds conn: Arc<parking_lot::Mutex<rusqlite::Connection>> and is defined in store/mod.rs next to Store::open, Store::open_in_memory, and the StoreError enum. StoreError::Sqlite(#[from] rusqlite::Error) auto-converts rusqlite errors through ?. About 70 write call sites open conn.transaction_with_behavior(TransactionBehavior::Immediate) inline, then call tx.commit(). Exactly one read path uses TransactionBehavior::Deferred: verify_and_replay_aggregate in store/audit_support.rs. No generic transaction combinator exists yet. Concurrency tests use two independent Store::open(&path) connections against one file, busy_timeout, and a Barrier; nerve_concurrency_tests.rs is the established template. The kv_state table plus get_kv/set_kv and the test-only all_kv_for_test provide a minimal fixture for combinator unit tests.

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

Implement two new private (`pub(crate)`) combinator methods on `Store` in
`crates/openspine-kernel/src/store/mod.rs`, and centralize the shared
lock-open-run-commit logic.

- First, update the existing `use rusqlite::{...}` import line in `store/mod.rs`
to include `Transaction` and `TransactionBehavior`.
- Add `Store::with_immediate_tx<T>(&self, f: impl FnOnce(&rusqlite::Transaction)
-> Result<T, StoreError>) -> Result<T, StoreError>` near
`open`/`open_in_memory`. It must: lock `self.conn`, open a transaction with
`conn.transaction_with_behavior(TransactionBehavior::Immediate)`, run `f` on
that transaction, call `tx.commit()` when `f` returns `Ok`, and return the
closure's error when `f` returns `Err` (letting the transaction roll back via
`Drop`).
- Add a separate method `Store::with_deferred_read<T>(&self, f: impl
FnOnce(&rusqlite::Transaction) -> Result<T, StoreError>) -> Result<T,
StoreError>`. It must: lock `self.conn`, open a transaction with
`TransactionBehavior::Deferred`, run `f`, and drop the transaction without
committing, regardless of whether `f` returns `Ok` or `Err`.
- Keep both methods `pub(crate)`. Do not change the visibility of `conn`, which
stays `pub(crate)`.
- Do not call `.unwrap()` on `self.conn.lock()`, since `parking_lot::Mutex`
cannot be poisoned. Propagate rusqlite errors through `?` so they convert into
`StoreError::Sqlite`, matching existing error-handling style in the file.
===============================================================================

Task: 2

Migrate two existing call sites onto the new combinators from Phase 1: one write
path and the single deferred read path. Do not migrate any other write sites in
this ticket.

For `claim_pending_draft_write` in
`crates/openspine-kernel/src/store/pending_draft.rs`:
- Move the current function body (the `SELECT EXISTS` check, the early
`Ok(false)` return, and the `INSERT`) into a closure passed to
`with_immediate_tx`.
- Preserve the exact return contract: `Ok(true)` when the row is inserted,
`Ok(false)` when a pending row already exists.
- Remove the now-unused direct `conn.lock()` and `transaction_with_behavior`
calls.
- If `TransactionBehavior` becomes unused in this file's imports after the
migration, remove it from the `use rusqlite::{...}` line.

For `verify_and_replay_aggregate` in
`crates/openspine-kernel/src/store/audit_support.rs`:
- Move the hash-chain verification logic and the `Store::replay_audit_conn` call
into a `with_deferred_read` closure that receives the `&Transaction`.
- Ensure the whole-ledger verification and the aggregate replay run on the same
transaction, inside one closure, to keep the single-snapshot guarantee.
- Preserve all current error outcomes exactly: `StoreError::LedgerCorrupted` on
chain or metadata mismatch, and propagated `StoreError::Sqlite` and
`replay_audit_conn` errors.
- Remove the now-unused direct `conn.lock()` and
`transaction_with_behavior(TransactionBehavior::Deferred)` calls, and remove
`TransactionBehavior` from this file's imports if it becomes unused.
===============================================================================

Task: 3

Add tests that verify combinator behavior in isolation and verify real
two-writer serialization for the pilot write site.

For commit and rollback unit tests, add tests in
`crates/openspine-kernel/src/store/tests.rs` (already wired via `#[cfg(test)]
mod tests;`), using the `kv_state` table as a minimal fixture and
`Store::open_in_memory()`:
- Add a test that calls `with_immediate_tx`, inserts a `kv_state` row inside the
closure, returns `Ok`, and then asserts the row is present using `get_kv`.
- Add a test that calls `with_immediate_tx`, inserts a `kv_state` row inside the
closure, but returns `Err`, and then asserts the row is absent using `get_kv`,
to prove rollback.
- Add a test that calls `with_deferred_read`, reads a `kv_state` value inside
the closure, and asserts the returned value matches a row set beforehand.
- Follow the assertion style of the existing `kv_state_round_trips_and_upserts`
test.

For the two-writer serialization test, update the existing racing test for
`claim_pending_draft_write` (currently in `pending_draft.rs`) to follow the
`nerve_concurrency_tests.rs` template:
- Open two independent `Store::open(&path)` connections against one
`tempfile::tempdir()` path, instead of one shared `Arc<Store>`.
- Set `busy_timeout` explicitly on both connections' locked `conn` before
starting the race.
- Use a `Barrier::new(2)` together with one `std::thread::spawn` per `Store`,
moving each `Store` into its thread by value.
- Assert that exactly one thread returns `Ok(true)`, and assert that
`count_pending_draft_writes()` equals `1`.
- Handle each thread's result as a `Result` instead of calling `.unwrap()`
unconditionally, because the losing thread may block until the winner commits
and then return `Ok(false)`.

💡 Iterate on the plan with: `@coderabbitai` <feedback>

Example Feedback
- `@coderabbitai` You can skip phase 3. Add a simple unit test case for phase 2.
- `@coderabbitai` For design choice 1 go ahead with option 3 and replan.

---

💬 Have feedback or questions? Drop into our discord!