From 013d44f9b270d85a3b0ac7531d248dae25b10eb8 Mon Sep 17 00:00:00 2001 From: Mayank Mehra Date: Thu, 20 Aug 2026 12:11:51 -0700 Subject: [PATCH 1/3] Add structured handshake rejections --- PROTOCOL.md | 8 +++++ README.md | 24 ++++++++++++++- __tests__/e2e.test.ts | 24 ++++++++++----- protobuf/handshake.ts | 4 ++- protobuf/index.ts | 5 ++++ router/handshake.ts | 42 +++++++++++++++++++++++++-- router/index.ts | 5 ++++ transport/client.ts | 9 ++++++ transport/events.ts | 7 ++++- transport/index.ts | 1 + transport/message.test.ts | 51 ++++++++++++++++++++++++++++++++ transport/message.ts | 7 +++++ transport/server.ts | 54 ++++++++++++++++++++++++++++++++-- transport/transport.test.ts | 58 +++++++++++++++++++++++++++++++++++++ 14 files changed, 285 insertions(+), 14 deletions(-) diff --git a/PROTOCOL.md b/PROTOCOL.md index 453b50e2..9b568841 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -258,6 +258,13 @@ interface ControlHandshakeResponse { // fatal, returned by the custom handshake handler | 'REJECTED_BY_CUSTOM_HANDLER' | 'REJECTED_UNSUPPORTED_CLIENT'; + // Application-defined rejection details. Older peers ignore this + // optional field. + details?: { + code: string; + message: string; + extras?: unknown; + }; }; } @@ -628,6 +635,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 `details` to a rejection. River preserves its own `code` for protocol behavior and exposes `details` on the client's `handshake_failed` protocol error event. Applications can use `details.code` for decisions without parsing the human-readable `reason` or `message`. ### Re-handshaking (live credential refresh) diff --git a/README.md b/README.md index e185fd34..d21668d7 100644 --- a/README.md +++ b/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 structured details) // | 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,27 @@ createServer(serverTransport, services, { }); ``` +Use `rejectHandshake` when the client needs a machine-readable reason for an application-level rejection: + +```ts +createServerHandshakeOptions(handshakeSchema, async (metadata) => { + const authenticated = await authenticate(metadata.token); + if (!authenticated.ok) { + return rejectHandshake({ + code: 'TOKEN_EXPIRED', + message: 'The authentication token expired', + extras: { expiredAt: authenticated.expiredAt }, + }); + } + + return { parsedToken: metadata.token }; +}); +``` + +River sends these details on the optional `details` field of the failed handshake response and exposes them on the client's `handshake_failed` protocol error event. Existing failure codes remain available for simple handlers. Do not put secrets or raw internal errors in `message` or `extras` because River sends them to the peer. + +During a re-handshake, River exposes structured rejection details only on the server's `handshake_failed` event before it closes the session. The client's next fresh handshake can receive the details. + `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 diff --git a/__tests__/e2e.test.ts b/__tests__/e2e.test.ts index be360d73..460ea888 100644 --- a/__tests__/e2e.test.ts +++ b/__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 rejectionDetails = { + code: 'TOKEN_EXPIRED', + message: 'The refreshed token expired', + }; + const validate = vi.fn((metadata: ParsedMetadata) => + metadata.token === 'token-v1' + ? { token: metadata.token } + : rejectHandshake(rejectionDetails), ); 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', + details: rejectionDetails, + }); // let the client's now-disconnected session lapse before cleanup await advanceFakeTimersBySessionGrace(); diff --git a/protobuf/handshake.ts b/protobuf/handshake.ts index f441ee4c..bb6ab02e 100644 --- a/protobuf/handshake.ts +++ b/protobuf/handshake.ts @@ -8,6 +8,7 @@ import { createClientHandshakeOptions as createTransportClientHandshakeOptions, createServerHandshakeOptions as createTransportServerHandshakeOptions, type ClientHandshakeOptions, + type HandshakeRejection, type ServerHandshakeOptions, } from '../router/handshake'; import { @@ -34,7 +35,8 @@ type ValidateHandshake = ( ) => | ParsedMetadata | ProtobufHandshakeFailureCode - | Promise; + | HandshakeRejection + | Promise; /** * Create client-side handshake options backed by a protobuf message type. diff --git a/protobuf/index.ts b/protobuf/index.ts index 53d1a8ff..5d2caba3 100644 --- a/protobuf/index.ts +++ b/protobuf/index.ts @@ -21,6 +21,11 @@ export { createClientHandshakeOptions, createServerHandshakeOptions, } from './handshake'; +export { rejectHandshake } from '../router/handshake'; +export type { + HandshakeRejection, + HandshakeRejectionDetails, +} from '../router/handshake'; export { createProtoService } from './service'; export type { AnyProtoService, diff --git a/router/handshake.ts b/router/handshake.ts index 427b86b6..05a5acdd 100644 --- a/router/handshake.ts +++ b/router/handshake.ts @@ -1,9 +1,45 @@ import type { Static, TSchema } from 'typebox'; import { HandshakeErrorCustomHandlerFatalResponseCodes, + HandshakeRejectionDetailsSchema, type TransportClientId, } from '../transport/message'; +const handshakeRejectionBrand: unique symbol = Symbol('handshakeRejection'); + +export type HandshakeRejectionDetails = Static< + typeof HandshakeRejectionDetailsSchema +>; + +export interface HandshakeRejection { + readonly [handshakeRejectionBrand]: true; + responseCode: Static; + details: HandshakeRejectionDetails; +} + +export function rejectHandshake( + details: HandshakeRejectionDetails, + responseCode: Static< + typeof HandshakeErrorCustomHandlerFatalResponseCodes + > = 'REJECTED_BY_CUSTOM_HANDLER', +): HandshakeRejection { + return { + [handshakeRejectionBrand]: true, + responseCode, + details, + }; +} + +export function isHandshakeRejection( + value: unknown, +): value is HandshakeRejection { + return ( + typeof value === 'object' && + value !== null && + handshakeRejectionBrand in value + ); +} + type ConstructHandshake = () => | Static | Promise>; @@ -14,9 +50,11 @@ type ValidateHandshake = ( from?: TransportClientId, ) => | Static + | HandshakeRejection | ParsedMetadata | Promise< | Static + | HandshakeRejection | ParsedMetadata >; @@ -57,8 +95,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 diff --git a/router/index.ts b/router/index.ts index f748101b..4540f02c 100644 --- a/router/index.ts +++ b/router/index.ts @@ -64,5 +64,10 @@ export type { export { createClientHandshakeOptions, createServerHandshakeOptions, + rejectHandshake, +} from './handshake'; +export type { + HandshakeRejection, + HandshakeRejectionDetails, } from './handshake'; export { version as RIVER_VERSION } from '../package.json'; diff --git a/transport/client.ts b/transport/client.ts index 4f9e440a..d5e2c29e 100644 --- a/transport/client.ts +++ b/transport/client.ts @@ -381,6 +381,12 @@ export abstract class ClientTransport< this.rejectHandshakeResponse(session, reason, { ...session.loggingMetadata, transportMessage: msg, + ...(msg.payload.status.details && { + extras: { + ...session.loggingMetadata.extras, + handshakeRejectionDetails: msg.payload.status.details, + }, + }), }); if (retriable) { @@ -390,6 +396,9 @@ export abstract class ClientTransport< type: ProtocolError.HandshakeFailed, code: msg.payload.status.code, message: reason, + ...(msg.payload.status.details && { + details: msg.payload.status.details, + }), }); } diff --git a/transport/events.ts b/transport/events.ts index 5da1f81e..56a16d7f 100644 --- a/transport/events.ts +++ b/transport/events.ts @@ -1,6 +1,10 @@ import type { Static } from 'typebox'; import { Connection } from './connection'; -import { OpaqueTransportMessage, HandshakeErrorResponseCodes } from './message'; +import { + OpaqueTransportMessage, + HandshakeErrorResponseCodes, + HandshakeRejectionDetailsSchema, +} from './message'; import { Session, SessionState } from './sessionStateMachine'; import { SessionId } from './sessionStateMachine/common'; import { TransportStatus } from './transport'; @@ -38,6 +42,7 @@ export interface EventMap { type: (typeof ProtocolError)['HandshakeFailed']; code: Static; message: string; + details?: Static; } | { type: Omit< diff --git a/transport/index.ts b/transport/index.ts index 9bd911e1..843874c7 100644 --- a/transport/index.ts +++ b/transport/index.ts @@ -24,6 +24,7 @@ export { export { TransportMessageSchema, OpaqueTransportMessageSchema, + HandshakeRejectionDetailsSchema, isStreamOpen, isStreamClose, } from './message'; diff --git a/transport/message.test.ts b/transport/message.test.ts index cc2426da..d02b513c 100644 --- a/transport/message.test.ts +++ b/transport/message.test.ts @@ -1,6 +1,8 @@ import { TransportMessage } from '.'; import { + ControlMessageHandshakeResponseSchema, ControlFlags, + HandshakeErrorResponseCodes, handshakeRequestMessage, handshakeResponseMessage, isAck, @@ -8,6 +10,8 @@ import { 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,53 @@ describe('message helpers', () => { expect(mFail.payload.status.ok).toBe(false); }); + test('structured handshake rejections 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', + details: { + code: 'TOKEN_EXPIRED', + message: 'The authentication token expired', + }, + }, + }; + + expect(Value.Check(oldHandshakeResponseSchema, payload)).toBe(true); + expect(Value.Check(ControlMessageHandshakeResponseSchema, payload)).toBe( + true, + ); + }); + + test('handshake rejections without details 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'); diff --git a/transport/message.ts b/transport/message.ts index 5dbd0cd2..5efe9cbf 100644 --- a/transport/message.ts +++ b/transport/message.ts @@ -123,6 +123,12 @@ export const HandshakeErrorResponseCodes = Type.Union([ HandshakeErrorFatalResponseCodes, ]); +export const HandshakeRejectionDetailsSchema = Type.Object({ + code: Type.String(), + message: Type.String(), + extras: Type.Optional(Type.Unknown()), +}); + export const ControlMessageHandshakeResponseSchema = Type.Object({ type: Type.Literal('HANDSHAKE_RESP'), status: Type.Union([ @@ -134,6 +140,7 @@ export const ControlMessageHandshakeResponseSchema = Type.Object({ ok: Type.Literal(false), reason: Type.String(), code: HandshakeErrorResponseCodes, + details: Type.Optional(HandshakeRejectionDetailsSchema), }), ]), }); diff --git a/transport/server.ts b/transport/server.ts index ef42d22d..fe10ba93 100644 --- a/transport/server.ts +++ b/transport/server.ts @@ -1,10 +1,14 @@ import { SpanStatusCode } from '@opentelemetry/api'; -import { ServerHandshakeOptions } from '../router/handshake'; +import { + isHandshakeRejection, + type ServerHandshakeOptions, +} from '../router/handshake'; import { validationErrorToRiverErrors } from '../router/errors'; import { ControlMessageHandshakeRequestSchema, ControlMessageRehandshakeResponseSchema, HandshakeErrorCustomHandlerFatalResponseCodes, + HandshakeRejectionDetailsSchema, HandshakeErrorResponseCodes, OpaqueTransportMessage, acceptedProtocolVersions, @@ -204,6 +208,17 @@ export abstract class ServerTransport< return; } + if (isHandshakeRejection(parsedMetadataOrFailureCode)) { + this.teardownForFailedRehandshake( + session, + 're-handshake metadata rejected by handshake handler', + parsedMetadataOrFailureCode.responseCode, + parsedMetadataOrFailureCode.details, + ); + + return; + } + if ( Value.Check( HandshakeErrorCustomHandlerFatalResponseCodes, @@ -246,6 +261,10 @@ export abstract class ServerTransport< private teardownForFailedRehandshake( session: ServerSession, reason: string, + code: Static< + typeof HandshakeErrorCustomHandlerFatalResponseCodes + > = 'REJECTED_BY_CUSTOM_HANDLER', + details?: Static, ) { if (this.sessions.get(session.to) !== session) { return; @@ -255,12 +274,19 @@ export abstract class ServerTransport< this.log?.warn(`tearing down session to ${to}: ${reason}`, { ...session.loggingMetadata, connectedTo: to, + ...(details && { + extras: { + ...session.loggingMetadata.extras, + handshakeRejectionDetails: details, + }, + }), }); this.protocolError({ type: ProtocolError.HandshakeFailed, - code: 'REJECTED_BY_CUSTOM_HANDLER', + code, message: reason, + ...(details && { details }), }); this.deleteSession(session, { unhealthy: true }); } @@ -352,6 +378,7 @@ export abstract class ServerTransport< reason: string, code: Static, metadata: MessageMetadata, + details?: Static, ) { session.conn.telemetry?.span.setStatus({ code: SpanStatusCode.ERROR, @@ -367,6 +394,7 @@ export abstract class ServerTransport< ok: false, code, reason, + ...(details && { details }), }, }); @@ -390,6 +418,7 @@ export abstract class ServerTransport< type: ProtocolError.HandshakeFailed, code, message: reason, + ...(details && { details }), }); this.deletePendingSession(session); } @@ -493,6 +522,27 @@ export abstract class ServerTransport< } // handler rejected the connection + if (isHandshakeRejection(parsedMetadataOrFailureCode)) { + this.rejectHandshakeRequest( + session, + msg.from, + 'rejected by handshake handler', + parsedMetadataOrFailureCode.responseCode, + { + ...session.loggingMetadata, + connectedTo: msg.from, + clientId: this.clientId, + extras: { + ...session.loggingMetadata.extras, + handshakeRejectionDetails: parsedMetadataOrFailureCode.details, + }, + }, + parsedMetadataOrFailureCode.details, + ); + + return; + } + if ( Value.Check( HandshakeErrorCustomHandlerFatalResponseCodes, diff --git a/transport/transport.test.ts b/transport/transport.test.ts index ba1bc0d8..238b592f 100644 --- a/transport/transport.test.ts +++ b/transport/transport.test.ts @@ -29,6 +29,7 @@ import { ProvidedClientTransportOptions, ProvidedTransportOptions, } from './options'; +import { rejectHandshake } from '../router/handshake'; describe.each(testMatrix())( 'transport connection behaviour tests ($transport.name transport, $codec.name codec)', @@ -1945,5 +1946,62 @@ describe.each(testMatrix())( serverTransport, }); }); + + test('parse can reject connection with structured details', async () => { + const schema = Type.Object({ foo: Type.String() }); + const details = { + code: 'TOKEN_EXPIRED', + message: 'The authentication token expired', + extras: { expiredAt: '2026-08-20T12:00:00Z' }, + }; + const serverTransport = getServerTransport('SERVER', { + schema, + validate: async () => rejectHandshake(details), + }); + const clientTransport = getClientTransport('client', { + schema, + construct: async () => ({ foo: 'foo' }), + }); + const clientHandshakeFailed = vi.fn(); + clientTransport.addEventListener('protocolError', clientHandshakeFailed); + const serverRejectedConnection = vi.fn(); + serverTransport.addEventListener( + 'protocolError', + serverRejectedConnection, + ); + clientTransport.connect(serverTransport.clientId); + + addPostTestCleanup(async () => { + clientTransport.removeEventListener( + 'protocolError', + clientHandshakeFailed, + ); + serverTransport.removeEventListener( + 'protocolError', + serverRejectedConnection, + ); + await cleanupTransports([clientTransport, serverTransport]); + }); + + await waitFor(() => { + expect(clientHandshakeFailed).toHaveBeenCalledWith({ + type: ProtocolError.HandshakeFailed, + code: 'REJECTED_BY_CUSTOM_HANDLER', + message: 'handshake failed: rejected by handshake handler', + details, + }); + expect(serverRejectedConnection).toHaveBeenCalledWith({ + type: ProtocolError.HandshakeFailed, + code: 'REJECTED_BY_CUSTOM_HANDLER', + message: 'rejected by handshake handler', + details, + }); + }); + + await testFinishesCleanly({ + clientTransports: [clientTransport], + serverTransport, + }); + }); }, ); From e72249fda30c5d3a9e04e8b414666165fa8fd608 Mon Sep 17 00:00:00 2001 From: Mayank Mehra Date: Thu, 20 Aug 2026 12:25:03 -0700 Subject: [PATCH 2/3] Simplify handshake rejection handling --- protobuf/handshake.ts | 20 +++++------------- router/handshake.ts | 15 +++++++------- transport/client.ts | 9 ++++---- transport/server.ts | 48 ++++++++++++++++++++++--------------------- 4 files changed, 41 insertions(+), 51 deletions(-) diff --git a/protobuf/handshake.ts b/protobuf/handshake.ts index bb6ab02e..0ede30bf 100644 --- a/protobuf/handshake.ts +++ b/protobuf/handshake.ts @@ -3,27 +3,19 @@ import type { MessageInitShape, MessageShape, } from '@bufbuild/protobuf'; -import { type Static } from 'typebox'; import { createClientHandshakeOptions as createTransportClientHandshakeOptions, createServerHandshakeOptions as createTransportServerHandshakeOptions, type ClientHandshakeOptions, - type HandshakeRejection, + 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 = () => | MessageInitShape | Promise>; @@ -33,10 +25,8 @@ type ValidateHandshake = ( previousParsedMetadata?: ParsedMetadata, from?: TransportClientId, ) => - | ParsedMetadata - | ProtobufHandshakeFailureCode - | HandshakeRejection - | Promise; + | HandshakeValidationResult + | Promise>; /** * Create client-side handshake options backed by a protobuf message type. @@ -75,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); diff --git a/router/handshake.ts b/router/handshake.ts index 05a5acdd..d04dbe49 100644 --- a/router/handshake.ts +++ b/router/handshake.ts @@ -40,6 +40,11 @@ export function isHandshakeRejection( ); } +export type HandshakeValidationResult = + | Static + | HandshakeRejection + | ParsedMetadata; + type ConstructHandshake = () => | Static | Promise>; @@ -49,14 +54,8 @@ type ValidateHandshake = ( previousParsedMetadata?: ParsedMetadata, from?: TransportClientId, ) => - | Static - | HandshakeRejection - | ParsedMetadata - | Promise< - | Static - | HandshakeRejection - | ParsedMetadata - >; + | HandshakeValidationResult + | Promise>; export interface ClientHandshakeOptions< MetadataSchema extends TSchema = TSchema, diff --git a/transport/client.ts b/transport/client.ts index d5e2c29e..1bc73290 100644 --- a/transport/client.ts +++ b/transport/client.ts @@ -377,14 +377,15 @@ export abstract class ClientTransport< ); const reason = `handshake failed: ${msg.payload.status.reason}`; + const { details } = msg.payload.status; const to = session.to; this.rejectHandshakeResponse(session, reason, { ...session.loggingMetadata, transportMessage: msg, - ...(msg.payload.status.details && { + ...(details && { extras: { ...session.loggingMetadata.extras, - handshakeRejectionDetails: msg.payload.status.details, + handshakeRejectionDetails: details, }, }), }); @@ -396,9 +397,7 @@ export abstract class ClientTransport< type: ProtocolError.HandshakeFailed, code: msg.payload.status.code, message: reason, - ...(msg.payload.status.details && { - details: msg.payload.status.details, - }), + ...(details && { details }), }); } diff --git a/transport/server.ts b/transport/server.ts index fe10ba93..181ed5b0 100644 --- a/transport/server.ts +++ b/transport/server.ts @@ -188,9 +188,9 @@ export abstract class ServerTransport< const previousParsedMetadata = this.sessionHandshakeMetadata.get(from); - let parsedMetadataOrFailureCode; + let validationResult; try { - parsedMetadataOrFailureCode = await handshakeExtensions.validate( + validationResult = await handshakeExtensions.validate( metadata, previousParsedMetadata, from, @@ -208,12 +208,12 @@ export abstract class ServerTransport< return; } - if (isHandshakeRejection(parsedMetadataOrFailureCode)) { + if (isHandshakeRejection(validationResult)) { this.teardownForFailedRehandshake( session, 're-handshake metadata rejected by handshake handler', - parsedMetadataOrFailureCode.responseCode, - parsedMetadataOrFailureCode.details, + validationResult.responseCode, + validationResult.details, ); return; @@ -222,7 +222,7 @@ export abstract class ServerTransport< if ( Value.Check( HandshakeErrorCustomHandlerFatalResponseCodes, - parsedMetadataOrFailureCode, + validationResult, ) ) { this.teardownForFailedRehandshake( @@ -240,10 +240,7 @@ export abstract class ServerTransport< return; } - this.storeSessionMetadata( - session, - parsedMetadataOrFailureCode as ParsedMetadata, - ); + this.storeSessionMetadata(session, validationResult as ParsedMetadata); this.log?.info(`re-handshake from ${from} ok`, { ...session.loggingMetadata, @@ -385,7 +382,16 @@ export abstract class ServerTransport< message: reason, }); - this.log?.warn(reason, metadata); + const logMetadata = details + ? { + ...metadata, + extras: { + ...metadata.extras, + handshakeRejectionDetails: details, + }, + } + : metadata; + this.log?.warn(reason, logMetadata); const responseMsg = handshakeResponseMessage({ from: this.clientId, @@ -492,9 +498,9 @@ export abstract class ServerTransport< msg.from, ); - let parsedMetadataOrFailureCode; + let validationResult; try { - parsedMetadataOrFailureCode = await this.handshakeExtensions.validate( + validationResult = await this.handshakeExtensions.validate( msg.payload.metadata, previousParsedMetadata, msg.from, @@ -522,22 +528,18 @@ export abstract class ServerTransport< } // handler rejected the connection - if (isHandshakeRejection(parsedMetadataOrFailureCode)) { + if (isHandshakeRejection(validationResult)) { this.rejectHandshakeRequest( session, msg.from, 'rejected by handshake handler', - parsedMetadataOrFailureCode.responseCode, + validationResult.responseCode, { ...session.loggingMetadata, connectedTo: msg.from, clientId: this.clientId, - extras: { - ...session.loggingMetadata.extras, - handshakeRejectionDetails: parsedMetadataOrFailureCode.details, - }, }, - parsedMetadataOrFailureCode.details, + validationResult.details, ); return; @@ -546,14 +548,14 @@ export abstract class ServerTransport< if ( Value.Check( HandshakeErrorCustomHandlerFatalResponseCodes, - parsedMetadataOrFailureCode, + validationResult, ) ) { this.rejectHandshakeRequest( session, msg.from, 'rejected by handshake handler', - parsedMetadataOrFailureCode, + validationResult, { ...session.loggingMetadata, connectedTo: msg.from, @@ -565,7 +567,7 @@ export abstract class ServerTransport< } // success! - parsedMetadata = parsedMetadataOrFailureCode as ParsedMetadata; + parsedMetadata = validationResult as ParsedMetadata; } // 4 connect cases From 3d6d7452f6c30f96445f556a33440dd3251d9251 Mon Sep 17 00:00:00 2001 From: Mayank Mehra Date: Mon, 24 Aug 2026 11:51:26 -0700 Subject: [PATCH 3/3] Make handshake rejection extras opaque and application-defined --- PROTOCOL.md | 12 ++++-------- README.md | 9 ++++----- __tests__/e2e.test.ts | 6 +++--- protobuf/index.ts | 5 +---- router/handshake.ts | 16 ++++++++-------- router/index.ts | 5 +---- transport/client.ts | 8 ++++---- transport/events.ts | 13 +++++++------ transport/index.ts | 1 - transport/message.test.ts | 27 +++++++++++++++++++++++--- transport/message.ts | 10 +++------- transport/server.ts | 38 ++++++++++++++++++------------------- transport/transport.test.ts | 11 +++++------ 13 files changed, 83 insertions(+), 78 deletions(-) diff --git a/PROTOCOL.md b/PROTOCOL.md index 9b568841..0c1dc22b 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -258,13 +258,9 @@ interface ControlHandshakeResponse { // fatal, returned by the custom handshake handler | 'REJECTED_BY_CUSTOM_HANDLER' | 'REJECTED_UNSUPPORTED_CLIENT'; - // Application-defined rejection details. Older peers ignore this - // optional field. - details?: { - code: string; - message: string; - extras?: unknown; - }; + // Application-defined rejection data. River populates this only for + // custom-handler rejections. Older peers ignore this optional field. + extras?: unknown; }; } @@ -635,7 +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 `details` to a rejection. River preserves its own `code` for protocol behavior and exposes `details` on the client's `handshake_failed` protocol error event. Applications can use `details.code` for decisions without parsing the human-readable `reason` or `message`. +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) diff --git a/README.md b/README.md index d21668d7..ee0a9597 100644 --- a/README.md +++ b/README.md @@ -807,7 +807,7 @@ createServer(serverTransport, services, { // from?: TransportClientId, // ) => // | 'REJECTED_BY_CUSTOM_HANDLER' | 'REJECTED_UNSUPPORTED_CLIENT' (if you reject it) - // | HandshakeRejection (if you reject it with structured details) + // | HandshakeRejection (if you reject it with application-defined extras) // | ParsedMetadata (if you allow it) // | a Promise of any of the above // @@ -821,7 +821,7 @@ createServer(serverTransport, services, { }); ``` -Use `rejectHandshake` when the client needs a machine-readable reason for an application-level rejection: +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) => { @@ -830,7 +830,6 @@ createServerHandshakeOptions(handshakeSchema, async (metadata) => { return rejectHandshake({ code: 'TOKEN_EXPIRED', message: 'The authentication token expired', - extras: { expiredAt: authenticated.expiredAt }, }); } @@ -838,9 +837,9 @@ createServerHandshakeOptions(handshakeSchema, async (metadata) => { }); ``` -River sends these details on the optional `details` field of the failed handshake response and exposes them on the client's `handshake_failed` protocol error event. Existing failure codes remain available for simple handlers. Do not put secrets or raw internal errors in `message` or `extras` because River sends them to the peer. +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 structured rejection details only on the server's `handshake_failed` event before it closes the session. The client's next fresh handshake can receive the details. +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` diff --git a/__tests__/e2e.test.ts b/__tests__/e2e.test.ts index 460ea888..abded760 100644 --- a/__tests__/e2e.test.ts +++ b/__tests__/e2e.test.ts @@ -1484,14 +1484,14 @@ describe.each(testMatrix())( 'client', createClientHandshakeOptions(requestSchema, construct), ); - const rejectionDetails = { + const rejectionExtras = { code: 'TOKEN_EXPIRED', message: 'The refreshed token expired', }; const validate = vi.fn((metadata: ParsedMetadata) => metadata.token === 'token-v1' ? { token: metadata.token } - : rejectHandshake(rejectionDetails), + : rejectHandshake(rejectionExtras), ); const serverTransport = getServerTransport< typeof requestSchema, @@ -1547,7 +1547,7 @@ describe.each(testMatrix())( type: 'handshake_failed', code: 'REJECTED_BY_CUSTOM_HANDLER', message: 're-handshake metadata rejected by handshake handler', - details: rejectionDetails, + extras: rejectionExtras, }); // let the client's now-disconnected session lapse before cleanup diff --git a/protobuf/index.ts b/protobuf/index.ts index 5d2caba3..5a3df140 100644 --- a/protobuf/index.ts +++ b/protobuf/index.ts @@ -22,10 +22,7 @@ export { createServerHandshakeOptions, } from './handshake'; export { rejectHandshake } from '../router/handshake'; -export type { - HandshakeRejection, - HandshakeRejectionDetails, -} from '../router/handshake'; +export type { HandshakeRejection } from '../router/handshake'; export { createProtoService } from './service'; export type { AnyProtoService, diff --git a/router/handshake.ts b/router/handshake.ts index d04dbe49..9c65f090 100644 --- a/router/handshake.ts +++ b/router/handshake.ts @@ -1,24 +1,24 @@ import type { Static, TSchema } from 'typebox'; import { HandshakeErrorCustomHandlerFatalResponseCodes, - HandshakeRejectionDetailsSchema, type TransportClientId, } from '../transport/message'; const handshakeRejectionBrand: unique symbol = Symbol('handshakeRejection'); -export type HandshakeRejectionDetails = Static< - typeof HandshakeRejectionDetailsSchema ->; - export interface HandshakeRejection { readonly [handshakeRejectionBrand]: true; responseCode: Static; - details: HandshakeRejectionDetails; + /** + * 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( - details: HandshakeRejectionDetails, + extras: unknown, responseCode: Static< typeof HandshakeErrorCustomHandlerFatalResponseCodes > = 'REJECTED_BY_CUSTOM_HANDLER', @@ -26,7 +26,7 @@ export function rejectHandshake( return { [handshakeRejectionBrand]: true, responseCode, - details, + extras, }; } diff --git a/router/index.ts b/router/index.ts index 4540f02c..d21ca96d 100644 --- a/router/index.ts +++ b/router/index.ts @@ -66,8 +66,5 @@ export { createServerHandshakeOptions, rejectHandshake, } from './handshake'; -export type { - HandshakeRejection, - HandshakeRejectionDetails, -} from './handshake'; +export type { HandshakeRejection } from './handshake'; export { version as RIVER_VERSION } from '../package.json'; diff --git a/transport/client.ts b/transport/client.ts index 1bc73290..177c2cee 100644 --- a/transport/client.ts +++ b/transport/client.ts @@ -377,15 +377,15 @@ export abstract class ClientTransport< ); const reason = `handshake failed: ${msg.payload.status.reason}`; - const { details } = msg.payload.status; + const { extras } = msg.payload.status; const to = session.to; this.rejectHandshakeResponse(session, reason, { ...session.loggingMetadata, transportMessage: msg, - ...(details && { + ...(extras !== undefined && { extras: { ...session.loggingMetadata.extras, - handshakeRejectionDetails: details, + handshakeRejectionExtras: extras, }, }), }); @@ -397,7 +397,7 @@ export abstract class ClientTransport< type: ProtocolError.HandshakeFailed, code: msg.payload.status.code, message: reason, - ...(details && { details }), + ...(extras !== undefined && { extras }), }); } diff --git a/transport/events.ts b/transport/events.ts index 56a16d7f..27ed2663 100644 --- a/transport/events.ts +++ b/transport/events.ts @@ -1,10 +1,6 @@ import type { Static } from 'typebox'; import { Connection } from './connection'; -import { - OpaqueTransportMessage, - HandshakeErrorResponseCodes, - HandshakeRejectionDetailsSchema, -} from './message'; +import { OpaqueTransportMessage, HandshakeErrorResponseCodes } from './message'; import { Session, SessionState } from './sessionStateMachine'; import { SessionId } from './sessionStateMachine/common'; import { TransportStatus } from './transport'; @@ -42,7 +38,12 @@ export interface EventMap { type: (typeof ProtocolError)['HandshakeFailed']; code: Static; message: string; - details?: Static; + /** + * 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< diff --git a/transport/index.ts b/transport/index.ts index 843874c7..9bd911e1 100644 --- a/transport/index.ts +++ b/transport/index.ts @@ -24,7 +24,6 @@ export { export { TransportMessageSchema, OpaqueTransportMessageSchema, - HandshakeRejectionDetailsSchema, isStreamOpen, isStreamClose, } from './message'; diff --git a/transport/message.test.ts b/transport/message.test.ts index d02b513c..f4e72b7d 100644 --- a/transport/message.test.ts +++ b/transport/message.test.ts @@ -109,7 +109,7 @@ describe('message helpers', () => { expect(mFail.payload.status.ok).toBe(false); }); - test('structured handshake rejections are compatible with older clients', () => { + test('handshake rejection extras are compatible with older clients', () => { const oldHandshakeResponseSchema = Type.Object({ type: Type.Literal('HANDSHAKE_RESP'), status: Type.Union([ @@ -130,7 +130,7 @@ describe('message helpers', () => { ok: false, reason: 'rejected by handshake handler', code: 'REJECTED_BY_CUSTOM_HANDLER', - details: { + extras: { code: 'TOKEN_EXPIRED', message: 'The authentication token expired', }, @@ -143,7 +143,28 @@ describe('message helpers', () => { ); }); - test('handshake rejections without details remain valid', () => { + 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', diff --git a/transport/message.ts b/transport/message.ts index 5efe9cbf..a78e3ae7 100644 --- a/transport/message.ts +++ b/transport/message.ts @@ -123,12 +123,6 @@ export const HandshakeErrorResponseCodes = Type.Union([ HandshakeErrorFatalResponseCodes, ]); -export const HandshakeRejectionDetailsSchema = Type.Object({ - code: Type.String(), - message: Type.String(), - extras: Type.Optional(Type.Unknown()), -}); - export const ControlMessageHandshakeResponseSchema = Type.Object({ type: Type.Literal('HANDSHAKE_RESP'), status: Type.Union([ @@ -140,7 +134,9 @@ export const ControlMessageHandshakeResponseSchema = Type.Object({ ok: Type.Literal(false), reason: Type.String(), code: HandshakeErrorResponseCodes, - details: Type.Optional(HandshakeRejectionDetailsSchema), + // Application-defined rejection data. River populates this only for + // custom-handler rejections; older peers ignore it. + extras: Type.Optional(Type.Unknown()), }), ]), }); diff --git a/transport/server.ts b/transport/server.ts index 181ed5b0..bb76c426 100644 --- a/transport/server.ts +++ b/transport/server.ts @@ -8,7 +8,6 @@ import { ControlMessageHandshakeRequestSchema, ControlMessageRehandshakeResponseSchema, HandshakeErrorCustomHandlerFatalResponseCodes, - HandshakeRejectionDetailsSchema, HandshakeErrorResponseCodes, OpaqueTransportMessage, acceptedProtocolVersions, @@ -213,7 +212,7 @@ export abstract class ServerTransport< session, 're-handshake metadata rejected by handshake handler', validationResult.responseCode, - validationResult.details, + validationResult.extras, ); return; @@ -261,7 +260,7 @@ export abstract class ServerTransport< code: Static< typeof HandshakeErrorCustomHandlerFatalResponseCodes > = 'REJECTED_BY_CUSTOM_HANDLER', - details?: Static, + rejectionExtras?: unknown, ) { if (this.sessions.get(session.to) !== session) { return; @@ -271,10 +270,10 @@ export abstract class ServerTransport< this.log?.warn(`tearing down session to ${to}: ${reason}`, { ...session.loggingMetadata, connectedTo: to, - ...(details && { + ...(rejectionExtras !== undefined && { extras: { ...session.loggingMetadata.extras, - handshakeRejectionDetails: details, + handshakeRejectionExtras: rejectionExtras, }, }), }); @@ -283,7 +282,7 @@ export abstract class ServerTransport< type: ProtocolError.HandshakeFailed, code, message: reason, - ...(details && { details }), + ...(rejectionExtras !== undefined && { extras: rejectionExtras }), }); this.deleteSession(session, { unhealthy: true }); } @@ -375,22 +374,23 @@ export abstract class ServerTransport< reason: string, code: Static, metadata: MessageMetadata, - details?: Static, + rejectionExtras?: unknown, ) { session.conn.telemetry?.span.setStatus({ code: SpanStatusCode.ERROR, message: reason, }); - const logMetadata = details - ? { - ...metadata, - extras: { - ...metadata.extras, - handshakeRejectionDetails: details, - }, - } - : metadata; + const logMetadata = + rejectionExtras !== undefined + ? { + ...metadata, + extras: { + ...metadata.extras, + handshakeRejectionExtras: rejectionExtras, + }, + } + : metadata; this.log?.warn(reason, logMetadata); const responseMsg = handshakeResponseMessage({ @@ -400,7 +400,7 @@ export abstract class ServerTransport< ok: false, code, reason, - ...(details && { details }), + ...(rejectionExtras !== undefined && { extras: rejectionExtras }), }, }); @@ -424,7 +424,7 @@ export abstract class ServerTransport< type: ProtocolError.HandshakeFailed, code, message: reason, - ...(details && { details }), + ...(rejectionExtras !== undefined && { extras: rejectionExtras }), }); this.deletePendingSession(session); } @@ -539,7 +539,7 @@ export abstract class ServerTransport< connectedTo: msg.from, clientId: this.clientId, }, - validationResult.details, + validationResult.extras, ); return; diff --git a/transport/transport.test.ts b/transport/transport.test.ts index 238b592f..e66ebf9a 100644 --- a/transport/transport.test.ts +++ b/transport/transport.test.ts @@ -1947,16 +1947,15 @@ describe.each(testMatrix())( }); }); - test('parse can reject connection with structured details', async () => { + test('parse can reject connection with application-defined extras', async () => { const schema = Type.Object({ foo: Type.String() }); - const details = { + const rejectionExtras = { code: 'TOKEN_EXPIRED', message: 'The authentication token expired', - extras: { expiredAt: '2026-08-20T12:00:00Z' }, }; const serverTransport = getServerTransport('SERVER', { schema, - validate: async () => rejectHandshake(details), + validate: async () => rejectHandshake(rejectionExtras), }); const clientTransport = getClientTransport('client', { schema, @@ -1988,13 +1987,13 @@ describe.each(testMatrix())( type: ProtocolError.HandshakeFailed, code: 'REJECTED_BY_CUSTOM_HANDLER', message: 'handshake failed: rejected by handshake handler', - details, + extras: rejectionExtras, }); expect(serverRejectedConnection).toHaveBeenCalledWith({ type: ProtocolError.HandshakeFailed, code: 'REJECTED_BY_CUSTOM_HANDLER', message: 'rejected by handshake handler', - details, + extras: rejectionExtras, }); });