Fix leak linked to event listeners on quality change

The #1778 and #1779 issues / PR noticed a leak that seem to arise when
multiple quality switches happen.

I'm still unsure of the severity (looking at it what this fixes seems
very minimal, and we did not notice this yet on production at Canal+
including on low-memory devices for what seems to be a change that has
been here for 2 years - but external contributors actually did notice a
leak so maybe a set of conditions amplify the issue), but looking closely
at the code in question, there does seem to be an improper event listener
clean-up on a quality switch.

The issue is rooted in the complexity behind how quality switch happen:

- depending on heuristics, we may either perform an "urgent" quality
  switch (where we directly cancel the requests linked to the older
  quality) or a non-urgent one (where we will wait for the current
  requests to finish and only after load the new quality).

- If non-urgent, we want to still do the requests for the new quality as
  soon as we can, thus we parallelize it with the pushing operations of
  the segments we just loaded from the previous quality.

Thus when a "non-urgent" quality switch happen, there might be a short
time where several quality-linked modules are running at the same time
(the old one to push segments, the new one to load them), whereas at
first glance they seemed conflicting (one loads and push one quality,
the other loads and push another quality of the same thing).

This lead to an awkward architecture where the clean-up process of those
modules is subtly different than in other RxPlayer modules - this one
has actually 2 means to terminate:

- its `terminate` parameter, kind of like a SIGTERM: just finish what
  you're doing (e.g. finish loading segments and/or pushing them then
  stop).

  Once the `RepresentationStream` (the module in question) has finished
  loading segments, it sends a `terminating` event - but it might still
  be pushing segments.

  It however has no event to indicate that segments have been pushed,
  for now.

- its `cancelSignal` parameter, more akin to a SIGKILL: terminate
  everything now without delay.

  This one is e.g. triggered when stopping the content, changing the
  track etc.

The leaking event listener was wrongly linked to that "SIGKILL" signal,
even if it was intended to be cleaned up when the module is not needed
anymore. When the module was only "SIGTERMed", it was not cleaned up.

---

I chose to clean it up not right when "SIGTERMed", but when the module
itself anounced that it is "terminating" (it is done loading and is now
pushing segments).

I found it to be more appropriate for the logic in question and a
corresponding `CancellationSignal` was already used for other similar
logic linked to the same lifetime.

Author: peaBerberian

src/core/stream/adaptation/adaptation_stream.ts

@@ -215,12 +215,12 @@ export default function AdaptationStream(
    * error or on some cancellation.
    * @param {Object} choice - The last Representations choice that has been
    * made.
-   * @param {Object} fnCancelSignal - `CancellationSignal` allowing to cancel
-   * everything this function is doing and free all related resources.
+   * @param {Object} repsChoiceCancelSignal - `CancellationSignal` allowing to
+   * cancel everything this function is doing and free all related resources.
    */
   async function onRepresentationsChoiceChange(
     choice: IRepresentationsChoice,
-    fnCancelSignal: CancellationSignal,
+    repsChoiceCancelSignal: CancellationSignal,
   ): Promise<void> {
     // First check if we should perform any action regarding what was previously
     // in the buffer
@@ -243,7 +243,7 @@ export default function AdaptationStream(
         return queueMicrotask(() => {
           playbackObserver.listen(
             () => {
-              if (fnCancelSignal.isCancelled()) {
+              if (repsChoiceCancelSignal.isCancelled()) {
                 return;
               }
               const { DELTA_POSITION_AFTER_RELOAD } = config.getCurrent();
@@ -255,21 +255,21 @@ export default function AdaptationStream(
                 stayInPeriod: true,
               });
             },
-            { includeLastObservation: true, clearSignal: fnCancelSignal },
+            { includeLastObservation: true, clearSignal: repsChoiceCancelSignal },
           );
         });
 
       case "flush-buffer": // Clean + flush
       case "clean-buffer": // Just clean
         for (const range of switchStrat.value) {
           await segmentSink.removeBuffer(range.start, range.end);
-          if (fnCancelSignal.isCancelled()) {
+          if (repsChoiceCancelSignal.isCancelled()) {
             return;
           }
         }
         if (switchStrat.type === "flush-buffer") {
           callbacks.needsBufferFlush();
-          if (fnCancelSignal.isCancelled()) {
+          if (repsChoiceCancelSignal.isCancelled()) {
             return;
           }
         }
@@ -278,7 +278,7 @@ export default function AdaptationStream(
         assertUnreachable(switchStrat);
     }
 
-    recursivelyCreateRepresentationStreams(fnCancelSignal);
+    recursivelyCreateRepresentationStreams(repsChoiceCancelSignal);
   }
 
   /**
@@ -410,30 +410,41 @@ export default function AdaptationStream(
    * indicating that the `RepresentationStream` should stop what it's doing.
    * @param {Object} representationStreamCallbacks - Callbacks to call on
    * various `RepresentationStream` events.
-   * @param {Object} fnCancelSignal - `CancellationSignal` which will abort
-   * anything this function is doing and free allocated resources.
+   * @param {Object} globalCancelSignal - `CancellationSignal` which will
+   * immediately clean every resources allocated by this function.
    */
   function createRepresentationStream(
     representation: IRepresentation,
     terminateCurrentStream: IReadOnlySharedReference<ITerminationOrder | null>,
     representationStreamCallbacks: IRepresentationStreamCallbacks,
-    fnCancelSignal: CancellationSignal,
+    globalCancelSignal: CancellationSignal,
   ): void {
     /** Set to `true` if we've encountered an error with this `RepresentationStream` */
     let hasEncounteredError = false;
 
-    const bufferGoalCanceller = new TaskCanceller(
-      "AdaptationStream: BufferGoal " + adaptation.type,
+    /**
+     * Construct a `TaskCanceller`, triggered once the `RepresentationStream` we
+     * will create here announces that it is "terminating" (implies that it is
+     * done loading new data and will clean itself automatically once it has
+     * pushed all loaded segments).
+     *
+     * We keep it distinct from `globalCancelSignal` as the latter's lifetime may
+     * be much much longer than our `RepresentationStream`'s.
+     * Thus it wouldn't be adapted as a canceller for the listeners we're
+     * registering here.
+     */
+    const terminatingCanceller = new TaskCanceller(
+      "RepresentationStream-linked listeners in AdaptationStream",
     );
-    bufferGoalCanceller.linkToSignal(fnCancelSignal);
+    terminatingCanceller.linkToSignal(globalCancelSignal);
 
     /** Actually built buffer size, in seconds. */
     const bufferGoal = createMappedReference(
       wantedBufferAhead,
       (prev) => {
         return getBufferGoal(representation, prev);
       },
-      bufferGoalCanceller.signal,
+      terminatingCanceller.signal,
     );
 
     const maxBufferSize =
@@ -480,20 +491,21 @@ export default function AdaptationStream(
 
           // We wait 4 seconds to let the situation evolve by itself before
           // retrying loading segments with a lower buffer goal
-          cancellableSleep(4000, fnCancelSignal)
+          // If the `RepresentationStream` was terminating anyway, just exits
+          cancellableSleep(4000, terminatingCanceller.signal)
             .then(() => {
               return createRepresentationStream(
                 representation,
                 terminateCurrentStream,
                 representationStreamCallbacks,
-                fnCancelSignal,
+                globalCancelSignal,
               );
             })
             .catch(noop);
         }
       },
       terminating() {
-        bufferGoalCanceller.cancel("Representation terminating");
+        terminatingCanceller.cancel("Representation terminating");
         representationStreamCallbacks.terminating();
       },
     });
@@ -512,7 +524,16 @@ export default function AdaptationStream(
         },
       },
       updatedCallbacks,
-      fnCancelSignal,
+      // NOTE: We give the long-lived `globalCancelSignal` here (and not
+      // `terminatingCanceller.signal`) on purpose.
+      // `RepresentationStream` should clean-up themselves automatically based
+      // on their `terminate` parameter.
+      //
+      // This `CancellationSignal` is a killswitch which if triggered too
+      // soon might interrupt some async operations done when this
+      // `RepresentationStream` is terminating: e.g. stop pushing the segments
+      // it has just loaded.
+      globalCancelSignal,
     );
 
     // reload if the Representation disappears from the Manifest
@@ -525,7 +546,7 @@ export default function AdaptationStream(
               if (updated.adaptation === adaptation.id) {
                 for (const rep of updated.removedRepresentations) {
                   if (rep === representation.id) {
-                    if (fnCancelSignal.isCancelled()) {
+                    if (terminatingCanceller.isUsed()) {
                       return;
                     }
                     return callbacks.waitingMediaSourceReload({
@@ -543,7 +564,7 @@ export default function AdaptationStream(
           }
         }
       },
-      fnCancelSignal,
+      terminatingCanceller.signal,
     );
   }