Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
17 kB · 322 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323import fs from "node:fs/promises";import { afterEach, describe, expect, test, vi } from "vitest";import { DeterministicBatcher, acquireBatchOwnerLocks, batchDeclarationFingerprint, batchProgressId } from "../src/batches/runtime.js";import type { EventCandidate, ThoughtEvent } from "../src/events/types.js";import type { JazzThoughtStore } from "../src/jazz/store.js";import type { BatchDeclarationManifest } from "../src/runtime/manifest.js";import { temporaryProject, testStore } from "./helpers.js";
const stores: JazzThoughtStore[] = [];const roots: string[] = [];
afterEach(async () => { vi.restoreAllMocks(); await Promise.all(stores.splice(0).map((store) => store.close())); await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })));});
describe("deterministic derived-event batching", () => { test("flushes one isolated commit after restart-stable quiet time", async () => { const { store } = await setup(); await appendCommit(store, 1, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z"); const declaration = batchDeclaration({ quietWindowMs: 1_000 });
expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:00.500Z")).cycle(declaration)) .toMatchObject({ waiting: true, emitted: 0 }); const restarted = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:02.000Z")); expect(await restarted.cycle(declaration)).toMatchObject({ flushReason: "quiet-window", memberCount: 1 }); expect(await restarted.cycle(declaration)).toMatchObject({ emitted: 0, examined: 0 }); });
test("cursor and lifecycle source events do not postpone quiet flush", async () => { const { store } = await setup(); await appendCommit(store, 100, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z"); await appendNonmatching(store, "cursor-101", "2026-07-21T00:00:00.100Z"); await appendNonmatching(store, "lifecycle-102", "2026-07-21T00:00:01.500Z"); const result = await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:02.000Z")) .cycle(batchDeclaration({ quietWindowMs: 1_000 })); expect(result).toMatchObject({ flushReason: "quiet-window", memberCount: 1 }); });
test("sparse isolated matching commits each flush without raw-source head blocking", async () => { const { store } = await setup(); const declaration = batchDeclaration({ quietWindowMs: 100 }); await appendCommit(store, 1, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z"); await appendNonmatching(store, "cursor-after-first", "2026-07-21T00:00:00.010Z"); const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z")); expect(await batcher.cycle(declaration)).toMatchObject({ memberCount: 1 }); await appendCommit(store, 2, "app.bsky.feed.post", "2026-07-21T00:00:10.000Z"); await appendNonmatching(store, "cursor-after-second", "2026-07-21T00:00:10.010Z"); expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:11.000Z")).cycle(declaration)) .toMatchObject({ memberCount: 1 }); });
test("max-items forces a burst batch and replay is idempotent", async () => { const { store } = await setup(); for (let index = 1; index <= 3; index += 1) await appendCommit(store, index, "app.bsky.feed.like", `2026-07-21T00:00:0${index}.000Z`); const declaration = batchDeclaration({ maxItems: 3, quietWindowMs: 60_000, maxAgeMs: 120_000 }); const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:04.000Z")); const first = await batcher.cycle(declaration); expect(first).toMatchObject({ flushReason: "max-items", memberCount: 3, emitted: 1 }); const batch = await store.getEvent(first.batchEventId!); expect((batch!.payload.members as unknown[])).toHaveLength(3); expect(JSON.stringify(batch!.payload)).not.toContain("LIKE-BODY"); expect(await batcher.cycle(declaration)).toMatchObject({ emitted: 0 }); });
test("max-age prevents starvation under a continuously advancing source", async () => { const { store } = await setup(); for (let index = 1; index <= 4; index += 1) await appendCommit(store, index, "app.bsky.feed.like", `2026-07-21T00:00:0${index}.000Z`); const declaration = batchDeclaration({ maxItems: 50, quietWindowMs: 10_000, maxAgeMs: 3_000 }); expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:04.500Z")).cycle(declaration)) .toMatchObject({ flushReason: "max-age", memberCount: 4 }); });
test("crash before atomic settlement leaves progress unchanged and replay succeeds", async () => { const { store } = await setup(); await appendCommit(store, 1, "app.bsky.feed.like", "2026-07-21T00:00:00.000Z"); const declaration = batchDeclaration({ quietWindowMs: 100 }); const original = store.settleDerivedBatch.bind(store); vi.spyOn(store, "settleDerivedBatch").mockRejectedValueOnce(new Error("synthetic crash boundary")); const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z")); await expect(batcher.cycle(declaration)).rejects.toThrow("synthetic crash boundary"); expect(await store.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE))).toBeUndefined(); vi.mocked(store.settleDerivedBatch).mockImplementation(original); expect(await batcher.cycle(declaration)).toMatchObject({ emitted: 1, memberCount: 1 }); });
test("filtered progress crosses nonmatching rows but rejects a hidden matching interval member", async () => { const { store } = await setup(); const declaration = batchDeclaration({ quietWindowMs: 100 }); const commit10 = await appendCommit(store, 10, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z"); await appendNonmatching(store, "cursor-11", "2026-07-21T00:00:00.010Z"); const commit12 = await appendCommit(store, 12, "app.bsky.feed.like", "2026-07-21T00:00:00.020Z"); expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z")).cycle(declaration)) .toMatchObject({ memberCount: 2 });
const { store: hiddenStore } = await setup(); const hidden10 = await appendCommit(hiddenStore, 10, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z"); const hidden11 = await appendCommit(hiddenStore, 11, "app.bsky.feed.like", "2026-07-21T00:00:00.010Z"); const hidden12 = await appendCommit(hiddenStore, 12, "app.bsky.feed.like", "2026-07-21T00:00:00.020Z"); const candidate = batchCandidate(declaration, [hidden10, hidden12]); await expect(hiddenStore.settleDerivedBatch({ candidate, progress: { id: batchProgressId(declaration, RAW_SOURCE), consumerId: declaration.id, consumerVersion: declaration.version, source: RAW_SOURCE, lastSequence: hidden12.sourceSequence, lastEventId: hidden12.id, updatedAt: "2026-07-21T00:00:01.000Z", }, members: [hidden10, hidden12], priorFilteredSequence: 0, declaration: { inputEventTypes: declaration.input.eventTypes, acceptedPrivacy: ["public-source", "private", "sensitive"], source: RAW_SOURCE, fingerprint: String((candidate.payload.declaration as Record<string, unknown>).fingerprint), }, })).rejects.toThrow("complete matching filtered interval"); expect(await hiddenStore.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE))).toBeUndefined(); expect(await hiddenStore.listEvents({ source: declaration.output.sourceId })).toHaveLength(0); expect(hidden11.sourceSequence).toBeGreaterThan(hidden10.sourceSequence); expect(commit12.sourceSequence).toBeGreaterThan(commit10.sourceSequence); });
test("producer append after replay-now captured head remains pending", async () => { const { store } = await setup(); const capturedHead = await appendNonmatching(store, "captured-head", "2026-07-21T00:00:00.000Z"); const declaration = batchDeclaration({ replay: "now", quietWindowMs: 100 }); const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:00.050Z")); expect(await batcher.cycle(declaration)).toMatchObject({ emitted: 0 }); expect(await store.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE))) .toMatchObject({ lastSequence: capturedHead.sourceSequence, lastEventId: capturedHead.id }); await appendCommit(store, 2, "app.bsky.feed.post", "2026-07-21T00:00:00.060Z"); expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z")).cycle(declaration)) .toMatchObject({ emitted: 1, memberCount: 1 }); });
test("fails closed rather than declassifying mixed privacy", async () => { const { store } = await setup(); await appendCommit(store, 1, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z"); const candidate = commitCandidate(2, "app.bsky.feed.like", "2026-07-21T00:00:01.000Z"); await store.appendEvent({ ...candidate, privacy: "private" }); const declaration = batchDeclaration({ quietWindowMs: 100, privacy: "preserve" }); await expect(new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:02.000Z")).cycle(declaration)) .rejects.toThrow("different privacy classes"); expect(await store.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE))).toBeUndefined(); });
test("atomically merges complete prefixes from multiple sources into one most-private window", async () => { const { store } = await setup(); const secondSource = "x:fixture"; const first = await appendCommit(store, 1, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z"); const second = (await store.appendEvent({ ...commitCandidate(2, "app.bsky.feed.like", "2026-07-21T00:00:00.100Z"), source: secondSource, sourceKind: "x-webhook", externalId: "x-event-2", idempotencyKey: "x-event-2", privacy: "sensitive", })).event; const third = await appendCommit(store, 3, "app.bsky.feed.like", "2026-07-21T00:00:00.200Z"); const declaration = batchDeclaration({ id: "fixture-activity-window", input: { eventTypes: ["stream.thought.source.atproto.commit"], sourceIds: [RAW_SOURCE, secondSource] }, output: { sourceId: "batch:fixture-activity", eventType: "stream.thought.derived.event.batch" }, quietWindowMs: 100, privacy: "most-private", }); const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z")); const result = await batcher.cycle(declaration); expect(result).toMatchObject({ emitted: 1, memberCount: 3, flushReason: "quiet-window" }); const batch = await store.getEvent(result.batchEventId!); expect(batch).toMatchObject({ privacy: "sensitive", rootEventId: result.batchEventId }); expect((batch!.payload.members as Array<{ eventId: string }>).map((member) => member.eventId)).toEqual([ first.id, second.id, third.id, ]); expect(await store.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE))) .toMatchObject({ lastSequence: third.sourceSequence, lastEventId: third.id }); expect(await store.getConsumerProgress(batchProgressId(declaration, secondSource))) .toMatchObject({ lastSequence: second.sourceSequence, lastEventId: second.id }); expect(await batcher.cycle(declaration)).toMatchObject({ emitted: 0, examined: 0 }); });});
describe("batch declaration owner locks", () => { test("rejects a live owner and releases cleanly", async () => { const root = await temporaryProject("thoughtstream-batch-lock-"); roots.push(root); const declaration = batchDeclaration(); const identity = { bootId: async () => "boot-a", processStart: async () => "start-a", pid: 123 }; const first = await acquireBatchOwnerLocks(root, [declaration], identity); await expect(acquireBatchOwnerLocks(root, [declaration], identity)).rejects.toThrow("live batch owner"); await first.release(); const second = await acquireBatchOwnerLocks(root, [declaration], identity); await second.release(); });
test("recovers dead pid and boot or process-start mismatches", async () => { const root = await temporaryProject("thoughtstream-batch-lock-"); roots.push(root); const declaration = batchDeclaration(); const first = await acquireBatchOwnerLocks(root, [declaration], { bootId: async () => "boot-old", processStart: async () => "start-old", pid: 111, }); const rebooted = await acquireBatchOwnerLocks(root, [declaration], { bootId: async () => "boot-new", processStart: async () => "start-new", pid: 222, }); await rebooted.release(); await first.release();
const stale = await acquireBatchOwnerLocks(root, [declaration], { bootId: async () => "boot-same", processStart: async () => "old-start", pid: 333, }); const reused = await acquireBatchOwnerLocks(root, [declaration], { bootId: async () => "boot-same", processStart: async (pid) => pid === 333 ? "reused-start" : "new-start", pid: 444, }); await reused.release(); await stale.release();
const dead = await acquireBatchOwnerLocks(root, [declaration], { bootId: async () => "boot-dead", processStart: async () => "dead-start", pid: 555, }); const recovered = await acquireBatchOwnerLocks(root, [declaration], { bootId: async () => "boot-dead", processStart: async (pid) => pid === 555 ? undefined : "live-start", pid: 666, }); await recovered.release(); await dead.release(); });});
const RAW_SOURCE = "jetstream:fixture";
function batchDeclaration(overrides: Partial<BatchDeclarationManifest> = {}): BatchDeclarationManifest { return { id: "fixture-atproto-batch", version: 1, enabled: true, input: { eventTypes: ["stream.thought.source.atproto.commit"], sourceIds: [RAW_SOURCE] }, output: { sourceId: "batch:fixture-atproto", eventType: "stream.thought.derived.event.batch" }, quietWindowMs: 1_000, maxAgeMs: 10_000, maxItems: 10, privacy: "preserve", replay: "beginning", pollIntervalMs: 100, ...overrides, };}
async function appendCommit(store: JazzThoughtStore, index: number, collection: string, occurredAt: string): Promise<ThoughtEvent> { return (await store.appendEvent(commitCandidate(index, collection, occurredAt))).event;}
async function appendNonmatching(store: JazzThoughtStore, key: string, occurredAt: string): Promise<ThoughtEvent> { return (await store.appendEvent({ ...commitCandidate(999, "app.bsky.feed.post", occurredAt), type: "stream.thought.connector.cursor.advanced", externalId: key, idempotencyKey: key, payload: { cursor: key }, })).event;}
function batchCandidate(declaration: BatchDeclarationManifest, members: ThoughtEvent[]): EventCandidate { const fingerprint = sha256ForTest(declaration); return { type: declaration.output.eventType, schemaVersion: 1, source: declaration.output.sourceId, sourceKind: "system", externalId: `batch-${members.map((event) => event.id).join("-")}`, idempotencyKey: `batch-${members.map((event) => event.id).join("-")}`, occurredAt: members.at(-1)!.occurredAt, actor: `batch:${declaration.id}`, correlationId: "batch-test", privacy: "public-source", payload: { declaration: { id: declaration.id, version: declaration.version, fingerprint }, flushReason: "quiet-window", firstOccurredAt: members[0]!.occurredAt, lastOccurredAt: members.at(-1)!.occurredAt, members: members.map((event) => ({ eventId: event.id, source: event.source, sourceSequence: event.sourceSequence, type: event.type, schemaVersion: event.schemaVersion, privacy: event.privacy, occurredAt: event.occurredAt, observedAt: event.observedAt, payloadHash: event.payloadHash, })), }, };}
function sha256ForTest(declaration: BatchDeclarationManifest): string { return batchDeclarationFingerprint(declaration);}
function commitCandidate(index: number, collection: string, occurredAt: string): EventCandidate { return { type: "stream.thought.source.atproto.commit", schemaVersion: 1, source: RAW_SOURCE, sourceKind: "jetstream", externalId: `at://did:plc:fixture/${collection}/${index}`, idempotencyKey: `commit-${index}`, occurredAt, observedAt: occurredAt, actor: "did:plc:fixture", correlationId: "synthetic-burst", privacy: "public-source", payload: { atUri: `at://did:plc:fixture/${collection}/${index}`, collection, operation: "create", record: collection === "app.bsky.feed.like" ? { subject: { uri: `at://did:plc:subject/app.bsky.feed.post/${index}`, cid: `bafy-${index}` }, note: "LIKE-BODY" } : { text: "POST-BODY" }, }, };}
async function setup(): Promise<{ store: JazzThoughtStore }> { const root = await temporaryProject("thoughtstream-batches-"); roots.push(root); const store = await testStore(root); stores.push(store); return { store };}