Include river errors in server side spans

router/client.ts

@@ -288,6 +288,7 @@ function handleProc(
   const procClosesWithInit = procType === 'rpc' || procType === 'subscription';
   const streamId = generateId();
   const { span, ctx } = createProcTelemetryInfo(
+    transport.tracer,
     session,
     procType,
     serviceName,

router/server.ts

@@ -36,8 +36,12 @@ import { Value, ValueError } from '@sinclair/typebox/value';
 import { Err, Result, Ok, ErrResult } from './result';
 import { EventMap } from '../transport/events';
 import { coerceErrorString } from '../transport/stringifyError';
-import { Span, SpanStatusCode } from '@opentelemetry/api';
-import { createHandlerSpan, PropagationContext } from '../tracing';
+import { Span } from '@opentelemetry/api';
+import {
+  createHandlerSpan,
+  PropagationContext,
+  recordRiverError,
+} from '../tracing';
 import { ServerHandshakeOptions } from './handshake';
 import { Connection } from '../transport/connection';
 import { ServerTransport } from '../transport/server';
@@ -198,6 +202,7 @@ class RiverServer<Services extends AnyServiceSchemaMap>
 
       // if its not a cancelled stream, validate and create a new stream
       createHandlerSpan(
+        transport.tracer,
         newStreamProps.initialSession,
         newStreamProps.procedure.type,
         newStreamProps.serviceName,
@@ -423,6 +428,8 @@ class RiverServer<Services extends AnyServiceSchemaMap>
     };
 
     const onServerCancel = (e: Static<typeof ReaderErrorSchema>) => {
+      recordRiverError(span, e);
+
       if (reqReadable.isClosed() && resWritable.isClosed()) {
         // Everything already closed, no-op.
         return;
@@ -476,6 +483,10 @@ class RiverServer<Services extends AnyServiceSchemaMap>
       Result<Static<PayloadType>, Static<BaseErrorSchemaType>>
     >({
       writeCb: (response) => {
+        if (!response.ok) {
+          recordRiverError(span, response.payload);
+        }
+
         sessionScopedSend({
           streamId,
           controlFlags: procClosesWithResponse
@@ -519,7 +530,6 @@ class RiverServer<Services extends AnyServiceSchemaMap>
       const errorMsg = coerceErrorString(err);
 
       span.recordException(err instanceof Error ? err : new Error(errorMsg));
-      span.setStatus({ code: SpanStatusCode.ERROR });
 
       this.log?.error(
         `${serviceName}.${procedureName} handler threw an uncaught error`,

testUtil/index.ts

@@ -19,6 +19,7 @@ import { SessionStateGraph } from '../transport/sessionStateMachine/transitions'
 import { BaseErrorSchemaType } from '../router/errors';
 import { ClientTransport } from '../transport/client';
 import { ServerTransport } from '../transport/server';
+import { getTracer } from '../tracing';
 
 export {
   createMockTransportNetwork,
@@ -186,6 +187,7 @@ export function dummySession() {
     },
     testingSessionOptions,
     currentProtocolVersion,
+    getTracer(),
   );
 }
 

tracing/index.ts

@@ -2,16 +2,18 @@ import {
   Context,
   Span,
   SpanKind,
+  SpanStatusCode,
   context,
   propagation,
   trace,
+  Tracer,
 } from '@opentelemetry/api';
-import { version as RIVER_VERSION } from '../package.json';
-import { ValidProcType } from '../router';
+import { BaseErrorSchemaType, RIVER_VERSION, ValidProcType } from '../router';
 import { Connection } from '../transport';
 import { MessageMetadata } from '../logging';
 import { ClientSession } from '../transport/sessionStateMachine/transitions';
 import { IdentifiedSession } from '../transport/sessionStateMachine/common';
+import { Static } from '@sinclair/typebox';
 
 export interface PropagationContext {
   traceparent: string;
@@ -36,6 +38,7 @@ export function getPropagationContext(
 }
 
 export function createSessionTelemetryInfo(
+  tracer: Tracer,
   sessionId: string,
   to: string,
   from: string,
@@ -46,7 +49,7 @@ export function createSessionTelemetryInfo(
     : context.active();
 
   const span = tracer.startSpan(
-    `session ${sessionId}`,
+    `river.session.${sessionId}`,
     {
       attributes: {
         component: 'river',
@@ -64,6 +67,7 @@ export function createSessionTelemetryInfo(
 }
 
 export function createConnectionTelemetryInfo(
+  tracer: Tracer,
   connection: Connection,
   info: TelemetryInfo,
 ): TelemetryInfo {
@@ -85,6 +89,7 @@ export function createConnectionTelemetryInfo(
 }
 
 export function createProcTelemetryInfo(
+  tracer: Tracer,
   session: ClientSession<Connection>,
   kind: ValidProcType,
   serviceName: string,
@@ -93,7 +98,7 @@ export function createProcTelemetryInfo(
 ): TelemetryInfo {
   const baseCtx = context.active();
   const span = tracer.startSpan(
-    `procedure call ${serviceName}.${procedureName}`,
+    `river.client.${serviceName}.${procedureName}`,
     {
       attributes: {
         component: 'river',
@@ -131,6 +136,7 @@ export function createProcTelemetryInfo(
 }
 
 export function createHandlerSpan<Fn extends (span: Span) => unknown>(
+  tracer: Tracer,
   session: IdentifiedSession,
   kind: ValidProcType,
   serviceName: string,
@@ -144,7 +150,7 @@ export function createHandlerSpan<Fn extends (span: Span) => unknown>(
     : context.active();
 
   return tracer.startActiveSpan<Fn>(
-    `procedure handler ${serviceName}.${procedureName}`,
+    `river.server.${serviceName}.${procedureName}`,
     {
       attributes: {
         component: 'river',
@@ -162,5 +168,20 @@ export function createHandlerSpan<Fn extends (span: Span) => unknown>(
   );
 }
 
-const tracer = trace.getTracer('river', RIVER_VERSION);
-export default tracer;
+export function recordRiverError(
+  span: Span,
+  error: Static<BaseErrorSchemaType>,
+): void {
+  span.setStatus({
+    code: SpanStatusCode.ERROR,
+    message: error.message,
+  });
+  span.setAttributes({
+    'river.error_code': error.code,
+    'river.error_message': error.message,
+  });
+}
+
+export function getTracer(): Tracer {
+  return trace.getTracer('river', RIVER_VERSION);
+}

tracing/tracing.test.ts

@@ -1,4 +1,10 @@
-import { trace, context, propagation, Span } from '@opentelemetry/api';
+import {
+  trace,
+  context,
+  propagation,
+  Span,
+  SpanStatusCode,
+} from '@opentelemetry/api';
 import { describe, test, expect, vi, assert, beforeEach } from 'vitest';
 import { dummySession, readNextResult } from '../testUtil';
 
@@ -9,10 +15,7 @@ import {
 } from '@opentelemetry/sdk-trace-base';
 import { W3CTraceContextPropagator } from '@opentelemetry/core';
 import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks';
-import tracer, {
-  createSessionTelemetryInfo,
-  getPropagationContext,
-} from './index';
+import { createSessionTelemetryInfo, getPropagationContext } from './index';
 import { testMatrix } from '../testUtil/fixtures/matrix';
 import {
   cleanupTransports,
@@ -27,26 +30,27 @@ import { createClient } from '../router/client';
 import { UNCAUGHT_ERROR_CODE } from '../router';
 import { LogFn } from '../logging';
 
-describe('Basic tracing tests', () => {
-  const provider = new BasicTracerProvider();
-  provider.addSpanProcessor(
-    new SimpleSpanProcessor(new InMemorySpanExporter()),
-  );
-  const contextManager = new AsyncHooksContextManager();
-  contextManager.enable();
-  trace.setGlobalTracerProvider(provider);
-  context.setGlobalContextManager(contextManager);
-  propagation.setGlobalPropagator(new W3CTraceContextPropagator());
+const provider = new BasicTracerProvider();
+const spanExporter = new InMemorySpanExporter();
+provider.addSpanProcessor(new SimpleSpanProcessor(spanExporter));
+const contextManager = new AsyncHooksContextManager();
+contextManager.enable();
+trace.setGlobalTracerProvider(provider);
+context.setGlobalContextManager(contextManager);
+propagation.setGlobalPropagator(new W3CTraceContextPropagator());
 
+describe('Basic tracing tests', () => {
   test('createSessionTelemetryInfo', () => {
     const parentCtx = context.active();
+    const tracer = trace.getTracer('test');
     const span = tracer.startSpan('empty span', {}, parentCtx);
     const ctx = trace.setSpan(parentCtx, span);
 
     const propCtx = getPropagationContext(ctx);
     expect(propCtx?.traceparent).toBeTruthy();
     const session = dummySession();
     const teleInfo = createSessionTelemetryInfo(
+      tracer,
       session.id,
       session.to,
       session.from,
@@ -75,6 +79,7 @@ describe.each(testMatrix())(
       const setup = await transport.setup({ client: opts, server: opts });
       getClientTransport = setup.getClientTransport;
       getServerTransport = setup.getServerTransport;
+      spanExporter.reset();
 
       return async () => {
         await postTestCleanup();
@@ -189,5 +194,74 @@ describe.each(testMatrix())(
         server,
       });
     });
+
+    test('river errors are recorded on handler spans', async () => {
+      // setup
+      const clientTransport = getClientTransport('client');
+      const clientMockLogger = vi.fn<LogFn>();
+      clientTransport.bindLogger(clientMockLogger);
+      const serverTransport = getServerTransport();
+      const serverMockLogger = vi.fn<LogFn>();
+      serverTransport.bindLogger(serverMockLogger);
+      const services = {
+        fallible: FallibleServiceSchema,
+      };
+      const server = createServer(serverTransport, services);
+      const client = createClient<typeof services>(
+        clientTransport,
+        serverTransport.clientId,
+      );
+      addPostTestCleanup(async () => {
+        await cleanupTransports([clientTransport, serverTransport]);
+      });
+
+      const { reqWritable, resReadable } = client.fallible.echo.stream({});
+
+      reqWritable.write({
+        msg: 'abc',
+        throwResult: false,
+        throwError: false,
+      });
+      let result = await readNextResult(resReadable);
+      expect(result).toStrictEqual({
+        ok: true,
+        payload: {
+          response: 'abc',
+        },
+      });
+
+      // this isn't the first message so doesn't have telemetry info on the message itself
+      reqWritable.write({
+        msg: 'def',
+        throwResult: false,
+        throwError: true,
+      });
+
+      result = await readNextResult(resReadable);
+      expect(result).toStrictEqual({
+        ok: false,
+        payload: {
+          code: UNCAUGHT_ERROR_CODE,
+          message: 'some message',
+        },
+      });
+
+      const spans = spanExporter.getFinishedSpans();
+
+      const errSpan = spans.find(
+        (span) =>
+          span.name === 'river.server.fallible.echo' &&
+          span.status.code === SpanStatusCode.ERROR,
+      );
+      expect(errSpan).toBeTruthy();
+      expect(errSpan?.attributes['river.error_code']).toBe(UNCAUGHT_ERROR_CODE);
+      expect(errSpan?.attributes['river.error_message']).toBe('some message');
+
+      await testFinishesCleanly({
+        clientTransports: [clientTransport],
+        serverTransport,
+        server,
+      });
+    });
   },
 );

transport/client.ts

@@ -18,7 +18,7 @@ import { Transport } from './transport';
 import { coerceErrorString } from './stringifyError';
 import { ProtocolError } from './events';
 import { Value } from '@sinclair/typebox/value';
-import tracer, { getPropagationContext } from '../tracing';
+import { getPropagationContext } from '../tracing';
 import { Connection } from './connection';
 import { MessageMetadata } from '../logging';
 import { SessionConnecting } from './sessionStateMachine/SessionConnecting';
@@ -110,6 +110,7 @@ export abstract class ClientTransport<
       },
       this.options,
       currentProtocolVersion,
+      this.tracer,
       this.log,
     );
 
@@ -377,20 +378,23 @@ export abstract class ClientTransport<
   }
 
   protected onBackoffFinished(session: SessionBackingOff) {
-    const connPromise = tracer.startActiveSpan('connect', async (span) => {
-      try {
-        return await this.createNewOutgoingConnection(session.to);
-      } catch (err) {
-        // rethrow the error so that the promise is rejected
-        // as it was before we wrapped it in a span
-        const errStr = coerceErrorString(err);
-        span.recordException(errStr);
-        span.setStatus({ code: SpanStatusCode.ERROR });
-        throw err;
-      } finally {
-        span.end();
-      }
-    });
+    const connPromise = session.tracer.startActiveSpan(
+      'connect',
+      async (span) => {
+        try {
+          return await this.createNewOutgoingConnection(session.to);
+        } catch (err) {
+          // rethrow the error so that the promise is rejected
+          // as it was before we wrapped it in a span
+          const errStr = coerceErrorString(err);
+          span.recordException(errStr);
+          span.setStatus({ code: SpanStatusCode.ERROR });
+          throw err;
+        } finally {
+          span.end();
+        }
+      },
+    );
 
     // transition to connecting
     const connectingSession =

transport/server.ts

@@ -164,6 +164,7 @@ export abstract class ServerTransport<
         },
       },
       this.options,
+      this.tracer,
       this.log,
     );