mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
e5c6182bab
A Session no longer holds its line while a commit is stored. Commits that arrive while one storage call is in flight go out together in the next call, each with its own seq and outcome; a job runs until it touches a key an unpublished commit wrote, then waits for that commit's publication. Outcomes are delivered in line order; publication is in line order, one commit per macrotask. The scheduler's decisions run as regions that record the keys they read and rerun after a wait, with side effects only after a clean check. Storage.commit(batches, receive, context) replaces commit(writes, adopt, context): prefix semantics (Durable's own checks refuse a batch and keep the ones before it; any other failure rejects the call), receipts per stored batch, optional publishing() scope. Memory, SQLite, JSONL, IndexedDB and StoreStorage implement it natively; StoreStorage through the new ChordStore.commitSeries (one store transaction per call), replacing commitInBatch. A normal run makes no storage reads: bounded record cache, conversation tails, task-document index, kept context ranges. A tool slot's running status and a generation's attempt are stored with the next commit (12 instead of 14 commits per run). One Harness with 128 conversations over storage with a 4 ms round trip: about 5 -> 440 agent runs per second. Design: packages/durable/docs/group-commit.md.
71 lines
2.7 KiB
TypeScript
71 lines
2.7 KiB
TypeScript
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
|
|
import { registerStorageConformance } from "@earendil-works/pi-durable/testing";
|
|
import { describe, expect, it } from "vitest";
|
|
import { idFromNumber, seqFromNumber } from "../src/ids.ts";
|
|
import { MemoryStorage } from "../src/storage/memory.ts";
|
|
import { commitWrites } from "../src/testing/commit-writes.ts";
|
|
import { type EntryId, ROOT_CONVERSATION_ID, type Seq, type StorageWrite } from "../src/types.ts";
|
|
|
|
registerStorageConformance({ describe, expect, it }, "MemoryStorage", (use) => use(new MemoryStorage()));
|
|
|
|
it("does not expose retained state through a prepared commit", async () => {
|
|
const storage = new MemoryStorage();
|
|
await commitWrites(storage, [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], BACKGROUND_CONTEXT);
|
|
const entryId = idFromNumber<EntryId>(2);
|
|
const prepared = storage.prepareCommit([
|
|
{
|
|
type: "entry",
|
|
value: {
|
|
id: entryId,
|
|
conversationId: ROOT_CONVERSATION_ID,
|
|
kind: "test",
|
|
data: { nested: [1] },
|
|
},
|
|
},
|
|
]);
|
|
const exposed = prepared.writes[0];
|
|
if (exposed.type !== "entry") throw new Error("Expected an entry write");
|
|
expect(() => (exposed.value.data as { nested: number[] }).nested.push(2)).toThrow();
|
|
|
|
expect(prepared.apply()).toBe(seqFromNumber(2));
|
|
expect(prepared.apply()).toBe(seqFromNumber(2));
|
|
expect((await storage.entry(entryId, BACKGROUND_CONTEXT))?.entry.data).toEqual({ nested: [1] });
|
|
});
|
|
|
|
it("delivers the receipts of the batches applied before one whose apply throws, then rejects", async () => {
|
|
const failure = new Error("apply failed");
|
|
class FailingApply extends MemoryStorage {
|
|
failAt: Seq | undefined;
|
|
override prepareCommit(writes: readonly StorageWrite[], seq?: Seq) {
|
|
const prepared = super.prepareCommit(writes, seq);
|
|
if (prepared.seq !== this.failAt) return prepared;
|
|
return {
|
|
...prepared,
|
|
apply: (): Seq => {
|
|
throw failure;
|
|
},
|
|
};
|
|
}
|
|
}
|
|
const storage = new FailingApply();
|
|
const root = await commitWrites(
|
|
storage,
|
|
[{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }],
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
const [first, second] = [await storage.mintId<EntryId>(), await storage.mintId<EntryId>()];
|
|
storage.failAt = seqFromNumber(root + 2);
|
|
const receipts: [number, Seq][] = [];
|
|
await expect(
|
|
storage.commit(
|
|
[first, second].map((id) => [
|
|
{ type: "entry", value: { id, conversationId: ROOT_CONVERSATION_ID, kind: "m" } },
|
|
]),
|
|
(index, seq) => receipts.push([index, seq]),
|
|
BACKGROUND_CONTEXT,
|
|
),
|
|
).rejects.toBe(failure);
|
|
expect(receipts).toEqual([[0, root + 1]]);
|
|
expect((await storage.entry(first, BACKGROUND_CONTEXT))?.commitSeq).toBe(root + 1);
|
|
expect(await storage.entry(second, BACKGROUND_CONTEXT)).toBeUndefined();
|
|
});
|