Add uncaught error handling for streaming procedures

README.md

@@ -250,6 +250,17 @@ if (!result.ok && result.payload.code === 'UNCAUGHT_ERROR') {
 }
 ```
 
+For errors raised later by a callback or background task, use `ctx.uncaught(error)` to end the
+procedure with the same built-in error:
+
+```ts
+async handler({ ctx }) {
+  setTimeout(() => {
+    ctx.uncaught(new Error('Something went wrong!'));
+  }, 1000);
+}
+```
+
 ### Logging
 
 To add logging, you can bind a logging function to a transport.

__tests__/cancellation.test.ts

@@ -159,6 +159,50 @@ describe.each(testMatrix())(
         });
       });
 
+      test('stream can report uncaught errors from background tasks', async () => {
+        const clientTransport = getClientTransport('client');
+        const serverTransport = getServerTransport();
+        addPostTestCleanup(async () => {
+          await cleanupTransports([clientTransport, serverTransport]);
+        });
+
+        let signal: AbortSignal | undefined;
+        const services = {
+          service: ServiceSchema.define({
+            stream: Procedure.stream({
+              requestInit: Type.Object({}),
+              requestData: Type.Object({}),
+              responseData: Type.Object({}),
+              handler: async ({ ctx }) => {
+                signal = ctx.signal;
+                setTimeout(() => {
+                  ctx.uncaught(new Error('background task failed'));
+                }, 0);
+              },
+            }),
+          }),
+        };
+
+        const server = createServer(serverTransport, services);
+        const client = createClient<typeof services>(
+          clientTransport,
+          serverTransport.clientId,
+        );
+
+        const { resReadable } = client.service.stream.stream({});
+
+        await expect(resReadable.collect()).resolves.toEqual([
+          Err({ code: UNCAUGHT_ERROR_CODE, message: 'background task failed' }),
+        ]);
+        expect(signal?.aborted).toEqual(true);
+
+        await testFinishesCleanly({
+          clientTransports: [clientTransport],
+          serverTransport,
+          server,
+        });
+      });
+
       test('upload', async () => {
         const clientTransport = getClientTransport('client');
         const serverTransport = getServerTransport();

router/context.ts

@@ -2,7 +2,7 @@ import { Span } from '@opentelemetry/api';
 import { TransportClientId } from '../transport/message';
 import { SessionId } from '../transport/sessionStateMachine/common';
 import { ErrResult } from './result';
-import { CancelErrorSchema } from './errors';
+import { CancelErrorSchema, ReaderErrorSchema } from './errors';
 import type { Static } from 'typebox';
 
 /**
@@ -40,6 +40,12 @@ export type ProcedureHandlerContext<State, Context, ParsedMetadata> =
      * the river documentation to understand the difference between the two concepts.
      */
     cancel: (message?: string) => ErrResult<Static<typeof CancelErrorSchema>>;
+    /**
+     * End the procedure with an uncaught server error and notify the client.
+     * This is useful for errors raised by callbacks or background tasks that
+     * run after the handler has returned control to the event loop.
+     */
+    uncaught: (error: unknown) => ErrResult<Static<typeof ReaderErrorSchema>>;
     /**
      * Register a cleanup function that will run after the procedure handler
      * completes (whether it returns normally, throws, or is cancelled).

router/server.ts

@@ -730,6 +730,16 @@ class RiverServer<
 
         return Err(errRes);
       },
+      uncaught: (error: unknown) => {
+        const errRes = {
+          code: UNCAUGHT_ERROR_CODE,
+          message: coerceErrorString(error),
+        } as const;
+
+        onServerCancel(errRes);
+
+        return Err(errRes);
+      },
       deferCleanup,
       signal: finishedController.signal,
     };
@@ -1217,7 +1227,10 @@ function getStreamCloseBackwardsCompat(protocolVersion: ProtocolVersion) {
 
 export interface MiddlewareContext
   extends Readonly<
-    Omit<ProcedureHandlerContext<unknown, unknown, unknown>, 'cancel'>
+    Omit<
+      ProcedureHandlerContext<unknown, unknown, unknown>,
+      'cancel' | 'uncaught'
+    >
   > {
   readonly streamId: StreamId;
   readonly procedureName: string;