hubcio commented on code in PR #3763:
URL: https://github.com/apache/iggy/pull/3763#discussion_r3682350777
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -125,53 +130,124 @@ 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.readBuffers = Buffer.allocUnsafe(0);
+ this.connectPromise = undefined;
+ this.reconnectPromise = undefined;
+ this.responseDecoder = new ResponseFrameDecoder(
+ config.protocol ?? 'classic',
+ config.maxResponseFrameSize ?? DEFAULT_MAX_RESPONSE_FRAME_SIZE
+ );
+ 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', (data) => {
+ if (this.socket !== socket)
+ return;
+ if (!Buffer.isBuffer(data)) {
+ this.emit(
+ 'error',
+ new ProtocolFrameError('socket returned text instead of binary data')
+ );
+ socket.destroy();
+ return;
+ }
+ this._onData(data);
+ });
- this.socket.on('error', async (err: SocketError) => {
+ socket.on('error', (err: SocketError) => {
+ if (this.socket !== socket)
+ return;
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.emit('error', err);
+ });
- this.reconnect(err);
+ socket.once('connect', () => {
+ if (this.socket !== socket)
+ return;
+ debug('socket/connect event');
+ this.connected = true;
+ this.connecting = false;
+ this.reconnectCount = 0;
+ this.emit('connect');
});
- this.socket.once('end', async (hadError?: boolean) => {
- debug('socket/close#END event', hadError);
+ socket.once('close', (hadError?: boolean) => {
+ if (this.socket !== socket)
+ return;
+ debug('socket/close event', hadError);
this.connected = false;
+ this.connecting = false;
+ this.connectPromise = undefined;
+ this._endResponseWait();
this.emit('disconnected', hadError);
- this.reconnect();
+ if (!this.ending)
+ void this.reconnect();
Review Comment:
`void this.reconnect()` drops a promise that can reject. on re-entry
`reconnect()` returns `this.reconnectPromise` directly (line 264), and since it
is an `async` method the promise it hands back adopts the shared one, so it
rejects too. only the first caller wraps the await in try/catch (lines
282-292), so every re-entry through this close handler leaves an unhandled
rejection behind.
re-entry happens on each retry: `_reconnectUntilConnected` installs a new
socket, the dial fails, that socket's close handler lands here,
`reconnectPromise` is already set, and the adopted promise gets voided. with
the defaults at lines 76-80 (enabled, 5000 ms, 12 retries) an outage that
outlives the whole retry budget leaves 11 unhandled rejections, and node
terminates the process on the first one. the unit tests miss it because the
only dead-endpoint case (`client.socket.test.ts:380`) uses `maxRetries: 1`,
which never re-enters.
one-line fix: `void this.reconnect().catch(() => undefined);`
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -191,37 +272,116 @@ export class IggyConnection extends EventEmitter {
}
);
- if (!enabled || this.reconnectCount > maxRetries) {
- debug(`reconnect reached maxRetries of ${maxRetries}`, err);
- return this.emit(
- 'error',
- new Error(
- `reconnect maxRetries exceeded (count: ${this.reconnectCount})`,
- { cause: err }
- ));
+ const reconnectPromise = this._reconnectUntilConnected(
+ enabled,
+ interval,
+ maxRetries,
+ err
+ );
+ this.reconnectPromise = reconnectPromise;
+ try {
+ return await reconnectPromise;
+ } catch (error) {
+ if (!this.ending)
+ this.emit('error', error);
+ return;
+ } finally {
+ if (this.reconnectPromise === reconnectPromise)
+ this.reconnectPromise = undefined;
+ this.connecting = false;
}
+ }
- /** recreate socket */
- this.connecting = true;
- this.reconnectCount += 1;
- this.socket = await recreate(this.config, interval);
- this.connect();
+ private async _reconnectUntilConnected(
+ enabled: boolean,
+ interval: number,
+ maxRetries: number,
+ initialError?: Error
+ ): Promise<this> {
+ let lastError = initialError;
+ while (enabled && this.reconnectCount < maxRetries) {
+ this.connecting = true;
+ this.reconnectCount += 1;
+ const socketBeforeBackoff = this.socket;
+ await waitForReconnect(interval);
+ if (this.ending)
+ throw new Error('connection is closed', { cause: lastError });
+ // A redirect may replace the socket during the backoff. Defer to the
+ // active connection instead of dialing the superseded endpoint.
+ if (this.connected || this.socket !== socketBeforeBackoff)
+ return this.connect();
+
+ const socket = this._installSocket(getTransport(this.config));
+ this.socket = socket;
+ try {
+ return await this._waitForConnection(socket);
+ } catch (error) {
+ lastError = error instanceof Error
+ ? error
+ : new Error(String(error));
+ debug('reconnect attempt failed', lastError);
+ }
+ }
+
+ debug(`reconnect reached maxRetries of ${maxRetries}`, lastError);
+ throw new Error(
+ `reconnect maxRetries exceeded (count: ${this.reconnectCount})`,
+ { cause: lastError }
+ );
+ }
+
+ async redirect(host: string, port: number) {
+ const redirectedOptions = { ...this.config.options, host, port };
+ const redirectedConfig = {
+ ...this.config,
+ options: redirectedOptions
+ };
+ this.socket.removeAllListeners();
Review Comment:
`removeAllListeners()` also strips the listeners `_waitForConnection`
registered on this same socket, so a dial in flight never settles - and there
is no connect timeout anywhere, so that promise and everything awaiting it
hangs for good.
it also hides a second problem: the abandoned `_reconnectUntilConnected`
loop is not fenced to the socket it installed. today the orphaned await wedges
that loop so it cannot clobber the socket the redirect installs, but drop this
line alone and the clobber goes live. fix both together - remove the blanket
listener wipe (the `this.socket !== socket` guards already keep the old socket
inert, and the manual `disconnected` emit below stays), and have the loop bail
out when `this.socket` is no longer the socket it created.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -191,37 +272,116 @@ export class IggyConnection extends EventEmitter {
}
);
- if (!enabled || this.reconnectCount > maxRetries) {
- debug(`reconnect reached maxRetries of ${maxRetries}`, err);
- return this.emit(
- 'error',
- new Error(
- `reconnect maxRetries exceeded (count: ${this.reconnectCount})`,
- { cause: err }
- ));
+ const reconnectPromise = this._reconnectUntilConnected(
+ enabled,
+ interval,
+ maxRetries,
+ err
+ );
+ this.reconnectPromise = reconnectPromise;
+ try {
+ return await reconnectPromise;
+ } catch (error) {
+ if (!this.ending)
+ this.emit('error', error);
+ return;
+ } finally {
+ if (this.reconnectPromise === reconnectPromise)
+ this.reconnectPromise = undefined;
+ this.connecting = false;
}
+ }
- /** recreate socket */
- this.connecting = true;
- this.reconnectCount += 1;
- this.socket = await recreate(this.config, interval);
- this.connect();
+ private async _reconnectUntilConnected(
+ enabled: boolean,
+ interval: number,
+ maxRetries: number,
+ initialError?: Error
+ ): Promise<this> {
+ let lastError = initialError;
+ while (enabled && this.reconnectCount < maxRetries) {
+ this.connecting = true;
+ this.reconnectCount += 1;
+ const socketBeforeBackoff = this.socket;
+ await waitForReconnect(interval);
+ if (this.ending)
+ throw new Error('connection is closed', { cause: lastError });
+ // A redirect may replace the socket during the backoff. Defer to the
+ // active connection instead of dialing the superseded endpoint.
+ if (this.connected || this.socket !== socketBeforeBackoff)
+ return this.connect();
+
+ const socket = this._installSocket(getTransport(this.config));
+ this.socket = socket;
+ try {
+ return await this._waitForConnection(socket);
+ } catch (error) {
+ lastError = error instanceof Error
+ ? error
+ : new Error(String(error));
+ debug('reconnect attempt failed', lastError);
+ }
+ }
+
+ debug(`reconnect reached maxRetries of ${maxRetries}`, lastError);
+ throw new Error(
+ `reconnect maxRetries exceeded (count: ${this.reconnectCount})`,
+ { cause: lastError }
+ );
+ }
+
+ async redirect(host: string, port: number) {
+ const redirectedOptions = { ...this.config.options, host, port };
+ const redirectedConfig = {
+ ...this.config,
+ options: redirectedOptions
+ };
+ this.socket.removeAllListeners();
+ this.socket.destroy();
+ this.connected = false;
+ this.connecting = false;
+ this.connectPromise = undefined;
+ this.reconnectPromise = 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.socket = this._installSocket(getTransport(redirectedConfig));
+ await this.connect();
+ this.config.options = redirectedOptions;
Review Comment:
committing `config.options` after a successful redirect throws away the seed
endpoint for good. once the client has followed the leader here, `this.config`
only knows that leader, so when it dies `_reconnectUntilConnected` builds every
socket from `getTransport(this.config)` and dials the dead ex-leader until
maxRetries, then gives up - while the rest of the quorum is sitting there
healthy. that is the failover path vsr exists for.
keep the configured endpoint as a seed and fall back to it in the reconnect
loop (or hold a list and rotate) instead of overwriting the only endpoint the
client knows. moving the write after `connect()` fixed the failed-redirect half
of this, but the successful-redirect half is the one that costs you failover.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -191,37 +272,116 @@ export class IggyConnection extends EventEmitter {
}
);
- if (!enabled || this.reconnectCount > maxRetries) {
- debug(`reconnect reached maxRetries of ${maxRetries}`, err);
- return this.emit(
- 'error',
- new Error(
- `reconnect maxRetries exceeded (count: ${this.reconnectCount})`,
- { cause: err }
- ));
+ const reconnectPromise = this._reconnectUntilConnected(
+ enabled,
+ interval,
+ maxRetries,
+ err
+ );
+ this.reconnectPromise = reconnectPromise;
+ try {
+ return await reconnectPromise;
+ } catch (error) {
+ if (!this.ending)
+ this.emit('error', error);
+ return;
+ } finally {
+ if (this.reconnectPromise === reconnectPromise)
+ this.reconnectPromise = undefined;
+ this.connecting = false;
}
+ }
- /** recreate socket */
- this.connecting = true;
- this.reconnectCount += 1;
- this.socket = await recreate(this.config, interval);
- this.connect();
+ private async _reconnectUntilConnected(
+ enabled: boolean,
+ interval: number,
+ maxRetries: number,
+ initialError?: Error
+ ): Promise<this> {
+ let lastError = initialError;
+ while (enabled && this.reconnectCount < maxRetries) {
+ this.connecting = true;
+ this.reconnectCount += 1;
+ const socketBeforeBackoff = this.socket;
+ await waitForReconnect(interval);
+ if (this.ending)
+ throw new Error('connection is closed', { cause: lastError });
+ // A redirect may replace the socket during the backoff. Defer to the
+ // active connection instead of dialing the superseded endpoint.
+ if (this.connected || this.socket !== socketBeforeBackoff)
+ return this.connect();
+
+ const socket = this._installSocket(getTransport(this.config));
+ this.socket = socket;
+ try {
+ return await this._waitForConnection(socket);
+ } catch (error) {
+ lastError = error instanceof Error
+ ? error
+ : new Error(String(error));
+ debug('reconnect attempt failed', lastError);
+ }
+ }
+
+ debug(`reconnect reached maxRetries of ${maxRetries}`, lastError);
+ throw new Error(
+ `reconnect maxRetries exceeded (count: ${this.reconnectCount})`,
+ { cause: lastError }
+ );
+ }
+
+ async redirect(host: string, port: number) {
+ const redirectedOptions = { ...this.config.options, host, port };
+ const redirectedConfig = {
+ ...this.config,
+ options: redirectedOptions
+ };
+ this.socket.removeAllListeners();
+ this.socket.destroy();
+ this.connected = false;
+ this.connecting = false;
+ this.connectPromise = undefined;
+ this.reconnectPromise = 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.socket = this._installSocket(getTransport(redirectedConfig));
+ await this.connect();
+ this.config.options = redirectedOptions;
Review Comment:
two things to add to my earlier comment on this line.
both dial sites read the mutated config - `_reconnectUntilConnected` at line
314 and `connect()` at line 216 - and nothing routes around the dead endpoint:
`redirect()` is only reachable from `_ensureVsrLeader`, which needs a working
connection to fetch the cluster metadata first, and vsr pins the pool to a
single connection (`client.config.ts:54`) while `createPool` gets no `validate`
or eviction options (`client.ts:52`), so the broken client is never replaced.
also these retry dials feed the unhandled rejection flagged at line 196, so
with the defaults the process can die before the retry budget even runs out.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -125,53 +130,124 @@ 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.readBuffers = Buffer.allocUnsafe(0);
+ this.connectPromise = undefined;
+ this.reconnectPromise = undefined;
+ this.responseDecoder = new ResponseFrameDecoder(
+ config.protocol ?? 'classic',
+ config.maxResponseFrameSize ?? DEFAULT_MAX_RESPONSE_FRAME_SIZE
+ );
+ 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', (data) => {
+ if (this.socket !== socket)
+ return;
+ if (!Buffer.isBuffer(data)) {
+ this.emit(
+ 'error',
+ new ProtocolFrameError('socket returned text instead of binary data')
+ );
+ socket.destroy();
+ return;
+ }
+ this._onData(data);
+ });
- this.socket.on('error', async (err: SocketError) => {
+ socket.on('error', (err: SocketError) => {
+ if (this.socket !== socket)
+ return;
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.emit('error', err);
+ });
- this.reconnect(err);
+ socket.once('connect', () => {
+ if (this.socket !== socket)
+ return;
+ debug('socket/connect event');
+ this.connected = true;
+ this.connecting = false;
+ this.reconnectCount = 0;
+ this.emit('connect');
});
- this.socket.once('end', async (hadError?: boolean) => {
- debug('socket/close#END event', hadError);
+ socket.once('close', (hadError?: boolean) => {
+ if (this.socket !== socket)
+ return;
+ debug('socket/close event', hadError);
this.connected = false;
+ this.connecting = false;
+ this.connectPromise = undefined;
+ this._endResponseWait();
this.emit('disconnected', hadError);
- this.reconnect();
+ if (!this.ending)
+ void this.reconnect();
Review Comment:
correcting one clause in my earlier comment on this line - there are two
reconnect tests with a dead endpoint, and the one i named as the only case is
not the relevant one.
`client.connection.test.ts:242` (`maxRetries: 1`) points the *redirect
target* at a closed port, so the retry dial goes back to the live seed and
succeeds - no retry ever fails there. `client.socket.test.ts:380` (`maxRetries:
1`, server closed before the client connects) is the only test where a retry
dial actually fails, and with one retry it never re-enters. so the real gap is
that no test ever has two consecutive failed attempts, which is the minimum
this needs.
the rest of that comment stands: the leak count is `maxRetries - 1`, so 11
at the defaults.
--
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]
