fix: Intermittent subscription update loss and socket lifecycle reliability

src/Ember/Client/ConnectionStatus.ts

@@ -0,0 +1,6 @@
+export enum ConnectionStatus {
+	Error,
+	Disconnected,
+	Connecting,
+	Connected,
+}

src/Ember/Client/__tests__/index.spec.ts

@@ -420,6 +420,103 @@ 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()
+			})
+		})
+
+		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', () => {
 		it('registers stream parameter when subscribing', async () => {
 			await runWithConnection(async (client, socket) => {

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<T> = Promise<RequestPromiseArguments<T>>
 export interface RequestPromiseArguments<T> {
@@ -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<EmberClientEvents> {
 
 		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<Root>(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<EmberClientEvents> {
 			}
 		}
 
-		if (Array.isArray(node)) {
-			return this._sendRequest<Root>(new NumberedTreeNodeImpl(0, command), ExpectResponse.Any)
-		}
-
 		return this._sendCommand<void>(node, command, ExpectResponse.None)
 	}
 	async invoke(
@@ -564,8 +558,12 @@ export class EmberClient extends EventEmitter<EmberClientEvents> {
 
 		// 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,31 @@ export class EmberClient extends EventEmitter<EmberClientEvents> {
 					break
 			}
 		}
-		if (update.children && tree.children) {
-			// Update children
-			for (const child of Object.values<NumberedTreeNode<EmberElement>>(update.children)) {
-				const i = child.number
-				const oldChild = tree.children[i] // as NumberedTreeNode<EmberElement> | 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<NumberedTreeNode<EmberElement>>(update.children)) {
-				c.parent = tree
+		if (update.children) {
+			const treePath = getPath(tree)
+			if (!tree.children) {
+				changes.push({ path: treePath, node: tree })
+				tree.children = update.children
+				for (const c of Object.values<NumberedTreeNode<EmberElement>>(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<NumberedTreeNode<EmberElement>>(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
+						insertedChild = true
+					}
+				}
+				if (insertedChild && !changes.some((change) => change.path === treePath && change.node === tree)) {
+					changes.push({ path: treePath, node: tree })
+				}
 			}
 		}
 

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'

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<S101SocketEvents> {
 	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<S101SocketEvents> {
 		})
 
 		this.codec.on('keepaliveResp', () => {
-			clearInterval(<NodeJS.Timeout>this.keepaliveResponseWindowTimer)
+			if (this.keepaliveResponseWindowTimer) {
+				clearTimeout(this.keepaliveResponseWindowTimer)
+				this.keepaliveResponseWindowTimer = null
+			}
 		})
 
 		this.codec.on('emberPacket', (packet) => {
@@ -78,10 +81,18 @@ export default class S101Socket extends EventEmitter<S101SocketEvents> {
 			})
 
 			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
+				this.emit('disconnected')
 			})
 
 			this.socket.on('error', (e) => {
@@ -102,6 +113,10 @@ export default class S101Socket extends EventEmitter<S101SocketEvents> {
 				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<S101SocketEvents> {
 	 *
 	 */
 	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')
 	}