fix(bus): acquire PubSub subscription eagerly to close /event race (#27959)

Author: kitlangton

packages/opencode/src/bus/index.ts

@@ -37,8 +37,16 @@ export interface Interface {
     properties: BusProperties<D>,
     options?: { id?: string },
   ) => Effect.Effect<void>
-  readonly subscribe: <D extends BusEvent.Definition>(def: D) => Stream.Stream<Payload<D>>
-  readonly subscribeAll: () => Stream.Stream<Payload>
+  // subscribe / subscribeAll are eager: the underlying PubSub subscription is
+  // acquired in the caller's Scope at `yield*` time. Any publish after the
+  // yield is delivered, even if stream consumption starts later. The previous
+  // Stream-returning shape acquired the subscription lazily on first pull,
+  // opening a race window during which publishes were lost — see
+  // test/bus/bus-effect.test.ts RACE tests.
+  readonly subscribe: <D extends BusEvent.Definition>(
+    def: D,
+  ) => Effect.Effect<Stream.Stream<Payload<D>>, never, Scope.Scope>
+  readonly subscribeAll: () => Effect.Effect<Stream.Stream<Payload>, never, Scope.Scope>
   readonly subscribeCallback: <D extends BusEvent.Definition>(
     def: D,
     callback: (event: Payload<D>) => unknown,
@@ -109,26 +117,26 @@ export const layer = Layer.effect(
       })
     }
 
-    function subscribe<D extends BusEvent.Definition>(def: D): Stream.Stream<Payload<D>> {
-      log.info("subscribing", { type: def.type })
-      return Stream.unwrap(
-        Effect.gen(function* () {
-          const s = yield* InstanceState.get(state)
-          const ps = yield* getOrCreate(s, def)
-          return Stream.fromPubSub(ps)
-        }),
-      ).pipe(Stream.ensuring(Effect.sync(() => log.info("unsubscribing", { type: def.type }))))
-    }
+    const subscribe = <D extends BusEvent.Definition>(
+      def: D,
+    ): Effect.Effect<Stream.Stream<Payload<D>>, never, Scope.Scope> =>
+      Effect.gen(function* () {
+        log.info("subscribing", { type: def.type })
+        const s = yield* InstanceState.get(state)
+        const ps = yield* getOrCreate(s, def)
+        const subscription = yield* PubSub.subscribe(ps)
+        yield* Effect.addFinalizer(() => Effect.sync(() => log.info("unsubscribing", { type: def.type })))
+        return Stream.fromSubscription(subscription)
+      })
 
-    function subscribeAll(): Stream.Stream<Payload> {
-      log.info("subscribing", { type: "*" })
-      return Stream.unwrap(
-        Effect.gen(function* () {
-          const s = yield* InstanceState.get(state)
-          return Stream.fromPubSub(s.wildcard)
-        }),
-      ).pipe(Stream.ensuring(Effect.sync(() => log.info("unsubscribing", { type: "*" }))))
-    }
+    const subscribeAll = (): Effect.Effect<Stream.Stream<Payload>, never, Scope.Scope> =>
+      Effect.gen(function* () {
+        log.info("subscribing", { type: "*" })
+        const s = yield* InstanceState.get(state)
+        const subscription = yield* PubSub.subscribe(s.wildcard)
+        yield* Effect.addFinalizer(() => Effect.sync(() => log.info("unsubscribing", { type: "*" })))
+        return Stream.fromSubscription(subscription)
+      })
 
     function on<T>(pubsub: PubSub.PubSub<T>, type: string, callback: (event: T) => unknown) {
       return Effect.gen(function* () {

packages/opencode/src/plugin/index.ts

@@ -243,7 +243,7 @@ export const layer = Layer.effect(
         }
 
         // Subscribe to bus events, fiber interrupted when scope closes
-        yield* bus.subscribeAll().pipe(
+        yield* (yield* bus.subscribeAll()).pipe(
           Stream.runForEach((input) =>
             Effect.sync(() => {
               for (const hook of hooks) {

packages/opencode/src/project/project.ts

@@ -425,7 +425,7 @@ export const layer: Layer.Layer<
 
     const initState = yield* InstanceState.make(
       Effect.fn("Project.initState")(function* (ctx) {
-        yield* bus.subscribe(Command.Event.Executed).pipe(
+        yield* (yield* bus.subscribe(Command.Event.Executed)).pipe(
           Stream.runForEach((payload) =>
             payload.properties.name === Command.Default.INIT ? setInitialized(ctx.project.id) : Effect.void,
           ),

packages/opencode/src/project/vcs.ts

@@ -298,7 +298,7 @@ export const layer: Layer.Layer<Service, never, Git.Service | Bus.Service> = Lay
         const value = { current, root }
         log.info("initialized", { branch: value.current, default_branch: value.root?.name })
 
-        yield* bus.subscribe(FileWatcher.Event.Updated).pipe(
+        yield* (yield* bus.subscribe(FileWatcher.Event.Updated)).pipe(
           Stream.filter((evt) => evt.properties.file.endsWith("HEAD")),
           Stream.runForEach((_evt) =>
             Effect.gen(function* () {

packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts

@@ -20,10 +20,11 @@ function eventData(data: unknown): Sse.Event {
 
 function eventResponse(bus: Bus.Interface) {
   return Effect.gen(function* () {
-    const context = yield* Effect.context()
-
-    const events = bus.subscribeAll().pipe(
-      Stream.provideContext(context),
+    // Subscribe eagerly: the bus subscription is acquired in the request scope
+    // at this yield, so any publish from now on is queued for the body-pump
+    // fiber to drain — closing the race where Stream.concat(server.connected,
+    // lazy-subscribe) used to drop publishes in the prefix-consume window.
+    const events = (yield* bus.subscribeAll()).pipe(
       Stream.takeUntil((event) => event.type === Bus.InstanceDisposed.type),
     )
     const heartbeat = Stream.tick("10 seconds").pipe(

packages/opencode/src/share/share-next.ts

@@ -168,16 +168,20 @@ export const layer = Layer.effect(
           fn: (evt: { properties: any }) => Effect.Effect<void, unknown>,
         ) =>
           bus.subscribe(def as never).pipe(
-            Stream.runForEach((evt) =>
-              fn(evt).pipe(
-                Effect.catchCause((cause) =>
-                  Effect.sync(() => {
-                    log.error("share subscriber failed", { type: def.type, cause })
-                  }),
+            Effect.flatMap((stream) =>
+              stream.pipe(
+                Stream.runForEach((evt) =>
+                  fn(evt).pipe(
+                    Effect.catchCause((cause) =>
+                      Effect.sync(() => {
+                        log.error("share subscriber failed", { type: def.type, cause })
+                      }),
+                    ),
+                  ),
                 ),
+                Effect.forkScoped,
               ),
             ),
-            Effect.forkScoped,
           )
 
         yield* watch(Session.Event.Updated, (evt) =>

packages/opencode/test/bus/bus-effect.test.ts

@@ -44,7 +44,7 @@ describe("Bus (Effect-native)", () => {
       const done = yield* Deferred.make<void>()
       const ready = yield* Latch.make()
 
-      yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
+      yield* Stream.runForEach(yield* bus.subscribe(TestEvent.Ping), (evt) =>
         Effect.gen(function* () {
           if (evt.properties.value < 0) {
             yield* ready.open
@@ -71,7 +71,7 @@ describe("Bus (Effect-native)", () => {
       const done = yield* Deferred.make<void>()
       const ready = yield* Latch.make()
 
-      yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
+      yield* Stream.runForEach(yield* bus.subscribe(TestEvent.Ping), (evt) =>
         Effect.gen(function* () {
           if (evt.properties.value < 0) {
             yield* ready.open
@@ -98,7 +98,7 @@ describe("Bus (Effect-native)", () => {
       const done = yield* Deferred.make<void>()
       const ready = yield* Latch.make()
 
-      yield* Stream.runForEach(bus.subscribeAll(), (evt) =>
+      yield* Stream.runForEach(yield* bus.subscribeAll(), (evt) =>
         Effect.gen(function* () {
           if (evt.type === TestEvent.Warmup.type) {
             yield* ready.open
@@ -129,7 +129,7 @@ describe("Bus (Effect-native)", () => {
       const readyA = yield* Latch.make()
       const readyB = yield* Latch.make()
 
-      yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
+      yield* Stream.runForEach(yield* bus.subscribe(TestEvent.Ping), (evt) =>
         Effect.gen(function* () {
           if (evt.properties.value < 0) {
             yield* readyA.open
@@ -140,7 +140,7 @@ describe("Bus (Effect-native)", () => {
         }),
       ).pipe(Effect.forkScoped)
 
-      yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
+      yield* Stream.runForEach(yield* bus.subscribe(TestEvent.Ping), (evt) =>
         Effect.gen(function* () {
           if (evt.properties.value < 0) {
             yield* readyB.open
@@ -162,6 +162,92 @@ describe("Bus (Effect-native)", () => {
     }),
   )
 
+  // RACE 1: eager subscription means publishing immediately after yield*
+  // bus.subscribe is delivered. Regression for the old lazy `Stream.unwrap`
+  // shape where PubSub.subscribe ran on first pull and missed any publish
+  // in the hand-off window.
+  it.instance("eager subscribe: publish after yield* is delivered without consumer-activation race", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const stream = yield* bus.subscribe(TestEvent.Ping)
+
+      // Hand-off window: subscription is alive (we yielded). Publish goes
+      // straight into the subscription queue, even with no consumer running.
+      yield* bus.publish(TestEvent.Ping, { value: 99 })
+
+      const collected = yield* stream.pipe(
+        Stream.take(1),
+        Stream.runCollect,
+        Effect.timeout("400 millis"),
+        Effect.option,
+      )
+
+      expect(collected._tag).toBe("Some")
+      if (collected._tag === "Some") {
+        const arr = Array.from(collected.value)
+        expect(arr[0].properties.value).toBe(99)
+      }
+    }),
+  )
+
+  // RACE 2: same property for subscribeAll.
+  it.instance("eager subscribeAll: publish after yield* is delivered", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const stream = yield* bus.subscribeAll()
+
+      yield* bus.publish(TestEvent.Ping, { value: 42 })
+
+      const collected = yield* stream.pipe(
+        Stream.take(1),
+        Stream.runCollect,
+        Effect.timeout("400 millis"),
+        Effect.option,
+      )
+
+      expect(collected._tag).toBe("Some")
+      if (collected._tag === "Some") {
+        const arr = Array.from(collected.value)
+        expect(arr[0].type).toBe(TestEvent.Ping.type)
+      }
+    }),
+  )
+
+  // RACE 3: the /event-handler shape exactly. With eager subscription, the
+  // bus subscription is alive before Stream.concat ever starts. Publishes
+  // during the prefix consumption window are queued and delivered.
+  it.instance("eager subscribe: Stream.concat(initial, subscribe) delivers publish during prefix", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const sawInitial = yield* Deferred.make<void>()
+      const sawPublish = yield* Deferred.make<number>()
+
+      type Frame = { marker?: "initial"; value?: number }
+      const subscriptionStream = yield* bus.subscribe(TestEvent.Ping)
+      const handlerStream: Stream.Stream<Frame> = Stream.make({ marker: "initial" } as Frame).pipe(
+        Stream.concat(subscriptionStream.pipe(Stream.map((evt): Frame => ({ value: evt.properties.value })))),
+      )
+
+      yield* Stream.runForEach(handlerStream, (frame) =>
+        Effect.gen(function* () {
+          if (frame.marker === "initial") {
+            Deferred.doneUnsafe(sawInitial, Effect.void)
+            return
+          }
+          if (frame.value !== undefined) Deferred.doneUnsafe(sawPublish, Effect.succeed(frame.value))
+        }),
+      ).pipe(Effect.forkScoped)
+
+      yield* Deferred.await(sawInitial).pipe(Effect.timeout("1 second"))
+
+      yield* bus.publish(TestEvent.Ping, { value: 7 })
+
+      const got = yield* Deferred.await(sawPublish).pipe(Effect.timeout("1 second"), Effect.option)
+      expect(got._tag).toBe("Some")
+      if (got._tag === "Some") expect(got.value).toBe(7)
+    }),
+  )
+
   it.live("subscribeAll stream sees InstanceDisposed on disposal", () =>
     Effect.gen(function* () {
       const dir = yield* tmpdirScoped()
@@ -174,7 +260,7 @@ describe("Bus (Effect-native)", () => {
       yield* Effect.gen(function* () {
         const bus = yield* Bus.Service
 
-        yield* Stream.runForEach(bus.subscribeAll(), (evt) =>
+        yield* Stream.runForEach(yield* bus.subscribeAll(), (evt) =>
           Effect.gen(function* () {
             if (evt.type === TestEvent.Warmup.type) {
               yield* ready.open