Add structured handshake rejections

PROTOCOL.md

@@ -258,6 +258,9 @@ interface ControlHandshakeResponse {
           // fatal, returned by the custom handshake handler
           | 'REJECTED_BY_CUSTOM_HANDLER'
           | 'REJECTED_UNSUPPORTED_CLIENT';
+        // Application-defined rejection data. River populates this only for
+        // custom-handler rejections. Older peers ignore this optional field.
+        extras?: unknown;
       };
 }
 
@@ -628,6 +631,7 @@ The server will send an error response if either:
   - server is in the future (`server.seq > client.nextExpectedSeq`)
 
 When the client receives a status with `ok: false`, it should consider the handshake failed and close the connection.
+Custom handshake handlers can attach application-defined `extras` to a rejection. River transports `extras` opaquely, preserves its own `code` for protocol behavior, and exposes `extras` on the client's `handshake_failed` protocol error event. Applications define and validate their own schema for `extras`, so they can make decisions (for example, stop retrying on a terminal failure) without parsing the human-readable `reason` or adding error types to River.
 
 ### Re-handshaking (live credential refresh)
 

README.md

@@ -807,8 +807,9 @@ createServer(serverTransport, services, {
       //   from?: TransportClientId,
       // ) =>
       //   | 'REJECTED_BY_CUSTOM_HANDLER' | 'REJECTED_UNSUPPORTED_CLIENT' (if you reject it)
+      //   | HandshakeRejection (if you reject it with application-defined extras)
       //   | ParsedMetadata (if you allow it)
-      //   | a Promise of either
+      //   | a Promise of any of the above
       //
       // next time a connection happens on the same session, previousMetadata will
       // be populated with the last returned value. `from` is the client id the peer
@@ -820,6 +821,26 @@ createServer(serverTransport, services, {
 });
 ```
 
+Use `rejectHandshake` when the client needs a machine-readable, application-defined reason for a rejection. The argument is an opaque `extras` value; your application owns its schema:
+
+```ts
+createServerHandshakeOptions(handshakeSchema, async (metadata) => {
+  const authenticated = await authenticate(metadata.token);
+  if (!authenticated.ok) {
+    return rejectHandshake({
+      code: 'TOKEN_EXPIRED',
+      message: 'The authentication token expired',
+    });
+  }
+
+  return { parsedToken: metadata.token };
+});
+```
+
+River sends `extras` on the optional `extras` field of the failed handshake response and exposes it on the client's `handshake_failed` protocol error event as `event.extras`. River does not interpret it: validate it with your own schema (for example with TypeBox's `Value.Check`) before acting on it. Existing failure codes remain available for simple handlers. Do not put secrets or raw internal errors in `extras` because River sends it to the peer.
+
+During a re-handshake, River exposes the rejection extras only on the server's `handshake_failed` event before it closes the session. The client's next fresh handshake can receive the extras.
+
 `createClientHandshakeOptions` also takes an optional third `eager` argument. When set, the
 client constructs handshake metadata as soon as it starts dialing, so a slow `construct`
 (e.g. fetching a fresh token) overlaps establishing the connection instead of running after

__tests__/e2e.test.ts

@@ -42,6 +42,7 @@ import {
 import {
   createClientHandshakeOptions,
   createServerHandshakeOptions,
+  rejectHandshake,
 } from '../router/handshake';
 import { RehandshakeStreamId } from '../transport/message';
 import { TestSetupHelpers } from '../testUtil/fixtures/transports';
@@ -1483,13 +1484,14 @@ describe.each(testMatrix())(
         'client',
         createClientHandshakeOptions(requestSchema, construct),
       );
-      const validate = vi.fn(
-        (
-          metadata: ParsedMetadata,
-        ): ParsedMetadata | 'REJECTED_BY_CUSTOM_HANDLER' =>
-          metadata.token === 'token-v1'
-            ? { token: metadata.token }
-            : 'REJECTED_BY_CUSTOM_HANDLER',
+      const rejectionExtras = {
+        code: 'TOKEN_EXPIRED',
+        message: 'The refreshed token expired',
+      };
+      const validate = vi.fn((metadata: ParsedMetadata) =>
+        metadata.token === 'token-v1'
+          ? { token: metadata.token }
+          : rejectHandshake(rejectionExtras),
       );
       const serverTransport = getServerTransport<
         typeof requestSchema,
@@ -1504,6 +1506,8 @@ describe.each(testMatrix())(
       addPostTestCleanup(async () => {
         await cleanupTransports([clientTransport, serverTransport]);
       });
+      const serverHandshakeFailed = vi.fn();
+      serverTransport.addEventListener('protocolError', serverHandshakeFailed);
 
       const ServiceSchema = createServiceSchema<
         MaybeDisposable,
@@ -1539,6 +1543,12 @@ describe.each(testMatrix())(
         expect(serverTransport.sessions.has('client')).toBe(false),
       );
       await waitFor(() => expect(numberOfConnections(clientTransport)).toBe(0));
+      expect(serverHandshakeFailed).toHaveBeenCalledWith({
+        type: 'handshake_failed',
+        code: 'REJECTED_BY_CUSTOM_HANDLER',
+        message: 're-handshake metadata rejected by handshake handler',
+        extras: rejectionExtras,
+      });
 
       // let the client's now-disconnected session lapse before cleanup
       await advanceFakeTimersBySessionGrace();

protobuf/handshake.ts

@@ -3,26 +3,19 @@ import type {
   MessageInitShape,
   MessageShape,
 } from '@bufbuild/protobuf';
-import { type Static } from 'typebox';
 import {
   createClientHandshakeOptions as createTransportClientHandshakeOptions,
   createServerHandshakeOptions as createTransportServerHandshakeOptions,
   type ClientHandshakeOptions,
+  type HandshakeValidationResult,
   type ServerHandshakeOptions,
 } from '../router/handshake';
-import {
-  HandshakeErrorCustomHandlerFatalResponseCodes,
-  type TransportClientId,
-} from '../transport/message';
+import { type TransportClientId } from '../transport/message';
 import { decodeMessageBytes, encodeMessageBytes } from './shared';
 import { Uint8ArrayType } from '../customSchemas';
 
 const HandshakeBytesSchema = Uint8ArrayType();
 
-type ProtobufHandshakeFailureCode = Static<
-  typeof HandshakeErrorCustomHandlerFatalResponseCodes
->;
-
 type ConstructHandshake<Schema extends DescMessage> = () =>
   | MessageInitShape<Schema>
   | Promise<MessageInitShape<Schema>>;
@@ -32,9 +25,8 @@ type ValidateHandshake<Schema extends DescMessage, ParsedMetadata> = (
   previousParsedMetadata?: ParsedMetadata,
   from?: TransportClientId,
 ) =>
-  | ParsedMetadata
-  | ProtobufHandshakeFailureCode
-  | Promise<ParsedMetadata | ProtobufHandshakeFailureCode>;
+  | HandshakeValidationResult<ParsedMetadata>
+  | Promise<HandshakeValidationResult<ParsedMetadata>>;
 
 /**
  * Create client-side handshake options backed by a protobuf message type.
@@ -73,7 +65,7 @@ export function createServerHandshakeOptions<
       try {
         decoded = decodeMessageBytes(schema, metadata);
       } catch {
-        return 'REJECTED_BY_CUSTOM_HANDLER' as ProtobufHandshakeFailureCode;
+        return 'REJECTED_BY_CUSTOM_HANDLER';
       }
 
       return await validate(decoded, previousParsedMetadata, from);

protobuf/index.ts

@@ -21,6 +21,8 @@ export {
   createClientHandshakeOptions,
   createServerHandshakeOptions,
 } from './handshake';
+export { rejectHandshake } from '../router/handshake';
+export type { HandshakeRejection } from '../router/handshake';
 export { createProtoService } from './service';
 export type {
   AnyProtoService,

router/handshake.ts

@@ -4,6 +4,47 @@ import {
   type TransportClientId,
 } from '../transport/message';
 
+const handshakeRejectionBrand: unique symbol = Symbol('handshakeRejection');
+
+export interface HandshakeRejection {
+  readonly [handshakeRejectionBrand]: true;
+  responseCode: Static<typeof HandshakeErrorCustomHandlerFatalResponseCodes>;
+  /**
+   * Application-defined rejection data, forwarded to the peer in the failed
+   * handshake response. River transports it opaquely; the application defines
+   * and validates its own schema for it.
+   */
+  extras: unknown;
+}
+
+export function rejectHandshake(
+  extras: unknown,
+  responseCode: Static<
+    typeof HandshakeErrorCustomHandlerFatalResponseCodes
+  > = 'REJECTED_BY_CUSTOM_HANDLER',
+): HandshakeRejection {
+  return {
+    [handshakeRejectionBrand]: true,
+    responseCode,
+    extras,
+  };
+}
+
+export function isHandshakeRejection(
+  value: unknown,
+): value is HandshakeRejection {
+  return (
+    typeof value === 'object' &&
+    value !== null &&
+    handshakeRejectionBrand in value
+  );
+}
+
+export type HandshakeValidationResult<ParsedMetadata> =
+  | Static<typeof HandshakeErrorCustomHandlerFatalResponseCodes>
+  | HandshakeRejection
+  | ParsedMetadata;
+
 type ConstructHandshake<T extends TSchema> = () =>
   | Static<T>
   | Promise<Static<T>>;
@@ -13,12 +54,8 @@ type ValidateHandshake<T extends TSchema, ParsedMetadata> = (
   previousParsedMetadata?: ParsedMetadata,
   from?: TransportClientId,
 ) =>
-  | Static<typeof HandshakeErrorCustomHandlerFatalResponseCodes>
-  | ParsedMetadata
-  | Promise<
-      | Static<typeof HandshakeErrorCustomHandlerFatalResponseCodes>
-      | ParsedMetadata
-    >;
+  | HandshakeValidationResult<ParsedMetadata>
+  | Promise<HandshakeValidationResult<ParsedMetadata>>;
 
 export interface ClientHandshakeOptions<
   MetadataSchema extends TSchema = TSchema,
@@ -57,8 +94,8 @@ export interface ServerHandshakeOptions<
 
   /**
    * Parses the metadata sent by the client during the handshake into the
-   * server-side {@link ParsedMetadata}, or returns a handshake failure code to
-   * reject the connection.
+   * server-side {@link ParsedMetadata}, or returns a handshake failure code or
+   * {@link HandshakeRejection} to reject the connection.
    *
    * @param metadata - The metadata sent by the client.
    * @param previousParsedMetadata - The parsed metadata from the previous

router/index.ts

@@ -64,5 +64,7 @@ export type {
 export {
   createClientHandshakeOptions,
   createServerHandshakeOptions,
+  rejectHandshake,
 } from './handshake';
+export type { HandshakeRejection } from './handshake';
 export { version as RIVER_VERSION } from '../package.json';

transport/client.ts

@@ -377,10 +377,17 @@ export abstract class ClientTransport<
       );
 
       const reason = `handshake failed: ${msg.payload.status.reason}`;
+      const { extras } = msg.payload.status;
       const to = session.to;
       this.rejectHandshakeResponse(session, reason, {
         ...session.loggingMetadata,
         transportMessage: msg,
+        ...(extras !== undefined && {
+          extras: {
+            ...session.loggingMetadata.extras,
+            handshakeRejectionExtras: extras,
+          },
+        }),
       });
 
       if (retriable) {
@@ -390,6 +397,7 @@ export abstract class ClientTransport<
           type: ProtocolError.HandshakeFailed,
           code: msg.payload.status.code,
           message: reason,
+          ...(extras !== undefined && { extras }),
         });
       }
 

transport/events.ts

@@ -38,6 +38,12 @@ export interface EventMap {
         type: (typeof ProtocolError)['HandshakeFailed'];
         code: Static<typeof HandshakeErrorResponseCodes>;
         message: string;
+        /**
+         * Application-defined rejection data from the server's handshake
+         * handler, transported opaquely. Present only for custom-handler
+         * rejections; validate it with the application's own schema.
+         */
+        extras?: unknown;
       }
     | {
         type: Omit<

transport/message.test.ts

@@ -1,13 +1,17 @@
 import { TransportMessage } from '.';
 import {
+  ControlMessageHandshakeResponseSchema,
   ControlFlags,
+  HandshakeErrorResponseCodes,
   handshakeRequestMessage,
   handshakeResponseMessage,
   isAck,
   isStreamClose,
   isStreamOpen,
 } from './message';
 import { describe, test, expect } from 'vitest';
+import { Type } from 'typebox';
+import { Value } from 'typebox/value';
 
 const msg = (
   to: string,
@@ -105,6 +109,74 @@ describe('message helpers', () => {
     expect(mFail.payload.status.ok).toBe(false);
   });
 
+  test('handshake rejection extras are compatible with older clients', () => {
+    const oldHandshakeResponseSchema = Type.Object({
+      type: Type.Literal('HANDSHAKE_RESP'),
+      status: Type.Union([
+        Type.Object({
+          ok: Type.Literal(true),
+          sessionId: Type.String(),
+        }),
+        Type.Object({
+          ok: Type.Literal(false),
+          reason: Type.String(),
+          code: HandshakeErrorResponseCodes,
+        }),
+      ]),
+    });
+    const payload = {
+      type: 'HANDSHAKE_RESP',
+      status: {
+        ok: false,
+        reason: 'rejected by handshake handler',
+        code: 'REJECTED_BY_CUSTOM_HANDLER',
+        extras: {
+          code: 'TOKEN_EXPIRED',
+          message: 'The authentication token expired',
+        },
+      },
+    };
+
+    expect(Value.Check(oldHandshakeResponseSchema, payload)).toBe(true);
+    expect(Value.Check(ControlMessageHandshakeResponseSchema, payload)).toBe(
+      true,
+    );
+  });
+
+  test('handshake rejection extras accept any application-defined shape', () => {
+    for (const extras of [
+      'TERMINAL_REPL_GONE',
+      { code: 'TOKEN_EXPIRED' },
+      42,
+      null,
+    ]) {
+      expect(
+        Value.Check(ControlMessageHandshakeResponseSchema, {
+          type: 'HANDSHAKE_RESP',
+          status: {
+            ok: false,
+            reason: 'rejected by handshake handler',
+            code: 'REJECTED_BY_CUSTOM_HANDLER',
+            extras,
+          },
+        }),
+      ).toBe(true);
+    }
+  });
+
+  test('handshake rejections without extras remain valid', () => {
+    expect(
+      Value.Check(ControlMessageHandshakeResponseSchema, {
+        type: 'HANDSHAKE_RESP',
+        status: {
+          ok: false,
+          reason: 'rejected by handshake handler',
+          code: 'REJECTED_BY_CUSTOM_HANDLER',
+        },
+      }),
+    ).toBe(true);
+  });
+
   test('default message has no control flags set', () => {
     const m = msg('a', 'b', 'stream', { test: 1 }, 'svc', 'proc');
 

transport/message.ts

@@ -134,6 +134,9 @@ export const ControlMessageHandshakeResponseSchema = Type.Object({
       ok: Type.Literal(false),
       reason: Type.String(),
       code: HandshakeErrorResponseCodes,
+      // Application-defined rejection data. River populates this only for
+      // custom-handler rejections; older peers ignore it.
+      extras: Type.Optional(Type.Unknown()),
     }),
   ]),
 });