feat: detect clock drift and handle INVALID_TIMESTAMP better for event streaming

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

@@ -275,6 +275,10 @@
                 android:name="com.flipcash.app.internal.startup.TraceInitializer"
                 android:value="androidx.startup" />
 
+            <meta-data
+                android:name="com.flipcash.app.internal.startup.ClockDriftInitializer"
+                android:value="androidx.startup" />
+
             <meta-data
                 android:name="com.flipcash.app.internal.startup.FirebaseInitializer"
                 android:value="androidx.startup" />

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

@@ -0,0 +1,48 @@
+package com.flipcash.app.internal.startup
+
+import android.content.Context
+import androidx.startup.Initializer
+import com.flipcash.app.internal.time.SntpClient
+import com.getcode.utils.ClockSource
+import com.getcode.utils.TraceManager
+import com.getcode.utils.trace
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+
+/**
+ * Best-effort SNTP measurement of how far the device clock has drifted from real time, taken once at
+ * startup. The result is recorded on [TraceManager] (so it appears in the exported log header) and
+ * emitted as a launch breadcrumb. A large drift is the likely cause of `INVALID_TIMESTAMP` request
+ * rejections, so capturing it on launch gives a diagnostic signal even before the event stream
+ * connects (and even when a bad clock would prevent that stream from authenticating at all).
+ *
+ * Depends on [TraceInitializer] so [TraceManager] is initialized before we record/trace.
+ */
+class ClockDriftInitializer : Initializer<Unit> {
+
+    override fun create(context: Context) {
+        CoroutineScope(Dispatchers.IO).launch {
+            val offsetMillis = SntpClient.queryClockOffsetMillis()
+            if (offsetMillis == null) {
+                trace(tag = "clock", message = "Clock drift on launch: unavailable (SNTP query failed)")
+                return@launch
+            }
+            // drift = device time - true time = negative of the NTP offset (true - device).
+            val driftMillis = -offsetMillis
+            TraceManager.recordClockDrift(driftMillis, source = ClockSource.Sntp)
+            trace(
+                tag = "clock",
+                message = "Clock drift on launch",
+                metadata = {
+                    "driftMs" to driftMillis
+                    "source" to "sntp"
+                },
+            )
+        }
+    }
+
+    override fun dependencies(): List<Class<out Initializer<*>?>?> {
+        return listOf(TraceInitializer::class.java)
+    }
+}

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

@@ -0,0 +1,104 @@
+package com.flipcash.app.internal.time
+
+import android.os.SystemClock
+import java.net.DatagramPacket
+import java.net.DatagramSocket
+import java.net.InetAddress
+
+/**
+ * Minimal, dependency-free SNTP (RFC 4330) client used purely for diagnostics — measuring how far
+ * the device wall clock has drifted from real time.
+ *
+ * NTP is unauthenticated and has no timestamp/replay check, so this works even when the device clock
+ * is wrong enough that backend auth rejects it with `INVALID_TIMESTAMP` — which is exactly the case
+ * we want to detect. Best-effort: returns `null` on any failure. Must be called off the main thread.
+ */
+internal object SntpClient {
+
+    private const val NTP_PORT = 123
+    private const val NTP_PACKET_SIZE = 48
+    private const val NTP_MODE_CLIENT = 3
+    private const val NTP_VERSION = 3
+    private const val NTP_MODE_SERVER = 4
+    private const val NTP_LEAP_NOSYNC = 3
+    private const val NTP_STRATUM_MAX = 15
+
+    // Seconds between the NTP epoch (1900) and the Unix epoch (1970).
+    private const val OFFSET_1900_TO_1970 = 2_208_988_800L
+
+    private const val INDEX_ORIGINATE_TIME = 24
+    private const val INDEX_RECEIVE_TIME = 32
+    private const val INDEX_TRANSMIT_TIME = 40
+
+    /**
+     * Returns the NTP clock offset in milliseconds (`trueTime - deviceTime`): positive means the
+     * device clock is BEHIND real time. Returns `null` on any failure (no network, timeout,
+     * malformed response).
+     */
+    fun queryClockOffsetMillis(host: String = "time.google.com", timeoutMs: Int = 3_000): Long? {
+        return try {
+            DatagramSocket().use { socket ->
+                socket.soTimeout = timeoutMs
+                val address = InetAddress.getByName(host)
+                val buffer = ByteArray(NTP_PACKET_SIZE)
+                buffer[0] = (NTP_MODE_CLIENT or (NTP_VERSION shl 3)).toByte()
+
+                val requestTime = System.currentTimeMillis()
+                val requestTicks = SystemClock.elapsedRealtime()
+                writeTimestamp(buffer, INDEX_TRANSMIT_TIME, requestTime)
+
+                socket.send(DatagramPacket(buffer, buffer.size, address, NTP_PORT))
+                socket.receive(DatagramPacket(buffer, buffer.size))
+                val responseTicks = SystemClock.elapsedRealtime()
+
+                // t4: estimated local time the response arrived, measured via the monotonic clock so
+                // a mid-flight wall-clock change can't corrupt the round trip.
+                val responseTime = requestTime + (responseTicks - requestTicks)
+
+                val leap = (buffer[0].toInt() shr 6) and 0x3
+                val mode = buffer[0].toInt() and 0x7
+                val stratum = buffer[1].toInt() and 0xff
+                if (leap == NTP_LEAP_NOSYNC || mode != NTP_MODE_SERVER || stratum < 1 || stratum > NTP_STRATUM_MAX) {
+                    return null
+                }
+
+                val originateTime = readTimestamp(buffer, INDEX_ORIGINATE_TIME) // t1 (echoed)
+                val receiveTime = readTimestamp(buffer, INDEX_RECEIVE_TIME)     // t2
+                val transmitTime = readTimestamp(buffer, INDEX_TRANSMIT_TIME)   // t3
+
+                // Clock offset = ((t2 - t1) + (t3 - t4)) / 2
+                ((receiveTime - originateTime) + (transmitTime - responseTime)) / 2
+            }
+        } catch (_: Exception) {
+            null
+        }
+    }
+
+    private fun writeTimestamp(buffer: ByteArray, offset: Int, timeMillis: Long) {
+        val seconds = timeMillis / 1_000L + OFFSET_1900_TO_1970
+        val milliseconds = timeMillis % 1_000L
+        buffer[offset] = (seconds shr 24).toByte()
+        buffer[offset + 1] = (seconds shr 16).toByte()
+        buffer[offset + 2] = (seconds shr 8).toByte()
+        buffer[offset + 3] = seconds.toByte()
+        val fraction = milliseconds * 0x1_0000_0000L / 1_000L
+        buffer[offset + 4] = (fraction shr 24).toByte()
+        buffer[offset + 5] = (fraction shr 16).toByte()
+        buffer[offset + 6] = (fraction shr 8).toByte()
+        buffer[offset + 7] = fraction.toByte()
+    }
+
+    private fun readTimestamp(buffer: ByteArray, offset: Int): Long {
+        val seconds = readUnsigned32(buffer, offset)
+        val fraction = readUnsigned32(buffer, offset + 4)
+        return (seconds - OFFSET_1900_TO_1970) * 1_000L + (fraction * 1_000L) / 0x1_0000_0000L
+    }
+
+    private fun readUnsigned32(buffer: ByteArray, offset: Int): Long {
+        val b0 = buffer[offset].toLong() and 0xff
+        val b1 = buffer[offset + 1].toLong() and 0xff
+        val b2 = buffer[offset + 2].toLong() and 0xff
+        val b3 = buffer[offset + 3].toLong() and 0xff
+        return (b0 shl 24) or (b1 shl 16) or (b2 shl 8) or b3
+    }
+}

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

@@ -153,8 +153,18 @@ private fun buildDeviceHeader(context: Context): String {
         appendLine("ABI:            ${Build.SUPPORTED_ABIS.joinToString()}")
         appendLine("Locale:         ${Locale.getDefault()}")
         appendLine("Timezone:       ${TimeZone.getDefault().id}")
+        appendLine("Clock Drift:    ${formatClockDrift()}")
         appendLine("Exported:       ${Instant.now()}")
         appendLine("=".repeat(60))
         appendLine()
     }
 }
+
+/** Renders the latest clock-drift measurement for the export header. */
+private fun formatClockDrift(): String {
+    val snapshot = TraceManager.clockDriftSnapshot()
+        ?: return "unknown (no server sync yet)"
+    val sign = if (snapshot.driftMillis >= 0) "+" else ""
+    val age = snapshot.ageMillis?.let { ", synced ${it / 1000}s ago" }.orEmpty()
+    return "$sign${snapshot.driftMillis} ms (${snapshot.source}$age)"
+}

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

@@ -2,6 +2,7 @@ package com.getcode.utils
 
 import android.annotation.SuppressLint
 import android.content.Context
+import android.os.SystemClock
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
@@ -84,6 +85,35 @@ object TraceManager {
         onUserIdChanged = listener
     }
 
+    @Volatile
+    private var clockDriftMillis: Long? = null
+    @Volatile
+    private var clockDriftSource: ClockSource = ClockSource.Unknown
+    @Volatile
+    private var clockDriftSyncElapsedRealtime: Long? = null
+
+    /**
+     * Records the most recent measured clock drift (device time minus authoritative server/true
+     * time; positive means the device clock is ahead). [source] identifies where the measurement
+     * came from, e.g. `"sntp"` or `"event-stream"`.
+     */
+    fun recordClockDrift(driftMillis: Long, source: ClockSource) {
+        clockDriftMillis = driftMillis
+        clockDriftSource = source
+        clockDriftSyncElapsedRealtime = SystemClock.elapsedRealtime()
+    }
+
+    /** Latest clock-drift measurement, or `null` if no server time has been observed yet. */
+    fun clockDriftSnapshot(): ClockDriftSnapshot? {
+        val drift = clockDriftMillis ?: return null
+        val ageMillis = clockDriftSyncElapsedRealtime?.let { SystemClock.elapsedRealtime() - it }
+        return ClockDriftSnapshot(
+            driftMillis = drift,
+            source = clockDriftSource,
+            ageMillis = ageMillis,
+        )
+    }
+
     fun addPlugin(plugin: TraceLogPlugin) {
         plugins.add(plugin)
     }
@@ -143,9 +173,26 @@ object TraceManager {
         _userId = null
         onUserIdChanged = null
         includeRpcBodies = false
+        clockDriftMillis = null
+        clockDriftSource = ClockSource.Unknown
+        clockDriftSyncElapsedRealtime = null
     }
 }
 
+/** A point-in-time clock-drift measurement captured by [TraceManager.recordClockDrift]. */
+data class ClockDriftSnapshot(
+    /** Device time minus true/server time, in milliseconds. Positive = device clock is ahead. */
+    val driftMillis: Long,
+    /** Where the measurement came from, e.g. `"sntp"` or `"event-stream"`. */
+    val source: ClockSource,
+    /** Milliseconds since the measurement was taken, or `null` if unknown. */
+    val ageMillis: Long?,
+)
+
+enum class ClockSource {
+    Unknown, Sntp, EventStream
+}
+
 /**
  * Primary logging entry point. Writes [message] to Timber and forwards a
  * breadcrumb to every registered [BreadcrumbSink].

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

@@ -10,6 +10,8 @@ import com.flipcash.services.models.chat.ChatUpdate
 import com.getcode.ed25519.Ed25519.KeyPair
 import com.getcode.opencode.internal.bidi.BidirectionalStreamReference
 import com.getcode.opencode.internal.bidi.openBidirectionalStream
+import com.getcode.utils.ClockSource
+import com.getcode.utils.TraceManager
 import com.getcode.utils.TraceType
 import com.getcode.utils.trace
 import com.google.protobuf.Timestamp
@@ -69,6 +71,8 @@ internal class EventStreamingService @Inject constructor(
             },
             reconnectOnUnavailable = true,
             reconnectOnCancelled = true,
+            reconnectOnAborted = true,
+            reconnectDelayMs = 1_000L,
             onError = { onError(it) },
             responseHandler = { response, sendRequest ->
                 when (response.typeCase) {
@@ -83,7 +87,9 @@ internal class EventStreamingService @Inject constructor(
                         streamRef.receivedPing(updatedTimeout = response.ping.pingDelay.seconds * 1_000L)
                         sendRequest(pong)
                         val serverTime = Instant.fromEpochSeconds(response.ping.timestamp.seconds, response.ping.timestamp.nanos)
-                        trace(tag = "event-stream", message = "Pong. Server timestamp: $serverTime")
+                        val driftMillis = Clock.System.now().toEpochMilliseconds() - serverTime.toEpochMilliseconds()
+                        TraceManager.recordClockDrift(driftMillis, source = ClockSource.EventStream)
+                        trace(tag = "event-stream", message = "Pong. Server timestamp: $serverTime (drift ${driftMillis}ms)")
                     }
 
                     RpcEventStreamingService.StreamEventsResponse.TypeCase.EVENTS -> {