fix(voice): key watch attempts by published track (#2681)

fluxer_app/src/features/voice/engine/VoiceMediaGraph.test.ts

@@ -21,6 +21,7 @@ import {
 	selectVoiceMediaGraphSubscriptionEntry,
 	selectVoiceMediaGraphViewerStreamKeys,
 	selectVoiceMediaGraphWatchGeneration,
+	selectVoiceMediaGraphWatchStopReason,
 	transitionVoiceMediaGraph,
 	transitionVoiceMediaGraphViewerStreamKeys,
 	type VoiceMediaGraphDeadline,
@@ -85,6 +86,7 @@ describe('VoiceMediaGraph watch generations', () => {
 			startedAt: 1000,
 			hasRenderedVideoFrame: false,
 			generation: 0,
+			holders: 1,
 		});
 	});
 
@@ -760,6 +762,131 @@ describe('VoiceMediaGraph deadlines', () => {
 		expect(recorded?.code).toBe(-2301);
 		expect(recorded?.reason).toBe('publication-missing-timeout');
 	});
+
+	it('records the expired republish grace as the stop reason for the dropped stream', () => {
+		let graph = transitionVoiceMediaGraph(createVoiceMediaGraphSnapshot(), {type: 'watchIntent.add', key: STREAM_A});
+		graph = transitionVoiceMediaGraph(graph, {type: 'watchIntent.deferRemove', key: STREAM_A, at: 1000});
+
+		graph = transitionVoiceMediaGraph(graph, {
+			type: 'time.deadlineFired',
+			key: voiceMediaGraphDeferredStopDeadlineKey(STREAM_A),
+			at: 1000 + PUBLISHER_REPUBLISH_GRACE_MS,
+		});
+
+		expect(selectVoiceMediaGraphViewerStreamKeys(graph)).toEqual([]);
+		expect(selectVoiceMediaGraphWatchStopReason(graph, STREAM_A)).toBe('republish-grace-expired');
+	});
+
+	it('forgets the recorded stop reason once the stream is watched again', () => {
+		let graph = transitionVoiceMediaGraph(createVoiceMediaGraphSnapshot(), {type: 'watchIntent.add', key: STREAM_A});
+		graph = transitionVoiceMediaGraph(graph, {type: 'watchIntent.deferRemove', key: STREAM_A, at: 1000});
+		graph = transitionVoiceMediaGraph(graph, {
+			type: 'time.deadlineFired',
+			key: voiceMediaGraphDeferredStopDeadlineKey(STREAM_A),
+			at: 1000 + PUBLISHER_REPUBLISH_GRACE_MS,
+		});
+
+		graph = transitionVoiceMediaGraph(graph, {type: 'watch.started', streamKey: STREAM_A, at: 6000});
+
+		expect(selectVoiceMediaGraphWatchStopReason(graph, STREAM_A)).toBeNull();
+	});
+
+	it('keeps the recorded stop reason through the watch stop that follows the grace expiry', () => {
+		let graph = transitionVoiceMediaGraph(createVoiceMediaGraphSnapshot(), {type: 'watchIntent.add', key: STREAM_A});
+		graph = transitionVoiceMediaGraph(graph, {type: 'watchIntent.deferRemove', key: STREAM_A, at: 1000});
+		graph = transitionVoiceMediaGraph(graph, {
+			type: 'time.deadlineFired',
+			key: voiceMediaGraphDeferredStopDeadlineKey(STREAM_A),
+			at: 1000 + PUBLISHER_REPUBLISH_GRACE_MS,
+		});
+
+		graph = transitionVoiceMediaGraph(graph, {type: 'watch.stopped', streamKey: STREAM_A});
+
+		expect(selectVoiceMediaGraphWatchStopReason(graph, STREAM_A)).toBe('republish-grace-expired');
+	});
+});
+
+describe('VoiceMediaGraph watch attempt holders', () => {
+	const ATTEMPT_KEY = 'attempt-1';
+
+	function ensureAttempt(graph: VoiceMediaGraphSnapshot, startedAt: number): VoiceMediaGraphSnapshot {
+		return transitionVoiceMediaGraph(graph, {
+			type: 'watch.attemptEnsured',
+			streamKey: STREAM_A,
+			attemptKey: ATTEMPT_KEY,
+			startedAt,
+		});
+	}
+
+	function releaseAttempt(graph: VoiceMediaGraphSnapshot): VoiceMediaGraphSnapshot {
+		return transitionVoiceMediaGraph(graph, {
+			type: 'watch.attemptReleased',
+			streamKey: STREAM_A,
+			attemptKey: ATTEMPT_KEY,
+		});
+	}
+
+	function watchedStream(at: number): VoiceMediaGraphSnapshot {
+		return transitionVoiceMediaGraph(createVoiceMediaGraphSnapshot(), {
+			type: 'watch.started',
+			streamKey: STREAM_A,
+			at,
+		});
+	}
+
+	it('keeps the attempt deadline while a second tile still holds the attempt', () => {
+		let graph = ensureAttempt(watchedStream(1000), 1000);
+		graph = ensureAttempt(graph, 1200);
+
+		graph = releaseAttempt(graph);
+
+		expect(selectVoiceMediaGraphAttempt(graph, STREAM_A)?.holders).toBe(1);
+		expect(selectVoiceMediaGraphDeadline(graph, voiceMediaGraphWatchAttemptDeadlineKey(STREAM_A))?.dueAt).toBe(
+			1000 + WATCH_ATTEMPT_TIMEOUT_MS,
+		);
+	});
+
+	it('drops the attempt deadline once the last tile releases the attempt', () => {
+		let graph = ensureAttempt(watchedStream(1000), 1000);
+		graph = ensureAttempt(graph, 1200);
+
+		graph = releaseAttempt(releaseAttempt(graph));
+
+		expect(selectVoiceMediaGraphAttempt(graph, STREAM_A)?.holders).toBe(0);
+		expect(selectVoiceMediaGraphDeadline(graph, voiceMediaGraphWatchAttemptDeadlineKey(STREAM_A))).toBeNull();
+	});
+
+	it('rearms a released attempt from the time the attempt started', () => {
+		let graph = ensureAttempt(watchedStream(1000), 1000);
+		graph = releaseAttempt(graph);
+
+		graph = ensureAttempt(graph, 9000);
+
+		expect(selectVoiceMediaGraphAttempt(graph, STREAM_A)?.startedAt).toBe(1000);
+		expect(selectVoiceMediaGraphDeadline(graph, voiceMediaGraphWatchAttemptDeadlineKey(STREAM_A))?.dueAt).toBe(
+			1000 + WATCH_ATTEMPT_TIMEOUT_MS,
+		);
+	});
+
+	it('counts one holder per tile when every mount runs its effect twice', () => {
+		let graph = ensureAttempt(watchedStream(1000), 1000);
+		graph = ensureAttempt(releaseAttempt(graph), 1400);
+		graph = ensureAttempt(graph, 1500);
+		graph = ensureAttempt(releaseAttempt(graph), 1600);
+
+		expect(selectVoiceMediaGraphAttempt(graph, STREAM_A)?.holders).toBe(2);
+
+		graph = releaseAttempt(releaseAttempt(graph));
+
+		expect(selectVoiceMediaGraphAttempt(graph, STREAM_A)?.holders).toBe(0);
+		expect(selectVoiceMediaGraphDeadline(graph, voiceMediaGraphWatchAttemptDeadlineKey(STREAM_A))).toBeNull();
+	});
+
+	it('ignores a release for an attempt the graph no longer tracks', () => {
+		const graph = watchedStream(1000);
+
+		expect(releaseAttempt(graph)).toBe(graph);
+	});
 });
 
 describe('VoiceMediaGraph desired and actual subscription state', () => {

fluxer_app/src/features/voice/engine/VoiceMediaGraph.ts

@@ -159,8 +159,11 @@ export interface VoiceMediaGraphWatchAttempt {
 	startedAt: number;
 	hasRenderedVideoFrame: boolean;
 	generation: number;
+	holders: number;
 }
 
+export type VoiceMediaGraphWatchStopReason = 'republish-grace-expired';
+
 export interface VoiceMediaGraphWatchIntent {
 	viewerStreamKeys: ReadonlyArray<string>;
 	deferredStopKeys: ReadonlySet<string>;
@@ -179,6 +182,7 @@ export interface VoiceMediaGraphSnapshot<TFailure extends VoiceMediaGraphFailure
 	failuresByKey: ReadonlyMap<string, TFailure>;
 	watchGenerationByStreamKey: ReadonlyMap<string, number>;
 	attemptsByStreamKey: ReadonlyMap<string, VoiceMediaGraphWatchAttempt>;
+	watchStopReasonByStreamKey: ReadonlyMap<string, VoiceMediaGraphWatchStopReason>;
 	watchIntent: VoiceMediaGraphWatchIntent;
 	subscriptionsByKey: ReadonlyMap<string, VoiceMediaGraphSubscriptionEntry>;
 	subscriptionCommands: ReadonlyArray<VoiceMediaGraphSubscriptionCommand>;
@@ -218,6 +222,7 @@ export function createVoiceMediaGraphSnapshot<
 		failuresByKey: new Map(),
 		watchGenerationByStreamKey: new Map(),
 		attemptsByStreamKey: new Map(),
+		watchStopReasonByStreamKey: new Map(),
 		watchIntent: {
 			viewerStreamKeys: [],
 			deferredStopKeys: EMPTY_STRING_SET,
@@ -402,6 +407,7 @@ function transitionWatchStarted<TFailure extends VoiceMediaGraphFailure>(
 		),
 		watchGenerationByStreamKey: mapSetBounded(snapshot.watchGenerationByStreamKey, streamKey, generation),
 		attemptsByStreamKey: mapDelete(snapshot.attemptsByStreamKey, streamKey),
+		watchStopReasonByStreamKey: mapDelete(snapshot.watchStopReasonByStreamKey, streamKey),
 		deadlinesByKey,
 	};
 }
@@ -430,6 +436,7 @@ function transitionFailureWatchClearAll<TFailure extends VoiceMediaGraphFailure>
 		failuresByKey: new Map(),
 		watchGenerationByStreamKey: new Map(),
 		attemptsByStreamKey: new Map(),
+		watchStopReasonByStreamKey: new Map(),
 		deadlinesByKey: new Map(),
 	};
 }
@@ -1205,7 +1212,15 @@ function applyDeferredStopDeadline<TFailure extends VoiceMediaGraphFailure>(
 ): VoiceMediaGraphSnapshot<TFailure> {
 	if (!deadline.streamKey) return snapshot;
 	if (!snapshot.watchIntent.deferredStopKeys.has(deadline.streamKey)) return snapshot;
-	return transitionWatchIntentRemoveMany(snapshot, [deadline.streamKey]);
+	const stopped: VoiceMediaGraphSnapshot<TFailure> = {
+		...snapshot,
+		watchStopReasonByStreamKey: mapSetBounded(
+			snapshot.watchStopReasonByStreamKey,
+			deadline.streamKey,
+			'republish-grace-expired',
+		),
+	};
+	return transitionWatchIntentRemoveMany(stopped, [deadline.streamKey]);
 }
 
 function applyPublicationMissingDeadline<TFailure extends VoiceMediaGraphFailure>(
@@ -1415,14 +1430,18 @@ function ensureVoiceMediaGraphWatchAttempt<TFailure extends VoiceMediaGraphFailu
 	const deadlineKey = voiceMediaGraphWatchAttemptDeadlineKey(event.streamKey);
 	const existing = snapshot.attemptsByStreamKey.get(event.streamKey);
 	if (existing?.attemptKey === event.attemptKey) {
-		if (existing.hasRenderedVideoFrame) return snapshot;
-		if (snapshot.deadlinesByKey.has(deadlineKey)) return snapshot;
+		const attemptsByStreamKey = mapSetBounded(snapshot.attemptsByStreamKey, event.streamKey, {
+			...existing,
+			holders: existing.holders + 1,
+		});
+		if (existing.hasRenderedVideoFrame) return {...snapshot, attemptsByStreamKey};
+		if (snapshot.deadlinesByKey.has(deadlineKey)) return {...snapshot, attemptsByStreamKey};
 		const deadlinesByKey = mapSetBounded(
 			snapshot.deadlinesByKey,
 			deadlineKey,
-			voiceMediaGraphWatchAttemptDeadline(event.streamKey, event.attemptKey, existing.generation, event.startedAt),
+			voiceMediaGraphWatchAttemptDeadline(event.streamKey, event.attemptKey, existing.generation, existing.startedAt),
 		);
-		return {...snapshot, deadlinesByKey};
+		return {...snapshot, attemptsByStreamKey, deadlinesByKey};
 	}
 	const generation = event.generation ?? snapshot.watchGenerationByStreamKey.get(event.streamKey) ?? 0;
 	const attempt: VoiceMediaGraphWatchAttempt = {
@@ -1430,6 +1449,7 @@ function ensureVoiceMediaGraphWatchAttempt<TFailure extends VoiceMediaGraphFailu
 		startedAt: event.startedAt,
 		hasRenderedVideoFrame: false,
 		generation,
+		holders: 1,
 	};
 	const deadlinesByKey = mapSetBounded(
 		snapshot.deadlinesByKey,
@@ -1448,10 +1468,15 @@ function releaseVoiceMediaGraphWatchAttempt<TFailure extends VoiceMediaGraphFail
 	event: {streamKey: string; attemptKey: string},
 ): VoiceMediaGraphSnapshot<TFailure> {
 	if (!event.streamKey) return snapshot;
+	const existing = snapshot.attemptsByStreamKey.get(event.streamKey);
+	if (!existing || existing.attemptKey !== event.attemptKey) return snapshot;
+	const holders = Math.max(existing.holders - 1, 0);
+	const attemptsByStreamKey = mapSetBounded(snapshot.attemptsByStreamKey, event.streamKey, {...existing, holders});
+	if (holders > 0) return {...snapshot, attemptsByStreamKey};
 	const deadlineKey = voiceMediaGraphWatchAttemptDeadlineKey(event.streamKey);
 	const deadline = snapshot.deadlinesByKey.get(deadlineKey);
-	if (!deadline || deadline.attemptKey !== event.attemptKey) return snapshot;
-	return {...snapshot, deadlinesByKey: mapDelete(snapshot.deadlinesByKey, deadlineKey)};
+	if (!deadline || deadline.attemptKey !== event.attemptKey) return {...snapshot, attemptsByStreamKey};
+	return {...snapshot, attemptsByStreamKey, deadlinesByKey: mapDelete(snapshot.deadlinesByKey, deadlineKey)};
 }
 
 function markRenderedFrameOnSubscriptionEntry<TFailure extends VoiceMediaGraphFailure>(
@@ -1482,6 +1507,7 @@ function markVoiceMediaGraphRenderedFrame<TFailure extends VoiceMediaGraphFailur
 		startedAt: existing?.startedAt ?? event.renderedAt,
 		hasRenderedVideoFrame: true,
 		generation: existing?.generation ?? snapshot.watchGenerationByStreamKey.get(event.streamKey) ?? 0,
+		holders: existing?.holders ?? 0,
 	};
 	const next: VoiceMediaGraphSnapshot<TFailure> = {
 		...snapshot,
@@ -1560,6 +1586,14 @@ export function selectVoiceMediaGraphAttempt(
 	return snapshot.attemptsByStreamKey.get(streamKey) ?? null;
 }
 
+export function selectVoiceMediaGraphWatchStopReason(
+	snapshot: VoiceMediaGraphSnapshot,
+	streamKey: string,
+): VoiceMediaGraphWatchStopReason | null {
+	if (!streamKey) return null;
+	return snapshot.watchStopReasonByStreamKey.get(streamKey) ?? null;
+}
+
 export function selectVoiceMediaGraphViewerStreamKeys(snapshot: VoiceMediaGraphSnapshot): ReadonlyArray<string> {
 	return snapshot.watchIntent.viewerStreamKeys;
 }