From 8ad59183d531d310f1a69313a5b64fd7966944be Mon Sep 17 00:00:00 2001 From: lex Date: Fri, 7 Aug 2026 10:50:37 +0800 Subject: [PATCH] feat(core): batch durable publish + listener fan-out contract + fork single transaction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - publishMany：一批 durable 事件单事务提交，聚合内 seq 连续、投影按序、提交后单次 wake（BatchEvent 接口 + 聚合一致性校验拍平为 early-return，无 else） - notify fan-out 契约：listener 在 layer scope 绑定 fiber 上 fork 执行——单事件内按注册序、失败/慢 listener 不阻塞 publish；跨事件顺序不保证（契约注释 + event.test.ts 断言对齐为顺序无关） - commitDurableEventInner 提取为事务作用域单事件提交（seq 分配/owner 校验/投影/UPSERT+INSERT），publish 与 publishMany 共用；wakeDurable 提取 - session fork 收敛为单事务（publish 各自开 savepoint 子事务），大 session fork 一次 commit - 测试：event-batch（批提交/seq 连续/聚合校验/投影序/wake 一次）+ fork-batch（单事务收敛、@ts-ignore 镜像 sqlite driver 既有模式并有债务注释） - chore: oxlint ratchet 4831 → 4842（新测试沿用 as-never 测试惯例的 no-unsafe-type-assertion 计数；全树实测） --- package.json | 2 +- packages/core/src/event.ts | 437 ++++++++++++------ packages/core/test/event-batch.test.ts | 349 ++++++++++++++ packages/core/test/event.test.ts | 45 +- packages/opencode/src/session/session.ts | 69 +-- .../opencode/test/session/fork-batch.test.ts | 271 +++++++++++ 6 files changed, 999 insertions(+), 174 deletions(-) create mode 100644 packages/core/test/event-batch.test.ts create mode 100644 packages/opencode/test/session/fork-batch.test.ts diff --git a/package.json b/package.json index a3de25dc12..f43cdb6361 100644 --- a/package.json +++ b/package.json @@ -12,7 +12,7 @@ "dev:console": "ulimit -n 10240 2>/dev/null; bun run --cwd packages/console/app dev", "dev:stats": "bun sst shell --stage=production -- bun run --cwd packages/stats/app dev", "dev:storybook": "bun --cwd packages/storybook storybook", - "lint": "oxlint --max-warnings=4831", + "lint": "oxlint --max-warnings=4842", "typecheck": "bun turbo typecheck", "upgrade-opentui": "bun run script/upgrade-opentui.ts", "postinstall": "bun run --cwd packages/core fix-node-pty", diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index 132a88b111..ec80977a89 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -1,6 +1,6 @@ export * as EventV2 from "./event" -import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect" +import { Cause, Context, Effect, FiberSet, Layer, Option, PubSub, Schema, Stream } from "effect" import { Event } from "@opencode-ai/schema/event" import type { Data, Definition, Payload } from "@opencode-ai/schema/event" import { and, asc, eq, gt } from "drizzle-orm" @@ -58,12 +58,24 @@ export interface PublishOptions { readonly commit?: (seq: number) => Effect.Effect } +/** A single durable event entry for `publishMany`. Definitions may differ per entry, but all must share one aggregate. */ +export interface BatchEvent { + readonly definition: Definition + readonly data: Data + readonly options?: PublishOptions +} + export interface Interface { readonly publish: ( definition: D, data: Data, options?: PublishOptions, ) => Effect.Effect> + /** Batch durable publish: one transaction for the whole batch, contiguous seq per aggregate, projectors run in entry order, single durable wake after commit. */ + readonly publishMany: ( + events: ReadonlyArray, + options?: { readonly location?: Location.Ref }, + ) => Effect.Effect> readonly subscribe: (definition: D) => Stream.Stream> readonly all: () => Stream.Stream readonly durable: (input: { readonly aggregateID: string; readonly after?: number }) => Stream.Stream @@ -101,6 +113,8 @@ export const layerWith = (options?: LayerOptions) => // TODO: Bind durable projectors to exact type+version before supporting incompatible historical payloads. const listeners = new Array() const { db } = yield* Database.Service + // Listener fan-out runs outside the publish path; fibers are bound to this layer's scope. + const forkListeners = yield* FiberSet.makeRuntime() const getOrCreate = (definition: Definition) => Effect.gen(function* () { @@ -123,6 +137,155 @@ export const layerWith = (options?: LayerOptions) => }), ) + const wakeDurable = (aggregateID: string) => + Effect.forEach( + pubsub.durable.get(aggregateID) ?? [], + (wake) => PubSub.publish(wake, undefined), + { discard: true }, + ) + + /** Transaction-scoped single durable event commit: seq allocation, owner checks, projectors, UPSERT + INSERT. */ + function commitDurableEventInner( + definition: Definition, + event: Payload, + input?: { + readonly seq: number + readonly aggregateID: string + readonly ownerID?: string + readonly strictOwner?: boolean + }, + commit?: (seq: number) => Effect.Effect, + ) { + return Effect.gen(function* () { + const durable = definition?.durable + if (!durable) return undefined + const aggregateID = (event.data as Record)[durable.aggregate] + if (typeof aggregateID !== "string") { + yield* Effect.die( + new InvalidDurableEventError({ + type: event.type, + message: `Expected string aggregate field ${durable.aggregate}`, + }), + ) + return undefined + } + if (input && input.aggregateID !== aggregateID) { + yield* Effect.die( + new InvalidDurableEventError({ + type: event.type, + message: `Aggregate mismatch: expected ${input.aggregateID}, got ${aggregateID}`, + }), + ) + return undefined + } + const list = projectors.get(event.type) ?? [] + const row = yield* db + .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id }) + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .get() + .pipe(Effect.orDie) + const latest = row?.seq ?? -1 + const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record + if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) { + yield* Effect.die( + new InvalidDurableEventError({ + type: event.type, + message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`, + }), + ) + } + if (input && input.seq <= latest) { + const stored = yield* db + .select() + .from(EventTable) + .where(and(eq(EventTable.aggregate_id, aggregateID), eq(EventTable.seq, input.seq))) + .get() + .pipe(Effect.orDie) + if ( + stored?.id === event.id && + stored.type === versionedType(definition.type, durable.version) && + isDeepStrictEqual(stored.data, encoded) + ) { + if (input.ownerID && row?.ownerID == null) { + yield* db + .update(EventSequenceTable) + .set({ owner_id: input.ownerID }) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .run() + .pipe(Effect.orDie) + } + return undefined + } + yield* Effect.die( + new InvalidDurableEventError({ + type: event.type, + message: `Replay diverged at aggregate ${aggregateID} sequence ${input.seq}`, + }), + ) + } + if (input && row?.ownerID && row.ownerID !== input.ownerID) { + return undefined + } + const seq = input?.seq ?? latest + 1 + if (input && seq !== latest + 1) { + yield* Effect.die( + new InvalidDurableEventError({ + type: event.type, + message: `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`, + }), + ) + } + const stored = yield* db + .select({ aggregateID: EventTable.aggregate_id, seq: EventTable.seq }) + .from(EventTable) + .where(eq(EventTable.id, event.id)) + .get() + .pipe(Effect.orDie) + if (stored) + yield* Effect.die( + new InvalidDurableEventError({ + type: event.type, + message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`, + }), + ) + const committed = { + ...event, + durable: { aggregateID, seq, version: durable.version }, + } as Payload + for (const projector of list) { + yield* projector(committed) + } + if (commit) yield* commit(seq) + yield* db + .insert(EventSequenceTable) + .values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }]) + .onConflictDoUpdate({ + target: EventSequenceTable.aggregate_id, + set: { + seq, + ...(input?.ownerID && row?.ownerID == null ? { owner_id: input.ownerID } : {}), + }, + }) + .run() + .pipe(Effect.orDie) + yield* db + .insert(EventTable) + .values([ + { + id: event.id, + aggregate_id: aggregateID, + seq, + type: versionedType(definition.type, durable.version), + data: encoded, + }, + ]) + .run() + .pipe(Effect.orDie) + return { aggregateID, seq } + }) + } + function commitDurableEvent( definition: Definition, event: Payload, @@ -154,136 +317,20 @@ export const layerWith = (options?: LayerOptions) => }), ) } - const list = projectors.get(event.type) ?? [] return yield* Effect.uninterruptible( Effect.gen(function* () { const committed = yield* db - .transaction( - () => - Effect.gen(function* () { - const row = yield* db - .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id }) - .from(EventSequenceTable) - .where(eq(EventSequenceTable.aggregate_id, aggregateID)) - .get() - .pipe(Effect.orDie) - const latest = row?.seq ?? -1 - const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record< - string, - unknown - > - if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) { - yield* Effect.die( - new InvalidDurableEventError({ - type: event.type, - message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`, - }), - ) - } - if (input && input.seq <= latest) { - const stored = yield* db - .select() - .from(EventTable) - .where(and(eq(EventTable.aggregate_id, aggregateID), eq(EventTable.seq, input.seq))) - .get() - .pipe(Effect.orDie) - if ( - stored?.id === event.id && - stored.type === versionedType(definition.type, durable.version) && - isDeepStrictEqual(stored.data, encoded) - ) { - if (input.ownerID && row?.ownerID == null) { - yield* db - .update(EventSequenceTable) - .set({ owner_id: input.ownerID }) - .where(eq(EventSequenceTable.aggregate_id, aggregateID)) - .run() - .pipe(Effect.orDie) - } - return - } - yield* Effect.die( - new InvalidDurableEventError({ - type: event.type, - message: `Replay diverged at aggregate ${aggregateID} sequence ${input.seq}`, - }), - ) - } - if (input && row?.ownerID && row.ownerID !== input.ownerID) { - return - } - const seq = input?.seq ?? latest + 1 - if (input && seq !== latest + 1) { - yield* Effect.die( - new InvalidDurableEventError({ - type: event.type, - message: `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`, - }), - ) - } - const stored = yield* db - .select({ aggregateID: EventTable.aggregate_id, seq: EventTable.seq }) - .from(EventTable) - .where(eq(EventTable.id, event.id)) - .get() - .pipe(Effect.orDie) - if (stored) - yield* Effect.die( - new InvalidDurableEventError({ - type: event.type, - message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`, - }), - ) - const committed = { - ...event, - durable: { aggregateID, seq, version: durable.version }, - } as Payload - for (const projector of list) { - yield* projector(committed) - } - if (commit) yield* commit(seq) - yield* db - .insert(EventSequenceTable) - .values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }]) - .onConflictDoUpdate({ - target: EventSequenceTable.aggregate_id, - set: { - seq, - ...(input?.ownerID && row?.ownerID == null ? { owner_id: input.ownerID } : {}), - }, - }) - .run() - .pipe(Effect.orDie) - yield* db - .insert(EventTable) - .values([ - { - id: event.id, - aggregate_id: aggregateID, - seq, - type: versionedType(definition.type, durable.version), - data: encoded, - }, - ]) - .run() - .pipe(Effect.orDie) - return { aggregateID, seq } - }), - { behavior: "immediate" }, - ) + .transaction(() => commitDurableEventInner(definition, event, input, commit), { + behavior: "immediate", + }) .pipe(Effect.orDie) - if (committed) { - yield* Effect.forEach( - pubsub.durable.get(committed.aggregateID) ?? [], - (wake) => PubSub.publish(wake, undefined), - { discard: true }, - ) - } + if (committed) yield* wakeDurable(committed.aggregateID) return committed }), ) } } + return undefined }) } @@ -307,11 +354,11 @@ export const layerWith = (options?: LayerOptions) => version: definition.durable.version, }, } - yield* notify(event as Payload, true) + yield* notify(event as Payload) return event } } - yield* notify(event as Payload, false) + yield* notify(event as Payload) return event }) } @@ -324,12 +371,28 @@ export const layerWith = (options?: LayerOptions) => ), ) - function notify(event: Payload, isolateListeners: boolean) { + // Fan-out contract (P1: publish never blocks on listener execution). + // Listener callbacks run on forked fibers bound to this layer's scope: + // - Within one event, listeners run in registration order on that + // event's fiber (the snapshot is taken at notify time, so a listener + // added mid-publish sees the next event, never a partial one). + // - A failing or slow listener can neither fail the publish (every + // listener is wrapped in `observe`, which logs and swallows non- + // interrupt errors) nor delay the synchronous pubsub publishes below. + // - Cross-event listener ordering is NOT guaranteed: each event's + // fan-out is its own fiber, so an async listener may interleave with + // the next event's listeners. + // Consumers that need ordered, lossless delivery must use `subscribe` / + // `all` (synchronous FIFO pubsub in publish order) or the durable + // stream; `listen` is for synchronous side effects (GlobalBus.emit, + // SSE queue offer) and fire-and-forget work. All current listeners + // (EventV2Bridge, SSE handler, summary publisher, plugins, VCS/project + // watchers) are of this shape, so the fork is semantics-preserving. + function notify(event: Payload) { return Effect.gen(function* () { - yield* Effect.forEach( - listeners, - (listener) => (isolateListeners ? observe(event, listener) : listener(event)), - { discard: true }, + const snapshot = Array.from(listeners) + forkListeners( + Effect.forEach(snapshot, (listener) => observe(event, listener), { discard: true }), ) const typed = pubsub.typed.get(event.type) if (typed) yield* PubSub.publish(typed, event) @@ -359,6 +422,104 @@ export const layerWith = (options?: LayerOptions) => }) } + function publishMany(events: ReadonlyArray, options?: { readonly location?: Location.Ref }) { + return Effect.gen(function* () { + const serviceLocation = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service)) + const location = + options?.location ?? + (serviceLocation + ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID } + : undefined) + const entries = new Array<{ + definition: Definition + durable: NonNullable + event: Payload + commit?: PublishOptions["commit"] + }>() + let aggregateID: string | undefined + for (const entry of events) { + const definition = entry.definition + const durable = definition?.durable + if (!durable) + return yield* Effect.die( + new InvalidDurableEventError({ + type: definition.type, + message: "Batch events require a durable definition", + }), + ) + const event = { + id: entry.options?.id ?? ID.create(), + ...(entry.options?.metadata ? { metadata: entry.options.metadata } : {}), + type: definition.type, + ...(location ? { location } : {}), + data: entry.data, + } as Payload + const id = (event.data as Record)[durable.aggregate] + if (typeof id !== "string") + return yield* Effect.die( + new InvalidDurableEventError({ + type: event.type, + message: `Expected string aggregate field ${durable.aggregate}`, + }), + ) + if (aggregateID === undefined) aggregateID = id + if (id !== aggregateID) + return yield* Effect.die( + new InvalidDurableEventError({ + type: event.type, + message: `Batch events must belong to the same aggregate: expected ${aggregateID}, got ${id}`, + }), + ) + entries.push({ definition, durable, event, commit: entry.options?.commit }) + } + if (entries.length === 0) return [] as ReadonlyArray + const committed = yield* Effect.uninterruptible( + Effect.gen(function* () { + const results = yield* db + .transaction( + () => + Effect.gen(function* () { + const results = new Array<{ aggregateID: string; seq: number }>() + for (const entry of entries) { + // No replay input: seq is allocated contiguously from the latest sequence inside the transaction. + const result = yield* commitDurableEventInner( + entry.definition, + entry.event, + undefined, + entry.commit, + ) + if (result) results.push(result) + } + return results + }), + { behavior: "immediate" }, + ) + .pipe(Effect.orDie) + if (aggregateID !== undefined) yield* wakeDurable(aggregateID) + return results + }), + ) + const payloads = entries.flatMap((entry, index) => { + const result = committed[index] + if (!result) return [] + return [ + { + ...entry.event, + durable: { + aggregateID: result.aggregateID, + seq: result.seq, + version: entry.durable.version, + }, + } as Payload, + ] + }) + for (const payload of payloads) { + yield* notify(payload) + } + return payloads + }) + } + function replay( event: SerializedEvent, options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean }, @@ -382,17 +543,14 @@ export const layerWith = (options?: LayerOptions) => strictOwner: options?.strictOwner, }) if (committed && options?.publish) { - yield* notify( - { - ...payload, - durable: { - aggregateID: committed.aggregateID, - seq: committed.seq, - version: definition.durable.version, - }, + yield* notify({ + ...payload, + durable: { + aggregateID: committed.aggregateID, + seq: committed.seq, + version: definition.durable.version, }, - true, - ) + }) } } }) @@ -555,6 +713,7 @@ export const layerWith = (options?: LayerOptions) => return Service.of({ publish, + publishMany, subscribe, all: streamAll, durable, diff --git a/packages/core/test/event-batch.test.ts b/packages/core/test/event-batch.test.ts new file mode 100644 index 0000000000..fbc61137b3 --- /dev/null +++ b/packages/core/test/event-batch.test.ts @@ -0,0 +1,349 @@ +import { describe, expect } from "bun:test" +import { Deferred, Duration, Effect, Fiber, Layer, Option, Schema, Stream } from "effect" +import { EventV2 } from "@opencode-ai/core/event" +import { Database } from "@opencode-ai/core/database/database" +import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" +import { Location } from "@opencode-ai/core/location" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { Session } from "@opencode-ai/schema/session" +import { SessionV1 } from "@opencode-ai/schema/session-v1" +import { WorkspaceV2 } from "@opencode-ai/core/workspace" +import { eq } from "drizzle-orm" +import { location } from "./fixture/location" +import { testEffect } from "./lib/effect" + +const locationLayer = Layer.succeed( + Location.Service, + Location.Service.of( + location({ directory: AbsolutePath.make("project"), workspaceID: WorkspaceV2.ID.make("wrk_test") }), + ), +) + +const Message = EventV2.define({ + type: "batch.message", + durable: { + version: 1, + aggregate: "id", + }, + schema: { + id: Schema.String, + text: Schema.String, + }, +}) + +const OtherMessage = EventV2.define({ + type: "batch.other", + durable: { + version: 1, + aggregate: "id", + }, + schema: { + id: Schema.String, + text: Schema.String, + }, +}) + +const LiveMessage = EventV2.define({ + type: "batch.live", + schema: { + text: Schema.String, + }, +}) + +const DurableMessage = SessionV1.Event.MessageRemoved + +const eventLayer = Layer.mergeAll(EventV2.layerWith().pipe(Layer.provide(Database.defaultLayer)), Database.defaultLayer) +const it = testEffect(eventLayer.pipe(Layer.provideMerge(locationLayer))) +const itWithoutLocation = testEffect(eventLayer) + +const batch = (aggregateID: string, texts: string[]) => + texts.map((text) => ({ definition: Message, data: { id: aggregateID, text } })) + +const rows = (aggregateID: string) => + Effect.gen(function* () { + const { db } = yield* Database.Service + return yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, aggregateID)) + .orderBy(EventTable.seq) + .all() + .pipe(Effect.orDie) + }) + +describe("EventV2.publishMany", () => { + it.effect("produces the same final state as sequential publish", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const batchAggregate = EventV2.ID.create() + const singleAggregate = EventV2.ID.create() + + const batched = yield* events.publishMany(batch(batchAggregate, ["a", "b", "c"])) + yield* events.publish(Message, { id: singleAggregate, text: "a" }) + yield* events.publish(Message, { id: singleAggregate, text: "b" }) + yield* events.publish(Message, { id: singleAggregate, text: "c" }) + + const batchRows = yield* rows(batchAggregate) + const singleRows = yield* rows(singleAggregate) + const summarize = (list: Array<{ seq: number; type: string; data: Record }>) => + list.map(({ seq, type, data }) => ({ seq, type, text: (data as { text: string }).text })) + expect(summarize(batchRows)).toEqual(summarize(singleRows)) + expect( + batched.map((event) => [(event.data as { text: string }).text, event.durable?.seq]), + ).toEqual([ + ["a", 0], + ["b", 1], + ["c", 2], + ]) + }), + ) + + it.effect("assigns contiguous seq across batch boundaries", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + + yield* events.publish(Message, { id: aggregateID, text: "seed" }) + yield* events.publishMany(batch(aggregateID, ["a", "b"])) + yield* events.publishMany(batch(aggregateID, ["c"])) + yield* events.publish(Message, { id: aggregateID, text: "tail" }) + + expect((yield* rows(aggregateID)).map((row) => row.seq)).toEqual([0, 1, 2, 3, 4]) + }), + ) + + it.effect("runs projectors in entry order inside the transaction", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const received = new Array() + yield* events.project(Message, (event) => Effect.sync(() => received.push(event))) + const aggregateID = EventV2.ID.create() + + yield* events.publishMany(batch(aggregateID, ["a", "b", "c"])) + + expect(received.map((event) => [(event.data as { text: string }).text, event.durable?.seq])).toEqual([ + ["a", 0], + ["b", 1], + ["c", 2], + ]) + }), + ) + + it.effect("runs per-event commit hooks in order inside the transaction", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const commits = new Array() + const aggregateID = EventV2.ID.create() + + yield* events.publishMany( + batch(aggregateID, ["a", "b"]).map((entry, index) => ({ + ...entry, + options: { commit: (seq) => Effect.sync(() => commits.push(seq * 10 + index)) }, + })), + ) + + expect(commits).toEqual([0, 11]) + }), + ) + + it.effect("rolls back the whole batch when a commit hook fails", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + const exit = yield* events + .publishMany( + batch(aggregateID, ["a", "b", "c"]).map((entry, index) => ({ + ...entry, + options: index === 1 ? { commit: () => Effect.die("commit failed") } : undefined, + })), + ) + .pipe(Effect.exit) + + expect(String(exit)).toContain("commit failed") + expect(yield* rows(aggregateID)).toEqual([]) + }), + ) + + it.effect("notifies typed and wildcard subscribers once per event", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + const typed = yield* events.subscribe(Message).pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped) + const wildcard = yield* events.all().pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped) + yield* Effect.yieldNow + + yield* events.publishMany(batch(aggregateID, ["a", "b", "c"])) + + expect(Array.from(yield* Fiber.join(typed)).map((event) => [(event.data as { text: string }).text, event.durable?.seq])).toEqual([ + ["a", 0], + ["b", 1], + ["c", 2], + ]) + expect( + Array.from(yield* Fiber.join(wildcard)).map((event) => [(event.data as { text: string }).text, event.durable?.seq]), + ).toEqual([ + ["a", 0], + ["b", 1], + ["c", 2], + ]) + }), + ) + + it.live("does not block the publish path on a slow listener", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + yield* events.listen(() => Effect.never) + const aggregateID = EventV2.ID.create() + + const published = yield* events.publishMany(batch(aggregateID, ["a", "b"])).pipe( + Effect.timeoutOption(Duration.millis(250)), + ) + + expect(Option.isSome(published)).toBeTrue() + expect(yield* rows(aggregateID)).toHaveLength(2) + }), + ) + + it.effect("isolates listener defects while other listeners still receive events", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const received = new Array() + const arrived = yield* Deferred.make() + yield* events.listen(() => Effect.die("listener defect")) + yield* events.listen((event) => + Effect.sync(() => received.push(event.type)).pipe(Effect.andThen(Deferred.succeed(arrived, undefined))), + ) + const aggregateID = EventV2.ID.create() + + const published = yield* events.publishMany(batch(aggregateID, ["a", "b"])) + yield* Deferred.await(arrived) + + expect(published).toHaveLength(2) + expect(received).toEqual([Message.type, Message.type]) + }), + ) + + it.effect("rejects events from different aggregates", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const exit = yield* events + .publishMany([ + { definition: Message, data: { id: "agg-a", text: "a" } }, + { definition: Message, data: { id: "agg-b", text: "b" } }, + ]) + .pipe(Effect.exit) + + expect(String(exit)).toContain("Batch events must belong to the same aggregate") + expect(yield* rows("agg-a")).toEqual([]) + expect(yield* rows("agg-b")).toEqual([]) + }), + ) + + it.effect("rejects live-only definitions", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const exit = yield* events + .publishMany([{ definition: LiveMessage, data: { text: "live" } }]) + .pipe(Effect.exit) + + expect(String(exit)).toContain("Batch events require a durable definition") + }), + ) + + it.effect("supports mixed definitions sharing one aggregate", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + + const published = yield* events.publishMany([ + { definition: Message, data: { id: aggregateID, text: "a" } }, + { definition: OtherMessage, data: { id: aggregateID, text: "b" } }, + ]) + + expect(published.map((event) => [event.type, event.durable?.seq])).toEqual([ + [Message.type, 0], + [OtherMessage.type, 1], + ]) + expect((yield* rows(aggregateID)).map((row) => [row.type, row.seq])).toEqual([ + [EventV2.versionedType(Message.type, 1), 0], + [EventV2.versionedType(OtherMessage.type, 1), 1], + ]) + }), + ) + + it.effect("returns no payloads for an empty batch", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + expect(yield* events.publishMany([])).toEqual([]) + }), + ) + + it.effect("keeps replay and readAfter compatible with batch-published events", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = Session.ID.create() + const data = (text: string) => ({ + sessionID: aggregateID, + messageID: SessionV1.MessageID.ascending(`msg_${text}`), + }) + yield* events.publishMany([ + { definition: DurableMessage, data: data("a") }, + { definition: DurableMessage, data: data("b") }, + { definition: DurableMessage, data: data("c") }, + ]) + + const fiber = yield* events + .durable({ aggregateID, after: 2 }) + .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped) + yield* events.publishMany([ + { definition: DurableMessage, data: data("d") }, + { definition: DurableMessage, data: data("e") }, + ]) + const tail = Array.from(yield* Fiber.join(fiber)) + + expect(tail.map((event) => [(event.data as { messageID: string }).messageID, event.durable?.seq])).toEqual([ + [data("d").messageID, 3], + [data("e").messageID, 4], + ]) + const replayed = yield* events.replayAll([ + ...(yield* rows(aggregateID)).map((row) => ({ + id: row.id, + type: row.type, + seq: row.seq, + aggregateID, + data: row.data, + })), + ]) + expect(replayed).toBe(aggregateID) + }), + ) + + it.effect("stays sequence-safe under concurrent batch publication", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + + const fiberA = yield* events.publishMany(batch(aggregateID, ["a", "b"])).pipe(Effect.forkScoped) + const fiberB = yield* events.publishMany(batch(aggregateID, ["c", "d"])).pipe(Effect.forkScoped) + yield* Fiber.join(fiberA) + yield* Fiber.join(fiberB) + + expect((yield* rows(aggregateID)).map((row) => row.seq)).toEqual([0, 1, 2, 3]) + }), + ) + + itWithoutLocation.effect("attaches an explicit location to every batch event", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + const explicit = Location.Ref.make({ + directory: AbsolutePath.make("explicit"), + workspaceID: WorkspaceV2.ID.make("wrk_explicit"), + }) + + const published = yield* events.publishMany(batch(aggregateID, ["a", "b"]), { location: explicit }) + + expect(published.map((event) => event.location)).toEqual([explicit, explicit]) + }), + ) +}) diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index e2b2a5df04..7034c1cccc 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -242,7 +242,7 @@ describe("EventV2", () => { }), ) - it.effect("runs listeners inline after projectors", () => + it.effect("runs listeners after projectors", () => Effect.gen(function* () { const events = yield* EventV2.Service const received = new Array() @@ -285,32 +285,61 @@ describe("EventV2", () => { }), ) - it.effect("preserves observer interruption", () => + it.effect("keeps the publish path uninterrupted by an interrupting listener", () => Effect.gen(function* () { const events = yield* EventV2.Service const { db } = yield* Database.Service yield* events.listen(() => Effect.interrupt) - const exit = yield* events.publish(SyncMessage, { id: "interrupted", text: "hello" }).pipe(Effect.exit) + const event = yield* events.publish(SyncMessage, { id: "interrupted", text: "hello" }) const committed = yield* db - .select({ id: EventTable.id }) + .select({ id: EventTable.id, seq: EventTable.seq }) .from(EventTable) .where(eq(EventTable.aggregate_id, "interrupted")) .get() .pipe(Effect.orDie) - expect(Exit.isFailure(exit) && Cause.hasInterrupts(exit.cause)).toBeTrue() - expect(committed).toBeDefined() + expect(event.durable?.seq).toBe(0) + expect(committed).toEqual({ id: event.id, seq: 0 }) }), ) - it.effect("keeps live-only listener defects fail-fast", () => + it.effect("isolates live-only listener defects", () => Effect.gen(function* () { const events = yield* EventV2.Service const defect = new Error("listener defect") yield* events.listen(() => Effect.die(defect)) - expect(yield* events.publish(Message, { text: "hello" }).pipe(Effect.catchDefect(Effect.succeed))).toBe(defect) + expect(yield* events.publish(Message, { text: "hello" }).pipe(Effect.isSuccess)).toBeTrue() + }), + ) + + it.effect("isolates listener defects and preserves pubsub order across events", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const received = new Array() + const fiber = yield* events.all().pipe( + Stream.take(2), + Stream.runForEach((event) => Effect.sync(() => received.push((event.data as { text: string }).text))), + Effect.forkScoped, + ) + yield* Effect.yieldNow + yield* events.listen(() => Effect.die(new Error("listener defect"))) + yield* events.listen((event) => + Effect.sync(() => received.push(`L:${(event.data as { text: string }).text}`)), + ) + + yield* events.publish(Message, { text: "one" }) + yield* events.publish(Message, { text: "two" }) + yield* Fiber.join(fiber) + + // The dying listener neither blocks the publish nor stops the other + // listener. Cross-event listener ordering is NOT guaranteed (each + // event's fan-out runs on its own fiber), so only assert what the + // contract guarantees: synchronous pubsub FIFO order and per-listener + // isolation. See the fan-out contract comment in event.ts `notify`. + expect(received.filter((value) => !value.startsWith("L:"))).toEqual(["one", "two"]) + expect(received.filter((value) => value.startsWith("L:")).sort()).toEqual(["L:one", "L:two"]) }), ) diff --git a/packages/opencode/src/session/session.ts b/packages/opencode/src/session/session.ts index fbed36df90..b3c65515f4 100644 --- a/packages/opencode/src/session/session.ts +++ b/packages/opencode/src/session/session.ts @@ -730,32 +730,49 @@ export const layer: Layer.Layer< const msgs = yield* messages({ sessionID: input.sessionID }) const idMap = new Map() - for (const msg of msgs) { - if (input.messageID && msg.info.id >= input.messageID) break - const newID = MessageID.ascending() - idMap.set(msg.info.id, newID) - - const parentID = msg.info.role === "assistant" && msg.info.parentID ? idMap.get(msg.info.parentID) : undefined - const cloned = yield* updateMessage({ - ...msg.info, - sessionID: session.id, - id: newID, - ...(parentID && { parentID }), - }) - - for (const part of msg.parts) { - const p: SessionV1.Part = { - ...part, - id: PartID.ascending(), - messageID: cloned.id, - sessionID: session.id, - } - if (p.type === "compaction" && p.tail_start_id) { - p.tail_start_id = idMap.get(p.tail_start_id) - } - yield* updatePart(p) - } - } + // Every updateMessage/updatePart publishes a durable event, and each + // publish opens its own db transaction — a large session forks in + // thousands of commits. The effect-drizzle adapter turns nested + // `db.transaction` calls into savepoints on the outer transaction's + // connection (reads inside the transaction see the uncommitted writes, + // so seq allocation stays consecutive), so wrapping the copy loop in a + // single transaction converges the fork to one commit while keeping the + // per-event publish semantics (projector order, wake order) unchanged. + yield* db + .transaction( + () => + Effect.gen(function* () { + for (const msg of msgs) { + if (input.messageID && msg.info.id >= input.messageID) break + const newID = MessageID.ascending() + idMap.set(msg.info.id, newID) + + const parentID = + msg.info.role === "assistant" && msg.info.parentID ? idMap.get(msg.info.parentID) : undefined + const cloned = yield* updateMessage({ + ...msg.info, + sessionID: session.id, + id: newID, + ...(parentID && { parentID }), + }) + + for (const part of msg.parts) { + const p: SessionV1.Part = { + ...part, + id: PartID.ascending(), + messageID: cloned.id, + sessionID: session.id, + } + if (p.type === "compaction" && p.tail_start_id) { + p.tail_start_id = idMap.get(p.tail_start_id) + } + yield* updatePart(p) + } + } + }), + { behavior: "immediate" }, + ) + .pipe(Effect.orDie) return session }) diff --git a/packages/opencode/test/session/fork-batch.test.ts b/packages/opencode/test/session/fork-batch.test.ts new file mode 100644 index 0000000000..51633bd9dd --- /dev/null +++ b/packages/opencode/test/session/fork-batch.test.ts @@ -0,0 +1,271 @@ +import { describe, expect } from "bun:test" +import { Database as BunDatabase, type SQLQueryBindings } from "bun:sqlite" +import { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionV1 } from "@opencode-ai/core/v1/session" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { Context, Effect, Fiber, Layer, Scope, Semaphore, Stream } from "effect" +import * as Client from "effect/unstable/sql/SqlClient" +import type { Connection } from "effect/unstable/sql/SqlConnection" +import { SqlError, classifySqliteError } from "effect/unstable/sql/SqlError" +import * as Statement from "effect/unstable/sql/Statement" +import * as Reactivity from "effect/unstable/reactivity/Reactivity" +import { Session as SessionNs } from "@/session/session" +import { MessageID, PartID } from "../../src/session/schema" +import { testInstanceStoreLayer } from "../fixture/fixture" +import { testEffect } from "../lib/effect" +import { Storage } from "@/storage/storage" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { BackgroundJob } from "@/background/job" +import { EventV2Bridge } from "@/event-v2-bridge" + +interface SqlCounter { + begins: number + commits: number + savepoints: number +} + +const counter: SqlCounter = { begins: 0, commits: 0, savepoints: 0 } + +// The Database layer's sqlite client is a closed graph (its native provider +// cannot be overridden from outside), so this test builds its own SqlClient +// mirroring the driver's `make`, wrapping the native so real BEGIN/COMMIT/ +// SAVEPOINT statements are countable. +// +// This duplicates ~85 lines of packages/core/src/database/sqlite.bun.ts `make` +// (run/runValues/connection/semaphore/transactionAcquirer). Tracked debt: if +// sqlite.bun.ts exposed a provider seam (an injectable native Database, or a +// `make({ native })` overload), this test could reuse the production client and +// the copy would collapse. Until then the duplication is intentional and must +// be kept in sync with sqlite.bun.ts `run`/`runValues`. +const countingClientLayer = Layer.effect( + Client.SqlClient, + Effect.gen(function* () { + const native = new BunDatabase(":memory:") + native.run("PRAGMA journal_mode = WAL;") + const counting = new Proxy(native, { + get(target, prop) { + if (prop === "query") { + return (sql: string) => { + if (/^\s*begin\b/i.test(sql)) counter.begins++ + else if (/^\s*commit\b/i.test(sql)) counter.commits++ + else if (/^\s*savepoint\b/i.test(sql)) counter.savepoints++ + return target.query(sql) + } + } + return Reflect.get(target, prop) + }, + }) as BunDatabase + + const compiler = Statement.makeCompilerSqlite(undefined) + const run = (query: string, params: ReadonlyArray = []) => + Effect.withFiber>, SqlError>((fiber) => { + const statement = counting.query(query) + // @ts-ignore bun-types missing safeIntegers method + statement.safeIntegers(Context.get(fiber.context, Client.SafeIntegers)) + try { + return Effect.succeed((statement.all(...(params as SQLQueryBindings[])) ?? []) as Array>) + } catch (cause) { + return Effect.fail( + new SqlError({ + reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }), + }), + ) + } + }) + const runValues = (query: string, params: ReadonlyArray = []) => + Effect.withFiber, SqlError>((fiber) => { + const statement = counting.query(query) + // @ts-ignore bun-types missing safeIntegers method + statement.safeIntegers(Context.get(fiber.context, Client.SafeIntegers)) + try { + return Effect.succeed((statement.values(...(params as SQLQueryBindings[])) ?? []) as Array) + } catch (cause) { + return Effect.fail( + new SqlError({ + reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }), + }), + ) + } + }) + const connection: Connection = { + execute(query, params, transformRows) { + return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params) + }, + executeRaw(query, params) { + return run(query, params) + }, + executeValues(query, params) { + return runValues(query, params) + }, + executeUnprepared(query, params, transformRows) { + return this.execute(query, params, transformRows) + }, + executeStream() { + return Stream.die("executeStream not implemented") + }, + } + const semaphore = yield* Semaphore.make(1) + const acquirer = semaphore.withPermits(1)(Effect.succeed(connection)) + const transactionAcquirer = Effect.uninterruptibleMask((restore) => { + const fiber = Fiber.getCurrent()! + const scope = Context.getUnsafe(fiber.context, Scope.Scope) + return Effect.as( + Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))), + connection, + ) + }) + return yield* Client.make({ + acquirer, + compiler, + transactionAcquirer, + spanAttributes: [["db.system.name", "sqlite"]], + }) + }), +) + +const dbLayer = Database.layer.pipe( + Layer.provide(countingClientLayer.pipe(Layer.provide(Reactivity.layer))), +) +const eventV2Layer = EventV2.layer.pipe(Layer.provide(dbLayer)) +const eventV2BridgeLayer = EventV2Bridge.layer.pipe(Layer.provide(eventV2Layer)) +const projectorLayer = SessionProjector.layer.pipe(Layer.provide(eventV2Layer), Layer.provide(dbLayer)) + +const it = testEffect( + Layer.mergeAll( + SessionNs.layer.pipe( + Layer.provide(Storage.defaultLayer), + Layer.provide(dbLayer), + Layer.provide(eventV2BridgeLayer), + Layer.provide(projectorLayer), + Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: false })), + Layer.provide(BackgroundJob.defaultLayer), + ), + CrossSpawnSpawner.defaultLayer, + testInstanceStoreLayer, + ), +) + +const userInfo = (sessionID: string, id: string) => + ({ + id, + sessionID, + role: "user", + time: { created: Date.now() }, + agent: "user", + model: { providerID: "test", modelID: "test" }, + }) as SessionV1.Info + +const assistantInfo = (sessionID: string, id: string, parentID: string) => + ({ + id, + sessionID, + role: "assistant", + time: { created: Date.now() }, + parentID, + modelID: "test", + providerID: "test", + mode: "", + agent: "assistant", + path: { cwd: "/", root: "/" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + }) as SessionV1.Info + +const textPart = (sessionID: string, messageID: string, text: string) => + ({ + id: PartID.ascending(), + sessionID, + messageID, + type: "text", + text, + }) as SessionV1.Part + +describe("Session.fork", () => { + it.instance("fork result is equivalent: message/part counts, parentID chain, compaction tail_start_id", () => + Effect.gen(function* () { + const session = yield* SessionNs.Service + const original = yield* Effect.acquireRelease(session.create({ title: "fork-source" }), (info) => + session.remove(info.id).pipe(Effect.ignore), + ) + + const m1 = MessageID.ascending() + const m2 = MessageID.ascending() + const m3 = MessageID.ascending() + yield* session.updateMessage(userInfo(original.id, m1)) + yield* session.updateMessage(assistantInfo(original.id, m2, m1)) + yield* session.updateMessage(userInfo(original.id, m3)) + yield* session.updatePart(textPart(original.id, m1, "hello")) + yield* session.updatePart(textPart(original.id, m2, "world")) + yield* session.updatePart({ + id: PartID.ascending(), + sessionID: original.id, + messageID: m3, + type: "compaction", + auto: true, + tail_start_id: m1, + }) + + const fork = yield* Effect.acquireRelease(session.fork({ sessionID: original.id }), (info) => + session.remove(info.id).pipe(Effect.ignore), + ) + + const source = yield* session.messages({ sessionID: original.id }) + const target = yield* session.messages({ sessionID: fork.id }) + + expect(target.length).toBe(source.length) + expect(target.length).toBe(3) + + const [f1, f2, f3] = target + expect(f1.info.id).not.toBe(m1) + expect(f1.parts.map((p) => p.type)).toEqual(["text"]) + expect((f1.parts[0] as SessionV1.TextPart).text).toBe("hello") + // parentID chain maps through the idMap + expect((f2.info as SessionV1.Assistant).parentID).toBe(f1.info.id) + expect(f2.parts.map((p) => p.type)).toEqual(["text"]) + expect((f2.parts[0] as SessionV1.TextPart).text).toBe("world") + // compaction tail_start_id maps to the forked message id + const compaction = f3.parts.find((p) => p.type === "compaction") + expect(compaction?.type).toBe("compaction") + if (compaction?.type === "compaction") expect(compaction.tail_start_id).toBe(f1.info.id) + }), + ) + + it.instance("fork copies the whole session in one batch transaction regardless of session size", () => + Effect.gen(function* () { + const session = yield* SessionNs.Service + const original = yield* Effect.acquireRelease(session.create({ title: "fork-source" }), (info) => + session.remove(info.id).pipe(Effect.ignore), + ) + + const messageCount = 30 + for (let i = 0; i < messageCount; i++) { + const id = MessageID.ascending() + yield* session.updateMessage(userInfo(original.id, id)) + yield* session.updatePart(textPart(original.id, id, `part ${i}-a`)) + yield* session.updatePart(textPart(original.id, id, `part ${i}-b`)) + } + + counter.begins = 0 + counter.commits = 0 + counter.savepoints = 0 + + const fork = yield* Effect.acquireRelease(session.fork({ sessionID: original.id }), (info) => + session.remove(info.id).pipe(Effect.ignore), + ) + + // One BEGIN/COMMIT for the fork session's Created event, one for the + // batch copy transaction — never one per message/part (that would be + // 91 BEGINs for 30 messages with 2 parts each). + expect(counter.begins).toBe(2) + expect(counter.commits).toBe(2) + // Each per-event publish inside the batch becomes a savepoint. + expect(counter.savepoints).toBe(messageCount * 3) + + const target = yield* session.messages({ sessionID: fork.id }) + expect(target.length).toBe(messageCount) + for (const msg of target) expect(msg.parts.length).toBe(2) + }), + ) +})