Skip to content

Commit 6bb8385

Browse files
bmc08gtbmc08gt
committed
feat: detect clock drift and handle INVALID_TIMESTAMP better for event streaming
Signed-off-by: Brandon McAnsh <git@bmcreations.dev>
1 parent d03434a commit 6bb8385

6 files changed

Lines changed: 220 additions & 1 deletion

File tree

‎apps/flipcash/app/src/main/AndroidManifest.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -275,6 +275,10 @@
275275
android:name="com.flipcash.app.internal.startup.TraceInitializer"
276276
android:value="androidx.startup" />
277277

278+
<meta-data
279+
android:name="com.flipcash.app.internal.startup.ClockDriftInitializer"
280+
android:value="androidx.startup" />
281+
278282
<meta-data
279283
android:name="com.flipcash.app.internal.startup.FirebaseInitializer"
280284
android:value="androidx.startup" />

‎apps/flipcash/app/src/main/kotlin/com/flipcash/app/internal/startup/ClockDriftInitializer.kt‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
package com.flipcash.app.internal.startup
2+
3+
import android.content.Context
4+
import androidx.startup.Initializer
5+
import com.flipcash.app.internal.time.SntpClient
6+
import com.getcode.utils.ClockSource
7+
import com.getcode.utils.TraceManager
8+
import com.getcode.utils.trace
9+
import kotlinx.coroutines.CoroutineScope
10+
import kotlinx.coroutines.Dispatchers
11+
import kotlinx.coroutines.launch
12+
13+
/**
14+
* Best-effort SNTP measurement of how far the device clock has drifted from real time, taken once at
15+
* startup. The result is recorded on [TraceManager] (so it appears in the exported log header) and
16+
* emitted as a launch breadcrumb. A large drift is the likely cause of `INVALID_TIMESTAMP` request
17+
* rejections, so capturing it on launch gives a diagnostic signal even before the event stream
18+
* connects (and even when a bad clock would prevent that stream from authenticating at all).
19+
*
20+
* Depends on [TraceInitializer] so [TraceManager] is initialized before we record/trace.
21+
*/
22+
class ClockDriftInitializer : Initializer<Unit> {
23+
24+
override fun create(context: Context) {
25+
CoroutineScope(Dispatchers.IO).launch {
26+
val offsetMillis = SntpClient.queryClockOffsetMillis()
27+
if (offsetMillis == null) {
28+
trace(tag = "clock", message = "Clock drift on launch: unavailable (SNTP query failed)")
29+
return@launch
30+
}
31+
// drift = device time - true time = negative of the NTP offset (true - device).
32+
val driftMillis = -offsetMillis
33+
TraceManager.recordClockDrift(driftMillis, source = ClockSource.Sntp)
34+
trace(
35+
tag = "clock",
36+
message = "Clock drift on launch",
37+
metadata = {
38+
"driftMs" to driftMillis
39+
"source" to "sntp"
40+
},
41+
)
42+
}
43+
}
44+
45+
override fun dependencies(): List<Class<out Initializer<*>?>?> {
46+
return listOf(TraceInitializer::class.java)
47+
}
48+
}

‎apps/flipcash/app/src/main/kotlin/com/flipcash/app/internal/time/SntpClient.kt‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
package com.flipcash.app.internal.time
2+
3+
import android.os.SystemClock
4+
import java.net.DatagramPacket
5+
import java.net.DatagramSocket
6+
import java.net.InetAddress
7+
8+
/**
9+
* Minimal, dependency-free SNTP (RFC 4330) client used purely for diagnostics — measuring how far
10+
* the device wall clock has drifted from real time.
11+
*
12+
* NTP is unauthenticated and has no timestamp/replay check, so this works even when the device clock
13+
* is wrong enough that backend auth rejects it with `INVALID_TIMESTAMP` — which is exactly the case
14+
* we want to detect. Best-effort: returns `null` on any failure. Must be called off the main thread.
15+
*/
16+
internal object SntpClient {
17+
18+
private const val NTP_PORT = 123
19+
private const val NTP_PACKET_SIZE = 48
20+
private const val NTP_MODE_CLIENT = 3
21+
private const val NTP_VERSION = 3
22+
private const val NTP_MODE_SERVER = 4
23+
private const val NTP_LEAP_NOSYNC = 3
24+
private const val NTP_STRATUM_MAX = 15
25+
26+
// Seconds between the NTP epoch (1900) and the Unix epoch (1970).
27+
private const val OFFSET_1900_TO_1970 = 2_208_988_800L
28+
29+
private const val INDEX_ORIGINATE_TIME = 24
30+
private const val INDEX_RECEIVE_TIME = 32
31+
private const val INDEX_TRANSMIT_TIME = 40
32+
33+
/**
34+
* Returns the NTP clock offset in milliseconds (`trueTime - deviceTime`): positive means the
35+
* device clock is BEHIND real time. Returns `null` on any failure (no network, timeout,
36+
* malformed response).
37+
*/
38+
fun queryClockOffsetMillis(host: String = "time.google.com", timeoutMs: Int = 3_000): Long? {
39+
return try {
40+
DatagramSocket().use { socket ->
41+
socket.soTimeout = timeoutMs
42+
val address = InetAddress.getByName(host)
43+
val buffer = ByteArray(NTP_PACKET_SIZE)
44+
buffer[0] = (NTP_MODE_CLIENT or (NTP_VERSION shl 3)).toByte()
45+
46+
val requestTime = System.currentTimeMillis()
47+
val requestTicks = SystemClock.elapsedRealtime()
48+
writeTimestamp(buffer, INDEX_TRANSMIT_TIME, requestTime)
49+
50+
socket.send(DatagramPacket(buffer, buffer.size, address, NTP_PORT))
51+
socket.receive(DatagramPacket(buffer, buffer.size))
52+
val responseTicks = SystemClock.elapsedRealtime()
53+
54+
// t4: estimated local time the response arrived, measured via the monotonic clock so
55+
// a mid-flight wall-clock change can't corrupt the round trip.
56+
val responseTime = requestTime + (responseTicks - requestTicks)
57+
58+
val leap = (buffer[0].toInt() shr 6) and 0x3
59+
val mode = buffer[0].toInt() and 0x7
60+
val stratum = buffer[1].toInt() and 0xff
61+
if (leap == NTP_LEAP_NOSYNC || mode != NTP_MODE_SERVER || stratum < 1 || stratum > NTP_STRATUM_MAX) {
62+
return null
63+
}
64+
65+
val originateTime = readTimestamp(buffer, INDEX_ORIGINATE_TIME) // t1 (echoed)
66+
val receiveTime = readTimestamp(buffer, INDEX_RECEIVE_TIME) // t2
67+
val transmitTime = readTimestamp(buffer, INDEX_TRANSMIT_TIME) // t3
68+
69+
// Clock offset = ((t2 - t1) + (t3 - t4)) / 2
70+
((receiveTime - originateTime) + (transmitTime - responseTime)) / 2
71+
}
72+
} catch (_: Exception) {
73+
null
74+
}
75+
}
76+
77+
private fun writeTimestamp(buffer: ByteArray, offset: Int, timeMillis: Long) {
78+
val seconds = timeMillis / 1_000L + OFFSET_1900_TO_1970
79+
val milliseconds = timeMillis % 1_000L
80+
buffer[offset] = (seconds shr 24).toByte()
81+
buffer[offset + 1] = (seconds shr 16).toByte()
82+
buffer[offset + 2] = (seconds shr 8).toByte()
83+
buffer[offset + 3] = seconds.toByte()
84+
val fraction = milliseconds * 0x1_0000_0000L / 1_000L
85+
buffer[offset + 4] = (fraction shr 24).toByte()
86+
buffer[offset + 5] = (fraction shr 16).toByte()
87+
buffer[offset + 6] = (fraction shr 8).toByte()
88+
buffer[offset + 7] = fraction.toByte()
89+
}
90+
91+
private fun readTimestamp(buffer: ByteArray, offset: Int): Long {
92+
val seconds = readUnsigned32(buffer, offset)
93+
val fraction = readUnsigned32(buffer, offset + 4)
94+
return (seconds - OFFSET_1900_TO_1970) * 1_000L + (fraction * 1_000L) / 0x1_0000_0000L
95+
}
96+
97+
private fun readUnsigned32(buffer: ByteArray, offset: Int): Long {
98+
val b0 = buffer[offset].toLong() and 0xff
99+
val b1 = buffer[offset + 1].toLong() and 0xff
100+
val b2 = buffer[offset + 2].toLong() and 0xff
101+
val b3 = buffer[offset + 3].toLong() and 0xff
102+
return (b0 shl 24) or (b1 shl 16) or (b2 shl 8) or b3
103+
}
104+
}

‎libs/logging/src/main/kotlin/com/getcode/utils/FileTree.kt‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -153,8 +153,18 @@ private fun buildDeviceHeader(context: Context): String {
153153
appendLine("ABI: ${Build.SUPPORTED_ABIS.joinToString()}")
154154
appendLine("Locale: ${Locale.getDefault()}")
155155
appendLine("Timezone: ${TimeZone.getDefault().id}")
156+
appendLine("Clock Drift: ${formatClockDrift()}")
156157
appendLine("Exported: ${Instant.now()}")
157158
appendLine("=".repeat(60))
158159
appendLine()
159160
}
160161
}
162+
163+
/** Renders the latest clock-drift measurement for the export header. */
164+
private fun formatClockDrift(): String {
165+
val snapshot = TraceManager.clockDriftSnapshot()
166+
?: return "unknown (no server sync yet)"
167+
val sign = if (snapshot.driftMillis >= 0) "+" else ""
168+
val age = snapshot.ageMillis?.let { ", synced ${it / 1000}s ago" }.orEmpty()
169+
return "$sign${snapshot.driftMillis} ms (${snapshot.source}$age)"
170+
}

‎libs/logging/src/main/kotlin/com/getcode/utils/Logging.kt‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package com.getcode.utils
22

33
import android.annotation.SuppressLint
44
import android.content.Context
5+
import android.os.SystemClock
56
import kotlinx.coroutines.channels.BufferOverflow
67
import kotlinx.coroutines.flow.MutableSharedFlow
78
import kotlinx.coroutines.flow.SharedFlow
@@ -84,6 +85,35 @@ object TraceManager {
8485
onUserIdChanged = listener
8586
}
8687

88+
@Volatile
89+
private var clockDriftMillis: Long? = null
90+
@Volatile
91+
private var clockDriftSource: ClockSource = ClockSource.Unknown
92+
@Volatile
93+
private var clockDriftSyncElapsedRealtime: Long? = null
94+
95+
/**
96+
* Records the most recent measured clock drift (device time minus authoritative server/true
97+
* time; positive means the device clock is ahead). [source] identifies where the measurement
98+
* came from, e.g. `"sntp"` or `"event-stream"`.
99+
*/
100+
fun recordClockDrift(driftMillis: Long, source: ClockSource) {
101+
clockDriftMillis = driftMillis
102+
clockDriftSource = source
103+
clockDriftSyncElapsedRealtime = SystemClock.elapsedRealtime()
104+
}
105+
106+
/** Latest clock-drift measurement, or `null` if no server time has been observed yet. */
107+
fun clockDriftSnapshot(): ClockDriftSnapshot? {
108+
val drift = clockDriftMillis ?: return null
109+
val ageMillis = clockDriftSyncElapsedRealtime?.let { SystemClock.elapsedRealtime() - it }
110+
return ClockDriftSnapshot(
111+
driftMillis = drift,
112+
source = clockDriftSource,
113+
ageMillis = ageMillis,
114+
)
115+
}
116+
87117
fun addPlugin(plugin: TraceLogPlugin) {
88118
plugins.add(plugin)
89119
}
@@ -143,9 +173,26 @@ object TraceManager {
143173
_userId = null
144174
onUserIdChanged = null
145175
includeRpcBodies = false
176+
clockDriftMillis = null
177+
clockDriftSource = ClockSource.Unknown
178+
clockDriftSyncElapsedRealtime = null
146179
}
147180
}
148181

182+
/** A point-in-time clock-drift measurement captured by [TraceManager.recordClockDrift]. */
183+
data class ClockDriftSnapshot(
184+
/** Device time minus true/server time, in milliseconds. Positive = device clock is ahead. */
185+
val driftMillis: Long,
186+
/** Where the measurement came from, e.g. `"sntp"` or `"event-stream"`. */
187+
val source: ClockSource,
188+
/** Milliseconds since the measurement was taken, or `null` if unknown. */
189+
val ageMillis: Long?,
190+
)
191+
192+
enum class ClockSource {
193+
Unknown, Sntp, EventStream
194+
}
195+
149196
/**
150197
* Primary logging entry point. Writes [message] to Timber and forwards a
151198
* breadcrumb to every registered [BreadcrumbSink].

‎services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/services/EventStreamingService.kt‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@ import com.flipcash.services.models.chat.ChatUpdate
1010
import com.getcode.ed25519.Ed25519.KeyPair
1111
import com.getcode.opencode.internal.bidi.BidirectionalStreamReference
1212
import com.getcode.opencode.internal.bidi.openBidirectionalStream
13+
import com.getcode.utils.ClockSource
14+
import com.getcode.utils.TraceManager
1315
import com.getcode.utils.TraceType
1416
import com.getcode.utils.trace
1517
import com.google.protobuf.Timestamp
@@ -69,6 +71,8 @@ internal class EventStreamingService @Inject constructor(
6971
},
7072
reconnectOnUnavailable = true,
7173
reconnectOnCancelled = true,
74+
reconnectOnAborted = true,
75+
reconnectDelayMs = 1_000L,
7276
onError = { onError(it) },
7377
responseHandler = { response, sendRequest ->
7478
when (response.typeCase) {
@@ -83,7 +87,9 @@ internal class EventStreamingService @Inject constructor(
8387
streamRef.receivedPing(updatedTimeout = response.ping.pingDelay.seconds * 1_000L)
8488
sendRequest(pong)
8589
val serverTime = Instant.fromEpochSeconds(response.ping.timestamp.seconds, response.ping.timestamp.nanos)
86-
trace(tag = "event-stream", message = "Pong. Server timestamp: $serverTime")
90+
val driftMillis = Clock.System.now().toEpochMilliseconds() - serverTime.toEpochMilliseconds()
91+
TraceManager.recordClockDrift(driftMillis, source = ClockSource.EventStream)
92+
trace(tag = "event-stream", message = "Pong. Server timestamp: $serverTime (drift ${driftMillis}ms)")
8793
}
8894

8995
RpcEventStreamingService.StreamEventsResponse.TypeCase.EVENTS -> {

0 commit comments

Comments
 (0)