Always recreate stream on failure. #205
Conversation
Recreate stream if timed out.
WalkthroughRecent updates involve altering the Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant CodeBlock
participant TimeoutException
participant StreamManager
Client->>CodeBlock: Execute Code Block
CodeBlock-->>Client: Success or Delay
alt Delay
Client->>TimeoutException: Time exceeds limit
TimeoutException-->>Client: Throws TimeoutException
else Success
Client-->>StreamManager: Manage streams
StreamManager-->>Client: Stream Response
end
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (invoked as PR comments)
Additionally, you can add CodeRabbit Configration File (
|
![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
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (1)
- yorkie/src/main/kotlin/dev/yorkie/core/Client.kt (5 hunks)
Additional comments not posted (2)
yorkie/src/main/kotlin/dev/yorkie/core/Client.kt (2)
43-43: The addition ofTimeoutExceptionandwithTimeoutOrNullare appropriate for the new timeout functionality.Also applies to: 79-79
419-426: The implementation ofsafeCloseusingNonCancellablecontext andrunCatchingis robust and ensures that the stream closure is handled gracefully without causing unintended exceptions.
| val stream = withTimeoutOrNull(1_000) { | ||
| service.watchDocument( | ||
| attachment.document.key.documentBasedRequestHeader, | ||
| ).also { | ||
| latestStream = it | ||
| } | ||
| } ?: continue | ||
| val streamJob = launch(start = CoroutineStart.UNDISPATCHED) { | ||
| val channel = stream.responseChannel() | ||
| var retry = 0 | ||
| while (!stream.isReceiveClosed() && !channel.isClosedForReceive) { | ||
| val receiveResult = channel.receiveCatching() | ||
| receiveResult.onSuccess { | ||
| attachment.document.publishEvent(StreamConnectionChanged.Connected) | ||
| handleWatchDocumentsResponse(attachment.document.key, it) | ||
| retry = 0 | ||
| }.onFailure { | ||
| if (receiveResult.isClosed) { | ||
| return@onFailure | ||
| withTimeoutOrNull(60_000) { | ||
| val receiveResult = channel.receiveCatching() | ||
| receiveResult.onSuccess { | ||
| attachment.document.publishEvent(StreamConnectionChanged.Connected) | ||
| handleWatchDocumentsResponse(attachment.document.key, it) | ||
| }.onFailure { | ||
| if (receiveResult.isClosed) { | ||
| stream.safeClose() | ||
| return@onFailure | ||
| } | ||
| handleWatchStreamFailure(attachment.document, stream, it) | ||
| }.onClosed { | ||
| handleWatchStreamFailure( | ||
| attachment.document, | ||
| stream, | ||
| it ?: ClosedReceiveChannelException("Channel was closed"), | ||
| ) | ||
| } | ||
| retry++ | ||
| handleWatchStreamFailure(attachment.document, stream, it, retry > 3) | ||
| }.onClosed { | ||
| } ?: run { | ||
| handleWatchStreamFailure( | ||
| attachment.document, | ||
| stream, | ||
| it ?: ClosedReceiveChannelException("Channel was closed"), | ||
| true, | ||
| TimeoutException("channel timed out"), |
There was a problem hiding this comment.
Tip
Codebase Verification
The ClientTest.kt file includes instances of withTimeoutOrNull, but they are used to test document synchronization and not directly related to createWatchJob. This indicates that the timeout handling in createWatchJob is not explicitly tested.
ClientTest.ktcontains tests for document synchronization usingwithTimeoutOrNull.- No specific tests cover the timeout scenarios within
createWatchJob.
Analysis chain
The refactoring of createWatchJob to include timeout handling and stream recreation upon failure is well-implemented. However, ensure proper testing of the timeout values to match expected behavior in production environments.
To ensure that the timeout values in createWatchJob are properly tested, we need to search for test cases that might be testing this method explicitly or indirectly. Let's look for test files that might be related to this functionality and check if they include any tests for timeouts.
The search results indicate that there are no specific test cases directly testing the createWatchJob method for timeout values, but there are instances of withTimeoutOrNull in ClientTest.kt. This suggests that some timeout behavior is being tested, but we need to verify if these tests are comprehensive and cover the scenarios in createWatchJob.
To proceed, let's review the contents of ClientTest.kt to determine if the timeout values and behaviors in createWatchJob are adequately tested.
Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Verify the timeout values are tested.
# Test: Search for the test cases related to `createWatchJob`.
rg --type kotlin --files-with-matches 'fun createWatchJob' | xargs rg 'withTimeoutOrNull'
Length of output: 241
Script:
#!/bin/bash
# Search for test files that might be related to the Client class or createWatchJob method.
fd -e kt -e kts -e java --type f -x grep -l 'Client' {} +
fd -e kt -e kts -e java --type f -x grep -l 'createWatchJob' {} +
# Check if any of these files contain tests for timeout values.
grep -E 'withTimeoutOrNull|TimeoutException' $(fd -e kt -e kts -e java --type f -x grep -l 'Client' {} +)
grep -E 'withTimeoutOrNull|TimeoutException' $(fd -e kt -e kts -e java --type f -x grep -l 'createWatchJob' {} +)
Length of output: 21210
Script:
#!/bin/bash
# Display the relevant lines from ClientTest.kt that include withTimeoutOrNull to verify if they test createWatchJob.
rg --context 10 'withTimeoutOrNull' ./yorkie/src/androidTest/kotlin/dev/yorkie/core/ClientTest.kt
Length of output: 2242
|
all the test cases are passing on the real server. |
Recreate stream if timed out.
What this PR does / why we need it?
When ClosedReceiveChannelException is received, reusing the same stream caused channel receive to suspend forever.
Therefore we only reused the stream on failure not related to ClosedReceiveChannelException.
But there is suspicion that other failures can also cause it to hang forever.
This PR always creates a new stream on any failures. Also it uses
withTimeoutto prevent such behavior and recreate the stream.Any background context you want to provide?
What are the relevant tickets?
Fixes #
Checklist
Summary by CodeRabbit
0.4.24-rc2for the upcoming release.