From 29693e573952786184604b83c8ede8bfe9445b5b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 18:22:43 +0000 Subject: [PATCH 5/6] feat(mesh): NetcodeBringUp kill-switch (default OFF) to isolate netcode from the mesh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Netcode bring-up was wired unconditionally into MeshCoordinator.init — the fused-sample collector, the claim/verdict transport, and emitClaims all ran in the mesh scope (before quorum), with no way to disable them short of editing code. A bring-up defect could therefore regress ranging, the solve, or Raft. Add MessagingConfig.netcodeBringUp (default OFF). When OFF the NetcodeSession is still constructed (render/aim APIs stay non-null and return empty), but the live seam is inert: no peer samples folded in, transport never starts, emitClaims is a no-op. Flip ON via the MeshNode descriptor's config to exercise go-live on-device; flip back OFF to revert without a rebuild. - MessagingConfig: add netcodeBringUp flag + doc - MeshCoordinator: gate the two init collectors + transport.start and emitClaims; expose netcodeBringUpEnabled for host/diagnostics - ClusterHarness: netcodeBringUp ctor flag threaded into the coordinator config - Go-live tests (Assembly/ClaimEmission/Arbitration) opt into bring-up - New NetcodeBringUpGateTest: same peer sample OFF -> inert vs ON -> live; claims suppressed when OFF Verified offline: :common:jvmTest green, 751 tests, 0 failures. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WppuiKZt4CuQxX4N7k6SVR --- .../aether/mofe/messaging/MeshCoordinator.kt | 42 ++++++--- .../mofe/messaging/MeshMessagingService.kt | 11 +++ .../mofe/messaging/NetcodeArbitrationTest.kt | 4 +- .../mofe/messaging/NetcodeAssemblyTest.kt | 2 +- .../mofe/messaging/NetcodeBringUpGateTest.kt | 91 +++++++++++++++++++ .../messaging/NetcodeClaimEmissionTest.kt | 2 +- .../mofe/messaging/support/ClusterHarness.kt | 5 + 7 files changed, 138 insertions(+), 19 deletions(-) create mode 100644 common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeBringUpGateTest.kt diff --git a/common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshCoordinator.kt b/common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshCoordinator.kt index 224aee4..ccf03e1 100644 --- a/common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshCoordinator.kt +++ b/common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshCoordinator.kt @@ -103,6 +103,7 @@ class MeshCoordinator( * [EventNotifier]. Non-arbitrable events map to no claims. */ fun emitClaims(event: MeshEvent) { + if (!config.netcodeBringUp) return // netcode disabled → no claims on the control plane for (claim in MofeNetcodeBridge.claimsFor(event, selfId)) { claimTracker.track(claim) val wire = claim.toWire() @@ -133,24 +134,35 @@ class MeshCoordinator( scope.launch { registry.deltas.collect { delta -> applyLocalSideEffects(delta) } } - // Fold every peer's fused sample (stamped in mesh - // time by the sender) into its history buffer. Runs for both roles; - // the leader also arbitrates against these buffers. - scope.launch { - messaging.fusedSamples.collect { netcodeAdapter.onFusedSample(it.second) } + // Netcode bring-up (kill-switch, default OFF — see MessagingConfig.netcodeBringUp). + // Gated as one unit so a bring-up defect stays fully isolated from the mesh: + // when OFF, none of the netcode collectors/transport run in the mesh scope, so + // ranging, the solve, and Raft proceed exactly as they did before go-live. + if (config.netcodeBringUp) { + // Fold every peer's fused sample (stamped in mesh + // time by the sender) into its history buffer. Runs for both roles; + // the leader also arbitrates against these buffers. + scope.launch { + messaging.fusedSamples.collect { netcodeAdapter.onFusedSample(it.second) } + } + // G4b: the leader arbitrates each inbound claim against its rewound history and + // broadcasts the verdict; every device applies inbound verdicts to its predictions. + netcodeTransport.start( + scope = scope, + eventClaims = messaging.eventClaims, + eventVerdicts = messaging.eventVerdicts, + isLeader = { isLeader }, + nowMesh = { netcode.toMeshTime(clock.now()) }, + broadcastControl = { messaging.broadcastControl(it) }, + ) } - // G4b: the leader arbitrates each inbound claim against its rewound history and - // broadcasts the verdict; every device applies inbound verdicts to its predictions. - netcodeTransport.start( - scope = scope, - eventClaims = messaging.eventClaims, - eventVerdicts = messaging.eventVerdicts, - isLeader = { isLeader }, - nowMesh = { netcode.toMeshTime(clock.now()) }, - broadcastControl = { messaging.broadcastControl(it) }, - ) } + /** Whether the netcode seam is live on this node (see [MessagingConfig.netcodeBringUp]). + * Exposed so the host/diagnostics can surface the bring-up mode; when false the + * [netcode] session is constructed but inert. */ + val netcodeBringUpEnabled: Boolean get() = config.netcodeBringUp + // ═══════════════════════════ VOTER (anchor) ═══════════════════════════ /** diff --git a/common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshMessagingService.kt b/common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshMessagingService.kt index 1be2fee..7bed53f 100644 --- a/common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshMessagingService.kt +++ b/common/src/commonMain/kotlin/com/aether/mofe/messaging/MeshMessagingService.kt @@ -41,6 +41,17 @@ data class MessagingConfig( val rangingPublishHz: Int = 10, val fusedSamplePublishHz: Int = 10, val timeSyncIntervalMillis: Long = 5_000, + /** + * Netcode bring-up kill-switch. Default **OFF**: the coordinator still + * *constructs* the [com.aether.mofe.engine.netcode.NetcodeSession] (so the + * render / aim-tracking APIs the UI reads stay non-null and simply return + * empty), but it does NOT fold the live fused-state plane into it, start the + * claim/verdict transport, or emit claims onto the control plane. This keeps + * the netcode seam fully inert so a bring-up defect cannot regress ranging, + * the solve, or Raft. Flip ON (via the [MeshNode] descriptor's config) to + * exercise go-live on-device; flip back OFF to revert without a rebuild. + */ + val netcodeBringUp: Boolean = false, ) /** A peer this node fans data-plane traffic out to. Maintained from the registry. */ diff --git a/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeArbitrationTest.kt b/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeArbitrationTest.kt index c610d8b..45501f8 100644 --- a/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeArbitrationTest.kt +++ b/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeArbitrationTest.kt @@ -43,7 +43,7 @@ class NetcodeArbitrationTest { @Test fun memberClaimConfirmedByLeaderRewindArbitration() = runTest { - val h = ClusterHarness(this) + val h = ClusterHarness(this, netcodeBringUp = true) // exercising the netcode go-live path val leader = h.bootstrapLeader("root", 9921); runCurrent() val member = h.joinAsMember("client", 9922, leaderHost = "root") h.pump(40) // membership @@ -67,7 +67,7 @@ class NetcodeArbitrationTest { @Test fun leaderArbitratesItsOwnDetectedClaim() = runTest { - val h = ClusterHarness(this) + val h = ClusterHarness(this, netcodeBringUp = true) // exercising the netcode go-live path val leader = h.bootstrapLeader("root", 9931) h.pump(10) leader.registerPredicateMeshWide(zone()) diff --git a/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeAssemblyTest.kt b/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeAssemblyTest.kt index 254e6bf..a5658a3 100644 --- a/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeAssemblyTest.kt +++ b/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeAssemblyTest.kt @@ -27,7 +27,7 @@ class NetcodeAssemblyTest { @Test fun coordinatorBuffersPeerFusedSamplesIntoTheNetcodeSession() = runTest { - val h = ClusterHarness(this) + val h = ClusterHarness(this, netcodeBringUp = true) // exercising the netcode go-live path h.bootstrapLeader("root", 9901) h.pump(5) val svc = h.services["root"]!! diff --git a/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeBringUpGateTest.kt b/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeBringUpGateTest.kt new file mode 100644 index 0000000..64b814b --- /dev/null +++ b/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeBringUpGateTest.kt @@ -0,0 +1,91 @@ +package com.aether.mofe.messaging + +import com.aether.mofe.messaging.support.ClusterHarness +import com.aether.mofe.messaging.support.TestFactories +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 +import kotlin.test.assertNull + +/** + * The NetcodeBringUp kill-switch ([MessagingConfig.netcodeBringUp], default OFF). + * + * Safeguard for netcode go-live: with the switch OFF the coordinator still owns a + * [com.aether.mofe.engine.netcode.NetcodeSession], but the live seam is inert — no + * peer fused samples are folded in, the claim/verdict transport never starts, and + * [MeshCoordinator.emitClaims] puts nothing on the control plane. So a netcode + * bring-up defect cannot regress ranging, the solve, or Raft. Flipping it ON + * restores full go-live behaviour (covered end to end by NetcodeAssemblyTest / + * NetcodeClaimEmissionTest / NetcodeArbitrationTest). + * + * Both cases deliver the SAME peer sample under opposite flag states and assert + * opposite outcomes, so the gate — not some incidental wiring — is what makes + * netcode live. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class NetcodeBringUpGateTest { + + private val meshT = 50_000_000L + + /** A peer's mesh-time-stamped fused sample arriving on [leaderHost]'s data plane. */ + private suspend fun ClusterHarness.deliverPeerSample(leaderHost: String, leaderPort: Int) { + val svc = services[leaderHost]!! + 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), + ))), + ) + ether.endpoint("peerX", 5555).send(NodeAddress(leaderHost, leaderPort), MeshCodec.encode(env)) + } + + @Test + fun default_off_keeps_the_netcode_seam_inert() = runTest { + val h = ClusterHarness(this) // default: netcodeBringUp = false + val leader = h.bootstrapLeader("root", 9941) + h.pump(5) + h.deliverPeerSample("root", 9941) + h.pump(5) + + // The fused-sample collector never launched, so nothing reached the session. + assertNull( + h.coords["root"]!!.netcode.poseAt(DeviceId("peerX"), Timestamp(meshT)), + "with bring-up OFF the peer's sample is not folded into the netcode session", + ) + assertEquals(false, leader.netcodeBringUpEnabled) + + // And emitClaims puts nothing on the control plane / tracks nothing. + leader.emitClaims(TestFactories.regionEntry("zone-1", "target-7", meshT)) + h.pump(5) + assertEquals(0, leader.claimTracker.pendingCount(), "with bring-up OFF no claim is emitted or tracked") + } + + @Test + fun bring_up_on_makes_the_same_sample_live() = runTest { + val h = ClusterHarness(this, netcodeBringUp = true) + val leader = h.bootstrapLeader("root", 9942) + h.pump(5) + h.deliverPeerSample("root", 9942) + h.pump(5) + + val pose = assertNotNull( + h.coords["root"]!!.netcode.poseAt(DeviceId("peerX"), Timestamp(meshT)), + "with bring-up ON the same peer sample is folded into the session", + ) + assertEquals(3.0, pose.position.x, 1e-9) + assertEquals(4.0, pose.position.y, 1e-9) + assertEquals(true, leader.netcodeBringUpEnabled) + } +} diff --git a/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeClaimEmissionTest.kt b/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeClaimEmissionTest.kt index 59506ff..d9db4e5 100644 --- a/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeClaimEmissionTest.kt +++ b/common/src/commonTest/kotlin/com/aether/mofe/messaging/NetcodeClaimEmissionTest.kt @@ -22,7 +22,7 @@ class NetcodeClaimEmissionTest { @Test fun memberEmitsClaimToLeaderAndTracksItPending() = runTest { - val h = ClusterHarness(this) + val h = ClusterHarness(this, netcodeBringUp = true) // exercising the netcode go-live path h.bootstrapLeader("root", 9911); runCurrent() val member = h.joinAsMember("client", 9912, leaderHost = "root") h.pump(40) // membership handshake diff --git a/common/src/commonTest/kotlin/com/aether/mofe/messaging/support/ClusterHarness.kt b/common/src/commonTest/kotlin/com/aether/mofe/messaging/support/ClusterHarness.kt index c5d6877..107fed9 100644 --- a/common/src/commonTest/kotlin/com/aether/mofe/messaging/support/ClusterHarness.kt +++ b/common/src/commonTest/kotlin/com/aether/mofe/messaging/support/ClusterHarness.kt @@ -7,6 +7,7 @@ import com.aether.mofe.engine.MultilaterationSolver import com.aether.mofe.messaging.MeshCoordinator import com.aether.mofe.messaging.MeshMessagingService import com.aether.mofe.messaging.MeshStateRegistry +import com.aether.mofe.messaging.MessagingConfig import com.aether.mofe.messaging.NodeAddress import com.aether.mofe.messaging.raft.InMemoryRaftPersistence import com.aether.mofe.messaging.security.GroupKeyManager @@ -40,6 +41,9 @@ import kotlinx.coroutines.test.runCurrent class ClusterHarness( private val test: TestScope, startMicros: Long = 1_000L, + /** Enable the netcode seam on every node built by this harness (default OFF, + * matching production). Go-live tests opt in; mesh/Raft scenarios leave it off. */ + private val netcodeBringUp: Boolean = false, ) { val clock = VirtualClock(startMicros) val bus = VirtualControlBus() @@ -63,6 +67,7 @@ class ClusterHarness( val coord = MeshCoordinator( selfId = DeviceId(id), engine = engine(), messaging = messaging, registry = registry, clock = clock, scope = nodeScope, persistence = InMemoryRaftPersistence(), groupKeys = groupKeys, + config = MessagingConfig(netcodeBringUp = netcodeBringUp), ) coords[id] = coord; registries[id] = registry; services[id] = messaging scopes[id] = nodeScope; ports[id] = port -- 2.43.0