| // 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 { 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 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 registerReplyBody = (): 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(TEST_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 + 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.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, |
| port: number, |
| role: number, |
| status = 0 |
| ): Buffer => { |
| const ip = Buffer.from('127.0.0.1'); |
| const encoded = Buffer.alloc(4 + nodeName.length + 4 + ip.length + 8 + 2); |
| let offset = 0; |
| 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'), leaderPort, leaderRole, leaderStatus), |
| node(Buffer.from('iggy-node-2'), followerPort, 1) |
| ]); |
| }; |
| |
| const threeNodeMetadataBody = ( |
| firstPort: number, |
| secondPort: number, |
| thirdPort: number |
| ): Buffer => { |
| const node = (name: string, port: number, role: number): Buffer => { |
| const nodeName = Buffer.from(name); |
| const ip = Buffer.from('127.0.0.1'); |
| const encoded = Buffer.alloc(4 + nodeName.length + 4 + ip.length + 8 + 2); |
| let offset = 0; |
| 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', firstPort, 0), |
| node('iggy-node-2', secondPort, 1), |
| node('iggy-node-3', 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.GetClusterMetadata) { |
| socket.write( |
| replyFrame(Operation.NonReplicated, singleNodeMetadataBody(port)) |
| ); |
| 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; |
| }; |
| |
| describe('VSR client socket', () => { |
| 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) { |
| socket.write(replyFrame(Operation.Register, registerReplyBody())); |
| 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.Register, Operation.NonReplicated] |
| ); |
| 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) { |
| socket.write(replyFrame(Operation.Register, registerReplyBody())); |
| 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)}`); |
| |
| const operations = survivor.frames.map( |
| (frame) => frame.readUInt8(REQUEST_OFFSET.operation) |
| ); |
| assert.ok( |
| operations.includes(Operation.Register), |
| 'the client signed in again on the survivor' |
| ); |
| 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.Register, |
| Operation.NonReplicated |
| ]); |
| 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) { |
| socket.write(replyFrame(Operation.Register, registerReplyBody())); |
| 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(); |
| } |
| } |
| ); |
| |
| 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(); |
| } |
| }); |
| }); |
