From 3265a151fd6ee92233388a3dceabad8cc8f58210 Mon Sep 17 00:00:00 2001 From: Minggang Wang Date: Tue, 27 Oct 2026 16:06:28 +0800 Subject: [PATCH 1/7] Preserve ready callbacks across errors and invalidate stopped batches --- lib/node.js | 407 ++++++++++++++++++++++++------------------ test/test-spinning.js | 271 ++++++++++++++++++++++++++++ types/node.d.ts | 4 + 3 files changed, 507 insertions(+), 175 deletions(-) diff --git a/lib/node.js b/lib/node.js index c3321b02..08d6a4d9 100644 --- a/lib/node.js +++ b/lib/node.js @@ -163,6 +163,7 @@ class Node extends rclnodejs.ShadowNode { this._postSetParametersCallbacks = []; this._logger = new Logging(rclnodejs.getNodeLoggerName(this.handle)); this._spinning = false; + this._dispatchGeneration = 0; this._enableRosout = options.enableRosout; if (this._enableRosout) { @@ -231,6 +232,16 @@ class Node extends rclnodejs.ShadowNode { } execute(handles) { + const dispatchGeneration = this._dispatchGeneration; + const callbackErrors = []; + const dispatch = (callback) => { + if (this._dispatchGeneration !== dispatchGeneration) return; + try { + callback(); + } catch (error) { + callbackErrors.push(error); + } + }; let timersReady = this._timers.filter((timer) => handles.includes(timer.handle) ); @@ -256,104 +267,112 @@ class Node extends rclnodejs.ShadowNode { handles.includes(event.handle) ); - timersReady.forEach((timer) => { - // Earlier callbacks may destroy timers or the node after this snapshot. - if (this._timers.includes(timer) && timer.isReady()) { - let timerInfo; - if (typeof rclnodejs.callTimerWithInfo === 'function') { - timerInfo = rclnodejs.callTimerWithInfo(timer.handle); - timer.callback(timerInfo); - } else { - rclnodejs.callTimer(timer.handle); - timer.callback(); + timersReady.forEach((timer) => + dispatch(() => { + // Earlier callbacks may destroy timers or the node after this snapshot. + if (this._timers.includes(timer) && timer.isReady()) { + let timerInfo; + if (typeof rclnodejs.callTimerWithInfo === 'function') { + timerInfo = rclnodejs.callTimerWithInfo(timer.handle); + timer.callback(timerInfo); + } else { + rclnodejs.callTimer(timer.handle); + timer.callback(); + } } - } - }); + }) + ); - eventsReady.forEach((event) => { - event.takeData(); - }); + eventsReady.forEach((event) => dispatch(() => event.takeData())); for (const subscription of subscriptionsReady) { if (subscription.isDestroyed()) continue; if (subscription.isRaw) { - if (subscription.wantsMessageInfo) { - let taken = rclnodejs.rclTakeRawWithInfo(subscription.handle); - if (taken) { - subscription.processResponse( - taken.buffer, - new MessageInfo(taken.info) - ); - } - } else { - let rawMessage = rclnodejs.rclTakeRaw(subscription.handle); - if (rawMessage) { - subscription.processResponse(rawMessage); - } - } - continue; - } - - this._runWithMessageType( - subscription.typeClass, - (message, deserialize) => { + dispatch(() => { if (subscription.wantsMessageInfo) { - let rawInfo = rclnodejs.rclTakeWithInfo( - subscription.handle, - message - ); - if (rawInfo) { + let taken = rclnodejs.rclTakeRawWithInfo(subscription.handle); + if (taken) { subscription.processResponse( - deserialize(), - new MessageInfo(rawInfo) + taken.buffer, + new MessageInfo(taken.info) ); } } else { - let success = rclnodejs.rclTake(subscription.handle, message); - if (success) { - subscription.processResponse(deserialize()); + let rawMessage = rclnodejs.rclTakeRaw(subscription.handle); + if (rawMessage) { + subscription.processResponse(rawMessage); } } - } + }); + continue; + } + + dispatch(() => + this._runWithMessageType( + subscription.typeClass, + (message, deserialize) => { + if (subscription.wantsMessageInfo) { + let rawInfo = rclnodejs.rclTakeWithInfo( + subscription.handle, + message + ); + if (rawInfo) { + subscription.processResponse( + deserialize(), + new MessageInfo(rawInfo) + ); + } + } else { + let success = rclnodejs.rclTake(subscription.handle, message); + if (success) { + subscription.processResponse(deserialize()); + } + } + } + ) ); } for (const guard of guardsReady) { if (guard.isDestroyed()) continue; - guard.callback(); + dispatch(() => guard.callback()); } for (const client of clientsReady) { if (client.isDestroyed()) continue; - this._runWithMessageType( - client.typeClass.Response, - (message, deserialize) => { - let sequenceNumber = rclnodejs.rclTakeResponse( - client.handle, - message - ); - if (sequenceNumber !== undefined) { - client.processResponse(sequenceNumber, deserialize()); + dispatch(() => + this._runWithMessageType( + client.typeClass.Response, + (message, deserialize) => { + let sequenceNumber = rclnodejs.rclTakeResponse( + client.handle, + message + ); + if (sequenceNumber !== undefined) { + client.processResponse(sequenceNumber, deserialize()); + } } - } + ) ); } for (const service of servicesReady) { if (service.isDestroyed()) continue; - this._runWithMessageType( - service.typeClass.Request, - (message, deserialize) => { - let header = rclnodejs.rclTakeRequest( - service.handle, - this.handle, - message - ); - if (header) { - service.processRequest(header, deserialize()); + dispatch(() => + this._runWithMessageType( + service.typeClass.Request, + (message, deserialize) => { + let header = rclnodejs.rclTakeRequest( + service.handle, + this.handle, + message + ); + if (header) { + service.processRequest(header, deserialize()); + } } - } + ) ); } @@ -363,77 +382,87 @@ class Node extends rclnodejs.ShadowNode { const properties = actionClient.handle.properties; if (properties.isGoalResponseReady) { - this._runWithMessageType( - actionClient.typeClass.impl.SendGoalService.Response, - (message, deserialize) => { - let sequence = rclnodejs.actionTakeGoalResponse( - actionClient.handle, - message - ); - if (sequence != undefined) { - actionClient.processGoalResponse(sequence, deserialize()); + dispatch(() => + this._runWithMessageType( + actionClient.typeClass.impl.SendGoalService.Response, + (message, deserialize) => { + let sequence = rclnodejs.actionTakeGoalResponse( + actionClient.handle, + message + ); + if (sequence != undefined) { + actionClient.processGoalResponse(sequence, deserialize()); + } } - } + ) ); } - if (properties.isCancelResponseReady) { - this._runWithMessageType( - actionClient.typeClass.impl.CancelGoal.Response, - (message, deserialize) => { - let sequence = rclnodejs.actionTakeCancelResponse( - actionClient.handle, - message - ); - if (sequence != undefined) { - actionClient.processCancelResponse(sequence, deserialize()); + if (properties.isCancelResponseReady && !actionClient.isDestroyed()) { + dispatch(() => + this._runWithMessageType( + actionClient.typeClass.impl.CancelGoal.Response, + (message, deserialize) => { + let sequence = rclnodejs.actionTakeCancelResponse( + actionClient.handle, + message + ); + if (sequence != undefined) { + actionClient.processCancelResponse(sequence, deserialize()); + } } - } + ) ); } - if (properties.isResultResponseReady) { - this._runWithMessageType( - actionClient.typeClass.impl.GetResultService.Response, - (message, deserialize) => { - let sequence = rclnodejs.actionTakeResultResponse( - actionClient.handle, - message - ); - if (sequence != undefined) { - actionClient.processResultResponse(sequence, deserialize()); + if (properties.isResultResponseReady && !actionClient.isDestroyed()) { + dispatch(() => + this._runWithMessageType( + actionClient.typeClass.impl.GetResultService.Response, + (message, deserialize) => { + let sequence = rclnodejs.actionTakeResultResponse( + actionClient.handle, + message + ); + if (sequence != undefined) { + actionClient.processResultResponse(sequence, deserialize()); + } } - } + ) ); } - if (properties.isFeedbackReady) { - this._runWithMessageType( - actionClient.typeClass.impl.FeedbackMessage, - (message, deserialize) => { - let success = rclnodejs.actionTakeFeedback( - actionClient.handle, - message - ); - if (success) { - actionClient.processFeedbackMessage(deserialize()); + if (properties.isFeedbackReady && !actionClient.isDestroyed()) { + dispatch(() => + this._runWithMessageType( + actionClient.typeClass.impl.FeedbackMessage, + (message, deserialize) => { + let success = rclnodejs.actionTakeFeedback( + actionClient.handle, + message + ); + if (success) { + actionClient.processFeedbackMessage(deserialize()); + } } - } + ) ); } - if (properties.isStatusReady) { - this._runWithMessageType( - actionClient.typeClass.impl.GoalStatusArray, - (message, deserialize) => { - let success = rclnodejs.actionTakeStatus( - actionClient.handle, - message - ); - if (success) { - actionClient.processStatusMessage(deserialize()); + if (properties.isStatusReady && !actionClient.isDestroyed()) { + dispatch(() => + this._runWithMessageType( + actionClient.typeClass.impl.GoalStatusArray, + (message, deserialize) => { + let success = rclnodejs.actionTakeStatus( + actionClient.handle, + message + ); + if (success) { + actionClient.processStatusMessage(deserialize()); + } } - } + ) ); } } @@ -444,71 +473,89 @@ class Node extends rclnodejs.ShadowNode { const properties = actionServer.handle.properties; if (properties.isGoalRequestReady) { - this._runWithMessageType( - actionServer.typeClass.impl.SendGoalService.Request, - (message, deserialize) => { - const result = rclnodejs.actionTakeGoalRequest( - actionServer.handle, - message - ); - if (result) { - actionServer.processGoalRequest(result, deserialize()); + dispatch(() => + this._runWithMessageType( + actionServer.typeClass.impl.SendGoalService.Request, + (message, deserialize) => { + const result = rclnodejs.actionTakeGoalRequest( + actionServer.handle, + message + ); + if (result) { + actionServer.processGoalRequest(result, deserialize()); + } } - } + ) ); } - if (properties.isCancelRequestReady) { - this._runWithMessageType( - actionServer.typeClass.impl.CancelGoal.Request, - (message, deserialize) => { - const result = rclnodejs.actionTakeCancelRequest( - actionServer.handle, - message - ); - if (result) { - actionServer.processCancelRequest(result, deserialize()); + if (properties.isCancelRequestReady && !actionServer.isDestroyed()) { + dispatch(() => + this._runWithMessageType( + actionServer.typeClass.impl.CancelGoal.Request, + (message, deserialize) => { + const result = rclnodejs.actionTakeCancelRequest( + actionServer.handle, + message + ); + if (result) { + actionServer.processCancelRequest(result, deserialize()); + } } - } + ) ); } - if (properties.isResultRequestReady) { - this._runWithMessageType( - actionServer.typeClass.impl.GetResultService.Request, - (message, deserialize) => { - const result = rclnodejs.actionTakeResultRequest( - actionServer.handle, - message - ); - if (result) { - actionServer.processResultRequest(result, deserialize()); + if (properties.isResultRequestReady && !actionServer.isDestroyed()) { + dispatch(() => + this._runWithMessageType( + actionServer.typeClass.impl.GetResultService.Request, + (message, deserialize) => { + const result = rclnodejs.actionTakeResultRequest( + actionServer.handle, + message + ); + if (result) { + actionServer.processResultRequest(result, deserialize()); + } } - } + ) ); } - if (properties.isGoalExpired) { - let numGoals = actionServer._goalHandles.size; - if (numGoals > 0) { - let GoalInfoArray = ActionInterfaces.GoalInfo.ArrayType; - let message = new GoalInfoArray(numGoals); - let count = rclnodejs.actionExpireGoals( - actionServer.handle, - numGoals, - message._refArray.buffer - ); - if (count > 0) { - actionServer.processGoalExpired(message, count); + if (properties.isGoalExpired && !actionServer.isDestroyed()) { + dispatch(() => { + let numGoals = actionServer._goalHandles.size; + if (numGoals > 0) { + let GoalInfoArray = ActionInterfaces.GoalInfo.ArrayType; + let message = new GoalInfoArray(numGoals); + try { + let count = rclnodejs.actionExpireGoals( + actionServer.handle, + numGoals, + message._refArray.buffer + ); + if (count > 0) { + actionServer.processGoalExpired(message, count); + } + } finally { + GoalInfoArray.freeArray(message); + } } - GoalInfoArray.freeArray(message); - } + }); } } // At this point it is safe to clear the cache of any // destroyed entity references Entity._gcHandles(); + if (callbackErrors.length === 1) throw callbackErrors[0]; + if (callbackErrors.length > 1) { + throw new AggregateError( + callbackErrors, + 'Multiple executor callbacks failed' + ); + } } /** @@ -522,6 +569,8 @@ class Node extends rclnodejs.ShadowNode { /** * Trigger the event loop to continuously check for and route. * incoming events. + * Callback errors are reported after the current ready batch is processed. + * Multiple callback errors in a batch are reported as an AggregateError. * @param {Node} node - The node to be spun up. * @param {number} [timeout=10] - Timeout to wait in milliseconds. Block forever if negative. Don't wait if 0. * @throws {Error} If the node is already spinning or native executor cleanup is still in progress. @@ -553,6 +602,7 @@ class Node extends rclnodejs.ShadowNode { * @returns {undefined} */ stop() { + this._dispatchGeneration++; try { super.stop(); } finally { @@ -566,6 +616,7 @@ class Node extends rclnodejs.ShadowNode { * @deprecated since 0.18.0, Use stop(). */ stopSpinning() { + this._dispatchGeneration++; try { super.stop(); } finally { @@ -576,6 +627,8 @@ class Node extends rclnodejs.ShadowNode { /** * Spin the node and trigger the event loop to check for one incoming event. Thereafter the node * will not received additional events until running additional calls to spin() or spinOnce(). + * Callback errors are thrown after the current ready batch is processed. + * Multiple callback errors in a batch are thrown as an AggregateError. * @param {Node} node - The node to be spun. * @param {number} [timeout=10] - Timeout to wait in milliseconds. Block forever if negative. Don't wait if 0. * @throws {Error} If the node is already spinning or native executor cleanup is still in progress. @@ -1205,6 +1258,7 @@ class Node extends rclnodejs.ShadowNode { * @return {undefined} */ destroy() { + this._dispatchGeneration++; if (this.spinning) { this.stop(); } @@ -2603,15 +2657,18 @@ class Node extends rclnodejs.ShadowNode { */ _runWithMessageType(Type, callback) { let message = new Type(); + let rawMessage = message.toRawROS(); - callback(message.toRawROS(), () => { - let result = new Type(); - result.deserialize(message.refObject); - - return result; - }); + try { + callback(rawMessage, () => { + let result = new Type(); + result.deserialize(message.refObject); - Type.destroyRawROS(message); + return result; + }); + } finally { + Type.destroyRawROS(message); + } } _addActionClient(actionClient) { diff --git a/test/test-spinning.js b/test/test-spinning.js index da68355a..114d1d91 100644 --- a/test/test-spinning.js +++ b/test/test-spinning.js @@ -12,7 +12,9 @@ import assert from 'assert'; import { execFile } from 'node:child_process'; +import sinon from 'sinon'; import rclnodejs from '../index.js'; +import nativeLoader from '../lib/native_loader.js'; describe('Spin testing', function () { var node; @@ -193,6 +195,84 @@ describe('Spin testing', function () { } } + it('releases raw messages when callbacks throw', function () { + const failure = new Error('message callback failed'); + let destroyedMessage; + class Message { + toRawROS() { + return {}; + } + + static destroyRawROS(message) { + destroyedMessage = message; + } + } + assert.throws( + () => + node._runWithMessageType(Message, () => { + throw failure; + }), + (error) => error === failure + ); + assert.ok(destroyedMessage instanceof Message); + }); + + for (const actionKind of ['client', 'server']) { + for (const destroyOnError of [false, true]) { + it(`handles ready action ${actionKind} events after ${destroyOnError ? 'destruction and ' : ''}an error`, function () { + const sandbox = sinon.createSandbox(); + const failure = new Error('action callback failed'); + const calls = []; + const messageKind = actionKind === 'client' ? 'Response' : 'Request'; + const collection = + actionKind === 'client' ? node._actionClients : node._actionServers; + let destroyed = false; + const action = { + handle: { + properties: { + [`isGoal${messageKind}Ready`]: true, + [`isCancel${messageKind}Ready`]: true, + }, + }, + typeClass: { + impl: { + SendGoalService: { [messageKind]: Object }, + CancelGoal: { [messageKind]: Object }, + }, + }, + isDestroyed: () => destroyed, + [`processGoal${messageKind}`]: () => { + calls.push('goal'); + destroyed = destroyOnError; + throw failure; + }, + [`processCancel${messageKind}`]: () => calls.push('cancel'), + }; + collection.push(action); + try { + sandbox.stub(nativeLoader, `actionTakeGoal${messageKind}`).returns(1); + sandbox + .stub(nativeLoader, `actionTakeCancel${messageKind}`) + .returns(1); + sandbox + .stub(node, '_runWithMessageType') + .callsFake((type, callback) => callback({}, () => ({}))); + assert.throws( + () => node.execute([action.handle]), + (error) => error === failure + ); + assert.deepEqual( + calls, + destroyOnError ? ['goal'] : ['goal', 'cancel'] + ); + } finally { + collection.splice(collection.indexOf(action), 1); + sandbox.restore(); + } + }); + } + } + describe('Interruptible spinning', function () { this.timeout(15000); @@ -593,6 +673,38 @@ describe('Spin testing', function () { }); } + it('does not execute an earlier ready batch after an immediate restart', function (done) { + runInChild( + ` + const calls = []; + let restarted; + const afterRestart = new Promise((resolve) => { restarted = resolve; }); + let complete; + const completed = new Promise((resolve) => { complete = resolve; }); + const first = node.createGuardCondition(() => { + calls.push('first'); + node.stop(); + node.spin(-1); + setImmediate(restarted); + }); + const second = node.createGuardCondition(() => { + calls.push('second'); + complete(); + }); + first.trigger(); + second.trigger(); + node.spin(-1); + await afterRestart; + assert.deepEqual(calls, ['first']); + second.trigger(); + await completed; + node.stop(); + assert.deepEqual(calls, ['first', 'second']); + `, + done + ); + }); + it('stops and restarts without pumping unrelated JavaScript callbacks', function (done) { runInChild( ` @@ -673,6 +785,165 @@ describe('Spin testing', function () { done ); }); + + for (const multipleErrors of [false, true]) { + it(`${spinMethod} preserves ready guards after ${multipleErrors ? 'multiple errors' : 'an error'}`, function (done) { + runInChild( + ` + const firstError = new Error('first guard failure'); + const secondError = new Error('second guard failure'); + const calls = []; + const first = node.createGuardCondition(() => { + calls.push('first'); + throw firstError; + }); + const second = node.createGuardCondition(() => { + calls.push('second'); + if (${multipleErrors}) throw secondError; + }); + const third = node.createGuardCondition(() => calls.push('third')); + first.trigger(); + second.trigger(); + third.trigger(); + let error; + if (${JSON.stringify(spinMethod)} === 'spin') { + const reported = new Promise((resolve) => { + process.once('uncaughtException', resolve); + }); + node.spin(-1); + error = await reported; + node.stop(); + } else { + try { node.spinOnce(0); } catch (failure) { error = failure; } + } + assert.deepEqual(calls, ['first', 'second', 'third']); + if (${multipleErrors}) { + assert.ok(error instanceof AggregateError); + assert.deepEqual(error.errors, [firstError, secondError]); + } else { + assert.equal(error, firstError); + } + `, + done + ); + }); + } + + it(`${spinMethod} preserves ready guards after a timer error`, function (done) { + runInChild( + ` + const failure = new Error('timer failure'); + const calls = []; + const timer = node.createTimer(0n, () => { + timer.cancel(); + calls.push('timer'); + throw failure; + }); + const guard = node.createGuardCondition(() => calls.push('guard')); + guard.trigger(); + let error; + if (${JSON.stringify(spinMethod)} === 'spin') { + const reported = new Promise((resolve) => { + process.once('uncaughtException', resolve); + }); + node.spin(-1); + error = await reported; + node.stop(); + } else { + try { node.spinOnce(0); } catch (caught) { error = caught; } + } + assert.equal(error, failure); + assert.deepEqual(calls, ['timer', 'guard']); + `, + done + ); + }); + + it(`${spinMethod} discards remaining ready callbacks after stop and an error`, function (done) { + runInChild( + ` + const failure = new Error('stopped callback failed'); + const calls = []; + const first = node.createGuardCondition(() => { + calls.push('first'); + node.stop(); + throw failure; + }); + let complete; + const completed = new Promise((resolve) => { complete = resolve; }); + const second = node.createGuardCondition(() => { + calls.push('second'); + complete(); + }); + first.trigger(); + second.trigger(); + let error; + if (${JSON.stringify(spinMethod)} === 'spin') { + const reported = new Promise((resolve) => { + process.once('uncaughtException', resolve); + }); + node.spin(-1); + error = await reported; + } else { + try { node.spinOnce(0); } catch (caught) { error = caught; } + } + assert.equal(error, failure); + assert.deepEqual(calls, ['first']); + second.trigger(); + if (${JSON.stringify(spinMethod)} === 'spin') { + node.spin(-1); + await completed; + node.stop(); + } else { + node.spinOnce(0); + await completed; + } + assert.deepEqual(calls, ['first', 'second']); + `, + done + ); + }); + + for (const operation of ['node.destroy()', 'context.shutdown()']) { + for (const throws of [false, true]) { + it(`${spinMethod} discards ready guards after ${operation}${throws ? ' and an error' : ''}`, function (done) { + runInChild( + ` + const failure = new Error('destroyed callback failed'); + const calls = []; + let complete; + const completed = new Promise((resolve) => { complete = resolve; }); + const timer = node.createTimer(0n, () => { + calls.push('timer'); + timer.cancel(); + ${operation}; + if (${throws}) throw failure; + complete(); + }); + const guard = node.createGuardCondition(() => calls.push('guard')); + guard.trigger(); + let error; + if (${JSON.stringify(spinMethod)} === 'spin') { + const reported = ${throws} ? new Promise((resolve) => { + process.once('uncaughtException', (caught) => resolve({ caught })); + }) : completed; + node.spin(-1); + const result = await reported; + if (${throws}) error = result.caught; + } else { + try { node.spinOnce(0); } catch (caught) { error = caught; } + } + assert.equal(error, ${throws} ? failure : undefined); + assert.deepEqual(calls, ['timer']); + assert.equal(node.spinning, false); + assert.equal(context.nodes.includes(node), false); + assert.equal(context.isShutdown(), ${JSON.stringify(operation)} === 'context.shutdown()'); + `, + done + ); + }); + } + } } }); }); diff --git a/types/node.d.ts b/types/node.d.ts index ae2b37a4..227c1a2e 100644 --- a/types/node.d.ts +++ b/types/node.d.ts @@ -325,6 +325,8 @@ declare module 'rclnodejs' { /** * Trigger the event loop to continuously check for and route. * incoming events. + * Callback errors are reported after the current ready batch is processed. + * Multiple callback errors in a batch are reported as an AggregateError. * @param node - The node to be spun up. * @param - Timeout to wait in milliseconds. Block forever if negative. Don't wait if 0. * @throws Error if the node is already spinning or native executor cleanup is still in progress. @@ -334,6 +336,8 @@ declare module 'rclnodejs' { /** * Spin the node and trigger the event loop to check for one incoming event. Thereafter the node * will not received additional events until running additional calls to spin() or spinOnce(). + * Callback errors are thrown after the current ready batch is processed. + * Multiple callback errors in a batch are thrown as an AggregateError. * @param node - The node to be spun. * @param - Timeout to wait in milliseconds. Block forever if negative. Don't wait if 0. * @throws An error if the node is already spinning or native executor cleanup is still in progress. From 6e3bc584e0c6aac5d7d005675b4cdafc29071a1e Mon Sep 17 00:00:00 2001 From: Minggang Wang Date: Tue, 27 Oct 2026 18:30:46 +0800 Subject: [PATCH 2/7] Address comments --- lib/node.js | 6 ++- test/test-spinning.js | 90 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 1 deletion(-) diff --git a/lib/node.js b/lib/node.js index 08d6a4d9..4ca14c41 100644 --- a/lib/node.js +++ b/lib/node.js @@ -283,7 +283,11 @@ class Node extends rclnodejs.ShadowNode { }) ); - eventsReady.forEach((event) => dispatch(() => event.takeData())); + eventsReady.forEach((event) => + dispatch(() => { + if (!event.isDestroyed()) event.takeData(); + }) + ); for (const subscription of subscriptionsReady) { if (subscription.isDestroyed()) continue; diff --git a/test/test-spinning.js b/test/test-spinning.js index 114d1d91..390fecd1 100644 --- a/test/test-spinning.js +++ b/test/test-spinning.js @@ -14,6 +14,11 @@ import assert from 'assert'; import { execFile } from 'node:child_process'; import sinon from 'sinon'; import rclnodejs from '../index.js'; +import DistroUtils from '../lib/distro.js'; +import { + PublisherEventCallbacks, + SubscriptionEventCallbacks, +} from '../lib/event_handler.js'; import nativeLoader from '../lib/native_loader.js'; describe('Spin testing', function () { @@ -195,6 +200,91 @@ describe('Spin testing', function () { } } + for (const entityKind of ['publisher', 'subscription']) { + for (const origin of ['timer', 'event']) { + for (const throws of [false, true]) { + it(`skips destroyed ${entityKind} events after ${origin} callbacks${throws ? ' throw' : ''}`, function () { + if (DistroUtils.getDistroId() < DistroUtils.getDistroId('jazzy')) { + this.skip(); + } + const EventCallbacks = + entityKind === 'publisher' + ? PublisherEventCallbacks + : SubscriptionEventCallbacks; + const calls = []; + const failure = new Error('event owner destroyed'); + const createEndpoint = (topic, callback) => { + const callbacks = new EventCallbacks(); + callbacks.deadline = callback; + return entityKind === 'publisher' + ? node.createPublisher( + 'std_msgs/msg/String', + topic, + undefined, + callbacks + ) + : node.createSubscription( + 'std_msgs/msg/String', + topic, + undefined, + () => {}, + callbacks + ); + }; + let releasedEndpoint; + const destroyEndpoint = () => { + calls.push('source'); + if (entityKind === 'publisher') { + node.destroyPublisher(releasedEndpoint); + } else { + node.destroySubscription(releasedEndpoint); + } + if (throws) throw failure; + }; + let sourceHandle; + if (origin === 'timer') { + const timer = node.createTimer(0n, () => { + timer.cancel(); + destroyEndpoint(); + }); + sourceHandle = timer.handle; + } else { + const source = createEndpoint('event_source', destroyEndpoint); + sourceHandle = source.events[0].handle; + } + releasedEndpoint = createEndpoint('event_released', () => + calls.push('released') + ); + const liveEndpoint = createEndpoint('event_live', () => + calls.push('live') + ); + const releasedEvent = releasedEndpoint.events[0]; + const takeData = sinon.spy(releasedEvent, 'takeData'); + const handles = [ + sourceHandle, + releasedEvent.handle, + liveEndpoint.events[0].handle, + ]; + try { + if (throws) { + assert.throws( + () => node.execute(handles), + (error) => error === failure + ); + } else { + node.execute(handles); + } + assert.equal(releasedEvent.isDestroyed(), true); + assert.equal(takeData.callCount, 0); + assert.deepEqual(calls, ['source', 'live']); + } finally { + takeData.restore(); + } + }); + } + } + } + it('releases raw messages when callbacks throw', function () { const failure = new Error('message callback failed'); let destroyedMessage; From d20e07ad94ebf865cc3d2f3a46effd2bc76d9277 Mon Sep 17 00:00:00 2001 From: Minggang Wang Date: Wed, 28 Oct 2026 13:16:16 +0800 Subject: [PATCH 3/7] Address comments --- lib/node.js | 18 +--- test/test-spinning.js | 218 ++++++++++++++++++++++++++++++++---------- types/node.d.ts | 7 +- 3 files changed, 173 insertions(+), 70 deletions(-) diff --git a/lib/node.js b/lib/node.js index 4ca14c41..22cac702 100644 --- a/lib/node.js +++ b/lib/node.js @@ -233,13 +233,13 @@ class Node extends rclnodejs.ShadowNode { execute(handles) { const dispatchGeneration = this._dispatchGeneration; - const callbackErrors = []; const dispatch = (callback) => { if (this._dispatchGeneration !== dispatchGeneration) return; try { callback(); } catch (error) { - callbackErrors.push(error); + Entity._gcHandles(); + throw error; } }; let timersReady = this._timers.filter((timer) => @@ -553,13 +553,6 @@ class Node extends rclnodejs.ShadowNode { // At this point it is safe to clear the cache of any // destroyed entity references Entity._gcHandles(); - if (callbackErrors.length === 1) throw callbackErrors[0]; - if (callbackErrors.length > 1) { - throw new AggregateError( - callbackErrors, - 'Multiple executor callbacks failed' - ); - } } /** @@ -573,8 +566,8 @@ class Node extends rclnodejs.ShadowNode { /** * Trigger the event loop to continuously check for and route. * incoming events. - * Callback errors are reported after the current ready batch is processed. - * Multiple callback errors in a batch are reported as an AggregateError. + * A synchronous callback exception stops the current ready batch and is + * reported through Node.js uncaught exception handling. * @param {Node} node - The node to be spun up. * @param {number} [timeout=10] - Timeout to wait in milliseconds. Block forever if negative. Don't wait if 0. * @throws {Error} If the node is already spinning or native executor cleanup is still in progress. @@ -631,8 +624,7 @@ class Node extends rclnodejs.ShadowNode { /** * Spin the node and trigger the event loop to check for one incoming event. Thereafter the node * will not received additional events until running additional calls to spin() or spinOnce(). - * Callback errors are thrown after the current ready batch is processed. - * Multiple callback errors in a batch are thrown as an AggregateError. + * A synchronous callback exception stops the current ready batch and is thrown to the caller. * @param {Node} node - The node to be spun. * @param {number} [timeout=10] - Timeout to wait in milliseconds. Block forever if negative. Don't wait if 0. * @throws {Error} If the node is already spinning or native executor cleanup is still in progress. diff --git a/test/test-spinning.js b/test/test-spinning.js index 390fecd1..5eeb4223 100644 --- a/test/test-spinning.js +++ b/test/test-spinning.js @@ -15,6 +15,7 @@ import { execFile } from 'node:child_process'; import sinon from 'sinon'; import rclnodejs from '../index.js'; import DistroUtils from '../lib/distro.js'; +import Entity from '../lib/entity.js'; import { PublisherEventCallbacks, SubscriptionEventCallbacks, @@ -276,7 +277,7 @@ describe('Spin testing', function () { } assert.equal(releasedEvent.isDestroyed(), true); assert.equal(takeData.callCount, 0); - assert.deepEqual(calls, ['source', 'live']); + assert.deepEqual(calls, throws ? ['source'] : ['source', 'live']); } finally { takeData.restore(); } @@ -285,6 +286,27 @@ describe('Spin testing', function () { } } + it('clears destroyed handle references when a callback throws', function () { + const failure = new Error('guard callback failed'); + let released; + const first = node.createGuardCondition(() => { + node.destroyGuardCondition(released); + throw failure; + }); + released = node.createGuardCondition(() => assert.fail('must not run')); + const clearHandles = sinon.spy(Entity, '_gcHandles'); + try { + assert.throws( + () => node.execute([first.handle, released.handle]), + (error) => error === failure + ); + assert.equal(released.isDestroyed(), true); + assert.equal(clearHandles.callCount, 1); + } finally { + clearHandles.restore(); + } + }); + it('releases raw messages when callbacks throw', function () { const failure = new Error('message callback failed'); let destroyedMessage; @@ -308,8 +330,13 @@ describe('Spin testing', function () { }); for (const actionKind of ['client', 'server']) { - for (const destroyOnError of [false, true]) { - it(`handles ready action ${actionKind} events after ${destroyOnError ? 'destruction and ' : ''}an error`, function () { + for (const [destroyDuringCallback, throws] of [ + [false, false], + [false, true], + [true, false], + [true, true], + ]) { + it(`handles ready action ${actionKind} events after ${destroyDuringCallback ? 'destruction' : 'a callback'}${throws ? ' and an error' : ''}`, function () { const sandbox = sinon.createSandbox(); const failure = new Error('action callback failed'); const calls = []; @@ -333,8 +360,8 @@ describe('Spin testing', function () { isDestroyed: () => destroyed, [`processGoal${messageKind}`]: () => { calls.push('goal'); - destroyed = destroyOnError; - throw failure; + destroyed = destroyDuringCallback; + if (throws) throw failure; }, [`processCancel${messageKind}`]: () => calls.push('cancel'), }; @@ -347,13 +374,17 @@ describe('Spin testing', function () { sandbox .stub(node, '_runWithMessageType') .callsFake((type, callback) => callback({}, () => ({}))); - assert.throws( - () => node.execute([action.handle]), - (error) => error === failure - ); + if (throws) { + assert.throws( + () => node.execute([action.handle]), + (error) => error === failure + ); + } else { + node.execute([action.handle]); + } assert.deepEqual( calls, - destroyOnError ? ['goal'] : ['goal', 'cancel'] + destroyDuringCallback || throws ? ['goal'] : ['goal', 'cancel'] ); } finally { collection.splice(collection.indexOf(action), 1); @@ -844,6 +875,78 @@ describe('Spin testing', function () { ); }); + it('continues spinning on new work after a handled callback error', function (done) { + runInChild( + ` + const failure = new Error('handled callback failure'); + const calls = []; + const reported = new Promise((resolve) => { + process.once('uncaughtException', resolve); + }); + let complete; + const completed = new Promise((resolve) => { complete = resolve; }); + const first = node.createGuardCondition(() => { + calls.push('first'); + throw failure; + }); + const later = node.createGuardCondition(() => { + calls.push('later'); + node.stop(); + complete(); + }); + first.trigger(); + node.spin(-1); + assert.equal(await reported, failure); + assert.equal(node.spinning, true); + assert.deepEqual(calls, ['first']); + later.trigger(); + await completed; + assert.deepEqual(calls, ['first', 'later']); + assert.equal(node.spinning, false); + `, + done + ); + }); + + it('exits on an unhandled callback error without running later callbacks', function (done) { + const moduleUrl = new URL('../index.js', import.meta.url).href; + const script = ` + import rclnodejs from ${JSON.stringify(moduleUrl)}; + await rclnodejs.init(); + const node = rclnodejs.createNode( + 'unhandled_callback_error', '', rclnodejs.Context.defaultContext(), + new rclnodejs.NodeOptions(false, [], false, false, false) + ); + const first = node.createGuardCondition(() => { + throw new Error('unhandled callback error'); + }); + const second = node.createGuardCondition(() => { + console.log('unexpected second callback'); + }); + first.trigger(); + second.trigger(); + node.spin(-1); + `; + execFile( + process.execPath, + ['--input-type=module', '--eval', script], + { timeout: 10000, killSignal: 'SIGKILL' }, + (error, stdout, stderr) => { + try { + assert.ok(error); + assert.equal(error.code, 1); + assert.equal(error.signal, null); + assert.match(stderr, /unhandled callback error/); + assert.doesNotMatch(stderr, /FATAL ERROR/); + assert.doesNotMatch(stdout, /unexpected second callback/); + done(); + } catch (failure) { + done(failure); + } + } + ); + }); + for (const spinMethod of ['spin', 'spinOnce']) { it(`${spinMethod} preserves non-Error thrown values`, function (done) { runInChild( @@ -876,50 +979,59 @@ describe('Spin testing', function () { ); }); - for (const multipleErrors of [false, true]) { - it(`${spinMethod} preserves ready guards after ${multipleErrors ? 'multiple errors' : 'an error'}`, function (done) { - runInChild( - ` - const firstError = new Error('first guard failure'); - const secondError = new Error('second guard failure'); - const calls = []; - const first = node.createGuardCondition(() => { - calls.push('first'); - throw firstError; - }); - const second = node.createGuardCondition(() => { - calls.push('second'); - if (${multipleErrors}) throw secondError; + it(`${spinMethod} stops the ready batch at the first guard error`, function (done) { + runInChild( + ` + const firstError = new Error('first guard failure'); + const calls = []; + let complete; + const completed = new Promise((resolve) => { complete = resolve; }); + const first = node.createGuardCondition(() => { + calls.push('first'); + throw firstError; + }); + const second = node.createGuardCondition(() => { + calls.push('second'); + throw new Error('second guard failure'); + }); + const third = node.createGuardCondition(() => { + calls.push('third'); + complete(); + }); + first.trigger(); + second.trigger(); + third.trigger(); + let error; + if (${JSON.stringify(spinMethod)} === 'spin') { + const reported = new Promise((resolve) => { + process.once('uncaughtException', resolve); }); - const third = node.createGuardCondition(() => calls.push('third')); - first.trigger(); - second.trigger(); - third.trigger(); - let error; - if (${JSON.stringify(spinMethod)} === 'spin') { - const reported = new Promise((resolve) => { - process.once('uncaughtException', resolve); - }); - node.spin(-1); - error = await reported; - node.stop(); - } else { - try { node.spinOnce(0); } catch (failure) { error = failure; } - } - assert.deepEqual(calls, ['first', 'second', 'third']); - if (${multipleErrors}) { - assert.ok(error instanceof AggregateError); - assert.deepEqual(error.errors, [firstError, secondError]); - } else { - assert.equal(error, firstError); - } - `, - done - ); - }); - } + node.spin(-1); + error = await reported; + node.stop(); + } else { + try { node.spinOnce(0); } catch (failure) { error = failure; } + } + assert.equal(error, firstError); + assert.deepEqual(calls, ['first']); + node.destroyGuardCondition(first); + node.destroyGuardCondition(second); + third.trigger(); + if (${JSON.stringify(spinMethod)} === 'spin') { + node.spin(-1); + await completed; + node.stop(); + } else { + node.spinOnce(0); + await completed; + } + assert.deepEqual(calls, ['first', 'third']); + `, + done + ); + }); - it(`${spinMethod} preserves ready guards after a timer error`, function (done) { + it(`${spinMethod} stops before ready guards after a timer error`, function (done) { runInChild( ` const failure = new Error('timer failure'); @@ -943,7 +1055,7 @@ describe('Spin testing', function () { try { node.spinOnce(0); } catch (caught) { error = caught; } } assert.equal(error, failure); - assert.deepEqual(calls, ['timer', 'guard']); + assert.deepEqual(calls, ['timer']); `, done ); diff --git a/types/node.d.ts b/types/node.d.ts index 227c1a2e..25b8f212 100644 --- a/types/node.d.ts +++ b/types/node.d.ts @@ -325,8 +325,8 @@ declare module 'rclnodejs' { /** * Trigger the event loop to continuously check for and route. * incoming events. - * Callback errors are reported after the current ready batch is processed. - * Multiple callback errors in a batch are reported as an AggregateError. + * A synchronous callback exception stops the current ready batch and is + * reported through Node.js uncaught exception handling. * @param node - The node to be spun up. * @param - Timeout to wait in milliseconds. Block forever if negative. Don't wait if 0. * @throws Error if the node is already spinning or native executor cleanup is still in progress. @@ -336,8 +336,7 @@ declare module 'rclnodejs' { /** * Spin the node and trigger the event loop to check for one incoming event. Thereafter the node * will not received additional events until running additional calls to spin() or spinOnce(). - * Callback errors are thrown after the current ready batch is processed. - * Multiple callback errors in a batch are thrown as an AggregateError. + * A synchronous callback exception stops the current ready batch and is thrown to the caller. * @param node - The node to be spun. * @param - Timeout to wait in milliseconds. Block forever if negative. Don't wait if 0. * @throws An error if the node is already spinning or native executor cleanup is still in progress. From 2246d9c6df76737bcc7dfdd23dbdda710681861f Mon Sep 17 00:00:00 2001 From: Minggang Wang Date: Wed, 28 Oct 2026 13:42:37 +0800 Subject: [PATCH 4/7] Address comments --- lib/node.js | 218 +++++++++++++++++++----------------------- test/test-spinning.js | 19 ++-- 2 files changed, 108 insertions(+), 129 deletions(-) diff --git a/lib/node.js b/lib/node.js index 22cac702..5e3e7fb3 100644 --- a/lib/node.js +++ b/lib/node.js @@ -233,42 +233,34 @@ class Node extends rclnodejs.ShadowNode { execute(handles) { const dispatchGeneration = this._dispatchGeneration; - const dispatch = (callback) => { - if (this._dispatchGeneration !== dispatchGeneration) return; - try { - callback(); - } catch (error) { - Entity._gcHandles(); - throw error; - } - }; - let timersReady = this._timers.filter((timer) => - handles.includes(timer.handle) - ); - let guardsReady = this._guards.filter((guard) => - handles.includes(guard.handle) - ); - let subscriptionsReady = this._subscriptions.filter((subscription) => - handles.includes(subscription.handle) - ); - let clientsReady = this._clients.filter((client) => - handles.includes(client.handle) - ); - let servicesReady = this._services.filter((service) => - handles.includes(service.handle) - ); - let actionClientsReady = this._actionClients.filter((actionClient) => - handles.includes(actionClient.handle) - ); - let actionServersReady = this._actionServers.filter((actionServer) => - handles.includes(actionServer.handle) - ); - let eventsReady = this._events.filter((event) => - handles.includes(event.handle) - ); + try { + let timersReady = this._timers.filter((timer) => + handles.includes(timer.handle) + ); + let guardsReady = this._guards.filter((guard) => + handles.includes(guard.handle) + ); + let subscriptionsReady = this._subscriptions.filter((subscription) => + handles.includes(subscription.handle) + ); + let clientsReady = this._clients.filter((client) => + handles.includes(client.handle) + ); + let servicesReady = this._services.filter((service) => + handles.includes(service.handle) + ); + let actionClientsReady = this._actionClients.filter((actionClient) => + handles.includes(actionClient.handle) + ); + let actionServersReady = this._actionServers.filter((actionServer) => + handles.includes(actionServer.handle) + ); + let eventsReady = this._events.filter((event) => + handles.includes(event.handle) + ); - timersReady.forEach((timer) => - dispatch(() => { + for (const timer of timersReady) { + if (this._dispatchGeneration !== dispatchGeneration) return; // Earlier callbacks may destroy timers or the node after this snapshot. if (this._timers.includes(timer) && timer.isReady()) { let timerInfo; @@ -280,19 +272,17 @@ class Node extends rclnodejs.ShadowNode { timer.callback(); } } - }) - ); + } - eventsReady.forEach((event) => - dispatch(() => { + for (const event of eventsReady) { + if (this._dispatchGeneration !== dispatchGeneration) return; if (!event.isDestroyed()) event.takeData(); - }) - ); + } - for (const subscription of subscriptionsReady) { - if (subscription.isDestroyed()) continue; - if (subscription.isRaw) { - dispatch(() => { + for (const subscription of subscriptionsReady) { + if (this._dispatchGeneration !== dispatchGeneration) return; + if (subscription.isDestroyed()) continue; + if (subscription.isRaw) { if (subscription.wantsMessageInfo) { let taken = rclnodejs.rclTakeRawWithInfo(subscription.handle); if (taken) { @@ -307,11 +297,9 @@ class Node extends rclnodejs.ShadowNode { subscription.processResponse(rawMessage); } } - }); - continue; - } + continue; + } - dispatch(() => this._runWithMessageType( subscription.typeClass, (message, deserialize) => { @@ -333,19 +321,19 @@ class Node extends rclnodejs.ShadowNode { } } } - ) - ); - } + ); + } - for (const guard of guardsReady) { - if (guard.isDestroyed()) continue; + for (const guard of guardsReady) { + if (this._dispatchGeneration !== dispatchGeneration) return; + if (guard.isDestroyed()) continue; - dispatch(() => guard.callback()); - } + guard.callback(); + } - for (const client of clientsReady) { - if (client.isDestroyed()) continue; - dispatch(() => + for (const client of clientsReady) { + if (this._dispatchGeneration !== dispatchGeneration) return; + if (client.isDestroyed()) continue; this._runWithMessageType( client.typeClass.Response, (message, deserialize) => { @@ -357,13 +345,12 @@ class Node extends rclnodejs.ShadowNode { client.processResponse(sequenceNumber, deserialize()); } } - ) - ); - } + ); + } - for (const service of servicesReady) { - if (service.isDestroyed()) continue; - dispatch(() => + for (const service of servicesReady) { + if (this._dispatchGeneration !== dispatchGeneration) return; + if (service.isDestroyed()) continue; this._runWithMessageType( service.typeClass.Request, (message, deserialize) => { @@ -376,17 +363,16 @@ class Node extends rclnodejs.ShadowNode { service.processRequest(header, deserialize()); } } - ) - ); - } + ); + } - for (const actionClient of actionClientsReady) { - if (actionClient.isDestroyed()) continue; + for (const actionClient of actionClientsReady) { + if (this._dispatchGeneration !== dispatchGeneration) return; + if (actionClient.isDestroyed()) continue; - const properties = actionClient.handle.properties; + const properties = actionClient.handle.properties; - if (properties.isGoalResponseReady) { - dispatch(() => + if (properties.isGoalResponseReady) { this._runWithMessageType( actionClient.typeClass.impl.SendGoalService.Response, (message, deserialize) => { @@ -398,12 +384,11 @@ class Node extends rclnodejs.ShadowNode { actionClient.processGoalResponse(sequence, deserialize()); } } - ) - ); - } + ); + } - if (properties.isCancelResponseReady && !actionClient.isDestroyed()) { - dispatch(() => + if (this._dispatchGeneration !== dispatchGeneration) return; + if (properties.isCancelResponseReady && !actionClient.isDestroyed()) { this._runWithMessageType( actionClient.typeClass.impl.CancelGoal.Response, (message, deserialize) => { @@ -415,12 +400,11 @@ class Node extends rclnodejs.ShadowNode { actionClient.processCancelResponse(sequence, deserialize()); } } - ) - ); - } + ); + } - if (properties.isResultResponseReady && !actionClient.isDestroyed()) { - dispatch(() => + if (this._dispatchGeneration !== dispatchGeneration) return; + if (properties.isResultResponseReady && !actionClient.isDestroyed()) { this._runWithMessageType( actionClient.typeClass.impl.GetResultService.Response, (message, deserialize) => { @@ -432,12 +416,11 @@ class Node extends rclnodejs.ShadowNode { actionClient.processResultResponse(sequence, deserialize()); } } - ) - ); - } + ); + } - if (properties.isFeedbackReady && !actionClient.isDestroyed()) { - dispatch(() => + if (this._dispatchGeneration !== dispatchGeneration) return; + if (properties.isFeedbackReady && !actionClient.isDestroyed()) { this._runWithMessageType( actionClient.typeClass.impl.FeedbackMessage, (message, deserialize) => { @@ -449,12 +432,11 @@ class Node extends rclnodejs.ShadowNode { actionClient.processFeedbackMessage(deserialize()); } } - ) - ); - } + ); + } - if (properties.isStatusReady && !actionClient.isDestroyed()) { - dispatch(() => + if (this._dispatchGeneration !== dispatchGeneration) return; + if (properties.isStatusReady && !actionClient.isDestroyed()) { this._runWithMessageType( actionClient.typeClass.impl.GoalStatusArray, (message, deserialize) => { @@ -466,18 +448,17 @@ class Node extends rclnodejs.ShadowNode { actionClient.processStatusMessage(deserialize()); } } - ) - ); + ); + } } - } - for (const actionServer of actionServersReady) { - if (actionServer.isDestroyed()) continue; + for (const actionServer of actionServersReady) { + if (this._dispatchGeneration !== dispatchGeneration) return; + if (actionServer.isDestroyed()) continue; - const properties = actionServer.handle.properties; + const properties = actionServer.handle.properties; - if (properties.isGoalRequestReady) { - dispatch(() => + if (properties.isGoalRequestReady) { this._runWithMessageType( actionServer.typeClass.impl.SendGoalService.Request, (message, deserialize) => { @@ -489,12 +470,11 @@ class Node extends rclnodejs.ShadowNode { actionServer.processGoalRequest(result, deserialize()); } } - ) - ); - } + ); + } - if (properties.isCancelRequestReady && !actionServer.isDestroyed()) { - dispatch(() => + if (this._dispatchGeneration !== dispatchGeneration) return; + if (properties.isCancelRequestReady && !actionServer.isDestroyed()) { this._runWithMessageType( actionServer.typeClass.impl.CancelGoal.Request, (message, deserialize) => { @@ -506,12 +486,11 @@ class Node extends rclnodejs.ShadowNode { actionServer.processCancelRequest(result, deserialize()); } } - ) - ); - } + ); + } - if (properties.isResultRequestReady && !actionServer.isDestroyed()) { - dispatch(() => + if (this._dispatchGeneration !== dispatchGeneration) return; + if (properties.isResultRequestReady && !actionServer.isDestroyed()) { this._runWithMessageType( actionServer.typeClass.impl.GetResultService.Request, (message, deserialize) => { @@ -523,12 +502,11 @@ class Node extends rclnodejs.ShadowNode { actionServer.processResultRequest(result, deserialize()); } } - ) - ); - } + ); + } - if (properties.isGoalExpired && !actionServer.isDestroyed()) { - dispatch(() => { + if (this._dispatchGeneration !== dispatchGeneration) return; + if (properties.isGoalExpired && !actionServer.isDestroyed()) { let numGoals = actionServer._goalHandles.size; if (numGoals > 0) { let GoalInfoArray = ActionInterfaces.GoalInfo.ArrayType; @@ -546,13 +524,11 @@ class Node extends rclnodejs.ShadowNode { GoalInfoArray.freeArray(message); } } - }); + } } + } finally { + Entity._gcHandles(); } - - // At this point it is safe to clear the cache of any - // destroyed entity references - Entity._gcHandles(); } /** diff --git a/test/test-spinning.js b/test/test-spinning.js index 5eeb4223..bcc8b517 100644 --- a/test/test-spinning.js +++ b/test/test-spinning.js @@ -330,13 +330,15 @@ describe('Spin testing', function () { }); for (const actionKind of ['client', 'server']) { - for (const [destroyDuringCallback, throws] of [ - [false, false], - [false, true], - [true, false], - [true, true], + for (const [operation, throws] of [ + ['callback', false], + ['callback', true], + ['destruction', false], + ['destruction', true], + ['stop', false], + ['stop', true], ]) { - it(`handles ready action ${actionKind} events after ${destroyDuringCallback ? 'destruction' : 'a callback'}${throws ? ' and an error' : ''}`, function () { + it(`handles ready action ${actionKind} events after ${operation === 'callback' ? 'a callback' : operation}${throws ? ' and an error' : ''}`, function () { const sandbox = sinon.createSandbox(); const failure = new Error('action callback failed'); const calls = []; @@ -360,7 +362,8 @@ describe('Spin testing', function () { isDestroyed: () => destroyed, [`processGoal${messageKind}`]: () => { calls.push('goal'); - destroyed = destroyDuringCallback; + destroyed = operation === 'destruction'; + if (operation === 'stop') node.stop(); if (throws) throw failure; }, [`processCancel${messageKind}`]: () => calls.push('cancel'), @@ -384,7 +387,7 @@ describe('Spin testing', function () { } assert.deepEqual( calls, - destroyDuringCallback || throws ? ['goal'] : ['goal', 'cancel'] + operation !== 'callback' || throws ? ['goal'] : ['goal', 'cancel'] ); } finally { collection.splice(collection.indexOf(action), 1); From dd2190fcc104ffb38fbe927cfe7c21f055834413 Mon Sep 17 00:00:00 2001 From: Minggang Wang Date: Wed, 28 Oct 2026 14:56:02 +0800 Subject: [PATCH 5/7] Add debug log --- lib/node.js | 1 + test/test-spinning.js | 20 ++++++++++++++++++++ 2 files changed, 21 insertions(+) diff --git a/lib/node.js b/lib/node.js index 5e3e7fb3..bf5fba81 100644 --- a/lib/node.js +++ b/lib/node.js @@ -163,6 +163,7 @@ class Node extends rclnodejs.ShadowNode { this._postSetParametersCallbacks = []; this._logger = new Logging(rclnodejs.getNodeLoggerName(this.handle)); this._spinning = false; + // Detects stop/destroy during a ready batch, even if spinning restarts. this._dispatchGeneration = 0; this._enableRosout = options.enableRosout; diff --git a/test/test-spinning.js b/test/test-spinning.js index bcc8b517..c50e1ff4 100644 --- a/test/test-spinning.js +++ b/test/test-spinning.js @@ -944,6 +944,26 @@ describe('Spin testing', function () { assert.doesNotMatch(stdout, /unexpected second callback/); done(); } catch (failure) { + console.error( + 'Unhandled callback child process diagnostics:', + JSON.stringify( + { + code: error?.code ?? null, + signal: error?.signal ?? null, + killed: error?.killed ?? false, + node: process.version, + rosDistro: process.env.ROS_DISTRO ?? null, + rmwImplementation: process.env.RMW_IMPLEMENTATION ?? null, + asanOptions: process.env.ASAN_OPTIONS ?? null, + lsanOptions: process.env.LSAN_OPTIONS ?? null, + ldPreload: process.env.LD_PRELOAD ?? null, + }, + null, + 2 + ) + ); + console.error('Child stdout:\n%s', stdout || ''); + console.error('Child stderr:\n%s', stderr || ''); done(failure); } } From b9fe42d2a92fc8ae147584541266eec2f6593951 Mon Sep 17 00:00:00 2001 From: Minggang Wang Date: Wed, 28 Oct 2026 15:47:24 +0800 Subject: [PATCH 6/7] Fix leak when exit --- index.js | 1 + test/test-context.js | 62 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+) diff --git a/index.js b/index.js index 14e99ca8..779f5f9e 100644 --- a/index.js +++ b/index.js @@ -770,6 +770,7 @@ const _sigHandler = () => { rcl.shutdownAll(); }; process.on('SIGINT', _sigHandler); +process.once('exit', () => rcl.shutdownAll()); /** {@link Node} class */ rcl.Node = Node; diff --git a/test/test-context.js b/test/test-context.js index 6c72cad8..652fdd28 100644 --- a/test/test-context.js +++ b/test/test-context.js @@ -13,6 +13,7 @@ // limitations under the License. import assert from 'assert'; +import { execFile } from 'node:child_process'; import rclnodejs from '../index.js'; describe('context test suite', function () { @@ -91,4 +92,65 @@ describe('context test suite', function () { await rclnodejs.init(context); assert.strictEqual(typeof context.domainId, 'bigint'); }); + + for (const [termination, operation, expectedCode] of [ + ['natural exit', 'for (const node of nodes) node.stop();', 0], + ['explicit exit', 'process.exit(7);', 7], + ['an unhandled callback', 'guard.trigger();', 1], + ]) { + it(`shuts down all contexts on ${termination}`, function (done) { + const moduleUrl = new URL('../index.js', import.meta.url).href; + const script = ` + import assert from 'node:assert/strict'; + import rclnodejs from ${JSON.stringify(moduleUrl)}; + const contexts = [ + rclnodejs.Context.defaultContext(), + new rclnodejs.Context(), + ]; + const nodes = []; + for (const [index, context] of contexts.entries()) { + await rclnodejs.init(context); + nodes.push(rclnodejs.createNode( + 'exit_cleanup_' + index, '', context, + new rclnodejs.NodeOptions(false, [], false, false, false) + )); + } + const guard = nodes[0].createGuardCondition(() => { + throw new Error('exit cleanup regression'); + }); + for (const node of nodes) node.spin(-1); + process.once('exit', (code) => { + assert.equal(code, ${expectedCode}); + assert.equal(rclnodejs.Context.instances.length, 0); + for (const context of contexts) { + assert.equal(context.isShutdown(), true); + assert.equal(context.isValid(), false); + assert.equal(context.nodes.length, 0); + } + for (const node of nodes) assert.equal(node.spinning, false); + console.log('all contexts finalized'); + }); + ${operation} + `; + execFile( + process.execPath, + ['--input-type=module', '--eval', script], + { timeout: 10000, killSignal: 'SIGKILL' }, + (error, stdout, stderr) => { + try { + assert.equal(error?.code ?? 0, expectedCode, stderr); + assert.equal(error?.signal ?? null, null, stderr); + assert.match(stdout, /all contexts finalized/); + assert.doesNotMatch(stderr, /ERROR: (?:Address|Leak)Sanitizer/); + if (expectedCode === 1) { + assert.match(stderr, /exit cleanup regression/); + } + done(); + } catch (failure) { + done(failure); + } + } + ); + }); + } }); From 73f4c171b5b1658e9e5729c45466851edf1ce9a5 Mon Sep 17 00:00:00 2001 From: Minggang Wang Date: Wed, 28 Oct 2026 16:49:28 +0800 Subject: [PATCH 7/7] Address comment --- src/rcl_context_bindings.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/rcl_context_bindings.cpp b/src/rcl_context_bindings.cpp index 7daad5b2..1df349fd 100644 --- a/src/rcl_context_bindings.cpp +++ b/src/rcl_context_bindings.cpp @@ -108,8 +108,8 @@ int DestroyContext(Napi::Env env, rcl_context_t* context) { Napi::Error::New(env, rcl_get_error_string().str) .ThrowAsJavaScriptException(); } - ret = rcl_context_fini(context); } + ret = rcl_context_fini(context); } return ret; }