From 53b3fa2f4a621fd80cadf0d797917c6d617809a5 Mon Sep 17 00:00:00 2001 From: Hampus Date: Fri, 11 Sep 2026 15:01:14 +0200 Subject: [PATCH] fix(voice): key watch attempts by published track (#2681) --- .../useScreenShareWatchFailure.test.ts | 129 ++++++++++++++++++ .../components/useScreenShareWatchFailure.ts | 25 +++- .../voice/engine/VoiceMediaGraph.test.ts | 127 +++++++++++++++++ .../features/voice/engine/VoiceMediaGraph.ts | 48 ++++++- 4 files changed, 319 insertions(+), 10 deletions(-) create mode 100644 fluxer_app/src/features/voice/components/useScreenShareWatchFailure.test.ts diff --git a/fluxer_app/src/features/voice/components/useScreenShareWatchFailure.test.ts b/fluxer_app/src/features/voice/components/useScreenShareWatchFailure.test.ts new file mode 100644 index 000000000..8b91777be --- /dev/null +++ b/fluxer_app/src/features/voice/components/useScreenShareWatchFailure.test.ts @@ -0,0 +1,129 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {screenShareWatchAttemptKey} from '@app/features/voice/components/useScreenShareWatchFailure'; +import { + createVoiceMediaGraphSnapshot, + selectVoiceMediaGraphAttempt, + selectVoiceMediaGraphDeadline, + transitionVoiceMediaGraph, + type VoiceMediaGraphSnapshot, + voiceMediaGraphAttemptKeyIsOperation, + voiceMediaGraphWatchAttemptDeadlineKey, + WATCH_ATTEMPT_TIMEOUT_MS, +} from '@app/features/voice/engine/VoiceMediaGraph'; +import { + selectVoiceMediaGraphStreamTileState, + type VoiceMediaGraphStreamTileState, +} from '@app/features/voice/engine/VoiceMediaGraphTileState'; +import {VoiceTrackSource} from '@app/features/voice/engine/VoiceTrackSource'; +import {describe, expect, it} from 'vitest'; + +const STREAM_KEY = 'dm:channel-a:connection-a'; +const PARTICIPANT_IDENTITY = 'user_2_connection-a'; + +function watchAttemptKey(trackSid: string | null): string { + return screenShareWatchAttemptKey({streamKey: STREAM_KEY, watchGeneration: 1, trackSid, operationKey: null}); +} + +describe('screenShareWatchAttemptKey', () => { + it('gives a republished track its own attempt key', () => { + expect(watchAttemptKey('TR_2')).not.toBe(watchAttemptKey('TR_1')); + }); + + it('puts the stream, the watch generation and the published track in the key', () => { + expect(watchAttemptKey('TR_1')).toBe(`${STREAM_KEY}:1:TR_1:watch`); + }); + + it('separates a stream with no published track from one with a track', () => { + expect(watchAttemptKey(null)).not.toBe(watchAttemptKey('TR_1')); + }); + + it('marks a republish buffering attempt as an operation attempt', () => { + const buffering = screenShareWatchAttemptKey({ + streamKey: STREAM_KEY, + watchGeneration: 1, + trackSid: 'TR_1', + operationKey: 'republish:3', + }); + + expect(voiceMediaGraphAttemptKeyIsOperation(buffering)).toBe(true); + expect(voiceMediaGraphAttemptKeyIsOperation(watchAttemptKey('TR_1'))).toBe(false); + }); +}); + +function subscribedScreenShareGraph(): VoiceMediaGraphSnapshot { + const graph = transitionVoiceMediaGraph(createVoiceMediaGraphSnapshot(), { + type: 'subscription.subscribe', + participantIdentity: PARTICIPANT_IDENTITY, + source: VoiceTrackSource.ScreenShare, + hasPublication: true, + observedElement: null, + context: 'focused', + }); + return transitionVoiceMediaGraph(graph, {type: 'subscription.clearCommands'}); +} + +function screenShareTileState(graph: VoiceMediaGraphSnapshot): VoiceMediaGraphStreamTileState { + return selectVoiceMediaGraphStreamTileState(graph, { + streamKey: STREAM_KEY, + participantIdentity: PARTICIPANT_IDENTITY, + source: VoiceTrackSource.ScreenShare, + }); +} + +describe('screen share watch attempts across a republish', () => { + it('arms a fresh first frame deadline when a new track replaces a rendered one', () => { + const firstAttemptKey = watchAttemptKey('TR_1'); + let graph = transitionVoiceMediaGraph(subscribedScreenShareGraph(), { + type: 'watch.started', + streamKey: STREAM_KEY, + at: 0, + }); + graph = transitionVoiceMediaGraph(graph, { + type: 'publication.observed', + participantIdentity: PARTICIPANT_IDENTITY, + source: VoiceTrackSource.ScreenShare, + trackSid: 'TR_1', + at: 0, + }); + graph = transitionVoiceMediaGraph(graph, { + type: 'watch.attemptEnsured', + streamKey: STREAM_KEY, + attemptKey: firstAttemptKey, + startedAt: 0, + }); + graph = transitionVoiceMediaGraph(graph, { + type: 'watch.renderedFrame', + streamKey: STREAM_KEY, + attemptKey: firstAttemptKey, + renderedAt: 500, + }); + expect(screenShareTileState(graph)).toBe('rendering'); + + graph = transitionVoiceMediaGraph(graph, { + type: 'publication.lost', + participantIdentity: PARTICIPANT_IDENTITY, + source: VoiceTrackSource.ScreenShare, + at: 4000, + }); + graph = transitionVoiceMediaGraph(graph, { + type: 'publication.observed', + participantIdentity: PARTICIPANT_IDENTITY, + source: VoiceTrackSource.ScreenShare, + trackSid: 'TR_2', + at: 4500, + }); + graph = transitionVoiceMediaGraph(graph, { + type: 'watch.attemptEnsured', + streamKey: STREAM_KEY, + attemptKey: watchAttemptKey('TR_2'), + startedAt: 4600, + }); + + expect(selectVoiceMediaGraphAttempt(graph, STREAM_KEY)?.hasRenderedVideoFrame).toBe(false); + expect(screenShareTileState(graph)).not.toBe('rendering'); + expect(selectVoiceMediaGraphDeadline(graph, voiceMediaGraphWatchAttemptDeadlineKey(STREAM_KEY))?.dueAt).toBe( + 4600 + WATCH_ATTEMPT_TIMEOUT_MS, + ); + }); +}); diff --git a/fluxer_app/src/features/voice/components/useScreenShareWatchFailure.ts b/fluxer_app/src/features/voice/components/useScreenShareWatchFailure.ts index 58125f4b7..3b6c4a320 100644 --- a/fluxer_app/src/features/voice/components/useScreenShareWatchFailure.ts +++ b/fluxer_app/src/features/voice/components/useScreenShareWatchFailure.ts @@ -13,6 +13,7 @@ import type React from 'react'; import {useEffect, useMemo} from 'react'; const SCREEN_SHARE_SOURCE = 'screen_share'; +const MISSING_TRACK_SID = 'no-track'; interface UseScreenShareWatchFailureOptions { enabled: boolean; @@ -64,6 +65,26 @@ function createFailureTarget({ return target; } +interface ScreenShareWatchAttemptKeyOptions { + streamKey: string; + watchGeneration: number; + trackSid?: string | null; + operationKey?: string | number | null; +} + +export function screenShareWatchAttemptKey({ + streamKey, + watchGeneration, + trackSid, + operationKey, +}: ScreenShareWatchAttemptKeyOptions): string { + if (!streamKey) return ''; + const track = trackSid || MISSING_TRACK_SID; + return operationKey == null + ? `${streamKey}:${watchGeneration}:${track}:watch` + : `${streamKey}:${watchGeneration}:${track}:operation:${operationKey}`; +} + export function useScreenShareWatchFailure({ enabled, streamKey, @@ -87,9 +108,7 @@ export function useScreenShareWatchFailure({ ); const watchGeneration = attemptEnabled ? ScreenShareWatchFailures.getWatchGeneration(streamKey) : 0; const attemptKey = attemptEnabled - ? operationKey == null - ? `${streamKey}:${watchGeneration}:watch` - : `${streamKey}:${watchGeneration}:operation:${operationKey}` + ? screenShareWatchAttemptKey({streamKey, watchGeneration, trackSid, operationKey}) : ''; const isOperationBuffering = operationKey != null; const failure = attemptEnabled ? ScreenShareWatchFailures.getFailure(target) : null; diff --git a/fluxer_app/src/features/voice/engine/VoiceMediaGraph.test.ts b/fluxer_app/src/features/voice/engine/VoiceMediaGraph.test.ts index 3b109c385..9241ed635 100644 --- a/fluxer_app/src/features/voice/engine/VoiceMediaGraph.test.ts +++ b/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', () => { diff --git a/fluxer_app/src/features/voice/engine/VoiceMediaGraph.ts b/fluxer_app/src/features/voice/engine/VoiceMediaGraph.ts index e33f47b3c..136b92b49 100644 --- a/fluxer_app/src/features/voice/engine/VoiceMediaGraph.ts +++ b/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; deferredStopKeys: ReadonlySet; @@ -179,6 +182,7 @@ export interface VoiceMediaGraphSnapshot; watchGenerationByStreamKey: ReadonlyMap; attemptsByStreamKey: ReadonlyMap; + watchStopReasonByStreamKey: ReadonlyMap; watchIntent: VoiceMediaGraphWatchIntent; subscriptionsByKey: ReadonlyMap; subscriptionCommands: ReadonlyArray; @@ -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( ), watchGenerationByStreamKey: mapSetBounded(snapshot.watchGenerationByStreamKey, streamKey, generation), attemptsByStreamKey: mapDelete(snapshot.attemptsByStreamKey, streamKey), + watchStopReasonByStreamKey: mapDelete(snapshot.watchStopReasonByStreamKey, streamKey), deadlinesByKey, }; } @@ -430,6 +436,7 @@ function transitionFailureWatchClearAll failuresByKey: new Map(), watchGenerationByStreamKey: new Map(), attemptsByStreamKey: new Map(), + watchStopReasonByStreamKey: new Map(), deadlinesByKey: new Map(), }; } @@ -1205,7 +1212,15 @@ function applyDeferredStopDeadline( ): VoiceMediaGraphSnapshot { if (!deadline.streamKey) return snapshot; if (!snapshot.watchIntent.deferredStopKeys.has(deadline.streamKey)) return snapshot; - return transitionWatchIntentRemoveMany(snapshot, [deadline.streamKey]); + const stopped: VoiceMediaGraphSnapshot = { + ...snapshot, + watchStopReasonByStreamKey: mapSetBounded( + snapshot.watchStopReasonByStreamKey, + deadline.streamKey, + 'republish-grace-expired', + ), + }; + return transitionWatchIntentRemoveMany(stopped, [deadline.streamKey]); } function applyPublicationMissingDeadline( @@ -1415,14 +1430,18 @@ function ensureVoiceMediaGraphWatchAttempt { 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( @@ -1482,6 +1507,7 @@ function markVoiceMediaGraphRenderedFrame = { ...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 { return snapshot.watchIntent.viewerStreamKeys; }