Rule proposal: resume registered on a multi-shot event source inside Effect.callback drops every event after the first

[mattiamanzati]

Problem

Effect.callback is single-shot: only the first resume wins. Flags callback bodies where resume is invoked from a handler registered via a repeating subscription API (emitter.on(...), addEventListener without { once: true }, socket.on('data'), setInterval) rather than a once-style registration. The multi-shot idiom is Stream.callback, which buffers each emission into a queue.

Why the compiler is silent / what breaks at runtime: Type-checks cleanly, but only the first emitted event is ever observed; subsequent resume calls are no-ops, later events vanish, and the un-removed listener leaks on the emitter.

Both examples below type-check with zero errors against effect@4.0.0-beta.104 (re-verified) under a strict tsconfig with an isolated per-proposal tsconfig — so the compiler offers no protection here and a diagnostic is the only static safety net.

Bad — compiles cleanly, the rule should flag this

// RULE: effect-callback-multishot-emitter
// BAD: resume is invoked from a handler registered with a repeating
// subscription API (emitter.on) inside Effect.callback. Effect.callback is
// single-shot: only the first resume(...) completes the fiber, every later
// "message" event is silently dropped, and the listener is never removed.
import { Effect } from "effect"
import { EventEmitter } from "node:events"

interface Msg {
  readonly id: number
  readonly body: string
}

const emitter = new EventEmitter()

// Type-checks cleanly, but observes only the FIRST message ever emitted.
const messages = Effect.callback<Msg>((resume) => {
  emitter.on("message", (msg: Msg) => {
    // Second and subsequent invocations are no-ops: events vanish.
    resume(Effect.succeed(msg))
  })
})

void messages

Good

// RULE: effect-callback-multishot-emitter
// GOOD: multi-shot event sources belong in Stream.callback, which buffers
// every emission into a queue. Registration is wrapped in acquireRelease so
// the listener is detached when the stream is interrupted or finished.
import { Effect, Queue, Stream } from "effect"
import { EventEmitter } from "node:events"

interface Msg {
  readonly id: number
  readonly body: string
}

const emitter = new EventEmitter()

// Every emitted "message" is offered to the queue and observed downstream.
const messages = Stream.callback<Msg>((queue) =>
  Effect.acquireRelease(
    Effect.sync(() => {
      const handler = (msg: Msg) => {
        Queue.offerUnsafe(queue, msg)
      }
      emitter.on("message", handler)
      return handler
    }),
    (handler) =>
      Effect.sync(() => {
        emitter.off("message", handler)
      })
  )
)

void messages

Where this came up

• Roasting PRs & ecosystem updates | Effect Office Hours 22 @ 1:31:30

Mined from the Effect Office Hours playlist; deduplicated against all implemented tsgo diagnostics and prior rule-proposal issues.

Proposed rule name

effectCallbackMultishotEmitter