feat(devframe): forward onFunctionError/onGeneralError to startHttpAndWs

package.json

@@ -14,17 +14,17 @@
   },
   "bugs": "https://github.com/devframes/devframe/issues",
   "scripts": {
-    "build": "turbo run build",
+    "build": "turbo run build --concurrency=3",
     "watch": "pnpm -r run watch",
     "docs": "pnpm -C docs run docs",
     "docs:build": "pnpm -C docs run docs:build",
     "docs:serve": "pnpm -C docs run docs:serve",
     "storybook": "pnpm -r --parallel --if-present run storybook",
     "storybook:build": "pnpm --filter @devframes/storybook run storybook:build",
     "lint": "eslint --cache",
-    "test": "turbo run build && vitest",
-    "test:e2e": "turbo run build && playwright test",
-    "test:e2e:ui": "turbo run build && playwright test --ui",
+    "test": "pnpm run build && vitest",
+    "test:e2e": "pnpm run build && playwright test",
+    "test:e2e:ui": "pnpm run build && playwright test --ui",
     "test:ecosystem": "tsx scripts/ecosystem-ci.ts",
     "release": "bumpp -r",
     "typecheck": "pnpm run verify:typecheck-coverage && turbo run typecheck",

packages/devframe/src/node/__tests__/server.test.ts

@@ -0,0 +1,102 @@
+import type { DevframeHost, DevframeNodeContext, DevframeRpcClientFunctions, DevframeRpcServerFunctions } from 'devframe/types'
+import { mkdtempSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { createRpcClient } from 'devframe/rpc/client'
+import { createWsRpcChannel } from 'devframe/rpc/transports/ws-client'
+import { getPort } from 'get-port-please'
+import { describe, expect, it, vi } from 'vitest'
+import { WebSocket } from 'ws'
+import { createHostContext } from '../context'
+import { startHttpAndWs } from '../server'
+
+function makeHost(storageDir: string): DevframeHost {
+  return {
+    mountStatic: () => {},
+    resolveOrigin: () => 'http://localhost',
+    getStorageDir: () => storageDir,
+  }
+}
+
+async function createTestContext(): Promise<DevframeNodeContext> {
+  const storageDir = mkdtempSync(join(tmpdir(), 'devframe-server-'))
+  return createHostContext({ cwd: storageDir, mode: 'dev', host: makeHost(storageDir) })
+}
+
+function connectClient(host: string, port: number) {
+  return createRpcClient<DevframeRpcServerFunctions, DevframeRpcClientFunctions>(
+    {} as DevframeRpcClientFunctions,
+    { channel: createWsRpcChannel({ url: `ws://${host}:${port}` }) },
+  )
+}
+
+describe('startHttpAndWs rpcOptions passthrough', () => {
+  it('forwards a thrown handler error to rpcOptions.onFunctionError without swallowing the response', async () => {
+    const context = await createTestContext()
+    context.rpc.register({
+      name: 'test:boom',
+      type: 'action',
+      handler: () => {
+        throw new Error('kaboom')
+      },
+    })
+
+    const onFunctionError = vi.fn()
+    const host = '127.0.0.1'
+    const port = await getPort({ port: 0, host })
+    const server = await startHttpAndWs({
+      context,
+      host,
+      port,
+      auth: false,
+      rpcOptions: { onFunctionError },
+    })
+
+    try {
+      const client = connectClient(host, port)
+      await expect(client.$call('test:boom' as any)).rejects.toThrow('kaboom')
+
+      expect(onFunctionError).toHaveBeenCalledTimes(1)
+      const [error, name] = onFunctionError.mock.calls[0]!
+      expect(name).toBe('test:boom')
+      expect((error as Error).message).toBe('kaboom')
+      client.$close()
+    }
+    finally {
+      await server.close()
+    }
+  })
+
+  it('forwards a deserialize failure to rpcOptions.onGeneralError', async () => {
+    const context = await createTestContext()
+    // Returning `true` tells birpc the error was handled, suppressing its
+    // default rethrow — matches how a "log and swallow" host would use this.
+    const onGeneralError = vi.fn(() => true)
+    const host = '127.0.0.1'
+    const port = await getPort({ port: 0, host })
+    const server = await startHttpAndWs({
+      context,
+      host,
+      port,
+      auth: false,
+      rpcOptions: { onGeneralError },
+    })
+
+    try {
+      const raw = new WebSocket(`ws://${host}:${port}`)
+      await new Promise<void>((resolve, reject) => {
+        raw.once('open', () => resolve())
+        raw.once('error', reject)
+      })
+      raw.send('not valid json and not structured-clone either')
+
+      await vi.waitFor(() => {
+        expect(onGeneralError).toHaveBeenCalledTimes(1)
+      })
+      raw.close()
+    }
+    finally {
+      await server.close()
+    }
+  })
+})

packages/devframe/src/node/server.ts

@@ -1,4 +1,4 @@
-import type { BirpcGroup } from 'birpc'
+import type { BirpcGroup, EventOptions } from 'birpc'
 import type { Peer } from 'crossws'
 import type { NodeAdapter } from 'crossws/adapters/node'
 import type { ConnectionMeta, DevframeNodeContext, DevframeNodeRpcSession, DevframeNodeRpcSessionMeta, DevframeRpcClientFunctions, DevframeRpcServerFunctions } from 'devframe/types'
@@ -80,6 +80,20 @@ export interface StartHttpAndWsOptions {
    * observe — but not override — the connect-time trust decision.
    */
   onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
+  /**
+   * Forwarded verbatim to the internal `createRpcServer`'s birpc
+   * `rpcOptions`, alongside the resolver `startHttpAndWs` installs for
+   * auth/session wiring. Use this so a host that owns its own structured
+   * diagnostics (e.g. a coded error reporter) keeps seeing RPC failures
+   * instead of them being silently absorbed by delegating to
+   * `startHttpAndWs`. Returning `true` from either callback suppresses
+   * birpc's own error response to the caller — see birpc's
+   * `EventOptions` for the full contract.
+   */
+  rpcOptions?: Pick<
+    EventOptions<DevframeRpcClientFunctions, DevframeRpcServerFunctions, false>,
+    'onFunctionError' | 'onGeneralError'
+  >
   /**
    * Extra origins to accept on the WS upgrade beyond the loopback default
    * (`localhost`/`127.0.0.1`/`::1` and any `Origin`-less request from a
@@ -150,6 +164,10 @@ export async function startHttpAndWs(options: StartHttpAndWsOptions): Promise<St
     rpcHost.functions,
     {
       rpcOptions: {
+        // Forwarded as-is so a host with its own structured diagnostics
+        // keeps seeing RPC failures; see `StartHttpAndWsOptions.rpcOptions`.
+        onFunctionError: options.rpcOptions?.onFunctionError,
+        onGeneralError: options.rpcOptions?.onGeneralError,
         // Wrap each RPC handler in an AsyncLocalStorage context so
         // `ctx.rpc.getCurrentRpcSession()` works inside handlers (used
         // by streaming subscribe/unsubscribe/cancel and shared-state