[stdlib]: Propagate AsyncStream termination to all consumers

As of the changes in #41713 to
enable Sendability for AsyncStream, it has been possible to create
multiple stream consumers operating concurrently. This change fixes
behavior in the case that the underlying stream is terminated while
multiple pending continuations are outstanding. Previously such
consumers would have been leaked (never resumed). Now, they are notified
of the stream's termination and resumed appropriately.

Resolves #66541 & #71412

Author: jamieQ

stdlib/public/Concurrency/AsyncStreamBuffer.swift

@@ -212,25 +212,21 @@ extension AsyncStream {
       state.onTermination = nil
       state.terminal = true
 
-      if let continuation = state.continuations.first {
-        if state.pending.count > 0 {
-          state.continuations.removeFirst()
-          let toSend = state.pending.removeFirst()
-          unlock()
-          handler?(.finished)
-          continuation.resume(returning: toSend)
-        } else if state.terminal {
-          state.continuations.removeFirst()
-          unlock()
-          handler?(.finished)
-          continuation.resume(returning: nil)
-        } else {
-          unlock()
-          handler?(.finished)
-        }
-      } else {
+      guard !state.continuations.isEmpty else {
         unlock()
         handler?(.finished)
+        return
+      }
+
+      // Hold on to the continuations to resume outside the lock.
+      let continuations = state.continuations
+      state.continuations = []
+
+      unlock()
+      handler?(.finished)
+
+      for continuation in continuations {
+        continuation.resume(returning: nil)
       }
     }
 

test/Concurrency/Runtime/async_stream.swift

@@ -435,6 +435,37 @@ class NotSendable {}
         expectTrue(expectation.fulfilled)
       }
 
+      // MARK: - Multiple consumers
+
+      tests.test("finish behavior with multiple consumers") {
+        let (stream, continuation) = AsyncStream<Int>.makeStream()
+        let (controlStream, controlContinuation) = AsyncStream<Int>.makeStream()
+        var controlIterator = controlStream.makeAsyncIterator()
+
+        func makeConsumingTaskWithIndex(_ index: Int) -> Task<Void, Never> {
+          Task {
+            controlContinuation.yield(index)
+            for await i in stream {
+              controlContinuation.yield(i)
+            }
+          }
+        }
+
+        // Wait for the consumers to be set up
+        let consumer1 = makeConsumingTaskWithIndex(1)
+        expectEqual(await controlIterator.next(isolation: #isolation), 1)
+
+        let consumer2 = makeConsumingTaskWithIndex(2)
+        expectEqual(await controlIterator.next(isolation: #isolation), 2)
+
+        // Terminate the stream
+        continuation.finish()
+
+        // Ensure the consuming Tasks both complete
+        _ = await consumer1.value
+        _ = await consumer2.value
+      }
+
       await runAllTestsAsync()
     }
   }