foreign/node/src/client/client.socket.test.ts

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

import assert from 'node:assert/strict';
import { once } from 'node:events';
import { readFileSync } from 'node:fs';
import type { AddressInfo, Socket } from 'node:net';
import { createServer, type Server } from 'node:net';
import { describe, it } from 'node:test';
import { createServer as createTlsServer } from 'node:tls';
import { COMMAND_CODE } from '../wire/command.code.js';
import { LOGIN, LOGIN_WITH_TOKEN, LOGOUT } from '../wire/index.js';
import { UPDATE_USER } from '../wire/user/update-user.command.js';
import { CHANGE_PASSWORD } from '../wire/user/change-password.command.js';
import { POLL_MESSAGES } from '../wire/message/poll-messages.command.js';
import { SEND_MESSAGES, sendMessages } from '../wire/message/send-messages.command.js';
import { GET_OFFSET } from '../wire/offset/get-offset.command.js';
import { STORE_OFFSET } from '../wire/offset/store-offset.command.js';
import { DELETE_OFFSET } from '../wire/offset/delete-offset.command.js';
import { Partitioning } from '../wire/message/partitioning.utils.js';
import { PollingStrategy } from '../wire/message/poll.utils.js';
import { ConsumerKind } from '../wire/offset/offset.utils.js';
import { readWireName } from '../wire/vsr/index.js';
import { BIND_SECRET_BYTES } from '../wire/vsr/register.js';
import { ResponseError } from '../wire/error.utils.js';
import {
  Command,
  EVICTION_OFFSET,
  EvictionReason,
  HEADER_SIZE,
  REPLY_OFFSET,
  REQUEST_OFFSET
} from '../wire/vsr/header.js';
import { Operation } from '../wire/vsr/operation.js';
import { VsrEvictionError } from '../wire/vsr/reply.js';
import { CommandResponseStream } from './client.socket.js';
import type { ClientConfig, CommandResponse } from './client.type.js';

const TEST_SESSION = 42n;
const HISTORY_UNAVAILABLE = 87;
const TRANSIENT_NOT_ACCEPTED = 58;
const TLS_CERTIFICATE = readFileSync(
  new URL('../../../../core/certs/iggy_cert.pem', import.meta.url)
);
const TLS_KEY = readFileSync(
  new URL('../../../../core/certs/iggy_key.pem', import.meta.url)
);
const TLS_CA_CERTIFICATE = readFileSync(
  new URL('../../../../core/certs/iggy_ca_cert.pem', import.meta.url)
);

type FrameHandler = (frame: Buffer, socket: Socket) => void;

type VsrTestServer = {
  port: number,
  frames: Buffer[],
  close: () => Promise<void>
};

/** Loopback server speaking just enough VSR framing for the client tests. */
const startVsrServer = async (
  handler: FrameHandler,
  transport: 'TCP' | 'TLS' = 'TCP'
): Promise<VsrTestServer> => {
  const frames: Buffer[] = [];
  const handleConnection = (socket: Socket) => {
    let pending = Buffer.alloc(0);
    socket.on('data', (data: Buffer) => {
      pending = Buffer.concat([pending, data]);
      while (pending.length >= HEADER_SIZE) {
        const size = pending.readUInt32LE(REQUEST_OFFSET.size);
        if (pending.length < size) break;
        const frame = pending.subarray(0, size);
        pending = pending.subarray(size);
        frames.push(Buffer.from(frame));
        handler(Buffer.from(frame), socket);
      }
    });
    socket.on('error', () => {});
  };
  const server: Server = transport === 'TLS'
    ? createTlsServer(
      { cert: TLS_CERTIFICATE, key: TLS_KEY },
      handleConnection
    )
    : createServer(handleConnection);
  server.listen(0, '127.0.0.1');
  await once(server, 'listening');
  return {
    port: (server.address() as AddressInfo).port,
    frames,
    close: () => new Promise((resolve) => server.close(() => resolve()))
  };
};

const replyFrame = (
  operation: number,
  body: Buffer = Buffer.alloc(0),
  status = 0
): Buffer => {
  const frame = Buffer.alloc(HEADER_SIZE + body.length);
  frame.writeUInt32LE(frame.length, REPLY_OFFSET.size);
  frame.writeUInt8(Command.Reply, REPLY_OFFSET.command);
  frame.writeUInt8(operation, REPLY_OFFSET.operation);
  frame.writeUInt32LE(status, REPLY_OFFSET.status);
  body.copy(frame, HEADER_SIZE);
  return frame;
};

const evictionFrame = (reason: number): Buffer => {
  const frame = Buffer.alloc(HEADER_SIZE);
  frame.writeUInt32LE(HEADER_SIZE, REPLY_OFFSET.size);
  frame.writeUInt8(Command.Eviction, REPLY_OFFSET.command);
  frame.writeUInt8(reason, EVICTION_OFFSET.reason);
  return frame;
};

const bindingIdentity = (frame: Buffer): Buffer => frame.subarray(-64, -32);

const assertSharedBinding = (register: Buffer, binding: Buffer): void => {
  assert.deepEqual(bindingIdentity(binding).subarray(0, 16),
    register.subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16));
  assert.equal(bindingIdentity(binding).readBigUInt64LE(16), TEST_SESSION);
  const sdkName = readWireName(register, HEADER_SIZE + 4);
  const sdkVersion = readWireName(register, sdkName.next);
  assert.deepEqual(binding.subarray(-BIND_SECRET_BYTES),
    register.subarray(sdkVersion.next, sdkVersion.next + BIND_SECRET_BYTES));
};

const registerReplyBody = (session = TEST_SESSION): Buffer => {
  const serverVersion = Buffer.from('0.0.0');
  const body = Buffer.alloc(4 + 17 + serverVersion.length);
  body.writeUInt32LE(0, 0);
  body.writeUInt32LE(7, 4);
  body.writeBigUInt64LE(session, 8);
  body.writeUInt32LE(0, 16);
  body.writeUInt8(serverVersion.length, 20);
  serverVersion.copy(body, 21);
  return body;
};

const singleNodeMetadataBody = (port: number): Buffer => {
  const name = Buffer.from('single-node');
  const nodeName = Buffer.from('iggy-node');
  const ip = Buffer.from('127.0.0.1');
  const body = Buffer.alloc(
    4 + name.length + 4 + 1 + 4 + nodeName.length + 4 + ip.length + 8 + 2
  );
  let offset = 0;
  body.writeUInt32LE(name.length, offset); offset += 4;
  name.copy(body, offset); offset += name.length;
  body.writeUInt32LE(1, offset); offset += 4;
  body.writeUInt8(0, offset); offset += 1;
  body.writeUInt32LE(nodeName.length, offset); offset += 4;
  nodeName.copy(body, offset); offset += nodeName.length;
  body.writeUInt32LE(ip.length, offset); offset += 4;
  ip.copy(body, offset); offset += ip.length;
  body.writeUInt16LE(port, offset); offset += 8;
  body.writeUInt8(0, offset); offset += 1;
  body.writeUInt8(0, offset);
  return body;
};

const twoNodeMetadataBody = (
  followerPort: number,
  leaderPort: number,
  leaderRole = 0,
  leaderStatus = 0
): Buffer => {
  const node = (
    nodeName: Buffer,
    replicaId: number,
    port: number,
    role: number,
    status = 0
  ): Buffer => {
    const ip = Buffer.from('127.0.0.1');
    const encoded = Buffer.alloc(1 + 4 + nodeName.length + 4 + ip.length + 8 + 2);
    let offset = 0;
    encoded.writeUInt8(replicaId, offset); offset += 1;
    encoded.writeUInt32LE(nodeName.length, offset); offset += 4;
    nodeName.copy(encoded, offset); offset += nodeName.length;
    encoded.writeUInt32LE(ip.length, offset); offset += 4;
    ip.copy(encoded, offset); offset += ip.length;
    encoded.writeUInt16LE(port, offset); offset += 8;
    encoded.writeUInt8(role, offset); offset += 1;
    encoded.writeUInt8(status, offset);
    return encoded;
  };
  const name = Buffer.from('iggy-cluster');
  const header = Buffer.alloc(4 + name.length + 4);
  header.writeUInt32LE(name.length, 0);
  name.copy(header, 4);
  header.writeUInt32LE(2, 4 + name.length);
  return Buffer.concat([
    header,
    node(Buffer.from('iggy-node-1'), 0, leaderPort, leaderRole, leaderStatus),
    node(Buffer.from('iggy-node-2'), 7, followerPort, 1)
  ]);
};

const threeNodeMetadataBody = (
  firstPort: number,
  secondPort: number,
  thirdPort: number
): Buffer => {
  const node = (name: string, replicaId: number, port: number, role: number): Buffer => {
    const nodeName = Buffer.from(name);
    const ip = Buffer.from('127.0.0.1');
    const encoded = Buffer.alloc(1 + 4 + nodeName.length + 4 + ip.length + 8 + 2);
    let offset = 0;
    encoded.writeUInt8(replicaId, offset); offset += 1;
    encoded.writeUInt32LE(nodeName.length, offset); offset += 4;
    nodeName.copy(encoded, offset); offset += nodeName.length;
    encoded.writeUInt32LE(ip.length, offset); offset += 4;
    ip.copy(encoded, offset); offset += ip.length;
    encoded.writeUInt16LE(port, offset); offset += 8;
    encoded.writeUInt8(role, offset); offset += 1;
    encoded.writeUInt8(0, offset);
    return encoded;
  };
  const name = Buffer.from('iggy-cluster');
  const header = Buffer.alloc(4 + name.length + 4);
  header.writeUInt32LE(name.length, 0);
  name.copy(header, 4);
  header.writeUInt32LE(3, 4 + name.length);
  return Buffer.concat([
    header,
    node('iggy-node-1', 0, firstPort, 0),
    node('iggy-node-2', 7, secondPort, 1),
    node('iggy-node-3', 9, thirdPort, 1)
  ]);
};

/** Register, metadata, and echo behavior of a healthy single VSR node. */
const singleNodeHandler = (port: number): FrameHandler =>
  (frame, socket) => {
    const operation = frame.readUInt8(REQUEST_OFFSET.operation);
    if (operation === Operation.Register) {
      socket.write(replyFrame(Operation.Register, registerReplyBody()));
      return;
    }
    const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
    if (code === COMMAND_CODE.BindSession) {
      socket.write(replyFrame(Operation.NonReplicated, registerReplyBody().subarray(4)));
      return;
    }
    if (code === COMMAND_CODE.GetClusterMetadata) {
      socket.write(
        replyFrame(Operation.NonReplicated, singleNodeMetadataBody(port))
      );
      return;
    }
    if (code === COMMAND_CODE.GetPartitionRoute) {
      socket.write(replyFrame(Operation.NonReplicated, routingBody(frame.readBigUInt64LE(REQUEST_OFFSET.minimumMetadataOp) || 1n)));
      return;
    }
    socket.write(replyFrame(operation));
  };

const vsrConfig = (port: number): ClientConfig => ({
  transport: 'TCP',
  options: { host: '127.0.0.1', port },
  credentials: { username: 'iggy', password: 'iggy' },
  reconnect: { enabled: false, interval: 100, maxRetries: 1 }
});

/** Shrinks the leaderless poll so a test observes it without waiting on it. */
/** The queue a command waits in, for parking one the way the client does. */
const execQueue = (client: CommandResponseStream): {
  command: number,
  payload: Buffer,
  handleResponse: boolean,
  deadline: number,
  resolve: (v: CommandResponse | PromiseLike<CommandResponse>) => void,
  reject: (e: unknown) => void
}[] => (client as unknown as { _execQueue: never[] })._execQueue;

/** The connection under a stream, for driving a redirect the way a move does. */
const connectionOf = (client: CommandResponseStream): {
  redirect: (host: string, port: number) => Promise<void>
} => (client as unknown as {
  connection: { redirect: (host: string, port: number) => Promise<void> }
}).connection;

const compressLeaderlessPoll = (
  client: CommandResponseStream,
  budget: number
): void => {
  const settlement = client as unknown as {
    leaderlessWaitBudget: number,
    leaderlessPollInterval: number
  };
  settlement.leaderlessWaitBudget = budget;
  settlement.leaderlessPollInterval = 1;
};

const groupPollPayload = (autocommit = true, partitionId = 0): Buffer =>
  POLL_MESSAGES.serialize({
    streamId: 1,
    topicId: 1,
    consumer: { kind: ConsumerKind.Group, id: 1 },
    partitionId,
    pollingStrategy: PollingStrategy.Next,
    count: 1,
    autocommit
  });

const routingBody = (watermark: bigint, incarnation = 0n,
  primaryReplica = 7): Buffer => {
  const route = Buffer.alloc(25);
  route.writeBigUInt64LE(incarnation, 0);
  route.writeBigUInt64LE(watermark, 8);
  route.writeUInt32LE(4, 20);
  route.writeUInt8(primaryReplica, 24);
  return route;
};

const countCommand = (server: VsrTestServer, code: number): number =>
  server.frames.filter((frame) =>
    frame.readUInt8(REQUEST_OFFSET.operation) === Operation.NonReplicated &&
    frame.readUInt32LE(REQUEST_OFFSET.reserved) === code).length;

const registerCredentialsBody = (frame: Buffer): Buffer => {
  const sdkName = readWireName(frame, HEADER_SIZE + 4);
  const sdkVersion = readWireName(frame, sdkName.next);
  return frame.subarray(sdkVersion.next + BIND_SECRET_BYTES);
};

const startPollCluster = async (
  primaryOverride?: (frame: Buffer, socket: Socket) => boolean,
  coordinatorOverride?: (frame: Buffer, socket: Socket) => boolean,
  transport: 'TCP' | 'TLS' = 'TCP'
) => {
  const primary = await startVsrServer((frame, socket) => {
    if (primaryOverride?.(frame, socket))
      return;
    if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.PollMessages) {
      socket.write(replyFrame(Operation.NonReplicated, Buffer.from('primary')));
      return;
    }
    singleNodeHandler(primary.port)(frame, socket);
  }, transport);
  const coordinator = await startVsrServer((frame, socket) => {
    if (coordinatorOverride?.(frame, socket))
      return;
    const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
    if (code === COMMAND_CODE.GetClusterMetadata) {
      socket.write(replyFrame(Operation.NonReplicated,
        twoNodeMetadataBody(primary.port, coordinator.port)));
      return;
    }
    if (code === COMMAND_CODE.GetPartitionRoute) {
      socket.write(replyFrame(Operation.NonReplicated, routingBody(frame.readBigUInt64LE(REQUEST_OFFSET.minimumMetadataOp) || 1n)));
      return;
    }
    singleNodeHandler(coordinator.port)(frame, socket);
  }, transport);
  const config = {
    ...vsrConfig(coordinator.port), transport, heartbeatInterval: 0,
    options: {
      ...vsrConfig(coordinator.port).options,
      ...(transport === 'TLS' ? { ca: TLS_CA_CERTIFICATE, servername: 'localhost' } : {})
    }
  };
  const client = new CommandResponseStream(config);
  return {
    client, primary, coordinator,
    close: async () => {
      client.destroy();
      await Promise.all([coordinator.close(), primary.close()]);
    }
  };
};

const sendTo = (partition: Partitioning): Buffer => SEND_MESSAGES.serialize({
  streamId: 1, topicId: 2, messages: [{ payload: 'message' }], partition
});

const explicitSend = sendTo(Partitioning.PartitionId(3));

const storeOffsetPayload = STORE_OFFSET.serialize({
  streamId: 1, topicId: 2, consumer: { kind: ConsumerKind.Single, id: 1 }, partitionId: 3, offset: 5n
});

const deleteOffsetPayload = DELETE_OFFSET.serialize({
  streamId: 1, topicId: 2, consumer: { kind: ConsumerKind.Single, id: 1 }, partitionId: 3
});

const CALLER_CONTEXT = { incarnation: 7n, ownerGeneration: 8n, metadataOp: 9n };

const stampedContext = (frame: Buffer) => ({
  incarnation: frame.readBigUInt64LE(REQUEST_OFFSET.partitionIncarnation),
  ownerGeneration: frame.readBigUInt64LE(REQUEST_OFFSET.ownerGeneration),
  metadataOp: frame.readBigUInt64LE(REQUEST_OFFSET.minimumMetadataOp)
});

const framesOf = (server: VsrTestServer, operation: number): Buffer[] =>
  server.frames.filter((frame) => frame.readUInt8(REQUEST_OFFSET.operation) === operation);

/** Body of a committed metadata reply whose single result is `code`. */
const metadataResult = (code: number): Buffer => {
  const body = Buffer.alloc(12);
  body.writeUInt32LE(1, 0);
  body.writeUInt32LE(code, 8);
  return body;
};

/** A single node that also commits offset writes. */
const offsetCommittingHandler = (port: number): FrameHandler => (frame, socket) => {
  const operation = frame.readUInt8(REQUEST_OFFSET.operation);
  if (operation === Operation.StoreConsumerOffset || operation === Operation.DeleteConsumerOffset) {
    socket.write(replyFrame(operation, metadataResult(0)));
    return;
  }
  singleNodeHandler(port)(frame, socket);
};

const contextServer = async (state: {
  live: bigint, refuseAll: boolean, contextStatus: number, onSend?: () => void
}) => {
  const server: VsrTestServer = await startVsrServer((frame, socket) => {
    if (frame.readUInt8(REQUEST_OFFSET.operation) === Operation.SendMessages) {
      state.onSend?.();
      const sent = frame.readBigUInt64LE(REQUEST_OFFSET.partitionIncarnation);
      socket.write(replyFrame(Operation.SendMessages, Buffer.alloc(0),
        !state.refuseAll && sent === state.live ? 0 : HISTORY_UNAVAILABLE));
      return;
    }
    if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.GetPartitionRoute) {
      const context = routingBody(0n, 0n, 0);
      context.writeBigUInt64LE(state.live, 0);
      socket.write(replyFrame(Operation.NonReplicated, context, state.contextStatus));
      return;
    }
    singleNodeHandler(server.port)(frame, socket);
  });
  return server;
};

describe('unified partition routing', () => {
  const offset = { streamId: 1, topicId: 2, partitionId: 3,
    consumer: { kind: ConsumerKind.Single, id: 1 } };
  const poll = POLL_MESSAGES.serialize({ ...offset, count: 1,
    pollingStrategy: PollingStrategy.Next, autocommit: false });
  const requests = [
    [COMMAND_CODE.SendMessages, explicitSend, Operation.SendMessages],
    [COMMAND_CODE.PollMessages, poll, Operation.NonReplicated],
    [COMMAND_CODE.GetOffset, GET_OFFSET.serialize(offset), Operation.NonReplicated],
    [COMMAND_CODE.StoreOffset, storeOffsetPayload, Operation.StoreConsumerOffset],
    [COMMAND_CODE.DeleteConsumerOffset, deleteOffsetPayload, Operation.DeleteConsumerOffset]
  ] as const;

  for (const [command, payload, operation] of requests) {
    it(`reroutes command ${command} after89 with the identical request`, async () => {
      const replacement = await startVsrServer((frame, socket) => {
        if (frame[REQUEST_OFFSET.operation] === operation &&
            (operation !== Operation.NonReplicated || frame.readUInt32LE(REQUEST_OFFSET.reserved) === command)) {
          socket.write(replyFrame(operation, operation === Operation.StoreConsumerOffset ||
            operation === Operation.DeleteConsumerOffset ? metadataResult(0) : Buffer.alloc(0)));
          return;
        }
        singleNodeHandler(replacement.port)(frame, socket);
      });
      let routes = 0;
      const cluster = await startPollCluster((frame, socket) => {
        if (frame[REQUEST_OFFSET.operation] !== operation ||
            (operation === Operation.NonReplicated && frame.readUInt32LE(REQUEST_OFFSET.reserved) !== command))
          return false;
        socket.write(replyFrame(operation, Buffer.alloc(0), 89));
        return true;
      }, (frame, socket) => {
        const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
        if (code === COMMAND_CODE.GetClusterMetadata) {
          socket.write(replyFrame(Operation.NonReplicated,
            threeNodeMetadataBody(cluster.coordinator.port, cluster.primary.port, replacement.port)));
          return true;
        }
        if (code !== COMMAND_CODE.GetPartitionRoute)
          return false;
        routes += 1;
        socket.write(replyFrame(Operation.NonReplicated,
          routingBody(BigInt(routes), BigInt(routes), routes === 1 ? 7 : 9)));
        return true;
      });
      try {
        await cluster.client.sendCommand(command, payload, { deadline: Date.now() + 1_000 });
        const matching = (server: VsrTestServer) => framesOf(server, operation).filter((frame) =>
          operation !== Operation.NonReplicated || frame.readUInt32LE(REQUEST_OFFSET.reserved) === command);
        assert.equal(matching(cluster.primary).length, 1);
        assert.equal(matching(replacement).length, 1);
        assert.deepEqual(matching(cluster.primary)[0], matching(replacement)[0]);
        assert.deepEqual(stampedContext(matching(replacement)[0]),
          { incarnation: 1n, ownerGeneration: 0n, metadataOp: 1n });
        assert.equal(routes, 2);
        assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetClusterMetadata), 1);
        assert.equal(framesOf(cluster.coordinator, Operation.Register).length, 1);
        const initial = cluster.coordinator.frames[0];
        for (const server of [cluster.primary, replacement]) {
          const binding = server.frames.find((frame) =>
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession)!;
          assertSharedBinding(initial, binding);
        }
      } finally {
        await cluster.close();
        await replacement.close();
      }
    });
  }

  it('stamps the latest metadata acknowledgement on typed send route discovery', async () => {
    const cluster = await startPollCluster(undefined, (frame, socket) => {
      if (frame[REQUEST_OFFSET.operation] !== Operation.CreateStream)
        return false;
      const response = replyFrame(Operation.CreateStream, metadataResult(0));
      response.writeBigUInt64LE(19n, REPLY_OFFSET.commit);
      socket.write(response);
      return true;
    });
    try {
      await cluster.client.sendCommand(COMMAND_CODE.CreateStream, Buffer.alloc(0));
      await sendMessages(async () => cluster.client)({
        streamId: 1, topicId: 2, messages: [{ payload: 'balanced' }]
      });
      const lookups = cluster.coordinator.frames.filter((frame) =>
        frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.GetPartitionRoute);
      assert.equal(lookups.length, 1);
      assert.equal(stampedContext(lookups[0]).metadataOp, 19n);
    } finally {
      await cluster.close();
    }
  });

  for (const command of [COMMAND_CODE.PollMessages, COMMAND_CODE.GetOffset]) {
    it(`refreshes cached history after observing a newer metadata commit for ${command}`, async () => {
      let watermark = 1n;
      let incarnation = 7n;
      const histories: bigint[] = [];
      const cluster = await startPollCluster((frame, socket) => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== command)
          return false;
        histories.push(stampedContext(frame).incarnation);
        socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0)));
        return true;
      }, (frame, socket) => {
        if (frame[REQUEST_OFFSET.operation] === Operation.UpdateUser) {
          watermark = 19n;
          incarnation = 8n;
          const response = replyFrame(Operation.UpdateUser, metadataResult(0));
          response.writeBigUInt64LE(watermark, REPLY_OFFSET.commit);
          socket.write(response);
          return true;
        }
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.GetPartitionRoute)
          return false;
        socket.write(replyFrame(Operation.NonReplicated, routingBody(watermark, incarnation)));
        return true;
      });
      try {
        const payload = command === COMMAND_CODE.PollMessages
          ? groupPollPayload() : GET_OFFSET.serialize({ streamId: 1, topicId: 2,
            consumer: { kind: ConsumerKind.Single, id: 1 }, partitionId: 3 });
        await cluster.client.sendCommand(command, payload);
        await cluster.client.sendCommand(COMMAND_CODE.UpdateUser, Buffer.alloc(0));
        await cluster.client.sendCommand(command, payload);
        assert.deepEqual(histories, [7n, 8n]);
        assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 2);
        const lookups = cluster.coordinator.frames.filter((frame) =>
          frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.GetPartitionRoute);
        assert.equal(stampedContext(lookups[1]).metadataOp, watermark);
      } finally {
        await cluster.close();
      }
    });
  }

  it('shares one route and request counter across all five partition commands', async () => {
    const cluster = await startPollCluster((frame, socket) => {
      const operation = frame[REQUEST_OFFSET.operation];
      if (operation !== Operation.StoreConsumerOffset && operation !== Operation.DeleteConsumerOffset)
        return false;
      socket.write(replyFrame(operation, metadataResult(0)));
      return true;
    });
    try {
      for (const [command, payload] of requests)
        await cluster.client.sendCommand(command, payload);
      assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 1);
      const boundFrames = [...cluster.coordinator.frames, ...cluster.primary.frames]
        .filter((frame) => frame.readBigUInt64LE(REQUEST_OFFSET.session) === TEST_SESSION);
      const ids = boundFrames.map((frame) => frame.readBigUInt64LE(REQUEST_OFFSET.request));
      assert.equal(new Set(ids).size, ids.length);
      assert.ok(ids.every((id) => id > 0n));
      assert.equal(framesOf(cluster.primary, Operation.Register).length, 0);
    } finally {
      await cluster.close();
    }
  });
});

describe('primary auto-commit polling', () => {
  for (const transport of ['TCP', 'TLS'] as const) {
    for (const refused of [false, true]) {
      it(`preserves raw poll responses over ${transport}${refused ? ' after a refusal' : ''}`, async () => {
        const expected = replyFrame(Operation.NonReplicated, Buffer.from('raw primary'));
        let polls = 0;
        const cluster = await startPollCluster((frame, socket) => {
          if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.PollMessages)
            return false;
          polls += 1;
          socket.write(refused && polls === 1
            ? replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58)
            : expected);
          return true;
        }, undefined, transport);
        let resets = 0;
        cluster.client.on('sessionReset', () => { resets += 1; });
        try {
          for (let poll = 0; poll < 2; poll += 1) {
            const response = await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload(),
              { handleResponse: false });
            assert.ok(Buffer.isBuffer(response), 'raw polls must return the complete VSR frame');
            assert.deepEqual(response, expected);
          }
          assert.equal(polls, refused ? 3 : 2);
          assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), refused ? 2 : 1);
          assert.equal(countCommand(cluster.primary, COMMAND_CODE.BindSession), 1);
          assert.equal(resets, 0);
        } finally {
          await cluster.close();
        }
      });
    }
  }

  for (const transport of ['TCP', 'TLS'] as const) {
    it(`retains the coordinator session and reuses authenticated primary polls over ${transport}`, async () => {
      const cluster = await startPollCluster(undefined, undefined, transport);
      const { client, primary, coordinator } = cluster;
      let resets = 0;
      client.on('sessionReset', () => { resets += 1; });
      try {
        const first = await client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        const second = await client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        assert.equal(first.data.toString(), 'primary');
        assert.equal(second.data.toString(), 'primary');
        assert.equal(countCommand(coordinator, COMMAND_CODE.GetPartitionRoute), 1);
        assert.equal(countCommand(coordinator, COMMAND_CODE.PollMessages), 0);
        assert.equal(countCommand(primary, COMMAND_CODE.PollMessages), 2);
        assert.equal(countCommand(primary, COMMAND_CODE.BindSession), 1);
        assert.equal(countCommand(primary, COMMAND_CODE.GetClusterMetadata), 0,
          'auxiliary login must stay on the partition primary');
        assert.equal(primary.frames.filter((frame) =>
          frame[REQUEST_OFFSET.operation] === Operation.Register).length, 0);
        const route = coordinator.frames.find((frame) =>
          frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.GetPartitionRoute)!;
        const attach = primary.frames.find((frame) =>
          frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession)!;
        assert.deepEqual(bindingIdentity(attach).subarray(0, 16),
          route.subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16));
        assert.equal(bindingIdentity(attach).readBigUInt64LE(16), TEST_SESSION);
        assert.deepEqual(attach.subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16),
          route.subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16),
          'the auxiliary must bind the coordinator identity');
        assert.equal(resets, 0);
      } finally {
        await cluster.close();
      }
    });
  }

  it('routes manual-commit polls to the primary and keeps the standalone path', async () => {
    const cluster = await startPollCluster();
    const server = await startVsrServer((frame, socket) => singleNodeHandler(server.port)(frame, socket));
    const standalone = new CommandResponseStream({ ...vsrConfig(server.port), heartbeatInterval: 0 });
    try {
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload(false));
      await standalone.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.PollMessages), 0);
      assert.equal(countCommand(cluster.primary, COMMAND_CODE.PollMessages), 1);
      assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 1);
      assert.equal(countCommand(server, COMMAND_CODE.PollMessages), 1);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 1);
    } finally {
      standalone.destroy();
      await cluster.close();
      await server.close();
    }
  });

  it('refreshes the attachment after a metadata acknowledgement but ignores partition commits', async () => {
    const cluster = await startPollCluster((frame, socket) => {
      if (frame[REQUEST_OFFSET.operation] !== Operation.SendMessages)
        return false;
      const response = replyFrame(Operation.SendMessages);
      response.writeBigUInt64LE(900n, REPLY_OFFSET.commit);
      socket.write(response);
      return true;
    }, (frame, socket) => {
      if (frame[REQUEST_OFFSET.operation] !== Operation.UpdateTopic)
        return false;
      const response = replyFrame(Operation.UpdateTopic, Buffer.alloc(4));
      response.writeBigUInt64LE(15n, REPLY_OFFSET.commit);
      socket.write(response);
      return true;
    });
    try {
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      await cluster.client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 2);
      await cluster.client.sendCommand(COMMAND_CODE.UpdateTopic, Buffer.alloc(0), { handleResponse: false });
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      const attachments = cluster.primary.frames.filter((frame) =>
        frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession);
      assert.deepEqual(attachments.map((frame) => bindingIdentity(frame).readBigUInt64LE(24)), [1n, 15n]);
      assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 3);
      await cluster.client.sendCommand(LOGOUT.code, LOGOUT.serialize());
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      const replacement = cluster.primary.frames.filter((frame) =>
        frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession).at(-1)!;
      assert.equal(bindingIdentity(replacement).readBigUInt64LE(24), 15n,
        'a new coordinator session must retain previously acknowledged metadata');
    } finally {
      await cluster.close();
    }
  });

  it('refreshes a not-admitted route without moving the coordinator', async () => {
    let polls = 0;
    const cluster = await startPollCluster((frame, socket) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.PollMessages)
        return false;
      polls += 1;
      if (polls > 1)
        return false;
      socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
      return true;
    });
    let resets = 0;
    cluster.client.on('sessionReset', () => { resets += 1; });
    try {
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      assert.equal(polls, 2);
      assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 2);
      assert.equal(countCommand(cluster.primary, COMMAND_CODE.BindSession), 1);
      assert.equal(resets, 0);
    } finally {
      await cluster.close();
    }
  });

  for (const refusing of [COMMAND_CODE.GetPartitionRoute, COMMAND_CODE.BindSession,
    COMMAND_CODE.PollMessages]) {
    it(`reports not accepted after command ${refusing} refuses the whole poll budget`, async () => {
      const refuse = (frame: Buffer, socket: Socket): boolean => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== refusing)
          return false;
        socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
        return true;
      };
      const cluster = await startPollCluster(refuse, refuse);
      let resets = 0;
      cluster.client.on('sessionReset', () => { resets += 1; });
      try {
        await assert.rejects(cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload(),
          { deadline: Date.now() + 1_000 }), (error: unknown) =>
          error instanceof ResponseError && error.errorCode === 58);
        assert.ok(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute) >= 2);
        if (refusing === COMMAND_CODE.GetPartitionRoute)
          assert.equal(cluster.primary.frames.length, 0);
        if (refusing === COMMAND_CODE.BindSession)
          assert.equal(countCommand(cluster.primary, COMMAND_CODE.PollMessages), 0);
        assert.equal(resets, 0);
      } finally {
        await cluster.close();
      }
    });
  }

  it('refreshes the roster once when the primary has no TCP endpoint', async () => {
    const cluster = await startPollCluster(undefined, (frame, socket) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.GetClusterMetadata)
        return false;
      socket.write(replyFrame(Operation.NonReplicated, twoNodeMetadataBody(0, cluster.coordinator.port)));
      return true;
    });
    try {
      await assert.rejects(cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload(),
        { deadline: Date.now() + 500 }),
      (error: unknown) => error instanceof ResponseError && error.errorCode === 58);
      assert.equal(cluster.primary.frames.length, 0);
      assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetClusterMetadata), 2);
    } finally {
      await cluster.close();
    }
  });

  for (const standalone of [false, true]) {
    for (const failure of ['empty reply', 'empty roster', 'unavailable']) {
      it(`requires valid topology after ${failure} before ${standalone ? 'standalone' : 'cluster'} polling`, async () => {
        const clusterName = Buffer.from('cluster-without-nodes');
        const emptyRoster = Buffer.alloc(4 + clusterName.length + 4);
        emptyRoster.writeUInt32LE(clusterName.length, 0);
        clusterName.copy(emptyRoster, 4);
        let rejectMetadata = true;
        let coordinatorPort = 0;
        const cluster = await startPollCluster(undefined, (frame, socket) => {
          if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.GetClusterMetadata)
            return false;
          if (rejectMetadata) {
            socket.write(replyFrame(Operation.NonReplicated,
              failure === 'empty roster' ? emptyRoster : Buffer.alloc(0),
              failure === 'unavailable' ? 5 : 0));
            return true;
          }
          if (standalone) {
            socket.write(replyFrame(Operation.NonReplicated, singleNodeMetadataBody(coordinatorPort)));
            return true;
          }
          return false;
        });
        coordinatorPort = cluster.coordinator.port;
        let resets = 0;
        cluster.client.on('sessionReset', () => { resets += 1; });
        try {
          await cluster.client.authenticate(vsrConfig(coordinatorPort).credentials);
          await assert.rejects(cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload()));
          assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.PollMessages), 0);
          assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 0);
          assert.equal(cluster.primary.frames.length, 0);
          rejectMetadata = false;
          await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
          rejectMetadata = true;
          const refresh = cluster.client.sendCommand(COMMAND_CODE.GetClusterMetadata, Buffer.alloc(0));
          if (failure === 'unavailable')
            await assert.rejects(refresh, (error: unknown) => error instanceof ResponseError && error.errorCode === 5);
          else
            await refresh;
          const reads = countCommand(cluster.coordinator, COMMAND_CODE.GetClusterMetadata);
          await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
          assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetClusterMetadata), reads);
          assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.PollMessages), standalone ? 2 : 0);
          assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 1);
          assert.equal(countCommand(cluster.primary, COMMAND_CODE.PollMessages), standalone ? 0 : 2);
          assert.equal(resets, 0);
        } finally {
          await cluster.close();
        }
      });
    }
  }

  it('refreshes a stalled auxiliary TLS dial within the shared poll budget', async () => {
    const held: Socket[] = [];
    const silent = createServer((socket) => {
      held.push(socket);
      socket.resume();
    });
    silent.listen(0, '127.0.0.1');
    await once(silent, 'listening');
    const silentPort = (silent.address() as AddressInfo).port;
    let routes = 0;
    const cluster = await startPollCluster(undefined, (frame, socket) => {
      const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
      if (code === COMMAND_CODE.GetClusterMetadata) {
        socket.write(replyFrame(Operation.NonReplicated,
          threeNodeMetadataBody(cluster.coordinator.port, cluster.primary.port, silentPort)));
        return true;
      }
      if (code !== COMMAND_CODE.GetPartitionRoute || ++routes > 1)
        return false;
      socket.write(replyFrame(Operation.NonReplicated, routingBody(1n, 0n, 9)));
      return true;
    }, 'TLS');
    let resets = 0;
    cluster.client.on('sessionReset', () => { resets += 1; });
    try {
      const response = await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload(),
        { deadline: Date.now() + 5_000 });
      assert.equal(response.data.toString(), 'primary');
      assert.equal(routes, 2);
      assert.equal(held.length, 1);
      assert.equal(held[0].destroyed, true);
      assert.equal(countCommand(cluster.primary, COMMAND_CODE.PollMessages), 1);
      assert.equal(resets, 0);
    } finally {
      await cluster.close();
      held.forEach((socket) => socket.destroy());
      await new Promise<void>((resolve) => silent.close(() => resolve()));
    }
  });

  for (const operation of [Operation.TruncatePartition, Operation.DeleteSegments]) {
    it(`raises the attachment floor for a DeleteSegments acknowledgement with operation ${operation}`, async () => {
      const cluster = await startPollCluster(undefined, (frame, socket) => {
        if (frame[REQUEST_OFFSET.operation] !== Operation.DeleteSegments)
          return false;
        const response = replyFrame(operation,
          operation === Operation.TruncatePartition ? Buffer.alloc(4) : Buffer.alloc(0));
        response.writeBigUInt64LE(50n, REPLY_OFFSET.commit);
        socket.write(response);
        return true;
      });
      try {
        await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        await cluster.client.sendCommand(COMMAND_CODE.DeleteSegments, Buffer.alloc(0));
        await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        const attachments = cluster.primary.frames.filter((frame) =>
          frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession);
        assert.deepEqual(attachments.map((frame) => bindingIdentity(frame).readBigUInt64LE(24)), [1n, 50n]);
        assert.equal(cluster.primary.frames.filter((frame) => frame[REQUEST_OFFSET.operation] === Operation.Register).length, 0);
      } finally {
        await cluster.close();
      }
    });
  }

  it('rechecks metadata after waiting behind another primary poll', async () => {
    let firstPoll!: (socket: Socket) => void;
    const admitted = new Promise<Socket>((resolve) => { firstPoll = resolve; });
    let polls = 0;
    const cluster = await startPollCluster((frame, socket) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.PollMessages || ++polls !== 1)
        return false;
      firstPoll(socket);
      return true;
    }, (frame, socket) => {
      if (frame[REQUEST_OFFSET.operation] !== Operation.UpdateTopic)
        return false;
      const response = replyFrame(Operation.UpdateTopic, Buffer.alloc(4));
      response.writeBigUInt64LE(21n, REPLY_OFFSET.commit);
      socket.write(response);
      return true;
    });
    try {
      const first = cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      const socket = await admitted;
      const second = cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      await cluster.client.sendCommand(COMMAND_CODE.UpdateTopic, Buffer.alloc(0));
      assert.equal(polls, 1, 'concurrent polls must share a serialized primary connection');
      socket.write(replyFrame(Operation.NonReplicated, Buffer.from('primary')));
      await Promise.all([first, second]);
      assert.equal(polls, 2);
      const attachments = cluster.primary.frames.filter((frame) =>
        frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession);
      assert.deepEqual(attachments.map((frame) => bindingIdentity(frame).readBigUInt64LE(24)), [1n, 21n]);
    } finally {
      await cluster.close();
    }
  });

  for (const failure of ['disconnect', 'not committed', 'timeout', 'partial reply', 'stale session', 'unauthenticated'] as const) {
    it(`does not replay an uncertain primary poll after ${failure}`, async () => {
      let polls = 0;
      const cluster = await startPollCluster((frame, socket) => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.PollMessages)
          return false;
        polls += 1;
        if (polls > 1)
          return false;
        if (failure === 'disconnect') socket.destroy();
        if (failure === 'not committed')
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 57));
        if (failure === 'stale session')
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 30));
        if (failure === 'unauthenticated')
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 40));
        if (failure === 'partial reply')
          socket.write(replyFrame(Operation.NonReplicated, Buffer.from('partial')).subarray(0, HEADER_SIZE + 1));
        return true;
      });
      let resets = 0;
      cluster.client.on('sessionReset', () => { resets += 1; });
      try {
        await assert.rejects(cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload(),
          { deadline: Date.now() + 300 }), (error: unknown) =>
          error instanceof ResponseError && error.errorCode === 57);
        assert.equal(polls, 1);
        assert.equal(resets, 0);
        await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        assert.equal(polls, 2);
        assert.equal(countCommand(cluster.primary, COMMAND_CODE.BindSession), failure === 'not committed' ? 1 : 2);
      } finally {
        await cluster.close();
      }
    });
  }

  for (const status of [30, 40]) {
    it(`rebuilds an auxiliary rejected with session status ${status} before poll admission`, async () => {
      let attachments = 0;
      const cluster = await startPollCluster((frame, socket) => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.BindSession || ++attachments > 1)
          return false;
        socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), status));
        return true;
      });
      let resets = 0;
      cluster.client.on('sessionReset', () => { resets += 1; });
      try {
        await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        assert.equal(attachments, 2);
        assert.equal(countCommand(cluster.primary, COMMAND_CODE.PollMessages), 1);
        assert.equal(cluster.primary.frames.filter((frame) => frame[REQUEST_OFFSET.operation] === Operation.Register).length, 0);
        assert.equal(resets, 0);
      } finally {
        await cluster.close();
      }
    });

    it(`returns control session status ${status} without rerouting the lookup`, async () => {
      const registered = new Set<Socket>();
      let routes = 0;
      const cluster = await startPollCluster(undefined, (frame, socket) => {
        if (frame[REQUEST_OFFSET.operation] === Operation.Register ||
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession) {
          assert.equal(registered.has(socket), false,
            'session recovery must replace the rejected control transport');
          registered.add(socket);
        }
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.GetPartitionRoute || ++routes > 1)
          return false;
        socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), status));
        return true;
      });
      try {
        await assert.rejects(cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload()),
          (error: unknown) => error instanceof ResponseError && error.errorCode === status);
        assert.equal(routes, 1);
        assert.equal(countCommand(cluster.primary, COMMAND_CODE.PollMessages), 0);
        await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        assert.equal(registered.size, 2);
        assert.equal(countCommand(cluster.primary, COMMAND_CODE.PollMessages), 1);
      } finally {
        await cluster.close();
      }
    });
  }

  it('binds primary connections after credential changes without another login', async () => {
    const cluster = await startPollCluster(undefined, (frame, socket) => {
      const operation = frame[REQUEST_OFFSET.operation];
      if (operation !== Operation.UpdateUser && operation !== Operation.ChangePassword)
        return false;
      socket.write(replyFrame(operation, Buffer.alloc(4)));
      return true;
    });
    try {
      await cluster.client.sendCommand(LOGIN.code, LOGIN.serialize({ username: 'manual', password: 'first' }));
      await cluster.client.sendCommand(UPDATE_USER.code, UPDATE_USER.serialize({ userId: 7, username: 'renamed' }));
      await cluster.client.sendCommand(CHANGE_PASSWORD.code, CHANGE_PASSWORD.serialize({
        userId: 7, currentPassword: 'first', newPassword: 'second'
      }));
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      const login = cluster.coordinator.frames.find((frame) => frame[REQUEST_OFFSET.operation] === Operation.Register)!;
      const credentials = registerCredentialsBody(login);
      const username = readWireName(credentials, 0);
      const password = readWireName(credentials, username.next);
      assert.equal(username.value, 'manual');
      assert.equal(password.value, 'first');
      const binding = cluster.primary.frames.find((frame) =>
        frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession)!;
      assertSharedBinding(login, binding);
      assert.equal(cluster.primary.frames.filter((frame) =>
        frame[REQUEST_OFFSET.operation] === Operation.Register).length, 0);
      await cluster.client.sendCommand(LOGOUT.code, LOGOUT.serialize());
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      const replacement = cluster.coordinator.frames.filter((frame) =>
        frame[REQUEST_OFFSET.operation] === Operation.Register).at(-1)!;
      assert.equal(readWireName(registerCredentialsBody(replacement), 0).value, 'iggy');
      assertSharedBinding(replacement, cluster.primary.frames.filter((frame) =>
        frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession).at(-1)!);
    } finally {
      await cluster.close();
    }
  });

  it('binds auxiliaries with the proof from PAT registration', async () => {
    const cluster = await startPollCluster();
    try {
      await cluster.client.sendCommand(LOGIN_WITH_TOKEN.code, LOGIN_WITH_TOKEN.serialize({ token: 'manual-token' }));
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      const login = cluster.coordinator.frames.find((frame) => frame[REQUEST_OFFSET.operation] === Operation.Register)!;
      const sdkName = readWireName(login, HEADER_SIZE + 4);
      const sdkVersion = readWireName(login, sdkName.next);
      assert.equal(readWireName(login, sdkVersion.next + BIND_SECRET_BYTES).value, 'manual-token');
      assertSharedBinding(login, cluster.primary.frames.find((frame) =>
        frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession)!);
    } finally {
      await cluster.close();
    }
  });

  for (const manual of [
    { command: LOGIN.code, payload: LOGIN.serialize({ username: 'manual', password: 'first' }), identity: 'manual' },
    { command: LOGIN_WITH_TOKEN.code, payload: LOGIN_WITH_TOKEN.serialize({ token: 'manual-token' }), identity: 'manual-token' }
  ]) {
    it(`retains ${manual.identity} across recovery and restores configured credentials after logout`, async () => {
      let coordinatorSocket!: Socket;
      const cluster = await startPollCluster(undefined, (_frame, socket) => {
        coordinatorSocket = socket;
        return false;
      });
      try {
        await cluster.client.sendCommand(manual.command, manual.payload);
        await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        const reset = once(cluster.client, 'sessionReset');
        coordinatorSocket.destroy();
        await reset;
        await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        for (const server of [cluster.coordinator]) {
          const login = server.frames.filter((frame) => frame[REQUEST_OFFSET.operation] === Operation.Register).at(-1)!;
          assert.equal(readWireName(registerCredentialsBody(login), 0).value, manual.identity);
          assertSharedBinding(login, cluster.primary.frames.filter((frame) =>
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession).at(-1)!);
        }
        await cluster.client.sendCommand(LOGOUT.code, LOGOUT.serialize());
        await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
        for (const server of [cluster.coordinator]) {
          const login = server.frames.filter((frame) => frame[REQUEST_OFFSET.operation] === Operation.Register).at(-1)!;
          const credentials = registerCredentialsBody(login);
          const username = readWireName(credentials, 0);
          assert.equal(username.value, 'iggy');
          assert.equal(readWireName(credentials, username.next).value, 'iggy');
          assertSharedBinding(login, cluster.primary.frames.filter((frame) =>
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession).at(-1)!);
        }
      } finally {
        await cluster.close();
      }
    });
  }

  it('updates the configured identity after its password and username change', async () => {
    const cluster = await startPollCluster(undefined, (frame, socket) => {
      const operation = frame[REQUEST_OFFSET.operation];
      if (operation !== Operation.UpdateUser && operation !== Operation.ChangePassword)
        return false;
      socket.write(replyFrame(operation, Buffer.alloc(4)));
      return true;
    });
    try {
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      await cluster.client.sendCommand(UPDATE_USER.code, UPDATE_USER.serialize({ userId: 7, username: 'renamed' }));
      await cluster.client.sendCommand(CHANGE_PASSWORD.code, CHANGE_PASSWORD.serialize({
        userId: 7, currentPassword: 'iggy', newPassword: 'second'
      }));
      await cluster.client.sendCommand(LOGOUT.code, LOGOUT.serialize());
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      for (const server of [cluster.coordinator]) {
        const login = server.frames.filter((frame) => frame[REQUEST_OFFSET.operation] === Operation.Register).at(-1)!;
        const credentials = registerCredentialsBody(login);
        const username = readWireName(credentials, 0);
        assert.equal(username.value, 'renamed');
        assert.equal(readWireName(credentials, username.next).value, 'second');
        assertSharedBinding(login, cluster.primary.frames.filter((frame) =>
          frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession).at(-1)!);
      }
    } finally {
      await cluster.close();
    }
  });

  it('closes a waiting primary exchange on coordinator logout without replay', async () => {
    let admitted!: () => void;
    const pendingAdmission = new Promise<void>((resolve) => { admitted = resolve; });
    const cluster = await startPollCluster((frame) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.PollMessages)
        return false;
      admitted();
      return true;
    });
    try {
      const polling = cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload());
      const failed = assert.rejects(polling, (error: unknown) =>
        error instanceof ResponseError && error.errorCode === 57);
      await pendingAdmission;
      await cluster.client.sendCommand(LOGOUT.code, LOGOUT.serialize());
      await failed;
      assert.equal(countCommand(cluster.primary, COMMAND_CODE.PollMessages), 1);
      assert.equal(cluster.client.isAuthenticated, false);
    } finally {
      await cluster.close();
    }
  });
});

describe('VSR client socket', () => {
  for (const refusal of [0, 30, 40, 57, 58]) {
    it(`resumes or replaces a lost connection only after bind verdict ${refusal}`, async () => {
      let registrations = 0;
      let sends = 0;
      let bindStatus = refusal;
      const server = await startVsrServer((frame, socket) => {
        const operation = frame.readUInt8(REQUEST_OFFSET.operation);
        const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
        if (operation === Operation.Register) {
          registrations += 1;
          socket.write(replyFrame(operation, registerReplyBody(TEST_SESSION + BigInt(registrations - 1))));
        } else if (code === COMMAND_CODE.BindSession) {
          socket.write(replyFrame(operation, bindStatus === 0 ? registerReplyBody().subarray(4) : undefined, bindStatus));
        } else if (code === COMMAND_CODE.GetClusterMetadata) {
          socket.write(replyFrame(operation, singleNodeMetadataBody(server.port)));
        } else if (code === COMMAND_CODE.GetPartitionRoute) {
          socket.write(replyFrame(operation, routingBody(0n, 0n, 0)));
        } else if (operation === Operation.SendMessages) {
          sends += 1;
          if (sends === 1)
            socket.destroy();
          else
            socket.write(replyFrame(operation, Buffer.alloc(4)));
        }
      });
      const config = vsrConfig(server.port);
      const client = new CommandResponseStream(config);
      try {
        await client.authenticate(config.credentials);
        await assert.rejects(client.sendCommand(COMMAND_CODE.SendMessages, explicitSend));
        if (refusal === 57 || refusal === 58) {
          await assert.rejects(client.authenticate(config.credentials), (error: unknown) =>
            error instanceof ResponseError && error.errorCode === refusal);
          assert.equal(registrations, 1, 'lagging metadata must not replace the logical session');
          bindStatus = 0;
        }
        await client.authenticate(config.credentials);
        assert.equal(sends, 1, 'authentication must not replay the uncertain mutation');
        const terminal = refusal === 30 || refusal === 40;
        assert.equal(registrations, terminal ? 2 : 1);
        const binding = server.frames.find((frame) =>
          frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession)!;
        assertSharedBinding(server.frames[0], binding);
        await client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
        const writes = server.frames.filter((frame) => frame.readUInt8(REQUEST_OFFSET.operation) === Operation.SendMessages);
        assert.equal(writes.length, 2);
        assert.equal(writes[1].readBigUInt64LE(REQUEST_OFFSET.session), terminal ? TEST_SESSION + 1n : TEST_SESSION);
        const nextRequest = writes[1].readBigUInt64LE(REQUEST_OFFSET.request);
        assert.ok(nextRequest > 0n);
        if (!terminal)
          assert.ok(nextRequest > writes[0].readBigUInt64LE(REQUEST_OFFSET.request));
        assert.equal(writes[0].subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16)
          .equals(writes[1].subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16)), !terminal);
      } finally {
        client.destroy();
        await server.close();
      }
    });
  }

  it('exchanges VSR frames over TLS', async () => {
    const server = await startVsrServer(
      (frame, socket) => singleNodeHandler(server.port)(frame, socket),
      'TLS'
    );
    const client = new CommandResponseStream({
      ...vsrConfig(server.port),
      transport: 'TLS',
      options: {
        host: '127.0.0.1',
        port: server.port,
        servername: 'localhost',
        ca: TLS_CA_CERTIFICATE
      }
    });
    try {
      const response = await client.sendCommand(60_015, Buffer.from('tls'));

      assert.equal(response.status, 0);
      assert.equal(server.frames.length, 3);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('registers one session before the first authenticated command', async () => {
    const server = await startVsrServer(
      (frame, socket) => singleNodeHandler(server.port)(frame, socket)
    );
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      const response = await client.sendCommand(
        60_001,
        Buffer.from('opaque')
      );
      assert.equal(response.status, 0);

      const operations = server.frames.map(
        (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
      );
      assert.deepEqual(operations, [
        Operation.Register,
        Operation.NonReplicated,
        Operation.NonReplicated
      ]);
      const register = server.frames[0];
      assert.equal(register.readBigUInt64LE(REQUEST_OFFSET.request), 0n);
      assert.equal(register.readBigUInt64LE(REQUEST_OFFSET.session), 0n);
      const settlement = server.frames[1];
      assert.equal(
        settlement.readUInt32LE(REQUEST_OFFSET.reserved),
        COMMAND_CODE.GetClusterMetadata
      );
      const request = server.frames[2];
      assert.equal(
        request.readBigUInt64LE(REQUEST_OFFSET.session),
        TEST_SESSION
      );
      assert.equal(request.readUInt32LE(REQUEST_OFFSET.reserved), 60_001);
      assert.equal(client.isAuthenticated, true);
      assert.equal(client.userId, 7);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('replays the identical frame on a transient response', async () => {
    let attempts = 0;
    const server = await startVsrServer((frame, socket) => {
      const operation = frame.readUInt8(REQUEST_OFFSET.operation);
      if (operation !== Operation.NonReplicated ||
          frame.readUInt32LE(REQUEST_OFFSET.reserved) !== 60_002)
        return singleNodeHandler(server.port)(frame, socket);
      attempts += 1;
      socket.write(
        attempts === 1
          ? replyFrame(operation, Buffer.alloc(0), 57)
          : replyFrame(operation, Buffer.from('done'))
      );
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      const response = await client.sendCommand(
        60_002,
        Buffer.from('retry-me')
      );
      assert.equal(attempts, 2);
      assert.deepEqual(response.data, Buffer.from('done'));
      const sent = server.frames.filter(
        (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_002
      );
      assert.equal(sent.length, 2);
      assert.deepEqual(sent[0], sent[1]);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('routes eviction out of band without desynchronizing replies', async () => {
    const server = await startVsrServer((frame, socket) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_003) {
        socket.write(Buffer.concat([
          evictionFrame(EvictionReason.NoSession),
          replyFrame(Operation.NonReplicated, Buffer.from('stale'))
        ]));
        return;
      }
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_014) {
        socket.write(
          replyFrame(Operation.NonReplicated, Buffer.from('fresh'))
        );
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      let sessionResets = 0;
      let evictions = 0;
      client.on('sessionReset', () => { sessionResets += 1; });
      client.on('eviction', () => { evictions += 1; });
      await assert.rejects(
        () => client.sendCommand(
          60_003,
          Buffer.alloc(0)
        ),
        (error: unknown) =>
          error instanceof VsrEvictionError && error.errorCode === 40
      );
      assert.ok(sessionResets >= 1);
      assert.equal(evictions, 1);
      assert.equal(client.isAuthenticated, false);
      const response = await client.sendCommand(60_014, Buffer.alloc(0));
      assert.deepEqual(response.data, Buffer.from('fresh'));
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('fails an in-flight request when the connection drops', async () => {
    const server = await startVsrServer((frame, socket) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_004) {
        socket.destroy();
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await assert.rejects(
        () => client.sendCommand(
          60_004,
          Buffer.alloc(0)
        )
      );
      assert.equal(client.isAuthenticated, false);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('redirects a login to the advertised leader and registers there',
    async () => {
      const leader = await startVsrServer(
        (frame, socket) => singleNodeHandler(leader.port)(frame, socket)
      );
      const follower = await startVsrServer((frame, socket) => {
        const operation = frame.readUInt8(REQUEST_OFFSET.operation);
        if (operation === Operation.Register) {
          socket.write(replyFrame(Operation.Register, registerReplyBody()));
          return;
        }
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) ===
            COMMAND_CODE.GetClusterMetadata) {
          socket.write(replyFrame(
            Operation.NonReplicated,
            twoNodeMetadataBody(follower.port, leader.port)
          ));
          return;
        }
        socket.write(replyFrame(operation, Buffer.alloc(0), 58));
      });
      const client = new CommandResponseStream(vsrConfig(follower.port));
      try {
        const response = await client.sendCommand(
          COMMAND_CODE.LoginUser,
          Buffer.concat([
            Buffer.from([4]),
            Buffer.from('iggy'),
            Buffer.from([4]),
            Buffer.from('iggy')
          ]));
        assert.equal(response.status, 0);
        assert.equal(client.isAuthenticated, true);
        const followerOperations = follower.frames.map(
          (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
        );
        assert.deepEqual(followerOperations, [
          Operation.Register,
          Operation.NonReplicated
        ]);

        await client.sendCommand(60_018, Buffer.alloc(0));

        const leaderOperations = leader.frames.map(
          (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
        );
        assert.deepEqual(leaderOperations, [
          Operation.Register,
          Operation.NonReplicated,
          Operation.NonReplicated
        ]);
        assert.equal(
          leader.frames[2].readUInt32LE(REQUEST_OFFSET.reserved),
          60_018
        );
        assert.equal(follower.frames.length, 2);
      } finally {
        client.destroy();
        await leader.close();
        await follower.close();
      }
    });

  it('rechecks leadership after a redirected login', async () => {
    const leader = await startVsrServer(
      (frame, socket) => singleNodeHandler(leader.port)(frame, socket)
    );
    const intermediate = await startVsrServer((frame, socket) => {
      const operation = frame.readUInt8(REQUEST_OFFSET.operation);
      if (operation === Operation.Register ||
          frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession) {
        singleNodeHandler(intermediate.port)(frame, socket);
        return;
      }
      socket.write(replyFrame(
        Operation.NonReplicated,
        twoNodeMetadataBody(intermediate.port, leader.port)
      ));
    });
    const follower = await startVsrServer((frame, socket) => {
      const operation = frame.readUInt8(REQUEST_OFFSET.operation);
      if (operation === Operation.Register) {
        socket.write(replyFrame(Operation.Register, registerReplyBody()));
        return;
      }
      socket.write(replyFrame(
        Operation.NonReplicated,
        twoNodeMetadataBody(follower.port, intermediate.port)
      ));
    });
    const client = new CommandResponseStream(vsrConfig(follower.port));
    try {
      await client.authenticate(vsrConfig(follower.port).credentials);
      await client.sendCommand(60_019, Buffer.alloc(0));

      assert.deepEqual(
        follower.frames.map((frame) => frame.readUInt8(REQUEST_OFFSET.operation)),
        [Operation.Register, Operation.NonReplicated]
      );
      assert.deepEqual(
        intermediate.frames.map(
          (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
        ),
        [Operation.NonReplicated, Operation.NonReplicated]
      );
      assertSharedBinding(follower.frames[0], intermediate.frames[0]);
      assert.equal(
        leader.frames[2].readUInt32LE(REQUEST_OFFSET.reserved),
        60_019
      );
    } finally {
      client.destroy();
      await follower.close();
      await intermediate.close();
      await leader.close();
    }
  });

  // The node a client authenticated on dies; its next command has to complete
  // on a survivor the roster named, under a session established there.
  // Mirrors `core/integration/tests/cluster/failover_client_continuity.rs`.
  it('resumes on a survivor after the node it authenticated on dies',
    async () => {
      const primarySockets = new Set<Socket>();
      let primaryDead = false;

      const survivor = await startVsrServer((frame, socket) => {
        const operation = frame.readUInt8(REQUEST_OFFSET.operation);
        if (operation === Operation.Register ||
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession) {
          singleNodeHandler(survivor.port)(frame, socket);
          return;
        }
        const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
        if (code === COMMAND_CODE.GetClusterMetadata) {
          // The survivor leads once the primary is gone.
          socket.write(replyFrame(
            Operation.NonReplicated,
            twoNodeMetadataBody(primary.port, survivor.port)
          ));
          return;
        }
        socket.write(replyFrame(operation));
      });

      const primary = await startVsrServer((frame, socket) => {
        primarySockets.add(socket);
        if (primaryDead) {
          socket.destroy();
          return;
        }
        const operation = frame.readUInt8(REQUEST_OFFSET.operation);
        if (operation === Operation.Register) {
          socket.write(replyFrame(Operation.Register, registerReplyBody()));
          return;
        }
        const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
        if (code === COMMAND_CODE.GetClusterMetadata) {
          // The primary leads, so the login settles here and the roster is
          // only remembered, not acted on, until the node dies.
          socket.write(replyFrame(
            Operation.NonReplicated,
            twoNodeMetadataBody(survivor.port, primary.port)
          ));
          return;
        }
        socket.write(replyFrame(operation));
      });

      const config: ClientConfig = {
        ...vsrConfig(primary.port),
        reconnect: { enabled: true, interval: 1, maxRetries: 3 }
      };
      const client = new CommandResponseStream(config);
      try {
        await client.authenticate(config.credentials);
        await client.sendCommand(60_021, Buffer.alloc(0));
        assert.ok(
          primary.frames.some(
            (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_021
          ),
          'the live primary answered the first command'
        );

        primaryDead = true;
        for (const socket of primarySockets)
          socket.destroy();
        await primary.close();

        // The attempt in flight when the socket died is allowed to fail; the
        // one after it has to land on the survivor. Two attempts, not a
        // polling loop: the comment above promises at most one failed
        // submission, and a loop of twenty would pass with nineteen failures.
        let resumed = false;
        let lastError: unknown;
        for (let attempt = 0; attempt < 2 && !resumed; attempt += 1) {
          try {
            await client.sendCommand(60_021, Buffer.alloc(0));
            resumed = true;
          } catch (error) {
            lastError = error;
          }
        }
        assert.ok(resumed, `the client never resumed: ${String(lastError)}`);

        assert.equal(countCommand(survivor, COMMAND_CODE.BindSession), 1);
        assertSharedBinding(primary.frames[0], survivor.frames[0]);
        assert.ok(
          survivor.frames.some(
            (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_021
          ),
          'the command landed on the survivor the roster named'
        );
      } finally {
        client.destroy();
        await survivor.close();
      }
    });

  it('keeps a single-node login on its node', async () => {
    const server = await startVsrServer(
      (frame, socket) => singleNodeHandler(server.port)(frame, socket)
    );
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await client.authenticate(vsrConfig(server.port).credentials);

      const operations = server.frames.map(
        (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
      );
      assert.deepEqual(operations, [
        Operation.Register,
        Operation.NonReplicated
      ]);
      assert.equal(
        server.frames[1].readUInt32LE(REQUEST_OFFSET.reserved),
        COMMAND_CODE.GetClusterMetadata
      );
      assert.equal(client.isAuthenticated, true);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('polls a leaderless roster before redirecting to the elected leader',
    async () => {
      const leader = await startVsrServer(
        (frame, socket) => singleNodeHandler(leader.port)(frame, socket)
      );
      let rosterReads = 0;
      const follower = await startVsrServer((frame, socket) => {
        const operation = frame.readUInt8(REQUEST_OFFSET.operation);
        if (operation === Operation.Register) {
          socket.write(replyFrame(Operation.Register, registerReplyBody()));
          return;
        }
        rosterReads += 1;
        // The first answer is mid-election: neither node holds the leader role.
        socket.write(replyFrame(
          Operation.NonReplicated,
          rosterReads === 1
            ? twoNodeMetadataBody(follower.port, leader.port, 1)
            : twoNodeMetadataBody(follower.port, leader.port)
        ));
      });
      const client = new CommandResponseStream(vsrConfig(follower.port));
      compressLeaderlessPoll(client, 1_000);
      try {
        await client.authenticate(vsrConfig(follower.port).credentials);

        assert.equal(rosterReads, 2);
        const leaderOperations = leader.frames.map(
          (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
        );
        assert.deepEqual(leaderOperations, [
          Operation.NonReplicated,
          Operation.NonReplicated
        ]);
        assertSharedBinding(follower.frames[0], leader.frames[0]);
        assert.equal(client.isAuthenticated, true);
      } finally {
        client.destroy();
        await leader.close();
        await follower.close();
      }
    });

  it('keeps a login alive when the session dies mid-poll', async () => {
    let rosterReads = 0;
    const server = await startVsrServer((frame, socket) => {
      const operation = frame.readUInt8(REQUEST_OFFSET.operation);
      if (operation === Operation.Register) {
        socket.write(replyFrame(Operation.Register, registerReplyBody()));
        return;
      }
      rosterReads += 1;
      // The roster stays leaderless and the session dies in the same breath.
      // Polling on without a session would re-enter authentication, which
      // awaits the login being settled, and the login would never return.
      socket.write(Buffer.concat([
        replyFrame(
          Operation.NonReplicated,
          twoNodeMetadataBody(server.port, server.port, 1)
        ),
        evictionFrame(EvictionReason.NoSession)
      ]));
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    compressLeaderlessPoll(client, 1_000);
    try {
      const outcome = await Promise.race([
        client.authenticate(vsrConfig(server.port).credentials)
          .then(() => 'authenticated'),
        // Unreferenced so a passing run is not held open by the stall timer.
        new Promise((resolve) => {
          setTimeout(() => resolve('stalled'), 500).unref();
        })
      ]);

      assert.equal(outcome, 'authenticated');
      assert.equal(rosterReads, 1);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('keeps a login on its node when no leader appears in the budget',
    async () => {
      const unavailable = await startVsrServer(() => {});
      const unavailablePort = unavailable.port;
      await unavailable.close();
      // The roster marks its leader-role node unhealthy on a dead port, so a
      // redirect would fail the dial instead of passing unnoticed.
      const server = await startVsrServer((frame, socket) => {
        const operation = frame.readUInt8(REQUEST_OFFSET.operation);
        if (operation === Operation.Register) {
          socket.write(replyFrame(Operation.Register, registerReplyBody()));
          return;
        }
        socket.write(replyFrame(
          Operation.NonReplicated,
          twoNodeMetadataBody(server.port, unavailablePort, 0, 1)
        ));
      });
      const client = new CommandResponseStream(vsrConfig(server.port));
      compressLeaderlessPoll(client, 0);
      try {
        await client.authenticate(vsrConfig(server.port).credentials);

        const operations = server.frames.map(
          (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
        );
        assert.deepEqual(operations, [
          Operation.Register,
          Operation.NonReplicated
        ]);
        assert.equal(client.isAuthenticated, true);
      } finally {
        client.destroy();
        await server.close();
      }
    });

  it('rejects instead of hanging while a connection attempt is unresolved', async () => {
    const server = await startVsrServer(() => {});
    const port = server.port;
    await server.close();
    const client = new CommandResponseStream({
      ...vsrConfig(port),
      reconnect: { enabled: true, interval: 300, maxRetries: 1 }
    });
    try {
      await assert.rejects(
        () => client.sendCommand(
          60_006,
          Buffer.alloc(0)
        )
      );
      await assert.rejects(
        () => client.sendCommand(
          60_006,
          Buffer.alloc(0)
        )
      );
    } finally {
      client.destroy();
    }
  });

  it('resets the session before an authenticated VSR login', async () => {
    const server = await startVsrServer(
      (frame, socket) => singleNodeHandler(server.port)(frame, socket)
    );
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await client.sendCommand(60_007, Buffer.alloc(0));
      let sessionResets = 0;
      client.on('sessionReset', () => { sessionResets += 1; });

      await client.sendCommand(
        COMMAND_CODE.LoginUser,
        Buffer.concat([
          Buffer.from([4]),
          Buffer.from('iggy'),
          Buffer.from([4]),
          Buffer.from('iggy')
        ])
      );

      const operations = server.frames.map(
        (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
      );
      assert.equal(
        operations.filter((operation) => operation === Operation.Register)
          .length,
        2
      );
      assert.ok(operations.includes(Operation.Logout));
      assert.equal(sessionResets, 1);
      assert.equal(client.isAuthenticated, true);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('keeps logout and replacement login adjacent in the queue', async () => {
    const server = await startVsrServer((frame, socket) => {
      const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
      if (code === 60_015) {
        setTimeout(() => {
          socket.write(
            replyFrame(Operation.NonReplicated, Buffer.from('first'))
          );
        }, 10);
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await client.authenticate(vsrConfig(server.port).credentials);
      const first = client.sendCommand(60_015, Buffer.alloc(0));
      const second = client.sendCommand(60_016, Buffer.alloc(0));
      const login = client.sendCommand(
        COMMAND_CODE.LoginUser,
        Buffer.concat([
          Buffer.from([4]),
          Buffer.from('iggy'),
          Buffer.from([4]),
          Buffer.from('iggy')
        ])
      );
      await Promise.all([first, second, login]);

      const operations = server.frames.map(
        (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
      );
      const logoutIndex = operations.lastIndexOf(Operation.Logout);
      assert.equal(operations[logoutIndex + 1], Operation.Register);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('supports raw VSR responses without decoding', async () => {
    const server = await startVsrServer(
      (frame, socket) => singleNodeHandler(server.port)(frame, socket)
    );
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await client.authenticate(vsrConfig(server.port).credentials);
      const raw = await client.sendCommand(
        60_008,
        Buffer.alloc(0),
        { handleResponse: false }
      );
      assert.ok(Buffer.isBuffer(raw));
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('remaps a terminal VSR response to the original command', async () => {
    const server = await startVsrServer((frame, socket) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_009) {
        socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 3));
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await assert.rejects(
        () => client.sendCommand(60_009, Buffer.alloc(0)),
        (error: unknown) =>
          error instanceof ResponseError &&
          error.commandCode === 60_009 &&
          error.errorCode === 3
      );
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('rejects synchronous writes and timed-out exchanges', async () => {
    const server = await startVsrServer(() => {});
    const client = new CommandResponseStream(vsrConfig(server.port));
    const exchange = (
      client as unknown as {
        _exchange: (write: () => void, timeout?: number) => Promise<Buffer>
      }
    )._exchange.bind(client);
    try {
      const writeError = new Error('write failed');
      await assert.rejects(
        () => exchange(() => { throw writeError; }),
        (error: unknown) => error === writeError
      );
      const connection = (
        client as unknown as {
          connection: { emit: (event: string, error: Error) => boolean }
        }
      ).connection;
      const responseError = new Error('response failed');
      const pending = exchange(() => {});
      connection.emit('error', responseError);
      await assert.rejects(
        () => pending,
        (error: unknown) => error === responseError
      );
      await assert.rejects(
        () => exchange(() => {}, 1),
        /timed out after 1 ms/
      );
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('expires a VSR request before writing after its deadline', async () => {
    const server = await startVsrServer(
      (frame, socket) => singleNodeHandler(server.port)(frame, socket)
    );
    const client = new CommandResponseStream(vsrConfig(server.port));
    const realNow = Date.now;
    try {
      await client.authenticate(vsrConfig(server.port).credentials);
      let sessionResets = 0;
      client.on('sessionReset', () => { sessionResets += 1; });
      let now = 0;
      Date.now = () => {
        now += 30_001;
        return now;
      };
      await assert.rejects(
        () => client.sendCommand(60_013, Buffer.alloc(0)),
        /timed out after 30000 ms/
      );
      assert.equal(
        server.frames.some(
          (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_013
        ),
        false
      );
      assert.equal(client.isAuthenticated, true);
      assert.equal(sessionResets, 0);
    } finally {
      Date.now = realNow;
      client.destroy();
      await server.close();
    }
  });

  it('re-issues every request refused by a demoted node, not just the first',
    async () => {
      // One demotion, several refused requests: each of them re-checking on
      // its own would move the client once per request, and the first
      // redirect's drop would fail the others' roster reads - reporting a
      // refusal they never had to.
      const leader = await startVsrServer((frame, socket) => {
        singleNodeHandler(leader.port)(frame, socket);
      });
      // Leader at login, so the settlement leaves the client here, then
      // demoted: the refusals below are what tells the client to look again.
      let demotedYet = false;
      const demoted = await startVsrServer((frame, socket) => {
        const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
        if (code === COMMAND_CODE.GetClusterMetadata) {
          socket.write(replyFrame(
            Operation.NonReplicated,
            demotedYet
              ? twoNodeMetadataBody(demoted.port, leader.port)
              : twoNodeMetadataBody(leader.port, demoted.port)
          ));
          return;
        }
        if (code === 60_032) {
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
          return;
        }
        singleNodeHandler(demoted.port)(frame, socket);
      });

      const client = new CommandResponseStream(vsrConfig(demoted.port));
      try {
        await client.authenticate(vsrConfig(demoted.port).credentials);
        demotedYet = true;
        const rosterReadsBefore = demoted.frames.filter(
          (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) ===
            COMMAND_CODE.GetClusterMetadata
        ).length;

        // Two commands, both refused by the demoted node: one re-check between
        // them, and both answered on the node it moved to.
        const answers = await Promise.all([
          client.sendCommand(60_032, Buffer.alloc(0)),
          client.sendCommand(60_032, Buffer.alloc(0))
        ]);

        assert.deepEqual(answers.map((answer) => answer.status), [0, 0]);
        const rosterReads = demoted.frames.filter(
          (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) ===
            COMMAND_CODE.GetClusterMetadata
        ).length - rosterReadsBefore;
        assert.equal(rosterReads, 1,
          'each refusal re-read the roster on its own'
        );
        const reissued = leader.frames.filter(
          (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_032
        ).length;
        assert.equal(reissued, 2,
          'a refused command was not re-issued on the node the move landed on'
        );
        const connection = (client as unknown as {
          connection: { isConnectedTo: (host: string, port: number) => boolean }
        }).connection;
        assert.equal(connection.isConnectedTo('127.0.0.1', leader.port), true);
      } finally {
        client.destroy();
        await leader.close();
        await demoted.close();
      }
    }
  );

  it('walks past two refusing replicas to the partition primary',
    async () => {
      const third = await startVsrServer((frame, socket) => {
        singleNodeHandler(third.port)(frame, socket);
      });
      const second = await startVsrServer((frame, socket) => {
        const operation = frame.readUInt8(REQUEST_OFFSET.operation);
        if (operation === Operation.Register ||
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession) {
          singleNodeHandler(second.port)(frame, socket);
          return;
        }
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_040) {
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
          return;
        }
        socket.write(replyFrame(operation));
      });
      const metadataLeader = await startVsrServer((frame, socket) => {
        const operation = frame.readUInt8(REQUEST_OFFSET.operation);
        const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
        if (operation === Operation.Register) {
          socket.write(replyFrame(Operation.Register, registerReplyBody()));
          return;
        }
        if (code === COMMAND_CODE.GetClusterMetadata) {
          socket.write(replyFrame(
            Operation.NonReplicated,
            threeNodeMetadataBody(
              metadataLeader.port,
              second.port,
              third.port
            )
          ));
          return;
        }
        if (code === 60_040) {
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
          return;
        }
        socket.write(replyFrame(operation));
      });
      const client = new CommandResponseStream(vsrConfig(metadataLeader.port));
      try {
        await client.authenticate(vsrConfig(metadataLeader.port).credentials);

        const response = await client.sendCommand(60_040, Buffer.alloc(0), {
          deadline: Date.now() + 15_000
        });

        assert.equal(response.status, 0);
        assert.ok(
          second.frames.some((frame) =>
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_040),
          'the roster walk skipped the second replica'
        );
        assert.ok(
          third.frames.some((frame) =>
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_040),
          'the roster walk never reached the partition primary'
        );
      } finally {
        client.destroy();
        await metadataLeader.close();
        await second.close();
        await third.close();
      }
    }
  );

  it('holds a queued command instead of writing it to the node being left',
    async () => {
      // A refusal sends its caller to re-read the roster, and the drain that
      // handed it out keeps going. Written in that window, the next queued
      // command goes to the socket the move is about to replace: in flight when
      // that happens, it dies with a lost-connection error nobody can act on
      // instead of being re-issued on the node the move lands on.
      const leader = await startVsrServer((frame, socket) => {
        singleNodeHandler(leader.port)(frame, socket);
      });
      let demotedYet = false;
      const demoted = await startVsrServer((frame, socket) => {
        const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
        if (code === COMMAND_CODE.GetClusterMetadata) {
          socket.write(replyFrame(
            Operation.NonReplicated,
            demotedYet
              ? twoNodeMetadataBody(demoted.port, leader.port)
              : twoNodeMetadataBody(leader.port, demoted.port)
          ));
          return;
        }
        if (code === 60_037) {
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
          return;
        }
        if (code === 60_038) {
          // Accepted and never answered: a command written here is stuck until
          // the move replaces the socket under it.
          return;
        }
        singleNodeHandler(demoted.port)(frame, socket);
      });

      const client = new CommandResponseStream(vsrConfig(demoted.port));
      try {
        await client.authenticate(vsrConfig(demoted.port).credentials);
        demotedYet = true;

        // The second command is queued while the first is in flight, which is
        // where a command caught in a move comes from.
        const refused = client.sendCommand(60_037, Buffer.alloc(0));
        const behind = client.sendCommand(60_038, Buffer.alloc(0));
        refused.catch(() => undefined);
        behind.catch(() => undefined);

        const settled = await Promise.race([
          Promise.all([refused, behind]).then(() => 'answered'),
          new Promise((resolve) => {
            setTimeout(() => resolve('stalled'), 10_000).unref();
          })
        ]);
        assert.equal(settled, 'answered',
          'the command behind the refusal went out on the node being left'
        );
        assert.ok(
          !demoted.frames.some((frame) =>
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_038),
          'the command behind the refusal was written to the node being left'
        );
        assert.ok(
          leader.frames.some((frame) =>
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_038),
          'the command behind the refusal never reached the node moved to'
        );
      } finally {
        client.destroy();
        await leader.close();
        await demoted.close();
      }
    }
  );

  it('re-issues a command queued behind a leader move instead of failing it',
    async () => {
      // A move replaces the socket, which looks like a drop to everything
      // waiting in the queue. Nothing queued was written, though, so it belongs
      // on the node the client moves to rather than in a lost-connection error
      // the caller can do nothing about.
      const leader = await startVsrServer((frame, socket) => {
        singleNodeHandler(leader.port)(frame, socket);
      });
      const demoted = await startVsrServer((frame, socket) => {
        singleNodeHandler(demoted.port)(frame, socket);
      });

      const client = new CommandResponseStream(vsrConfig(demoted.port));
      try {
        await client.authenticate(vsrConfig(demoted.port).credentials);

        // Parked the way a command is while something else holds the queue.
        const queued = new Promise<CommandResponse>((resolve, reject) => {
          execQueue(client).push({
            command: 60_034,
            payload: Buffer.alloc(0),
            handleResponse: true,
            deadline: Date.now() + 30_000,
            resolve,
            reject
          });
        });

        await connectionOf(client).redirect('127.0.0.1', leader.port);
        await queued;

        const landedOnLeader = leader.frames.some(
          (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_034
        );
        assert.ok(landedOnLeader,
          'the queued command never reached the node the client moved to'
        );
      } finally {
        client.destroy();
        await leader.close();
        await demoted.close();
      }
    }
  );

  it('surfaces the refusal rather than a timeout when the budget runs out',
    async () => {
      const server = await startVsrServer((frame, socket) => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_036) {
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
          return;
        }
        singleNodeHandler(server.port)(frame, socket);
      });
      const client = new CommandResponseStream(vsrConfig(server.port));
      const realNow = Date.now;
      try {
        await client.authenticate(vsrConfig(server.port).credentials);
        // The request's budget, the first exchange, the window that hands the
        // refusal out, and then a clock 10ms short of the deadline: too little
        // to carry another attempt, so the caller has to see the answer the
        // server gave rather than the timeout a doomed re-issue would produce.
        const times = [0, 1, 2_001, 29_990];
        Date.now = () => times.shift() ?? 30_050;

        await assert.rejects(
          () => client.sendCommand(60_036, Buffer.alloc(0)),
          (error: unknown) =>
            error instanceof ResponseError &&
            error.commandCode === 60_036 &&
            error.errorCode === 58
        );
      } finally {
        Date.now = realNow;
        client.destroy();
        await server.close();
      }
    }
  );

  it('does not suppress leader settlement after a failed roster walk',
    async () => {
      const server = await startVsrServer(
        (frame, socket) => singleNodeHandler(server.port)(frame, socket)
      );
      const client = new CommandResponseStream(vsrConfig(server.port));
      try {
        await client.authenticate(vsrConfig(server.port).credentials);
        const routing = client as unknown as {
          walkSettleSuppressed: boolean,
          _followLeaderMove: (walkPastLeader: boolean) => Promise<unknown>,
          connection: {
            nextRosterEndpoint: () => { host: string, port: number },
            redirect: (host: string, port: number) => Promise<void>
          }
        };
        routing.connection.nextRosterEndpoint = () => ({
          host: '127.0.0.1',
          port: server.port + 1
        });
        routing.connection.redirect = async () => {
          throw new Error('redirect failed');
        };

        assert.deepEqual(await routing._followLeaderMove(true), {
          endpoint: { host: '127.0.0.1', port: server.port + 1 },
          moved: false
        });
        assert.equal(routing.walkSettleSuppressed, false,
          'a failed walk leaked its one-shot suppression into a later login'
        );
      } finally {
        client.destroy();
        await server.close();
      }
    }
  );

  it('surfaces the refusal when the move left too little of the budget',
    async () => {
      // The move itself costs budget: a roster read, an election it waited out,
      // a redial. What is left can be positive and still too small to carry
      // another exchange, and re-issued into it the request times out -- the
      // caller then sees a timeout where the answer was "not admitted".
      const leader = await startVsrServer((frame, socket) => {
        singleNodeHandler(leader.port)(frame, socket);
      });
      let demotedYet = false;
      const demoted = await startVsrServer((frame, socket) => {
        const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
        if (code === COMMAND_CODE.GetClusterMetadata) {
          socket.write(replyFrame(
            Operation.NonReplicated,
            demotedYet
              ? twoNodeMetadataBody(demoted.port, leader.port)
              : twoNodeMetadataBody(leader.port, demoted.port)
          ));
          return;
        }
        if (code === 60_039) {
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
          return;
        }
        singleNodeHandler(demoted.port)(frame, socket);
      });

      const client = new CommandResponseStream(vsrConfig(demoted.port));
      const realNow = Date.now;
      let offset = 0;
      try {
        await client.authenticate(vsrConfig(demoted.port).credentials);
        demotedYet = true;

        const deadline = realNow() + 30_000;
        const connection = connectionOf(client);
        const move = connection.redirect.bind(connection);
        connection.redirect = async (host: string, port: number) => {
          await move(host, port);
          // 30ms of budget left the moment the client lands: positive, and
          // below the interval one exchange needs.
          offset = deadline - realNow() - 30;
        };
        Date.now = () => realNow() + offset;

        await assert.rejects(
          () => client.sendCommand(60_039, Buffer.alloc(0), { deadline }),
          (error: unknown) =>
            error instanceof ResponseError &&
            error.commandCode === 60_039 &&
            error.errorCode === 58
        );
        assert.ok(
          !leader.frames.some((frame) =>
            frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_039),
          'the request was re-issued into a budget too small to answer it'
        );
      } finally {
        Date.now = realNow;
        client.destroy();
        await leader.close();
        await demoted.close();
      }
    }
  );

  it('paces the re-issues while the roster still names this node',
    async () => {
      // Re-issuing is right, spinning is not: the in-connection replay window
      // belongs to the request's budget and is spent after the first pass, so
      // without a wait the client would hammer the node for the whole budget.
      let refusals = 0;
      const server = await startVsrServer((frame, socket) => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_035) {
          refusals += 1;
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
          return;
        }
        singleNodeHandler(server.port)(frame, socket);
      });
      const client = new CommandResponseStream(vsrConfig(server.port));
      try {
        await client.authenticate(vsrConfig(server.port).credentials);

        const pending = client.sendCommand(60_035, Buffer.alloc(0));
        pending.catch(() => undefined);
        await new Promise((resolve) => {
          setTimeout(resolve, 5_000).unref();
        });

        // The first 2s window replays on the connection at its own interval;
        // every window after it costs one refusal per pace.
        assert.ok(refusals < 200,
          `the re-issues were not paced: ${refusals} refusals in 5s`
        );
      } finally {
        client.destroy();
        await server.close();
      }
    }
  );

  it('keeps re-issuing a not-admitted request while the roster still names this node',
    async () => {
      const server = await startVsrServer((frame, socket) => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_031) {
          socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
          return;
        }
        singleNodeHandler(server.port)(frame, socket);
      });
      const client = new CommandResponseStream(vsrConfig(server.port));
      try {
        await client.authenticate(vsrConfig(server.port).credentials);

        // A refusal the roster cannot explain is a wait, not a verdict: an
        // election may still be in flight, so the request keeps going for its
        // whole budget instead of failing after the first re-check window.
        const pending = client.sendCommand(60_031, Buffer.alloc(0));
        const outcome = await Promise.race([
          pending.then(() => 'answered', () => 'gave up'),
          new Promise((resolve) => {
            setTimeout(() => resolve('still trying'), 4_000).unref();
          })
        ]);

        assert.equal(outcome, 'still trying');
      } finally {
        client.destroy();
        await server.close();
      }
    }
  );

  it('keeps a typed transient error and session at its retry deadline',
    async () => {
      const server = await startVsrServer((frame, socket) => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_017) {
          socket.write(
            replyFrame(Operation.NonReplicated, Buffer.alloc(0), 57)
          );
          return;
        }
        singleNodeHandler(server.port)(frame, socket);
      });
      const client = new CommandResponseStream(vsrConfig(server.port));
      const realNow = Date.now;
      try {
        await client.authenticate(vsrConfig(server.port).credentials);
        let sessionResets = 0;
        client.on('sessionReset', () => { sessionResets += 1; });
        const times = [0, 1, 30_001];
        Date.now = () => times.shift() ?? 30_001;

        await assert.rejects(
          () => client.sendCommand(60_017, Buffer.alloc(0)),
          (error: unknown) =>
            error instanceof ResponseError &&
            error.commandCode === 60_017 &&
            error.errorCode === 57
        );
        assert.equal(client.isAuthenticated, true);
        assert.equal(sessionResets, 0);
      } finally {
        Date.now = realNow;
        client.destroy();
        await server.close();
      }
    }
  );

  for (const refusal of [58, 89, 41, 40, 30, 87, 4000]) {
  it(`keeps an uncertain write uncertain after a later refusal ${refusal}`, async () => {
    const EXPIRED_REQUEST_TIME = 30_001;
    let now = 0;
    let attempts = 0;
    const server = await startVsrServer((frame, socket) => {
      if (frame.readUInt8(REQUEST_OFFSET.operation) === Operation.SendMessages) {
        attempts += 1;
        now = attempts === 1 ? 1 : EXPIRED_REQUEST_TIME;
        socket.write(replyFrame(
          Operation.SendMessages, Buffer.alloc(0), attempts === 1 ? 57 : refusal
        ));
        return;
      }
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.GetPartitionRoute) {
        socket.write(replyFrame(Operation.NonReplicated, routingBody(0n, 0n, 0)));
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    const realNow = Date.now;
    try {
      await client.authenticate(vsrConfig(server.port).credentials);
      let sessionResets = 0;
      client.on('sessionReset', () => { sessionResets += 1; });
      Date.now = () => now;
      await assert.rejects(
        () => client.sendCommand(COMMAND_CODE.SendMessages, explicitSend),
        (error: unknown) =>
          error instanceof ResponseError &&
          error.commandCode === COMMAND_CODE.SendMessages &&
          error.errorCode === 57
      );
      const writes = server.frames.filter(
        (frame) => frame.readUInt8(REQUEST_OFFSET.operation) === Operation.SendMessages
      );
      assert.equal(writes.length, 2);
      assert.deepEqual(writes[1], writes[0]);
      assert.equal(client.isAuthenticated, true);
      assert.equal(sessionResets, 0);
    } finally {
      Date.now = realNow;
      client.destroy();
      await server.close();
    }
  });
  }

  it('reuses a send context and refreshes it once after a history refusal', async () => {
    const state = { live: 1n, refuseAll: false, contextStatus: 0 };
    const server = await contextServer(state);
    const client = new CommandResponseStream(vsrConfig(server.port));
    const incarnations = () => server.frames
      .filter((frame) => frame.readUInt8(REQUEST_OFFSET.operation) === Operation.SendMessages)
      .map((frame) => frame.readBigUInt64LE(REQUEST_OFFSET.partitionIncarnation));
    try {
      await client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
      await client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 1);
      state.live = 2n;
      await client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 2);
      assert.deepEqual(incarnations(), [1n, 1n, 1n, 2n]);
      state.refuseAll = true;
      await assert.rejects(() => client.sendCommand(COMMAND_CODE.SendMessages, explicitSend),
        (error: unknown) => error instanceof ResponseError &&
          error.commandCode === COMMAND_CODE.SendMessages && error.errorCode === HISTORY_UNAVAILABLE);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 3);
      assert.equal(incarnations().length, 6);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('reports a refused context query as an error of the send', async () => {
    const server = await contextServer({ live: 1n, refuseAll: false, contextStatus: 2010 });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await assert.rejects(() => client.sendCommand(COMMAND_CODE.SendMessages, explicitSend),
        (error: unknown) => error instanceof ResponseError &&
          error.commandCode === COMMAND_CODE.SendMessages && error.errorCode === 2010);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('keeps an eviction during the send context query typed', async () => {
    const server = await startVsrServer((frame, socket) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.GetPartitionRoute) {
        socket.write(evictionFrame(EvictionReason.NoSession));
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await assert.rejects(() => client.sendCommand(COMMAND_CODE.SendMessages, explicitSend),
        (error: unknown) => error instanceof VsrEvictionError && error.errorCode === 40);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('surfaces a history refusal the remaining budget cannot refresh', async () => {
    const realNow = Date.now;
    const deadline = realNow() + 30_000;
    let offset = 0;
    const server = await contextServer({
      live: 1n, refuseAll: true, contextStatus: 0,
      // 10ms of budget left when the refusal comes back: too little to carry a
      // context query and another send.
      onSend: () => { offset = deadline - realNow() - 10; }
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await client.authenticate(vsrConfig(server.port).credentials);
      Date.now = () => realNow() + offset;
      await assert.rejects(
        () => client.sendCommand(COMMAND_CODE.SendMessages, explicitSend, { deadline }),
        (error: unknown) => error instanceof ResponseError &&
          error.commandCode === COMMAND_CODE.SendMessages && error.errorCode === HISTORY_UNAVAILABLE);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 1);
    } finally {
      Date.now = realNow;
      client.destroy();
      await server.close();
    }
  });

  it('keeps the captured context of an offset store re-issued after a leader move', async () => {
    // The store captures incarnation 1 and then waits behind a command the old
    // node never answers. The move re-issues it on a node whose route already
    // reports incarnation 2: an offset taken from 1 must not commit under 2.
    const leader = await startVsrServer((frame, socket) => {
      if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.GetPartitionRoute) {
        socket.write(replyFrame(Operation.NonReplicated, routingBody(1n, 2n)));
        return;
      }
      offsetCommittingHandler(leader.port)(frame, socket);
    });
    const demoted = await startVsrServer((frame, socket) => {
      const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
      if (code === COMMAND_CODE.GetPartitionRoute) {
        socket.write(replyFrame(Operation.NonReplicated, routingBody(1n, 1n)));
        return;
      }
      if (code === 60_038) {
        void connectionOf(client).redirect('127.0.0.1', leader.port);
        return;
      }
      singleNodeHandler(demoted.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(demoted.port));
    try {
      await client.authenticate(vsrConfig(demoted.port).credentials);
      const store = client.sendCommand(COMMAND_CODE.StoreOffset, storeOffsetPayload);
      client.sendCommand(60_038, Buffer.alloc(0)).catch(() => undefined);
      await store;
      const stores = framesOf(leader, Operation.StoreConsumerOffset);
      assert.equal(stores.length, 1);
      assert.equal(stampedContext(stores[0]).incarnation, 1n);
      assert.equal(countCommand(leader, COMMAND_CODE.GetPartitionRoute), 0);
      assert.equal(framesOf(demoted, Operation.StoreConsumerOffset).length, 0);
    } finally {
      client.destroy();
      await leader.close();
      await demoted.close();
    }
  });

  it('refreshes a send context at most once when a leader move re-issues the send', async () => {
    const leader = await contextServer({ live: 0n, refuseAll: true, contextStatus: 0 });
    let contextQueries = 0;
    const demoted = await startVsrServer((frame, socket) => {
      if (frame.readUInt8(REQUEST_OFFSET.operation) === Operation.SendMessages) {
        socket.write(replyFrame(Operation.SendMessages, Buffer.alloc(0), HISTORY_UNAVAILABLE));
        return;
      }
      const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
      if (code === COMMAND_CODE.GetPartitionRoute) {
        contextQueries += 1;
        // The refresh: park the refreshed send behind a command that this
        // node never answers, so the move finds it in the queue.
        if (contextQueries === 2)
          client.sendCommand(60_038, Buffer.alloc(0)).catch(() => undefined);
        const context = routingBody(0n, 0n, 0);
        context.writeBigUInt64LE(BigInt(contextQueries), 0);
        socket.write(replyFrame(Operation.NonReplicated, context));
        return;
      }
      if (code === 60_038) {
        void connectionOf(client).redirect('127.0.0.1', leader.port);
        return;
      }
      singleNodeHandler(demoted.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(demoted.port));
    try {
      await client.authenticate(vsrConfig(demoted.port).credentials);
      await assert.rejects(() => client.sendCommand(COMMAND_CODE.SendMessages, explicitSend),
        (error: unknown) => error instanceof ResponseError &&
          error.commandCode === COMMAND_CODE.SendMessages && error.errorCode === HISTORY_UNAVAILABLE);
      assert.equal(contextQueries, 2);
      assert.equal(countCommand(leader, COMMAND_CODE.GetPartitionRoute), 0);
      assert.deepEqual(framesOf(leader, Operation.SendMessages)
        .map((frame) => stampedContext(frame).incarnation), [2n]);
    } finally {
      client.destroy();
      await leader.close();
      await demoted.close();
    }
  });

  for (const partition of [Partitioning.Balanced, Partitioning.MessageKey('key')]) {
    it(`refuses a raw send with partitioning kind ${partition.kind} before writing it`, async () => {
      const server = await startVsrServer((frame, socket) => singleNodeHandler(server.port)(frame, socket));
      const client = new CommandResponseStream(vsrConfig(server.port));
      try {
        await assert.rejects(() => client.sendCommand(COMMAND_CODE.SendMessages, sendTo(partition)),
          (error: unknown) => error instanceof ResponseError &&
            error.commandCode === COMMAND_CODE.SendMessages && error.errorCode === 5 &&
            error.message.includes('explicit partition id') && error.message.includes('typed send API'));
        assert.equal(framesOf(server, Operation.SendMessages).length, 0);
        assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 0);
      } finally {
        client.destroy();
        await server.close();
      }
    });
  }

  for (const [name, code, payload, operation] of [
    ['an explicit-partition send', COMMAND_CODE.SendMessages, explicitSend, Operation.SendMessages],
    ['an offset store', COMMAND_CODE.StoreOffset, storeOffsetPayload, Operation.StoreConsumerOffset],
    ['an offset delete', COMMAND_CODE.DeleteConsumerOffset, deleteOffsetPayload, Operation.DeleteConsumerOffset],
    ['a standalone poll', COMMAND_CODE.PollMessages, groupPollPayload(false), Operation.NonReplicated]
  ] as const) {
    it(`stamps the caller's context on ${name} while looking up its primary`, async () => {
      const server = await startVsrServer((frame, socket) => offsetCommittingHandler(server.port)(frame, socket));
      const client = new CommandResponseStream(vsrConfig(server.port));
      try {
        await client.sendCommand(code, payload, { context: CALLER_CONTEXT });
        const requests = framesOf(server, operation).filter((frame) =>
          operation !== Operation.NonReplicated || frame.readUInt32LE(REQUEST_OFFSET.reserved) === code);
        assert.equal(requests.length, 1);
        assert.deepEqual(stampedContext(requests[0]), CALLER_CONTEXT);
        assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 1);
      } finally {
        client.destroy();
        await server.close();
      }
    });
  }

  it("stamps the caller's context on a primary poll", async () => {
    const cluster = await startPollCluster();
    try {
      await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload(), { context: CALLER_CONTEXT });
      const polls = cluster.primary.frames.filter((frame) =>
        frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.PollMessages);
      assert.equal(polls.length, 1);
      assert.deepEqual(stampedContext(polls[0]), CALLER_CONTEXT);
    } finally {
      await cluster.close();
    }
  });

  it('keeps a caller context on a send refused as history unavailable', async () => {
    const server = await contextServer({ live: 1n, refuseAll: true, contextStatus: 0 });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await assert.rejects(
        () => client.sendCommand(COMMAND_CODE.SendMessages, explicitSend, { context: CALLER_CONTEXT }),
        (error: unknown) => error instanceof ResponseError && error.errorCode === HISTORY_UNAVAILABLE);
      assert.deepEqual(framesOf(server, Operation.SendMessages).map(stampedContext), [CALLER_CONTEXT]);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 1);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('retries primary lifecycle refusals with new ids and the captured context while the coordinator proceeds', async () => {
    const order: string[] = [];
    let refusals = 2;
    let routes = 0;
    let firstRefusal!: () => void;
    const refused = new Promise<void>((resolve) => { firstRefusal = resolve; });
    const otherCommand = 60_041;
    const cluster = await startPollCluster((frame, socket) => {
      if (frame.readUInt8(REQUEST_OFFSET.operation) !== Operation.SendMessages)
        return false;
      order.push('send');
      socket.write(replyFrame(Operation.SendMessages, Buffer.alloc(0), refusals-- > 0 ? 88 : 0));
      firstRefusal();
      return true;
    }, (frame, socket) => {
      const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
      if (code === COMMAND_CODE.GetPartitionRoute) {
        routes += 1;
        socket.write(replyFrame(Operation.NonReplicated, routingBody(BigInt(routes), BigInt(routes))));
        return true;
      }
      if (code === otherCommand)
        order.push('other');
      return false;
    });
    try {
      await cluster.client.authenticate(vsrConfig(cluster.coordinator.port).credentials);
      const started = Date.now();
      const send = cluster.client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
      await refused;
      await cluster.client.sendCommand(otherCommand, Buffer.alloc(0));
      await send;
      const sends = framesOf(cluster.primary, Operation.SendMessages);
      assert.equal(sends.length, 3);
      assert.equal(new Set(sends.map((frame) => frame.readBigUInt64LE(REQUEST_OFFSET.request))).size, 3);
      assert.deepEqual(sends.map(stampedContext), Array(3).fill({
        incarnation: 1n, ownerGeneration: 0n, metadataOp: 1n
      }), 'a refreshed route must not replace the captured operation context');
      assert.deepEqual(order, ['send', 'other', 'send', 'send']);
      assert.ok(Date.now() - started >= 145, 'the pauses grow from 50 ms to 100 ms');
      assert.equal(countCommand(cluster.primary, COMMAND_CODE.GetClusterMetadata), 0);
    } finally {
      await cluster.close();
    }
  });

  for (const code of [COMMAND_CODE.GetPartitionRoute, COMMAND_CODE.BindSession]) {
    it(`does not retry a lifecycle refusal from routing command ${code}`, async () => {
      const override = (frame: Buffer, socket: Socket): boolean => {
        if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== code)
          return false;
        socket.write(replyFrame(frame.readUInt8(REQUEST_OFFSET.operation), Buffer.alloc(0), 88));
        return true;
      };
      const cluster = await startPollCluster(override, override);
      try {
        await cluster.client.authenticate(vsrConfig(cluster.coordinator.port).credentials);
        await assert.rejects(
          () => cluster.client.sendCommand(COMMAND_CODE.SendMessages, explicitSend,
            { deadline: Date.now() + 200 }),
          (error: unknown) => error instanceof ResponseError &&
            error.commandCode === COMMAND_CODE.SendMessages && error.errorCode === 88);
        assert.equal(countCommand(code === COMMAND_CODE.GetPartitionRoute
          ? cluster.coordinator : cluster.primary, code), 1);
        assert.equal(framesOf(cluster.primary, Operation.SendMessages).length, 0);
      } finally {
        await cluster.close();
      }
    });
  }

  it('returns a primary lifecycle refusal within the original request deadline', async () => {
    const cluster = await startPollCluster((frame, socket) => {
      if (frame.readUInt8(REQUEST_OFFSET.operation) !== Operation.SendMessages)
        return false;
      socket.write(replyFrame(Operation.SendMessages, Buffer.alloc(0), 88));
      return true;
    });
    try {
      await cluster.client.authenticate(vsrConfig(cluster.coordinator.port).credentials);
      const deadline = Date.now() + 400;
      await assert.rejects(
        () => cluster.client.sendCommand(COMMAND_CODE.SendMessages, explicitSend, { deadline }),
        (error: unknown) => error instanceof ResponseError &&
          error.commandCode === COMMAND_CODE.SendMessages && error.errorCode === 88);
      assert.ok(Date.now() <= deadline, 'the next exponential pause must fit the original deadline');
      assert.ok(framesOf(cluster.primary, Operation.SendMessages).length >= 2);
    } finally {
      await cluster.close();
    }
  });

  it('retries a lifecycle refusal as new requests while other commands proceed', async () => {
    let refusals = 2;
    const server = await startVsrServer((frame, socket) => {
      if (frame.readUInt8(REQUEST_OFFSET.operation) === Operation.CreateStream) {
        socket.write(replyFrame(Operation.CreateStream, metadataResult(refusals-- > 0 ? 88 : 0)));
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await client.authenticate(vsrConfig(server.port).credentials);
      const started = Date.now();
      const create = client.sendCommand(COMMAND_CODE.CreateStream, Buffer.alloc(0));
      const other = client.sendCommand(60_040, Buffer.alloc(0));
      await Promise.all([create, other]);
      const elapsed = Date.now() - started;
      const creates = framesOf(server, Operation.CreateStream);
      assert.equal(creates.length, 3);
      assert.equal(new Set(creates.map((frame) =>
        frame.readBigUInt64LE(REQUEST_OFFSET.request))).size, 3, 'each retry is a new request');
      assert.ok(elapsed >= 145, `the pauses grow from 50 ms, took ${elapsed} ms`);
      const order = server.frames.map((frame) => frame.readUInt8(REQUEST_OFFSET.operation) === Operation.CreateStream
        ? 'create' : frame.readUInt32LE(REQUEST_OFFSET.reserved)).filter((entry) => entry === 'create' || entry === 60_040);
      assert.deepEqual(order, ['create', 60_040, 'create', 'create'], 'the pause must not hold the queue');
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('returns the lifecycle refusal once the request deadline leaves no room for a retry', async () => {
    const server = await startVsrServer((frame, socket) => {
      if (frame.readUInt8(REQUEST_OFFSET.operation) === Operation.CreateStream) {
        socket.write(replyFrame(Operation.CreateStream, metadataResult(88)));
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await client.authenticate(vsrConfig(server.port).credentials);
      const deadline = Date.now() + 400;
      await assert.rejects(
        () => client.sendCommand(COMMAND_CODE.CreateStream, Buffer.alloc(0), { deadline }),
        (error: unknown) => error instanceof ResponseError &&
          error.commandCode === COMMAND_CODE.CreateStream && error.errorCode === 88);
      assert.ok(Date.now() <= deadline, 'the last pause must fit the budget');
      assert.ok(framesOf(server, Operation.CreateStream).length >= 2);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('drops the cached send contexts of a topic after a history refusal', async () => {
    const state = { live: 1n, refuseAll: false, contextStatus: 0 };
    const server = await contextServer(state);
    const client = new CommandResponseStream(vsrConfig(server.port));
    const otherPartition = sendTo(Partitioning.PartitionId(4));
    try {
      await client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
      await client.sendCommand(COMMAND_CODE.SendMessages, otherPartition);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 2);
      state.live = 2n;
      await client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 3);
      await client.sendCommand(COMMAND_CODE.SendMessages, otherPartition);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 4,
        'a refusal for one partition drops its whole topic');
      assert.deepEqual(framesOf(server, Operation.SendMessages).map((frame) =>
        stampedContext(frame).incarnation), [1n, 1n, 1n, 2n, 2n]);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  for (const code of [COMMAND_CODE.DeleteStream, COMMAND_CODE.DeleteTopic,
    COMMAND_CODE.CreatePartitions, COMMAND_CODE.DeletePartitions]) {
    it(`drops every cached send context after the client's own command ${code}`, async () => {
      const server = await contextServer({ live: 1n, refuseAll: false, contextStatus: 0 });
      const client = new CommandResponseStream(vsrConfig(server.port));
      let resets = 0;
      client.on('topicDiscoveryReset', () => { resets += 1; });
      try {
        await client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
        await client.sendCommand(code, Buffer.alloc(0), { handleResponse: false });
        await client.sendCommand(COMMAND_CODE.SendMessages, explicitSend);
        assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 2);
        assert.equal(resets, 1, 'typed sends read the partition count again');
      } finally {
        client.destroy();
        await server.close();
      }
    });
  }

  for (const [name, code, operation, payload] of [
    ['store', COMMAND_CODE.StoreOffset, Operation.StoreConsumerOffset, storeOffsetPayload],
    ['delete', COMMAND_CODE.DeleteConsumerOffset, Operation.DeleteConsumerOffset, deleteOffsetPayload]
  ] as const) {
    it(`writes an offset ${name} on the partition primary and keeps the main connection`, async () => {
      const cluster = await startPollCluster(
        (frame, socket) => {
          if (frame.readUInt8(REQUEST_OFFSET.operation) !== operation)
            return false;
          socket.write(replyFrame(operation, metadataResult(0)));
          return true;
        },
        (frame, socket) => {
          if (frame.readUInt8(REQUEST_OFFSET.operation) !== operation)
            return false;
          socket.write(replyFrame(operation, Buffer.alloc(0), TRANSIENT_NOT_ACCEPTED));
          return true;
        });
      try {
        await cluster.client.sendCommand(code, payload, { deadline: Date.now() + 2_000 });
        assert.equal(framesOf(cluster.coordinator, operation).length, 0);
        assert.equal(framesOf(cluster.primary, operation).length, 1);
        const rosterReads = countCommand(cluster.coordinator, COMMAND_CODE.GetClusterMetadata);
        await cluster.client.sendCommand(COMMAND_CODE.GetClusterMetadata, Buffer.alloc(0));
        assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetClusterMetadata), rosterReads + 1,
          'the main connection stays on its node');
        assert.equal(framesOf(cluster.coordinator, Operation.Register).length, 1, 'the session stays');
        assert.equal(framesOf(cluster.primary, Operation.Register).length, 0);
      } finally {
        await cluster.close();
      }
    });
  }

  it('takes the request ids of primary offset writes from the session counter', async () => {
    // The partition deduplicates by client id and request id, so an id the
    // session already used would answer a write with the cached reply of
    // another. Reading routes also consumes ids from the same counter.
    let stores = 0;
    const cluster = await startPollCluster(
      (frame, socket) => {
        if (frame.readUInt8(REQUEST_OFFSET.operation) !== Operation.StoreConsumerOffset)
          return false;
        stores += 1;
        socket.write(replyFrame(Operation.StoreConsumerOffset, metadataResult(0),
          stores === 1 ? HISTORY_UNAVAILABLE : 0));
        return true;
      },
      (frame, socket) => {
        if (frame.readUInt8(REQUEST_OFFSET.operation) !== Operation.CreateStream)
          return false;
        socket.write(replyFrame(Operation.CreateStream, metadataResult(0)));
        return true;
      });
    try {
      await cluster.client.sendCommand(COMMAND_CODE.CreateStream, Buffer.alloc(0));
      await assert.rejects(cluster.client.sendCommand(COMMAND_CODE.StoreOffset, storeOffsetPayload),
        (error: unknown) => error instanceof ResponseError && error.errorCode === HISTORY_UNAVAILABLE);
      await cluster.client.sendCommand(COMMAND_CODE.StoreOffset, storeOffsetPayload);
      assert.equal(countCommand(cluster.primary, COMMAND_CODE.BindSession), 1,
        'a history refusal retains the authenticated connection');
      const requests = [
        ...framesOf(cluster.coordinator, Operation.CreateStream),
        ...framesOf(cluster.primary, Operation.StoreConsumerOffset)
      ].map((frame) => frame.readBigUInt64LE(REQUEST_OFFSET.request));
      assert.equal(new Set(requests).size, requests.length);
      assert.ok(requests[0] < requests[1] && requests[1] < requests[2]);
    } finally {
      await cluster.close();
    }
  });

  for (const callerContext of [undefined, CALLER_CONTEXT]) {
    it(`keeps one ${callerContext ? 'caller' : 'route'} context across primary retries of an offset store`, async () => {
      let routes = 0;
      let writes = 0;
      const cluster = await startPollCluster(
        (frame, socket) => {
          if (frame.readUInt8(REQUEST_OFFSET.operation) !== Operation.StoreConsumerOffset)
            return false;
          writes += 1;
          socket.write(writes === 1
            ? replyFrame(Operation.StoreConsumerOffset, Buffer.alloc(0), TRANSIENT_NOT_ACCEPTED)
            : replyFrame(Operation.StoreConsumerOffset, metadataResult(0)));
          return true;
        },
        (frame, socket) => {
          if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.GetPartitionRoute)
            return false;
          routes += 1;
          socket.write(replyFrame(Operation.NonReplicated,
            routingBody(1n, BigInt(routes))));
          return true;
        });
      let resets = 0;
      cluster.client.on('sessionReset', () => { resets += 1; });
      try {
        await cluster.client.sendCommand(COMMAND_CODE.StoreOffset, storeOffsetPayload,
          { context: callerContext });
        assert.equal(routes, 2, 'a refused write routes again');
        assert.equal(resets, 0, 'a primary refusal leaves the main connection alone');
        assert.equal(framesOf(cluster.coordinator, Operation.Register).length, 1);
        assert.deepEqual(framesOf(cluster.primary, Operation.StoreConsumerOffset).map(stampedContext),
          Array(2).fill(callerContext ?? { incarnation: 1n, ownerGeneration: 0n, metadataOp: 1n }));
      } finally {
        await cluster.close();
      }
    });
  }

  it('refreshes standalone routes after metadata commits', async () => {
    const server = await startVsrServer((frame, socket) => {
      if (frame.readUInt8(REQUEST_OFFSET.operation) === Operation.CreateStream) {
        const reply = replyFrame(Operation.CreateStream, metadataResult(0));
        reply.writeBigUInt64LE(5n, REPLY_OFFSET.commit);
        socket.write(reply);
        return;
      }
      offsetCommittingHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      for (let call = 0; call < 2; call += 1) {
        await client.sendCommand(POLL_MESSAGES.code, groupPollPayload(false));
        await client.sendCommand(COMMAND_CODE.StoreOffset, storeOffsetPayload);
        await client.sendCommand(COMMAND_CODE.DeleteConsumerOffset, deleteOffsetPayload);
      }
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 2);
      await client.sendCommand(COMMAND_CODE.CreateStream, Buffer.alloc(0));
      await client.sendCommand(POLL_MESSAGES.code, groupPollPayload(false));
      await client.sendCommand(COMMAND_CODE.StoreOffset, storeOffsetPayload);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 4);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('surfaces a refusal under a stale route once and routes the next call again', async () => {
    let live = 1n;
    const server = await startVsrServer((frame, socket) => {
      const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
      if (code === COMMAND_CODE.GetPartitionRoute) {
        socket.write(replyFrame(Operation.NonReplicated, routingBody(1n, live)));
        return;
      }
      if (code === COMMAND_CODE.PollMessages) {
        socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0),
          stampedContext(frame).incarnation === live ? 0 : HISTORY_UNAVAILABLE));
        return;
      }
      singleNodeHandler(server.port)(frame, socket);
    });
    const client = new CommandResponseStream(vsrConfig(server.port));
    try {
      await client.sendCommand(POLL_MESSAGES.code, groupPollPayload(false));
      live = 2n;
      await assert.rejects(() => client.sendCommand(POLL_MESSAGES.code, groupPollPayload(false)),
        (error: unknown) => error instanceof ResponseError && error.errorCode === HISTORY_UNAVAILABLE);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 1, 'no silent retry');
      await client.sendCommand(POLL_MESSAGES.code, groupPollPayload(false));
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 2);
      assert.deepEqual(server.frames
        .filter((frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.PollMessages)
        .map((frame) => stampedContext(frame).incarnation), [1n, 1n, 2n]);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('rediscovers an evicted route when the partition cache is full', async () => {
    const server = await contextServer({ live: 1n, refuseAll: false, contextStatus: 0 });
    const client = new CommandResponseStream(vsrConfig(server.port));
    const partitions = Array.from({ length: 4097 }, (_, partitionId) => sendTo(Partitioning.PartitionId(partitionId)));
    try {
      await Promise.all(partitions.map((payload) => client.sendCommand(COMMAND_CODE.SendMessages, payload)));
      await client.sendCommand(COMMAND_CODE.SendMessages, partitions[0]);
      assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), partitions.length + 1);
    } finally {
      client.destroy();
      await server.close();
    }
  });

  it('shares token authentication between concurrent callers', async () => {
    const server = await startVsrServer(
      (frame, socket) => singleNodeHandler(server.port)(frame, socket)
    );
    const client = new CommandResponseStream({
      ...vsrConfig(server.port),
      credentials: { token: 'secret' }
    });
    try {
      await Promise.all([
        client.sendCommand(60_011, Buffer.alloc(0)),
        client.sendCommand(60_012, Buffer.alloc(0))
      ]);
      assert.equal(
        server.frames.filter(
          (frame) =>
            frame.readUInt8(REQUEST_OFFSET.operation) === Operation.Register
        ).length,
        1
      );
      assert.equal(client.isAuthenticated, true);
    } finally {
      client.destroy();
      await server.close();
    }
  });
});