Project

General

Profile

Download (6.74 KB)

User Story #47 » phase2-g2-clock-sync-seam.patch

knight8241, 08/07/2026 18:32

View differences:

common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshMessagingService.kt
/** A peer this node fans data-plane traffic out to. Maintained from the registry. */
data class PeerEndpoint(val nodeId: String, val datagram: NodeAddress)
/**
* One completed clock-sync round-trip: [t1Micros] local send, [t2Micros] leader
* receive, [t3Micros] leader send, [t4Micros] local receive — the four stamps
* `MeshClock.observe` folds into an offset/RTT sample.
*/
data class TimeSyncSample(val t1Micros: Long, val t2Micros: Long, val t3Micros: Long, val t4Micros: Long)
/**
* Message hub: assigns sequence numbers + epoch on the way out, applies the
* StalenessGate on the way in, and demultiplexes payload types into typed
......
private val _groupKeys = MutableSharedFlow<GroupKeyDistribution>(extraBufferCapacity = 4)
private val _eventClaims = MutableSharedFlow<Pair<String, WireEventClaim>>(extraBufferCapacity = 32)
private val _eventVerdicts = MutableSharedFlow<WireEventVerdict>(extraBufferCapacity = 32)
private val _timeSyncResponses = MutableSharedFlow<TimeSyncSample>(extraBufferCapacity = 16)
val rangingBatches: SharedFlow<Pair<String, RangingBatch>> = _rangingBatches.asSharedFlow()
val fusedSamples: SharedFlow<Pair<String, FusedStateSample>> = _fusedSamples.asSharedFlow()
......
/** Inbound netcode claims (leader arbitrates) and verdicts (all devices apply). */
val eventClaims: SharedFlow<Pair<String, WireEventClaim>> = _eventClaims.asSharedFlow()
val eventVerdicts: SharedFlow<WireEventVerdict> = _eventVerdicts.asSharedFlow()
/** Completed clock-sync round-trips (t1..t4). Sole consumer: the mesh clock. */
val timeSyncResponses: SharedFlow<TimeSyncSample> = _timeSyncResponses.asSharedFlow()
val currentEpoch: Long get() = gate.localEpoch
fun advanceEpoch(epoch: Long) = gate.advanceEpoch(epoch)
......
link.sendFrame(MeshCodec.encode(envelope(MeshChannel.CONTROL, payload)))
}
/** Member side: start one clock-sync round-trip to the leader (stamps t1 now).
* The leader stamps t2/t3 and replies; the completed sample surfaces on
* [timeSyncResponses]. Drive this from the member's periodic probe loop. */
suspend fun probeTimeSync() = sendControl(TimeSync(t1Micros = clock.now().microseconds))
/** Control plane, leader side: reliable frame to one member. */
suspend fun sendControlTo(nodeId: String, payload: MeshPayload) {
links[nodeId]?.sendFrame(MeshCodec.encode(envelope(MeshChannel.CONTROL, payload)))
......
),
),
)
} else {
// Completed round-trip: t1/t2/t3 rode the response; t4 is now. Surface
// it for the mesh clock to fold into an offset/RTT estimate.
_timeSyncResponses.tryEmit(
TimeSyncSample(p.t1Micros, p.t2Micros, p.t3Micros, clock.now().microseconds),
)
}
// responses are consumed by the ClockOffsetEstimator
}
}
common/src/commonTest/kotlin/com/aether/mofe/messaging/ClockSyncSeamTest.kt
package com.aether.mofe.messaging
import com.aether.mofe.engine.netcode.MeshClock
import com.aether.mofe.messaging.support.SkewController
import com.aether.mofe.messaging.support.VirtualControlBus
import com.aether.mofe.messaging.support.VirtualEther
import com.aether.mofe.model.DeviceId
import com.aether.mofe.model.messaging.NodeHello
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
/**
* G2 clock seam. A member's `TimeSync` round-trip to the leader used to die on
* arrival ("consumed by the ClockOffsetEstimator" that never existed); now the
* completed four-stamp exchange surfaces on [MeshMessagingService.timeSyncResponses]
* and `MeshClock` recovers the leader−member offset from it.
*
* Per-node skewed clocks (via [SkewController]) make the recovered offset a real,
* non-zero value — proving the wiring, not a tautology.
*/
@OptIn(ExperimentalCoroutinesApi::class)
class ClockSyncSeamTest {
private fun hello(id: String, port: Int) = NodeHello(
nodeId = id, deviceType = "CLIENT", capabilities = emptyList(),
datagramHost = id, datagramPort = port, appliedRevision = 0L,
)
@Test
fun memberRecoversLeaderOffsetFromCompletedRoundTrip() = runTest {
val skew = SkewController(startMicros = 1_000_000)
val leaderClock = skew.clock(offsetMicros = +37_000) // leader 37ms ahead of the reference
val memberClock = skew.clock(offsetMicros = -12_000) // member 12ms behind
val bus = VirtualControlBus()
val ether = VirtualEther()
val leader = MeshMessagingService(DeviceId("L"), ether.endpoint("L", 8100), bus.transportFor("L"), leaderClock, backgroundScope)
val member = MeshMessagingService(DeviceId("m"), ether.endpoint("m", 8101), bus.transportFor("m"), memberClock, backgroundScope)
leader.startAsLeader(); runCurrent()
member.startAsMember(NodeAddress("L", 0), hello("m", 8101)); runCurrent()
val meshClock = MeshClock()
backgroundScope.launch {
member.timeSyncResponses.collect { meshClock.observe(it.t1Micros, it.t2Micros, it.t3Micros, it.t4Micros) }
}
runCurrent()
repeat(4) { member.probeTimeSync(); runCurrent() }
assertTrue(meshClock.synced, "the round-trip completed and fed the clock (it used to be dropped)")
assertEquals(49_000L, meshClock.offsetMicros, "offset = leader(+37ms) − member(−12ms) = 49ms")
assertEquals(
leaderClock.now().microseconds,
meshClock.toMeshTime(memberClock.now()).microseconds,
"member local + offset resolves to the leader's reading of the same instant",
)
}
}
(4-4/4)