Replace sfu connect exception with sfu connect failure domain model#1745
Closed
rahul-lohra wants to merge 3 commits into
Closed
Conversation
makes the producer/consumer boundary clearer: `connectInternal()` reports a typed SFU connect failure, and `_join()` decides how to orchestrate recovery from that failure.
Contributor
PR checklist ❌The following issues were detected:
What we check
|
rahul-lohra
changed the title
Contributor
SDK Size Comparison 📏
|
Contributor
WalkthroughSFU connection failures now use typed causes for socket observation timeouts, recoverable failures, and terminal failures. Join and reconnect flows branch on these causes, while analytics and tests consume the updated failure structure. ChangesSFU failure handling
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant RtcSession
participant Call
participant ReconnectLoop
RtcSession->>Call: Return typed SFU failure cause
Call->>ReconnectLoop: Start REJOIN or await recovery
Call->>Call: Fail immediately for terminal failure
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 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}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt (1)
976-991: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider preserving the original throwable as
causewhen wrapping.
Exception(msg)here drops the underlyingnetworkError.cause(e.g.SocketTimeoutException/InterruptedIOException), so any consumer inspectingSfuConnectFailure.error.cause(crash reporting, logs) loses the original exception type/stack trace once wrapped asRecoverableSocketFailure/TerminalSocketFailure.♻️ Proposed fix to preserve the original cause
- val error = Exception(msg) + val error = Exception(msg, networkError?.cause) val failure = if ((sfuSocketState as? SfuSocketState.Disconnected)?.isRecoverable() == true) {🤖 Prompt for 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. In `@stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt` around lines 976 - 991, Preserve the underlying throwable when constructing the connection failure in the `RtcSession` connect failure handling block. Replace the `Exception(msg)` construction with an exception that uses `networkError` as its cause when available, while retaining the existing message and fallback behavior, so `RecoverableSocketFailure` and `TerminalSocketFailure` expose the original error chain.
🤖 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.kt`:
- Around line 783-828: Update the SocketStateObservationTimeout branch in
_join() to await the REJOIN result instead of launching it asynchronously and
falling through. Mirror the RecoverableSocketFailure handling: invoke reconnect
with REJOIN, await didReconnectSucceed(), and return an appropriate Failure
after recovery settles unsuccessfully, ensuring _join() does not proceed to
monitorSession or return Success prematurely.
---
Nitpick comments:
In
`@stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt`:
- Around line 976-991: Preserve the underlying throwable when constructing the
connection failure in the `RtcSession` connect failure handling block. Replace
the `Exception(msg)` construction with an exception that uses `networkError` as
its cause when available, while retaining the existing message and fallback
behavior, so `RecoverableSocketFailure` and `TerminalSocketFailure` expose the
original error chain.
🪄 Autofix (Beta)
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
Run ID: 0f5b7f55-5160-448f-a3a3-040a7f74ddc4
📒 Files selected for processing (6)
stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.ktstream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.ktstream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/SfuConnectFailure.ktstream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/FastReconnectIceRestartTest.ktstream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/JoinRecoverableFailureTest.ktstream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt
Comment on lines
+783
to
828
| when (sfuConnectionResult.cause) { | ||
| is SfuConnectFailure.SocketStateObservationTimeout -> { | ||
| // REJOIN (not FAST) on purpose: a connect timeout means the initial | ||
| // join never completed. There is no established SFU session or | ||
| // negotiated media path to resume, so a FAST resume would likely hit | ||
| // PARTICIPANT_NOT_FOUND and burn attempts before the loop escalates. | ||
| // A full REJOIN re-fetches credentials and starts a clean join, which | ||
| // is the only thing that can actually succeed here. | ||
| logger.w { | ||
| "[_join] Recoverable SFU connect timeout with no recovery started — triggering REJOIN" | ||
| "[_join] SFU socket state observation timed out with no recovery started — triggering REJOIN" | ||
| } | ||
| scope.launch { | ||
| reconnect( | ||
| WebsocketReconnectStrategy.WEBSOCKET_RECONNECT_STRATEGY_REJOIN, | ||
| "join-recoverable-connect-failure", | ||
| ) | ||
| } | ||
| } else { | ||
| logger.w { "[_join] Recoverable SFU connection failure — awaiting recovery outcome" } | ||
| } | ||
| if (!didReconnectSucceed()) { | ||
| logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" } | ||
|
|
||
| is SfuConnectFailure.RecoverableSocketFailure -> { | ||
| logger.w { "[_join] Recoverable SFU socket failure — awaiting recovery outcome" } | ||
| if (!didReconnectSucceed()) { | ||
| logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" } | ||
| sendJoinErrorAnalytics(sfuConnectionResult) | ||
| return Failure( | ||
| Error.GenericError( | ||
| sfuConnectionResult.cause.error.message ?: "SFU connection failed", | ||
| ), | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| is SfuConnectFailure.TerminalSocketFailure -> { | ||
| logger.e { | ||
| "[_join] Got terminal error while connecting to SFU. Error : $sfuConnectionResult" | ||
| } | ||
| sendJoinErrorAnalytics(sfuConnectionResult) | ||
| return Failure( | ||
| Error.GenericError( | ||
| sfuConnectionResult.error.message ?: "SFU connection failed", | ||
| sfuConnectionResult.cause.error.message ?: "RtcSession error occurred.", | ||
| ), | ||
| ) | ||
| } | ||
| } else { | ||
| logger.e { | ||
| "[_join] Got non recoverable error while connecting to SFU. Error : $sfuConnectionResult" | ||
| } | ||
| sendJoinErrorAnalytics(sfuConnectionResult) | ||
| return Failure( | ||
| Error.GenericError( | ||
| sfuConnectionResult.error.message ?: "RtcSession error occurred.", | ||
| ), | ||
| ) | ||
| } | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
SocketStateObservationTimeout branch never awaits the REJOIN it triggers — _join() falls through to Success.
Unlike the RecoverableSocketFailure branch, this branch fires scope.launch { reconnect(REJOIN, ...) } and then falls out of the when with no return, straight into val connectedSession = session.value ?: return Failure(...) (Line 829). Since session.value is already non-null (set at Line 773 before connectInternal() ran), execution proceeds to monitorSession(...) and return Success(...) immediately — without ever observing whether the REJOIN succeeded.
This contradicts JoinRecoverableFailureTest.socket state observation timeout starts a REJOIN itself, which expects _join() to stay suspended after triggering REJOIN and resolve to Failure only once the connection settles as ReconnectingFailed. As written, _join() returns Success synchronously, so that test would fail, and in production the join flow would falsely report success while the SFU connection actually timed out.
🐛 Proposed fix to await the triggered REJOIN, mirroring the RecoverableSocketFailure branch
is SfuConnectFailure.SocketStateObservationTimeout -> {
logger.w {
"[_join] SFU socket state observation timed out with no recovery started — triggering REJOIN"
}
scope.launch {
reconnect(
WebsocketReconnectStrategy.WEBSOCKET_RECONNECT_STRATEGY_REJOIN,
"join-recoverable-connect-failure",
)
}
+ if (!didReconnectSucceed()) {
+ logger.e { "[_join] Could not recover from socket state observation timeout. Error : $sfuConnectionResult" }
+ sendJoinErrorAnalytics(sfuConnectionResult)
+ return Failure(
+ Error.GenericError(
+ sfuConnectionResult.cause.error.message ?: "SFU connection timed out",
+ ),
+ )
+ }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| when (sfuConnectionResult.cause) { | |
| is SfuConnectFailure.SocketStateObservationTimeout -> { | |
| // REJOIN (not FAST) on purpose: a connect timeout means the initial | |
| // join never completed. There is no established SFU session or | |
| // negotiated media path to resume, so a FAST resume would likely hit | |
| // PARTICIPANT_NOT_FOUND and burn attempts before the loop escalates. | |
| // A full REJOIN re-fetches credentials and starts a clean join, which | |
| // is the only thing that can actually succeed here. | |
| logger.w { | |
| "[_join] Recoverable SFU connect timeout with no recovery started — triggering REJOIN" | |
| "[_join] SFU socket state observation timed out with no recovery started — triggering REJOIN" | |
| } | |
| scope.launch { | |
| reconnect( | |
| WebsocketReconnectStrategy.WEBSOCKET_RECONNECT_STRATEGY_REJOIN, | |
| "join-recoverable-connect-failure", | |
| ) | |
| } | |
| } else { | |
| logger.w { "[_join] Recoverable SFU connection failure — awaiting recovery outcome" } | |
| } | |
| if (!didReconnectSucceed()) { | |
| logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" } | |
| is SfuConnectFailure.RecoverableSocketFailure -> { | |
| logger.w { "[_join] Recoverable SFU socket failure — awaiting recovery outcome" } | |
| if (!didReconnectSucceed()) { | |
| logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" } | |
| sendJoinErrorAnalytics(sfuConnectionResult) | |
| return Failure( | |
| Error.GenericError( | |
| sfuConnectionResult.cause.error.message ?: "SFU connection failed", | |
| ), | |
| ) | |
| } | |
| } | |
| is SfuConnectFailure.TerminalSocketFailure -> { | |
| logger.e { | |
| "[_join] Got terminal error while connecting to SFU. Error : $sfuConnectionResult" | |
| } | |
| sendJoinErrorAnalytics(sfuConnectionResult) | |
| return Failure( | |
| Error.GenericError( | |
| sfuConnectionResult.error.message ?: "SFU connection failed", | |
| sfuConnectionResult.cause.error.message ?: "RtcSession error occurred.", | |
| ), | |
| ) | |
| } | |
| } else { | |
| logger.e { | |
| "[_join] Got non recoverable error while connecting to SFU. Error : $sfuConnectionResult" | |
| } | |
| sendJoinErrorAnalytics(sfuConnectionResult) | |
| return Failure( | |
| Error.GenericError( | |
| sfuConnectionResult.error.message ?: "RtcSession error occurred.", | |
| ), | |
| ) | |
| } | |
| } | |
| } | |
| when (sfuConnectionResult.cause) { | |
| is SfuConnectFailure.SocketStateObservationTimeout -> { | |
| // REJOIN (not FAST) on purpose: a connect timeout means the initial | |
| // join never completed. There is no established SFU session or | |
| // negotiated media path to resume, so a FAST resume would likely hit | |
| // PARTICIPANT_NOT_FOUND and burn attempts before the loop escalates. | |
| // A full REJOIN re-fetches credentials and starts a clean join, which | |
| // is the only thing that can actually succeed here. | |
| logger.w { | |
| "[_join] SFU socket state observation timed out with no recovery started — triggering REJOIN" | |
| } | |
| scope.launch { | |
| reconnect( | |
| WebsocketReconnectStrategy.WEBSOCKET_RECONNECT_STRATEGY_REJOIN, | |
| "join-recoverable-connect-failure", | |
| ) | |
| } | |
| if (!didReconnectSucceed()) { | |
| logger.e { "[_join] Could not recover from socket state observation timeout. Error : $sfuConnectionResult" } | |
| sendJoinErrorAnalytics(sfuConnectionResult) | |
| return Failure( | |
| Error.GenericError( | |
| sfuConnectionResult.cause.error.message ?: "SFU connection timed out", | |
| ), | |
| ) | |
| } | |
| } | |
| is SfuConnectFailure.RecoverableSocketFailure -> { | |
| logger.w { "[_join] Recoverable SFU socket failure — awaiting recovery outcome" } | |
| if (!didReconnectSucceed()) { | |
| logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" } | |
| sendJoinErrorAnalytics(sfuConnectionResult) | |
| return Failure( | |
| Error.GenericError( | |
| sfuConnectionResult.cause.error.message ?: "SFU connection failed", | |
| ), | |
| ) | |
| } | |
| } | |
| is SfuConnectFailure.TerminalSocketFailure -> { | |
| logger.e { | |
| "[_join] Got terminal error while connecting to SFU. Error : $sfuConnectionResult" | |
| } | |
| sendJoinErrorAnalytics(sfuConnectionResult) | |
| return Failure( | |
| Error.GenericError( | |
| sfuConnectionResult.cause.error.message ?: "RtcSession error occurred.", | |
| ), | |
| ) | |
| } | |
| } | |
| } | |
| } |
🤖 Prompt for 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.
In
`@stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt`
around lines 783 - 828, Update the SocketStateObservationTimeout branch in
_join() to await the REJOIN result instead of launching it asynchronously and
falling through. Mirror the RecoverableSocketFailure handling: invoke reconnect
with REJOIN, await didReconnectSucceed(), and return an appropriate Failure
after recovery settles unsuccessfully, ensuring _join() does not proceed to
monitorSession or return Success prematurely.
Contributor
|
Merged the other PR #1744, instead of this as that looked cleaner |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Goal
Refactor SFU connect failure classification from exception subclasses to typed domain failures.
The implementation on PR #1741 uses
SfuConnectException.TimeoutandSfuConnectException.Disconnectedto tell_join()how to handle a failed SFUconnect attempt. That works mechanically, but the model is misleading:
Timeoutis not a network-layer exception here; it meansconnectInternal()gave up waiting for the socket to reach a terminal state.
Disconnectedis an SFU socket state, not an exception._join()is really branching on domain failure causes, not exception types.This PR keeps the recovery behavior, but makes the producer/consumer boundary
clearer:
connectInternal()reports a typed SFU connect failure, and_join()decides how to orchestrate recovery from that failure.
Implementation
Replace
SfuConnectExceptionwith a typedSfuConnectFailuredomain model:Update SfuConnectionResult.Failure to carry the typed failure:
RtcSession.connectInternal()now produces one of three domain failures:SocketStateObservationTimeout:connectInternal()stopped waiting for thesocket to reach
ConnectedorDisconnectedRecoverableSocketFailure: the socket reached a recoverable disconnectedstate, and
stateJobalready owns reconnectTerminalSocketFailure: the socket reached a terminal disconnected state andshould fail without recovery
Call._join()branches on the typed failure instead of exception subclasses:REJOINforSocketStateObservationTimeoutRecoverableSocketFailureTerminalSocketFailureThis keeps
Exceptionas error detail and usesSfuConnectFailureas the domain-level connection result.🎨 UI Changes
None
Testing
None
Summary by CodeRabbit