From c03058a06af75e8ce0805d2c79c78de40ed35920 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Mon, 14 Sep 2026 17:00:46 -0700 Subject: [PATCH] Let a durable event tail re-read on a tick, for a subscriber another process writes past. The wake that follows a commit is published in-process, so a tail on one process never sees events a second process appended to the same log. The tick is off unless a composition root asks for it, and OPENCODE_EVENT_POLL_MS overrides either way. --- packages/core/src/event.ts | 47 +++++++++++++++++++++++++++-- packages/core/src/flag/flag.ts | 6 ++++ packages/core/test/event.test.ts | 51 +++++++++++++++++++++++++++++++- 3 files changed, 101 insertions(+), 3 deletions(-) diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index c92ac0ac2ce3..3a64217f6009 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -1,11 +1,12 @@ export * as EventV2 from "./event" -import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect" +import { Cause, Context, Duration, Effect, Layer, Option, PubSub, Queue, 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, inArray } from "drizzle-orm" import { Database } from "./database/database" import { EventSequenceTable, EventTable } from "./event/sql" +import { Flag } from "./flag/flag" import { Location } from "./location" import { makeGlobalNode } from "./effect/app-node" import { isDeepStrictEqual } from "node:util" @@ -165,6 +166,27 @@ export const allBounded = (events: Interface, capacity: number) => export interface LayerOptions { readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect + /** How often a durable tail re-reads on its own, on top of the in-process wake. The wake only + * fires for commits made in THIS process, so without a tick a subscriber cannot see events + * another process appended: a standalone worker's whole turn is invisible to a tail on the + * HTTP server. + * Zero (the default) relies on the wake alone, which is the whole story for a deployment that + * runs in one process. */ + readonly livePollInterval?: Duration.Input +} + +/** Chosen to be well under what a person notices in a transcript while staying one cheap indexed + * read per subscribed session. In-process commits still wake instantly; this only catches what the + * wake cannot see, so it is worth its cost only where another process writes: see `pollingNode`. */ +const DEFAULT_LIVE_POLL = Duration.seconds(1) + +// An operator's override, in milliseconds, for either node. Read at layer build rather than at +// import, so a test or a CLI that sets it late still gets it. +const configuredPoll = (fallback: Duration.Input): Duration.Input => { + const raw = Flag.OPENCODE_EVENT_POLL_MS + if (raw === undefined) return fallback + const millis = Number(raw) + return Number.isFinite(millis) && millis >= 0 ? millis : fallback } export const layerWith = (options?: LayerOptions) => @@ -595,7 +617,18 @@ export const layerWith = (options?: LayerOptions) => ), ) const historical = yield* read - const live = Stream.fromSubscription(wakes).pipe( + // Wake on either an in-process commit or the tick. A tick that finds nothing new reads + // zero rows and emits nothing, so an idle subscriber costs one indexed query per + // period. + const pollInterval = configuredPoll(options?.livePollInterval ?? Duration.zero) + const woken = Stream.fromSubscription(wakes) + // haltStrategy "left" keeps the wake stream as what ends the tail. The tick never ends, + // so the default ("both") would leave a subscriber hanging past the layer's own + // PubSub.shutdown, still reading from a database being torn down. + const source = Duration.isZero(Duration.fromInputUnsafe(pollInterval)) + ? woken + : woken.pipe(Stream.merge(Stream.tick(pollInterval), { haltStrategy: "left" })) + const live = source.pipe( Stream.mapEffect(() => read), Stream.flattenIterable, ) @@ -636,3 +669,13 @@ export const layerWith = (options?: LayerOptions) => const layer = layerWith() export const node = makeGlobalNode({ service: Service, layer: layer, deps: [Database.node] }) + +// For a deployment where the process serving subscribers is not the only one appending: a serve +// process driving sessions on standalone workers. The wake is published in-process, so without the +// tick a worker's whole turn is invisible to a tail on the server. Composition roots that know they +// are in that shape swap this in for `node`; everything else keeps the wake alone. +export const pollingNode = makeGlobalNode({ + service: Service, + layer: layerWith({ livePollInterval: DEFAULT_LIVE_POLL }), + deps: [Database.node], +}) diff --git a/packages/core/src/flag/flag.ts b/packages/core/src/flag/flag.ts index a0eb78a13e2a..b8a5533e7a0e 100644 --- a/packages/core/src/flag/flag.ts +++ b/packages/core/src/flag/flag.ts @@ -75,4 +75,10 @@ export const Flag = { get OPENCODE_CLIENT() { return process.env["OPENCODE_CLIENT"] ?? "cli" }, + // Milliseconds between a live event tail's own re-reads, `0` to rely on the in-process wake + // alone. Only a deployment where another process appends to the log needs it, so it is off unless + // the composition root asks for it; this overrides either way. + get OPENCODE_EVENT_POLL_MS() { + return process.env["OPENCODE_EVENT_POLL_MS"] + }, } diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index e4329a2dde98..6877520fc48c 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "bun:test" -import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, Stream } from "effect" +import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, Scope, Stream } from "effect" import { EventV2 } from "@opencode-ai/core/event" import { Event } from "@opencode-ai/schema/event" import { Session } from "@opencode-ai/schema/session" @@ -1097,6 +1097,55 @@ describe("EventV2", () => { }), ) + it.live("tails another writer's events only when the layer asked for a tick", () => + Effect.gen(function* () { + // A second service over the SAME database with its OWN pubsub: exactly what a standalone + // worker is to the HTTP server. Its commits cannot reach the reader's wake map, so a tail + // that only listens for wakes sits there forever. + const tailSees = (options: EventV2.LayerOptions) => + Effect.gen(function* () { + const reader = yield* EventV2.Service.pipe(Effect.provide(EventV2.layerWith(options))) + const writer = yield* EventV2.Service.pipe(Effect.provide(EventV2.layerWith())) + const aggregateID = Session.ID.create() + const tail = yield* reader + .durable({ aggregateID, after: -1 }) + .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + yield* Effect.sleep("50 millis") + yield* writer.publish(DurableMessage, durableData(aggregateID, "from-elsewhere")) + const collected = yield* Fiber.join(tail).pipe(Effect.timeoutOption("1 second")) + return Option.isSome(collected) + }) + + expect(yield* tailSees({ livePollInterval: "50 millis" })).toBe(true) + // Off unless asked for. One process wakes its own subscribers, so a tick there is a query per + // second per subscribed session for events that cannot exist. + expect(yield* tailSees({})).toBe(false) + }), + ) + + it.live("ends a durable tail when the event layer is released under it", () => + Effect.gen(function* () { + const aggregateID = Session.ID.create() + // The layer needs its own closeable scope, and the consumer has to be forked into a scope that + // OUTLIVES it. Fork into the test's own scope and closing that scope interrupts the consumer + // before the finalizer runs, so a hung tail and a killed one look the same. + const outer = yield* Effect.scope + const layerScope = yield* Scope.make() + const ctx = yield* Layer.buildWithScope(EventV2.layerWith(), layerScope) + const events = yield* EventV2.Service.pipe(Effect.provideContext(ctx as never)) + yield* events.publish(DurableMessage, durableData(aggregateID, "seed")) + + const tail = yield* events.durable({ aggregateID, after: -1 }).pipe(Stream.runDrain, Effect.forkIn(outer)) + yield* Effect.sleep("150 millis") + yield* Scope.close(layerScope, Exit.void) + + // The tick never ends on its own, so it must not become what holds the tail open: a subscriber + // would outlive the layer and keep reading a database being torn down. + const settled = yield* Fiber.await(tail).pipe(Effect.timeout("5 seconds"), Effect.exit) + expect(Exit.isSuccess(settled)).toBe(true) + }), + ) + it.effect("remove clears durable event sequence", () => Effect.gen(function* () { const events = yield* EventV2.Service