JohnSColeman commented on code in PR #10:
URL:
https://github.com/apache/ignite-nodejs-thin-client/pull/10#discussion_r3742478568
##########
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();
+ });
Review Comment:
Good catch — fixed in 85ff8ca. You're right: `request` is removed from
`_requests` before the detached `_finalizeResponse` runs, so `_disconnect()` in
the catch can't reject it, and `_finalizeResponse` has throw paths *outside*
its own try/catch (the pre-`try` header reads
`readShort`/`readInteger`/`readString`, and the awaited
`_onAffinityTopologyChange`). On any of those a throw would leave the caller
awaiting the request forever.
The catch now rejects the request before disconnecting:
```ts
this._finalizeResponse(freshBuffer, request).catch(err => {
request.reject(err); // request already removed from _requests;
_disconnect() can't reject it
this._error = err.message;
this._disconnect();
});
```
`request.reject` is the raw Promise reject fn, so if `_finalizeResponse`
already settled the request on its handled paths (`!isSuccess`, or the
payloadReader catch) this is a harmless no-op. Verified: `ColdComplexRead`
(live) and `ColdComplexReadDeadlock` (mock socket) specs both still pass.
##########
spec/cache/ColdComplexReadDeadlock.spec.js:
##########
@@ -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.
+ */
+
+'use strict';
+
+require('jasmine-expect');
+
+const net = require('net');
+const EventEmitter = require('events');
+const Long = require('long');
+
+// Drive the REAL ClientSocket against a mock TCP socket so this regression is
+// deterministic and needs no cluster. It reproduces the exact async topology
of
+// the cold complex-object read: a single socket, one response whose
payloadReader
+// issues a nested request on that same socket and awaits its reply, where the
+// reply only arrives as a later 'data' event. If response finalization is
awaited
+// on the socket's serialized processing queue, the nested reply is chained
behind
+// the still-pending outer entry and can never be processed -> the outer
request
+// hangs forever. With finalization dispatched off the queue, it resolves.
+const ClientSocket =
require('apache-ignite-client/dist/internal/ClientSocket').default;
+const MessageBuffer =
require('apache-ignite-client/dist/internal/MessageBuffer').default;
+
+const HANDSHAKE_SUCCESS_STATUS_CODE = 1;
+const OP_OUTER = 2001;
+const OP_INNER = 2002;
+const DEADLOCK_TIMEOUT_MS = 5000;
+
+// Handshake reply frame: [length:int][status:byte]. The >= 1.4.0 path then
reads a
+// node-UUID via communicator.readObject, which the mock communicator stubs
out, so
+// no UUID bytes are required (frames are length-delimited, not reader-position
+// delimited).
+function buildHandshakeResponse() {
+ const buf = new MessageBuffer();
+ buf.position = 4;
+ buf.writeByte(HANDSHAKE_SUCCESS_STATUS_CODE);
+ const len = buf.length - 4;
+ buf.position = 0;
+ buf.writeInteger(len);
+ return buf.data;
+}
+
+// Success response frame for protocol >= 1.4.0:
[length:int][requestId:long][flags:short=0].
+function buildResponse(requestId) {
+ const buf = new MessageBuffer();
+ buf.position = 4;
+ buf.writeLong(requestId);
+ buf.writeShort(0);
+ const len = buf.length - 4;
+ buf.position = 0;
+ buf.writeInteger(len);
+ return buf.data;
+}
+
+// Outgoing request frame layout (see ClientSocket Request.getMessage):
+// [length:int][opCode:short][requestId:long][payload].
+function parseOutgoing(data) {
+ const buf = MessageBuffer.from(data, 0);
+ buf.readInteger();
+ const opCode = buf.readShort();
+ const requestId = buf.readLong();
+ return { opCode, requestId };
+}
+
+function withTimeout(promise, ms, message) {
+ let timer;
+ const guard = new Promise((resolve, reject) => {
+ timer = setTimeout(() => reject(new Error(message)), ms);
+ });
+ return Promise.race([promise, guard]).finally(() => clearTimeout(timer));
Review Comment:
Fixed in 85ff8ca. `Promise.prototype.finally` is Node 10+ while
`package.json` declares `engines.node >= 8.0.0`, so `withTimeout` is now
`async` with a plain `try/finally` — `clearTimeout` still runs on both resolve
and reject:
```js
async function withTimeout(promise, ms, message) {
let timer;
const guard = new Promise((resolve, reject) => {
timer = setTimeout(() => reject(new Error(message)), ms);
});
try {
return await Promise.race([promise, guard]);
} finally {
clearTimeout(timer);
}
}
```
Applied identically here and in `ColdComplexRead.spec.js` (the only two
`.finally` uses in the PR). Both specs still pass.
##########
spec/cache/ColdComplexRead.spec.js:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.
+ */
+
+'use strict';
+
+require('jasmine-expect');
+
+const TestingHelper = require('../TestingHelper');
+const {
+ IgniteClientConfiguration, ObjectType, ComplexObjectType
+} = require('apache-ignite-client');
+
+const CACHE_NAME = '__test_cold_complex_read';
+
+// Deadlock guard: a fresh client reading a complex object whose binary type is
+// not yet in its own BinaryTypeStorage must issue a nested GET_BINARY_TYPE on
the
+// same socket from inside the get() payloadReader. If response finalization is
+// awaited on the socket processing queue, that nested reply can never be
parsed
+// and get() hangs forever, so we race it against an explicit timeout.
+const DEADLOCK_TIMEOUT_MS = 10000;
+
+class ColdValue {
+ constructor() {
+ this.id = null;
+ this.name = null;
+ }
+}
+
+function coldValueType() {
+ return new ComplexObjectType(new ColdValue(), 'ColdValue').
+ setFieldType('id', ObjectType.PRIMITIVE_TYPE.INTEGER).
+ setFieldType('name', ObjectType.PRIMITIVE_TYPE.STRING);
+}
+
+function withTimeout(promise, ms, message) {
+ let timer;
+ const guard = new Promise((resolve, reject) => {
+ timer = setTimeout(() => reject(new Error(message)), ms);
+ });
+ return Promise.race([promise, guard]).finally(() => clearTimeout(timer));
Review Comment:
Fixed in 85ff8ca — same change as the `ColdComplexReadDeadlock.spec.js`
thread: `withTimeout` is now `async` + `try/finally` instead of
`Promise.prototype.finally` (Node 10+), honouring `engines.node >= 8.0.0`.
`clearTimeout` still runs on both resolve and reject. Spec still passes.
--
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]
