This is an automated email from the ASF dual-hosted git repository. zstan pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/ignite-nodejs-thin-client.git
commit df0fe4cfdcf55278c07787d8cd456d29dd5fd00b Author: JohnSColeman <[email protected]> AuthorDate: Sun Apr 5 19:01:42 2026 +0700 IGNITE-14550 Fix four bugs in ClientSocket, Cursor and Errors --- spec/cache/ColdComplexRead.spec.js | 136 +++++++++++++++++++++++ spec/cache/ColdComplexReadDeadlock.spec.js | 171 +++++++++++++++++++++++++++++ src/Cursor.ts | 39 ++++++- src/Errors.ts | 6 +- src/internal/ClientSocket.ts | 111 ++++++++++++++++--- 5 files changed, 439 insertions(+), 24 deletions(-) diff --git a/spec/cache/ColdComplexRead.spec.js b/spec/cache/ColdComplexRead.spec.js new file mode 100644 index 0000000..348c920 --- /dev/null +++ b/spec/cache/ColdComplexRead.spec.js @@ -0,0 +1,136 @@ +/* + * 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); +} + +async function withTimeout(promise, ms, message) { + let timer; + const guard = new Promise((resolve, reject) => { + timer = setTimeout(() => reject(new Error(message)), ms); + }); + // try/finally rather than Promise.prototype.finally, which is Node 10+ + // (package.json declares engines.node >= 8.0.0). + try { + return await Promise.race([promise, guard]); + } finally { + clearTimeout(timer); + } +} + +describe('cold complex object read test suite >', () => { + let igniteClient = null; + + beforeAll((done) => { + Promise.resolve(). + then(async () => { + await TestingHelper.init(); + igniteClient = TestingHelper.igniteClient; + await igniteClient.destroyCache(CACHE_NAME).catch(() => {}); + await igniteClient.getOrCreateCache(CACHE_NAME); + }). + then(done). + catch(error => done.fail(error)); + }, TestingHelper.TIMEOUT); + + afterAll((done) => { + Promise.resolve(). + then(async () => { + if (igniteClient) { + await igniteClient.destroyCache(CACHE_NAME).catch(() => {}); + } + await TestingHelper.cleanUp(); + }). + then(done). + catch(error => done.fail(error)); + }, TestingHelper.TIMEOUT); + + it('fresh client reads a complex object it did not write without deadlock', (done) => { + let clientB = null; + Promise.resolve(). + then(async () => { + const key = 1; + const value = new ColdValue(); + value.id = 42; + value.name = 'cold-read'; + + // Writer client A: put registers the binary type in A's own type + // storage (addType) and on the server, but NOT in any other client. + const cacheA = igniteClient.getCache(CACHE_NAME). + setValueType(coldValueType()); + await cacheA.put(key, value); + + // Reader client B: a separate, freshly-connected client whose + // BinaryTypeStorage._types is empty, so reading the value forces a + // nested GET_BINARY_TYPE on B's single socket — the cold path that + // the existing suites never exercise (they put before they read on + // the same client, making getType a cache hit). + clientB = TestingHelper.makeClient(); + const endpoints = TestingHelper.getEndpoints(1); + await clientB.connect(new IgniteClientConfiguration(...endpoints). + setConnectionOptions(false, null, false)); + + const cacheB = clientB.getCache(CACHE_NAME). + setValueType(coldValueType()); + + const result = await withTimeout( + cacheB.get(key), + DEADLOCK_TIMEOUT_MS, + 'cache.get() of an uncached complex object did not resolve within ' + + DEADLOCK_TIMEOUT_MS + 'ms — the response processing queue deadlocked'); + + expect(result).not.toBeNull(); + expect(result.id).toBe(value.id); + expect(result.name).toBe(value.name); + }). + then(done). + catch(error => done.fail(error)). + then(async () => { + if (clientB) { + await clientB.disconnect(); + } + }); + }, TestingHelper.TIMEOUT); +}); diff --git a/spec/cache/ColdComplexReadDeadlock.spec.js b/spec/cache/ColdComplexReadDeadlock.spec.js new file mode 100644 index 0000000..176e0e0 --- /dev/null +++ b/spec/cache/ColdComplexReadDeadlock.spec.js @@ -0,0 +1,171 @@ +/* + * 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 }; +} + +async function withTimeout(promise, ms, message) { + let timer; + const guard = new Promise((resolve, reject) => { + timer = setTimeout(() => reject(new Error(message)), ms); + }); + // try/finally rather than Promise.prototype.finally, which is Node 10+ + // (package.json declares engines.node >= 8.0.0). + try { + return await Promise.race([promise, guard]); + } finally { + clearTimeout(timer); + } +} + +describe('cold complex object read deadlock (mock socket) test suite >', () => { + let origCreateConnection; + + beforeEach(() => { + origCreateConnection = net.createConnection; + }); + + afterEach(() => { + net.createConnection = origCreateConnection; + }); + + it('response whose payloadReader awaits a nested same-socket request resolves', (done) => { + let clientSocket = null; + Promise.resolve(). + then(async () => { + const mockSocket = new EventEmitter(); + mockSocket.end = () => {}; + + let handshakeSent = false; + mockSocket.write = (data) => { + if (!handshakeSent) { + // First write is the handshake (no opCode/id). + handshakeSent = true; + setImmediate(() => mockSocket.emit('data', buildHandshakeResponse())); + return true; + } + // Every subsequent request gets its reply on a LATER event loop + // tick — modelling the real "reply arrives as a future data event" + // ordering, including the nested request the outer reader issues. + const { opCode, requestId } = parseOutgoing(data); + if (opCode === OP_OUTER || opCode === OP_INNER) { + setImmediate(() => mockSocket.emit('data', buildResponse(requestId))); + } + return true; + }; + + net.createConnection = (options, onConnected) => { + setImmediate(onConnected); + return mockSocket; + }; + + const config = { options: {}, useTLS: false }; + const communicator = { readObject: async () => null }; + clientSocket = new ClientSocket( + '127.0.0.1:10800', config, communicator, () => {}, async () => {}); + + await withTimeout(clientSocket.connect(), DEADLOCK_TIMEOUT_MS, + 'handshake did not complete'); + + let nestedIssued = false; + let nestedResolved = false; + const outer = clientSocket.sendRequest( + OP_OUTER, + async () => {}, + async (buffer) => { + // Inline read of the outer response issues a nested request on + // this same socket and awaits it — exactly what readObject does + // for a COMPLEX_OBJECT whose binary type is not yet cached. + nestedIssued = true; + await clientSocket.sendRequest(OP_INNER, async () => {}, async () => {}); + nestedResolved = true; + }); + + await withTimeout(outer, DEADLOCK_TIMEOUT_MS, + 'outer request did not resolve within ' + DEADLOCK_TIMEOUT_MS + + 'ms — the response processing queue deadlocked on the nested ' + + 'same-socket request'); + + expect(nestedIssued).toBe(true); + expect(nestedResolved).toBe(true); + }). + then(done). + catch(error => done.fail(error)). + then(() => { + if (clientSocket) { + try { clientSocket.disconnect(); } catch (_e) { /* ignore */ } + } + }); + }); +}); diff --git a/src/Cursor.ts b/src/Cursor.ts index 30bdc0a..75ed342 100644 --- a/src/Cursor.ts +++ b/src/Cursor.ts @@ -17,7 +17,7 @@ 'use strict'; -const Long = require('long'); +import Long = require('long'); import BinaryUtils, { OPERATION } from './internal/BinaryUtils'; import BinaryCommunicator from "./internal/BinaryCommunicator"; import {PRIMITIVE_TYPE} from "./internal/Constants"; @@ -74,8 +74,28 @@ export abstract class BaseCursor<T> { * @return {boolean} - true if more cache entries are available, false otherwise. */ hasMore(): boolean { - return this._hasNext || - this._values && this._valueIndex < this._values.length; + if (this._hasNext) { + return true; + } + if (this._values != null && this._valueIndex < this._values.length) { + return true; + } + if (this._buffer != null) { + // Peek the buffered first page without consuming it. A page is laid out as + // [rowCount:int][rows...][hasNext:bool]; rowCount === 0 with a trailing + // hasNext === false is an empty result, so hasMore() must be false here. + const savedPosition = this._buffer.position; + try { + const rowCount = this._buffer.readInteger(); + return rowCount > 0 || this._buffer.readBoolean(); + } + finally { + // Restore even if a read throws on a short/truncated page, so the + // peek never advances the buffer position (keeps hasMore() total). + this._buffer.position = savedPosition; + } + } + return false; } /** @@ -160,8 +180,17 @@ 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. Return an empty + // array (not null) to honour the declared Promise<T[]> contract: an + // exhausted cursor has no more entries, not a missing collection. + // getValue() still returns null naturally (length 0) and hasMore() + // stays false, so old entries are never replayed. + this._values = []; + } return this._values; } diff --git a/src/Errors.ts b/src/Errors.ts index 30ee6f6..789767e 100644 --- a/src/Errors.ts +++ b/src/Errors.ts @@ -32,7 +32,7 @@ export class IgniteClientError extends Error { * @ignore */ static unsupportedTypeError(type) { - const BinaryUtils = require('./internal/BinaryUtils'); + const BinaryUtils = require('./internal/BinaryUtils').default; return new IgniteClientError(Util.format('Type %s is not supported', BinaryUtils.getTypeName(type))); } @@ -41,7 +41,7 @@ export class IgniteClientError extends Error { * @ignore */ static typeCastError(fromType, toType) { - const BinaryUtils = require('./internal/BinaryUtils'); + const BinaryUtils = require('./internal/BinaryUtils').default; return new IgniteClientError(Util.format('Type "%s" can not be cast to %s', BinaryUtils.getTypeName(fromType), BinaryUtils.getTypeName(toType))); } @@ -51,7 +51,7 @@ export class IgniteClientError extends Error { * @ignore */ static valueCastError(value, toType) { - const BinaryUtils = require('./internal/BinaryUtils'); + const BinaryUtils = require('./internal/BinaryUtils').default; return new IgniteClientError(Util.format('Value "%s" can not be cast to %s', value, BinaryUtils.getTypeName(toType))); } diff --git a/src/internal/ClientSocket.ts b/src/internal/ClientSocket.ts index e9a1074..5af139b 100644 --- a/src/internal/ClientSocket.ts +++ b/src/internal/ClientSocket.ts @@ -86,7 +86,6 @@ class ProtocolVersion { } } -const PROTOCOL_VERSION_1_0_0 = new ProtocolVersion(1, 0, 0); const PROTOCOL_VERSION_1_1_0 = new ProtocolVersion(1, 1, 0); const PROTOCOL_VERSION_1_2_0 = new ProtocolVersion(1, 2, 0); const PROTOCOL_VERSION_1_3_0 = new ProtocolVersion(1, 3, 0); @@ -136,6 +135,7 @@ export default class ClientSocket { private _protocolVersion: ProtocolVersion; private _port: number | string; private _version: number; + private _processingQueue: Promise<void>; constructor(endpoint: string, config: IgniteClientConfiguration, communicator: BinaryCommunicator, onSocketDisconnect: Function, onAffinityTopologyChange: Function) { ArgumentChecker.notEmpty(endpoint, 'endpoints'); @@ -158,6 +158,7 @@ export default class ClientSocket { this._error = null; this._nodeUuid = null; + this._processingQueue = Promise.resolve(); } async connect() { @@ -216,14 +217,19 @@ export default class ClientSocket { this._socket = net.createConnection(<NetConnectOpts>options, onConnected); } - this._socket.on('data', async (data: Buffer) => { - try { - await this._processResponse(data); - } - catch (err) { - this._error = err.message; - this._disconnect(); - } + // Serialize response processing — each 'data' event is chained onto the + // previous one so that _processResponse never runs concurrently. This + // protects the synchronous frame-splitting that mutates the shared + // _buffer/_offset state, and the awaited handshake finalize. Matched + // responses are finalized off this chain (see _processResponse) so that a + // payloadReader's nested same-socket request cannot deadlock the queue. + this._socket.on('data', (data: Buffer) => { + this._processingQueue = this._processingQueue + .then(() => this._processResponse(data)) + .catch(err => { + this._error = err.message; + this._disconnect(); + }); }); this._socket.on('close', () => { this._disconnect(false); @@ -266,6 +272,13 @@ export default class ClientSocket { while (this._buffer && this._offset < this._buffer.length) { const buffer = this._buffer; + + // Always start each message parse from the correct position. + // A previous payloadReader may have consumed only part of the prior + // message payload (e.g. a scan cursor only reads the cursor-ID), so + // buffer.position could be anywhere inside the previous message. + buffer.position = this._offset; + // Response length const length = buffer.readInteger() + BinaryUtils.getSize(BinaryUtils.TYPE_CODE.INTEGER); @@ -286,7 +299,14 @@ export default class ClientSocket { requestId = buffer.readLong().toString(); } - this._logMessage(requestId, false, buffer.getSlice(this._offset - length, length)); + // Record boundaries before the socket buffer is potentially cleared. + // getSlice(start, end) takes an end offset, not a length, so the message + // bytes are [msgStart, msgEnd); passing `length` as the end logs empty + // bytes for any 2nd+ frame in a segment (where msgStart > 0). + const msgEnd = this._offset; + const msgStart = msgEnd - length; + + this._logMessage(requestId, false, buffer.getSlice(msgStart, msgEnd)); if (this._offset === buffer.length) { this._buffer = null; @@ -296,15 +316,74 @@ 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 => { + // `request` was already removed from _requests above, so the + // _disconnect() below cannot reject it. _finalizeResponse has + // throw paths outside its own try/catch (the header reads and + // _onAffinityTopologyChange), so an error here would otherwise + // leave the caller awaiting this request forever. Reject it + // explicitly first. If _finalizeResponse already settled the + // request, this reject is a harmless no-op. + request.reject(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); } } } @@ -350,11 +429,11 @@ export default class ClientSocket { } async _finalizeResponse(buffer: MessageBuffer, request: Request) { - let statusCode, isSuccess; + let isSuccess; if (this._protocolVersion.compareTo(PROTOCOL_VERSION_1_4_0) < 0) { // Check status code - statusCode = buffer.readInteger(); + const statusCode = buffer.readInteger(); isSuccess = statusCode === REQUEST_SUCCESS_STATUS_CODE; } else { @@ -368,7 +447,7 @@ export default class ClientSocket { } if (!isSuccess) { - statusCode = buffer.readInteger(); + buffer.readInteger(); // advance past status code; error detail is in the message string } }
