Calculate createdLocallyAt using server clock offset #6199
+664
−12
There are no files selected for viewing
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
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,177 @@ | ||
| /* | ||
| * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. | ||
| * | ||
| * Licensed under the Stream License; | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://github.com/GetStream/stream-chat-android/blob/main/LICENSE | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package io.getstream.chat.android.client.utils.internal | ||
|
|
||
| import io.getstream.chat.android.client.events.ConnectedEvent | ||
| import io.getstream.chat.android.client.events.HealthEvent | ||
| import io.getstream.chat.android.core.internal.InternalStreamChatApi | ||
| import java.util.Date | ||
|
|
||
| /** | ||
| * Tracks the offset between the local device clock and the server clock using | ||
| * NTP-style estimation from WebSocket health check round-trips. | ||
| * | ||
| * The algorithm keeps only the sample with the lowest observed RTT, since a | ||
| * smaller round-trip means less room for network asymmetry to distort the | ||
| * measurement. Under the assumption that clock skew is constant for the | ||
| * duration of a session, the estimate monotonically improves over time. | ||
| * | ||
| * Thread-safe: single-field writes use [Volatile] for visibility; compound | ||
| * read-modify-write sequences are guarded by [lock] for atomicity. | ||
| * | ||
| * @param localTimeMs Clock source for the local device time (injectable for tests). | ||
| * @param maxRttMs Upper bound on plausible RTT. Samples exceeding this are | ||
| * discarded as stale or mismatched. Defaults to the health check cycle | ||
| * interval (MONITOR_INTERVAL + HEALTH_CHECK_INTERVAL = 11 000 ms). | ||
| * @param maxOffsetMs Upper bound on the absolute value of the computed clock offset. | ||
| * If the derived offset exceeds this threshold the sample is considered unreliable | ||
| * (e.g. a stale / static server timestamp in a test environment) and the offset is | ||
| * reset to zero so that [estimatedServerTime] falls back to the raw local time. | ||
| * Defaults to 1 hour, which is already far beyond any real-world NTP drift. | ||
| */ | ||
| @InternalStreamChatApi | ||
| public class ServerClockOffset( | ||
| private val localTimeMs: () -> Long = { System.currentTimeMillis() }, | ||
| private val maxRttMs: Long = DEFAULT_MAX_RTT_MS, | ||
| private val maxOffsetMs: Long = DEFAULT_MAX_OFFSET_MS, | ||
| ) { | ||
|
|
||
| private val lock = Any() | ||
|
|
||
| @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 | ||
|
|
||
| /** | ||
| * Record the local time immediately before starting a WebSocket connection. | ||
| * When the next [ConnectedEvent] arrives, [onConnected] will pair with this | ||
| * timestamp to compute the offset using the NTP midpoint formula. | ||
| */ | ||
| internal fun onConnectionStarted() { | ||
| connectionStartedAtMs = localTimeMs() | ||
| } | ||
|
|
||
| /** | ||
| * Record the local time immediately before sending a health check echo. | ||
| * The next [onHealthCheck] call will pair with this timestamp to compute RTT. | ||
| */ | ||
| internal fun onHealthCheckSent() { | ||
| healthCheckSentAtMs = localTimeMs() | ||
| } | ||
|
|
||
| /** | ||
| * Calibration from a [ConnectedEvent]. | ||
| * | ||
| * If [onConnectionStarted] was called before this connection (e.g. right before | ||
| * opening the WebSocket), uses the NTP midpoint of (connectionStartedAt, receivedAt) | ||
| * and serverTime for a more accurate offset. Otherwise falls back to a naive | ||
| * `localTime - serverTime` estimate. | ||
| * | ||
| * Resets health check state, since a new connection means any in-flight health | ||
| * check from the previous connection is stale. | ||
| */ | ||
| internal fun onConnected(serverTime: Date) { | ||
| synchronized(lock) { | ||
| bestRttMs = Long.MAX_VALUE | ||
| healthCheckSentAtMs = 0L | ||
| offsetMs = 0L | ||
|
|
||
| val receivedAtMs = localTimeMs() | ||
| val startedAtMs = connectionStartedAtMs | ||
| connectionStartedAtMs = 0L | ||
|
|
||
| if (startedAtMs > 0L) { | ||
| val rtt = receivedAtMs - startedAtMs | ||
| if (rtt in 1..maxRttMs) { | ||
| acceptOffset((startedAtMs + receivedAtMs) / 2 - serverTime.time) | ||
| bestRttMs = rtt | ||
| return | ||
| } | ||
| } | ||
| acceptOffset(receivedAtMs - serverTime.time) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Refine the offset using a [HealthEvent] paired with [onHealthCheckSent]. | ||
| * | ||
| * Computes RTT from the stored send time and the current receive time, | ||
| * then applies the NTP midpoint formula: | ||
| * ``` | ||
| * offset = (sentAt + receivedAt) / 2 - serverTime | ||
| * ``` | ||
| * | ||
| * The sample is accepted only if: | ||
| * - There is a pending [onHealthCheckSent] timestamp. | ||
| * - RTT is positive (guards against clock anomalies). | ||
| * - RTT is below [maxRttMs] (rejects stale / mismatched pairs). | ||
| * - RTT is lower than any previous sample (min-RTT selection). | ||
| */ | ||
| internal fun onHealthCheck(serverTime: Date) { | ||
| synchronized(lock) { | ||
| val sentAtMs = healthCheckSentAtMs | ||
| if (sentAtMs <= 0L) return | ||
| healthCheckSentAtMs = 0L | ||
|
|
||
| val receivedAtMs = localTimeMs() | ||
| val rtt = receivedAtMs - sentAtMs | ||
| if (rtt !in 1..maxRttMs) return | ||
|
|
||
| if (rtt < bestRttMs) { | ||
| bestRttMs = rtt | ||
| acceptOffset((sentAtMs + receivedAtMs) / 2 - serverTime.time) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Returns the current time adjusted to the server timescale. | ||
| * | ||
| * Before the first [onConnected] call, this returns the raw local time | ||
| * (offset = 0). | ||
| */ | ||
| @InternalStreamChatApi | ||
| public fun estimatedServerTime(): Date = | ||
| Date(localTimeMs() - offsetMs) | ||
|
|
||
| /** | ||
| * Accepts [candidate] as the new [offsetMs] only when its absolute value is within | ||
| * [maxOffsetMs]. Offsets that are implausibly large (e.g. produced by a stale or | ||
| * static server timestamp) are silently discarded and [offsetMs] is left unchanged. | ||
| * | ||
| * Note: callers that want a rejected offset to reset to zero (e.g. [onConnected]) | ||
| * should set [offsetMs] = 0 before calling this function. | ||
| */ | ||
| private fun acceptOffset(candidate: Long) { | ||
| if (kotlin.math.abs(candidate) <= maxOffsetMs) { | ||
| offsetMs = candidate | ||
| } | ||
| } | ||
|
|
||
| internal companion object { | ||
| internal const val DEFAULT_MAX_RTT_MS = 11_000L | ||
| internal const val DEFAULT_MAX_OFFSET_MS = 3_600_000L // 1 hour | ||
| } | ||
| } | ||
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
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
Oops, something went wrong.
Oops, something went wrong.
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.
Uh oh!
There was an error while loading. Please reload this page.