From c545932ba5cdc46ff58986df50e911a064b0da03 Mon Sep 17 00:00:00 2001 From: Peter C <12292660+PeterC89@users.noreply.github.com> Date: Fri, 26 Jun 2026 12:44:16 +0100 Subject: [PATCH 1/3] fix: Intermittent subscription update loss and socket lifecycle reliability --- src/Ember/Client/ConnectionStatus.ts | 6 ++ src/Ember/Client/__tests__/index.spec.ts | 73 +++++++++++++++++++ src/Ember/Client/index.ts | 70 ++++++++++-------- src/Ember/Socket/S101Client.ts | 2 +- src/Ember/Socket/S101Socket.ts | 36 +++++++-- src/Ember/Socket/__tests__/S101Socket.spec.ts | 57 +++++++++++++++ src/__mocks__/S101Client.ts | 2 +- 7 files changed, 206 insertions(+), 40 deletions(-) create mode 100644 src/Ember/Client/ConnectionStatus.ts create mode 100644 src/Ember/Socket/__tests__/S101Socket.spec.ts diff --git a/src/Ember/Client/ConnectionStatus.ts b/src/Ember/Client/ConnectionStatus.ts new file mode 100644 index 0000000..7eb54ae --- /dev/null +++ b/src/Ember/Client/ConnectionStatus.ts @@ -0,0 +1,6 @@ +export enum ConnectionStatus { + Error, + Disconnected, + Connecting, + Connected, +} diff --git a/src/Ember/Client/__tests__/index.spec.ts b/src/Ember/Client/__tests__/index.spec.ts index 8ac637f..b9cb7bc 100644 --- a/src/Ember/Client/__tests__/index.spec.ts +++ b/src/Ember/Client/__tests__/index.spec.ts @@ -420,6 +420,79 @@ describe('client', () => { }) }) + describe('Subscription behavior regressions', () => { + it('invokes all callbacks subscribed to the same path', async () => { + await runWithConnection(async (client, socket) => { + const cb1 = jest.fn() + const cb2 = jest.fn() + + const parameter = new ParameterImpl(ParameterType.Integer, 'Level', undefined, 1) + const node = new NumberedTreeNodeImpl(1, parameter) + + await client.subscribe(node, cb1) + await client.subscribe(node, cb2) + + // Prime the tree first so subsequent qualified updates match path "1". + socket.mockData({ + value: { + 1: new NumberedTreeNodeImpl(1, new ParameterImpl(ParameterType.Integer, 'Level', undefined, 1)), + }, + }) + await new Promise(setImmediate) + + socket.mockData( + createQualifiedNodeResponse('1', new ParameterImpl(ParameterType.Integer, 'Level', undefined, 2), undefined) + ) + + await new Promise(setImmediate) + + expect(cb1).toHaveBeenCalledTimes(1) + expect(cb2).toHaveBeenCalledTimes(1) + }) + }) + + it('removes all matching subscriptions when unsubscribing a path', async () => { + await runWithConnection(async (client) => { + const parameter = new ParameterImpl(ParameterType.Integer, 'Level', undefined, 1) + const node = new NumberedTreeNodeImpl(1, parameter) + + await client.subscribe(node, jest.fn()) + await client.subscribe(node, jest.fn()) + + //@ts-expect-error - private member + expect(client._subscriptions.filter((s) => s.path === '1')).toHaveLength(2) + + await client.unsubscribe(node) + + //@ts-expect-error - private member + expect(client._subscriptions.filter((s) => s.path === '1')).toHaveLength(0) + }) + }) + + it('handles qualified updates that introduce new child nodes', async () => { + await runWithConnection(async (client, socket) => { + socket.mockData({ + value: { + 1: new NumberedTreeNodeImpl(1, new EmberNodeImpl('Root', undefined, undefined, true), { + 1: new NumberedTreeNodeImpl(1, new EmberNodeImpl('Existing', undefined, undefined, true)), + }), + }, + }) + await new Promise(setImmediate) + + const update = createQualifiedNodeResponse('1', new EmberNodeImpl('Root', undefined, undefined, true), { + 1: new NumberedTreeNodeImpl(1, new EmberNodeImpl('Existing', undefined, undefined, true)), + 2: new NumberedTreeNodeImpl(2, new EmberNodeImpl('Inserted', undefined, undefined, true)), + }) + + expect(() => socket.mockData(update)).not.toThrow() + await new Promise(setImmediate) + + expect(client.tree[1].children?.[2]).toBeDefined() + }) + }) + }) + describe('StreamManager Integration', () => { it('registers stream parameter when subscribing', async () => { await runWithConnection(async (client, socket) => { diff --git a/src/Ember/Client/index.ts b/src/Ember/Client/index.ts index 7f947e1..baa9f3e 100644 --- a/src/Ember/Client/index.ts +++ b/src/Ember/Client/index.ts @@ -36,6 +36,9 @@ import { EmberFunction } from '../../model/EmberFunction' import { DecodeResult } from '../../encodings/ber/decoder/DecodeResult' import { StreamEntry } from '../../model/StreamEntry' import { StreamManager } from './StreamManager' +import { ConnectionStatus } from './ConnectionStatus' + +export { ConnectionStatus } from './ConnectionStatus' export type RequestPromise = Promise> export interface RequestPromiseArguments { @@ -75,13 +78,6 @@ export interface Change { emptyNode?: boolean } -export enum ConnectionStatus { - Error, - Disconnected, - Connecting, - Connected, -} - export type EmberClientEvents = { error: [Error] warn: [Error] @@ -273,15 +269,17 @@ export class EmberClient extends EventEmitter { const command: Unsubscribe = new UnsubscribeImpl() - const path = Array.isArray(node) ? '' : getPath(node) - - // Clean up subscriptions - for (const i in this._subscriptions) { - if (this._subscriptions[i].path === path) { - this._subscriptions.splice(Number(i), 1) - } + if (Array.isArray(node)) { + // root subscriptions are tracked with undefined path + this._subscriptions = this._subscriptions.filter((subscription) => subscription.path !== undefined) + return this._sendRequest(new NumberedTreeNodeImpl(0, command), ExpectResponse.Any) } + const path = getPath(node) + + // Remove all matching subscriptions for the path in one pass + this._subscriptions = this._subscriptions.filter((subscription) => subscription.path !== path) + // Deregister from StreamManager if this was a Parameter with streamIdentifier if (!Array.isArray(node) && node.contents.type === ElementType.Parameter) { const parameter = node.contents @@ -290,10 +288,6 @@ export class EmberClient extends EventEmitter { } } - if (Array.isArray(node)) { - return this._sendRequest(new NumberedTreeNodeImpl(0, command), ExpectResponse.Any) - } - return this._sendCommand(node, command, ExpectResponse.None) } async invoke( @@ -564,8 +558,12 @@ export class EmberClient extends EventEmitter { // check for subscriptiions: for (const change of changes) { - const subscription = this._subscriptions.find((s) => s.path === change.path) - if (subscription && change.node) subscription.cb(change.node) + const subscriptions = this._subscriptions.filter((s) => s.path === change.path) + if (change.node) { + for (const subscription of subscriptions) { + subscription.cb(change.node) + } + } } // check for any outstanding requests and resolve them @@ -708,18 +706,26 @@ export class EmberClient extends EventEmitter { break } } - if (update.children && tree.children) { - // Update children - for (const child of Object.values>(update.children)) { - const i = child.number - const oldChild = tree.children[i] // as NumberedTreeNode | undefined // TODO - changes.push(...this._updateTree(child, oldChild)) - } - } else if (update.children) { - changes.push({ path: getPath(tree), node: tree }) - tree.children = update.children - for (const c of Object.values>(update.children)) { - c.parent = tree + if (update.children) { + if (!tree.children) { + changes.push({ path: getPath(tree), node: tree }) + tree.children = update.children + for (const c of Object.values>(update.children)) { + c.parent = tree + } + } else { + // Update existing children and insert missing children without recursing into undefined nodes. + for (const child of Object.values>(update.children)) { + const i = child.number + const oldChild = tree.children[i] + if (oldChild) { + changes.push(...this._updateTree(child, oldChild)) + } else { + child.parent = tree + tree.children[i] = child + changes.push({ path: getPath(tree), node: tree }) + } + } } } diff --git a/src/Ember/Socket/S101Client.ts b/src/Ember/Socket/S101Client.ts index 306e78d..005a794 100644 --- a/src/Ember/Socket/S101Client.ts +++ b/src/Ember/Socket/S101Client.ts @@ -1,6 +1,6 @@ import net from 'net' import S101Socket from './S101Socket' -import { ConnectionStatus } from '../Client' +import { ConnectionStatus } from '../Client/ConnectionStatus' import { normalizeError } from '../Lib/util' import Debug from 'debug' diff --git a/src/Ember/Socket/S101Socket.ts b/src/Ember/Socket/S101Socket.ts index d7db5ad..862ba49 100644 --- a/src/Ember/Socket/S101Socket.ts +++ b/src/Ember/Socket/S101Socket.ts @@ -2,8 +2,8 @@ import { EventEmitter } from 'eventemitter3' import { Socket } from 'net' import { S101Codec } from '../../S101' -import { berDecode } from '../..' -import { ConnectionStatus } from '../Client' +import { berDecode } from '../../encodings/ber' +import { ConnectionStatus } from '../Client/ConnectionStatus' import { normalizeError } from '../Lib/util' import { Root } from '../../types' import { DecodeResult } from '../../encodings/ber/decoder/DecodeResult' @@ -24,7 +24,7 @@ export default class S101Socket extends EventEmitter { private readonly keepaliveInterval = 10 private readonly keepaliveMaxResponseTime = 500 protected keepaliveIntervalTimer: NodeJS.Timeout | undefined - private keepaliveResponseWindowTimer: NodeJS.Timer | null + private keepaliveResponseWindowTimer: NodeJS.Timeout | null status: ConnectionStatus protected readonly codec = new S101Codec() @@ -40,7 +40,10 @@ export default class S101Socket extends EventEmitter { }) this.codec.on('keepaliveResp', () => { - clearInterval(this.keepaliveResponseWindowTimer) + if (this.keepaliveResponseWindowTimer) { + clearTimeout(this.keepaliveResponseWindowTimer) + this.keepaliveResponseWindowTimer = null + } }) this.codec.on('emberPacket', (packet) => { @@ -79,7 +82,15 @@ export default class S101Socket extends EventEmitter { this.socket.on('close', () => { this.emit('disconnected') - this.status = ConnectionStatus.Connected + this.status = ConnectionStatus.Disconnected + if (this.keepaliveIntervalTimer) { + clearInterval(this.keepaliveIntervalTimer) + this.keepaliveIntervalTimer = undefined + } + if (this.keepaliveResponseWindowTimer) { + clearTimeout(this.keepaliveResponseWindowTimer) + this.keepaliveResponseWindowTimer = null + } this.socket?.removeAllListeners() this.socket = undefined }) @@ -102,6 +113,10 @@ export default class S101Socket extends EventEmitter { clearInterval(this.keepaliveIntervalTimer) this.keepaliveIntervalTimer = undefined } + if (this.keepaliveResponseWindowTimer != null) { + clearTimeout(this.keepaliveResponseWindowTimer) + this.keepaliveResponseWindowTimer = null + } if (this.socket) { let done = false const cb = () => { @@ -129,8 +144,17 @@ export default class S101Socket extends EventEmitter { * */ protected handleClose(): void { + this.socket?.removeAllListeners() + this.socket?.destroy() this.socket = undefined - if (this.keepaliveIntervalTimer) clearInterval(this.keepaliveIntervalTimer) + if (this.keepaliveIntervalTimer) { + clearInterval(this.keepaliveIntervalTimer) + this.keepaliveIntervalTimer = undefined + } + if (this.keepaliveResponseWindowTimer) { + clearTimeout(this.keepaliveResponseWindowTimer) + this.keepaliveResponseWindowTimer = null + } this.status = ConnectionStatus.Disconnected this.emit('disconnected') } diff --git a/src/Ember/Socket/__tests__/S101Socket.spec.ts b/src/Ember/Socket/__tests__/S101Socket.spec.ts new file mode 100644 index 0000000..f013dd4 --- /dev/null +++ b/src/Ember/Socket/__tests__/S101Socket.spec.ts @@ -0,0 +1,57 @@ +import { EventEmitter } from 'eventemitter3' +import S101Socket from '../S101Socket' +import { ConnectionStatus } from '../../Client/ConnectionStatus' + +class FakeSocket extends EventEmitter { + public destroyed = false + public wrote: Buffer[] = [] + + write(data: Buffer): boolean { + this.wrote.push(data) + return true + } + + end(cb?: () => void): void { + cb?.() + this.emit('close') + } + + destroy(): void { + this.destroyed = true + this.emit('close') + } +} + +describe('S101Socket lifecycle', () => { + it('sets disconnected status on socket close', () => { + const socket = new FakeSocket() + const s101Socket = new S101Socket(socket as any) + const onDisconnected = jest.fn() + + s101Socket.on('disconnected', onDisconnected) + expect(s101Socket.status).toBe(ConnectionStatus.Connected) + + socket.emit('close') + + expect(onDisconnected).toHaveBeenCalledTimes(1) + expect(s101Socket.status).toBe(ConnectionStatus.Disconnected) + }) + + it('handleClose tears down socket and marks disconnected', () => { + const socket = new FakeSocket() + const s101Socket = new S101Socket(socket as any) + const onDisconnected = jest.fn() + + s101Socket.on('disconnected', onDisconnected) + ;(s101Socket as any).keepaliveIntervalTimer = setInterval(() => null, 1000) + ;(s101Socket as any).keepaliveResponseWindowTimer = setTimeout(() => null, 1000) + ;(s101Socket as any).handleClose() + + expect(socket.destroyed).toBeTruthy() + expect((s101Socket as any).socket).toBeUndefined() + expect((s101Socket as any).keepaliveIntervalTimer).toBeUndefined() + expect((s101Socket as any).keepaliveResponseWindowTimer).toBeNull() + expect(s101Socket.status).toBe(ConnectionStatus.Disconnected) + expect(onDisconnected).toHaveBeenCalled() + }) +}) diff --git a/src/__mocks__/S101Client.ts b/src/__mocks__/S101Client.ts index 6109082..5d7aa7c 100644 --- a/src/__mocks__/S101Client.ts +++ b/src/__mocks__/S101Client.ts @@ -1,4 +1,4 @@ -import { ConnectionStatus } from '../Ember/Client' +import { ConnectionStatus } from '../Ember/Client/ConnectionStatus' import type OrigS101Client from '../Ember/Socket/S101Client' import { EventEmitter } from 'eventemitter3' import { S101SocketEvents } from '../Ember/Socket/S101Socket' From b0d86011c2e8d1e258e2bbece3370ff31d9b951c Mon Sep 17 00:00:00 2001 From: Peter C <12292660+PeterC89@users.noreply.github.com> Date: Fri, 26 Jun 2026 15:04:34 +0100 Subject: [PATCH 2/3] chore: Address review comments --- src/Ember/Client/__tests__/index.spec.ts | 24 +++++++++++++++++ src/Ember/Client/index.ts | 9 +++++-- src/Ember/Socket/S101Socket.ts | 2 +- src/Ember/Socket/__tests__/S101Socket.spec.ts | 26 +++++++++++++++++++ 4 files changed, 58 insertions(+), 3 deletions(-) diff --git a/src/Ember/Client/__tests__/index.spec.ts b/src/Ember/Client/__tests__/index.spec.ts index b9cb7bc..ef80c48 100644 --- a/src/Ember/Client/__tests__/index.spec.ts +++ b/src/Ember/Client/__tests__/index.spec.ts @@ -491,6 +491,30 @@ describe('client', () => { expect(client.tree[1].children?.[2]).toBeDefined() }) }) + + it('queues one parent-path change when several missing children are inserted', async () => { + await runWithConnection(async (client) => { + const root = new NumberedTreeNodeImpl(1, new EmberNodeImpl('Root', undefined, undefined, true), { + 1: new NumberedTreeNodeImpl(1, new EmberNodeImpl('Existing', undefined, undefined, true)), + }) + if (!root.children?.[1]) throw new Error('Expected seeded child') + root.children[1].parent = root + client.tree[1] = root + + const update = createQualifiedNodeResponse('1', new EmberNodeImpl('Root', undefined, undefined, true), { + 1: new NumberedTreeNodeImpl(1, new EmberNodeImpl('Existing', undefined, undefined, true)), + 2: new NumberedTreeNodeImpl(2, new EmberNodeImpl('Inserted A', undefined, undefined, true)), + 3: new NumberedTreeNodeImpl(3, new EmberNodeImpl('Inserted B', undefined, undefined, true)), + }) + + //@ts-expect-error - private method under regression test + const changes = client._applyRootToTree(update.value) + + expect(changes.filter((change) => change.path === '1')).toHaveLength(1) + expect(client.tree[1].children?.[2]?.parent).toBe(client.tree[1]) + expect(client.tree[1].children?.[3]?.parent).toBe(client.tree[1]) + }) + }) }) describe('StreamManager Integration', () => { diff --git a/src/Ember/Client/index.ts b/src/Ember/Client/index.ts index baa9f3e..733b38b 100644 --- a/src/Ember/Client/index.ts +++ b/src/Ember/Client/index.ts @@ -707,14 +707,16 @@ export class EmberClient extends EventEmitter { } } if (update.children) { + const treePath = getPath(tree) if (!tree.children) { - changes.push({ path: getPath(tree), node: tree }) + changes.push({ path: treePath, node: tree }) tree.children = update.children for (const c of Object.values>(update.children)) { c.parent = tree } } else { // Update existing children and insert missing children without recursing into undefined nodes. + let insertedChild = false for (const child of Object.values>(update.children)) { const i = child.number const oldChild = tree.children[i] @@ -723,9 +725,12 @@ export class EmberClient extends EventEmitter { } else { child.parent = tree tree.children[i] = child - changes.push({ path: getPath(tree), node: tree }) + insertedChild = true } } + if (insertedChild && !changes.some((change) => change.path === treePath && change.node === tree)) { + changes.push({ path: treePath, node: tree }) + } } } diff --git a/src/Ember/Socket/S101Socket.ts b/src/Ember/Socket/S101Socket.ts index 862ba49..151d555 100644 --- a/src/Ember/Socket/S101Socket.ts +++ b/src/Ember/Socket/S101Socket.ts @@ -81,7 +81,6 @@ export default class S101Socket extends EventEmitter { }) this.socket.on('close', () => { - this.emit('disconnected') this.status = ConnectionStatus.Disconnected if (this.keepaliveIntervalTimer) { clearInterval(this.keepaliveIntervalTimer) @@ -93,6 +92,7 @@ export default class S101Socket extends EventEmitter { } this.socket?.removeAllListeners() this.socket = undefined + this.emit('disconnected') }) this.socket.on('error', (e) => { diff --git a/src/Ember/Socket/__tests__/S101Socket.spec.ts b/src/Ember/Socket/__tests__/S101Socket.spec.ts index f013dd4..e3eb5e7 100644 --- a/src/Ember/Socket/__tests__/S101Socket.spec.ts +++ b/src/Ember/Socket/__tests__/S101Socket.spec.ts @@ -37,6 +37,32 @@ describe('S101Socket lifecycle', () => { expect(s101Socket.status).toBe(ConnectionStatus.Disconnected) }) + it('notifies disconnected listeners after close teardown state is finalized', () => { + const socket = new FakeSocket() + const s101Socket = new S101Socket(socket as any) + const observedStates: Array<{ status: ConnectionStatus; keepaliveIntervalTimer: unknown; keepaliveResponseWindowTimer: unknown }> = [] + + ;(s101Socket as any).keepaliveIntervalTimer = setInterval(() => null, 1000) + ;(s101Socket as any).keepaliveResponseWindowTimer = setTimeout(() => null, 1000) + + s101Socket.on('disconnected', () => { + observedStates.push({ + status: s101Socket.status, + keepaliveIntervalTimer: (s101Socket as any).keepaliveIntervalTimer, + keepaliveResponseWindowTimer: (s101Socket as any).keepaliveResponseWindowTimer, + }) + }) + + socket.emit('close') + + expect(observedStates).toHaveLength(1) + expect(observedStates[0]).toMatchObject({ + status: ConnectionStatus.Disconnected, + keepaliveIntervalTimer: undefined, + keepaliveResponseWindowTimer: null, + }) + }) + it('handleClose tears down socket and marks disconnected', () => { const socket = new FakeSocket() const s101Socket = new S101Socket(socket as any) From c044b6732b7ddb18fd2758a1bd9c60a2dbaa57b7 Mon Sep 17 00:00:00 2001 From: Peter C <12292660+PeterC89@users.noreply.github.com> Date: Fri, 26 Jun 2026 16:32:17 +0100 Subject: [PATCH 3/3] chore: Lint --- src/Ember/Socket/__tests__/S101Socket.spec.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/Ember/Socket/__tests__/S101Socket.spec.ts b/src/Ember/Socket/__tests__/S101Socket.spec.ts index e3eb5e7..2f5e7bc 100644 --- a/src/Ember/Socket/__tests__/S101Socket.spec.ts +++ b/src/Ember/Socket/__tests__/S101Socket.spec.ts @@ -40,7 +40,11 @@ describe('S101Socket lifecycle', () => { it('notifies disconnected listeners after close teardown state is finalized', () => { const socket = new FakeSocket() const s101Socket = new S101Socket(socket as any) - const observedStates: Array<{ status: ConnectionStatus; keepaliveIntervalTimer: unknown; keepaliveResponseWindowTimer: unknown }> = [] + const observedStates: Array<{ + status: ConnectionStatus + keepaliveIntervalTimer: unknown + keepaliveResponseWindowTimer: unknown + }> = [] ;(s101Socket as any).keepaliveIntervalTimer = setInterval(() => null, 1000) ;(s101Socket as any).keepaliveResponseWindowTimer = setTimeout(() => null, 1000)