Subscribe to the source before startCollection returns (#247)

[monkopedia-coder]

Fixes #247.

asKsFlow() started its server-side collection with a plain scope.launch, which only schedules the body, and returned the KsCollectionToken right away. So startCollection could return before the coroutine had subscribed to the source.

A cold source is unaffected — nothing emits until the collection starts. A hot source (a MutableSharedFlow with replay = 0, or any flow the user shares) drops everything emitted in that window, and the client sees a stalled collection rather than an error. That matters because startCollection returning a token is the only "I am subscribed now" signal ksrpc gives a caller, and it did not mean that.

Fix

scope.launch(start = CoroutineStart.UNDISPATCHED). The body runs on the calling coroutine up to its first real suspension, which for SharedFlow.collect is after the subscriber slot is registered — so the token cannot be handed back before the subscription exists.

onSubscription is not usable here: it is defined on SharedFlow, and asKsFlow accepts any Flow.

The trade-off is that a cold source's synchronous prologue (the part of a flow { } builder before its first suspension) now runs on the caller's dispatcher instead of the launched one. In practice the first collector.onItem is a suspending sub-service call, so a source that emits at all suspends almost immediately.

Private implementation detail — no ABI change, apiCheck is byte-identical.

Measurement

2000 rounds in-process on Dispatchers.Default, startCollection(collector) then emit immediately, 200 ms per round:

variant	drops
before (launch { })	1877 / 2000
control: await subscriptionCount > 0 before emitting	0 / 2000
after (launch(start = UNDISPATCHED))	0 / 2000

The control isolates the cause to the dispatch gap. In-process the whole window is that gap, which is why the rate is so high; over a real connection the trigger costs a wire round trip, so the rate falls but the window does not close, and it widens under dispatcher load.

Test

KsFlowServiceTest.testStartCollectionSubscribesBeforeReturning asserts subscriptionCount.value == 1 the instant startCollection returns, then emits and requires the item to arrive. It asserts the invariant itself rather than sampling for the symptom, so there are no sleeps or timing thresholds.

It repeats 100 rounds, because a scheduled collection can still win the race by luck — it subscribed in time in about 6% of the measured rounds (the 123 of 2000 that did not drop), so a single round would let the regression through about that often. 100 rounds run in ~1.2s. Verified to fail on the unfixed code (round N: startCollection returned before subscribing to the source expected:<1> but was:<0>) and pass with the change.

Verification

ktlintCheck (CI's -x ktlintKotlinScriptCheck form), licenseCheckForKotlin, apiCheck, full jvmTest, and :ksrpc-test:linuxX64Test all green locally on JDK 21.

[monkopedia-coder]

One open question for whoever reviews this: does the JS/wasm side matter for a dispatch-timing change? Answering it here so it doesn't have to be rediscovered.

It matters, and it argues for the fix rather than against it. CoroutineStart.UNDISPATCHED is a coroutines-core semantic with the same meaning on every target — it is not a JVM threading trick. On JS and wasmJs the situation before this change is strictly worse than on the JVM: those targets are single-threaded, so a CoroutineStart.DEFAULT launch cannot run before the caller resumes. There is no other thread to win the race, so the collection is guaranteed not to have subscribed when startCollection returns, and a hot source would drop every emission rather than the 94% measured on Dispatchers.Default. The 6% of JVM rounds that got away with it are exactly the ones where another thread happened to run the body in time — a thing that cannot happen there.

The regression test lives in commonTest, so it covers those targets whenever their test tasks are run. CI is ubuntu-only and does not run the JS/wasm suites (see #246), so this is not verified by the checks on this PR; :ksrpc-test:linuxX64Test green does cover a native target.