Single-flight Call.join to stop concurrent-join race - #1764
Single-flight Call.join to stop concurrent-join race#1764PratimMallick wants to merge 19 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>
|
Goal section: |
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.
Keep already-joined Success at joinInternal for direct callers, detach stale flights when the last waiter leaves even if the deferred is dead, and record SFU traces plus warnings for double-join and coalesced concurrent joins. Co-authored-by: Cursor <cursoragent@cursor.com>
Unsafe casts after a nullable publish crashed join/ringing E2E when the publisher was missing or had no matching publish options. Co-authored-by: Cursor <cursoragent@cursor.com>
The publishStream null guard moved setMuteState after the publish attempt, so a null publish skipped UpdateMuteStates entirely. Without it the SFU never emits TrackPublished, ParticipantState.audioEnabled stays false and the participant tile shows a muted mic while the local toggle shows enabled. Signal the mute state first again, as before, and keep only the safe cast. The joinInternal already-joined guard sat after cancelSfuObservers(), so returning the live session cancelled its SFU event subscription with nothing left to re-register it (monitorSession only runs on the new-session path) and never moved the connection to Joined. Gate before the teardown instead. Co-authored-by: Cursor <cursoragent@cursor.com>
Incoming accept can finish or recreate the Activity after the SFU session is already in. Last-waiter cancel then discarded that session, ringing stayed Idle, and Connecting never left. Leave still aborts join by cancelling the call scope. Co-authored-by: Cursor <cursoragent@cursor.com>
RtcSession is installed before JoinCallResponseEvent, so startNoiseCancellation hit PARTICIPANT_NOT_FOUND and triggered a rejoin that left ringing stuck on Connecting. Co-authored-by: Cursor <cursoragent@cursor.com>
Keep coalesced and already-joined join() calls visible in telemetry without rotating the in-flight joinStageAttemptId used to correlate coordinator and SFU events. Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Report the SDK join at Call.join() entry so coalesced and already-joined callers stay visible, minting a new joinStageAttemptId each time. Co-authored-by: Cursor <cursoragent@cursor.com>
setMuteState(true) was sending UpdateMuteStates before asPublishedOrNull could return, so a failed publish still looked live on the SFU. Co-authored-by: Cursor <cursoragent@cursor.com>
Cancelling a join waiter does not abort the shared job — only leave() does. Capture the leader interceptor under the flight lock so coalesced callers do not warn about a drop against a not-yet-assigned state field, and install it before awaiting the guest token. Co-authored-by: Cursor <cursoragent@cursor.com>
|
| * [executeJoin] assigns [CallState.callJoinInterceptor]. | ||
| */ | ||
| @Volatile | ||
| private var inFlightJoinInterceptor: CallJoinInterceptor? = null |
There was a problem hiding this comment.
[P2] - I think this property shouldn't be here. Instead, it should be scoped to Flight.
|
|
||
| // Before any suspend so coalesced waiters and SFU observers see this join's | ||
| // interceptor rather than a stale/null [CallState.callJoinInterceptor]. | ||
| state.callJoinInterceptor = callJoinInterceptor |
There was a problem hiding this comment.
I think we will receive stale callJoinInterceptor here
Consider the case when the Caller A's Activity is destroyed and we move to activity B with call.join(interceptor) again
Then in that case a dead activity's interceptor will be invoked.
Please think around this. What behaviour should we follow?
There was a problem hiding this comment.
yes this was intentional, are taking the first interceptor that was sent and knowingly discarding the interceptor of next join calls, This is the same pattern used for other join parameters
There was a problem hiding this comment.
I have listed 4 concerns below. Please check
1 CallJoinInterceptor is lifecycle-sensitive
CallJoinInterceptor is different from the other join parameters. The other parameters are request configuration, while the interceptor is integrator-provided executable code. Its implementation may reference an Activity, UI components, navigation, or lifecycle-bound state.
Consider this scenario:
- Activity A calls
join(interceptorA).- Activity A is destroyed and recreated as Activity B.
- Activity B calls
join(interceptorB).
With the current implementation, the second call coalesces into the existing flight. interceptorA remains selected and interceptorB is discarded. The SDK may later invoke code associated with the destroyed Activity A. This could update stale UI, navigate using the old Activity, throw an exception, or skip the intended user interaction.
2 This changes integrator-visible behaviour
Before join coalescing, concurrent join attempts could each assign CallState.callJoinInterceptor, so a later execution could replace the previous interceptor. With single-flight, only the first join block executes; subsequent interceptors are silently discarded.
This removes the previous race, but it also establishes first-interceptor-wins as new deterministic behaviour.
Since this PR explicitly resolves concurrent join() behaviour, interceptor ownership still seems like a loose end.
3 Interceptor ownership trade-offs
Since CallJoinInterceptor is integrator-provided executable code. The main risk is invoking an interceptor supplied by a caller whose coroutine has already been cancelled. The interceptor itself runs in the SDK scope, so caller cancellation does not automatically stop it.
I have listed down Interceptor ownership trade-offs
| Policy | Advantages | Disadvantages | Risk |
|---|---|---|---|
| First interceptor wins — current | Simple; consistent with other first-flight arguments | Caller A’s interceptor remains selected even after A’s Job is cancelled. Caller B’s interceptor is ignored. | High: SDK can invoke code belonging to a cancelled caller. That code may use cancelled work, stale state, or throw CancellationException, potentially preventing the active-state transition. |
| Latest non-null wins | Simple replacement rule; join(null) does not erase an interceptor | The latest interceptor may also be cancelled before invocation, and the previous interceptor is no longer available as fallback. | Medium: Reduces stale-first-caller risk but does not verify whether the latest caller is still valid. |
| First non-cancelled wins | Preserves first-wins when the original caller remains valid; falls back to the next candidate after cancellation | Must store interceptor candidates with their caller Jobs and select them atomically. | Low: A cancellation can still race with selection. Once invocation starts, the selected interceptor must be frozen and allowed to finish. |
| Latest non-cancelled wins | Uses the most recent eligible interceptor and retains older candidates as fallback | Changes the existing first-wins policy and requires synchronized ordering and selection. | Low–Medium: Avoids known cancelled callers, but concurrent arrival order determines which valid interceptor wins. |
My Recommendation: first non-cancelled wins. It preserves the current policy unless the original caller is cancelled, while avoiding invocation of an interceptor already known to be associated with a cancelled caller.
4. Concurrent interceptor behaviour is not documented
Call.join() KDoc states that concurrent callers share one in-flight attempt, but it does not explain what happens when those callers provide different arguments or different CallJoinInterceptor instances.
CallJoinInterceptor KDoc also does not state that the first interceptor is retained and later interceptors are ignored. Therefore, an integrator cannot predict this behaviour from the public API contract.
Once the ownership policy is agreed, it should be documented explicitly. For example, if first-wins remains:
Concurrent join() calls share the same in-flight operation. The arguments and CallJoinInterceptor from the caller that creates the operation are used. Arguments supplied by coalesced callers are ignored.
If latest-non-null wins, that replacement behaviour should be documented instead.
3 participants
Goal
Fixes AND-1376
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). Last waiter always detaches the map entry even if the deferred is already dead.CallJoinCoordinator.join()through that processor so concurrent callers share one join attempt and once-only setup (telemetry, interceptor, leave guard,InProgress).join()/joinInternal()while a session already exists returnsSuccess(existing)(idempotent) instead of failing and tearing down the call.logger.wfor already-joined (join-already-joined) and coalesced concurrent joins (join-coalesced), including a warning when a coalesced caller’s interceptor differs from the in-flight one.discardFailedSession()cleanup on SFU connect failure during join so failed sessions do not keep issuing RPCs after eviction.onCoalesced, cancel-one / cancel-last, cancelled-scope detach) and join coordinator (concurrent join, already-joined Success, traces, cleanup).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).stream-android-coreis follow-up, not this PR.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.