Merge pull request #3537 from robintown/connection-leaks

Fix resource leaks when we stop using a connection

Author: robintown

src/main.tsx

@@ -24,6 +24,7 @@ import { App } from "./App";
 import { init as initRageshake } from "./settings/rageshake";
 import { Initializer } from "./initializer";
 import { AppViewModel } from "./state/AppViewModel";
+import { globalScope } from "./state/ObservableScope";
 
 window.setLKLogLevel = setLKLogLevel;
 
@@ -61,7 +62,7 @@ Initializer.initBeforeReact()
   .then(() => {
     root.render(
       <StrictMode>
-        <App vm={new AppViewModel()} />,
+        <App vm={new AppViewModel(globalScope)} />,
       </StrictMode>,
     );
   })

src/reactions/ReactionsReader.test.tsx

@@ -23,7 +23,7 @@ import {
   localRtcMember,
 } from "../utils/test-fixtures";
 import { getBasicRTCSession } from "../utils/test-viewmodel";
-import { withTestScheduler } from "../utils/test";
+import { testScope, withTestScheduler } from "../utils/test";
 import { ElementCallReactionEventType, ReactionSet } from ".";
 
 afterEach(() => {
@@ -37,6 +37,7 @@ test("handles a hand raised reaction", () => {
   withTestScheduler(({ schedule, expectObservable }) => {
     renderHook(() => {
       const { raisedHands$ } = new ReactionsReader(
+        testScope(),
         rtcSession.asMockedSession(),
       );
       schedule("ab", {
@@ -85,6 +86,7 @@ test("handles a redaction", () => {
   withTestScheduler(({ schedule, expectObservable }) => {
     renderHook(() => {
       const { raisedHands$ } = new ReactionsReader(
+        testScope(),
         rtcSession.asMockedSession(),
       );
       schedule("abc", {
@@ -148,6 +150,7 @@ test("handles waiting for event decryption", () => {
   withTestScheduler(({ schedule, expectObservable }) => {
     renderHook(() => {
       const { raisedHands$ } = new ReactionsReader(
+        testScope(),
         rtcSession.asMockedSession(),
       );
       schedule("abc", {
@@ -217,6 +220,7 @@ test("hands rejecting events without a proper membership", () => {
   withTestScheduler(({ schedule, expectObservable }) => {
     renderHook(() => {
       const { raisedHands$ } = new ReactionsReader(
+        testScope(),
         rtcSession.asMockedSession(),
       );
       schedule("ab", {
@@ -261,7 +265,10 @@ test("handles a reaction", () => {
 
   withTestScheduler(({ schedule, time, expectObservable }) => {
     renderHook(() => {
-      const { reactions$ } = new ReactionsReader(rtcSession.asMockedSession());
+      const { reactions$ } = new ReactionsReader(
+        testScope(),
+        rtcSession.asMockedSession(),
+      );
       schedule(`abc`, {
         a: () => {},
         b: () => {
@@ -317,7 +324,10 @@ test("ignores bad reaction events", () => {
 
   withTestScheduler(({ schedule, expectObservable }) => {
     renderHook(() => {
-      const { reactions$ } = new ReactionsReader(rtcSession.asMockedSession());
+      const { reactions$ } = new ReactionsReader(
+        testScope(),
+        rtcSession.asMockedSession(),
+      );
       schedule("ab", {
         a: () => {},
         b: () => {
@@ -439,7 +449,10 @@ test("that reactions cannot be spammed", () => {
 
   withTestScheduler(({ schedule, expectObservable }) => {
     renderHook(() => {
-      const { reactions$ } = new ReactionsReader(rtcSession.asMockedSession());
+      const { reactions$ } = new ReactionsReader(
+        testScope(),
+        rtcSession.asMockedSession(),
+      );
       schedule("abcd", {
         a: () => {},
         b: () => {

src/reactions/ReactionsReader.ts

@@ -18,7 +18,7 @@ import {
   EventType,
   RoomEvent as MatrixRoomEvent,
 } from "matrix-js-sdk";
-import { BehaviorSubject, delay, type Subscription } from "rxjs";
+import { BehaviorSubject, delay } from "rxjs";
 
 import {
   ElementCallReactionEventType,
@@ -28,6 +28,7 @@ import {
   type RaisedHandInfo,
   type ReactionInfo,
 } from ".";
+import { type ObservableScope } from "../state/ObservableScope";
 
 export const REACTION_ACTIVE_TIME_MS = 3000;
 
@@ -54,12 +55,13 @@ export class ReactionsReader {
    */
   public readonly reactions$ = this.reactionsSubject$.asObservable();
 
-  private readonly reactionsSub: Subscription;
-
-  public constructor(private readonly rtcSession: MatrixRTCSession) {
+  public constructor(
+    private readonly scope: ObservableScope,
+    private readonly rtcSession: MatrixRTCSession,
+  ) {
     // Hide reactions after a given time.
-    this.reactionsSub = this.reactionsSubject$
-      .pipe(delay(REACTION_ACTIVE_TIME_MS))
+    this.reactionsSubject$
+      .pipe(delay(REACTION_ACTIVE_TIME_MS), this.scope.bind())
       .subscribe((reactions) => {
         const date = new Date();
         const nextEntries = Object.fromEntries(
@@ -71,27 +73,62 @@ export class ReactionsReader {
         this.reactionsSubject$.next(nextEntries);
       });
 
+    // TODO: Convert this class to the functional reactive style and get rid of
+    // all this manual setup and teardown for event listeners
+
     this.rtcSession.room.on(MatrixRoomEvent.Timeline, this.handleReactionEvent);
+    this.scope.onEnd(() =>
+      this.rtcSession.room.off(
+        MatrixRoomEvent.Timeline,
+        this.handleReactionEvent,
+      ),
+    );
+
     this.rtcSession.room.on(
       MatrixRoomEvent.Redaction,
       this.handleReactionEvent,
     );
+    this.scope.onEnd(() =>
+      this.rtcSession.room.off(
+        MatrixRoomEvent.Redaction,
+        this.handleReactionEvent,
+      ),
+    );
+
     this.rtcSession.room.client.on(
       MatrixEventEvent.Decrypted,
       this.handleReactionEvent,
     );
+    this.scope.onEnd(() =>
+      this.rtcSession.room.client.off(
+        MatrixEventEvent.Decrypted,
+        this.handleReactionEvent,
+      ),
+    );
 
     // We listen for a local echo to get the real event ID, as timeline events
     // may still be sending.
     this.rtcSession.room.on(
       MatrixRoomEvent.LocalEchoUpdated,
       this.handleReactionEvent,
     );
+    this.scope.onEnd(() =>
+      this.rtcSession.room.off(
+        MatrixRoomEvent.LocalEchoUpdated,
+        this.handleReactionEvent,
+      ),
+    );
 
-    rtcSession.on(
+    this.rtcSession.on(
       MatrixRTCSessionEvent.MembershipsChanged,
       this.onMembershipsChanged,
     );
+    this.scope.onEnd(() =>
+      this.rtcSession.off(
+        MatrixRTCSessionEvent.MembershipsChanged,
+        this.onMembershipsChanged,
+      ),
+    );
 
     // Run this once to ensure we have fetched the state from the call.
     this.onMembershipsChanged([]);
@@ -309,31 +346,4 @@ export class ReactionsReader {
       this.removeRaisedHand(targetUser);
     }
   };
-
-  /**
-   * Stop listening for events.
-   */
-  public destroy(): void {
-    this.rtcSession.off(
-      MatrixRTCSessionEvent.MembershipsChanged,
-      this.onMembershipsChanged,
-    );
-    this.rtcSession.room.off(
-      MatrixRoomEvent.Timeline,
-      this.handleReactionEvent,
-    );
-    this.rtcSession.room.off(
-      MatrixRoomEvent.Redaction,
-      this.handleReactionEvent,
-    );
-    this.rtcSession.room.client.off(
-      MatrixEventEvent.Decrypted,
-      this.handleReactionEvent,
-    );
-    this.rtcSession.room.off(
-      MatrixRoomEvent.LocalEchoUpdated,
-      this.handleReactionEvent,
-    );
-    this.reactionsSub.unsubscribe();
-  }
 }

src/room/InCallView.tsx

@@ -110,6 +110,7 @@ import ringtoneMp3 from "../sound/ringtone.mp3?url";
 import ringtoneOgg from "../sound/ringtone.ogg?url";
 import { useTrackProcessorObservable$ } from "../livekit/TrackProcessorContext.tsx";
 import { type Layout } from "../state/layout-types.ts";
+import { ObservableScope } from "../state/ObservableScope.ts";
 
 const maxTapDurationMs = 400;
 
@@ -129,8 +130,10 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
 
   const trackProcessorState$ = useTrackProcessorObservable$();
   useEffect(() => {
-    const reactionsReader = new ReactionsReader(props.rtcSession);
+    const scope = new ObservableScope();
+    const reactionsReader = new ReactionsReader(scope, props.rtcSession);
     const vm = new CallViewModel(
+      scope,
       props.rtcSession,
       props.matrixRoom,
       mediaDevices,
@@ -146,11 +149,9 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
     );
     setVm(vm);
 
-    const sub = vm.leave$.subscribe(props.onLeft);
+    vm.leave$.pipe(scope.bind()).subscribe(props.onLeft);
     return (): void => {
-      vm.destroy();
-      sub.unsubscribe();
-      reactionsReader.destroy();
+      scope.end();
     };
   }, [
     props.rtcSession,

src/room/MuteStates.test.tsx

@@ -108,9 +108,7 @@ function mockMediaDevices(
         throw new Error("Unimplemented");
     }
   });
-  const scope = new ObservableScope();
-  onTestFinished(() => scope.end());
-  return new MediaDevices(scope);
+  return new MediaDevices(testScope());
 }
 
 describe("useMuteStates VITE_PACKAGE='full' (SPA) mode", () => {

src/state/AppViewModel.ts

@@ -6,14 +6,16 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { MediaDevices } from "./MediaDevices";
-import { ViewModel } from "./ViewModel";
+import { type ObservableScope } from "./ObservableScope";
 
 /**
  * The top-level state holder for the application.
  */
-export class AppViewModel extends ViewModel {
+export class AppViewModel {
   public readonly mediaDevices = new MediaDevices(this.scope);
 
   // TODO: Move more application logic here. The CallViewModel, at the very
   // least, ought to be accessible from this object.
+
+  public constructor(private readonly scope: ObservableScope) {}
 }

src/state/CallViewModel.test.ts

@@ -60,6 +60,7 @@ import {
   mockMediaDevices,
   mockMuteStates,
   mockConfig,
+  testScope,
 } from "../utils/test";
 import {
   ECAddonConnectionState,
@@ -89,7 +90,6 @@ import {
   localRtcMember,
   localRtcMemberDevice2,
 } from "../utils/test-fixtures";
-import { ObservableScope } from "./ObservableScope";
 import { MediaDevices } from "./MediaDevices";
 import { getValue } from "../utils/observable";
 import { type Behavior, constant } from "./Behavior";
@@ -347,6 +347,7 @@ function withCallViewModel(
   const reactions$ = new BehaviorSubject<Record<string, ReactionInfo>>({});
 
   const vm = new CallViewModel(
+    testScope(),
     rtcSession.asMockedSession(),
     room,
     mediaDevices,
@@ -361,7 +362,6 @@ function withCallViewModel(
   );
 
   onTestFinished(() => {
-    vm!.destroy();
     participantsSpy!.mockRestore();
     mediaSpy!.mockRestore();
     eventsSpy!.mockRestore();
@@ -402,6 +402,7 @@ test("test missing RTC config error", async () => {
   vi.spyOn(AutoDiscovery, "getRawClientConfig").mockResolvedValue({});
 
   const callVM = new CallViewModel(
+    testScope(),
     fakeRtcSession.asMockedSession(),
     matrixRoom,
     mockMediaDevices({}),
@@ -1630,9 +1631,7 @@ test("audio output changes when toggling earpiece mode", () => {
     getUrlParams.mockReturnValue({ controlledAudioDevices: true });
     vi.mocked(ComponentsCore.createMediaDeviceObserver).mockReturnValue(of([]));
 
-    const scope = new ObservableScope();
-    onTestFinished(() => scope.end());
-    const devices = new MediaDevices(scope);
+    const devices = new MediaDevices(testScope());
 
     window.controls.setAvailableAudioDevices([
       { id: "speaker", name: "Speaker", isSpeaker: true },