DOM: Fix 'ref-counted producer' Subscriber crash

This CL fixes a crash in Subscriber, which was introduced in
https://crrev.com/c/6221901, when we implemented ref-counted producers.
With ref-counted producers, Subscriber::next()/error()/complete() have
to iterate over the list of internal observers and call the respective
methods on them. However, because these methods can terminate the
subscription for a given observer, the list of internal observers can
mutate while iterating over it, which is unsafe and causes a crash.

This is essentially the implementation version of
whatwg/infra#396.

This CL fixes this bug by taking a copy of the list of internal
observers before iterating over it in each of these methods, so we can
call the methods on each registered observer, while iterating over a
stable vector that cannot be mutated (since it is a copy).

R=masonf

Bug: 40282760
Change-Id: I9ade96a95370120b4c9f7309a78d3222398aed6b
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/6311209
Commit-Queue: Dominic Farolino <[email protected]>
Reviewed-by: Mason Freed <[email protected]>
Cr-Commit-Position: refs/heads/main@{#1426289}

Author: domfarolino

dom/observable/tentative/observable-take.any.js

@@ -106,3 +106,28 @@ test(() => {
 
   assert_array_equals(results, ["source subscribe", 1, 2, 3, "complete"]);
 }, "take(): Negative count is treated as maximum value");
+
+// This tests a regression in Chromium's implementation. In ref-counted
+// producers, when Subscriber#next() is called, the Subscriber iterates over all
+// of its "internal observers" [1] and calls "next" on them. However, "next" can
+// complete the subscription, and modify the "internal observers" list while
+// Subscriber is iterating over it. This mutation-during-iteration caused a
+// crash regression in Chromium, which this test covers.
+//
+// [1]: https://wicg.github.io/observable/#subscriber-internal-observers
+promise_test(async () => {
+  async function* asyncNumbers() {
+    yield* [1,2,3,4];
+  }
+
+  const source = Observable.from(asyncNumbers());
+  const results = [];
+
+  source.take(1).toArray().then(result => results.push(result));
+  await source.take(3).toArray().then(result => results.push(result));
+
+  assert_equals(results.length, 2);
+  assert_array_equals(results[0], [1]);
+  assert_array_equals(results[1], [1, 2, 3]);
+}, "take(): No crash when take(1) unsubscribes from its source when next() " +
+   "is called, and the Subscriber iterates over the rest of the Observables");