Single-flight Call.join to stop concurrent-join race - #1764
Single-flight Call.join to stop concurrent-join race#1764PratimMallick wants to merge 8 commits into
Conversation
Coalesce overlapping join() callers onto one in-flight attempt and clean up sessions that fail to connect, preventing SFU-evicted zombie publishers. Co-authored-by: Cursor <cursoragent@cursor.com>
PR checklist ✅All required conditions are satisfied:
🎉 Great job! This PR is ready for review. |
Walkthrough
ChangesCall join coordination
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant ConcurrentCallers
participant CallJoinCoordinator
participant executeJoin
participant API
participant SFUConnection
ConcurrentCallers->>CallJoinCoordinator: call join
CallJoinCoordinator->>executeJoin: execute one join
executeJoin->>API: request join
executeJoin->>SFUConnection: connect once
CallJoinCoordinator-->>ConcurrentCallers: share join result
sequenceDiagram
participant executeJoin
participant SFUConnection
participant CallJoinCoordinator
participant Session
executeJoin->>SFUConnection: report terminal failure
executeJoin->>CallJoinCoordinator: discard failed session
CallJoinCoordinator->>Session: clear and clean up session
CallJoinCoordinator-->>executeJoin: return failure
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt`:
- Line 390: Update executeJoin’s failed-join cleanup so it clears the active
session only when it is still the same localSession; preserve any replacement
installed by discardFailedSession during recovery. Add a recovery-failure test
that installs a replacement session before returning Failure and verifies the
replacement remains active.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 5b98171e-ff06-4133-a27e-e543cf2d2d64
📒 Files selected for processing (2)
stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.ktstream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt
SDK Size Comparison 📏
|
Remove the discardFailedSession ownership guard. Once join is returning Failure (including after failed join-time recovery), clear the active slot and cleanup both the join session and any reconnect replacement. Co-authored-by: Cursor <cursoragent@cursor.com>
|
I think porting the SingleFlight mechanism from core and then using it here would be easier for migration than this in-line implementation. WDYT? |
The one from core runs on its own scope(which is the call scope), whereas for join we want to run in the caller's scope(UI/viewmodel). Hence used a newer way |
But you can create a Or do you mean to use the caller scope, like the UI scope for example to rely on the scope cancellation for join cancellation also? |
aleksandar-apostolov
changed the title
Move join coalescing to StreamRefCountedSingleFlightProcessor so work runs on the call scope, survives individual waiter cancellation, and cancels only when the last waiter leaves. Subsequent join() on an already-joined call returns the existing session instead of failing and tearing down the live call. Co-authored-by: Cursor <cursoragent@cursor.com>
Make flights ConcurrentHashMap-safe, remove+cancel under one lock so newcomers cannot attach to a Cancelling flight, refactor run into acquire/select/await helpers, and add regression tests. Co-authored-by: Cursor <cursoragent@cursor.com>
Last-/sole-waiter cancel aborts the call-scoped join. When that landed after setActiveSession, the half-joined session and Joined state stayed behind and the idempotent join() path then returned Success on that zombie. Tear it down on cancel, and keep the already-joined check in executeJoin only so joinInternal has a single caller-owned precondition. Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Reuse only isActive flights, and cancel/clear/stop now remove then cancel under the same mutex as the closed check so a new run cannot join a dying job or start after stop. Co-authored-by: Cursor <cursoragent@cursor.com>
|
| * (e.g. screen handoff, UI join + call-scoped auto-join) but should not keep running after | ||
| * nobody is waiting — unless [scope] itself is cancelled (leave / call cleanup). | ||
| * | ||
| * Candidate for Stream Android Core v2 alongside [StreamSingleFlightProcessorImpl]. |
There was a problem hiding this comment.
Re-ran this at HEAD after e0094292 — the cancelling-attach and stop() races are both closed, and the two new tests are the deterministic shape. Verified rather than taken on the diff.
On the KDoc though: stream-android-core already ships StreamSingleFlightProcessor with this exact run/has/cancel/clear/stop surface, and video already carries a copy of StreamSingleFlightProcessorImpl. This makes three. The ref-counting is the genuinely new idea and core is where it belongs — worth a follow-up rather than growing a third copy here.
Two things that point the same way: cancel/clear/stop just became suspend to serve the mutex, which makes them awkward from a non-suspending teardown path, and nothing calls them. Core avoids the mutex entirely by evicting on invokeOnCompletion with a conditional remove(key, value) (core#73) — non-suspending, and the Cancelling window cannot exist.
| // Single already-joined gate for the whole join flow: subsequent join() calls while a | ||
| // session is live return that session instead of building a second one. Every retry | ||
| // below clears the session first, so [joinInternal] always starts without one. | ||
| sessionManager.session.value?.let { existing -> |
There was a problem hiding this comment.
Telemetry deltas worth a deliberate call, since join reporting matters:
- This gate returns before
onJoinFunctionStart(), sojoin()on a live call now emits nothing and returns Success. Previously it emitted the join start and reported a Failure. Accidental double-join — the footgun this PR fixes — goes invisible, so we lose the ability to measure how often it happens in the field. - N concurrent joins now emit one
onJoinFunctionStart()instead of N. Right as a count of real joins, but a step change in the metric. - A coalesced caller's
callJoinInterceptoris dropped silently —state.callJoinInterceptoris a single slot assigned inside the flight.
Suggest a distinct signal on the already-joined and coalesced paths instead of going quiet, plus a logger.w when a coalescing caller passes a different interceptor than the in-flight one.
| * [RtcSession] for the same `sessionId` and produce the SFU-evicted zombie this coordinator | ||
| * exists to prevent. | ||
| */ | ||
| suspend fun joinInternal( |
There was a problem hiding this comment.
This drops the guard and replaces it with a comment. joinInternal is still module-visible with default args and is called directly by CallJoinCoordinatorTest and JoinRecoverableFailureTest, so the only thing stopping a second RtcSession on the same sessionId is now KDoc. executeJoin clears the session before every retry, so putting the early return back is free and keeps the invariant enforced in code.
|
Goal section: |
| mutex.withLock { | ||
| flight.waiters = (flight.waiters - 1).coerceAtLeast(0) | ||
| if (cancelIfLast && flight.waiters == 0 && flight.deferred.isActive) { | ||
| cancelAndDetachLocked(flight) |
There was a problem hiding this comment.
[P2] Detach cancelled flights even when the deferred is no longer active
Issue
A deferred created from an already-cancelled scope may be cancelled before its coroutine body starts. In that case, the body’s finally block may never run.
The current waiter cleanup removes the flight only when the deferred is still active:
if (cancelIfLast && flight.waiters == 0 && flight.deferred.isActive) {
cancelAndDetachLocked(flight)
}Because an already-cancelled deferred has isActive == false, its entry can remain in flights after the final waiter leaves.
Risk
This is particularly risky when the processor will support replacing or reusing its scope:
has(key)can incorrectly report that an inactive flight exists.- Flights created by the old scope remain associated with that scope.
- Supplying a new scope cannot restart or migrate old deferreds.
- Calls using unique keys can accumulate stale entries.
- Old and new scope work could overlap unless replacement clears existing flights first.
Solution
Always remove a flight when its final waiter leaves. Cancellation should remain conditional because completed or already-cancelled deferreds do not need to be cancelled again.
flight.waiters--
if (flight.waiters == 0) {
if (cancelIfLast && flight.deferred.isActive) {
cancelAndDetachLocked(flight)
} else {
removeFlightIfCurrentLocked(
key = flight.key,
deferred = flight.deferred,
)
}
}When scope replacement is added, we will clear and cancel all existing flights before exposing the new scope. The replacement must use the same mutex so it cannot race with run():
mutex.withLock {
val oldFlights = flights.values.toList()
flights.clear()
oldFlights.forEach { it.deferred.cancel() }
scope = newScope
}A Deferred permanently belongs to the scope that created it, so old flights must never be carried over to the new scope.
rahul-lohra
left a comment
rahul-lohra
left a comment
There was a problem hiding this comment.
Nice work, left [P2] comment about clearing stale flights before introducing a reusable coroutine scope. It does not block this PR, so I’m approving it.
3 participants
Goal
closes AND-1379
Prevent overlapping
Call.join()calls from creating multipleRtcSessions that share the samesessionId. The SFU keeps only the latest participant and evicts the others, which leaves zombie publishers that cannot publish A/V and often fail subsequent RPCs withPARTICIPANT_NOT_FOUND, triggering reconnect/rejoin loops.Also fix a related footgun: calling
join()again while already joined used to returnFailureand clear the live session / setRealtimeConnection.Failed, which tore down a healthy call (easy to hit with accidental double-join).Implementation
StreamRefCountedSingleFlightProcessor: keyed single-flight that runs shared work on the call scope, tracks waiters, and cancels the shared job only when the last waiter is cancelled (one UI cancel does not kill other waiters / auto-join).CallJoinCoordinator.join()through that processor so concurrent callers share one join attempt and once-only setup (telemetry, interceptor, leave guard,InProgress).join()while a session already exists returnsSuccess(existing)(idempotent) instead of failing and tearing down the call.discardFailedSession()cleanup on SFU connect failure during join so failed sessions do not keep issuing RPCs after eviction.Behavior notes for reviewers
join()waiters no longer cancels the shared join; cancelling the last waiter does.join()while already joined:Failure("already been joined")→Success(existing session)(intentional API softening; avoids destroying a live call).Testing
./gradlew :stream-video-android-core:spotlessApply./gradlew :stream-video-android-core:testDebugUnitTest --tests 'io.getstream.video.android.core.call.components.CallJoinCoordinatorTest'— passed./gradlew :stream-video-android-core:testDebugUnitTest --tests 'io.getstream.video.android.core.utils.StreamRefCountedSingleFlightProcessorTest'— passedFailure modes this mitigates
☑️Contributor Checklist
General
developbranchCode & documentation
stream-video-examples)☑️Reviewer Checklist
🎉 GIF
N/A — core join orchestration fix, no UI changes.