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.
47 lines
2.1 KiB
TypeScript
47 lines
2.1 KiB
TypeScript
import { describe, expect, it } from "vitest";
|
|
import { advanceRange, readContext, readContextFrom } from "../src/harness/context.ts";
|
|
import { SessionImpl } from "../src/session/session.ts";
|
|
import { MemoryStorage } from "../src/storage/memory.ts";
|
|
import { context } from "./session-support.ts";
|
|
|
|
describe("kept context ranges", () => {
|
|
it("is not advanced when an entry was committed while its scan ran off the line", async () => {
|
|
const entered = Promise.withResolvers<void>();
|
|
const release = Promise.withResolvers<void>();
|
|
class PausedStorage extends MemoryStorage {
|
|
pause = false;
|
|
override async scanEntries(...args: Parameters<MemoryStorage["scanEntries"]>) {
|
|
const page = await super.scanEntries(...args);
|
|
if (this.pause) {
|
|
this.pause = false;
|
|
entered.resolve();
|
|
await release.promise;
|
|
}
|
|
return page;
|
|
}
|
|
}
|
|
const storage = new PausedStorage();
|
|
const session = new SessionImpl(storage);
|
|
const conversation = await session.commit(
|
|
(tx) => tx.createConversation({ ownership: { kind: "ownerless" } }),
|
|
context,
|
|
);
|
|
await session.commit((tx) => tx.appendEntry(conversation.id, { kind: "first" }), context);
|
|
storage.pause = true;
|
|
const read = readContextFrom(session, storage, conversation.id, context, undefined, undefined);
|
|
await entered.promise;
|
|
await session.commit((tx) => tx.appendEntry(conversation.id, { kind: "missed" }), context);
|
|
release.resolve();
|
|
const { range } = await read;
|
|
expect(range?.current).toBe(false);
|
|
const last = await session.commit((tx) => tx.appendEntry(conversation.id, { kind: "last" }), context);
|
|
// Not current: left as it is, so the next read extends it from Storage instead of skipping "missed".
|
|
expect(advanceRange(range!, [last])).toBe(range);
|
|
const again = await readContextFrom(session, storage, conversation.id, context, undefined, range);
|
|
const fresh = await readContext(session, storage, conversation.id, context);
|
|
expect(again.view.entries.map((entry) => entry.kind)).toEqual(["first", "missed", "last"]);
|
|
expect(again.view).toEqual(fresh);
|
|
expect(again.range?.current).toBe(true);
|
|
await session.close(context);
|
|
});
|
|
});
|