simplify proxy, move seq check to no session path

__tests__/fixtures/transports.ts

@@ -1,7 +1,6 @@
 import {
  ClientTransport,
  Connection,
- OpaqueTransportMessageSchema,
  ServerTransport,
  TransportClientId,
 } from '../../transport';
@@ -29,8 +28,6 @@ import {
  ServerHandshakeOptions,
 } from '../../router/handshake';
 import { MessageFramer } from '../../transport/transforms/messageFraming';
-import { BinaryCodec } from '../../codec';
-import { Value } from '@sinclair/typebox/value';
 
 export type ValidTransports = 'ws' | 'unix sockets' | 'ws + uds proxy';
 
@@ -205,8 +202,6 @@ export const transports: Array<TransportMatrixEntry> = [
  let port: number;
  let wss: NodeWs.Server;
 
- const codec = opts?.client?.codec ?? BinaryCodec;
-
  async function setupProxyServer() {
  udsServer = net.createServer();
  await onUdsServeReady(udsServer, socketPath);
@@ -222,7 +217,6 @@ export const transports: Array<TransportMatrixEntry> = [
  wss = createWebSocketServer(proxyServer);
 
  // dumb proxy
- // assume that we are using the binary msgpack protocol
  wss.on('connection', (ws) => {
  const framer = MessageFramer.createFramedStream();
  const uds = net.createConnection(socketPath);
@@ -236,23 +230,11 @@ export const transports: Array<TransportMatrixEntry> = [
  // ws -> uds
  ws.onmessage = (msg) => {
  const data = msg.data as Uint8Array;
- const res = codec.fromBuffer(data);
- if (!res) return;
- if (!Value.Check(OpaqueTransportMessageSchema, res)) {
- return;
- }
-
  uds.write(MessageFramer.write(data));
  };
 
  // ws <- uds
  uds.pipe(framer).on('data', (data: Uint8Array) => {
- const res = codec.fromBuffer(data);
- if (!res) return;
- if (!Value.Check(OpaqueTransportMessageSchema, res)) {
- return;
- }
-
  ws.send(data);
  });
 

transport/server.ts

@@ -282,15 +282,14 @@ export abstract class ServerTransport<
  | 'unknown session'
  | 'transparent reconnection'
  | 'hard reconnection' = 'new session';
+ const clientNextExpectedSeq =
+ msg.payload.expectedSessionState.nextExpectedSeq;
+ const clientNextSentSeq = msg.payload.expectedSessionState.nextSentSeq ?? 0;
+
  if (oldSession && oldSession.id === msg.payload.sessionId) {
  connectCase = 'transparent reconnection';
 
  // invariant: ordering must be correct
- const clientNextExpectedSeq =
- msg.payload.expectedSessionState.nextExpectedSeq;
- // TODO: remove nullish coalescing when we're sure this is always set
- const clientNextSentSeq =
- msg.payload.expectedSessionState.nextSentSeq ?? 0;
  const ourNextSeq = oldSession.nextSeq();
  const ourAck = oldSession.ack;
 
@@ -367,33 +366,34 @@ export abstract class ServerTransport<
  connectCase = 'hard reconnection';
 
  // just nuke the old session entirely and proceed as if this was new
+ this.log?.info(
+ `client is reconnecting to a new session (${msg.payload.sessionId}) with an old session (${oldSession.id}) already existing, closing old session`,
+ {
+ ...session.loggingMetadata,
+ connectedTo: msg.from,
+ sessionId: msg.payload.sessionId,
+ },
+ );
  this.deleteSession(oldSession);
  oldSession = undefined;
- } else {
- connectCase = 'unknown session';
-
- const clientNextExpectedSeq =
- msg.payload.expectedSessionState.nextExpectedSeq;
- // TODO: remove nullish coalescing when we're sure this is always set
- const clientNextSentSeq =
- msg.payload.expectedSessionState.nextSentSeq ?? 0;
+ }
 
-  if (clientNextSentSeq > 0 || clientNextExpectedSeq > 0) {
-  // we don't have a session, but the client is trying to reconnect
-  // to an old session. we can't do anything about this, so we reject
-  this.rejectHandshakeRequest(
-  session,
-  msg.from,
-  `client is trying to reconnect to a session the server don't know about: ${msg.payload.sessionId}`,
-  'SESSION_STATE_MISMATCH',
-  {
-  ...session.loggingMetadata,
-  connectedTo: msg.from,
-  transportMessage: msg,
- },
- );
-  return;
- }
+ if (!oldSession && (clientNextSentSeq > 0 || clientNextExpectedSeq > 0)) {
+ // we don't have a session, but the client is trying to reconnect
+ // to an old session. we can't do anything about this, so we reject
+ connectCase = 'unknown session';
+ this.rejectHandshakeRequest(
+ session,
+ msg.from,
+ `client is trying to reconnect to a session the server don't know about: ${msg.payload.sessionId}`,
+ 'SESSION_STATE_MISMATCH',
+ {
+ ...session.loggingMetadata,
+ connectedTo: msg.from,
+ transportMessage: msg,
+ },
+ );
+ return;
  }
 
  // from this point on, we're committed to connecting

transport/sessionStateMachine/SessionConnected.ts

@@ -101,6 +101,7 @@ export class SessionConnected<
  }
 
  private sendHeartbeat() {
+ this.log?.debug('sending heartbeat', this.loggingMetadata);
  this.send({
  streamId: 'heartbeat',
  controlFlags: ControlFlags.AckBit,

transport/transport.test.ts

@@ -733,10 +733,8 @@ describe.each(testMatrix())(
  await cleanupTransports([clientTransport, serverTransport]);
  });
 
- await waitFor(() => expect(onConnect).toHaveBeenCalledTimes(1));
- closeAllConnections(clientTransport);
  await waitFor(() => {
- expect(onConnect).toHaveBeenCalledTimes(2);
+ expect(onConnect).toHaveBeenCalledTimes(1);
  expect(numberOfConnections(clientTransport)).toEqual(1);
  expect(numberOfConnections(serverTransport)).toEqual(1);
  });
@@ -746,10 +744,9 @@ describe.each(testMatrix())(
  expect(oldClientSessionId).not.toBeUndefined();
  expect(oldServerSessionId).not.toBeUndefined();
 
- // make sure our connection is still intact even after session grace elapses
- await advanceFakeTimersBySessionGrace();
-
+ closeAllConnections(clientTransport);
  await waitFor(() => {
+ expect(onConnect).toHaveBeenCalledTimes(2);
  expect(numberOfConnections(clientTransport)).toEqual(1);
  expect(numberOfConnections(serverTransport)).toEqual(1);
  });