Fix: never stop ring feedback on the sender side. (#3502)

* make ring$ a behavior and add code comments to justify/explain the change.

Signed-off-by: Timo K <toger5@hotmail.de>

* Add test: reproduce "ring does not stop" race.

Signed-off-by: Timo K <toger5@hotmail.de>

---------

Signed-off-by: Timo K <toger5@hotmail.de>

src/state/CallViewModel.test.ts

@@ -266,7 +266,7 @@ const mockLegacyRingEvent = {} as { event_id: string } & ICallNotifyContent;
 interface CallViewModelInputs {
   remoteParticipants$: Behavior<RemoteParticipant[]>;
   rtcMembers$: Behavior<Partial<CallMembership>[]>;
-  connectionState$: Observable<ECConnectionState>;
+  livekitConnectionState$: Observable<ECConnectionState>;
   speaking: Map<Participant, Observable<boolean>>;
   mediaDevices: MediaDevices;
   initialSyncState: SyncState;
@@ -276,7 +276,7 @@ function withCallViewModel(
   {
     remoteParticipants$ = constant([]),
     rtcMembers$ = constant([localRtcMember]),
-    connectionState$ = of(ConnectionState.Connected),
+    livekitConnectionState$: connectionState$ = of(ConnectionState.Connected),
     speaking = new Map(),
     mediaDevices = mockMediaDevices({}),
     initialSyncState = SyncState.Syncing,
@@ -384,7 +384,7 @@ test("participants are retained during a focus switch", () => {
           b: [],
         }),
         rtcMembers$: constant([localRtcMember, aliceRtcMember, bobRtcMember]),
-        connectionState$: behavior(connectionInputMarbles, {
+        livekitConnectionState$: behavior(connectionInputMarbles, {
           c: ConnectionState.Connected,
           s: ECAddonConnectionState.ECSwitchingFocus,
         }),
@@ -1251,6 +1251,41 @@ describe("waitForCallPickup$", () => {
     });
   });
 
+  test("regression test: does stop ringing in case livekitConnectionState$ emits after didSendCallNotification$ has already emitted", () => {
+    withTestScheduler(({ schedule, expectObservable, behavior }) => {
+      withCallViewModel(
+        {
+          livekitConnectionState$: behavior("d 9ms c", {
+            d: ConnectionState.Disconnected,
+            c: ConnectionState.Connected,
+          }),
+        },
+        (vm, rtcSession) => {
+          // Fire a call notification IMMEDIATELY (its important for this test, that this happens before the livekitConnectionState$ emits)
+          schedule("n", {
+            n: () => {
+              rtcSession.emit(
+                MatrixRTCSessionEvent.DidSendCallNotification,
+                mockRingEvent("$notif1", 30),
+                mockLegacyRingEvent,
+              );
+            },
+          });
+
+          expectObservable(vm.callPickupState$).toBe("a 9ms b 29ms c", {
+            a: "unknown",
+            b: "ringing",
+            c: "timeout",
+          });
+        },
+        {
+          waitForCallPickup: true,
+          encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
+        },
+      );
+    });
+  });
+
   test("ringing -> success if someone joins before timeout", () => {
     withTestScheduler(({ behavior, schedule, expectObservable }) => {
       // Someone joins at 20ms (both LiveKit participant and MatrixRTC member)
@@ -1305,7 +1340,7 @@ describe("waitForCallPickup$", () => {
             a: [localRtcMember],
             b: [localRtcMember, aliceRtcMember],
           }),
-          connectionState$,
+          livekitConnectionState$: connectionState$,
         },
         (vm, rtcSession) => {
           // Notify at 5ms so we enter ringing, then get disconnected 5ms later

src/state/CallViewModel.ts

@@ -880,23 +880,28 @@ export class CallViewModel extends ViewModel {
     ? this.allOthersLeft$
     : NEVER;
 
-  /**
+  private readonly didSendCallNotification$ = fromEvent(
-   * Whenever the RTC session tells us that it intends to ring the remote
-   * participant's devices, this emits an Observable tracking the current state of
-   * that ringing process.
-   */
-  private readonly ring$: Observable<
-    Observable<"ringing" | "timeout" | "decline">
-  > = (
-    fromEvent(
     this.matrixRTCSession,
     MatrixRTCSessionEvent.DidSendCallNotification,
   ) as Observable<
     Parameters<
       MatrixRTCSessionEventHandlerMap[MatrixRTCSessionEvent.DidSendCallNotification]
     >
-    >
+  >;
-  ).pipe(
+  /**
+   * Whenever the RTC session tells us that it intends to ring the remote
+   * participant's devices, this emits an Observable tracking the current state of
+   * that ringing process.
+   */
+  // This is a behavior since we need to store the latest state for when we subscribe to this after `didSendCallNotification$`
+  // has already emitted but we still need the latest observable with a timeout timer that only gets created on after receiving `notificationEvent`.
+  // A behavior will emit the latest observable with the running timer to new subscribers.
+  // see also: callPickupState$ and in particular the line: `return this.ring$.pipe(mergeAll());` here we otherwise might get an EMPTY observable if
+  // `ring$` would not be a behavior.
+  private readonly ring$: Behavior<
+    Observable<"ringing" | "timeout" | "decline"> | Observable<never>
+  > = this.scope.behavior(
+    this.didSendCallNotification$.pipe(
       filter(
         ([notificationEvent]) => notificationEvent.notification_type === "ring",
       ),
@@ -926,7 +931,8 @@ export class CallViewModel extends ViewModel {
                 ([event]) =>
                   event.getType() === EventType.RTCDecline &&
                   event.getRelation()?.rel_type === "m.reference" &&
-                event.getRelation()?.event_id === notificationEvent.event_id &&
+                  event.getRelation()?.event_id ===
+                    notificationEvent.event_id &&
                   event.getSender() !== this.userId,
               ),
             ),
@@ -934,6 +940,8 @@ export class CallViewModel extends ViewModel {
           endWith("decline" as const),
         );
       }),
+    ),
+    EMPTY,
   );
 
   /**
@@ -972,6 +980,8 @@ export class CallViewModel extends ViewModel {
               return of("success" as const);
             }
             // Show the ringing state of the most recent ringing attempt.
+            // ring$ is a behavior so it will emit the latest observable which very well might already have a running timer.
+            // this is important in case livekitConnectionState$ after didSendCallNotification$ has already emitted.
             return this.ring$.pipe(switchAll());
           }),
           // The state starts as 'unknown' because we don't know if the RTC