JohnSColeman commented on code in PR #10:
URL:
https://github.com/apache/ignite-nodejs-thin-client/pull/10#discussion_r3540701598
##########
package.json:
##########
@@ -1,6 +1,6 @@
{
"name": "apache-ignite-client",
- "version": "1.1.0",
+ "version": "1.1.1",
Review Comment:
Dropped the version bump — reverted `package.json` to 1.1.0 in ac6092a so
this PR is scoped to the fixes. The version change can be proposed and
discussed separately on the dev list.
##########
src/Cursor.ts:
##########
@@ -160,8 +180,15 @@ export abstract class BaseCursor<T> {
if (!this._buffer && this._hasNext) {
await this._getNext();
}
- await this._read(this._buffer)
- this._buffer = null;
+ if (this._buffer) {
+ await this._read(this._buffer);
+ this._buffer = null;
+ } else {
+ // No buffer and no next page — cursor is exhausted.
+ // Set _values to null so getValue() returns null and
+ // hasMore() returns false rather than replaying old entries.
+ this._values = null;
+ }
Review Comment:
Fixed in bfcf03d — `_getValues()` now returns an empty array on exhaustion
instead of `null`, honouring the declared `Promise<T[]>` (and the generated
`.d.ts`). `getValue()` still returns `null` naturally (length 0) and
`hasMore()` stays `false`, so exhausted cursors do not replay entries.
##########
src/internal/ClientSocket.ts:
##########
@@ -296,15 +316,66 @@ export default class ClientSocket {
if (this._requests.has(requestId)) {
const request = this._requests.get(requestId);
this._requests.delete(requestId);
+
+ // Carve a fresh, independent MessageBuffer from just this
message's
+ // payload bytes (after length field + request-id). getSlice()
returns
+ // a view over the shared socket buffer, but
MessageBuffer.from() copies
+ // those bytes (via Buffer.from), so freshBuffer owns an
independent
+ // buffer with its own position pointer. That independence
prevents two
+ // cursors created from the same TCP segment from aliasing the
same
+ // position and corrupting each other's reads under parallel
scan
+ // workloads. Built only on the matched-request path so
unmatched frames
+ // cost no copy.
+ const headerConsumed = isHandshake
+ ? BinaryUtils.getSize(BinaryUtils.TYPE_CODE.INTEGER)
// 4 B: length only
+ : BinaryUtils.getSize(BinaryUtils.TYPE_CODE.INTEGER) +
// 4 B: length
+ BinaryUtils.getSize(BinaryUtils.TYPE_CODE.LONG);
// 8 B: request-id
+ const freshBuffer = MessageBuffer.from(
+ buffer.getSlice(msgStart + headerConsumed, msgEnd),
+ 0
+ );
+
if (isHandshake) {
- await this._finalizeHandshake(buffer, request);
+ // Handshake is single-in-flight, transitions _state and
issues no
+ // nested request, so it is safe to await inline on the
queue.
+ await this._finalizeHandshake(freshBuffer, request);
}
else {
- await this._finalizeResponse(buffer, request);
+ // Do NOT await on the processing queue: a payloadReader
may issue a
+ // nested request on this same socket and await its reply
(e.g.
+ // GET_BINARY_TYPE when reading a COMPLEX_OBJECT whose
type is not yet
+ // cached in this client's BinaryTypeStorage). That reply
arrives as a
+ // later 'data' event chained behind this very queue
entry, so awaiting
+ // here would deadlock — the entry can only complete once
the reply is
+ // processed, but the reply can only be processed by a
later entry.
+ // freshBuffer is an independent copy of this message's
payload, so
+ // finalizing it off the parse chain cannot corrupt
_buffer/_offset.
+ // With finalize detached the CONNECTED path of
_processResponse has no
+ // remaining await and runs to completion synchronously,
so two
+ // invocations still cannot interleave on _buffer/_offset
and the parse
+ // race stays closed.
+ this._finalizeResponse(freshBuffer, request).catch(err => {
+ this._error = err.message;
+ this._disconnect();
+ });
}
}
else {
- throw IgniteClientError.internalError('Invalid response id: '
+ requestId);
+ // No pending request matches this response id. At the
protocol version
+ // this client negotiates (<= 1.4.0) the server never sends
unsolicited
+ // frames: affinity-topology updates ride on response flags
(handled in
+ // _finalizeResponse), and notification / heartbeat frames
only exist in
+ // later protocol versions this client does not speak.
Requests are also
+ // never removed while still awaiting a response (there is no
client-side
+ // timeout), so an unmatched id cannot be a late or duplicate
reply.
+ // It therefore means the response byte stream has desynced,
after which
+ // every subsequent frame is garbage and the originating
request would
+ // otherwise hang forever. Fail fast: throwing propagates to
the socket
+ // 'data' handler's catch, which disconnects and rejects all
pending
+ // requests with LostConnectionError so callers can recover
instead of
+ // hanging.
+ throw IgniteClientError.internalError(
+ 'Response stream desync: received a frame with unmatched
request id ' + requestId);
Review Comment:
Aligned the PR description with the implemented behaviour: the
unmatched-`requestId` path fails fast (throws → disconnect → rejects pending
requests), and the description now documents that fail-fast behaviour rather
than "log and discard".
--
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]
