SharedFlow: Fix scenario with concurrent emitters and cancellation of subscriber

Fixes #2356

Author: elizarov

kotlinx-coroutines-core/common/src/flow/SharedFlow.kt

@@ -497,7 +497,13 @@ private class SharedFlowImpl<T>(
             }
         }
         // Compute new buffer size -> how many values we now actually have after resume
-        val newBufferSize1 = (newBufferEndIndex - head).toInt()
+        var newBufferSize1 = (newBufferEndIndex - head).toInt()
+        // Note: When nCollectors == 0 we resume all queued emitted and might have resumed more that max size of
+        //       the buffer, so here is why we take coerce the resulting size to the buffer capacity
+        if (nCollectors == 0 && newBufferSize1 > bufferCapacity) {
+            newMinCollectorIndex += newBufferSize1 - bufferCapacity // adjust minCollectorIndex, too, to skip items
+            newBufferSize1 = bufferCapacity
+        }
         // Compute new replay size -> limit to replay the number of items we need, take into account that it can only grow
         var newReplayIndex = maxOf(replayIndex, newBufferEndIndex - minOf(replay, newBufferSize1))
         // adjustment for synchronous case with cancelled emitter (NO_VALUE)

kotlinx-coroutines-core/common/test/flow/sharing/SharedFlowScenarioTest.kt

@@ -201,6 +201,26 @@ class SharedFlowScenarioTest : TestBase() {
             emitResumes(e3); expectReplayOf(3)
         }
 
+    @Test
+    fun testSuspendedConcurrentEmitAndCancelSubscriber() =
+        testSharedFlow<Int>(MutableSharedFlow(1)) {
+            val a = subscribe("a");
+            emitRightNow(0); expectReplayOf(0)
+            collect(a, 0)
+            emitRightNow(1); expectReplayOf(1)
+            val e2 = emitSuspends(2) // suspends until 1 is collected
+            val e3 = emitSuspends(3) // suspends until 1 is collected, too
+            cancel(a) // must resume emitters 2 & 3
+            emitResumes(e2)
+            emitResumes(e3)
+            expectReplayOf(3) // but replay size is 1 so only 3 should be kept
+            // Note: originally, SharedFlow was in a broken state here with 3 elements in the buffer
+            val b = subscribe("b")
+            collect(b, 3)
+            emitRightNow(4); expectReplayOf(4)
+            collect(b, 4)
+        }
+
     private fun <T> testSharedFlow(
         sharedFlow: MutableSharedFlow<T>,
         scenario: suspend ScenarioDsl<T>.() -> Unit