Merge pull request 'fix(remote): scope connection teardown by connection, not device' (#190) from fix/remote-connection-lifecycle into main

Reviewed-on: https://git.i.wylde.net/markwylde/terminay/pulls/190

Author: markwylde

.github/workflows/trigger-release.yml

@@ -477,6 +477,16 @@ jobs:
           VERSION="${TAG#v}"
           node scripts/sync-package-version.mjs "$VERSION"
 
+      - name: Prepare integrity-pinned Secure Werift source mirror
+        run: |
+          node scripts/prove-secure-werift-offline-rebuild.mjs --prepare-mirror "$RUNNER_TEMP/secure-werift-source-mirror"
+
+      - name: Prove clean network-independent Secure Werift rebuild
+        env:
+          npm_config_offline: "true"
+        run: |
+          node scripts/prove-secure-werift-offline-rebuild.mjs --prove-mirror "$RUNNER_TEMP/secure-werift-source-mirror"
+
       - name: Build and pack exact standalone server artifact
         env:
           TAG: ${{ needs.release.outputs.tag }}

apps/terminay-desktop/test/embedded-runtime-artifact.test.mjs

@@ -46,26 +46,6 @@ function run(command, args, options = {}) {
   });
 }
 
-function runFailure(command, args, options = {}) {
-  return new Promise((resolve, reject) => {
-    const child = spawn(command, args, {
-      cwd: options.cwd,
-      env: options.env,
-      stdio: ["ignore", "pipe", "pipe"],
-    });
-    let stdout = "";
-    let stderr = "";
-    child.stdout.setEncoding("utf8");
-    child.stderr.setEncoding("utf8");
-    child.stdout.on("data", (chunk) => { stdout += chunk; });
-    child.stderr.on("data", (chunk) => { stderr += chunk; });
-    child.once("error", reject);
-    child.once("close", (code) => {
-      if (code !== 0) resolve({ code, stdout, stderr });
-      else reject(new Error(`${command} ${args.join(" ")} unexpectedly succeeded`));
-    });
-  });
-}
 
 function startForeground(command, args, options = {}) {
   const child = spawn(command, args, {

apps/terminay-desktop/test/server-bundle-host.test.mjs

@@ -121,7 +121,7 @@ function tar(entries) {
 		header.write('0000000\0', 108, 'ascii');
 		header.write('0000000\0', 116, 'ascii');
 		header.write(
-			body.length.toString(8).padStart(11, '0') + '\0',
+			`${body.length.toString(8).padStart(11, '0')}\0`,
 			124,
 			'ascii',
 		);
@@ -132,7 +132,7 @@ function tar(entries) {
 		header.write('00', 263, 'ascii');
 		let sum = 0;
 		for (const byte of header) sum += byte;
-		header.write(sum.toString(8).padStart(6, '0') + '\0 ', 148, 'ascii');
+		header.write(`${sum.toString(8).padStart(6, '0')}\0 `, 148, 'ascii');
 		records.push(header, body, Buffer.alloc((512 - (body.length % 512)) % 512));
 	}
 	return Buffer.concat([...records, Buffer.alloc(1024)]);

apps/terminay-server/src/index.ts

@@ -14,10 +14,8 @@ export * from "./localUiServer.js";
 export * from "./healthServer.js";
 export * from "./remote/nodeDataChannelRuntime.js";
 export * from "./remote/nodeDataChannelPeer.js";
-export * from "./remote/nodeDataChannelHost.js";
 export * from "./remote/secureWeriftRuntime.js";
 export * from "./remote/secureWeriftPeer.js";
-export * from "./remote/secureWeriftHost.js";
 export * from "./remote/serverExposure.js";
 export * from "./remote/signalingHostBoundary.js";
 export * from "./remote/hostedSignalingRegistrar.js";

apps/terminay-server/src/remote/hostedPairingHost.ts

@@ -30,33 +30,33 @@ import {
 import type { ServerPairingHandoff, ServerRemoteExposure } from './serverExposure.js';
 import { bindUiArchiveChannels, safeChannelSend } from './uiArchiveTransfer.js';
 import {
-	applyHostedLaneDiagnostic,
 	collectHostIceAddresses,
 	createHandshakeJoinQueue,
 	DEVICE_HOST_AVAILABILITY_MS,
 	deviceHostRefreshDelayMs,
 	type HostedIceServer,
-	HostedGenerationSet,
+	HostedLivePeerRegistry,
 	HostedPeerLifecycle,
 	hostedPeerConfiguration,
+	requiredLaneClosed,
 	resolveIceRecoveryGraceMs,
-	shouldFailHostedStall,
 } from './hostedPeerLifecycle.js';
 import { createHostedStreamDiagnostics, frameByteLength } from './hostedStreamDiagnostics.js';
 
 export {
-	applyHostedLaneDiagnostic,
 	collectHostIceAddresses,
 	createHandshakeJoinQueue,
 	DEFAULT_HOSTED_ICE_SERVERS,
 	DEFAULT_ICE_RECOVERY_GRACE_MS,
 	DEVICE_HOST_AVAILABILITY_MS,
 	DEVICE_REFRESH_LEAD_MS,
 	deviceHostRefreshDelayMs,
-	HostedGenerationSet,
+	HostedLivePeerRegistry,
 	HostedPeerLifecycle,
 	hostedPeerConfiguration,
 	parseHostedIceServers,
+	REQUIRED_LANES,
+	requiredLaneClosed,
 	resolveHostedIceServers,
 	resolveIceRecoveryGraceMs,
 } from './hostedPeerLifecycle.js';
@@ -166,14 +166,12 @@ export type HostedPairingDiagnostic = Readonly<{
 	readonly firstInboundAgeMs?: number | null;
 	readonly firstOutboundAgeMs?: number | null;
 	readonly liveGenerationCount?: number;
-	readonly stallIgnored?: boolean;
 	readonly hangup?: boolean;
 	readonly inboundKind?: 'bytes' | 'blob' | 'string' | 'empty' | 'other' | undefined;
 	readonly droppedFrames?: number;
 	readonly droppedClass?: 'bytes' | 'blob' | 'string' | 'empty' | 'other';
 	readonly sendFailures?: number;
 	readonly sendFailure?: boolean;
-	readonly stallClass?: 'no-outbound' | 'outbound-stalled';
 	readonly first?: 'inbound' | 'outbound';
 	readonly summary?: boolean;
 	readonly reasonClass?: string;
@@ -204,8 +202,8 @@ export async function startHostedPairingHost(
 	const archive = options.getUiArchive
 		? await options.getUiArchive()
 		: createMinimalUiArchive();
-	const generations = new HostedGenerationSet();
-	const context = { archive, options, generations };
+	const livePeers = new HostedLivePeerRegistry();
+	const context = { archive, options, livePeers };
 	const joinQueue = createHandshakeJoinQueue();
 	let handshakeGeneration = 0;
 	let handshake:
@@ -234,14 +232,30 @@ export async function startHostedPairingHost(
 		clearTimeout(pairingRefreshTimer);
 		clearTimeout(deviceRefreshTimer);
 		handshake?.peer.close();
-		generations.closeAll();
+		await livePeers.closeAll();
 		pairingGeneration += 1;
 		deviceGeneration += 1;
 		closeSocket(pairingSocket);
 		closeSocket(deviceSocket);
 	};
 
 	async function addHandshakePeer(socket: WebSocket, scope: SignalScope): Promise<void> {
+		// A device gets one live peer. Retire the connection this join replaces
+		// before its replacement exists, so the two never overlap and a late
+		// teardown cannot reach the new session's terminals.
+		if (scope.kind === 'device') {
+			const replaced = await livePeers.close(scope.deviceId);
+			if (replaced !== undefined) {
+				diagnose({ type: 'peer-closed', reasonClass: 'replaced-by-rejoin' });
+				// Report the retirement here rather than relying on the native
+				// datachannel emitting `close` before its peer is torn down. That
+				// event is not guaranteed to arrive, and a missed one would leave a
+				// replaced connection listed as live for the rest of the session.
+				if (replaced.connectionId !== undefined) {
+					options.onPeerDisconnected?.(replaced.connectionId);
+				}
+			}
+		}
 		const generation = ++handshakeGeneration;
 		if (handshake) {
 			handshake.peer.close();
@@ -253,10 +267,18 @@ export async function startHostedPairingHost(
 			scope,
 			context,
 			(connection, peer) => {
-				if (handshake?.peer === next) {
-					generations.add({ peer: next, connection });
-					handshake = undefined;
+				// A newer join replaced this handshake, or the host stopped, while
+				// this one was authenticating. Retire it rather than leaving an
+				// untracked live connection for the device: nothing would close it,
+				// and it is exactly the superseded generation this design removes.
+				if (handshake?.peer !== next || closed) {
+					void connection.close();
+					next.close();
+					diagnose({ type: 'peer-closed', reasonClass: 'replaced-by-rejoin' });
+					return;
 				}
+				livePeers.set(peer.deviceId, { peer: next, connection, connectionId: peer.connectionId });
+				handshake = undefined;
 				options.onPeerConnected?.(peer);
 				if (scope.kind === 'pairing') {
 					clearTimeout(pairingRefreshTimer);
@@ -266,8 +288,8 @@ export async function startHostedPairingHost(
 					pairingRefreshTimer.unref?.();
 				}
 			},
-			(retired) => {
-				generations.drop(retired);
+			(deviceId, retired) => {
+				if (deviceId !== undefined) livePeers.drop(deviceId, retired);
 			},
 		);
 		if (closed || generation !== handshakeGeneration) {
@@ -742,11 +764,11 @@ async function startPeer(
 	scope: SignalScope,
 	context: Readonly<{
 		archive: MinimalArchive;
-		generations: HostedGenerationSet;
+		livePeers: HostedLivePeerRegistry;
 		options: HostedPairingHostOptions;
 	}>,
 	onApplication: (connection: ServerConnectionLike, peer: HostedConnectedPeer) => void,
-	onRetire: (peer: WeriftPeer) => void,
+	onRetire: (deviceId: string | undefined, peer: WeriftPeer) => void,
 ): Promise<WeriftPeer> {
 	const native = new Peer(
 		hostedPeerConfiguration(
@@ -760,15 +782,10 @@ async function startPeer(
 	let wrapped: WeriftPeer;
 	const stream = createHostedStreamDiagnostics({
 		emit: (event) => {
-			const enriched = {
+			context.options.onDiagnostic?.({
 				...event,
-				liveGenerationCount: context.generations.size,
-				...(event.stallClass === undefined
-					? {}
-					: { stallIgnored: !shouldFailHostedStall(event) }),
-			};
-			context.options.onDiagnostic?.(enriched);
-			applyHostedLaneDiagnostic(lifecycle, enriched);
+				liveGenerationCount: context.livePeers.size,
+			});
 		},
 	});
 	lifecycle = new HostedPeerLifecycle(
@@ -777,7 +794,7 @@ async function startPeer(
 		(reason) => {
 			stream.peerClosed(reason);
 			const connectionId = session.peer?.connectionId;
-			onRetire(wrapped);
+			onRetire(session.peer?.deviceId, wrapped);
 			void session.connection?.close();
 			try {
 				native.close();
@@ -807,12 +824,14 @@ async function startPeer(
 	});
 	for (const label of CHANNELS) {
 		const channel = channels[label]!;
+		// A lane that never opened is still negotiating; only a lane that has
+		// carried traffic and then left `open` proves this peer cannot deliver.
+		let everOpened = false;
 		const emitState = () => {
-			stream.channelState(label, channel.readyState);
-			applyHostedLaneDiagnostic(lifecycle, {
-				channel: label,
-				channelState: channel.readyState,
-			});
+			if (channel.readyState === 'open') everOpened = true;
+			const hangup = requiredLaneClosed(label, channel.readyState, everOpened);
+			stream.channelState(label, channel.readyState, hangup);
+			if (hangup) lifecycle.fail(`WebRTC ${label} lane ${channel.readyState}.`);
 		};
 		channel.addEventListener('open', emitState);
 		channel.addEventListener('close', emitState);
@@ -1177,7 +1196,11 @@ function bindControl(
 				deviceName: device?.deviceName?.trim() || 'Browser',
 			});
 			onApplication(connection, peer);
+			// The application lane closing ends this generation. Releasing the
+			// server connection here frees exactly this connection's attachments;
+			// the peer-level required-lane handler retires the peer itself.
 			application.addEventListener('close', () => {
+				void connection.close();
 				context.options.onPeerDisconnected?.(peer.connectionId);
 			});
 			void connection.start().catch((error) => {