hubcio commented on code in PR #3763:
URL: https://github.com/apache/iggy/pull/3763#discussion_r3680795422
##########
foreign/node/src/client/client.socket.ts:
##########
@@ -167,25 +238,177 @@ export class CommandResponseStream extends EventEmitter {
payload: Buffer,
handleResp = true
): Promise<CommandResponse> {
- const lastWrite = this.connection.writeCommand(command, payload);
- debug('==> writeCommand', lastWrite);
- return new Promise((resolve, reject) => {
- if (!lastWrite)
- return reject(new Error('failed to write to socket'));
- const errCb = (err: unknown) => reject(err);
- this.connection.once('error', errCb);
- this.connection.once('response', (resp) => {
- this.connection.removeListener('error', errCb);
- if (!handleResp) return resolve(resp);
- const r = handleResponse(resp);
- if (r.status !== 0) {
- return reject(responseError(command, r.status));
+ if (this.options.protocol !== 'vsr')
+ return this._processClassic(command, payload, handleResp);
+ return this._processVsr(command, payload, handleResp);
+ }
+
+ private async _processClassic(
+ command: number,
+ payload: Buffer,
+ handleResp: boolean
+ ): Promise<CommandResponse> {
+ const response = await this._exchange(
+ () => this.connection.writeCommand(command, payload)
+ );
+ if (!handleResp)
+ return response as unknown as CommandResponse;
+ const parsed = handleResponse(response);
+ if (parsed.status !== 0)
+ throw responseError(command, parsed.status);
+ return parsed;
+ }
+
+ private async _processVsr(
+ command: number,
+ payload: Buffer,
+ 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);
+ const deadline = Date.now() + VSR_RESPONSE_TIMEOUT_MS;
+ let parsed: CommandResponse;
+ while (true) {
+ const remaining = deadline - Date.now();
+ if (remaining <= 0)
+ throw new Error(
+ `timed out after ${VSR_RESPONSE_TIMEOUT_MS} ms ` +
+ 'waiting for VSR response'
+ );
+ const response = await this._exchange(
+ () => this.connection.writeFrame(frame),
+ remaining
+ );
+ if (!handleResp)
+ return response as unknown as CommandResponse;
+ try {
+ parsed = decodeVsrResponse(response);
+ break;
+ } catch (error) {
+ if (!(error instanceof ResponseError) ||
+ !isTransientVsrError(error.errorCode))
+ throw error;
+ const retryDelay = Math.min(
+ VSR_RETRY_INTERVAL_MS,
+ Math.max(0, deadline - Date.now())
+ );
+ if (retryDelay === 0)
+ throw error;
+ await delay(retryDelay);
}
- return resolve(r);
- });
+ }
+
+ if (prepared.command === COMMAND_CODE.LoginRegister ||
+ prepared.command === COMMAND_CODE.LoginRegisterWithAccessToken) {
+ this.vsrSession.bind(readRegisteredSession(parsed));
+ this.isAuthenticated = true;
+ this.userId = parsed.data.readUInt32LE(0);
+ }
+ if (prepared.command === COMMAND_CODE.LogoutUser) {
+ this.isAuthenticated = false;
+ this.userId = undefined;
+ this.vsrSession.reset();
+ this.emit('sessionReset');
+ }
+ return parsed;
+ } catch (error) {
+ if (error instanceof VsrEvictionError) {
+ this.isAuthenticated = false;
+ this.userId = undefined;
+ this.vsrSession.reset();
+ this.emit('sessionReset');
+ } else if (!(error instanceof ResponseError)) {
+ // A local failure after encoding may have consumed a request ID
+ // without a server verdict; keeping the session would leave a
+ // request-ID gap the primary silently drops. Register afresh instead
+ // of replaying an ambiguous request under the old session.
+ this.isAuthenticated = false;
+ this.userId = undefined;
+ this.vsrSession.reset();
+ this.emit('sessionReset');
+ }
+ if (error instanceof ResponseError)
+ throw responseError(command, error.errorCode);
+ throw error;
+ }
+ }
+
+ private _exchange(write: () => void, timeout?: number): Promise<Buffer> {
Review Comment:
`_exchange` is first-frame-wins - `once('response')` with no request/reply
correlation, so whatever frame shows up next resolves whatever exchange is
currently open.
the server's `evict_stale_client` (`core/server-ng/src/dispatch.rs:1588`)
sends an unsolicited `Eviction` frame and never closes the socket. if one lands
while a command is in flight it eats that exchange, and the genuine reply lands
on the *next* one. from then on every command resolves the previous command's
body as a success. silently, no error, forever.
reproduced against a real server: command B resolved with the reply for
command A, and it chained through five commands.
it needs operator-enabled `[heartbeat]` (ships `enabled = false` in
`core/server-ng/config.toml:359`) plus group membership plus a client without
`heartbeatInterval` - which is the client default, and the README doesn't
mention it. that's why the suite never gets near it.
cheapest fix is routing `Eviction` out of band in `_onData` at connection
scope instead of through the exchange slot. that also gives idle eviction the
event it currently lacks. `abort()` in the eviction branch works too. real
request/operation correlation only helps replicated ops - `.client` is
caller-dependent and NonReplicated/partition ops share `currentRequestId()` -
so correlation alone can't cover reads.
##########
foreign/node/src/client/client.config.ts:
##########
@@ -0,0 +1,56 @@
+// 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 type { ClientConfig, Protocol } from './client.type.js';
+
+export const DEFAULT_MAX_RESPONSE_FRAME_SIZE = 64 * 1024 * 1024;
Review Comment:
this cap plus destroy-on-oversize now governs the classic path too - master
had no cap and never destroyed the socket. the cap is right (uncapped is an OOM
vector), but it's a behavior change for existing classic users shipped under a
patch bump with no BREAKING note, and the README explains it under `### VSR
framing` where a classic user won't look. move that paragraph and call it out
in the PR body.
##########
foreign/node/src/client/client.frame.ts:
##########
@@ -0,0 +1,77 @@
+// 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 type { Protocol } from './client.type.js';
+import {
+ HEADER_SIZE as VSR_HEADER_SIZE,
+ readSize as readVsrSize
+} from '../wire/vsr/header.js';
+
+const CLASSIC_HEADER_SIZE = 8;
+
+export class ProtocolFrameError extends Error {
+ constructor(message: string) {
+ super(message);
+ this.name = 'ProtocolFrameError';
+ }
+}
+
+export type ExtractedFrames = {
+ frames: Buffer[],
+ remainder: Buffer
+};
+
+export const extractResponseFrames = (
+ protocol: Protocol,
+ buffer: Buffer,
+ maximumFrameSize: number
+): ExtractedFrames => {
+ const headerSize =
+ protocol === 'vsr' ? VSR_HEADER_SIZE : CLASSIC_HEADER_SIZE;
+ const frames: Buffer[] = [];
+ let offset = 0;
+
+ while (buffer.length - offset >= headerSize) {
+ const available = buffer.length - offset;
+ const declaredSize = protocol === 'vsr'
+ ? readVsrSize(buffer.subarray(offset, offset + headerSize))
+ : CLASSIC_HEADER_SIZE + buffer.readUInt32LE(offset + 4);
+
+ if (declaredSize < headerSize)
+ throw new ProtocolFrameError(
+ `declared ${protocol} frame size ${declaredSize} is below header size`
+ );
+ if (!Number.isSafeInteger(declaredSize) ||
+ declaredSize > maximumFrameSize)
+ throw new ProtocolFrameError(
+ `declared ${protocol} frame size ${declaredSize} exceeds ` +
+ `the ${maximumFrameSize} byte limit`
+ );
+ if (available < declaredSize)
+ break;
+
+ frames.push(buffer.subarray(offset, offset + declaredSize));
+ offset += declaredSize;
+ }
+
+ return {
+ frames,
+ remainder: offset === buffer.length
+ ? Buffer.alloc(0)
+ : Buffer.from(buffer.subarray(offset))
Review Comment:
`Buffer.from(...)` copies the entire pending remainder on every socket read,
so a frame arriving in many chunks gets recopied each time.
`buffer.subarray(offset)` is the same value with no copy - safe here since the
source buffer isn't reused after extraction. measured roughly X.Xx slower than
master on a large blocked read.
the quadratic `Buffer.concat` accumulator behind it
(`client.connection.ts:318-329`) is pre-existing and master is strictly worse
(uncapped), so that can be a follow-up - a chunk list kills it outright. worth
noting the stall also feeds the eviction desync above: a merely busy client can
cross the stale window.
##########
foreign/node/README.md:
##########
@@ -30,6 +30,58 @@ npm i --save apache-iggy
## basic usage
+### VSR framing
+
+Classic framing remains the default. Select VSR explicitly when connecting to
+an Iggy VSR server:
+
+```typescript
+import { SimpleClient, getRawClient } from "apache-iggy";
+
+const config = {
+ protocol: "vsr" as const,
+ transport: "TCP" as const,
+ options: { host: "127.0.0.1", port: 8090 },
+ credentials: { username: "iggy", password: "iggy" },
+};
+const client = new SimpleClient(getRawClient(config));
+const response = await client.sendBinaryRequest(
Review Comment:
this headline example always throws.
`core/server-ng/src/responses.rs:585-589` fail-closes unknown codes, and the
PR's own e2e asserts exactly that rejection. leading the vsr section with a
snippet that can't run is rough - use a typed example here and keep the raw one
below as an explicit "this gets rejected" demo.
##########
foreign/node/src/client/client.socket.ts:
##########
@@ -167,25 +238,177 @@ export class CommandResponseStream extends EventEmitter {
payload: Buffer,
handleResp = true
): Promise<CommandResponse> {
- const lastWrite = this.connection.writeCommand(command, payload);
- debug('==> writeCommand', lastWrite);
- return new Promise((resolve, reject) => {
- if (!lastWrite)
- return reject(new Error('failed to write to socket'));
- const errCb = (err: unknown) => reject(err);
- this.connection.once('error', errCb);
- this.connection.once('response', (resp) => {
- this.connection.removeListener('error', errCb);
- if (!handleResp) return resolve(resp);
- const r = handleResponse(resp);
- if (r.status !== 0) {
- return reject(responseError(command, r.status));
+ if (this.options.protocol !== 'vsr')
+ return this._processClassic(command, payload, handleResp);
+ return this._processVsr(command, payload, handleResp);
+ }
+
+ private async _processClassic(
+ command: number,
+ payload: Buffer,
+ handleResp: boolean
+ ): Promise<CommandResponse> {
+ const response = await this._exchange(
+ () => this.connection.writeCommand(command, payload)
+ );
+ if (!handleResp)
+ return response as unknown as CommandResponse;
+ const parsed = handleResponse(response);
+ if (parsed.status !== 0)
+ throw responseError(command, parsed.status);
+ return parsed;
+ }
+
+ private async _processVsr(
+ command: number,
+ payload: Buffer,
+ 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);
+ const deadline = Date.now() + VSR_RESPONSE_TIMEOUT_MS;
+ let parsed: CommandResponse;
+ while (true) {
+ const remaining = deadline - Date.now();
+ if (remaining <= 0)
+ throw new Error(
+ `timed out after ${VSR_RESPONSE_TIMEOUT_MS} ms ` +
+ 'waiting for VSR response'
+ );
+ const response = await this._exchange(
+ () => this.connection.writeFrame(frame),
+ remaining
+ );
+ if (!handleResp)
+ return response as unknown as CommandResponse;
+ try {
+ parsed = decodeVsrResponse(response);
+ break;
+ } catch (error) {
+ if (!(error instanceof ResponseError) ||
+ !isTransientVsrError(error.errorCode))
+ throw error;
+ const retryDelay = Math.min(
+ VSR_RETRY_INTERVAL_MS,
+ Math.max(0, deadline - Date.now())
+ );
+ if (retryDelay === 0)
+ throw error;
+ await delay(retryDelay);
}
- return resolve(r);
- });
+ }
+
+ if (prepared.command === COMMAND_CODE.LoginRegister ||
+ prepared.command === COMMAND_CODE.LoginRegisterWithAccessToken) {
+ this.vsrSession.bind(readRegisteredSession(parsed));
+ this.isAuthenticated = true;
+ this.userId = parsed.data.readUInt32LE(0);
+ }
+ if (prepared.command === COMMAND_CODE.LogoutUser) {
+ this.isAuthenticated = false;
+ this.userId = undefined;
+ this.vsrSession.reset();
+ this.emit('sessionReset');
+ }
+ return parsed;
+ } catch (error) {
+ if (error instanceof VsrEvictionError) {
+ this.isAuthenticated = false;
+ this.userId = undefined;
+ this.vsrSession.reset();
+ this.emit('sessionReset');
+ } else if (!(error instanceof ResponseError)) {
Review Comment:
this branch decides to reset the session by sniffing the error type instead
of tracking whether anything was actually written. a transient-deadline exit
throws a plain `Error`, not `ResponseError(57)`, so a 30s deadline on a slow
command tears down the session and mints a fresh clientId - which kills group
membership. rust keeps the typed 57 and the session
(`core/sdk/src/tcp/tcp_client.rs`, its reconnect list omits 57). an encode
`RangeError` from `wire/vsr/index.ts:67-68` also resets, even though it fires
before any request id is spent.
the comment is also wrong. `core/server-ng/src/client_table.rs:185-187` says
it outright: "Watermark, not contiguity... There is no RequestGap". same stale
rationale is repeated at `wire/vsr/session.ts:26-32`. the wording exists
elsewhere in the repo so it wasn't invented here, but it shouldn't get
propagated into a new module.
fix: a `written` flag in `_exchange`, a typed deadline error, and rewrite
both comments.
##########
foreign/node/src/wire/error.utils.ts:
##########
@@ -19,10 +19,24 @@
import { translateCommandCode } from './command.code.js';
import { translateErrorCode } from './error.code.js';
-export const responseError = (cmdCode: number, errCode: number) => new Error(
- `command: { code: ${cmdCode}, name: ${translateCommandCode(cmdCode)} } ` +
- `error: {code: ${errCode}, message: ${translateErrorCode(errCode)} }`
-);
+export class ResponseError extends Error {
Review Comment:
`ResponseError`, `DeserializeError`, `VsrEvictionError` and
`ProtocolFrameError` aren't reachable from the package root - `src/index.ts`
re-exports `client/index.js` and `stream/index.js`, and neither pulls in the
error classes. the whole vsr error story is "catch it and match on it", so
users want `instanceof`.
not fatal since there's no exports map (deep dist imports work) and
`err.name` / `err.errorCode` duck-typing works, but re-exporting them is a
one-liner.
##########
foreign/node/src/wire/command-set.ts:
##########
@@ -165,6 +166,7 @@ const groupAPI = (c: ClientProvider) => ({
create: createGroup(c),
join: joinGroup(c),
leave: leaveGroup(c),
+ sync: syncGroup(c),
Review Comment:
`group.sync()` is new public API that always fails on classic -
`core/server/src/binary/dispatch.rs` has arms for get/create/delete/join/leave
consumer group but none for 606, so it falls through to `_ => InvalidCommand`
at `:451-453`. either gate it on protocol or keep it internal.
##########
foreign/node/src/wire/message/poll-messages.command.ts:
##########
@@ -67,7 +85,79 @@ export const POLL_MESSAGES = {
}
};
+const groupKey = ({ streamId, topicId, consumer }: PollMessages): string =>
+ `${String(streamId)}\0${String(topicId)}\0${String(consumer.id)}`;
+
+const syncAssignment = async (
+ client: RawClient,
+ request: PollMessages,
+): Promise<GroupCursor> => {
+ const response = await client.sendCommand(
+ SYNC_GROUP.code,
+ SYNC_GROUP.serialize({
+ streamId: request.streamId,
+ topicId: request.topicId,
+ groupId: request.consumer.id,
+ }),
+ );
+ const assignment = SYNC_GROUP.deserialize(response);
+ if (assignment === null)
+ throw responseError(SYNC_GROUP.code, 5006);
+
+ let cursors = groupCursors.get(client);
+ if (!cursors) {
+ cursors = new Map();
+ groupCursors.set(client, cursors);
+ client.once('sessionReset', () => groupCursors.delete(client));
+ }
+ const key = groupKey(request);
+ const current = cursors.get(key);
+ if (current && current.generation === assignment.generation) {
+ current.partitions = assignment.partitions;
+ if (current.position >= current.partitions.length)
+ current.position = 0;
+ return current;
+ }
+ const cursor = { ...assignment, position: 0 };
+ cursors.set(key, cursor);
+ return cursor;
+};
+
+const pollConsumerGroup = async (
+ client: RawClient,
+ request: PollMessages,
+): Promise<PollMessagesResponse> => {
+ for (let attempt = 0; attempt < GROUP_POLL_MAX_ATTEMPTS; attempt += 1) {
+ const cursor = await syncAssignment(client, request);
+ if (cursor.partitions.length === 0)
+ return { partitionId: 0, currentOffset: 0n, count: 0, messages: [] };
+
+ const partitionId = cursor.partitions[cursor.position];
+ cursor.position = (cursor.position + 1) % cursor.partitions.length;
+ const response = await client.sendCommand(
Review Comment:
the group poll spans several `sendCommand` calls but `finishQueue` fires
between them - `pendingSubmissions` only covers nested submissions, and it
fires twice for two sequential commands. so the pool releases the client
mid-operation. bounded blast radius, but real under the vsr pool of 1: a
session-reset command can land in the middle of the sequence, and the cursor is
shared.
##########
.github/actions/node-npm/pre-merge/action.yml:
##########
@@ -115,6 +113,47 @@ runs:
IGGY_SERVER_TCP_PORT: 8090
shell: bash
+ - name: Start Iggy VSR server
+ id: iggy-vsr
+ if: inputs.task == 'e2e-vsr'
+ uses: ./.github/actions/utils/server-start
Review Comment:
the vsr server starts with no cluster config - the plain e2e step sets
`replica-id` and `IGGY_CLUSTER_ENABLED`, this one sets neither. so
`_ensureVsrLeader`, `redirect()`, `isConnectedTo` and the no-leader throw path
get zero real-server coverage, which is exactly the code flagged elsewhere in
this review. worth either a cluster variant or enabling it here with a 2-node
expectation.
the build is also missing `--features vsr`. inert today since the feature
gates nothing in server-ng, but it's a latent false-green plus a rust-cache
flavor mismatch waiting to happen - `_test_bdd.yml:54-60` does it right.
minor: `IGGY_SERVER_HOST` / `IGGY_SERVER_TCP_PORT` on the step above are
dead, only `IGGY_TCP_ADDRESS` is read. copied from the existing step, not
introduced here.
##########
foreign/node/src/client/client.socket.ts:
##########
@@ -93,8 +123,17 @@ export class CommandResponseStream extends EventEmitter {
*/
_init() {
this.heartbeat(this.options.heartbeatInterval);
- this.connection.on('disconnected', async () => {
+ this.connection.on('error', (error: Error) => {
+ this._failQueue(error);
+ });
+ this.connection.on('disconnected', () => {
Review Comment:
the same four statements - clear auth, clear userId, reset the vsr session,
emit sessionReset - appear here and twice more at `:309-314` and `:316-331`.
one `_resetSession()` saves about 14 lines. one caveat: if the eviction branch
grows an `abort()` per the correlation fix, keep that one split out.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -125,54 +129,86 @@ export class IggyConnection extends EventEmitter {
constructor(config: ClientConfig) {
super();
this.config = config;
- this.socket = getTransport(config);
this.connected = false;
this.connecting = false;
this.ending = false;
this.waitingResponseEnd = false;
this.reconnectOption = { ...DefaultReconnectOption, ...config.reconnect };
this.reconnectCount = 0;
+ this.connectPromise = undefined;
this.readBuffers = Buffer.allocUnsafe(0);
+ this.socket = this._installSocket(getTransport(config));
}
/**
- * Establishes the connection to the server.
- * Sets up event handlers for data, errors, and disconnection.
- *
- * @returns Promise that resolves when connected
+ * Attaches the lifecycle listeners exactly once per socket instance.
+ * Attaching them in `connect()` would stack duplicate handlers whenever a
+ * failed attempt is retried on the same socket.
*/
- connect() {
- this.connecting = true;
-
- this.socket.on('data', this._onData.bind(this));
+ private _installSocket(socket: Socket): Socket {
+ socket.on('data', this._onData.bind(this));
- this.socket.on('error', async (err: SocketError) => {
+ socket.on('error', (err: SocketError) => {
debug('socket/error event', err, err.code, this.ending);
- // errors about disconnections should be ignored during disconnect
if (this.ending && (err?.code === 'ECONNRESET' || err?.code === 'EPIPE'))
return
-
- this.reconnect(err);
+ this.emit('error', err);
});
- this.socket.once('end', async (hadError?: boolean) => {
- debug('socket/close#END event', hadError);
+ socket.once('close', (hadError?: boolean) => {
+ debug('socket/close event', hadError);
this.connected = false;
+ this.connecting = false;
+ this.connectPromise = undefined;
this.emit('disconnected', hadError);
- this.reconnect();
+ if (!this.ending)
+ void this.reconnect();
});
+ return socket;
+ }
+
+ /**
+ * Establishes the connection to the server.
+ *
+ * @returns Promise that resolves when connected
+ */
+ connect(): Promise<this> {
Review Comment:
`connect()` never settles if the socket is already open. the constructor
opens the socket at `:140`, but the 'connect' listener is only registered in
here - if the event already fired, nothing resolves the promise, and
`connectPromise` caches the hang for every later caller. a pooled Client wedged
this way never resolves `destroy()` or `drain()` either.
master has the same shape so this isn't introduced here and shouldn't block
the PR, but `connect()` was rewritten wholesale and the fix is one guard:
resolve immediately when `!socket.connecting && readyState === 'open'`, or
register 'connect' in `_installSocket` alongside the other lifecycle listeners.
CI passing is iteration luck - a microtask boundary is safe, a `setTimeout(0)`
or an I/O await hangs, and both were observed in a single run.
##########
foreign/node/scripts/check-vsr-protocol.mjs:
##########
@@ -0,0 +1,301 @@
+// 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 { access, readFile } from 'node:fs/promises';
+import { resolve } from 'node:path';
+import { pathToFileURL } from 'node:url';
+
+const rootCandidates = [
Review Comment:
the first candidate resolves to `foreign/`, which can never hold
`core/binary_protocol/src/codes.rs`. the comment describes an isolated test
layout that doesn't exist in this tree. drop it and keep the monorepo root.
##########
foreign/node/src/wire/message/poll-messages.command.ts:
##########
@@ -67,7 +85,79 @@ export const POLL_MESSAGES = {
}
};
+const groupKey = ({ streamId, topicId, consumer }: PollMessages): string =>
+ `${String(streamId)}\0${String(topicId)}\0${String(consumer.id)}`;
+
+const syncAssignment = async (
+ client: RawClient,
+ request: PollMessages,
+): Promise<GroupCursor> => {
+ const response = await client.sendCommand(
+ SYNC_GROUP.code,
+ SYNC_GROUP.serialize({
+ streamId: request.streamId,
+ topicId: request.topicId,
+ groupId: request.consumer.id,
+ }),
+ );
+ const assignment = SYNC_GROUP.deserialize(response);
+ if (assignment === null)
+ throw responseError(SYNC_GROUP.code, 5006);
+
+ let cursors = groupCursors.get(client);
+ if (!cursors) {
+ cursors = new Map();
+ groupCursors.set(client, cursors);
+ client.once('sessionReset', () => groupCursors.delete(client));
+ }
+ const key = groupKey(request);
+ const current = cursors.get(key);
+ if (current && current.generation === assignment.generation) {
+ current.partitions = assignment.partitions;
+ if (current.position >= current.partitions.length)
+ current.position = 0;
+ return current;
+ }
+ const cursor = { ...assignment, position: 0 };
+ cursors.set(key, cursor);
+ return cursor;
+};
+
+const pollConsumerGroup = async (
+ client: RawClient,
+ request: PollMessages,
+): Promise<PollMessagesResponse> => {
+ for (let attempt = 0; attempt < GROUP_POLL_MAX_ATTEMPTS; attempt += 1) {
+ const cursor = await syncAssignment(client, request);
+ if (cursor.partitions.length === 0)
+ return { partitionId: 0, currentOffset: 0n, count: 0, messages: [] };
Review Comment:
returning `{partitionId: 0, count: 0}` for "no partitions assigned" is a
fine wire shape - rust's `PolledMessages::empty()` looks the same. the problem
is the stream layer can't tell it apart from "drained":
`stream/consumer-stream.ts:42-55` treats count 0 as end of data, so a member
legitimately holding zero partitions mid-rebalance ends an `endOnLastOffset`
stream on its very first poll. partitionId 0 being a real partition id makes it
worse. needs a way to distinguish no-assignment from drained - a sentinel, or
the cursor's real partition.
##########
foreign/node/src/wire/vsr/namespace.ts:
##########
@@ -0,0 +1,214 @@
+// 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.
+//
+
+/**
+ * Namespace packing and the partition-plane payload peeks needed to derive
+ * it, ported from `core/binary_protocol/src/namespace.rs` and
+ * `namespace_for_request` in `core/sdk/src/vsr.rs`.
+ */
+
+import { COMMAND_CODE } from '../command.code.js';
+import { responseError } from '../error.utils.js';
+import { Operation, isMetadata } from './operation.js';
+
+/** `IggyError::InvalidCommand`. */
+const INVALID_COMMAND = 3;
+/** `IggyError::FeatureUnavailable`. */
+const FEATURE_UNAVAILABLE = 5;
+/** `IggyError::InvalidIdentifier`. */
+const INVALID_IDENTIFIER = 6;
+
+const MAX_STREAMS = 4096;
+const MAX_TOPICS = 4096;
+const MAX_PARTITIONS = 1_000_000;
+const STREAM_SHIFT = 32n;
+const TOPIC_SHIFT = 20n;
+
+/**
+ * Control-plane requests target the metadata replica (shard 0), selected by
+ * this exact sentinel. Plain 0 would fall into namespace hashing and land a
+ * Register on a peer shard.
+ */
+export const METADATA_CONSENSUS_NAMESPACE = 1n << 63n;
+
+/** Packs stream / topic / partition ids into a routing namespace. */
+export const packNamespace = (
+ streamId: number,
+ topicId: number,
+ partitionId: number
+): bigint => {
+ validateField(streamId, MAX_STREAMS);
+ validateField(topicId, MAX_TOPICS);
+ validateField(partitionId, MAX_PARTITIONS);
+ return (BigInt(streamId) << STREAM_SHIFT) |
+ (BigInt(topicId) << TOPIC_SHIFT) |
+ BigInt(partitionId);
+};
+
+/**
+ * Selects the routing namespace for a request. Partition-plane commands
+ * derive it from their own payload; a named stream or topic identifier
+ * yields 0 so the server resolves the name.
+ *
+ * @throws Error mirroring the Rust SDK: invalid-identifier for an
+ * out-of-range field, invalid-command for an undecodable payload, and
+ * feature-unavailable for a partition operation this SDK cannot derive.
+ */
+export const namespaceForRequest = (
+ code: number,
+ payload: Buffer,
+ operation: number
+): bigint => {
+ if (operation === Operation.Register || operation === Operation.Logout)
+ return METADATA_CONSENSUS_NAMESPACE;
+ if (operation === Operation.NonReplicated || isMetadata(operation))
+ return 0n;
+
+ switch (code) {
+ case COMMAND_CODE.SendMessages:
+ return namespaceFromSendMessages(payload);
+ case COMMAND_CODE.StoreOffset:
+ case COMMAND_CODE.DeleteConsumerOffset:
+ case COMMAND_CODE.StoreOffset2:
+ case COMMAND_CODE.DeleteConsumerOffset2:
+ return namespaceFromConsumerOffset(payload);
+ case COMMAND_CODE.DeleteSegments:
+ return namespaceFromDeleteSegments(payload);
+ default:
+ // The guard that keeps custom partition operations unreachable.
+ throw responseError(code, FEATURE_UNAVAILABLE);
+ }
+};
+
+/** A decoded identifier: numeric value, or null for a name (server resolves).
*/
+type PeekedIdentifier = {
+ numeric: number | null,
+ length: number
+};
+
+const IDENTIFIER_KIND_NUMERIC = 1;
+const IDENTIFIER_KIND_STRING = 2;
+
+const peekIdentifier = (payload: Buffer, offset: number): PeekedIdentifier => {
+ if (payload.length < offset + 2)
+ throw responseError(0, INVALID_COMMAND);
+ const kind = payload.readUInt8(offset);
+ const length = payload.readUInt8(offset + 1);
+ if (payload.length < offset + 2 + length)
+ throw responseError(0, INVALID_COMMAND);
+ if (kind === IDENTIFIER_KIND_NUMERIC) {
+ if (length !== 4)
+ throw responseError(0, INVALID_COMMAND);
+ return { numeric: payload.readUInt32LE(offset + 2), length: 2 + length };
+ }
+ if (kind === IDENTIFIER_KIND_STRING && length > 0)
+ return { numeric: null, length: 2 + length };
+ throw responseError(0, INVALID_COMMAND);
+};
+
+const validateField = (value: number, exclusiveMax: number): void => {
+ if (value >= exclusiveMax)
+ throw responseError(0, INVALID_IDENTIFIER);
+};
+
+const namespaceFromIds = (
+ stream: PeekedIdentifier,
+ topic: PeekedIdentifier,
+ partitionId: number
+): bigint => {
+ // Named identifiers defer resolution to the server.
+ if (stream.numeric === null || topic.numeric === null) return 0n;
+ return packNamespace(stream.numeric, topic.numeric, partitionId);
+};
+
+/**
+ * `SendMessages`: `[metadata_len u32][stream ident][topic ident]
+ * [partitioning kind u8, len u8, value]...`. Only explicit `PartitionId`
+ * partitioning is routable under VSR; the broker never picks a partition.
+ */
+const namespaceFromSendMessages = (payload: Buffer): bigint => {
+ if (payload.length < 4)
+ throw responseError(COMMAND_CODE.SendMessages, INVALID_COMMAND);
+ const metadataLength = payload.readUInt32LE(0);
+ if (payload.length < 4 + metadataLength)
+ throw responseError(COMMAND_CODE.SendMessages, INVALID_COMMAND);
+ // Rust peeks inside payload[4..4 + metadata_length]; a read past the
+ // declared metadata region must fail rather than spill into message bytes
+ // and derive a namespace the server would never compute.
+ const metadata = payload.subarray(4, 4 + metadataLength);
+
+ let offset = 0;
+ const stream = peekIdentifier(metadata, offset);
+ offset += stream.length;
+ const topic = peekIdentifier(metadata, offset);
+ offset += topic.length;
+
+ if (metadata.length < offset + 2)
+ throw responseError(COMMAND_CODE.SendMessages, INVALID_COMMAND);
+ const partitioningKind = metadata.readUInt8(offset);
+ const partitioningLength = metadata.readUInt8(offset + 1);
+ const PARTITIONING_PARTITION_ID = 2;
+ if (partitioningKind !== PARTITIONING_PARTITION_ID)
+ throw responseError(COMMAND_CODE.SendMessages, FEATURE_UNAVAILABLE);
Review Comment:
default typed produce doesn't work under vsr. `message.send({streamId,
topicId, messages})` leaves `partition` unset (optional in
`send-messages.command.ts:37`), `partitioning.utils.ts:134-138` fills in
Balanced, and this line rejects it with code 5 `Feature is unavailable` before
anything reaches the wire. `MessageKey` hits the same wall.
the restriction itself is not node-specific - `core/sdk/src/vsr.rs:420`
rejects everything but `PartitionId` too, and `IggyProducer` defaults to
`Partitioning::balanced()` (`core/sdk/src/clients/producer.rs:467`), so rust
fails the same way unless you wire a custom partitioner. what's node-specific
is that there's no way out: no partition-count API to build a resolver on, no
producer layer, and `message.send` is the only send API there is. the PR's own
e2e quietly swaps to `PartitionId` in exactly the suites that would have caught
this, and the README says nothing.
`offset.store` on a group with `partitionId: null` has the same shape -
throws code 6 even though the JSDoc at `offset.utils.ts:93` promises null means
"the group's assigned partition". `pollConsumerGroup` resolves the partition
internally and never surfaces it, so a vsr group app can poll but cannot commit
the documented way.
minimum: document the constraint and fix that JSDoc. better: surface the
cursor's partitionId so commits have something to pass.
##########
foreign/node/src/wire/message/poll-messages.command.ts:
##########
@@ -67,7 +85,79 @@ export const POLL_MESSAGES = {
}
};
+const groupKey = ({ streamId, topicId, consumer }: PollMessages): string =>
+ `${String(streamId)}\0${String(topicId)}\0${String(consumer.id)}`;
+
+const syncAssignment = async (
+ client: RawClient,
+ request: PollMessages,
+): Promise<GroupCursor> => {
+ const response = await client.sendCommand(
+ SYNC_GROUP.code,
+ SYNC_GROUP.serialize({
+ streamId: request.streamId,
+ topicId: request.topicId,
+ groupId: request.consumer.id,
+ }),
+ );
+ const assignment = SYNC_GROUP.deserialize(response);
+ if (assignment === null)
+ throw responseError(SYNC_GROUP.code, 5006);
+
+ let cursors = groupCursors.get(client);
+ if (!cursors) {
+ cursors = new Map();
+ groupCursors.set(client, cursors);
+ client.once('sessionReset', () => groupCursors.delete(client));
+ }
+ const key = groupKey(request);
+ const current = cursors.get(key);
+ if (current && current.generation === assignment.generation) {
+ current.partitions = assignment.partitions;
+ if (current.position >= current.partitions.length)
+ current.position = 0;
+ return current;
+ }
+ const cursor = { ...assignment, position: 0 };
+ cursors.set(key, cursor);
+ return cursor;
+};
+
+const pollConsumerGroup = async (
+ client: RawClient,
+ request: PollMessages,
+): Promise<PollMessagesResponse> => {
+ for (let attempt = 0; attempt < GROUP_POLL_MAX_ATTEMPTS; attempt += 1) {
+ const cursor = await syncAssignment(client, request);
Review Comment:
`syncAssignment` runs a SyncGroup RPC before every single group poll - 2
round trips per poll, where rust only syncs on cache miss, fence, or heartbeat
refresh.
the catch is the unconditional sync is load-bearing right now, since node
has no other refresh trigger. dropping straight to a cache would strand widened
assignments after a rebalance. so it needs the bundle: assignment cache, a
refresh trigger, exposing the cursor's partitionId (same plumbing the
offset-commit fix wants), and a kind-tagged `groupKey`.
##########
foreign/node/src/wire/message/poll-messages.command.ts:
##########
@@ -67,7 +85,79 @@ export const POLL_MESSAGES = {
}
};
+const groupKey = ({ streamId, topicId, consumer }: PollMessages): string =>
+ `${String(streamId)}\0${String(topicId)}\0${String(consumer.id)}`;
+
+const syncAssignment = async (
+ client: RawClient,
+ request: PollMessages,
+): Promise<GroupCursor> => {
+ const response = await client.sendCommand(
+ SYNC_GROUP.code,
+ SYNC_GROUP.serialize({
+ streamId: request.streamId,
+ topicId: request.topicId,
+ groupId: request.consumer.id,
+ }),
+ );
+ const assignment = SYNC_GROUP.deserialize(response);
+ if (assignment === null)
+ throw responseError(SYNC_GROUP.code, 5006);
+
+ let cursors = groupCursors.get(client);
+ if (!cursors) {
+ cursors = new Map();
+ groupCursors.set(client, cursors);
+ client.once('sessionReset', () => groupCursors.delete(client));
Review Comment:
this is the only `sessionReset` reaction anywhere in the SDK, and all it
does is drop the cursor.
any disconnect fires it (`client.socket.ts:129-137`), re-register mints a
fresh clientId, and the server keys group membership by that id
(`core/server-ng/src/dispatch.rs:1032` - "`vsr_client_id` keys the
consumer-group offset fence (the member id)"). so `SyncGroup` comes back empty
forever, then 5006, and the stream `Readable` errors permanently.
`groupConsumerStream` joins once at `stream/consumer-stream.ts:119` and
never rejoins, so one routine TCP blip kills the feature's main consumer API
with no recovery path. reproduced end to end.
rust covers both halves - rejoin on `SignedIn` after reconnect
(`core/sdk/src/clients/consumer.rs:612-646`) and a 5006-triggered rejoin
(`:810`). node needs the same: rejoin on sessionReset, or off 5006.
##########
foreign/node/scripts/check-vsr-protocol.mjs:
##########
@@ -0,0 +1,301 @@
+// 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 { access, readFile } from 'node:fs/promises';
+import { resolve } from 'node:path';
+import { pathToFileURL } from 'node:url';
+
+const rootCandidates = [
+ resolve(import.meta.dirname, '../..'),
+ resolve(import.meta.dirname, '../../..')
+];
+let root;
+for (const candidate of rootCandidates) {
+ try {
+ await access(resolve(candidate, 'core/binary_protocol/src/codes.rs'));
+ root = candidate;
+ break;
+ } catch {
+ // Try the monorepo layout after the isolated test layout.
+ }
+}
+assert.ok(root, 'Apache Iggy repository root was not found');
+const read = (path) => readFile(resolve(root, path), 'utf8');
+const nodeRoot = resolve(import.meta.dirname, '..');
+const readNode = (path) => readFile(resolve(nodeRoot, path), 'utf8');
+
+const [
+ rustCodes,
+ rustDispatch,
+ rustHeader,
+ rustCommand,
+ rustOperation,
+ rustProtocolCargo,
+ nodeCodes,
+ nodeHeader,
+ nodeOperation,
+ nodeRegister,
+] = await Promise.all([
+ read('core/binary_protocol/src/codes.rs'),
+ read('core/binary_protocol/src/dispatch.rs'),
+ read('core/binary_protocol/src/consensus/header.rs'),
+ read('core/binary_protocol/src/consensus/command.rs'),
+ read('core/binary_protocol/src/consensus/operation.rs'),
+ read('core/binary_protocol/Cargo.toml'),
+ readNode('src/wire/command.code.ts'),
+ readNode('src/wire/vsr/header.ts'),
+ readNode('src/wire/vsr/operation.ts'),
+ readNode('src/wire/vsr/register.ts'),
+]);
+
+const numericValues = (source, pattern) =>
+ [...source.matchAll(pattern)].map((match) => Number(match[1]));
+
+const rustCommandCodes = numericValues(
+ rustCodes,
+ /^pub const [A-Z0-9_]+_CODE: u32 = ([0-9]+);$/gm
+).sort((left, right) => left - right);
+const nodeCommandBlock =
+ nodeCodes.match(/export const COMMAND_CODE = \{([\s\S]*?)\n\};/)?.[1] ?? '';
+const nodeCommandCodes = numericValues(
+ nodeCommandBlock,
+ /^\s*[A-Za-z0-9]+:\s*([0-9]+),/gm
+).sort((left, right) => left - right);
+assert.deepEqual(
+ nodeCommandCodes,
+ rustCommandCodes,
+ 'Node COMMAND_CODE differs from the Rust command registry'
+);
+
+const enumValues = (source, pattern) =>
+ new Map(
+ [...source.matchAll(pattern)].map((match) => [
+ match[1],
+ Number(match[2])
+ ])
+ );
+const rustOperations = enumValues(
+ rustOperation,
+ /^\s+([A-Za-z0-9]+)\s*=\s*([0-9]+),$/gm
+);
+const nodeOperations = enumValues(
+ nodeOperation,
+ /^\s+([A-Za-z0-9]+):\s*([0-9]+),?$/gm
+);
+assert.deepEqual(
+ nodeOperations,
+ rustOperations,
+ 'Node Operation differs from the Rust consensus enum'
+);
+
+const rustReplicated = new Set(
Review Comment:
two holes, both confirmed by mutating the rust sources and watching the gate
stay green.
first, the gate never reads `core/binary_protocol/src/namespace.rs`. rust
derives the shifts from `MAX_PARTITIONS` via `bits_required`, node hardcodes
`STREAM_SHIFT = 32n` / `TOPIC_SHIFT = 20n` (`wire/vsr/namespace.ts:39-40`).
bump `MAX_PARTITIONS` and every partition write goes to the wrong shard while
the gate reports fine. recompute the shifts from namespace.rs instead of
hardcoding them.
second, this replicated check compares sets of operation *names*, not the
code to operation pairing. swap CreateStream and DeleteStream and it passes.
compare numeric `{codeValue -> operation}` maps instead - const names don't map
1:1 to the node keys, so a name-based comparison structurally can't catch
mis-pairing.
##########
foreign/node/scripts/check-vsr-protocol.mjs:
##########
@@ -0,0 +1,301 @@
+// 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 { access, readFile } from 'node:fs/promises';
+import { resolve } from 'node:path';
+import { pathToFileURL } from 'node:url';
+
+const rootCandidates = [
+ resolve(import.meta.dirname, '../..'),
+ resolve(import.meta.dirname, '../../..')
+];
+let root;
+for (const candidate of rootCandidates) {
+ try {
+ await access(resolve(candidate, 'core/binary_protocol/src/codes.rs'));
+ root = candidate;
+ break;
+ } catch {
+ // Try the monorepo layout after the isolated test layout.
+ }
+}
+assert.ok(root, 'Apache Iggy repository root was not found');
+const read = (path) => readFile(resolve(root, path), 'utf8');
+const nodeRoot = resolve(import.meta.dirname, '..');
+const readNode = (path) => readFile(resolve(nodeRoot, path), 'utf8');
+
+const [
+ rustCodes,
+ rustDispatch,
+ rustHeader,
+ rustCommand,
+ rustOperation,
+ rustProtocolCargo,
+ nodeCodes,
+ nodeHeader,
+ nodeOperation,
+ nodeRegister,
+] = await Promise.all([
+ read('core/binary_protocol/src/codes.rs'),
+ read('core/binary_protocol/src/dispatch.rs'),
+ read('core/binary_protocol/src/consensus/header.rs'),
+ read('core/binary_protocol/src/consensus/command.rs'),
+ read('core/binary_protocol/src/consensus/operation.rs'),
+ read('core/binary_protocol/Cargo.toml'),
+ readNode('src/wire/command.code.ts'),
+ readNode('src/wire/vsr/header.ts'),
+ readNode('src/wire/vsr/operation.ts'),
+ readNode('src/wire/vsr/register.ts'),
+]);
+
+const numericValues = (source, pattern) =>
+ [...source.matchAll(pattern)].map((match) => Number(match[1]));
+
+const rustCommandCodes = numericValues(
+ rustCodes,
+ /^pub const [A-Z0-9_]+_CODE: u32 = ([0-9]+);$/gm
+).sort((left, right) => left - right);
+const nodeCommandBlock =
+ nodeCodes.match(/export const COMMAND_CODE = \{([\s\S]*?)\n\};/)?.[1] ?? '';
+const nodeCommandCodes = numericValues(
+ nodeCommandBlock,
+ /^\s*[A-Za-z0-9]+:\s*([0-9]+),/gm
+).sort((left, right) => left - right);
+assert.deepEqual(
+ nodeCommandCodes,
+ rustCommandCodes,
+ 'Node COMMAND_CODE differs from the Rust command registry'
+);
+
+const enumValues = (source, pattern) =>
+ new Map(
+ [...source.matchAll(pattern)].map((match) => [
+ match[1],
+ Number(match[2])
+ ])
+ );
+const rustOperations = enumValues(
+ rustOperation,
+ /^\s+([A-Za-z0-9]+)\s*=\s*([0-9]+),$/gm
+);
+const nodeOperations = enumValues(
+ nodeOperation,
+ /^\s+([A-Za-z0-9]+):\s*([0-9]+),?$/gm
+);
+assert.deepEqual(
+ nodeOperations,
+ rustOperations,
+ 'Node Operation differs from the Rust consensus enum'
+);
+
+const rustReplicated = new Set(
+ [...rustDispatch.matchAll(
+ /CommandMeta::replicated\([\s\S]*?Operation::([A-Za-z0-9]+)/g
+ )].map((match) => match[1])
+);
+const nodeReplicatedBlock =
+ nodeOperation.match(
+ /const REPLICATED_OPERATION[\s\S]*?new Map\(\[([\s\S]*?)\]\);/
+ )?.[1] ?? '';
+const nodeReplicated = new Set(
+ [...nodeReplicatedBlock.matchAll(/Operation\.([A-Za-z0-9]+)/g)]
+ .map((match) => match[1])
+);
+assert.deepEqual(
+ nodeReplicated,
+ rustReplicated,
+ 'Node replicated operations differ from the Rust dispatch table'
+);
+
+const rustEvictionBlock =
+ rustHeader.match(/pub enum EvictionReason \{([\s\S]*?)\n\}/)?.[1] ?? '';
+const rustEvictions = enumValues(
+ rustEvictionBlock,
+ /^\s+([A-Za-z0-9]+)\s*=\s*([0-9]+),$/gm
+);
+const nodeEvictionBlock =
+ nodeHeader.match(/export const EvictionReason = \{([\s\S]*?)\n\}/)?.[1] ??
+ '';
+const nodeEvictions = enumValues(
+ nodeEvictionBlock,
+ /^\s+([A-Za-z0-9]+):\s*([0-9]+),?$/gm
+);
+assert.deepEqual(
+ nodeEvictions,
+ rustEvictions,
+ 'Node EvictionReason differs from the Rust consensus enum'
+);
+
+const rustCommand2Block =
+ rustCommand.match(/pub enum Command2 \{([\s\S]*?)\n\}/)?.[1] ?? '';
+const rustCommand2 = enumValues(
+ rustCommand2Block,
+ /^\s+([A-Za-z0-9]+)\s*=\s*([0-9]+),$/gm
+);
+const nodeCommand2 = enumValues(
+ nodeHeader.match(/export const Command2 = \{([\s\S]*?)\n\}/)?.[1] ?? '',
+ /^\s+([A-Za-z0-9]+):\s*([0-9]+),?$/gm
+);
+assert.ok(nodeCommand2.size > 0, 'Node Command2 table was not found');
+for (const [name, value] of nodeCommand2)
+ assert.equal(
+ rustCommand2.get(name),
+ value,
+ `Node Command2.${name} differs from Rust`
+ );
+
+const rustHeaderSize = Number(
+ rustHeader.match(/pub const HEADER_SIZE: usize = ([0-9]+);/)?.[1]
+);
+const nodeHeaderSize = Number(
+ nodeHeader.match(/export const HEADER_SIZE = ([0-9]+);/)?.[1]
+);
+assert.equal(nodeHeaderSize, rustHeaderSize, 'VSR header size differs');
+
+// Header field offsets: recompute the #[repr(C)] layout from the Rust struct
+// declarations so a field inserted before the consumed offsets fails here
+// instead of desyncing the hardcoded Node tables.
+const FIELD_LAYOUT = new Map([
+ ['u8', [1, 1]],
+ ['u16', [2, 2]],
+ ['u32', [4, 4]],
+ ['u64', [8, 8]],
+ ['u128', [16, 16]],
+ ['Command2', [1, 1]],
+ ['Operation', [1, 1]],
+ ['EvictionReason', [1, 1]]
+]);
+
+const rustStructOffsets = (structName) => {
+ const body = rustHeader.match(
+ new RegExp(`pub struct ${structName} \\{([\\s\\S]*?)\\n\\}`)
+ )?.[1];
+ assert.ok(body, `Rust struct ${structName} was not found`);
+ const offsets = new Map();
+ let offset = 0;
+ for (const [, field, type] of body.matchAll(
+ /pub ([a-z_0-9]+): ([A-Za-z0-9_]+|\[u8; [0-9]+\]),/g
+ )) {
+ const arrayLength = type.match(/\[u8; ([0-9]+)\]/)?.[1];
+ const [size, align] = arrayLength
+ ? [Number(arrayLength), 1]
+ : FIELD_LAYOUT.get(type) ?? [];
+ assert.ok(
+ size !== undefined,
+ `unknown Rust field type ${type} in ${structName}`
+ );
+ offset = Math.ceil(offset / align) * align;
+ offsets.set(field, offset);
+ offset += size;
+ }
+ assert.equal(
+ offset,
+ rustHeaderSize,
+ `computed ${structName} layout does not fill HEADER_SIZE`
+ );
+ return offsets;
+};
+
+const nodeOffsetTable = (tableName) => {
+ const block = nodeHeader.match(
+ new RegExp(`export const ${tableName} = \\{([\\s\\S]*?)\\n\\} as const;`)
+ )?.[1];
+ assert.ok(block, `Node offset table ${tableName} was not found`);
+ return enumValues(block, /^\s+([A-Za-z0-9]+):\s*([0-9]+),?$/gm);
+};
+
+const toSnakeCase = (name) =>
+ name.replace(/([A-Z])/g, '_$1').toLowerCase();
+
+for (const [tableName, structName] of [
+ ['REQUEST_OFFSET', 'RequestHeader'],
+ ['REPLY_OFFSET', 'ReplyHeader'],
+ ['EVICTION_OFFSET', 'EvictionHeader']
+]) {
+ const rustOffsets = rustStructOffsets(structName);
+ for (const [field, value] of nodeOffsetTable(tableName))
+ assert.equal(
+ value,
+ rustOffsets.get(toSnakeCase(field)),
+ `Node ${tableName}.${field} differs from the Rust ${structName} layout`
+ );
+}
+
+// Operation classification: evaluate the compiled Node predicates against
+// the band constants and allowlists declared by the Rust enum.
+const operationModule = await import(
+ pathToFileURL(resolve(nodeRoot, 'dist/wire/vsr/operation.js')).href
+);
+const internalStart = rustOperations.get('CreateTopicWithAssignments');
+const metadataStart = rustOperations.get('CreateStream');
+const partitionStart = rustOperations.get('SendMessages');
+const rustMetadataNames = new Set(
+ [...(rustOperation.match(
+ /fn is_metadata[\s\S]*?matches!\(\s*self,([\s\S]*?)\)\s*\n\s*\}/
+ )?.[1] ?? '').matchAll(/Self::([A-Za-z0-9]+)/g)].map((match) => match[1])
+);
+assert.ok(rustMetadataNames.size > 0, 'Rust is_metadata allowlist not found');
+const rustResultFramedNames = new Set(
+ [...(rustOperation.match(
+ /fn is_result_framed[\s\S]*?matches!\(\s*self,([\s\S]*?)\)\s*\n\s*\}/
+ )?.[1] ?? '').matchAll(/Self::([A-Za-z0-9]+)/g)].map((match) => match[1])
+);
+assert.ok(
+ rustResultFramedNames.size > 0,
+ 'Rust is_result_framed allowlist not found'
+);
+for (const [name, value] of rustOperations) {
+ const internal = value >= internalStart && value < metadataStart;
+ const metadata = internal || rustMetadataNames.has(name);
+ assert.equal(
+ operationModule.isMetadata(value),
+ metadata,
+ `Node isMetadata(${name}) differs from Rust is_metadata`
+ );
+ assert.equal(
+ operationModule.isPartition(value),
+ value >= partitionStart,
+ `Node isPartition(${name}) differs from Rust is_partition`
+ );
+ assert.equal(
+ operationModule.isResultFramed(value),
+ metadata || rustResultFramedNames.has(name),
+ `Node isResultFramed(${name}) differs from Rust is_result_framed`
+ );
+}
+
+const protocolVersion =
+ rustProtocolCargo.match(/^version = "([0-9]+)\.([0-9]+)\.([0-9]+)/m);
+assert.ok(protocolVersion, 'binary protocol crate version is missing');
+const packedVersion =
+ Number(protocolVersion[1]) << 20 |
+ Number(protocolVersion[2]) << 10 |
+ Number(protocolVersion[3]);
+const nodePackedVersion = nodeRegister.match(
+ /export const IGGY_PROTOCOL_VERSION =\s*\(([0-9]+) << 20\) \| \(([0-9]+) <<
10\) \| ([0-9]+);/
+);
+assert.ok(nodePackedVersion, 'Node packed protocol version source changed');
+assert.equal(
Review Comment:
exact patch equality is stricter than the actual contract. rust masks the
patch out of the minimum (`core/binary_protocol/src/version.rs:117`) and
ignores it for compatibility (`:129-132`), so a patch bump on
`iggy_binary_protocol` never changes the wire but does red-build foreign/node.
compare major.minor only.
keep minor strict though - on 0.x a minor bump can legitimately break the
wire, which `version.rs:113-116` calls out.
##########
foreign/node/src/wire/error.code.ts:
##########
@@ -189,7 +189,7 @@ export const translateErrorCode = (code: number): string =>
{
case '5003': return "Failed to close file";
case '5004': return "Failed to delete file";
case '5005': return "Cannot read file";
- case '5006': return "Invalid file size";
+ case '5006': return "Consumer group member not found";
case '5007': return "Cannot create file";
case '5008': return "Cannot rename file";
case '5009': return "Cannot get file info";
Review Comment:
the whole 5000 block is wrong. rust's 5000-5009 range is consumer group
errors, not file/storage - 5001 doesn't exist in rust at all, and 5010-5020 are
invented here. this PR fixed 5006 but left the rest, so the resync path it adds
renders 5009 `ConsumerGroupPartitionNotOwned` as "Cannot get file info".
diagnostics only and mostly pre-existing, but it's specifically the fence this
PR depends on that misreports.
##########
foreign/node/src/client/client.socket.ts:
##########
@@ -104,54 +143,86 @@ export class CommandResponseStream extends EventEmitter {
*
* @param command - Command code to send
* @param payload - Command payload buffer
- * @param handleResponse - Whether to parse the response (default: true)
- * @param last - Whether to add to end of queue (default: true)
+ * @param options - Response and queue options
* @returns Promise resolving to the command response
*/
async sendCommand(
command: number,
payload: Buffer,
- handleResponse = true,
- last = true
+ options: SendCommandOptions = {}
): Promise<CommandResponse> {
+ this.pendingSubmissions += 1;
+ try {
+ const {
+ handleResponse = true,
+ last = true
+ } = options;
- if (!this.connection.connected)
- await this.connection.connect()
+ if (!this.connection.connected)
+ await this.connection.connect()
- if (!this.isAuthenticated && !UNLOGGED_COMMAND_CODE.includes(command))
- await this.authenticate(this.options.credentials);
+ if (this.options.protocol === 'vsr' &&
+ isLoginCommand(command) &&
+ this.isAuthenticated)
+ await this.sendCommand(
Review Comment:
this LOGOUT is unshifted to the front of the queue while the LOGIN that
follows goes through `wrapCommand` and lands at the back, so the whole
pre-existing queue runs between them. under an already-reset session every one
of those comes back `Unauthenticated`. keep the pair adjacent.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -204,8 +243,45 @@ export class IggyConnection extends EventEmitter {
/** recreate socket */
this.connecting = true;
this.reconnectCount += 1;
- this.socket = await recreate(this.config, interval);
- this.connect();
+ this.socket = this._installSocket(await recreate(this.config, interval));
+ this.connecting = false;
+ try {
+ await this.connect();
+ } catch (error) {
+ debug('reconnect attempt failed', error);
+ }
+ }
+
+ async redirect(host: string, port: number) {
+ this.socket.removeAllListeners();
+ this.socket.destroy();
+ this.connected = false;
+ this.connecting = false;
+ this.connectPromise = undefined;
+ this._endResponseWait();
+ // The old socket's close handler was just detached, so surface the drop
+ // to any in-flight exchange and queued work ourselves.
+ this.emit('disconnected', false);
+ this.config.options = { ...this.config.options, host, port };
Review Comment:
config is committed before `await connect()`, and this is the only
production write to `config.options`. if the redirect connect fails, the
configured seed endpoint is gone for good - `reconnect()` then re-dials the
dead ex-leader 12 times, and `isConnectedTo`'s config fallback reports the
wrong thing (the test at `client.connection.test.ts:85-95` canonizes that
behavior). commit the config only after connect resolves, and reset
`reconnectCount` on success.
##########
foreign/node/src/client/client.ts:
##########
@@ -43,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();
Review Comment:
if anything throws between `pool.acquire()` and the `finishQueue` release,
the connection never comes back - there's no try/catch, and `wrapCommand`
serializes *after* acquire, so a serializer throw is enough. on classic that's
a slow leak. under vsr the pool is pinned to max 1, so the first throw
deadlocks the Client permanently. reachable through `serializeGetOffset`'s
single/null guard. either release on throw, or serialize before acquiring.
##########
codecov.yml:
##########
@@ -62,7 +62,13 @@ flag_management:
- name: php
paths:
- foreign/php/
- - name: node
+ - name: node-test
Review Comment:
renaming the `node` flag to `node-test` orphans the old one - with
`carryforward: true` it sticks around in the codecov UI reporting stale
coverage until someone deletes it by hand. worth a note in the PR body.
##########
foreign/node/src/wire/vsr/reply.ts:
##########
@@ -0,0 +1,165 @@
+// 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.
+//
+
+/**
+ * Reply decode funnel, ported from `decode_response` in
+ * `core/sdk/src/vsr.rs`. Strict order: frame length, frame discriminant
+ * (evictions surface as typed errors, never a read timeout), declared size,
+ * pre-commit denial status before any body decode, then the committed
+ * result section for result-framed operations.
+ */
+
+import { ResponseError, responseError } from '../error.utils.js';
+import {
+ Command2, EvictionReason, HEADER_SIZE,
+ peekCommand, readEviction, readReplyOperation, readSize, readStatus
+} from './header.js';
+import { Operation, isKnownOperation, isResultFramed } from './operation.js';
+
+/** `IggyError` codes this funnel raises client-side. */
+const INVALID_COMMAND = 3;
+const UNAUTHENTICATED = 40;
+const INVALID_CREDENTIALS = 42;
+const INVALID_PERSONAL_ACCESS_TOKEN = 53;
+const STALE_CLIENT = 30;
+const INVALID_FORMAT = 4;
+const EMPTY_RESPONSE = 304;
+const INCOMPATIBLE_PROTOCOL_VERSION = 14003;
+
+const RESULT_COUNT_LEN = 4;
+const RESULT_ENTRY_LEN = 8;
+
+export class VsrEvictionError extends ResponseError {
+ constructor(errorCode: number) {
+ super(0, errorCode);
+ this.name = 'VsrEvictionError';
+ Object.setPrototypeOf(this, VsrEvictionError.prototype);
+ }
+}
+
+/**
+ * Decodes one complete consensus frame into the reply body, stripping the
+ * committed result section for result-framed operations and mapping every
+ * denial channel to a thrown error.
+ */
+export const decodeResponse = (frame: Buffer): Buffer => {
+ if (frame.length < HEADER_SIZE)
+ throw responseError(0, EMPTY_RESPONSE);
+
+ switch (peekCommand(frame)) {
+ case Command2.Eviction:
+ throw evictionError(frame);
+ case Command2.Reply:
+ break;
+ default:
+ throw responseError(0, INVALID_COMMAND);
+ }
+
+ const size = readSize(frame);
+ if (size < HEADER_SIZE || frame.length < size)
+ throw responseError(0, INVALID_COMMAND);
+
+ const status = readStatus(frame);
+ // A pre-commit denial always ships an empty body; the status channel and
+ // the committed result section are mutually exclusive.
+ if (status !== 0)
+ throw responseError(0, status);
+
+ const operation = readReplyOperation(frame);
+ if (!isKnownOperation(operation))
+ throw responseError(0, INVALID_COMMAND);
+ const body = frame.subarray(HEADER_SIZE, size);
+ return splitMetadataResult(operation, body);
+};
+
+/**
+ * Strips the committed result section leading a result-framed reply body.
+ * A Register reply is result-framed too, except a terminal register failure
+ * ships an empty body, passed through so the typed decode fails. A metadata
+ * body too short for the entries its count claims is corruption, never a
+ * silent success.
+ */
+export const splitMetadataResult = (
+ operation: number,
+ body: Buffer
+): Buffer => {
+ const resultFramed = isResultFramed(operation) ||
+ (operation === Operation.Register && body.length > 0);
+ if (!resultFramed) return body;
+
+ if (body.length < RESULT_COUNT_LEN)
+ throw responseError(0, INVALID_COMMAND);
+ const count = body.readUInt32LE(0);
+ const sectionLength = RESULT_COUNT_LEN + count * RESULT_ENTRY_LEN;
+ if (body.length < sectionLength)
+ throw responseError(0, INVALID_COMMAND);
+ if (count === 0)
Review Comment:
this throws on any count >= 1, but rust treats a first result of 0 as
success and strips the full `4 + count*8` section - see `result_code` in
`core/binary_protocol/src/consensus/reply_result.rs:55-65` and
`split_metadata_result` in `core/sdk/src/vsr.rs:293-301`. node turns that same
body into a throw. (the count == 0 path does agree, since the section is
exactly 4 bytes there.)
unreachable today - the server only emits count <= 1, rejection only - and
it fails closed, so this is forward-compat rather than a live bug. still worth
a `code === 0` success arm. same for threading the real command code: right now
the failure carries `errorCode` 0, which `translateErrorCode` renders as the
literal string "error".
##########
foreign/node/src/client/client.frame.ts:
##########
@@ -0,0 +1,77 @@
+// 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 type { Protocol } from './client.type.js';
+import {
+ HEADER_SIZE as VSR_HEADER_SIZE,
+ readSize as readVsrSize
+} from '../wire/vsr/header.js';
+
+const CLASSIC_HEADER_SIZE = 8;
+
+export class ProtocolFrameError extends Error {
+ constructor(message: string) {
+ super(message);
+ this.name = 'ProtocolFrameError';
+ }
+}
+
+export type ExtractedFrames = {
+ frames: Buffer[],
+ remainder: Buffer
+};
+
+export const extractResponseFrames = (
+ protocol: Protocol,
+ buffer: Buffer,
+ maximumFrameSize: number
+): ExtractedFrames => {
+ const headerSize =
+ protocol === 'vsr' ? VSR_HEADER_SIZE : CLASSIC_HEADER_SIZE;
+ const frames: Buffer[] = [];
+ let offset = 0;
+
+ while (buffer.length - offset >= headerSize) {
+ const available = buffer.length - offset;
+ const declaredSize = protocol === 'vsr'
+ ? readVsrSize(buffer.subarray(offset, offset + headerSize))
+ : CLASSIC_HEADER_SIZE + buffer.readUInt32LE(offset + 4);
+
+ if (declaredSize < headerSize)
+ throw new ProtocolFrameError(
+ `declared ${protocol} frame size ${declaredSize} is below header size`
+ );
+ if (!Number.isSafeInteger(declaredSize) ||
Review Comment:
`Number.isSafeInteger(declaredSize)` can't fail - `declaredSize` comes from
`readUInt32LE`, so it's bounded by 2^32. the adjacent bound check does the real
work. same dead guard at `wire/vsr/index.ts:67` and
`wire/consumer-group/sync-group.command.ts:50`.
keep the `partitionsLength > data.length - 12` half of the sync-group one,
two tests assert its message.
##########
foreign/node/src/wire/vsr/index.ts:
##########
@@ -0,0 +1,151 @@
+// 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 { createRequire } from 'node:module';
+import type { CommandResponse } from '../../client/client.type.js';
+import { COMMAND_CODE } from '../command.code.js';
+import { responseError } from '../error.utils.js';
+import { HEADER_SIZE, encodeRequestHeader } from './header.js';
+import { namespaceForRequest } from './namespace.js';
+import {
+ Operation,
+ isPartition,
+ operationForCode,
+} from './operation.js';
+import {
+ deserializeLoginRegister,
+ serializeLoginRegister,
+ serializeLoginRegisterWithPat,
+} from './register.js';
+import { decodeResponse } from './reply.js';
+import { ConsensusSession } from './session.js';
+
+const packageMetadata = createRequire(import.meta.url)(
+ '../../../package.json'
+) as { version: string };
+const SDK_VERSION = packageMetadata.version;
+const MAX_U32 = 0xFFFF_FFFF;
+/** `IggyError::Unauthenticated`, matching the Rust SDK's unbound-session
error. */
+const UNAUTHENTICATED = 40;
+
+export class VsrSession {
+ private state: ConsensusSession;
+
+ constructor(clientId?: bigint) {
+ this.state = new ConsensusSession(clientId);
+ }
+
+ reset(): void {
+ this.state = new ConsensusSession();
+ }
+
+ bind(session: bigint): void {
+ this.state.bind(session);
+ }
+
+ encode(command: number, payload: Buffer): Buffer {
+ const operation = registerCommand(command)
+ ? Operation.Register
+ : operationForCode(command);
+ const namespace = namespaceForRequest(command, payload, operation);
+ const size = HEADER_SIZE + payload.length;
+ if (!Number.isSafeInteger(size) || size > MAX_U32)
+ throw new RangeError('VSR request exceeds the u32 frame-size limit');
+
+ let request: bigint;
+ let session: bigint;
+
+ if (operation === Operation.Register) {
+ request = this.state.beginRegister();
+ session = 0n;
+ } else if (operation === Operation.NonReplicated) {
+ request = this.state.currentRequestId();
+ session = this.state.session ?? 0n;
+ } else {
+ if (this.state.session === null)
+ throw responseError(command, UNAUTHENTICATED);
+ request = isPartition(operation)
+ ? this.state.currentRequestId()
+ : this.state.nextRequestId();
+ session = this.state.session;
+ }
+
+ const header = encodeRequestHeader({
+ size: HEADER_SIZE + payload.length,
Review Comment:
`size` was already computed and validated at `:66` - pass it instead of
recomputing `HEADER_SIZE + payload.length`. and the conditional spread at
`:95-97` can be `nonReplicatedCode: operation === Operation.NonReplicated ?
command : undefined`, which compiles clean without `exactOptionalPropertyTypes`
and drops an intermediate object per request.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -239,47 +315,28 @@ export class IggyConnection extends EventEmitter {
this.waitingResponseEnd
);
- // Append new data to any buffered data
- if (this.waitingResponseEnd && this.readBuffers.length > 0) {
- data = Buffer.concat([this.readBuffers, data]);
- }
-
- // Keep processing while we have enough data
- let offset = 0;
-
- while (offset < data.length) {
- const remaining = data.length - offset;
-
- // Need at least 8 bytes for the header (4 bytes status + 4 bytes length)
- if (remaining < 8) {
- // Buffer the incomplete header and wait for more data
- this.waitingResponseEnd = true;
- this.readBuffers = data.subarray(offset);
- return;
- }
-
- // Read the header
- const responseSize = data.readUInt32LE(offset + 4);
- const totalSize = 8 + responseSize;
-
- // Check if we have the complete response (header + payload)
- if (remaining < totalSize) {
- // Buffer the incomplete response and wait for more data
- this.waitingResponseEnd = true;
- this.readBuffers = data.subarray(offset);
- return;
- }
-
- // We have a complete response, extract it and emit
- const response = data.subarray(offset, offset + totalSize);
- this.emit('response', response);
-
- // Move to the next response
- offset += totalSize;
+ const buffered = this.waitingResponseEnd && this.readBuffers.length > 0
+ ? Buffer.concat([this.readBuffers, data])
+ : data;
+
+ try {
+ const extracted = extractResponseFrames(
+ this.config.protocol ?? 'classic',
+ buffered,
+ this.config.maxResponseFrameSize ?? DEFAULT_MAX_RESPONSE_FRAME_SIZE
+ );
+ this.readBuffers = extracted.remainder;
+ this.waitingResponseEnd = extracted.remainder.length > 0;
Review Comment:
`waitingResponseEnd` is a pure derivative of `readBuffers.length > 0` - set
from `extracted.remainder.length > 0`, and only ever read alongside the same
condition. drop the field. moot anyway if the accumulator becomes a chunk list.
##########
foreign/node/src/wire/consumer-group/group.utils.ts:
##########
@@ -18,6 +18,13 @@
import { serializeIdentifier, type Id } from '../identifier.utils.js';
+/** Stream, topic, and consumer-group identifiers. */
+export type TargetGroup = {
Review Comment:
`TargetGroup` landed next to the four identical inline shapes instead of
replacing them - `join-group`, `leave-group`, `delete-group` and `get-group`
all spell out the same three fields. collapsing saves about 16 lines. keep the
old names as exported aliases (`export type JoinGroup = TargetGroup`) so it
isn't a breaking change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
