[protocol] Pre-session handshake

.github/workflows/ci.yaml

@@ -3,7 +3,7 @@ name: Build and Test
 on:
   pull_request:
     branches:
-      - main
+      - '**'
   push:
     branches:
       - main

__tests__/fixtures/cleanup.ts

@@ -117,7 +117,6 @@ export async function testFinishesCleanly({
 
   // server sits on top of server transport so we clean it up first
   if (server) {
-    await advanceFakeTimersByDisconnectGrace();
     await ensureServerIsClean(server);
     await server.close();
   }

__tests__/fixtures/transports.ts

@@ -1,4 +1,4 @@
-import { Connection, Transport } from '../../transport';
+import { ClientTransport, Connection, ServerTransport } from '../../transport';
 import http from 'node:http';
 import net from 'node:net';
 import {
@@ -14,8 +14,10 @@ import { UnixDomainSocketServerTransport } from '../../transport/impls/uds/serve
 export const transports: Array<{
   name: string;
   setup: () => Promise<{
-    // client, server
-    getTransports: () => [Transport<Connection>, Transport<Connection>];
+    getTransports: () => [
+      ClientTransport<Connection>,
+      ServerTransport<Connection>,
+    ];
     cleanup: () => Promise<void>;
   }>;
 }> = [

__tests__/typescript-stress.test.ts

@@ -3,7 +3,7 @@ import { Procedure, ServiceBuilder, serializeService } from '../router/builder';
 import { Type } from '@sinclair/typebox';
 import { OpaqueTransportMessage } from '../transport/message';
 import { createServer } from '../router/server';
-import { Transport, Connection } from '../transport';
+import { Transport, Connection, Session } from '../transport';
 import { createClient } from '../router/client';
 import { Ok } from '../router/result';
 import { buildServiceDefs } from '../router/defs';
@@ -97,6 +97,13 @@ export const StupidlyLargeService = <Name extends string>(name: Name) =>
 
 // mock transport
 export class MockTransport extends Transport<Connection> {
+  receiveWithBootSequence(
+    _conn: Connection,
+    _sessionCb: (sess: Session<Connection>) => void,
+  ): (data: Uint8Array) => void {
+    throw new Error('Method not implemented.');
+  }
+
   constructor(clientId: string) {
     super(clientId);
   }
@@ -105,6 +112,8 @@ export class MockTransport extends Transport<Connection> {
     return true;
   }
 
+  protected handleConnection(_conn: Connection, _to: string): void {}
+
   async createNewOutgoingConnection(): Promise<Connection> {
     throw new Error('unimplemented');
   }

router/context.ts

@@ -18,9 +18,7 @@
  * }
  * ```
  */
-export interface ServiceContext {
-  state: object | unknown;
-}
+export interface ServiceContext {}
 
 /**
  * The {@link ServiceContext} with state. This is what is passed to procedures.

transport/impls/stdio/client.ts

@@ -1,14 +1,14 @@
 import { log } from '../../../logging';
 import { TransportClientId } from '../../message';
-import { Transport, TransportOptions } from '../../transport';
+import { ClientTransport, TransportOptions } from '../../transport';
 import { StreamConnection } from './connection';
 
 /**
  * A client-side transport implementation that uses standard input and output streams.
  * Will eagerly connect as soon as it's initialized.
  * @extends Transport
  */
-export class StdioClientTransport extends Transport<StreamConnection> {
+export class StdioClientTransport extends ClientTransport<StreamConnection> {
   input: NodeJS.ReadableStream = process.stdin;
   output: NodeJS.WritableStream = process.stdout;
   serverId: TransportClientId;
@@ -36,15 +36,7 @@ export class StdioClientTransport extends Transport<StreamConnection> {
   async createNewOutgoingConnection(to: TransportClientId) {
     log?.info(`${this.clientId} -- establishing a new stream to ${to}`);
     const conn = new StreamConnection(this.input, this.output);
-    conn.addDataListener((data) => this.handleMsg(this.parseMsg(data)));
-    const cleanup = () => {
-      this.onDisconnect(conn, to);
-      this.connect(to);
-    };
-
-    this.input.addListener('close', cleanup);
-    this.output.addListener('close', cleanup);
-    this.onConnect(conn, to);
+    this.handleConnection(conn, to);
     return conn;
   }
 }

transport/impls/stdio/connection.ts

@@ -20,11 +20,20 @@ export class StreamConnection extends Connection {
     this.input.on('data', cb);
   }
 
+  addCloseListener(cb: () => void): void {
+    this.input.on('close', cb);
+    this.output.on('close', cb);
+  }
+
+  addErrorListener(cb: (err: Error) => void): void {
+    this.input.on('error', cb);
+    this.output.on('error', cb);
+  }
+
   send(payload: Uint8Array) {
     return this.output.write(MessageFramer.write(payload));
   }
 
-  // doesn't touch the underlying connection
   async close() {
     this.output.end();
     this.input.unpipe(this.framer);

transport/impls/stdio/server.ts

@@ -1,15 +1,13 @@
-import { log } from '../../../logging';
 import { TransportClientId } from '../../message';
-import { Session } from '../../session';
-import { Transport, TransportOptions } from '../../transport';
+import { ServerTransport, TransportOptions } from '../../transport';
 import { StreamConnection } from './connection';
 
 /**
  * A server-side transport implementation that uses standard input and output streams.
  * This will sit idle until a client connects.
  * @extends Transport
  */
-export class StdioServerTransport extends Transport<StreamConnection> {
+export class StdioServerTransport extends ServerTransport<StreamConnection> {
   input: NodeJS.ReadableStream = process.stdin;
   output: NodeJS.WritableStream = process.stdout;
 
@@ -28,48 +26,7 @@ export class StdioServerTransport extends Transport<StreamConnection> {
     super(clientId, providedOptions);
     this.input = input;
     this.output = output;
-
-    let session: Session<StreamConnection> | undefined = undefined;
-    const receiver = () => session?.connectedTo ?? 'unknown';
-
     const conn = new StreamConnection(this.input, this.output);
-    conn.addDataListener((data) => {
-      const parsed = this.parseMsg(data);
-      if (!parsed) return;
-      if (!session && !this.connections.has(parsed.from)) {
-        session = this.onConnect(conn, parsed.from);
-      }
-
-      this.handleMsg(parsed);
-    });
-
-    const cleanup = (session: Session<StreamConnection>) =>
-      this.onDisconnect(conn, session.connectedTo);
-
-    this.input.on('close', () => {
-      if (!session) return;
-      log?.info(
-        `${this.clientId} -- stream conn (id: ${
-          conn.debugId
-        }) to ${receiver()} disconnected`,
-      );
-      cleanup(session);
-    });
-
-    this.input.on('error', (err) => {
-      if (!session) return;
-      log?.warn(
-        `${this.clientId} -- error in stream connection (id: ${
-          conn.debugId
-        }) to ${receiver()}: ${err}`,
-      );
-      cleanup(session);
-    });
-  }
-
-  async createNewOutgoingConnection(to: string): Promise<StreamConnection> {
-    const err = `${this.clientId} -- failed to send msg to ${to}, client probably dropped`;
-    log?.warn(err);
-    throw new Error(err);
+    this.handleConnection(conn);
   }
 }

transport/impls/uds/client.ts

@@ -1,10 +1,10 @@
-import { Transport, TransportClientId } from '../..';
+import { TransportClientId } from '../..';
 import { Socket } from 'node:net';
 import { log } from '../../../logging';
 import { UdsConnection } from './connection';
-import { TransportOptions } from '../../transport';
+import { ClientTransport, TransportOptions } from '../../transport';
 
-export class UnixDomainSocketClientTransport extends Transport<UdsConnection> {
+export class UnixDomainSocketClientTransport extends ClientTransport<UdsConnection> {
   path: string;
   serverId: TransportClientId;
 
@@ -20,7 +20,7 @@ export class UnixDomainSocketClientTransport extends Transport<UdsConnection> {
     this.connect(serverId);
   }
 
-  async createNewOutgoingConnection(to: string) {
+  async createNewOutgoingConnection(to: TransportClientId) {
     const oldConnection = this.connections.get(to);
     if (oldConnection) {
       oldConnection.close();
@@ -35,20 +35,7 @@ export class UnixDomainSocketClientTransport extends Transport<UdsConnection> {
     });
 
     const conn = new UdsConnection(sock);
-    conn.addDataListener((data) => this.handleMsg(this.parseMsg(data)));
-    const cleanup = () => {
-      this.onDisconnect(conn, to);
-      this.connect(to);
-    };
-
-    sock.on('close', cleanup);
-    sock.on('error', (err) => {
-      log?.warn(
-        `${this.clientId} -- socket error in connection (id: ${conn.debugId}) to ${to}: ${err}`,
-      );
-    });
-
-    this.onConnect(conn, to);
+    this.handleConnection(conn, to);
     return conn;
   }
 }

transport/impls/uds/connection.ts

@@ -21,6 +21,14 @@ export class UdsConnection extends Connection {
     this.input.on('data', cb);
   }
 
+  addCloseListener(cb: () => void): void {
+    this.sock.on('close', cb);
+  }
+
+  addErrorListener(cb: (err: Error) => void): void {
+    this.sock.on('error', cb);
+  }
+
   send(payload: Uint8Array) {
     return this.sock.write(MessageFramer.write(payload));
   }

transport/impls/uds/server.ts

@@ -1,11 +1,10 @@
-import { Session } from '../../session';
 import { log } from '../../../logging';
 import { type Server, type Socket } from 'node:net';
-import { Transport, TransportOptions } from '../../transport';
+import { ServerTransport, TransportOptions } from '../../transport';
 import { TransportClientId } from '../../message';
 import { UdsConnection } from './connection';
 
-export class UnixDomainSocketServerTransport extends Transport<UdsConnection> {
+export class UnixDomainSocketServerTransport extends ServerTransport<UdsConnection> {
   server: Server;
 
   constructor(
@@ -22,48 +21,13 @@ export class UnixDomainSocketServerTransport extends Transport<UdsConnection> {
   }
 
   connectionHandler = (sock: Socket) => {
-    let session: Session<UdsConnection> | undefined = undefined;
     const conn = new UdsConnection(sock);
+    this.handleConnection(conn);
     log?.info(
       `${this.clientId} -- new incoming uds connection (id: ${conn.debugId})`,
     );
-
-    const client = () => session?.connectedTo ?? 'unknown';
-    conn.addDataListener((data) => {
-      const parsed = this.parseMsg(data);
-      if (!parsed) return;
-      if (!session) {
-        session = this.onConnect(conn, parsed.from);
-      }
-
-      this.handleMsg(parsed);
-    });
-
-    sock.on('close', () => {
-      if (!session) return;
-      log?.info(
-        `${this.clientId} -- uds (id: ${
-          conn.debugId
-        }) to ${client()} disconnected`,
-      );
-      this.onDisconnect(conn, session?.connectedTo);
-    });
-
-    sock.on('error', (err) => {
-      log?.warn(
-        `${this.clientId} -- uds (id: ${
-          conn.debugId
-        }) to ${client()} got an error: ${err}`,
-      );
-    });
   };
 
-  async createNewOutgoingConnection(to: string): Promise<UdsConnection> {
-    const err = `${this.clientId} -- failed to send msg to ${to}, client probably dropped`;
-    log?.warn(err);
-    throw new Error(err);
-  }
-
   async close() {
     super.close();
     this.server.removeListener('connection', this.connectionHandler);

transport/impls/ws/client.ts

@@ -1,5 +1,5 @@
 import WebSocket from 'isomorphic-ws';
-import { Transport, TransportOptions } from '../../transport';
+import { ClientTransport, TransportOptions } from '../../transport';
 import { TransportClientId } from '../../message';
 import { log } from '../../../logging';
 import { WebSocketConnection } from './connection';
@@ -11,7 +11,7 @@ type WebSocketResult = { ws: WebSocket } | { err: string };
  * @class
  * @extends Transport
  */
-export class WebSocketClientTransport extends Transport<WebSocketConnection> {
+export class WebSocketClientTransport extends ClientTransport<WebSocketConnection> {
   /**
    * A function that returns a Promise that resolves to a WebSocket instance.
    */
@@ -34,7 +34,6 @@ export class WebSocketClientTransport extends Transport<WebSocketConnection> {
     super(sessionId, providedOptions);
     this.wsGetter = wsGetter;
     this.serverId = serverId;
-    this.inflightConnectionPromises = new Map();
 
     // eagerly connect as soon as we initialize
     this.connect(this.serverId);
@@ -86,22 +85,7 @@ export class WebSocketClientTransport extends Transport<WebSocketConnection> {
       log?.info(
         `${this.clientId} -- websocket (id: ${conn.debugId}) to ${to} ok`,
       );
-      this.onConnect(conn, to);
-      conn.addDataListener((data) => this.handleMsg(this.parseMsg(data)));
-      wsRes.ws.onclose = () => {
-        log?.info(
-          `${this.clientId} -- websocket (id: ${conn.debugId}) to ${to} disconnected`,
-        );
-        this.onDisconnect(conn, to);
-        this.connect(to);
-      };
-
-      wsRes.ws.onerror = (msg) => {
-        log?.warn(
-          `${this.clientId} -- websocket (id: ${conn.debugId}) to ${to} had an error: ${msg}`,
-        );
-      };
-
+      this.handleConnection(conn, to);
       return conn;
     } else {
       throw new Error(wsRes.err);
@@ -110,11 +94,9 @@ export class WebSocketClientTransport extends Transport<WebSocketConnection> {
 
   async close() {
     super.close();
-    this.inflightConnectionPromises.clear();
   }
 
   async destroy() {
     super.destroy();
-    this.inflightConnectionPromises.clear();
   }
 }