User Story #42 » phase2-g1-netcode-assembly.patch
| common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshCoordinator.kt | ||
|---|---|---|
|
import com.aether.mofe.engine.MeshGroundTruth
|
||
|
import com.aether.mofe.engine.MultiObserverFusionEngine
|
||
|
import com.aether.mofe.engine.netcode.MeshClock
|
||
|
import com.aether.mofe.engine.netcode.NetcodeAdapter
|
||
|
import com.aether.mofe.engine.netcode.NetcodeSession
|
||
|
import com.aether.mofe.messaging.raft.RaftModule
|
||
|
import com.aether.mofe.messaging.raft.RaftPersistence
|
||
|
import com.aether.mofe.messaging.raft.RaftRole
|
||
| ... | ... | |
|
val meshClockSynced: Boolean get() = meshClock.synced
|
||
|
val meshOffsetMicros: Long get() = meshClock.offsetMicros
|
||
|
|
||
|
/** Netcode seam: per-device history + entity interpolation + leader-side
|
||
|
* arbitration. Shares the one [meshClock] so there is a single mesh time
|
||
|
* base — not a parallel clock. Fed below from the live fused-state plane,
|
||
|
* so a receiver's buffers fill with peers' mesh-time-stamped samples,
|
||
|
* ready for rewind and interpolated render. */
|
||
|
val netcode: NetcodeSession = NetcodeSession(clock = meshClock)
|
||
|
private val netcodeAdapter = NetcodeAdapter(netcode)
|
||
|
|
||
|
private val raft = RaftModule(
|
||
|
selfId, registry, messaging, persistence, clock, scope,
|
||
|
onRoleChange = { role, term -> onRaftRoleChange(role, term) },
|
||
| ... | ... | |
|
scope.launch {
|
||
|
registry.deltas.collect { delta -> applyLocalSideEffects(delta) }
|
||
|
}
|
||
|
// De-island the netcode: fold every peer's fused sample (stamped in mesh
|
||
|
// time by the sender — G3) into its history buffer. Runs for both roles;
|
||
|
// the leader also arbitrates against these buffers.
|
||
|
scope.launch {
|
||
|
messaging.fusedSamples.collect { netcodeAdapter.onFusedSample(it.second) }
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// ═══════════════════════════ VOTER (anchor) ═══════════════════════════
|
||
| common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeAssemblyTest.kt | ||
|---|---|---|
|
package com.aether.mofe.messaging
|
||
|
import com.aether.mofe.messaging.support.ClusterHarness
|
||
|
import com.aether.mofe.model.DeviceId
|
||
|
import com.aether.mofe.model.Timestamp
|
||
|
import com.aether.mofe.model.Vector3D
|
||
|
import com.aether.mofe.model.messaging.FusedStateSample
|
||
|
import com.aether.mofe.model.messaging.MeshChannel
|
||
|
import com.aether.mofe.model.messaging.MessageEnvelope
|
||
|
import com.aether.mofe.model.messaging.WireFusedSample
|
||
|
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||
|
import kotlinx.coroutines.test.runTest
|
||
|
import kotlin.test.Test
|
||
|
import kotlin.test.assertEquals
|
||
|
import kotlin.test.assertNotNull
|
||
|
/**
|
||
|
* G1 assembly. The netcode used to be dormant — nothing constructed or fed it.
|
||
|
* Now every MeshCoordinator owns a [com.aether.mofe.engine.netcode.NetcodeSession]
|
||
|
* (sharing the coordinator's one MeshClock, not a parallel clock) and folds the
|
||
|
* live fused-state plane into it. A peer's data-plane sample, stamped in mesh
|
||
|
* time (G3), reaches the session's history buffer end to end — the fixture every
|
||
|
* Area 01–06 scenario builds on.
|
||
|
*/
|
||
|
@OptIn(ExperimentalCoroutinesApi::class)
|
||
|
class NetcodeAssemblyTest {
|
||
|
@Test
|
||
|
fun coordinatorBuffersPeerFusedSamplesIntoTheNetcodeSession() = runTest {
|
||
|
val h = ClusterHarness(this)
|
||
|
h.bootstrapLeader("root", 9901)
|
||
|
h.pump(5)
|
||
|
val svc = h.services["root"]!!
|
||
|
// A peer's fused sample, already stamped in mesh time, arriving on the data plane.
|
||
|
val meshT = 50_000_000L
|
||
|
val env = MessageEnvelope(
|
||
|
messageId = "m1", sourceNodeId = "peerX", channel = MeshChannel.RANGING,
|
||
|
sequence = 1L, sentAtMicros = meshT, meshEpoch = svc.currentEpoch,
|
||
|
payload = FusedStateSample(listOf(WireFusedSample(
|
||
|
targetId = "peerX", timestampMicros = meshT,
|
||
|
position = Vector3D(3.0, 4.0, 0.0), velocity = Vector3D(1.0, 0.0, 0.0),
|
||
|
positionSigma = Vector3D(0.1, 0.1, 0.1),
|
||
|
))),
|
||
|
)
|
||
|
h.ether.endpoint("peerX", 5555).send(NodeAddress("root", 9901), MeshCodec.encode(env))
|
||
|
h.pump(5)
|
||
|
val pose = assertNotNull(
|
||
|
h.coords["root"]!!.netcode.poseAt(DeviceId("peerX"), Timestamp(meshT)),
|
||
|
"the assembled NetcodeSession buffered the peer's data-plane sample",
|
||
|
)
|
||
|
assertEquals(3.0, pose.position.x, 1e-9)
|
||
|
assertEquals(4.0, pose.position.y, 1e-9)
|
||
|
}
|
||
|
}
|
||
- « Previous
- 1
- 2
- Next »