This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch vsr_ts_sdk in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b6c781da9738af2f8ee073134c91b6c5d5bacc6c Author: spetz <[email protected]> AuthorDate: Tue Jul 28 15:27:12 2026 +0200 Update VSR --- foreign/node/README.md | 49 +++--------- foreign/node/src/bdd/raw.ts | 2 - foreign/node/src/client/client.socket.test.ts | 22 ++---- foreign/node/src/client/client.socket.ts | 35 +++------ foreign/node/src/client/client.ts | 76 +++++++++--------- foreign/node/src/client/client.type.ts | 5 +- foreign/node/src/e2e/tcp.raw.e2e.ts | 39 +--------- foreign/node/src/index.ts | 2 - foreign/node/src/wire/command-set.test.ts | 108 ++++---------------------- foreign/node/src/wire/command-set.ts | 27 +------ foreign/node/src/wire/vsr/index.ts | 5 +- foreign/node/src/wire/vsr/operation.test.ts | 61 +-------------- foreign/node/src/wire/vsr/operation.ts | 64 ++------------- foreign/node/src/wire/vsr/vsr.test.ts | 22 +----- 14 files changed, 106 insertions(+), 411 deletions(-) diff --git a/foreign/node/README.md b/foreign/node/README.md index ab5506d37..cca0426db 100644 --- a/foreign/node/README.md +++ b/foreign/node/README.md @@ -36,11 +36,7 @@ Classic framing remains the default. Select VSR explicitly when connecting to an Iggy VSR server: ```typescript -import { - BinaryRequestKind, - SimpleClient, - getRawClient, -} from "apache-iggy"; +import { SimpleClient, getRawClient } from "apache-iggy"; const config = { protocol: "vsr" as const, @@ -50,15 +46,15 @@ const config = { }; const client = new SimpleClient(getRawClient(config)); const response = await client.sendBinaryRequest( - BinaryRequestKind.NonReplicated, 60_000, Buffer.from("opaque mutation"), ); ``` -VSR is a runtime protocol choice in Node.js, not a build feature. Custom -non-replicated codes use `Operation::NonReplicated`; custom replicated codes -are rejected until a server-side replicated extension registry exists. +VSR is a runtime protocol choice in Node.js, not a build feature. Codes absent +from the SDK command table use `Operation::NonReplicated` and carry the command +code in the request header's reserved field. The server remains authoritative +for classifying or rejecting extension commands. The same npm package supports both framing modes. VSR currently supports TCP only and restricts `Client` to one pooled connection because authentication, @@ -77,37 +73,10 @@ new session. Transient not-committed responses retry the exact encoded request within one bounded deadline. A disconnected mutation is never replayed under a new session. -`sendBinaryRequest` has one intentionally breaking signature: - -```typescript -sendBinaryRequest( - kind: BinaryRequestKind, - code: number, - payload: Buffer, -): Promise<Buffer> -``` - -Migrate calls from: - -```typescript -await client.sendBinaryRequest(code, payload); -``` - -to: - -```typescript -await client.sendBinaryRequest( - BinaryRequestKind.NonReplicated, - code, - payload, -); -``` - -There is no compatibility overload or `sendBinaryRequestWithKind` method. -Known command tables remain authoritative under VSR: a conflicting declaration -is rejected, unknown non-replicated codes reach the server, and unknown -replicated codes fail locally until the extension registry exists. The kind is -not serialized by classic framing, so classic request bytes remain unchanged. +`sendBinaryRequest(code, payload)` has the same signature under classic and +VSR framing. Known replicated commands use their registered operation, while +unknown codes reach the server as non-replicated requests. Classic request +bytes remain unchanged. The client includes its npm package version and the binary protocol crate version in VSR registration. An incompatible server rejects registration with diff --git a/foreign/node/src/bdd/raw.ts b/foreign/node/src/bdd/raw.ts index 30e9d057d..f4daceb3f 100644 --- a/foreign/node/src/bdd/raw.ts +++ b/foreign/node/src/bdd/raw.ts @@ -17,7 +17,6 @@ import assert from 'node:assert/strict'; import { Then, When } from '@cucumber/cucumber'; -import { BINARY_REQUEST_KIND } from '../wire/command-set.js'; import type { TestWorld } from './world.js'; When( @@ -25,7 +24,6 @@ When( async function (this: TestWorld, code: number) { try { this.rawResponse = await this.client.sendBinaryRequest( - BINARY_REQUEST_KIND.NonReplicated, code, Buffer.alloc(0) ); diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts index 47b141049..d7292aa3f 100644 --- a/foreign/node/src/client/client.socket.test.ts +++ b/foreign/node/src/client/client.socket.test.ts @@ -22,7 +22,6 @@ import type { AddressInfo, Socket } from 'node:net'; import { createServer, type Server } from 'node:net'; import { describe, it } from 'node:test'; import { COMMAND_CODE } from '../wire/command.code.js'; -import { BINARY_REQUEST_KIND } from '../wire/command-set.js'; import { ResponseError } from '../wire/error.utils.js'; import { Command2, @@ -194,8 +193,7 @@ describe('VSR client socket', () => { try { const response = await client.sendCommand( 60_001, - Buffer.from('opaque'), - { rawKind: BINARY_REQUEST_KIND.NonReplicated } + Buffer.from('opaque') ); assert.equal(response.status, 0); @@ -242,8 +240,7 @@ describe('VSR client socket', () => { try { const response = await client.sendCommand( 60_002, - Buffer.from('retry-me'), - { rawKind: BINARY_REQUEST_KIND.NonReplicated } + Buffer.from('retry-me') ); assert.equal(attempts, 2); assert.deepEqual(response.data, Buffer.from('done')); @@ -273,8 +270,7 @@ describe('VSR client socket', () => { await assert.rejects( () => client.sendCommand( 60_003, - Buffer.alloc(0), - { rawKind: BINARY_REQUEST_KIND.NonReplicated } + Buffer.alloc(0) ), (error: unknown) => error instanceof ResponseError && error.errorCode === 40 @@ -300,8 +296,7 @@ describe('VSR client socket', () => { await assert.rejects( () => client.sendCommand( 60_004, - Buffer.alloc(0), - { rawKind: BINARY_REQUEST_KIND.NonReplicated } + Buffer.alloc(0) ) ); assert.equal(client.isAuthenticated, false); @@ -334,8 +329,7 @@ describe('VSR client socket', () => { try { const response = await client.sendCommand( 60_005, - Buffer.alloc(0), - { rawKind: BINARY_REQUEST_KIND.NonReplicated } + Buffer.alloc(0) ); assert.equal(response.status, 0); const followerOperations = follower.frames.map( @@ -369,15 +363,13 @@ describe('VSR client socket', () => { await assert.rejects( () => client.sendCommand( 60_006, - Buffer.alloc(0), - { rawKind: BINARY_REQUEST_KIND.NonReplicated } + Buffer.alloc(0) ) ); await assert.rejects( () => client.sendCommand( 60_006, - Buffer.alloc(0), - { rawKind: BINARY_REQUEST_KIND.NonReplicated } + Buffer.alloc(0) ) ); } finally { diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts index 17842673e..cdfdba573 100644 --- a/foreign/node/src/client/client.socket.ts +++ b/foreign/node/src/client/client.socket.ts @@ -23,7 +23,6 @@ import type { PasswordCredentials, Protocol, RawClient, SendCommandOptions, TokenCredentials } from '../client/client.type.js'; -import type { BinaryRequestKind } from '../wire/command-set.js'; import { handleResponse } from './client.utils.js'; import { ResponseError, responseError } from '../wire/error.utils.js'; import { debug } from './client.debug.js'; @@ -64,8 +63,6 @@ type Job = { payload: Buffer, /** Whether to parse the response */ handleResponse: boolean, - /** Execution model declared by a raw request */ - kind?: BinaryRequestKind, /** Promise resolve function */ resolve: (v: CommandResponse | PromiseLike<CommandResponse>) => void, /** Promise reject function */ @@ -108,10 +105,10 @@ export class CommandResponseStream extends EventEmitter { */ constructor(options: ClientConfig) { super(); - const normalized = normalizeClientConfig(options); - this.options = normalized; - this.protocol = normalized.protocol; - this.connection = new IggyConnection(normalized); + const normalizedConfig = normalizeClientConfig(options); + this.options = normalizedConfig; + this.protocol = normalizedConfig.protocol; + this.connection = new IggyConnection(normalizedConfig); this.busy = false; this.isAuthenticated = false; this._execQueue = []; @@ -146,7 +143,7 @@ export class CommandResponseStream extends EventEmitter { * * @param command - Command code to send * @param payload - Command payload buffer - * @param options - Response, queue, and raw-request options + * @param options - Response and queue options * @returns Promise resolving to the command response */ async sendCommand( @@ -158,8 +155,7 @@ export class CommandResponseStream extends EventEmitter { try { const { handleResponse = true, - last = true, - rawKind + last = true } = options; if (!this.connection.connected) @@ -182,7 +178,6 @@ export class CommandResponseStream extends EventEmitter { command, payload, handleResponse, - kind: rawKind, resolve, reject }; @@ -210,9 +205,9 @@ export class CommandResponseStream extends EventEmitter { while (this._execQueue.length > 0 && this.connection.socket.writable) { const next = this._execQueue.shift(); if (!next) break; - const { command, payload, handleResponse, kind, resolve, reject } = next; + const { command, payload, handleResponse, resolve, reject } = next; try { - resolve(await this._processNext(command, payload, handleResponse, kind)); + resolve(await this._processNext(command, payload, handleResponse)); } catch (err) { reject(err); } @@ -241,12 +236,11 @@ export class CommandResponseStream extends EventEmitter { _processNext( command: number, payload: Buffer, - handleResp = true, - kind?: BinaryRequestKind + handleResp = true ): Promise<CommandResponse> { if (this.options.protocol !== 'vsr') return this._processClassic(command, payload, handleResp); - return this._processVsr(command, payload, handleResp, kind); + return this._processVsr(command, payload, handleResp); } private async _processClassic( @@ -268,17 +262,12 @@ export class CommandResponseStream extends EventEmitter { private async _processVsr( command: number, payload: Buffer, - handleResp: boolean, - kind?: BinaryRequestKind + handleResp: boolean ): Promise<CommandResponse> { try { const prepared = prepareVsrCommand(command, payload); // A transient retry must preserve all request identity fields. - const frame = this.vsrSession.encode( - prepared.command, - prepared.payload, - kind - ); + const frame = this.vsrSession.encode(prepared.command, prepared.payload); const deadline = Date.now() + VSR_RESPONSE_TIMEOUT_MS; let parsed: CommandResponse; while (true) { diff --git a/foreign/node/src/client/client.ts b/foreign/node/src/client/client.ts index b67640a45..cfb7d2626 100644 --- a/foreign/node/src/client/client.ts +++ b/foreign/node/src/client/client.ts @@ -44,23 +44,25 @@ const createPoolFactory = (config: ClientConfig) => ({ * Automatically acquires and releases clients from the pool. * * @param config - Client configuration including pool size options - * @returns Client provider function with attached pool reference + * @returns Client provider and its connection pool */ -const poolClientProvider = (config: ClientConfig) => { - const min = config.poolSize?.min || 1; - const max = config.poolSize?.max || 4; - const pool = createPool(createPoolFactory(config), { min, max }); - const poolClientProvider = async () => { - const c = await pool.acquire(); +const createPooledClientProvider = (config: ClientConfig) => { + const minPoolSize = config.poolSize?.min || 1; + const maxPoolSize = config.poolSize?.max || 4; + const pool = createPool(createPoolFactory(config), { + min: minPoolSize, + max: maxPoolSize + }); + const clientProvider = async () => { + const client = await pool.acquire(); debug('client acquired from pool. pool size is', pool.size); - c.once('finishQueue', () => { - pool.release(c) + client.once('finishQueue', () => { + pool.release(client) debug('client released to pool. pool size is', pool.size); }); - return c; + return client; } - poolClientProvider._pool = pool; - return poolClientProvider; + return { clientProvider, pool }; }; @@ -77,15 +79,16 @@ export class Client extends CommandAPI { /** * Creates a new pooled client. * - * @param config - Client configuration - */ + * @param config - Client configuration + */ constructor(config: ClientConfig) { - const normalized = normalizeClientConfig(config); - const pcp = poolClientProvider(normalized); - super(pcp); - this._config = normalized; - this._pool = pcp._pool; - }; + const normalizedConfig = normalizeClientConfig(config); + const { clientProvider, pool } = + createPooledClientProvider(normalizedConfig); + super(clientProvider); + this._config = normalizedConfig; + this._pool = pool; + } /** * Destroys the client and drains all connections from the pool. @@ -104,10 +107,10 @@ export class Client extends CommandAPI { * @param config - Client configuration * @returns Client provider function that always returns the same client */ -const singleClientProvider = (config: ClientConfig) => { - const c = getRawClient(config); - return async function singleClientProvider() { - return c; +const createSingleClientProvider = (config: ClientConfig) => { + const client = getRawClient(config); + return async function clientProvider() { + return client; } } @@ -122,22 +125,21 @@ export class SingleClient extends CommandAPI { /** * Creates a new single-connection client. * - * @param config - Client configuration - */ + * @param config - Client configuration + */ constructor(config: ClientConfig) { - const normalized = normalizeClientConfig(config); - super(singleClientProvider(normalized)); - this._config = normalized; + super(createSingleClientProvider(config)); + this._config = config; } /** * Destroys the client connection. */ async destroy() { - const s = await this.clientProvider(); - s.destroy(); + const client = await this.clientProvider(); + client.destroy(); } -}; +} /** @@ -158,11 +160,11 @@ export class SimpleClient extends CommandAPI { * Destroys the underlying client connection. */ async destroy() { - const s = await this.clientProvider(); - s.destroy(); + const client = await this.clientProvider(); + client.destroy(); } -}; +} /** * Creates a SimpleClient with the given configuration. @@ -172,6 +174,6 @@ export class SimpleClient extends CommandAPI { * @returns SimpleClient instance */ export const getClient = async (config: ClientConfig) => { - const cli = getRawClient(normalizeClientConfig(config)); - return new SimpleClient(cli); + const client = getRawClient(config); + return new SimpleClient(client); }; diff --git a/foreign/node/src/client/client.type.ts b/foreign/node/src/client/client.type.ts index c3da1e82e..c634e31d5 100644 --- a/foreign/node/src/client/client.type.ts +++ b/foreign/node/src/client/client.type.ts @@ -19,7 +19,6 @@ import type { Readable } from 'stream'; import { type TcpSocketConnectOpts } from 'node:net'; import { type ConnectionOptions } from 'node:tls'; -import type { BinaryRequestKind } from '../wire/command-set.js'; /** * TCP socket connection options. @@ -49,9 +48,7 @@ export type SendCommandOptions = { /** Whether the response uses the standard command response decoder */ handleResponse?: boolean, /** Whether to append rather than prepend the command to the queue */ - last?: boolean, - /** Execution model supplied by the public raw-request API */ - rawKind?: BinaryRequestKind + last?: boolean }; /** diff --git a/foreign/node/src/e2e/tcp.raw.e2e.ts b/foreign/node/src/e2e/tcp.raw.e2e.ts index 309117eb7..e1fcf2737 100644 --- a/foreign/node/src/e2e/tcp.raw.e2e.ts +++ b/foreign/node/src/e2e/tcp.raw.e2e.ts @@ -19,7 +19,6 @@ import { after, describe, it } from 'node:test'; import assert from 'node:assert/strict'; import { getTestClient } from './test-client.utils.js'; -import { BINARY_REQUEST_KIND } from '../wire/command-set.js'; import { COMMAND_CODE } from '../wire/command.code.js'; describe('e2e -> raw', async () => { @@ -30,7 +29,6 @@ describe('e2e -> raw', async () => { it('e2e -> raw::ping', async () => { const response = await c.sendBinaryRequest( - BINARY_REQUEST_KIND.NonReplicated, COMMAND_CODE.Ping, Buffer.alloc(0) ); @@ -39,43 +37,25 @@ describe('e2e -> raw', async () => { it('e2e -> raw::getStats', async () => { const response = await c.sendBinaryRequest( - BINARY_REQUEST_KIND.NonReplicated, COMMAND_CODE.GetStats, Buffer.alloc(0) ); assert.ok(response.length > 0); }); - it('e2e -> raw::getStatsUsesTheProtocolKindRules', async () => { - const request = () => c.sendBinaryRequest( - BINARY_REQUEST_KIND.Replicated, - COMMAND_CODE.GetStats, - Buffer.alloc(0) - ); - if (process.env.IGGY_TEST_PROTOCOL === 'vsr') { - await assert.rejects(request); - return; - } - - // Classic framing has no operation field, so the read still succeeds. - assert.ok((await request()).length > 0); - }); - it('e2e -> raw::sessionControlCodeRejectedClientSide', async () => { - for (const kind of Object.values(BINARY_REQUEST_KIND)) - await assert.rejects( - () => c.sendBinaryRequest(kind, COMMAND_CODE.LoginUser, Buffer.alloc(0)) - ); + await assert.rejects( + () => c.sendBinaryRequest(COMMAND_CODE.LoginUser, Buffer.alloc(0)) + ); }); it('e2e -> raw::vendorCodeRejectedByServer', async () => { await assert.rejects( - () => c.sendBinaryRequest(BINARY_REQUEST_KIND.NonReplicated, VENDOR_CODE, Buffer.alloc(0)) + () => c.sendBinaryRequest(VENDOR_CODE, Buffer.alloc(0)) ); // The rejection is request-level, so the connection stays usable. const response = await c.sendBinaryRequest( - BINARY_REQUEST_KIND.NonReplicated, COMMAND_CODE.Ping, Buffer.alloc(0) ); @@ -86,7 +66,6 @@ describe('e2e -> raw', async () => { for (let attempt = 0; attempt < 3; attempt += 1) await assert.rejects( () => c.sendBinaryRequest( - BINARY_REQUEST_KIND.NonReplicated, VENDOR_CODE + attempt, Buffer.from([attempt]) ) @@ -98,16 +77,6 @@ describe('e2e -> raw', async () => { assert.equal(await c.stream.delete({ streamId: streamName }), true); }); - it('e2e -> raw::unknownReplicatedCodeRejected', async () => { - await assert.rejects( - () => c.sendBinaryRequest( - BINARY_REQUEST_KIND.Replicated, - VENDOR_CODE, - Buffer.alloc(0) - ) - ); - }); - after(() => { c.destroy(); }); diff --git a/foreign/node/src/index.ts b/foreign/node/src/index.ts index 299425aeb..e684b401a 100644 --- a/foreign/node/src/index.ts +++ b/foreign/node/src/index.ts @@ -23,8 +23,6 @@ export { Partitioning, HeaderValue, HeaderKeyFactory, - BINARY_REQUEST_KIND, - BinaryRequestKind, } from "./wire/index.js"; export * from "./client/index.js"; diff --git a/foreign/node/src/wire/command-set.test.ts b/foreign/node/src/wire/command-set.test.ts index ac4fe76d4..3c1cf6f75 100644 --- a/foreign/node/src/wire/command-set.test.ts +++ b/foreign/node/src/wire/command-set.test.ts @@ -21,11 +21,6 @@ import assert from 'node:assert/strict'; import { SimpleClient } from '../client/client.js'; import type { RawClient } from '../client/client.type.js'; import { COMMAND_CODE } from './command.code.js'; -import { - BINARY_REQUEST_KIND, - BinaryRequestKind, - type BinaryRequestKind as BinaryRequestKindType -} from './command-set.js'; const mockRawClient = (): RawClient => ({ protocol: 'classic', @@ -45,80 +40,22 @@ const mockRawClient = (): RawClient => ({ }); describe('CommandAPI.sendBinaryRequest', () => { - it('exports exactly the two protocol request kinds', () => { - assert.deepEqual(BinaryRequestKind, { - NonReplicated: 'non_replicated', - Replicated: 'replicated' - }); - }); - - it('requires the request kind in TypeScript', () => { - const client = new SimpleClient(mockRawClient()); - if (false) { - // @ts-expect-error the breaking API requires a replication kind - void client.sendBinaryRequest(COMMAND_CODE.Ping, Buffer.alloc(0)); - } - assert.equal('sendBinaryRequestWithKind' in client, false); - }); - describe('session-control guard', () => { - - Object.values(BINARY_REQUEST_KIND).forEach((kind) => { - [ - COMMAND_CODE.LoginUser, - COMMAND_CODE.LogoutUser, - COMMAND_CODE.LoginRegister, - COMMAND_CODE.LoginWithAccessToken, - COMMAND_CODE.LoginRegisterWithAccessToken, - ].forEach((code) => { - it(`rejects ${kind} code ${code} before reaching the client provider`, async () => { - const client = new SimpleClient(mockRawClient()); - await assert.rejects( - () => client.sendBinaryRequest(kind, code, Buffer.alloc(0)), - /code: 3, message: Invalid command/ - ); - }); + [ + COMMAND_CODE.LoginUser, + COMMAND_CODE.LogoutUser, + COMMAND_CODE.LoginRegister, + COMMAND_CODE.LoginWithAccessToken, + COMMAND_CODE.LoginRegisterWithAccessToken, + ].forEach((code) => { + it(`rejects code ${code} before reaching the client provider`, async () => { + const client = new SimpleClient(mockRawClient()); + await assert.rejects( + () => client.sendBinaryRequest(code, Buffer.alloc(0)), + /code: 3, message: Invalid command/ + ); }); }); - - }); - - it('rejects every invalid request kind before reaching the raw client', async () => { - const client = new SimpleClient(mockRawClient()); - const invalidKinds = [ - undefined, - null, - 'auto', - 0, - {}, - ]; - for (const invalidKind of invalidKinds) - await assert.rejects( - () => client.sendBinaryRequest( - invalidKind as BinaryRequestKindType, - COMMAND_CODE.Ping, - Buffer.alloc(0) - ), - /code: 3, message: Invalid command/ - ); - }); - - it('encodes both kinds identically because classic framing has no operation field', async () => { - const customCode = 60_001; - const payload = Buffer.from([0xAA, 0xBB, 0xCC]); - const frames: { code: number, payload: Buffer }[] = []; - const raw = mockRawClient(); - raw.sendCommand = async (code, sentPayload) => { - frames.push({ code, payload: Buffer.from(sentPayload) }); - return { status: 0, length: 1, data: Buffer.alloc(0) }; - }; - const client = new SimpleClient(raw); - - for (const kind of Object.values(BINARY_REQUEST_KIND)) - await client.sendBinaryRequest(kind, customCode, payload); - - assert.equal(frames.length, 2); - assert.deepEqual(frames[0], frames[1]); }); it('forwards a custom code and opaque payload to sendCommand', async () => { @@ -129,10 +66,7 @@ describe('CommandAPI.sendBinaryRequest', () => { raw.sendCommand = async (code, sentPayload, options) => { assert.equal(code, customCode); assert.deepEqual(sentPayload, payload); - assert.equal( - options?.rawKind, - BINARY_REQUEST_KIND.NonReplicated - ); + assert.equal(options, undefined); return { status: 0, length: expectedResponse.length, @@ -140,11 +74,7 @@ describe('CommandAPI.sendBinaryRequest', () => { }; }; const client = new SimpleClient(raw); - const response = await client.sendBinaryRequest( - BINARY_REQUEST_KIND.NonReplicated, - customCode, - payload - ); + const response = await client.sendBinaryRequest(customCode, payload); assert.deepEqual(response, expectedResponse); }); @@ -157,11 +87,7 @@ describe('CommandAPI.sendBinaryRequest', () => { return { status: 0, length: 1, data: Buffer.alloc(0) }; }; - const request = new SimpleClient(raw).sendBinaryRequest( - BINARY_REQUEST_KIND.NonReplicated, - 60_001, - payload - ); + const request = new SimpleClient(raw).sendBinaryRequest(60_001, payload); payload.fill(0); await request; @@ -176,12 +102,10 @@ describe('CommandAPI.sendBinaryRequest', () => { }); const response = await new SimpleClient(raw).sendBinaryRequest( - BINARY_REQUEST_KIND.NonReplicated, COMMAND_CODE.Ping, Buffer.alloc(0) ); assert.deepEqual(response, Buffer.alloc(0)); }); - }); diff --git a/foreign/node/src/wire/command-set.ts b/foreign/node/src/wire/command-set.ts index 5265f1184..ad8a2732b 100644 --- a/foreign/node/src/wire/command-set.ts +++ b/foreign/node/src/wire/command-set.ts @@ -213,26 +213,6 @@ const SESSION_CONTROL_CODES = new Set([ const INVALID_COMMAND_ERROR_CODE = 3; -/** - * How a raw binary request executes on the server. - * - * Classic framing carries no operation field, while VSR framing uses this - * declaration to route unknown extension codes. - */ -export const BINARY_REQUEST_KIND = { - NonReplicated: 'non_replicated', - Replicated: 'replicated' -} as const; - -/** PascalCase alias for enum-style call sites. */ -export const BinaryRequestKind = BINARY_REQUEST_KIND; - -export type BinaryRequestKind = - typeof BINARY_REQUEST_KIND[keyof typeof BINARY_REQUEST_KIND]; - -const BINARY_REQUEST_KINDS: ReadonlySet<string> = - new Set(Object.values(BINARY_REQUEST_KIND)); - export abstract class AbstractAPI { clientProvider: ClientProvider; @@ -264,24 +244,21 @@ export abstract class CommandAPI extends AbstractAPI { * Sends a command code with a payload and returns the raw response payload. * Session-control codes are rejected with an invalid-command error. * - * @param kind - How the command executes * @param code - Command code to send * @param payload - Raw command payload * @returns Raw response payload */ async sendBinaryRequest( - kind: BinaryRequestKind, code: number, payload: Buffer ): Promise<Buffer> { - if (!BINARY_REQUEST_KINDS.has(kind) || SESSION_CONTROL_CODES.has(code)) + if (SESSION_CONTROL_CODES.has(code)) throw responseError(code, INVALID_COMMAND_ERROR_CODE); const requestPayload = Buffer.from(payload); const response = await (await this.clientProvider()).sendCommand( code, - requestPayload, - { rawKind: kind } + requestPayload ); return response.length <= 1 ? Buffer.alloc(0) : response.data; } diff --git a/foreign/node/src/wire/vsr/index.ts b/foreign/node/src/wire/vsr/index.ts index 390097855..698d46dfa 100644 --- a/foreign/node/src/wire/vsr/index.ts +++ b/foreign/node/src/wire/vsr/index.ts @@ -18,7 +18,6 @@ import { createRequire } from 'node:module'; import type { CommandResponse } from '../../client/client.type.js'; -import type { BinaryRequestKind } from '../command-set.js'; import { COMMAND_CODE } from '../command.code.js'; import { responseError } from '../error.utils.js'; import { HEADER_SIZE, encodeRequestHeader } from './header.js'; @@ -59,10 +58,10 @@ export class VsrSession { this.state.bind(session); } - encode(command: number, payload: Buffer, kind?: BinaryRequestKind): Buffer { + encode(command: number, payload: Buffer): Buffer { const operation = registerCommand(command) ? Operation.Register - : operationForCode(command, kind); + : operationForCode(command); const namespace = namespaceForRequest(command, payload, operation); const size = HEADER_SIZE + payload.length; if (!Number.isSafeInteger(size) || size > MAX_U32) diff --git a/foreign/node/src/wire/vsr/operation.test.ts b/foreign/node/src/wire/vsr/operation.test.ts index 2189bbe80..260237344 100644 --- a/foreign/node/src/wire/vsr/operation.test.ts +++ b/foreign/node/src/wire/vsr/operation.test.ts @@ -17,9 +17,7 @@ import assert from 'node:assert/strict'; import { describe, it } from 'node:test'; -import { BINARY_REQUEST_KIND } from '../command-set.js'; import { COMMAND_CODE } from '../command.code.js'; -import { ResponseError } from '../error.utils.js'; import { isInternal, isKnownOperation, @@ -62,13 +60,8 @@ const replicated = new Map<number, number>([ describe('VSR operation classification', () => { it('matches every replicated command operation', () => { - for (const [code, operation] of replicated) { + for (const [code, operation] of replicated) assert.equal(operationForCode(code), operation); - assert.equal( - operationForCode(code, BINARY_REQUEST_KIND.Replicated), - operation - ); - } }); it('classifies every other known command as non-replicated', () => { @@ -79,59 +72,13 @@ describe('VSR operation classification', () => { } }); - it('keeps standard command tables authoritative', () => { - assert.throws( - () => operationForCode( - COMMAND_CODE.GetStats, - BINARY_REQUEST_KIND.Replicated - ), - (error: unknown) => - error instanceof ResponseError && error.errorCode === 3 - ); - assert.throws( - () => operationForCode( - COMMAND_CODE.CreateStream, - BINARY_REQUEST_KIND.NonReplicated - ), - (error: unknown) => - error instanceof ResponseError && error.errorCode === 3 - ); - }); - - it('forwards only unknown non-replicated extensions', () => { - assert.equal( - operationForCode(60_001, BINARY_REQUEST_KIND.NonReplicated), - Operation.NonReplicated - ); - assert.throws( - () => operationForCode(60_001, BINARY_REQUEST_KIND.Replicated), - (error: unknown) => - error instanceof ResponseError && error.errorCode === 5 - ); - assert.throws( - () => operationForCode(60_001), - (error: unknown) => - error instanceof ResponseError && error.errorCode === 3 - ); + it('forwards unknown extension codes as non-replicated', () => { + assert.equal(operationForCode(60_001), Operation.NonReplicated); + assert.equal(operationForCode(1_000_104), Operation.NonReplicated); }); it('routes logout through its consensus operation', () => { assert.equal(operationForCode(COMMAND_CODE.LogoutUser), Operation.Logout); - assert.equal( - operationForCode( - COMMAND_CODE.LogoutUser, - BINARY_REQUEST_KIND.Replicated - ), - Operation.Logout - ); - assert.throws( - () => operationForCode( - COMMAND_CODE.LogoutUser, - BINARY_REQUEST_KIND.NonReplicated - ), - (error: unknown) => - error instanceof ResponseError && error.errorCode === 3 - ); }); it('pins operation class predicates', () => { diff --git a/foreign/node/src/wire/vsr/operation.ts b/foreign/node/src/wire/vsr/operation.ts index 63f74bdd4..4765e5358 100644 --- a/foreign/node/src/wire/vsr/operation.ts +++ b/foreign/node/src/wire/vsr/operation.ts @@ -23,8 +23,6 @@ */ import { COMMAND_CODE } from '../command.code.js'; -import { BINARY_REQUEST_KIND, type BinaryRequestKind } from '../command-set.js'; -import { responseError } from '../error.utils.js'; /** `Operation` discriminants a client sends or receives. */ export const Operation = { @@ -70,15 +68,10 @@ const INTERNAL_START = 64; const METADATA_START = 128; const PARTITION_START = 160; -/** `IggyError::InvalidCommand` numeric code. */ -export const ERROR_CODE_INVALID_COMMAND = 3; -/** `IggyError::FeatureUnavailable` numeric code. */ -export const ERROR_CODE_FEATURE_UNAVAILABLE = 5; - /** * Replicated command code to `Operation` mapping, the client half of the - * dispatch table. Codes absent here are non-replicated when the classic - * command set knows them, unknown otherwise. + * dispatch table. Codes absent here are sent as non-replicated so the server + * remains authoritative for extension commands unknown to this SDK build. */ const REPLICATED_OPERATION: ReadonlyMap<number, number> = new Map([ [COMMAND_CODE.CreateUser, Operation.CreateUser], @@ -110,10 +103,6 @@ const REPLICATED_OPERATION: ReadonlyMap<number, number> = new Map([ [COMMAND_CODE.LeaveGroup, Operation.LeaveConsumerGroup] ]); -/** Classic command codes known to the SDK. */ -const KNOWN_COMMAND_CODES: ReadonlySet<number> = - new Set(Object.values(COMMAND_CODE)); - const KNOWN_OPERATIONS: ReadonlySet<number> = new Set(Object.values(Operation)); @@ -155,52 +144,15 @@ export const isResultFramed = (operation: number): boolean => operation === Operation.DeleteConsumerOffset2; /** - * Picks the header operation for a command code, honoring the caller's - * declaration only where the protocol tables have nothing to say. Standard - * codes resolve first so a caller cannot redirect a shipped command into the - * other execution model. Port of `operation_for_code` in - * `core/sdk/src/vsr.rs`, extended with the raw declaration rules for - * unknown extension codes. - * - * @throws Error with the invalid-command code for a conflicting declaration - * or an undeclared unknown code, and feature-unavailable for an unknown - * code declared replicated. + * Picks the header operation for a command code. Unknown extension codes are + * sent as non-replicated with their command code in the reserved header field, + * allowing the server to classify or reject them. */ -export const operationForCode = ( - code: number, - kind?: BinaryRequestKind -): number => { +export const operationForCode = (code: number): number => { // The dispatch table files logout as non-replicated, but VSR routes it // through its own consensus operation; the raw API blocks code 39 before // classification, so only the typed logout path reaches this branch. if (code === COMMAND_CODE.LogoutUser) - return acceptDeclaration(code, Operation.Logout, kind); - - const replicated = REPLICATED_OPERATION.get(code); - if (replicated !== undefined) - return acceptDeclaration(code, replicated, kind); - - if (KNOWN_COMMAND_CODES.has(code)) - return acceptDeclaration(code, Operation.NonReplicated, kind); - - if (kind === BINARY_REQUEST_KIND.NonReplicated) - return Operation.NonReplicated; - if (kind === BINARY_REQUEST_KIND.Replicated) - // A replicated extension needs a server-side handler registry that does - // not exist yet; failing closed beats a half-supported frame. - throw responseError(code, ERROR_CODE_FEATURE_UNAVAILABLE); - throw responseError(code, ERROR_CODE_INVALID_COMMAND); -}; - -const acceptDeclaration = ( - code: number, - operation: number, - kind?: BinaryRequestKind -): number => { - if (kind === undefined) return operation; - const standard = operation === Operation.NonReplicated ? - BINARY_REQUEST_KIND.NonReplicated : - BINARY_REQUEST_KIND.Replicated; - if (kind === standard) return operation; - throw responseError(code, ERROR_CODE_INVALID_COMMAND); + return Operation.Logout; + return REPLICATED_OPERATION.get(code) ?? Operation.NonReplicated; }; diff --git a/foreign/node/src/wire/vsr/vsr.test.ts b/foreign/node/src/wire/vsr/vsr.test.ts index 2df7b6326..d8acb329b 100644 --- a/foreign/node/src/wire/vsr/vsr.test.ts +++ b/foreign/node/src/wire/vsr/vsr.test.ts @@ -18,7 +18,6 @@ import { describe, it } from 'node:test'; import assert from 'node:assert/strict'; -import { BINARY_REQUEST_KIND } from '../command-set.js'; import { ResponseError } from '../error.utils.js'; import { HEADER_SIZE, REQUEST_OFFSET } from './header.js'; import { prepareVsrCommand, VsrSession } from './index.js'; @@ -29,29 +28,13 @@ describe('VSR custom request framing', () => { it('encodes a custom non-replicated code and opaque payload', () => { const session = new VsrSession(); const payload = Buffer.from([0xAA, 0xBB, 0xCC]); - const frame = session.encode( - 60_000, - payload, - BINARY_REQUEST_KIND.NonReplicated - ); + const frame = session.encode(60_000, payload); assert.equal(frame.readUInt8(REQUEST_OFFSET.operation), Operation.NonReplicated); assert.equal(frame.readUInt32LE(REQUEST_OFFSET.reserved), 60_000); assert.deepEqual(frame.subarray(256), payload); }); - it('rejects a custom replicated code without a server registry', () => { - const session = new VsrSession(); - assert.throws( - () => session.encode( - 60_000, - Buffer.alloc(0), - BINARY_REQUEST_KIND.Replicated - ), - /Feature is unavailable/ - ); - }); - it('does not consume a request ID when local routing fails', () => { const session = new VsrSession(7n); session.bind(42n); @@ -95,8 +78,7 @@ describe('VSR custom request framing', () => { session.bind(42n); const custom = session.encode( 60_001, - Buffer.alloc(0), - BINARY_REQUEST_KIND.NonReplicated + Buffer.alloc(0) ); assert.equal(custom.readBigUInt64LE(REQUEST_OFFSET.request), 1n);
