mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(agent): align SQLite ownership with session workers
This commit is contained in:
Mario Zechner
parent
f98c2850f5
commit
ef11444b6e
15 changed files with 860 additions and 314 deletions
|
|
@ -140,7 +140,7 @@ Internal pi objects are trusted typed values: Session, storage, operation proced
|
|||
|
||||
## 0.9 Implementation status
|
||||
|
||||
WP00–WP06 are complete (Part 8): the operation graph and public lane runtime are implemented. Part 9 states the required conformance matrix; it is not a claim that every listed row already has one dedicated test. Known missing behavior and current contract debt, each labeled again at its section:
|
||||
WP00–WP07 are complete (Part 8): the operation graph, public lane runtime, and SQLite host-ownership alignment are implemented. Part 9 states the required conformance matrix; it is not a claim that every listed row already has one dedicated test. Known missing behavior and current contract debt, each labeled again at its section:
|
||||
|
||||
- **J1 — JSONL snapshot compaction (§1.7):** specified, not implemented; dead bytes are never reclaimed today.
|
||||
- **C1 — raw RemoteSession (§2.8):** the specified remote mutation transport contradicts the shipped process-local product; a decision is required before implementing either direction.
|
||||
|
|
@ -148,7 +148,7 @@ WP00–WP06 are complete (Part 8): the operation graph and public lane runtime a
|
|||
- **T1 — telemetry (§5.8):** span vocabulary declared; production starts only the tool-hook span. RPC ingress has request-ID cancellation but no trace propagation.
|
||||
- **S3 — search (§2.8):** design only; the current `src/search/index.ts` skeleton conflicts with it and has no implementation.
|
||||
- **R11 — schema migrations (Part 7):** mechanism specified; activation-gated; no migration exists or is required.
|
||||
- **WP07 — SQLite host ownership (§1.7, §2.7):** the backend still contains a writer lease contrary to the host-authoritative ownership rule; live worker-owned fork sources also need an explicit independent read-only snapshot path. WP07 removes the lease and fixes repository-local deletion reservation, physical identity, paths, and close draining.
|
||||
- **WP08 — named-branch and streaming forks (§2.7):** actionable handoff, not implemented; the current contract still defaults branch forks to `main`, does not copy application state for tree forks, and materializes source-sized snapshots.
|
||||
- **SQLite branch divergence (§2.6):** the current compaction-bounded algorithm can copy O(history) on an uncompacted branch, contrary to its bounded-prefix goal.
|
||||
- **H1 — contract/test closure:** public `OperationStatus` includes `"running"` but current observations produce only `"open"`/`"aborting"` (§5.4); the pre-rewrite abort contract bound `operation_abort` before resolving/signalling but current code signals first and binds recipients before releasing the line (§4.6); Part 9 remains the required conformance matrix, not a claim that every row has a dedicated test.
|
||||
- **Source declaration corrections:** `CommitResult.stats` and `SessionReader.getStats()` are implemented; the old inline declarations omitted them even though other old sections depended on post-commit totals (§1.4, §2.8). The old execution-block declarations also predate the current source shapes: standalone `streamHarnessAssistant` permits an absent `afterResponse`, while the durable Harness caller always supplies it; tool phases directly carry `AgentHarnessTool`, `toolContext`, and invocation capabilities and create the canonical result message after an immediate raw result (§5.7). These are source-shape corrections, not changes to the durable boundaries.
|
||||
|
|
@ -368,9 +368,9 @@ Until J1 lands, deleted pending payloads, superseded state revisions, superseded
|
|||
|
||||
### SQLite
|
||||
|
||||
Backend: `packages/session-backends/sqlite-node`. **One database file per session is the default; a shared container is supported.** Without `databasePath` the repository creates one `{id}.sqlite` name per Session (WP07 replaces direct interpolation with path-safe encoding for arbitrary explicit ids); with `databasePath` any number of Sessions share one container. Every authoritative and projection row is scoped by `session_id`; shared containers are a supported deployment mode, not an implementation detail to remove. SQLite supplies atomic transactions and coherent WAL snapshots, not Session ownership.
|
||||
Backend: `packages/session-backends/sqlite-node`. **One database file per session is the default; a shared container is supported.** Without `databasePath`, safe alphanumeric/underscore/hyphen ids retain `{id}.sqlite`; every other explicit id uses a `~`-prefixed base64url encoding of its UTF-16 code units, so separators, dots, percent signs, and Unicode cannot escape `directory`. With `databasePath`, any number of Sessions share one container. Metadata reports the canonical physical container path. Every authoritative and projection row is scoped by `session_id`; shared containers are a supported deployment mode, not an implementation detail to remove. SQLite supplies atomic transactions and coherent WAL snapshots, not Session ownership.
|
||||
|
||||
Target `001_initial.sql` after WP07 (storage version 1), all scoped by `session_id`; current source still has the lease table identified below as removal debt:
|
||||
`001_initial.sql` (storage version 1), all scoped by `session_id`:
|
||||
|
||||
```sql
|
||||
entries(id, parent_id, seq, type, custom_type, timestamp, payload) WITHOUT ROWID;
|
||||
|
|
@ -400,9 +400,11 @@ One `commit()` is one SQL transaction: insert entries and ledger rows, replace/d
|
|||
|
||||
**Every transaction that may write must open with `BEGIN IMMEDIATE`.** A deferred `BEGIN` that reads before writing takes a read snapshot and must later upgrade to the write lock; if another writer committed in between, SQLite fails the upgrade — and `busy_timeout` cannot rescue it, because waiting cannot refresh a stale snapshot; the only recovery is rollback and full retry. Every commit reads the session row's `next_seq` before writing, so a read precedes a write in every writing transaction; branch creation (§2.6) also reads the newest compaction before inserting. Coherent read-only snapshot transactions — fork capture (§2.7) — may use a deferred `BEGIN` read transaction; they never upgrade to a write. The old blanket wording covered every transaction and conflicted with its own read-only fork rule; this narrows the rule to the source behavior without weakening any write path.
|
||||
|
||||
**Session ownership is host-authoritative.** Exactly one worker normally owns a writable Session; create/fork administration may own a destination only until it closes that Session and hands metadata to the worker. Memory, JSONL, and SQLite do not detect a second process opening the same Session for writes; bypassing the server/worker lifecycle is a trusted-host defect. A repository still rejects duplicate writable handles it owns in one process. The current SQLite backend's lease table, claims, renewal timer, and commit callback contradict this rule and are removed by WP07 rather than repaired.
|
||||
**Session ownership is host-authoritative.** Exactly one worker normally owns a writable Session; create/fork administration may own a destination only until it closes that Session and hands metadata to the worker. Memory, JSONL, and SQLite do not detect a second process opening the same Session for writes; bypassing the server/worker lifecycle is a trusted-host defect. SQLite has no lease, fence, heartbeat, or replacement ownership primitive. A repository still rejects duplicate writable handles and reserves create/open/fork/delete destinations it owns in one process. The host closes a worker before deletion; shared-container deletion removes only that Session's rows in one `BEGIN IMMEDIATE` transaction, while per-file deletion removes its database and WAL/SHM sidecars.
|
||||
|
||||
Read-only access is different: WAL permits the server's repository to capture a live worker-owned source while the worker continues committing. The source connection uses one deferred read transaction and never upgrades or claims writable authority; it sees every source transaction wholly before or wholly after its snapshot boundary. WAL frames become visible only when the commit record lands, so no fork sees part of a commit.
|
||||
Database access has three explicit modes: intentional create-or-open, no-create read-write, and no-create read-only. Metadata `open` and deletion use no-create read-write access; listing and external fork sources use no-create read-only access, so missing paths never become empty databases. Writable `open`/`delete` metadata must resolve to the repository-affine physical path. A foreign fork source is instead read from its exact physical path and can never alias an active local source with the same Session ID.
|
||||
|
||||
Read-only fork access may overlap the worker: WAL permits the server's repository to capture a live worker-owned source while the worker continues committing. The source connection uses one deferred read transaction and never upgrades or claims writable authority; it validates the Session row and storage version inside that transaction and sees every source transaction wholly before or wholly after its snapshot boundary. WAL frames become visible only when the commit record lands, so no fork sees part of a commit. A same-repository active source instead retains its queued `SqliteStorage.snapshot()` boundary. Repository close seals admission, starts every open Session close, waits for all to settle, and reports one error directly or several in an `AggregateError`.
|
||||
|
||||
Each physical segment of `scanBranch` uses one JOIN (§2.6 combines segment ranges):
|
||||
|
||||
|
|
@ -1275,13 +1277,14 @@ Workflow: keep a future package's row here until actionable; move exact files/te
|
|||
| WP04 | complete | Synchronous `emitBatch` publication; Session owns committed lane publication. | [Mutation publication and event delivery](work-packages/04-mutation-publication.md) |
|
||||
| WP05 | complete | The total direct durable graph, public/replicated lane surfaces, immutable results, atomic boundaries, cancellation reconciliation, lane-safe provider identity. M11 is its only recorded follow-up. | [Direct durable drive](work-packages/05-direct-durable-drive.md) |
|
||||
| WP06 | complete | Separated Session, Branch, AgentLane, AgentHarness; one keyless Session mutation line. | [Session, Branch, Lane separation](work-packages/06-session-branch-lane-separation.md) |
|
||||
| WP07 | planned | Remove SQLite storage-layer ownership; support live read-only source forks; fix deletion reservation, path/source identity, and close draining. | [SQLite host ownership and live forks](work-packages/07-sqlite-host-ownership-live-forks.md) |
|
||||
| WP07 | complete | Removed SQLite storage-layer ownership; added live read-only source forks, no-create opens, deletion reservation, physical/path safety, and all-settled close. | [SQLite host ownership and live forks](work-packages/07-sqlite-host-ownership-live-forks.md) |
|
||||
| WP08 | actionable, not implemented | Replace implicit-main forks with named-branch/tree semantics and bounded-memory backend copies. | [Named-branch and tree forks with streaming copies](work-packages/08-named-branch-streaming-forks.md) |
|
||||
|
||||
WP05 subsumed the former R2–R12 execution rows; their implemented contract is in Parts 0–5 and the completed handoff.
|
||||
|
||||
Future candidates (detail and order in the roadmap): **H1** — resolve the `OperationStatus.running`, abort signal/event-order, and private gate-close typing contracts and audit Part 9 coverage; **C1** — resolve the §2.8 raw-RemoteSession contradiction before implementing either direction; **L1** — repository ownership of open handles and all-settled close across the three backends; **J1** — implement the §1.7 snapshot rewrite, dead-byte triggers, preserved high-water/list sequences, physical reclamation; **M11** — measure Memory/SQLite/JSONL frame cost, then bound assistant-frame writes without weakening unknown-outcome recovery; **R12** — implement `watchSession`; **T1** — reconcile the declared telemetry schema, then implement retained local spans (RPC trace propagation and an exporter are separate follow-ups); **S3** — reconcile the draft search API, then implement the standalone service, repository catch-up utilities, and the reference SQLite FTS5 projection (§2.8); **R11** — chained migrate-on-open under exclusive host ownership with total mappings (Part 7), activated only before the first incompatible stabilized-format change.
|
||||
Future candidates (detail and order in the roadmap): **WP08** — implement the actionable named-branch/tree fork handoff; **H1** — resolve the `OperationStatus.running`, abort signal/event-order, and private gate-close typing contracts and audit Part 9 coverage; **C1** — resolve the §2.8 raw-RemoteSession contradiction before implementing either direction; **L1** — repository ownership of open handles and all-settled close across the three backends; **J1** — implement the §1.7 snapshot rewrite, dead-byte triggers, preserved high-water/list sequences, physical reclamation; **M11** — measure Memory/SQLite/JSONL frame cost, then bound assistant-frame writes without weakening unknown-outcome recovery; **R12** — implement `watchSession`; **T1** — reconcile the declared telemetry schema, then implement retained local spans (RPC trace propagation and an exporter are separate follow-ups); **S3** — reconcile the draft search API, then implement the standalone service, repository catch-up utilities, and the reference SQLite FTS5 projection (§2.8); **R11** — chained migrate-on-open under exclusive host ownership with total mappings (Part 7), activated only before the first incompatible stabilized-format change.
|
||||
|
||||
Client watch/subscription incarnation fencing, SQLite branch/fork/query performance, pending-payload measurement, and optional presentation/plugin capabilities are inventoried in the roadmap; they do not alter the Harness state machine. Protocol, client/server resnapshot, and lane reducer surfaces required by WP05 are already implemented; future protocol work extends them rather than redefining the lane contract.
|
||||
Client watch/subscription incarnation fencing, SQLite branch/query performance, pending-payload measurement, and optional presentation/plugin capabilities are inventoried in the roadmap; they do not alter the Harness state machine. Protocol, client/server resnapshot, and lane reducer surfaces required by WP05 are already implemented; future protocol work extends them rather than redefining the lane contract.
|
||||
|
||||
# Part 9 — Invariants and tests
|
||||
|
||||
|
|
|
|||
|
|
@ -14,23 +14,22 @@ This audit covers the durable AgentHarness and its directly coupled Session back
|
|||
- `packages/coding-agent/src/experimental` and its focused tests where they host that path;
|
||||
- telemetry plumbing in `packages/telemetry`, `packages/ai`, and `packages/agent`.
|
||||
|
||||
The inventory was checked against current source, tests, package READMEs, the complete `harness.md`, WP00–WP06 status, explicit stubs/TODOs/skips, and the shipped package boundaries. Historical handoffs are not treated as backlog when current source and the completed WP05 contract supersede them.
|
||||
The inventory was checked against current source, tests, package READMEs, the complete `harness.md`, completed WP00–WP07 status, the actionable WP08 handoff, explicit stubs/TODOs/skips, and the shipped package boundaries. Historical handoffs are not treated as backlog when current source and the completed WP05 contract supersede them.
|
||||
|
||||
## Executive result
|
||||
|
||||
WP05 is complete through M10. M11 is its only recorded follow-up. The current Harness execution graph has no unfinished runtime path: `watchSession()` is the sole `SliceNotImplemented` Harness method.
|
||||
|
||||
That does **not** mean the surrounding durable system is complete. The audit found:
|
||||
That does **not** mean the surrounding durable system is complete. The remaining audit findings are:
|
||||
|
||||
1. SQLite storage-layer writer ownership that contradicts the server/Session-worker authority model, plus no explicit read-only path proving a server-side fork can snapshot a live worker-owned source;
|
||||
2. normative JSONL snapshot-compaction behavior with no implementation;
|
||||
3. one required Harness method stub (`watchSession`);
|
||||
4. a deliberate removal of raw `RemoteSession` that conflicts with later normative WP06/`harness.md` text;
|
||||
5. a public search type skeleton that conflicts with the newer search design and has no implementation;
|
||||
6. a full telemetry vocabulary whose only production span is the tool-hook span;
|
||||
7. smaller repository, client-watch, query-bound, documentation, and end-to-end-test gaps.
|
||||
1. normative JSONL snapshot-compaction behavior with no implementation;
|
||||
2. one required Harness method stub (`watchSession`);
|
||||
3. a deliberate removal of raw `RemoteSession` that conflicts with later normative WP06/`harness.md` text;
|
||||
4. a public search type skeleton that conflicts with the newer search design and has no implementation;
|
||||
5. a full telemetry vocabulary whose only production span is the tool-hook span;
|
||||
6. smaller repository, client-watch, query-bound, documentation, and end-to-end-test gaps.
|
||||
|
||||
The immediate package is therefore SQLite host-ownership alignment and live-source fork support, not generic storage cleanup. Its handoff is [`work-packages/07-sqlite-host-ownership-live-forks.md`](work-packages/07-sqlite-host-ownership-live-forks.md).
|
||||
WP07 completed SQLite host-ownership alignment and live-source fork support after the audit baseline; its historical handoff is [`work-packages/07-sqlite-host-ownership-live-forks.md`](work-packages/07-sqlite-host-ownership-live-forks.md). WP08 now owns the separate named-branch, tree-state, and bounded-memory fork redesign; its actionable handoff is [`work-packages/08-named-branch-streaming-forks.md`](work-packages/08-named-branch-streaming-forks.md).
|
||||
|
||||
## Required missing functionality and contract contradictions
|
||||
|
||||
|
|
@ -118,27 +117,19 @@ When activated, it must provide ordered transactional migrate-on-open under excl
|
|||
|
||||
## Correctness and data-safety debt
|
||||
|
||||
### SQLite host ownership and live-source forks — immediate
|
||||
### SQLite host ownership and live-source forks — completed by WP07
|
||||
|
||||
The authoritative product rule is in `plugins.md`: exactly one host-assigned process owns writable Session authority; normally it is the Session worker, while the server may temporarily own a newly created or forked destination before closing it and handing it off. Storage backends do not implement writer ownership. The server closes a worker before destructive repository administration.
|
||||
|
||||
Current SQLite source contradicts that rule with a `writer_lease` table, claim/release calls, renewal timer, lease-loss admission path, and a pre-commit renewal callback. This duplicates host ownership without correctly fencing commits. Remove it rather than repair it; a second process directly opening the same Session for writes is an unsupported host defect, as with Memory and JSONL.
|
||||
SQLite now follows that rule: the writer-lease schema/module, claims, renewal timer, lease-loss path, and pre-commit callback are gone, with no replacement lock or ownership primitive. Create/open/fork/delete retain repository-local ID reservation. Metadata open and deletion use a true no-create read-write mode; listing and external fork sources use no-create read-only connections.
|
||||
|
||||
Forking is the supported cross-process overlap: the server repository may snapshot a source while its worker continues writing. Keep the same-repository source queue for its admitted-commit ordering seam, but use an independent no-create, read-only connection and one deferred transaction for a source owned elsewhere. WAL must allow later worker commits while the fork sees one complete before-or-after transaction boundary. WP07 owns the lease removal, live-source fork path, repository-local deletion reservation, physical identity/path safety, and SQLite close draining. It does not add a replacement storage lock, lease, tombstone, or takeover protocol.
|
||||
Same-repository forks retain the source `commitQueue` ordering seam. A source owned elsewhere, including a live worker, is read from its exact canonical container through one independent read-only deferred WAL transaction. Focused per-file and shared-container tests commit a complete later source transaction after the reader establishes its snapshot and before it closes: the first fork excludes the transaction wholly and a later fork includes it wholly.
|
||||
|
||||
### SQLite repository identity and path safety
|
||||
|
||||
- `databasePath` creation makes `options.directory`, not the actual custom path’s parent.
|
||||
- Per-session filenames interpolate arbitrary explicit IDs without encoding; separators can escape `directory`.
|
||||
- Active fork lookup is keyed only by Session ID, so caller-supplied metadata for the same ID at another path can select the wrong open source.
|
||||
- `repo.close()` uses fail-fast `Promise.all`, so it can reject before other Session closes finish draining and release their connections.
|
||||
- Equal-`createdAt` list ordering has no deterministic tie-break.
|
||||
|
||||
Path/identity/close items needed by the host-owned repository and live-fork path belong in WP07. Deterministic listing is a later behavior-preserving cleanup.
|
||||
WP07 also completed canonical `(containerPath, sessionId)` active identity, safe explicit-ID filenames, custom `databasePath` parent creation, Session-scoped shared deletion, WAL/SHM cleanup, and all-settled SQLite repository close. Writable open/delete reject foreign metadata; foreign fork sources are read only from their exact path. Equal-`createdAt` list ordering still has no deterministic tie-break and remains a later behavior-preserving cleanup.
|
||||
|
||||
### Repository close ownership is unspecified
|
||||
|
||||
`JsonlSessionRepo.close()` contains the only active Agent source TODO and closes no open Session handles. Memory and SQLite repositories do close owned resources, but both currently use fail-fast `Promise.all`; `SessionRepo` itself declares no `close()` method and shared conformance does not define repository-to-handle ownership. Resolve ownership and all-settled cleanup in one repository-lifecycle package; WP07 may harden SQLite's backend-local resource cleanup, but do not patch JSONL alone without deciding the common contract.
|
||||
`JsonlSessionRepo.close()` contains the only active Agent source TODO and closes no open Session handles. Memory still uses fail-fast `Promise.all`; SQLite now performs backend-local all-settled cleanup of currently open handles, but an already-admitted create/open/fork can still register a handle after repository close captured that set. `SessionRepo` itself declares no `close()` method and shared conformance does not define repository-to-handle ownership or draining of admitted repository operations. Resolve ownership and common cleanup in one repository-lifecycle package; do not patch one backend further without deciding the common contract.
|
||||
|
||||
### Client watch staleness after disconnect
|
||||
|
||||
|
|
@ -184,9 +175,9 @@ Candidate mechanisms remain design inputs, not approved architecture: coalesced
|
|||
|
||||
`createDivergentBranchForEntry()` copies every row after the newest compaction; with no compaction it copies root-through-parent. A first divergence from a long uncompacted transcript is therefore O(history) writes. This exposes an internal contradiction in `harness.md` §2.6: its opening bounded-prefix promise conflicts with its own compaction-based copy algorithm, which the implementation follows. Change both the specification and segment representation so a divergence can reference a covering segment at the parent boundary. Preserve shared-container support and add large uncompacted divergence plus chain-soundness tests/benchmarks.
|
||||
|
||||
### SQLite fork cost
|
||||
### Fork contract and materialization — WP08 actionable
|
||||
|
||||
Active and closed forks load every scalar value, including application values that are later excluded, and `createForkSnapshot()` repeatedly searches those arrays for branch/config/label state. Select only fork-relevant built-in namespaces, index once, and fetch labels only for copied entries. Measure large application-state forks before/after.
|
||||
All current backends materialize source-sized fork snapshots. Branch scope defaults to `main` without named-Branch ancestry validation; tree scope omits application values/lists; JSONL closed-source fork replay may repair its source. WP08 replaces that contract with required named-branch or tree scope, one closed built-in-state policy, complete current application state for tree forks, and backend-specific bounded-memory copy procedures. It preserves WP07's host ownership, physical identity, same-repository ordering, and independent live-source WAL boundary.
|
||||
|
||||
### SQLite catalog, statements, stats, and reclamation
|
||||
|
||||
|
|
@ -217,7 +208,7 @@ Remaining cleanup:
|
|||
- Consolidate duplicated SQLite branch payload/structure scan plumbing only after correctness tests pin both paths.
|
||||
- Decide whether unused `sessions.metadata` and unmeasured indexes have a future owner before schema removal; do not change schema casually.
|
||||
- Consolidate `startAiSpan()`/`startHarnessSpan()` implementation only if the telemetry surface is retained.
|
||||
- Mark historical durability documents as delivered where they still read as implementation queues. WP00–WP06, `runtime-simplification.md`, `values.md`’s old consumer deferrals, and external-finalization designs are not active runtime backlog.
|
||||
- Mark historical durability documents as delivered where they still read as implementation queues. WP00–WP07, `runtime-simplification.md`, `values.md`’s old consumer deferrals, and external-finalization designs are not active runtime backlog.
|
||||
|
||||
## Optional or deferred product capabilities
|
||||
|
||||
|
|
@ -238,15 +229,15 @@ These are not blockers for the durable Harness:
|
|||
|
||||
The order is by data safety first, then dependencies. Independent tracks may proceed in parallel only when they do not edit the same contracts.
|
||||
|
||||
1. **WP07 — SQLite host ownership and live forks.** Remove storage-layer writer leases; add no-create read-only snapshots of live worker-owned sources; preserve same-repository fork ordering; add repository-local deletion reservation, path/source identity, and close draining.
|
||||
2. **Harness contract/conformance closure.** Resolve `OperationStatus.running`, abort signal/event binding order, gate-close typing, and the Part 9 coverage matrix.
|
||||
3. **Remote Session decision (decision only).** Resolve the false normative boundary early. If process-local wins, repair the docs. If raw RemoteSession wins, later create a dedicated protocol/client/server/worker package; do not fold it into telemetry or R12.
|
||||
4. **Client watch/subscription staleness** and **repository lifecycle contract.** Small independent correctness packages; complete them before expanding server/worker lifecycle semantics. The lifecycle package must also address Memory's fail-fast repository close.
|
||||
5. **JSONL snapshot compaction.** Implement the already-normative physical reclamation path and metrics.
|
||||
6. **M11 durable frame volume.** Memory/SQLite measurement can start earlier; set final JSONL budgets only with compaction measured, and preserve all recovery boundaries.
|
||||
7. **R12 session-wide watch.** Complete the only Harness method stub before building revisioned Transcript/session-wide remote observation.
|
||||
8. **Telemetry, if retained:** reconcile schemas, then local instrumentation, then RPC propagation, then an optional exporter. RPC propagation follows the Remote Session/product-boundary decision.
|
||||
9. **SQLite branch/fork/query performance hardening.** Keep separate from WP07 ownership alignment and require benchmarks.
|
||||
1. **Harness contract/conformance closure.** Resolve `OperationStatus.running`, abort signal/event binding order, gate-close typing, and the Part 9 coverage matrix.
|
||||
2. **Remote Session decision (decision only).** Resolve the false normative boundary early. If process-local wins, repair the docs. If raw RemoteSession wins, later create a dedicated protocol/client/server/worker package; do not fold it into telemetry or R12.
|
||||
3. **Client watch/subscription staleness** and **repository lifecycle contract.** Small independent correctness packages; complete them before expanding server/worker lifecycle semantics. The lifecycle package must also address Memory's fail-fast repository close.
|
||||
4. **JSONL snapshot compaction.** Implement the already-normative physical reclamation path and metrics.
|
||||
5. **M11 durable frame volume.** Memory/SQLite measurement can start earlier; set final JSONL budgets only with compaction measured, and preserve all recovery boundaries.
|
||||
6. **R12 session-wide watch.** Complete the only Harness method stub before building revisioned Transcript/session-wide remote observation.
|
||||
7. **Telemetry, if retained:** reconcile schemas, then local instrumentation, then RPC propagation, then an optional exporter. RPC propagation follows the Remote Session/product-boundary decision.
|
||||
8. **WP08 — named-branch and streaming forks.** Implement the actionable handoff without reopening WP07 ownership or lifecycle decisions.
|
||||
9. **SQLite branch/query performance hardening.** Keep separate from completed WP07 ownership alignment and WP08 fork semantics; require benchmarks.
|
||||
10. **S3 search.** Resolve its API/filter/cursor decisions, then implement catch-up and the standalone FTS projection.
|
||||
11. **R11 migrations.** Activate immediately before the first incompatible stabilized durable schema change, not earlier.
|
||||
|
||||
|
|
@ -260,4 +251,5 @@ This inventory must be updated when any of these facts changes:
|
|||
- telemetry schemas are implemented or removed;
|
||||
- S3’s public API is reconciled;
|
||||
- a durable format change activates R11;
|
||||
- SQLite storage-layer ownership is removed or the host-authority contract changes.
|
||||
- WP08 lands or its fork contract changes;
|
||||
- the host-authority contract changes.
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
# WP07 — SQLite host ownership and live-source forks
|
||||
|
||||
**Status: actionable handoff; not implemented.**
|
||||
**Status: implemented.**
|
||||
|
||||
The delivered backend has no writer lease or replacement ownership primitive. It provides no-create read-write/read-only opens, queued same-repository snapshots, independent read-only WAL snapshots for live external sources, canonical physical identity, path-safe IDs, repository-local deletion reservation, and all-settled close. The tests cover both per-file and shared-container layouts, including a writer commit completed after a read snapshot boundary but before that reader closes.
|
||||
|
||||
This package aligns `packages/session-backends/sqlite-node` with the product ownership model: the server owns Session records and worker lifecycle, and exactly one host-assigned process owns writable Session authority at a time. Normally that process is the Session worker. The server may temporarily own a newly created or forked destination, but it closes that Session before handing its metadata to a worker.
|
||||
|
||||
|
|
|
|||
|
|
@ -24,6 +24,10 @@ await session.close(BACKGROUND_CONTEXT);
|
|||
await repository.close(BACKGROUND_CONTEXT);
|
||||
```
|
||||
|
||||
The default layout currently creates one `{sessionId}.sqlite` file per Session under `directory`; WP07 replaces that direct interpolation with path-safe encoding for arbitrary explicit IDs. Pass `databasePath` to place multiple Sessions in one supported shared container. Open Sessions own separate SQLite connections. A fork of a source open in the same repository queues on that source; WP07 adds an independent read-only WAL snapshot for a source owned by another process, including a live Session worker.
|
||||
The default layout creates one file per Session under `directory`. IDs containing only ASCII letters, digits, `_`, and `-` retain `{sessionId}.sqlite`; every other ID uses a `~`-prefixed base64url encoding of its UTF-16 code units. The durable ID is unchanged, and returned/listed metadata contains the canonical physical path. Pass `databasePath` to place multiple Sessions in one supported shared container; its parent is created when needed.
|
||||
|
||||
The host lifecycle, not this backend, guarantees one writable owner per Session. Directly opening the same Session for writes in another process is unsupported; read-only listing and fork-source access may overlap the worker. The backend provides atomic storage, branch projections, and maintained statistics. WP07 removes the current storage-layer writer lease and adds the read-only live-fork path, repository-local deletion reservation, physical/path safety, and complete close draining. The package does not currently export a search service or FTS index; search is the separate S3 projection.
|
||||
The database factory distinguishes intentional creation, no-create read-write open, and no-create read-only open. Session `open()` and deletion reject metadata outside the configured repository and never create a missing database. Listing is read-only and best-effort. Forking deliberately permits a foreign source metadata path: it reads that exact existing container read-only and never substitutes an active local Session with the same ID.
|
||||
|
||||
The host lifecycle, not this backend, guarantees one writable owner per Session. Directly opening the same Session for writes in another process is unsupported. The repository rejects overlapping local create/open/fork/delete ownership for one ID, but implements no cross-process lease, lock, fence, heartbeat, or takeover. The host must close a worker before deletion.
|
||||
|
||||
A fork of a source open in the same repository queues its snapshot on that source's commit queue. Any other source, including one held open by a live Session worker, uses an independent read-only connection and one deferred WAL transaction; later worker commits may complete while that snapshot remains open. Shared-container deletion removes only the selected Session's rows. Repository close waits for every open Session cleanup attempt before reporting errors. The package does not export a search service or FTS index; search is the separate S3 projection.
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import type { SQLInputValue } from "node:sqlite";
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { sql } from "./sqlite/sql.ts";
|
||||
import type { SqliteDatabase, SqliteDatabaseFactory, SqliteRunResult, SqliteStatement } from "./sqlite/types.ts";
|
||||
|
||||
|
|
@ -107,6 +108,14 @@ export function createNodeSqliteFactory(): SqliteDatabaseFactory {
|
|||
async open(path: string): Promise<SqliteDatabase> {
|
||||
return new NodeSqliteDatabase(new DatabaseSync(path));
|
||||
},
|
||||
async openExisting(path: string): Promise<SqliteDatabase> {
|
||||
const url = pathToFileURL(path);
|
||||
url.searchParams.set("mode", "rw");
|
||||
return new NodeSqliteDatabase(new DatabaseSync(url));
|
||||
},
|
||||
async openReadOnly(path: string): Promise<SqliteDatabase> {
|
||||
return new NodeSqliteDatabase(new DatabaseSync(path, { readOnly: true }));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -119,10 +119,3 @@ CREATE TABLE IF NOT EXISTS branch_meta (
|
|||
) WITHOUT ROWID;
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS ix_bm_tip ON branch_meta(session_id, tip_entry_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS writer_lease (
|
||||
session_id TEXT PRIMARY KEY,
|
||||
owner_id TEXT NOT NULL,
|
||||
fence INTEGER NOT NULL,
|
||||
expires_at_ms INTEGER NOT NULL
|
||||
) WITHOUT ROWID;
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { access, mkdir, open as openFile, readdir, rm } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { mkdir, open as openFile, readdir, realpath, rm } from "node:fs/promises";
|
||||
import { dirname, join } from "node:path";
|
||||
import type { Context, Entry, ForkOptions, SessionCreateOptions, StoredValue } from "@earendil-works/pi-agent-core";
|
||||
import { branchTip, createForkSnapshot, StorageBackedSession } from "@earendil-works/pi-agent-core";
|
||||
import { uuidv7 } from "@earendil-works/pi-ai";
|
||||
|
|
@ -16,13 +16,6 @@ import {
|
|||
type SqliteSessionMetadata,
|
||||
} from "./session/session-row.ts";
|
||||
import { readAllScalarValueRows, setScalarValueRow } from "./session/values.ts";
|
||||
import {
|
||||
claimWriterLease,
|
||||
readWriterLease,
|
||||
releaseWriterLease,
|
||||
renewWriterLease,
|
||||
type WriterLeaseRow,
|
||||
} from "./session/writer-lease.ts";
|
||||
import { SqliteOpenSession } from "./session.ts";
|
||||
import { sql } from "./sql.ts";
|
||||
import { SqliteStorage, type SqliteStorageSnapshot } from "./storage.ts";
|
||||
|
|
@ -31,22 +24,35 @@ import type { SqliteDatabase, SqliteDatabaseFactory } from "./types.ts";
|
|||
export const SQLITE_STORAGE_VERSION = 1;
|
||||
export const SQLITE_SESSION_EXTENSION = ".sqlite";
|
||||
|
||||
const DEFAULT_WRITER_LEASE_MS = 30_000;
|
||||
const WRITER_LEASE_RENEW_INTERVAL_MS = DEFAULT_WRITER_LEASE_MS / 2;
|
||||
const FIRST_AVAILABLE_COMMIT_SEQ = 1;
|
||||
const SAFE_SESSION_FILE_ID = /^[A-Za-z0-9_-]+$/;
|
||||
|
||||
export type SqliteSessionCreateOptions = SessionCreateOptions;
|
||||
|
||||
export interface SqliteSessionRepoOptions {
|
||||
directory: string;
|
||||
/** Optional single container path. Defaults to one `${id}.sqlite` file per session under directory. */
|
||||
/** Optional single container path. Defaults to one encoded `${id}.sqlite` file per session under directory. */
|
||||
databasePath?: string;
|
||||
databaseFactory: SqliteDatabaseFactory;
|
||||
now?: () => number;
|
||||
}
|
||||
|
||||
function sessionFileName(id: string): string {
|
||||
if (SAFE_SESSION_FILE_ID.test(id)) return `${id}${SQLITE_SESSION_EXTENSION}`;
|
||||
const encoded = Buffer.from(id, "utf16le").toString("base64url");
|
||||
return `~${encoded}${SQLITE_SESSION_EXTENSION}`;
|
||||
}
|
||||
|
||||
function sessionPath(directory: string, id: string): string {
|
||||
return join(directory, `${id}${SQLITE_SESSION_EXTENSION}`);
|
||||
return join(directory, sessionFileName(id));
|
||||
}
|
||||
|
||||
function storageIdentity(path: string, sessionId: string): string {
|
||||
return JSON.stringify([path, sessionId]);
|
||||
}
|
||||
|
||||
function isErrorWithCode(error: unknown, code: string): boolean {
|
||||
return typeof error === "object" && error !== null && "code" in error && error.code === code;
|
||||
}
|
||||
|
||||
async function removeSessionFiles(path: string, options: { force: boolean }): Promise<void> {
|
||||
|
|
@ -55,10 +61,14 @@ async function removeSessionFiles(path: string, options: { force: boolean }): Pr
|
|||
await rm(`${path}-shm`, { force: true });
|
||||
}
|
||||
|
||||
function configureConnection(db: SqliteDatabase): void {
|
||||
function configureWritableConnection(db: SqliteDatabase): void {
|
||||
db.exec("PRAGMA journal_mode = WAL; PRAGMA busy_timeout = 5000;");
|
||||
}
|
||||
|
||||
function configureReadOnlyConnection(db: SqliteDatabase): void {
|
||||
db.exec("PRAGMA busy_timeout = 5000;");
|
||||
}
|
||||
|
||||
interface ForkSnapshot {
|
||||
entries: Entry[];
|
||||
scalarValues: StoredValue<unknown>[];
|
||||
|
|
@ -107,14 +117,19 @@ function readForkSourceEntries(
|
|||
return requested === null ? [] : scanBranchEntries(db, sessionId, { start: requested, order: "oldestFirst" });
|
||||
}
|
||||
|
||||
function createSqliteForkSnapshot(sourceDb: SqliteDatabase, sessionId: string, options: ForkOptions): ForkSnapshot {
|
||||
function createSqliteForkSnapshot(
|
||||
sourceDb: SqliteDatabase,
|
||||
source: SqliteSessionMetadata,
|
||||
options: ForkOptions,
|
||||
): ForkSnapshot {
|
||||
sourceDb.exec("BEGIN");
|
||||
let committed = false;
|
||||
try {
|
||||
const scalarValues = readAllScalarValueRows(sourceDb, sessionId);
|
||||
metadataFromSessionRow(source.path, readSessionRow(sourceDb, source.id), SQLITE_STORAGE_VERSION);
|
||||
const scalarValues = readAllScalarValueRows(sourceDb, source.id);
|
||||
const snapshot = buildForkSnapshot(
|
||||
{
|
||||
entries: readForkSourceEntries(sourceDb, sessionId, scalarValues, options),
|
||||
entries: readForkSourceEntries(sourceDb, source.id, scalarValues, options),
|
||||
scalarValues,
|
||||
entriesComplete: options.scope === "tree",
|
||||
},
|
||||
|
|
@ -163,12 +178,11 @@ export class SqliteSessionRepo {
|
|||
this.reserveId(id);
|
||||
const path = this.pathForSession(id);
|
||||
let db: SqliteDatabase | undefined;
|
||||
let lease: WriterLeaseRow | undefined;
|
||||
let reservedFile = false;
|
||||
let initialized = false;
|
||||
let session: SqliteOpenSession | undefined;
|
||||
try {
|
||||
await mkdir(this.directory, { recursive: true });
|
||||
await mkdir(dirname(path), { recursive: true });
|
||||
if (!this.usesSharedDatabase()) {
|
||||
const file = await openFile(path, "wx");
|
||||
await file.close();
|
||||
|
|
@ -176,35 +190,33 @@ export class SqliteSessionRepo {
|
|||
}
|
||||
const activeDb = await this.databaseFactory.open(path);
|
||||
db = activeDb;
|
||||
configureConnection(activeDb);
|
||||
configureWritableConnection(activeDb);
|
||||
await applyInitialSchema(activeDb);
|
||||
const canonicalPath = await realpath(path);
|
||||
const metadata: SqliteSessionMetadata = {
|
||||
id,
|
||||
createdAt,
|
||||
storageVersion: SQLITE_STORAGE_VERSION,
|
||||
...(options.parentSessionId === undefined ? {} : { parentSessionId: options.parentSessionId }),
|
||||
path,
|
||||
path: canonicalPath,
|
||||
};
|
||||
lease = activeDb.transaction(() => {
|
||||
activeDb.transaction(() => {
|
||||
if (hasSessionRow(activeDb, id)) throw new Error(`SQLite session already exists: ${id}`);
|
||||
insertSessionRow(activeDb, metadata, SQLITE_STORAGE_VERSION, FIRST_AVAILABLE_COMMIT_SEQ);
|
||||
return claimWriterLease(activeDb, id, uuidv7(this.now()), this.now(), DEFAULT_WRITER_LEASE_MS);
|
||||
});
|
||||
initialized = true;
|
||||
session = this.openStorageBackedSession(metadata, activeDb, lease);
|
||||
session = this.openStorageBackedSession(metadata, activeDb);
|
||||
return session;
|
||||
} catch (error) {
|
||||
if (reservedFile && !initialized) await removeSessionFiles(path, { force: true });
|
||||
throw error;
|
||||
} finally {
|
||||
if (session === undefined) {
|
||||
const failedDb = db;
|
||||
const failedLease = lease;
|
||||
if (failedDb !== undefined && failedLease !== undefined) {
|
||||
failedDb.transaction(() => releaseWriterLease(failedDb, id, failedLease.owner_id, failedLease.fence));
|
||||
try {
|
||||
db?.close();
|
||||
} finally {
|
||||
this.pendingIds.delete(id);
|
||||
}
|
||||
db?.close();
|
||||
this.pendingIds.delete(id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -213,57 +225,52 @@ export class SqliteSessionRepo {
|
|||
this.assertOpen();
|
||||
this.reserveId(metadata.id);
|
||||
let db: SqliteDatabase | undefined;
|
||||
let lease: WriterLeaseRow | undefined;
|
||||
let session: SqliteOpenSession | undefined;
|
||||
try {
|
||||
await access(metadata.path);
|
||||
const activeDb = await this.databaseFactory.open(metadata.path);
|
||||
const path = await this.repositoryPathForMetadata(metadata);
|
||||
const activeDb = await this.databaseFactory.openExisting(path);
|
||||
db = activeDb;
|
||||
configureConnection(activeDb);
|
||||
const stored = activeDb.transaction(() => {
|
||||
const stored = metadataFromSessionRow(
|
||||
metadata.path,
|
||||
readSessionRow(activeDb, metadata.id),
|
||||
SQLITE_STORAGE_VERSION,
|
||||
);
|
||||
lease = claimWriterLease(activeDb, metadata.id, uuidv7(this.now()), this.now(), DEFAULT_WRITER_LEASE_MS);
|
||||
return stored;
|
||||
});
|
||||
if (lease === undefined) throw new Error("Failed to claim SQLite writer lease");
|
||||
session = this.openStorageBackedSession(stored, activeDb, lease);
|
||||
configureWritableConnection(activeDb);
|
||||
const stored = metadataFromSessionRow(path, readSessionRow(activeDb, metadata.id), SQLITE_STORAGE_VERSION);
|
||||
session = this.openStorageBackedSession(stored, activeDb);
|
||||
return session;
|
||||
} finally {
|
||||
if (session === undefined) {
|
||||
const failedDb = db;
|
||||
const failedLease = lease;
|
||||
if (failedDb !== undefined && failedLease !== undefined) {
|
||||
failedDb.transaction(() =>
|
||||
releaseWriterLease(failedDb, metadata.id, failedLease.owner_id, failedLease.fence),
|
||||
);
|
||||
try {
|
||||
db?.close();
|
||||
} finally {
|
||||
this.pendingIds.delete(metadata.id);
|
||||
}
|
||||
db?.close();
|
||||
this.pendingIds.delete(metadata.id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async list(_options: undefined, _context: Context): Promise<SqliteSessionMetadata[]> {
|
||||
this.assertOpen();
|
||||
await mkdir(this.directory, { recursive: true });
|
||||
const paths = this.usesSharedDatabase()
|
||||
? [this.pathForSession("")]
|
||||
: (await readdir(this.directory))
|
||||
.filter((name) => name.endsWith(SQLITE_SESSION_EXTENSION))
|
||||
.map((name) => join(this.directory, name));
|
||||
let paths: string[];
|
||||
if (this.usesSharedDatabase()) {
|
||||
paths = [this.databasePath!];
|
||||
} else {
|
||||
let names: string[];
|
||||
try {
|
||||
names = await readdir(this.directory);
|
||||
} catch (error) {
|
||||
if (isErrorWithCode(error, "ENOENT")) return [];
|
||||
throw error;
|
||||
}
|
||||
paths = names
|
||||
.filter((name) => name.endsWith(SQLITE_SESSION_EXTENSION))
|
||||
.map((name) => join(this.directory, name));
|
||||
}
|
||||
const sessions: SqliteSessionMetadata[] = [];
|
||||
for (const path of paths) {
|
||||
let db: SqliteDatabase | undefined;
|
||||
try {
|
||||
await access(path);
|
||||
db = await this.databaseFactory.open(path);
|
||||
configureConnection(db);
|
||||
const canonicalPath = await realpath(path);
|
||||
db = await this.databaseFactory.openReadOnly(canonicalPath);
|
||||
configureReadOnlyConnection(db);
|
||||
for (const row of readAllSessionRows(db)) {
|
||||
sessions.push(metadataFromSessionRow(path, row, SQLITE_STORAGE_VERSION));
|
||||
sessions.push(metadataFromSessionRow(canonicalPath, row, SQLITE_STORAGE_VERSION));
|
||||
}
|
||||
} catch {
|
||||
// Discovery is best-effort: corrupt files, incompatible versions, and
|
||||
|
|
@ -277,28 +284,26 @@ export class SqliteSessionRepo {
|
|||
|
||||
async delete(metadata: SqliteSessionMetadata, _context: Context): Promise<void> {
|
||||
this.assertOpen();
|
||||
if (this.pendingIds.has(metadata.id)) throw new Error(`Session is open: ${metadata.id}`);
|
||||
await access(metadata.path);
|
||||
const db = await this.databaseFactory.open(metadata.path);
|
||||
this.reserveId(metadata.id);
|
||||
try {
|
||||
configureConnection(db);
|
||||
const lease = readWriterLease(db, metadata.id);
|
||||
if (lease !== undefined && lease.expires_at_ms > this.now()) {
|
||||
throw new Error(`SQLite session is already claimed by writer ${lease.owner_id}`);
|
||||
}
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
if (this.usesSharedDatabase()) {
|
||||
const deleteDb = await this.databaseFactory.open(metadata.path);
|
||||
const path = await this.repositoryPathForMetadata(metadata);
|
||||
const db = await this.databaseFactory.openExisting(path);
|
||||
try {
|
||||
configureConnection(deleteDb);
|
||||
deleteDb.transaction(() => deleteSessionRows(deleteDb, metadata.id));
|
||||
configureWritableConnection(db);
|
||||
if (this.usesSharedDatabase()) {
|
||||
db.transaction(() => {
|
||||
metadataFromSessionRow(path, readSessionRow(db, metadata.id), SQLITE_STORAGE_VERSION);
|
||||
deleteSessionRows(db, metadata.id);
|
||||
});
|
||||
} else {
|
||||
metadataFromSessionRow(path, readSessionRow(db, metadata.id), SQLITE_STORAGE_VERSION);
|
||||
}
|
||||
} finally {
|
||||
deleteDb.close();
|
||||
db.close();
|
||||
}
|
||||
} else {
|
||||
await removeSessionFiles(metadata.path, { force: false });
|
||||
if (!this.usesSharedDatabase()) await removeSessionFiles(path, { force: false });
|
||||
} finally {
|
||||
this.pendingIds.delete(metadata.id);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -307,17 +312,16 @@ export class SqliteSessionRepo {
|
|||
const createdAt = this.now();
|
||||
const id = options.id ?? uuidv7(createdAt);
|
||||
this.reserveId(id);
|
||||
const sourceStorage = this.openStorages.get(source.id);
|
||||
const sourceStorage = this.openStorages.get(storageIdentity(source.path, source.id));
|
||||
const activeSourceSnapshot = sourceStorage?.snapshot(options, context);
|
||||
void activeSourceSnapshot?.catch(() => undefined);
|
||||
await mkdir(this.directory, { recursive: true });
|
||||
const path = this.pathForSession(id);
|
||||
let db: SqliteDatabase | undefined;
|
||||
let lease: WriterLeaseRow | undefined;
|
||||
let reservedFile = false;
|
||||
let initialized = false;
|
||||
let session: SqliteOpenSession | undefined;
|
||||
try {
|
||||
await mkdir(dirname(path), { recursive: true });
|
||||
if (!this.usesSharedDatabase()) {
|
||||
const file = await openFile(path, "wx");
|
||||
await file.close();
|
||||
|
|
@ -326,21 +330,22 @@ export class SqliteSessionRepo {
|
|||
|
||||
const snapshot =
|
||||
activeSourceSnapshot === undefined
|
||||
? await this.createForkSnapshotFromClosedSource(source, options)
|
||||
? await this.createForkSnapshotFromExternalSource(source, options)
|
||||
: buildForkSnapshot(await activeSourceSnapshot, options);
|
||||
|
||||
const activeDb = await this.databaseFactory.open(path);
|
||||
db = activeDb;
|
||||
configureConnection(activeDb);
|
||||
configureWritableConnection(activeDb);
|
||||
await applyInitialSchema(activeDb);
|
||||
const canonicalPath = await realpath(path);
|
||||
const metadata: SqliteSessionMetadata = {
|
||||
id,
|
||||
createdAt,
|
||||
storageVersion: SQLITE_STORAGE_VERSION,
|
||||
parentSessionId: source.id,
|
||||
path,
|
||||
path: canonicalPath,
|
||||
};
|
||||
lease = activeDb.transaction(() => {
|
||||
activeDb.transaction(() => {
|
||||
if (hasSessionRow(activeDb, id)) throw new Error(`SQLite session already exists: ${id}`);
|
||||
insertSessionRow(activeDb, metadata, SQLITE_STORAGE_VERSION, snapshot.nextSeq);
|
||||
const entryWriter = new EntryRowWriter(activeDb, id);
|
||||
|
|
@ -350,80 +355,83 @@ export class SqliteSessionRepo {
|
|||
}
|
||||
for (const stored of snapshot.scalarValues) insertForkValue(activeDb, id, stored);
|
||||
updateForkSessionStats(activeDb, id, snapshot.messageCount);
|
||||
return claimWriterLease(activeDb, id, uuidv7(this.now()), this.now(), DEFAULT_WRITER_LEASE_MS);
|
||||
});
|
||||
initialized = true;
|
||||
session = this.openStorageBackedSession(metadata, activeDb, lease);
|
||||
session = this.openStorageBackedSession(metadata, activeDb);
|
||||
return session;
|
||||
} catch (error) {
|
||||
if (reservedFile && !initialized) await removeSessionFiles(path, { force: true });
|
||||
throw error;
|
||||
} finally {
|
||||
if (session === undefined) {
|
||||
const failedDb = db;
|
||||
const failedLease = lease;
|
||||
if (failedDb !== undefined && failedLease !== undefined) {
|
||||
failedDb.transaction(() => releaseWriterLease(failedDb, id, failedLease.owner_id, failedLease.fence));
|
||||
try {
|
||||
db?.close();
|
||||
} finally {
|
||||
this.pendingIds.delete(id);
|
||||
}
|
||||
db?.close();
|
||||
this.pendingIds.delete(id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async createForkSnapshotFromClosedSource(
|
||||
source: SqliteSessionMetadata,
|
||||
options: ForkOptions,
|
||||
): Promise<ForkSnapshot> {
|
||||
const sourceDb = await this.databaseFactory.open(source.path);
|
||||
try {
|
||||
configureConnection(sourceDb);
|
||||
metadataFromSessionRow(source.path, readSessionRow(sourceDb, source.id), SQLITE_STORAGE_VERSION);
|
||||
return createSqliteForkSnapshot(sourceDb, source.id, options);
|
||||
} finally {
|
||||
sourceDb.close();
|
||||
}
|
||||
}
|
||||
|
||||
close(context: Context): Promise<void> {
|
||||
if (this.closePromise !== undefined) return this.closePromise;
|
||||
this.closed = true;
|
||||
this.closePromise = Promise.all([...this.openSessions].map((session) => session.close(context))).then(
|
||||
() => undefined,
|
||||
);
|
||||
this.closePromise = this.closeOpenSessions(context);
|
||||
return this.closePromise;
|
||||
}
|
||||
|
||||
private openStorageBackedSession(
|
||||
metadata: SqliteSessionMetadata,
|
||||
db: SqliteDatabase,
|
||||
initialLease: WriterLeaseRow,
|
||||
): SqliteOpenSession {
|
||||
let lease = initialLease;
|
||||
const renew = () => {
|
||||
lease = db.transaction(() =>
|
||||
renewWriterLease(db, metadata.id, lease.owner_id, lease.fence, this.now(), DEFAULT_WRITER_LEASE_MS),
|
||||
);
|
||||
};
|
||||
const storage = new SqliteStorage(db, { sessionId: metadata.id, now: this.now, beforeCommit: renew });
|
||||
this.openStorages.set(metadata.id, storage);
|
||||
private async createForkSnapshotFromExternalSource(
|
||||
source: SqliteSessionMetadata,
|
||||
options: ForkOptions,
|
||||
): Promise<ForkSnapshot> {
|
||||
const path = await realpath(source.path);
|
||||
const sourceDb = await this.databaseFactory.openReadOnly(path);
|
||||
try {
|
||||
configureReadOnlyConnection(sourceDb);
|
||||
return createSqliteForkSnapshot(sourceDb, { ...source, path }, options);
|
||||
} finally {
|
||||
sourceDb.close();
|
||||
}
|
||||
}
|
||||
|
||||
private async closeOpenSessions(context: Context): Promise<void> {
|
||||
const results = await Promise.allSettled([...this.openSessions].map((session) => session.close(context)));
|
||||
const errors = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
|
||||
if (errors.length === 1) throw errors[0];
|
||||
if (errors.length > 1) throw new AggregateError(errors, "Failed to close SQLite Sessions");
|
||||
}
|
||||
|
||||
private openStorageBackedSession(metadata: SqliteSessionMetadata, db: SqliteDatabase): SqliteOpenSession {
|
||||
const key = storageIdentity(metadata.path, metadata.id);
|
||||
const storage = new SqliteStorage(db, { sessionId: metadata.id, now: this.now });
|
||||
this.openStorages.set(key, storage);
|
||||
const session = new StorageBackedSession(metadata, storage);
|
||||
const openSession = new SqliteOpenSession(session, {
|
||||
renewWriterLease: renew,
|
||||
releaseWriterLease: () =>
|
||||
db.transaction(() => releaseWriterLease(db, metadata.id, lease.owner_id, lease.fence)),
|
||||
renewIntervalMs: WRITER_LEASE_RENEW_INTERVAL_MS,
|
||||
onClose: () => {
|
||||
db.close();
|
||||
this.openStorages.delete(metadata.id);
|
||||
this.openSessions.delete(openSession);
|
||||
this.pendingIds.delete(metadata.id);
|
||||
try {
|
||||
db.close();
|
||||
} finally {
|
||||
if (this.openStorages.get(key) === storage) this.openStorages.delete(key);
|
||||
this.openSessions.delete(openSession);
|
||||
this.pendingIds.delete(metadata.id);
|
||||
}
|
||||
},
|
||||
});
|
||||
this.openSessions.add(openSession);
|
||||
return openSession;
|
||||
}
|
||||
|
||||
private async repositoryPathForMetadata(metadata: SqliteSessionMetadata): Promise<string> {
|
||||
const [expected, actual] = await Promise.all([
|
||||
realpath(this.pathForSession(metadata.id)),
|
||||
realpath(metadata.path),
|
||||
]);
|
||||
if (expected !== actual) {
|
||||
throw new Error(`SQLite session metadata path is outside this repository: ${metadata.path}`);
|
||||
}
|
||||
return actual;
|
||||
}
|
||||
|
||||
private reserveId(id: string): void {
|
||||
if (this.pendingIds.has(id)) throw new Error(`Session is already open: ${id}`);
|
||||
this.pendingIds.add(id);
|
||||
|
|
|
|||
|
|
@ -18,9 +18,6 @@ import type { SqliteSessionMetadata } from "./session/session-row.ts";
|
|||
|
||||
export interface SqliteOpenSessionOptions {
|
||||
onClose: () => void;
|
||||
renewWriterLease: () => void;
|
||||
releaseWriterLease: () => void;
|
||||
renewIntervalMs: number;
|
||||
}
|
||||
|
||||
/** SQLite-specific open-session lifecycle wrapper. */
|
||||
|
|
@ -29,11 +26,7 @@ export class SqliteOpenSession implements Session<SqliteSessionMetadata> {
|
|||
readonly idGenerator: Session<SqliteSessionMetadata>["idGenerator"];
|
||||
private readonly session: Session<SqliteSessionMetadata>;
|
||||
private readonly onClose: () => void;
|
||||
private readonly renewWriterLease: () => void;
|
||||
private readonly releaseWriterLease: () => void;
|
||||
private readonly renewalTimer: ReturnType<typeof setInterval>;
|
||||
private readonly admitted = new Set<Promise<unknown>>();
|
||||
private leaseError: unknown;
|
||||
private readonly closedError = new Error("Session is closed");
|
||||
private state: "open" | "closing" | "closed" = "open";
|
||||
private closePromise: Promise<void> | undefined;
|
||||
|
|
@ -43,17 +36,6 @@ export class SqliteOpenSession implements Session<SqliteSessionMetadata> {
|
|||
this.metadata = session.metadata;
|
||||
this.idGenerator = session.idGenerator;
|
||||
this.onClose = options.onClose;
|
||||
this.renewWriterLease = options.renewWriterLease;
|
||||
this.releaseWriterLease = options.releaseWriterLease;
|
||||
this.renewalTimer = setInterval(() => {
|
||||
try {
|
||||
this.renewWriterLease();
|
||||
} catch (error) {
|
||||
this.leaseError = error;
|
||||
clearInterval(this.renewalTimer);
|
||||
}
|
||||
}, options.renewIntervalMs);
|
||||
this.renewalTimer.unref?.();
|
||||
}
|
||||
|
||||
async beginMutation(context: Context): Promise<SessionMutation> {
|
||||
|
|
@ -195,13 +177,8 @@ export class SqliteOpenSession implements Session<SqliteSessionMetadata> {
|
|||
this.closePromise = Promise.allSettled([...this.admitted])
|
||||
.then(() => this.session.close(context))
|
||||
.finally(() => {
|
||||
clearInterval(this.renewalTimer);
|
||||
try {
|
||||
this.releaseWriterLease();
|
||||
} finally {
|
||||
this.state = "closed";
|
||||
this.onClose();
|
||||
}
|
||||
this.state = "closed";
|
||||
this.onClose();
|
||||
});
|
||||
return this.closePromise;
|
||||
}
|
||||
|
|
@ -220,7 +197,6 @@ export class SqliteOpenSession implements Session<SqliteSessionMetadata> {
|
|||
|
||||
private admit<T>(operation: () => Promise<T>): Promise<T> {
|
||||
if (this.state !== "open") return Promise.reject(this.closedError);
|
||||
if (this.leaseError !== undefined) return Promise.reject(this.leaseError);
|
||||
let result: Promise<T>;
|
||||
try {
|
||||
result = operation();
|
||||
|
|
|
|||
|
|
@ -96,7 +96,6 @@ export function deleteSessionRows(db: SqliteDatabase, sessionId: string): void {
|
|||
sql`DELETE FROM usage_ledger WHERE session_id = ${sessionId}`.run(db);
|
||||
sql`DELETE FROM branch_entries WHERE session_id = ${sessionId}`.run(db);
|
||||
sql`DELETE FROM branch_meta WHERE session_id = ${sessionId}`.run(db);
|
||||
sql`DELETE FROM writer_lease WHERE session_id = ${sessionId}`.run(db);
|
||||
const result = sql`DELETE FROM sessions WHERE id = ${sessionId}`.run(db);
|
||||
if (result.changes !== 1)
|
||||
throw new Error(`Expected to delete one SQLite session ${sessionId}, deleted ${result.changes}`);
|
||||
|
|
|
|||
|
|
@ -1,60 +0,0 @@
|
|||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export interface WriterLeaseRow {
|
||||
session_id: string;
|
||||
owner_id: string;
|
||||
fence: number;
|
||||
expires_at_ms: number;
|
||||
}
|
||||
|
||||
export function readWriterLease(db: SqliteDatabase, sessionId: string): WriterLeaseRow | undefined {
|
||||
return sql`SELECT session_id, owner_id, fence, expires_at_ms FROM writer_lease
|
||||
WHERE session_id = ${sessionId}`.get<WriterLeaseRow>(db);
|
||||
}
|
||||
|
||||
export function claimWriterLease(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
ownerId: string,
|
||||
now: number,
|
||||
leaseMs: number,
|
||||
): WriterLeaseRow {
|
||||
const expiresAt = now + leaseMs;
|
||||
const current = readWriterLease(db, sessionId);
|
||||
if (current === undefined) {
|
||||
sql`INSERT INTO writer_lease (session_id, owner_id, fence, expires_at_ms)
|
||||
VALUES (${sessionId}, ${ownerId}, ${1}, ${expiresAt})`.run(db);
|
||||
return { session_id: sessionId, owner_id: ownerId, fence: 1, expires_at_ms: expiresAt };
|
||||
}
|
||||
if (current.owner_id !== ownerId && current.expires_at_ms > now) {
|
||||
throw new Error(`SQLite session is already claimed by writer ${current.owner_id}`);
|
||||
}
|
||||
const fence = current.owner_id === ownerId ? current.fence : current.fence + 1;
|
||||
sql`UPDATE writer_lease SET owner_id = ${ownerId}, fence = ${fence}, expires_at_ms = ${expiresAt}
|
||||
WHERE session_id = ${sessionId}`.run(db);
|
||||
return { session_id: sessionId, owner_id: ownerId, fence, expires_at_ms: expiresAt };
|
||||
}
|
||||
|
||||
export function renewWriterLease(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
ownerId: string,
|
||||
fence: number,
|
||||
now: number,
|
||||
leaseMs: number,
|
||||
): WriterLeaseRow {
|
||||
const expiresAt = now + leaseMs;
|
||||
const result = sql`UPDATE writer_lease SET expires_at_ms = ${expiresAt}
|
||||
WHERE session_id = ${sessionId} AND owner_id = ${ownerId} AND fence = ${fence}`.run(db);
|
||||
if (result.changes !== 1) throw new Error("SQLite writer lease was lost");
|
||||
const row = sql`SELECT session_id, owner_id, fence, expires_at_ms FROM writer_lease
|
||||
WHERE session_id = ${sessionId} AND owner_id = ${ownerId} AND fence = ${fence}`.get<WriterLeaseRow>(db);
|
||||
if (row === undefined) throw new Error("SQLite writer lease was lost");
|
||||
return row;
|
||||
}
|
||||
|
||||
export function releaseWriterLease(db: SqliteDatabase, sessionId: string, ownerId: string, fence: number): void {
|
||||
sql`DELETE FROM writer_lease
|
||||
WHERE session_id = ${sessionId} AND owner_id = ${ownerId} AND fence = ${fence}`.run(db);
|
||||
}
|
||||
|
|
@ -38,7 +38,6 @@ import type { SqliteDatabase } from "./types.ts";
|
|||
export interface SqliteStorageOptions {
|
||||
sessionId: string;
|
||||
now?: () => number;
|
||||
beforeCommit?: () => void;
|
||||
}
|
||||
|
||||
export interface SqliteStorageSnapshot {
|
||||
|
|
@ -51,7 +50,6 @@ export class SqliteStorage implements Storage {
|
|||
private readonly db: SqliteDatabase;
|
||||
private readonly sessionId: string;
|
||||
private readonly now: () => number;
|
||||
private readonly beforeCommit: () => void;
|
||||
private readonly entryWriter: EntryRowWriter;
|
||||
private readonly usageWriter: UsageLedgerRowWriter;
|
||||
private commitQueue: Promise<void> = Promise.resolve();
|
||||
|
|
@ -62,7 +60,6 @@ export class SqliteStorage implements Storage {
|
|||
this.db = db;
|
||||
this.sessionId = options.sessionId;
|
||||
this.now = options.now ?? Date.now;
|
||||
this.beforeCommit = options.beforeCommit ?? (() => undefined);
|
||||
this.entryWriter = new EntryRowWriter(db, this.sessionId);
|
||||
this.usageWriter = new UsageLedgerRowWriter(db, this.sessionId);
|
||||
}
|
||||
|
|
@ -166,7 +163,6 @@ export class SqliteStorage implements Storage {
|
|||
}
|
||||
|
||||
private applyCommit(writes: Write[]): CommitResult {
|
||||
this.beforeCommit();
|
||||
return this.db.transaction(() => {
|
||||
const firstSeq = readNextSeq(this.db, this.sessionId);
|
||||
const prepared = prepareStorageCommit(writes, firstSeq, this.now());
|
||||
|
|
|
|||
|
|
@ -22,5 +22,10 @@ export interface SqliteDatabase {
|
|||
}
|
||||
|
||||
export interface SqliteDatabaseFactory {
|
||||
/** Open a writable database, creating it when absent. */
|
||||
open(path: string): Promise<SqliteDatabase>;
|
||||
/** Open a writable database without creating it. */
|
||||
openExisting(path: string): Promise<SqliteDatabase>;
|
||||
/** Open an existing database read-only. */
|
||||
openReadOnly(path: string): Promise<SqliteDatabase>;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,52 @@
|
|||
import { access, mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { createNodeSqliteFactory } from "../src/index.ts";
|
||||
|
||||
async function withTempDir<T>(run: (directory: string) => Promise<T>): Promise<T> {
|
||||
const directory = await mkdtemp(join(tmpdir(), "pi-sqlite-adapter-"));
|
||||
try {
|
||||
return await run(directory);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
describe("node:sqlite adapter", () => {
|
||||
it("does not create files for existing or read-only opens", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const path = join(directory, "missing % #.sqlite");
|
||||
const factory = createNodeSqliteFactory();
|
||||
|
||||
await expect(factory.openExisting(path)).rejects.toThrow();
|
||||
await expect(access(path)).rejects.toThrow();
|
||||
await expect(factory.openReadOnly(path)).rejects.toThrow();
|
||||
await expect(access(path)).rejects.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
it("opens an existing database read-write or read-only without changing its mode", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const path = join(directory, "existing % #.sqlite");
|
||||
const factory = createNodeSqliteFactory();
|
||||
const created = await factory.open(path);
|
||||
created.exec("CREATE TABLE values_table (value INTEGER NOT NULL)");
|
||||
created.close();
|
||||
|
||||
const writable = await factory.openExisting(path);
|
||||
writable.exec("INSERT INTO values_table (value) VALUES (1)");
|
||||
writable.close();
|
||||
const readOnly = await factory.openReadOnly(path);
|
||||
try {
|
||||
expect(readOnly.prepare("SELECT value FROM values_table").all()).toEqual([{ value: 1 }]);
|
||||
expect(() => readOnly.exec("INSERT INTO values_table (value) VALUES (2)")).toThrow();
|
||||
} finally {
|
||||
readOnly.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("commits a synchronous transaction and returns its result", async () => {
|
||||
const db = await createNodeSqliteFactory().open(":memory:");
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -1,11 +1,11 @@
|
|||
import { access, mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { access, mkdtemp, realpath, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { dirname, isAbsolute, join, relative } from "node:path";
|
||||
import * as storedValues from "@earendil-works/pi-agent-core";
|
||||
import * as sessionWrites from "@earendil-works/pi-agent-core";
|
||||
import { BACKGROUND_CONTEXT } from "@earendil-works/pi-agent-core";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { SqliteDatabase } from "../src/index.ts";
|
||||
import type { SqliteDatabase, SqliteDatabaseFactory, SqliteStatement } from "../src/index.ts";
|
||||
import { createNodeSqliteFactory, SqliteSessionRepo, sql } from "../src/index.ts";
|
||||
|
||||
async function withTempDir<T>(run: (directory: string) => Promise<T>): Promise<T> {
|
||||
|
|
@ -26,6 +26,202 @@ async function withDb<T>(path: string, run: (db: SqliteDatabase) => Promise<T> |
|
|||
}
|
||||
}
|
||||
|
||||
async function pathExists(path: string): Promise<boolean> {
|
||||
return access(path).then(
|
||||
() => true,
|
||||
() => false,
|
||||
);
|
||||
}
|
||||
|
||||
class ForwardingDatabase implements SqliteDatabase {
|
||||
readonly source: SqliteDatabase;
|
||||
|
||||
constructor(source: SqliteDatabase) {
|
||||
this.source = source;
|
||||
}
|
||||
|
||||
exec(query: string): void {
|
||||
this.source.exec(query);
|
||||
}
|
||||
|
||||
prepare(query: string): SqliteStatement {
|
||||
return this.source.prepare(query);
|
||||
}
|
||||
|
||||
transaction<T>(callback: () => T): T {
|
||||
return this.source.transaction(callback);
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.source.close();
|
||||
}
|
||||
}
|
||||
|
||||
class SnapshotBoundaryStatement implements SqliteStatement {
|
||||
readonly source: SqliteStatement;
|
||||
readonly afterGet: () => void;
|
||||
private called = false;
|
||||
|
||||
constructor(source: SqliteStatement, afterGet: () => void) {
|
||||
this.source = source;
|
||||
this.afterGet = afterGet;
|
||||
}
|
||||
|
||||
run(...params: unknown[]) {
|
||||
return this.source.run(...params);
|
||||
}
|
||||
|
||||
get<TRow extends object>(...params: unknown[]): TRow | undefined {
|
||||
const row = this.source.get<TRow>(...params);
|
||||
if (!this.called) {
|
||||
this.called = true;
|
||||
this.afterGet();
|
||||
}
|
||||
return row;
|
||||
}
|
||||
|
||||
all<TRow extends object>(...params: unknown[]): TRow[] {
|
||||
return this.source.all<TRow>(...params);
|
||||
}
|
||||
|
||||
iterate<TRow extends object>(...params: unknown[]): Iterable<TRow> {
|
||||
return this.source.iterate<TRow>(...params);
|
||||
}
|
||||
}
|
||||
|
||||
class SnapshotBoundaryDatabase extends ForwardingDatabase {
|
||||
readonly afterSnapshotEstablished: () => void;
|
||||
|
||||
constructor(source: SqliteDatabase, afterSnapshotEstablished: () => void) {
|
||||
super(source);
|
||||
this.afterSnapshotEstablished = afterSnapshotEstablished;
|
||||
}
|
||||
|
||||
override prepare(query: string): SqliteStatement {
|
||||
const statement = this.source.prepare(query);
|
||||
return query.includes("FROM sessions")
|
||||
? new SnapshotBoundaryStatement(statement, this.afterSnapshotEstablished)
|
||||
: statement;
|
||||
}
|
||||
}
|
||||
|
||||
class SnapshotBoundaryFactory implements SqliteDatabaseFactory {
|
||||
readonly source = createNodeSqliteFactory();
|
||||
readonly afterSnapshotEstablished: () => void;
|
||||
readOnlyOpenCount = 0;
|
||||
|
||||
constructor(afterSnapshotEstablished: () => void) {
|
||||
this.afterSnapshotEstablished = afterSnapshotEstablished;
|
||||
}
|
||||
|
||||
open(path: string): Promise<SqliteDatabase> {
|
||||
return this.source.open(path);
|
||||
}
|
||||
|
||||
openExisting(path: string): Promise<SqliteDatabase> {
|
||||
return this.source.openExisting(path);
|
||||
}
|
||||
|
||||
async openReadOnly(path: string): Promise<SqliteDatabase> {
|
||||
this.readOnlyOpenCount++;
|
||||
return new SnapshotBoundaryDatabase(await this.source.openReadOnly(path), this.afterSnapshotEstablished);
|
||||
}
|
||||
}
|
||||
|
||||
function deferred(): { promise: Promise<void>; resolve: () => void } {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((resolvePromise) => {
|
||||
resolve = resolvePromise;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
class GatedOpenExistingFactory implements SqliteDatabaseFactory {
|
||||
readonly source = createNodeSqliteFactory();
|
||||
readonly entered = deferred();
|
||||
readonly release = deferred();
|
||||
private gated = false;
|
||||
|
||||
arm(): void {
|
||||
this.gated = true;
|
||||
}
|
||||
|
||||
open(path: string): Promise<SqliteDatabase> {
|
||||
return this.source.open(path);
|
||||
}
|
||||
|
||||
async openExisting(path: string): Promise<SqliteDatabase> {
|
||||
if (this.gated) {
|
||||
this.gated = false;
|
||||
this.entered.resolve();
|
||||
await this.release.promise;
|
||||
}
|
||||
return this.source.openExisting(path);
|
||||
}
|
||||
|
||||
openReadOnly(path: string): Promise<SqliteDatabase> {
|
||||
return this.source.openReadOnly(path);
|
||||
}
|
||||
}
|
||||
|
||||
class CloseTrackingDatabase extends ForwardingDatabase {
|
||||
closeAttempts = 0;
|
||||
closeError: Error | undefined;
|
||||
|
||||
override close(): void {
|
||||
this.closeAttempts++;
|
||||
this.source.close();
|
||||
if (this.closeError !== undefined) throw this.closeError;
|
||||
}
|
||||
}
|
||||
|
||||
class CloseTrackingFactory implements SqliteDatabaseFactory {
|
||||
readonly source = createNodeSqliteFactory();
|
||||
readonly writableConnections: CloseTrackingDatabase[] = [];
|
||||
|
||||
async open(path: string): Promise<SqliteDatabase> {
|
||||
const db = new CloseTrackingDatabase(await this.source.open(path));
|
||||
this.writableConnections.push(db);
|
||||
return db;
|
||||
}
|
||||
|
||||
async openExisting(path: string): Promise<SqliteDatabase> {
|
||||
const db = new CloseTrackingDatabase(await this.source.openExisting(path));
|
||||
this.writableConnections.push(db);
|
||||
return db;
|
||||
}
|
||||
|
||||
openReadOnly(path: string): Promise<SqliteDatabase> {
|
||||
return this.source.openReadOnly(path);
|
||||
}
|
||||
}
|
||||
|
||||
function commitLaterSourceState(db: SqliteDatabase): void {
|
||||
const tip = storedValues.branchTip("main");
|
||||
const name = storedValues.sessionName;
|
||||
const label = storedValues.entryLabel("root");
|
||||
db.transaction(() => {
|
||||
sql`INSERT INTO entries (session_id, id, parent_id, seq, type, custom_type, timestamp, payload)
|
||||
VALUES (${"source"}, ${"child"}, ${"root"}, ${5}, ${"message"}, ${null}, ${2}, ${JSON.stringify({ message: { role: "user", content: "after", timestamp: 2 } })})`.run(
|
||||
db,
|
||||
);
|
||||
sql`INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type)
|
||||
VALUES (${"source"}, ${"root"}, ${"child"}, ${5}, ${"message"})`.run(db);
|
||||
sql`UPDATE branch_meta SET tip_entry_id = ${"child"}, tip_seq = ${5}
|
||||
WHERE session_id = ${"source"} AND branch_id = ${"root"}`.run(db);
|
||||
for (const [namespace, key, seq, value] of [
|
||||
[tip.namespace, tip.key, 6, "child"],
|
||||
[name.namespace, name.key, 7, "after"],
|
||||
[label.namespace, label.key, 8, "after-label"],
|
||||
] as const) {
|
||||
sql`INSERT INTO scalar_values (session_id, namespace, key, seq, value)
|
||||
VALUES (${"source"}, ${namespace}, ${key}, ${seq}, ${JSON.stringify(value)})
|
||||
ON CONFLICT(session_id, namespace, key) DO UPDATE SET seq = excluded.seq, value = excluded.value`.run(db);
|
||||
}
|
||||
sql`UPDATE sessions SET message_count = ${1}, next_seq = ${9} WHERE id = ${"source"}`.run(db);
|
||||
});
|
||||
}
|
||||
|
||||
describe("SqliteSessionRepo", () => {
|
||||
it("creates one branchless initialized session file", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
|
|
@ -42,7 +238,7 @@ describe("SqliteSessionRepo", () => {
|
|||
createdAt: 1_700_000_000_000,
|
||||
storageVersion: 1,
|
||||
});
|
||||
expect(metadata.path).toBe(join(directory, "session.sqlite"));
|
||||
expect(metadata.path).toBe(await realpath(join(directory, "session.sqlite")));
|
||||
|
||||
await withDb(metadata.path, (db) => {
|
||||
expect(sql`SELECT COUNT(*) AS count FROM sessions`.get<{ count: number }>(db)).toEqual({ count: 1 });
|
||||
|
|
@ -144,7 +340,7 @@ describe("SqliteSessionRepo", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("lists sessions without changing the writer lease", async () => {
|
||||
it("lists an open session without storage-layer ownership state", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory,
|
||||
|
|
@ -153,23 +349,48 @@ describe("SqliteSessionRepo", () => {
|
|||
});
|
||||
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
||||
const { metadata } = session;
|
||||
let leaseBeforeList: unknown[] = [];
|
||||
await withDb(metadata.path, (db) => {
|
||||
leaseBeforeList = sql`SELECT owner_id, fence FROM writer_lease WHERE session_id = ${"session"}`.all(db);
|
||||
});
|
||||
|
||||
await expect(repo.list(undefined, BACKGROUND_CONTEXT)).resolves.toMatchObject([
|
||||
{ id: "session", path: metadata.path },
|
||||
]);
|
||||
await withDb(metadata.path, (db) => {
|
||||
expect(sql`SELECT owner_id, fence FROM writer_lease WHERE session_id = ${"session"}`.all(db)).toEqual(
|
||||
leaseBeforeList,
|
||||
);
|
||||
expect(
|
||||
sql`SELECT name FROM sqlite_master WHERE type = ${"table"} AND name = ${"writer_lease"}`.get(db),
|
||||
).toBeUndefined();
|
||||
});
|
||||
await session.close(BACKGROUND_CONTEXT);
|
||||
});
|
||||
});
|
||||
|
||||
it("ignores a stale writer_lease table from a pre-WP07 WIP database", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory,
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
const created = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
||||
await created.close(BACKGROUND_CONTEXT);
|
||||
await withDb(created.metadata.path, (db) => {
|
||||
db.exec(
|
||||
"CREATE TABLE writer_lease (session_id TEXT PRIMARY KEY, owner_id TEXT NOT NULL, fence INTEGER NOT NULL, expires_at_ms INTEGER NOT NULL) WITHOUT ROWID",
|
||||
);
|
||||
sql`INSERT INTO writer_lease (session_id, owner_id, fence, expires_at_ms)
|
||||
VALUES (${"session"}, ${"stale"}, ${7}, ${999})`.run(db);
|
||||
});
|
||||
|
||||
const reopened = await repo.open(created.metadata, BACKGROUND_CONTEXT);
|
||||
await reopened.setName("works", BACKGROUND_CONTEXT);
|
||||
await reopened.close(BACKGROUND_CONTEXT);
|
||||
await withDb(created.metadata.path, (db) => {
|
||||
expect(sql`SELECT owner_id, fence FROM writer_lease WHERE session_id = ${"session"}`.get(db)).toEqual({
|
||||
owner_id: "stale",
|
||||
fence: 7,
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
it("skips corrupt and incompatible files during list discovery", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const repo = new SqliteSessionRepo({
|
||||
|
|
@ -203,7 +424,7 @@ describe("SqliteSessionRepo", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("rejects delete for missing files and live external writer leases", async () => {
|
||||
it("rejects delete for missing files and deletes a closed session", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory,
|
||||
|
|
@ -214,16 +435,6 @@ describe("SqliteSessionRepo", () => {
|
|||
const { metadata } = session;
|
||||
await session.close(BACKGROUND_CONTEXT);
|
||||
|
||||
await withDb(metadata.path, (db) => {
|
||||
sql`INSERT INTO writer_lease (session_id, owner_id, fence, expires_at_ms) VALUES (${"session"}, ${"external"}, ${1}, ${1_000})`.run(
|
||||
db,
|
||||
);
|
||||
});
|
||||
await expect(repo.delete(metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already claimed");
|
||||
await withDb(metadata.path, (db) => {
|
||||
sql`DELETE FROM writer_lease WHERE session_id = ${"session"}`.run(db);
|
||||
});
|
||||
|
||||
await expect(
|
||||
repo.delete({ ...metadata, path: join(directory, "missing.sqlite") }, BACKGROUND_CONTEXT),
|
||||
).rejects.toThrow();
|
||||
|
|
@ -250,7 +461,7 @@ describe("SqliteSessionRepo", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("opens a session through the version gate and rejects a live external writer lease", async () => {
|
||||
it("opens a session through the version gate and rejects a duplicate local handle", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory,
|
||||
|
|
@ -263,14 +474,8 @@ describe("SqliteSessionRepo", () => {
|
|||
|
||||
const opened = await repo.open(metadata, BACKGROUND_CONTEXT);
|
||||
expect(opened.metadata).toMatchObject({ id: "session", path: metadata.path });
|
||||
await expect(repo.open(metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already open");
|
||||
await opened.close(BACKGROUND_CONTEXT);
|
||||
|
||||
await withDb(metadata.path, (db) => {
|
||||
sql`INSERT INTO writer_lease (session_id, owner_id, fence, expires_at_ms) VALUES (${"session"}, ${"external"}, ${1}, ${1_000})`.run(
|
||||
db,
|
||||
);
|
||||
});
|
||||
await expect(repo.open(metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already claimed");
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -286,8 +491,8 @@ describe("SqliteSessionRepo", () => {
|
|||
const left = await repo.create({ id: "left" }, BACKGROUND_CONTEXT);
|
||||
const right = await repo.create({ id: "right" }, BACKGROUND_CONTEXT);
|
||||
|
||||
expect(left.metadata.path).toBe(databasePath);
|
||||
expect(right.metadata.path).toBe(databasePath);
|
||||
expect(left.metadata.path).toBe(await realpath(databasePath));
|
||||
expect(right.metadata.path).toBe(await realpath(databasePath));
|
||||
await left.mutate(
|
||||
(mutator) =>
|
||||
mutator.commit(
|
||||
|
|
@ -382,4 +587,325 @@ describe("SqliteSessionRepo", () => {
|
|||
await Promise.all([source.close(BACKGROUND_CONTEXT), fork.close(BACKGROUND_CONTEXT)]);
|
||||
});
|
||||
});
|
||||
|
||||
it("does not create databases for missing open, list, fork, or delete targets", async () => {
|
||||
await withTempDir(async (root) => {
|
||||
const directory = join(root, "missing-directory");
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory,
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
const missing = {
|
||||
id: "missing",
|
||||
createdAt: 1,
|
||||
storageVersion: 1,
|
||||
path: join(directory, "missing.sqlite"),
|
||||
};
|
||||
|
||||
await expect(repo.list(undefined, BACKGROUND_CONTEXT)).resolves.toEqual([]);
|
||||
expect(await pathExists(directory)).toBe(false);
|
||||
await expect(repo.open(missing, BACKGROUND_CONTEXT)).rejects.toThrow();
|
||||
await expect(repo.delete(missing, BACKGROUND_CONTEXT)).rejects.toThrow();
|
||||
await expect(repo.fork(missing, { id: "fork", scope: "tree" }, BACKGROUND_CONTEXT)).rejects.toThrow();
|
||||
expect(await pathExists(missing.path)).toBe(false);
|
||||
expect(await pathExists(join(directory, "fork.sqlite"))).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
it("reserves deletion against local create, open, and fork destinations", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const databaseFactory = new GatedOpenExistingFactory();
|
||||
const repo = new SqliteSessionRepo({ directory, databaseFactory, now: () => 1 });
|
||||
const target = await repo.create({ id: "target" }, BACKGROUND_CONTEXT);
|
||||
const source = await repo.create({ id: "source" }, BACKGROUND_CONTEXT);
|
||||
await expect(repo.delete(target.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already open");
|
||||
// The host closes a worker before deletion; only same-repository exclusion is promised here.
|
||||
await target.close(BACKGROUND_CONTEXT);
|
||||
databaseFactory.arm();
|
||||
|
||||
const deleting = repo.delete(target.metadata, BACKGROUND_CONTEXT);
|
||||
await databaseFactory.entered.promise;
|
||||
await expect(repo.create({ id: "target" }, BACKGROUND_CONTEXT)).rejects.toThrow("already open");
|
||||
await expect(repo.open(target.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already open");
|
||||
await expect(repo.fork(source.metadata, { id: "target", scope: "tree" }, BACKGROUND_CONTEXT)).rejects.toThrow(
|
||||
"already open",
|
||||
);
|
||||
databaseFactory.release.resolve();
|
||||
await deleting;
|
||||
await source.close(BACKGROUND_CONTEXT);
|
||||
});
|
||||
});
|
||||
|
||||
it("deletes only the selected Session from a shared container", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const databasePath = join(directory, "shared.sqlite");
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory,
|
||||
databasePath,
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
const removed = await repo.create({ id: "removed" }, BACKGROUND_CONTEXT);
|
||||
const retained = await repo.create({ id: "retained" }, BACKGROUND_CONTEXT);
|
||||
await removed.setName("removed", BACKGROUND_CONTEXT);
|
||||
await retained.setName("retained", BACKGROUND_CONTEXT);
|
||||
await removed.close(BACKGROUND_CONTEXT);
|
||||
|
||||
await repo.delete(removed.metadata, BACKGROUND_CONTEXT);
|
||||
|
||||
expect((await repo.list(undefined, BACKGROUND_CONTEXT)).map(({ id }) => id)).toEqual(["retained"]);
|
||||
expect(await retained.getName(BACKGROUND_CONTEXT)).toBe("retained");
|
||||
await expect(repo.open(removed.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("Unknown SQLite session");
|
||||
await retained.close(BACKGROUND_CONTEXT);
|
||||
});
|
||||
});
|
||||
|
||||
it("removes per-file WAL and SHM sidecars", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory,
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
||||
await session.close(BACKGROUND_CONTEXT);
|
||||
await writeFile(`${session.metadata.path}-wal`, "");
|
||||
await writeFile(`${session.metadata.path}-shm`, "");
|
||||
|
||||
await repo.delete(session.metadata, BACKGROUND_CONTEXT);
|
||||
|
||||
await expect(access(session.metadata.path)).rejects.toThrow();
|
||||
await expect(access(`${session.metadata.path}-wal`)).rejects.toThrow();
|
||||
await expect(access(`${session.metadata.path}-shm`)).rejects.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
it("creates the parent of a custom shared-container path", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const databasePath = join(directory, "nested", "containers", "sessions.sqlite");
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory: join(directory, "unrelated"),
|
||||
databasePath,
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
|
||||
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
||||
|
||||
expect(session.metadata.path).toBe(await realpath(databasePath));
|
||||
await session.close(BACKGROUND_CONTEXT);
|
||||
});
|
||||
});
|
||||
|
||||
it("encodes unsafe explicit IDs inside the repository directory and round-trips them", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const repo = new SqliteSessionRepo({
|
||||
directory,
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
const ids = ["../escape", "slash/id", "back\\slash", "percent%id", "..", "dots...id", "ユニコード"];
|
||||
const metadata = [];
|
||||
for (const id of ids) {
|
||||
const session = await repo.create({ id }, BACKGROUND_CONTEXT);
|
||||
metadata.push(session.metadata);
|
||||
await session.close(BACKGROUND_CONTEXT);
|
||||
}
|
||||
const canonicalDirectory = await realpath(directory);
|
||||
for (const stored of metadata) {
|
||||
const fromDirectory = relative(canonicalDirectory, stored.path);
|
||||
expect(isAbsolute(fromDirectory)).toBe(false);
|
||||
expect(fromDirectory.startsWith(".."), stored.id).toBe(false);
|
||||
expect(dirname(stored.path)).toBe(canonicalDirectory);
|
||||
}
|
||||
expect((await repo.list(undefined, BACKGROUND_CONTEXT)).map(({ id }) => id).sort()).toEqual([...ids].sort());
|
||||
for (const stored of metadata) {
|
||||
const reopened = await repo.open(stored, BACKGROUND_CONTEXT);
|
||||
expect(reopened.metadata.id).toBe(stored.id);
|
||||
await reopened.close(BACKGROUND_CONTEXT);
|
||||
}
|
||||
const fork = await repo.fork(metadata[0]!, { id: "fork/../%", scope: "tree" }, BACKGROUND_CONTEXT);
|
||||
expect(fork.metadata).toMatchObject({ id: "fork/../%", parentSessionId: "../escape" });
|
||||
expect(dirname(fork.metadata.path)).toBe(canonicalDirectory);
|
||||
await fork.close(BACKGROUND_CONTEXT);
|
||||
});
|
||||
});
|
||||
|
||||
it("uses exact physical identity for foreign fork sources and rejects foreign writable metadata", async () => {
|
||||
await withTempDir(async (root) => {
|
||||
const leftRepo = new SqliteSessionRepo({
|
||||
directory: join(root, "left"),
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
const rightRepo = new SqliteSessionRepo({
|
||||
directory: join(root, "right"),
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
const left = await leftRepo.create({ id: "same" }, BACKGROUND_CONTEXT);
|
||||
const right = await rightRepo.create({ id: "same" }, BACKGROUND_CONTEXT);
|
||||
await left.mutate(
|
||||
(mutator) =>
|
||||
mutator.commit(
|
||||
[
|
||||
sessionWrites.insertEntry({ id: "left-root", parentId: null, type: "custom", customType: "left" }),
|
||||
storedValues.setValue(storedValues.branchTip("main"), "left-root"),
|
||||
storedValues.setValue(storedValues.sessionName, "left"),
|
||||
],
|
||||
BACKGROUND_CONTEXT,
|
||||
),
|
||||
BACKGROUND_CONTEXT,
|
||||
);
|
||||
await right.mutate(
|
||||
(mutator) =>
|
||||
mutator.commit(
|
||||
[
|
||||
sessionWrites.insertEntry({
|
||||
id: "right-root",
|
||||
parentId: null,
|
||||
type: "custom",
|
||||
customType: "right",
|
||||
}),
|
||||
storedValues.setValue(storedValues.branchTip("main"), "right-root"),
|
||||
storedValues.setValue(storedValues.sessionName, "right"),
|
||||
],
|
||||
BACKGROUND_CONTEXT,
|
||||
),
|
||||
BACKGROUND_CONTEXT,
|
||||
);
|
||||
|
||||
const fork = await rightRepo.fork(left.metadata, { id: "fork-left" }, BACKGROUND_CONTEXT);
|
||||
expect((await fork.findEntries({ order: "asc" }, BACKGROUND_CONTEXT)).map(({ id }) => id)).toEqual([
|
||||
"left-root",
|
||||
]);
|
||||
expect(await fork.getName(BACKGROUND_CONTEXT)).toBe("left");
|
||||
await right.close(BACKGROUND_CONTEXT);
|
||||
await expect(rightRepo.open(left.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("outside this repository");
|
||||
await expect(rightRepo.delete(left.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("outside this repository");
|
||||
await Promise.all([left.close(BACKGROUND_CONTEXT), fork.close(BACKGROUND_CONTEXT)]);
|
||||
});
|
||||
});
|
||||
|
||||
for (const layout of ["per-file", "shared"] as const) {
|
||||
it(`forks a live ${layout} source through one coherent read-only WAL snapshot`, async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const databasePath = layout === "shared" ? join(directory, "shared.sqlite") : undefined;
|
||||
const workerRepo = new SqliteSessionRepo({
|
||||
directory,
|
||||
...(databasePath === undefined ? {} : { databasePath }),
|
||||
databaseFactory: createNodeSqliteFactory(),
|
||||
now: () => 1,
|
||||
});
|
||||
const source = await workerRepo.create({ id: "source" }, BACKGROUND_CONTEXT);
|
||||
await source.mutate(
|
||||
(mutator) =>
|
||||
mutator.commit(
|
||||
[
|
||||
sessionWrites.insertEntry({ id: "root", parentId: null, type: "custom", customType: "root" }),
|
||||
storedValues.setValue(storedValues.branchTip("main"), "root"),
|
||||
storedValues.setValue(storedValues.sessionName, "before"),
|
||||
storedValues.setValue(storedValues.entryLabel("root"), "before-label"),
|
||||
],
|
||||
BACKGROUND_CONTEXT,
|
||||
),
|
||||
BACKGROUND_CONTEXT,
|
||||
);
|
||||
const writerDb = await createNodeSqliteFactory().openExisting(source.metadata.path);
|
||||
writerDb.exec("PRAGMA busy_timeout = 5000;");
|
||||
let laterCommitCompleted = false;
|
||||
const databaseFactory = new SnapshotBoundaryFactory(() => {
|
||||
if (laterCommitCompleted) return;
|
||||
commitLaterSourceState(writerDb);
|
||||
laterCommitCompleted = true;
|
||||
});
|
||||
const serverRepo = new SqliteSessionRepo({
|
||||
directory,
|
||||
...(databasePath === undefined ? {} : { databasePath }),
|
||||
databaseFactory,
|
||||
now: () => 2,
|
||||
});
|
||||
|
||||
const first = await serverRepo.fork(source.metadata, { id: "first-fork" }, BACKGROUND_CONTEXT);
|
||||
|
||||
expect(laterCommitCompleted).toBe(true);
|
||||
expect(databaseFactory.readOnlyOpenCount).toBe(1);
|
||||
expect((await first.findEntries({ order: "asc" }, BACKGROUND_CONTEXT)).map(({ id }) => id)).toEqual([
|
||||
"root",
|
||||
]);
|
||||
expect(await first.getName(BACKGROUND_CONTEXT)).toBe("before");
|
||||
expect(await first.getLabel("root", BACKGROUND_CONTEXT)).toBe("before-label");
|
||||
await expect((await first.branch("main", BACKGROUND_CONTEXT))?.getTipId(BACKGROUND_CONTEXT)).resolves.toBe(
|
||||
"root",
|
||||
);
|
||||
expect((await first.getStats(BACKGROUND_CONTEXT)).messageCount).toBe(0);
|
||||
|
||||
const second = await serverRepo.fork(source.metadata, { id: "second-fork" }, BACKGROUND_CONTEXT);
|
||||
expect(databaseFactory.readOnlyOpenCount).toBe(2);
|
||||
expect((await second.findEntries({ order: "asc" }, BACKGROUND_CONTEXT)).map(({ id }) => id)).toEqual([
|
||||
"root",
|
||||
"child",
|
||||
]);
|
||||
expect(await second.getName(BACKGROUND_CONTEXT)).toBe("after");
|
||||
expect(await second.getLabel("root", BACKGROUND_CONTEXT)).toBe("after-label");
|
||||
await expect((await second.branch("main", BACKGROUND_CONTEXT))?.getTipId(BACKGROUND_CONTEXT)).resolves.toBe(
|
||||
"child",
|
||||
);
|
||||
expect((await second.getStats(BACKGROUND_CONTEXT)).messageCount).toBe(1);
|
||||
|
||||
writerDb.close();
|
||||
await Promise.all([
|
||||
source.close(BACKGROUND_CONTEXT),
|
||||
first.close(BACKGROUND_CONTEXT),
|
||||
second.close(BACKGROUND_CONTEXT),
|
||||
]);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
it("waits for every Session close and reports one or multiple failures after settlement", async () => {
|
||||
await withTempDir(async (directory) => {
|
||||
const multipleFactory = new CloseTrackingFactory();
|
||||
const multipleRepo = new SqliteSessionRepo({
|
||||
directory: join(directory, "multiple"),
|
||||
databaseFactory: multipleFactory,
|
||||
now: () => 1,
|
||||
});
|
||||
await multipleRepo.create({ id: "first" }, BACKGROUND_CONTEXT);
|
||||
await multipleRepo.create({ id: "second" }, BACKGROUND_CONTEXT);
|
||||
await multipleRepo.create({ id: "third" }, BACKGROUND_CONTEXT);
|
||||
const firstError = new Error("first close failed");
|
||||
const secondError = new Error("second close failed");
|
||||
multipleFactory.writableConnections[0]!.closeError = firstError;
|
||||
multipleFactory.writableConnections[1]!.closeError = secondError;
|
||||
|
||||
const multipleClose = multipleRepo.close(BACKGROUND_CONTEXT);
|
||||
expect(multipleRepo.close(BACKGROUND_CONTEXT)).toBe(multipleClose);
|
||||
let multipleFailure: unknown;
|
||||
try {
|
||||
await multipleClose;
|
||||
} catch (error) {
|
||||
multipleFailure = error;
|
||||
}
|
||||
expect(multipleFailure).toBeInstanceOf(AggregateError);
|
||||
if (!(multipleFailure instanceof AggregateError)) throw new Error("Expected AggregateError");
|
||||
expect(multipleFailure.errors).toEqual([firstError, secondError]);
|
||||
expect(multipleFactory.writableConnections.map(({ closeAttempts }) => closeAttempts)).toEqual([1, 1, 1]);
|
||||
|
||||
const singleFactory = new CloseTrackingFactory();
|
||||
const singleRepo = new SqliteSessionRepo({
|
||||
directory: join(directory, "single"),
|
||||
databaseFactory: singleFactory,
|
||||
now: () => 1,
|
||||
});
|
||||
await singleRepo.create({ id: "first" }, BACKGROUND_CONTEXT);
|
||||
await singleRepo.create({ id: "second" }, BACKGROUND_CONTEXT);
|
||||
const singleError = new Error("single close failed");
|
||||
singleFactory.writableConnections[0]!.closeError = singleError;
|
||||
|
||||
await expect(singleRepo.close(BACKGROUND_CONTEXT)).rejects.toBe(singleError);
|
||||
expect(singleFactory.writableConnections.map(({ closeAttempts }) => closeAttempts)).toEqual([1, 1]);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -2,13 +2,45 @@ import * as storedValues from "@earendil-works/pi-agent-core";
|
|||
import * as sessionWrites from "@earendil-works/pi-agent-core";
|
||||
import { BACKGROUND_CONTEXT, prepareStorageCommit } from "@earendil-works/pi-agent-core";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { createNodeSqliteFactory, type SqliteDatabase, SqliteStorage, sql } from "../src/index.ts";
|
||||
import {
|
||||
createNodeSqliteFactory,
|
||||
type SqliteDatabase,
|
||||
type SqliteStatement,
|
||||
SqliteStorage,
|
||||
sql,
|
||||
} from "../src/index.ts";
|
||||
import { applyInitialSchema } from "../src/sqlite/migrations.ts";
|
||||
import { advanceNextSeq, readNextSeq } from "../src/sqlite/session/session-sequences.ts";
|
||||
import { listValueReadQuery } from "../src/sqlite/session/values.ts";
|
||||
|
||||
const SESSION_ID = "session";
|
||||
|
||||
class TransactionCountingDatabase implements SqliteDatabase {
|
||||
readonly source: SqliteDatabase;
|
||||
transactionCount = 0;
|
||||
|
||||
constructor(source: SqliteDatabase) {
|
||||
this.source = source;
|
||||
}
|
||||
|
||||
exec(query: string): void {
|
||||
this.source.exec(query);
|
||||
}
|
||||
|
||||
prepare(query: string): SqliteStatement {
|
||||
return this.source.prepare(query);
|
||||
}
|
||||
|
||||
transaction<T>(callback: () => T): T {
|
||||
this.transactionCount++;
|
||||
return this.source.transaction(callback);
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.source.close();
|
||||
}
|
||||
}
|
||||
|
||||
const ZERO_USAGE = {
|
||||
input: 0,
|
||||
output: 0,
|
||||
|
|
@ -59,6 +91,22 @@ function insertCommitSessionRow(db: SqliteDatabase, nextSeq = 1): void {
|
|||
}
|
||||
|
||||
describe("SqliteStorage", () => {
|
||||
it("uses one write transaction for an ordinary commit", async () => {
|
||||
const db = new TransactionCountingDatabase(await createNodeSqliteFactory().open(":memory:"));
|
||||
try {
|
||||
await applyInitialSchema(db);
|
||||
insertCommitSessionRow(db);
|
||||
const storage = new SqliteStorage(db, { sessionId: SESSION_ID, now: () => 1_700_000_000_000 });
|
||||
|
||||
await storage.commit([storedValues.setValue(storedValues.sessionName, "name")], BACKGROUND_CONTEXT);
|
||||
|
||||
expect(db.transactionCount).toBe(1);
|
||||
await storage.close(BACKGROUND_CONTEXT);
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("commits root entries and append-to-tip entries into the branch index", async () => {
|
||||
await withStorage(async (storage, db) => {
|
||||
insertCommitSessionRow(db);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue