refactor: abort signal (#46)

Author: luoling8192

src/adapters/broadcast-channel/index.ts

@@ -71,7 +71,10 @@ export function createContext(channel: BroadcastChannel, options?: BroadcastChan
 
   return {
     context: ctx,
-    dispose: () => {
+    dispose: (reason?: unknown) => {
+      // Cascade-cancel any in-flight `defineInvoke(...)` so callers don't hang
+      // on a torn-down channel (especially when `closeOnDispose: true`).
+      ctx.abort(reason ?? new Error('eventa: invoke cancelled, BroadcastChannel disposed'))
       cleanupRemoval.forEach(removal => removal.remove())
       if (closeOnDispose) {
         channel.close?.()

src/adapters/electron/main.ts

@@ -103,7 +103,10 @@ export function createContext(ipcMain: IpcMain, window?: BrowserWindow, options?
 
   return {
     context: ctx,
-    dispose: () => {
+    dispose: (reason?: unknown) => {
+      // Cascade-cancel any in-flight `defineInvoke(...)` so renderer-bound
+      // RPCs don't hang after the main-side adapter is torn down.
+      ctx.abort(reason ?? new Error('eventa: invoke cancelled, electron main ipc disposed'))
       cleanupRemoval.forEach(removal => removal.remove())
     },
   }

src/adapters/electron/renderer.ts

@@ -65,7 +65,10 @@ export function createContext(ipcRenderer: IpcRenderer, options?: {
 
   return {
     context: ctx,
-    dispose: () => {
+    dispose: (reason?: unknown) => {
+      // Cascade-cancel any in-flight `defineInvoke(...)` so main-bound
+      // RPCs don't hang after the renderer-side adapter is torn down.
+      ctx.abort(reason ?? new Error('eventa: invoke cancelled, electron renderer ipc disposed'))
       cleanupRemoval.forEach(removal => removal.remove())
     },
   }

src/adapters/event-emitter/index.ts

@@ -64,7 +64,8 @@ export function createContext(eventTarget: NodeJS.EventEmitter, options?: {
 
   return {
     context: ctx,
-    dispose: () => {
+    dispose: (reason?: unknown) => {
+      ctx.abort(reason ?? new Error('eventa: invoke cancelled, EventEmitter adapter disposed'))
       cleanupRemoval.forEach(removal => removal.remove())
     },
   }

src/adapters/event-target/index.ts

@@ -71,7 +71,8 @@ export function createContext(eventTarget: EventTarget, options?: {
 
   return {
     context: ctx,
-    dispose: () => {
+    dispose: (reason?: unknown) => {
+      ctx.abort(reason ?? new Error('eventa: invoke cancelled, EventTarget adapter disposed'))
       cleanupRemoval.forEach(removal => removal.remove())
     },
   }

src/adapters/websocket/h3/peer.test.ts

@@ -154,4 +154,60 @@ describe('h3 websocket adapter', { timeout: 2000 }, async () => {
       expect(res.output).toEqual('500')
     }
   })
+
+  // ROOT CAUSE:
+  //
+  // Before this fix, createPeerContext only wired the `message` hook, leaving
+  // `close` and `error` to consumers. A server-side defineInvoke() back to a
+  // client would hang forever when that peer disconnected, because nothing
+  // aborted the peer's ctx.signal. This mismatched the native ws adapter's
+  // self-aborting contract.
+  //
+  // We fixed this by wiring the `close` / `error` hooks directly in
+  // createPeerContext to call ctx.abort(...) in addition to emitting the
+  // business event — same shape as native ws onclose/onerror.
+  it('issue: rejects pending server-side invoke when peer disconnects', async (testCtx) => {
+    const port = randomBetween(40000, 50000)
+    const app = new H3()
+
+    const { untilLeastOneConnected, hooks } = createPeerHooks()
+    app.get('/ws', defineWebSocketHandler(hooks))
+
+    {
+      const server = serve(app, {
+        port,
+        plugins: [ws({
+          resolve: async (req) => {
+            const response = (await app.fetch(req)) as Response & { crossws: Partial<Hooks> }
+            return response.crossws
+          },
+        })],
+      })
+
+      testCtx.onTestFinished(() => {
+        server.close()
+      })
+    }
+
+    const opened = createUntil<void>()
+    const wsConn = new WebSocket(`ws://localhost:${port}/ws`)
+    wsConn.onopen = () => opened.handler()
+    await opened.promise
+
+    // Intentionally do NOT register a handler on the client side — the server
+    // will issue an invoke that has no responder, so the only path out of
+    // the pending promise is the disconnect-driven abort.
+    createContext(wsConn)
+    const { context: serverPeerContext } = await untilLeastOneConnected
+
+    const events = defineInvokeEventa<string, string>('test:peer-abort-cascade')
+    const invoke = defineInvoke(serverPeerContext, events)
+    const pending = invoke('hello')
+
+    // Drop the client side; crossws will fire the `close` hook on the server
+    // peer; createPeerContext.hooks.close calls ctx.abort(...).
+    wsConn.close()
+
+    await expect(pending).rejects.toThrowError(/peer disconnected/i)
+  })
 })

src/adapters/websocket/h3/peer.ts

@@ -4,7 +4,6 @@ import type { EventContext } from '../../../context'
 import type { DirectionalEventa, Eventa } from '../../../eventa'
 
 import { createContext as createBaseContext } from '../../../context'
-import { registerInvokeAbortEventListeners } from '../../../context-extension-invoke-internal'
 import { and, defineEventa, defineInboundEventa, defineOutboundEventa, EventaFlowDirection, matchBy } from '../../../eventa'
 import { generateWebsocketPayload, parseWebsocketPayload } from '../internal'
 
@@ -13,28 +12,12 @@ export const wsDisconnectedEvent = defineEventa<{ id: string }>('eventa:adapters
 export const wsErrorEvent = defineEventa<{ error: unknown }>('eventa:adapters:websocket-peer:error')
 
 export function createPeerContext(peer: Peer): {
-  hooks: Pick<Hooks, 'message'>
+  hooks: Pick<Hooks, 'message' | 'close' | 'error'>
   context: EventContext<any, { raw: { message: Message } }>
 } {
   const peerId = peer.id
   const ctx = createBaseContext<any, { raw: { message: Message } }>()
 
-  // Reject any in-flight `defineInvoke(...)` promises if this peer's transport
-  // dies. Mirrors the native ws adapter so server-side code that issues an
-  // invoke back to a client (push-style RPC) doesn't hang on close.
-  registerInvokeAbortEventListeners(ctx, wsDisconnectedEvent, (payload) => {
-    if (payload.id === wsDisconnectedEvent.id) {
-      const id = (payload as Eventa<{ id?: string }>).body?.id
-      return new Error(`eventa: invoke cancelled, peer disconnected${id ? ` (${id})` : ''}`)
-    }
-    if (payload.id === wsErrorEvent.id) {
-      const err = (payload as Eventa<{ error?: unknown }>).body?.error
-      return err instanceof Error ? err : new Error('eventa: invoke cancelled, peer error')
-    }
-    return undefined
-  })
-  registerInvokeAbortEventListeners(ctx, wsErrorEvent)
-
   ctx.on(and(
     matchBy((e: DirectionalEventa<any>) => e._flowDirection === EventaFlowDirection.Outbound || !e._flowDirection),
     matchBy('*'),
@@ -52,11 +35,30 @@ export function createPeerContext(peer: Peer): {
             ctx.emit(defineInboundEventa(type), payload.body, { raw: { message } })
           }
           catch (error) {
+            // Per-message parse failure — recoverable, do NOT abort lifetime.
             console.error('Failed to parse WebSocket message:', error)
             ctx.emit(wsErrorEvent, { error }, { raw: { message } })
           }
         }
       },
+      close(peer, details) {
+        // crossws fires close for ANY peer; filter to our own.
+        if (peer.id !== peerId) {
+          return
+        }
+        const reasonText = details.reason ? ` (${details.reason})` : ''
+        // Cascade-cancel any in-flight `defineInvoke(...)` so server-side code
+        // that issued an invoke back to this peer doesn't hang on close.
+        ctx.abort(new Error(`eventa: invoke cancelled, peer disconnected${reasonText}`))
+        ctx.emit(wsDisconnectedEvent, { id: peerId })
+      },
+      error(peer, error) {
+        if (peer.id !== peerId) {
+          return
+        }
+        ctx.abort(error instanceof Error ? error : new Error('eventa: invoke cancelled, peer error'))
+        ctx.emit(wsErrorEvent, { error })
+      },
     },
     context: ctx,
   }
@@ -70,18 +72,30 @@ export function createPeerHooks(): { hooks: Partial<Hooks>, untilLeastOneConnect
     resolve = r
   })
 
+  // NOTICE: single-peer model — these closure-scoped hook refs get overwritten
+  // when a second peer connects, so `createPeerHooks` only correctly serves the
+  // most-recently-opened peer. Multi-peer support requires a peerId-keyed Map
+  // and a different "untilLeastOneConnected" semantic; out of scope here.
   let message: Hooks['message'] | undefined
+  let close: Hooks['close'] | undefined
+  let error: Hooks['error'] | undefined
 
-  const hooks: Pick<Hooks, 'open' | 'message'> = {
+  const hooks: Pick<Hooks, 'open' | 'message' | 'close' | 'error'> = {
     open: (peer) => {
       const { context, hooks } = createPeerContext(peer)
       message = hooks.message
+      close = hooks.close
+      error = hooks.error
       resolve({ peer, context })
     },
     message: (peer, msg) => {
-      if (message != null) {
-        message(peer, msg)
-      }
+      message?.(peer, msg)
+    },
+    close: (peer, details) => {
+      close?.(peer, details)
+    },
+    error: (peer, err) => {
+      error?.(peer, err)
     },
   }
 

src/adapters/websocket/native/index.spec.ts

@@ -164,10 +164,11 @@ describe('browser websocket adapter', () => {
   // had to maintain its own pending-RPC tracker and reject manually on
   // disconnect.
   //
-  // We fixed this by registering wsDisconnectedEvent and wsErrorEvent as abort
-  // events on the context inside `createContext`, with a mapAbortError that
-  // produces real Error instances. defineInvoke's existing abortOnEvents
-  // machinery then rejects every in-flight invoke when either event fires.
+  // We fixed this by giving every EventContext a lifetime AbortSignal
+  // (`ctx.signal`) and an `abort(reason)` method. The native ws adapter calls
+  // `ctx.abort(error)` from `onclose` / `onerror`; defineInvoke hooks
+  // `ctx.signal` so transport death cascades into a synchronous reject of
+  // every in-flight invoke. Modeled after Go's context.Context.
   it('issue: rejects pending invoke when socket closes mid-flight', async (testCtx) => {
     const port = randomBetween(40000, 50000)
     const app = new H3()
@@ -212,8 +213,8 @@ describe('browser websocket adapter', () => {
     const invocation = invoke('hello')
 
     // Drop the underlying socket without ever delivering a response. The
-    // adapter emits wsDisconnectedEvent, defineInvoke sees it via abortOnEvents,
-    // and rejects with the mapAbortError-produced Error.
+    // adapter calls ctx.abort(error) from onclose; defineInvoke is hooked to
+    // ctx.signal and rejects with the adapter-supplied Error.
     wsConn.close()
 
     await expect(invocation).rejects.toThrowError(/websocket disconnected/i)

src/adapters/websocket/native/index.ts

@@ -2,7 +2,6 @@ import type { EventContext } from '../../../context'
 import type { DirectionalEventa, Eventa } from '../../../eventa'
 
 import { createContext as createBaseContext } from '../../../context'
-import { registerInvokeAbortEventListeners } from '../../../context-extension-invoke-internal'
 import { and, defineEventa, defineInboundEventa, defineOutboundEventa, EventaFlowDirection, matchBy } from '../../../eventa'
 import { generateWebsocketPayload, parseWebsocketPayload } from '../internal'
 
@@ -13,22 +12,6 @@ export const wsErrorEvent = defineEventa<{ error: unknown }>()
 export function createContext(wsConn: WebSocket) {
   const ctx = createBaseContext() as EventContext<any, { raw: { message?: any, open?: Event, error?: Event, close?: CloseEvent } }>
 
-  // Reject any in-flight `defineInvoke(...)` promises when the socket dies.
-  // Without this, callers wait forever for a response that the closed transport
-  // can never deliver
-  registerInvokeAbortEventListeners(ctx, wsDisconnectedEvent, (payload) => {
-    if (payload.id === wsDisconnectedEvent.id) {
-      const url = (payload as Eventa<{ url?: string }>).body?.url
-      return new Error(`eventa: invoke cancelled, websocket disconnected${url ? ` (${url})` : ''}`)
-    }
-    if (payload.id === wsErrorEvent.id) {
-      const err = (payload as Eventa<{ error?: unknown }>).body?.error
-      return err instanceof Error ? err : new Error('eventa: invoke cancelled, websocket error')
-    }
-    return undefined
-  })
-  registerInvokeAbortEventListeners(ctx, wsErrorEvent)
-
   ctx.on(and(
     matchBy((e: DirectionalEventa<any>) => e._flowDirection === EventaFlowDirection.Outbound || !e._flowDirection),
     matchBy('*'),
@@ -53,10 +36,15 @@ export function createContext(wsConn: WebSocket) {
   }
 
   wsConn.onerror = (error) => {
+    // Socket-level error (not a per-message parse failure — those stay
+    // recoverable in `onmessage` above). Abort lifetime so any in-flight
+    // invoke rejects; emit the business event for non-invoke listeners.
+    ctx.abort(new Error('eventa: invoke cancelled, websocket error'))
     ctx.emit(wsErrorEvent, { error }, { raw: { error } })
   }
 
   wsConn.onclose = (close) => {
+    ctx.abort(new Error(`eventa: invoke cancelled, websocket disconnected (${wsConn.url})`))
     ctx.emit(wsDisconnectedEvent, { url: wsConn.url }, { raw: { close } })
   }
 

src/adapters/webworkers/index.ts

@@ -2,7 +2,6 @@ import type { EventContext } from '../../context'
 import type { DirectionalEventa, Eventa } from '../../eventa'
 
 import { createContext as createBaseContext } from '../../context'
-import { registerInvokeAbortEventListeners } from '../../context-extension-invoke-internal'
 import { and, defineInboundEventa, defineOutboundEventa, EventaFlowDirection, matchBy } from '../../eventa'
 import { generateWorkerPayload, parseWorkerPayload } from './internal'
 import { isWorkerEventa, normalizeOnListenerParameters, workerErrorEvent } from './shared'
@@ -15,8 +14,6 @@ export function createContext(worker: Worker) {
     },
     { raw: { message?: MessageEvent, error?: ErrorEvent, messageError?: MessageEvent }, transfer?: Transferable[] }
   >
-  // Configure invoke to fail fast on fatal worker errors (load/syntax/runtime).
-  registerInvokeAbortEventListeners(ctx, workerErrorEvent)
 
   ctx.on(and(
     matchBy((e: DirectionalEventa<any>) => e._flowDirection === EventaFlowDirection.Outbound || !e._flowDirection),
@@ -49,10 +46,14 @@ export function createContext(worker: Worker) {
   }
 
   worker.onerror = (error) => {
+    // Fatal worker error (load / syntax / runtime). Abort lifetime so any
+    // in-flight invoke rejects; emit the business event for non-invoke listeners.
+    ctx.abort(error instanceof Error ? error : new Error('eventa: invoke cancelled, webworker error'))
     ctx.emit(workerErrorEvent, { error }, { raw: { error } })
   }
 
   worker.onmessageerror = (error) => {
+    ctx.abort(new Error('eventa: invoke cancelled, webworker messageerror'))
     ctx.emit(workerErrorEvent, { error }, { raw: { messageError: error } })
   }