Calculate createdLocallyAt using server clock offset#6199
Draft
VelikovPetar wants to merge 4 commits intodevelopfrom
Draft
Calculate createdLocallyAt using server clock offset#6199VelikovPetar wants to merge 4 commits intodevelopfrom
VelikovPetar wants to merge 4 commits intodevelopfrom
Conversation
Co-Authored-By: Claude <noreply@anthropic.com>
Contributor
PR checklist ✅All required conditions are satisfied:
🎉 Great job! This PR is ready for review. |
VelikovPetar
changed the title
VelikovPetar
changed the title
Contributor
SDK Size Comparison 📏
|
WalkthroughA new ServerClockOffset utility class is introduced to estimate server time via NTP-style synchronization using WebSocket health checks. The dependency is integrated through ChatClient, ChatModule, and ChatSocket to synchronize local timestamps with server time during message creation and connection lifecycle events. Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client
participant Socket as ChatSocket
participant Offset as ServerClockOffset
participant State as State Service
Client->>Socket: initiateConnection()
activate Socket
Socket->>Offset: onConnectionStarted()
activate Offset
Offset->>Offset: recordConnectionStart(localTime)
deactivate Offset
Socket->>Socket: createWebSocket()
Note over Socket: Connection Established
Socket->>Offset: onConnected(serverTime)
activate Offset
Offset->>Offset: calibrate(serverTime, localTime)
Offset->>Offset: calculateOffset via NTP midpoint
deactivate Offset
Socket->>State: notifyConnected()
rect rgba(100, 200, 150, 0.5)
Note over Socket: Health Check Loop
Socket->>Offset: onHealthCheckSent()
activate Offset
Offset->>Offset: recordHealthCheckSent(localTime)
deactivate Offset
Socket->>Socket: sendHealthMessage()
Socket->>Offset: onHealthCheck(serverTime)
activate Offset
Offset->>Offset: refineOffset via RTT & NTP
deactivate Offset
end
deactivate Socket
Client->>Client: createMessage()
Client->>Offset: estimatedServerTime()
activate Offset
Offset->>Offset: return localTime + offset
deactivate Offset
Client->>Client: setMessageCreatedLocallyAt()
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
Tip Try Coding Plans. Let us write the prompt for your AI agent so you can ship faster (with fewer bugs). 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: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In
`@stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/ChatSocket.kt`:
- Around line 66-70: The health-check timestamp is recorded before confirming
the echo was actually sent; update the checkCallback so that when
chatSocketStateService.currentState is a State.Connected and you obtain the
event, you call sendEvent(it) first and only invoke
serverClockOffset.onHealthCheckSent() if sendEvent(it) returns true (i.e., the
echo was successfully sent). Ensure you reference checkCallback,
State.Connected, sendEvent(it), and serverClockOffset.onHealthCheckSent() when
making the change.
In
`@stream-chat-android-client/src/main/java/io/getstream/chat/android/client/utils/internal/ServerClockOffset.kt`:
- Around line 44-55: The compound read/modify/write on volatile fields
(offsetMs, bestRttMs, healthCheckSentAtMs, connectionStartedAtMs) in
onConnected() and onHealthCheck() is racy; wrap those multi-field state
transitions in a mutual-exclusion block using a dedicated lock (e.g., private
val stateLock = Any()) and replace the check-then-set sequences with
synchronized(stateLock) { ... } blocks around the code that updates bestRttMs
and offsetMs so stale comparisons cannot overwrite better values; keep
single-field setters (onConnectionStarted(), onHealthCheckSent()) unchanged and
update the KDoc to state that compound operations are synchronized via stateLock
rather than relying solely on `@Volatile`.
ℹ️ Review info
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Linear integration is disabled
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (12)
stream-chat-android-client/src/main/java/io/getstream/chat/android/client/ChatClient.ktstream-chat-android-client/src/main/java/io/getstream/chat/android/client/di/ChatModule.ktstream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/ChatSocket.ktstream-chat-android-client/src/main/java/io/getstream/chat/android/client/utils/internal/ServerClockOffset.ktstream-chat-android-client/src/test/java/io/getstream/chat/android/client/ChatClientConnectionTests.ktstream-chat-android-client/src/test/java/io/getstream/chat/android/client/ChatClientTest.ktstream-chat-android-client/src/test/java/io/getstream/chat/android/client/DependencyResolverTest.ktstream-chat-android-client/src/test/java/io/getstream/chat/android/client/MockClientBuilder.ktstream-chat-android-client/src/test/java/io/getstream/chat/android/client/chatclient/BaseChatClientTest.ktstream-chat-android-client/src/test/java/io/getstream/chat/android/client/debugger/ChatClientDebuggerTest.ktstream-chat-android-client/src/test/java/io/getstream/chat/android/client/socket/FakeChatSocket.ktstream-chat-android-client/src/test/java/io/getstream/chat/android/client/utils/internal/ServerClockOffsetTest.kt
Comment on lines
+66
to
+70
| checkCallback = { | ||
| (chatSocketStateService.currentState as? State.Connected)?.event?.let { | ||
| serverClockOffset.onHealthCheckSent() | ||
| sendEvent(it) | ||
| } |
There was a problem hiding this comment.
Record health-check send time only when the echo is actually sent.
At Line 68, onHealthCheckSent() runs before checking whether sendEvent(it) succeeded. If sendEvent returns false, the sample can be orphaned and later paired with an unrelated HealthEvent, skewing offset estimation.
Proposed patch
checkCallback = {
(chatSocketStateService.currentState as? State.Connected)?.event?.let {
- serverClockOffset.onHealthCheckSent()
- sendEvent(it)
+ if (sendEvent(it)) {
+ serverClockOffset.onHealthCheckSent()
+ }
}
},📝 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
| checkCallback = { | |
| (chatSocketStateService.currentState as? State.Connected)?.event?.let { | |
| serverClockOffset.onHealthCheckSent() | |
| sendEvent(it) | |
| } | |
| checkCallback = { | |
| (chatSocketStateService.currentState as? State.Connected)?.event?.let { | |
| if (sendEvent(it)) { | |
| serverClockOffset.onHealthCheckSent() | |
| } | |
| } | |
| }, |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@stream-chat-android-client/src/main/java/io/getstream/chat/android/client/socket/ChatSocket.kt`
around lines 66 - 70, The health-check timestamp is recorded before confirming
the echo was actually sent; update the checkCallback so that when
chatSocketStateService.currentState is a State.Connected and you obtain the
event, you call sendEvent(it) first and only invoke
serverClockOffset.onHealthCheckSent() if sendEvent(it) returns true (i.e., the
echo was successfully sent). Ensure you reference checkCallback,
State.Connected, sendEvent(it), and serverClockOffset.onHealthCheckSent() when
making the change.
Comment on lines
+44
to
+55
| @Volatile | ||
| private var offsetMs: Long = 0L | ||
|
|
||
| @Volatile | ||
| private var bestRttMs: Long = Long.MAX_VALUE | ||
|
|
||
| @Volatile | ||
| private var healthCheckSentAtMs: Long = 0L | ||
|
|
||
| @Volatile | ||
| private var connectionStartedAtMs: Long = 0L | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
fd ServerClockOffset.ktRepository: GetStream/stream-chat-android
Length of output: 182
🏁 Script executed:
cat -n "stream-chat-android-client/src/main/java/io/getstream/chat/android/client/utils/internal/ServerClockOffset.kt"Repository: GetStream/stream-chat-android
Length of output: 6257
@Volatile is insufficient for compound state transitions—race conditions can corrupt RTT selection.
Lines 127–130 in onHealthCheck() perform a check-then-set pattern across two fields without mutual exclusion. A thread can read bestRttMs, be preempted while another thread updates both fields with a better value, then resume and overwrite with a worse RTT using the stale comparison. The same pattern appears in onConnected() (lines 92–98, 95–96).
The KDoc claim at line 32 ("Thread-safe: all mutable state is read/written via [Volatile] fields") is misleading—@volatile guarantees visibility, not atomicity of compound operations.
Apply synchronized locks around all multi-field state transitions:
Suggested fix
internal class ServerClockOffset(
private val localTimeMs: () -> Long = { System.currentTimeMillis() },
private val maxRttMs: Long = DEFAULT_MAX_RTT_MS,
) {
+ private val stateLock = Any()
`@Volatile`
private var offsetMs: Long = 0LThen wrap each compound update:
onConnectionStarted(): single field, already safeonHealthCheckSent(): single field, already safeonConnected(): wrap lines 84–101 withsynchronized(stateLock)onHealthCheck(): wrap lines 118–131 withsynchronized(stateLock)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@stream-chat-android-client/src/main/java/io/getstream/chat/android/client/utils/internal/ServerClockOffset.kt`
around lines 44 - 55, The compound read/modify/write on volatile fields
(offsetMs, bestRttMs, healthCheckSentAtMs, connectionStartedAtMs) in
onConnected() and onHealthCheck() is racy; wrap those multi-field state
transitions in a mutual-exclusion block using a dedicated lock (e.g., private
val stateLock = Any()) and replace the check-then-set sequences with
synchronized(stateLock) { ... } blocks around the code that updates bestRttMs
and offsetMs so stale comparisons cannot overwrite better values; keep
single-field setters (onConnectionStarted(), onHealthCheckSent()) unchanged and
update the KDoc to state that compound operations are synchronized via stateLock
rather than relying solely on `@Volatile`.
…_using_server_clock_offset_calculation
|
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
1 participant
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
When a device clock is skewed relative to the server,
createdLocallyAttimestamps assigned to locally-sent messages can be incorrect. This causes cross-user ordering issues — messages from a user with a fast clock appear in the future relative to messages from other users.Implementation
Introduced
ServerClockOffset, an NTP-style estimator that tracks the offset between the local device clock and the server clock using WebSocket round-trips:onConnectionStarted()— records local time before opening the WebSocket.onConnected(serverTime)— computes initial offset from theConnectedEventtimestamp using the NTP midpoint formula:offset = (sentAt + receivedAt) / 2 - serverTime.onHealthCheckSent()— records local time before each health check echo.onHealthCheck(serverTime)— refines the offset from eachHealthEvent, accepting only the sample with the lowest RTT (min-RTT selection minimises network-asymmetry error).estimatedServerTime()— returnsDate(localTime - offset), used inChatClient.ensureCreatedLocallyAt()instead of the raw local clock.ServerClockOffsetis constructed once perChatClientinstance, wired throughChatModuleintoChatSocket(for calibration) andChatClient(for consumption). All mutable state uses@Volatilefor thread safety.UI Changes
clock-before.mov
clock-after.mov
Testing
ServerClockOffsetTestcovers:onConnectedwith and without a pairedonConnectionStartedonHealthCheckmin-RTT selection and stale-sample rejectionmaxRttMsboundary conditionsServerClockOffsetviaFakeChatSocketandMockClientBuilder.Summary by CodeRabbit