Merge remote-tracking branch 'origin/main'
Some checks are pending
CI Benchmark - Fusion Loop Timing / Fusion Loop Timing Benchmark (push) Waiting to run

This commit is contained in:
jedarden 2026-07-03 20:29:43 -04:00
commit 2eab19b6e0
16 changed files with 852 additions and 104 deletions

View file

@ -342,7 +342,9 @@ func connectNodes(ctx context.Context, nodes []*VirtualNode, token string, rng *
log.Printf("[SIM] Node %d: connected to mothership", n.ID)
// Send hello message
// Send hello message. Announce the node's computed position
// (createVirtualNodes corner geometry) so the mothership persists it
// in the fleet/DB row instead of leaving it at the schema default (bf-24xp).
hello := map[string]interface{}{
"type": "hello",
"mac": macToString(n.MAC),
@ -351,6 +353,9 @@ func connectNodes(ctx context.Context, nodes []*VirtualNode, token string, rng *
"chip": "ESP32-S3",
"flash_mb": 16,
"uptime_ms": 1000,
"pos_x": n.Position.X,
"pos_y": n.Position.Y,
"pos_z": n.Position.Z,
}
if err := conn.WriteJSON(hello); err != nil {
log.Printf("[SIM] Node %d: failed to send hello: %v", n.ID, err)

View file

@ -351,21 +351,21 @@ func (a *fleetRoomConfigAdapter) GetRoom() (width, height, depth float64) {
// multiFleetNotifier fans out ingestion.FleetNotifier events to multiple fleet components.
type multiFleetNotifier struct {
notifiers []interface {
OnNodeConnected(mac, firmware, chip string)
OnNodeConnected(mac, firmware, chip string, posX, posY, posZ *float64)
OnNodeDisconnected(mac string)
}
}
func newMultiNotifier(notifiers ...interface {
OnNodeConnected(mac, firmware, chip string)
OnNodeConnected(mac, firmware, chip string, posX, posY, posZ *float64)
OnNodeDisconnected(mac string)
}) *multiFleetNotifier {
return &multiFleetNotifier{notifiers: notifiers}
}
func (m *multiFleetNotifier) OnNodeConnected(mac, firmware, chip string) {
func (m *multiFleetNotifier) OnNodeConnected(mac, firmware, chip string, posX, posY, posZ *float64) {
for _, n := range m.notifiers {
n.OnNodeConnected(mac, firmware, chip)
n.OnNodeConnected(mac, firmware, chip, posX, posY, posZ)
}
}

View file

@ -646,7 +646,9 @@ func connectNodes(ctx context.Context, nodes []*VirtualNode) error {
node.Conn = conn
log.Printf("[SIM] Node %d connected", node.ID)
// Send hello message
// Send hello message. Announce the node's computed position
// (createVirtualNodes perimeter geometry) so the mothership persists it
// in the fleet/DB row instead of leaving it at the schema default (bf-24xp).
hello := map[string]interface{}{
"type": "hello",
"mac": macToString(node.MAC),
@ -657,6 +659,9 @@ func connectNodes(ctx context.Context, nodes []*VirtualNode) error {
"uptime_ms": 1000,
"wifi_rssi": -45,
"ip": fmt.Sprintf("127.0.0.%d", node.ID+2),
"pos_x": node.Position.X,
"pos_y": node.Position.Y,
"pos_z": node.Position.Z,
}
helloBytes, err := json.Marshal(hello)

View file

@ -188,7 +188,7 @@ func TestRegistryGetAllNodes(t *testing.T) {
func TestManagerSingleNode_TxRx(t *testing.T) {
mgr, notif, _ := newTestManager(t)
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
role := notif.sentRole("aa:00:00:00:00:01")
if role != "tx_rx" {
@ -207,8 +207,8 @@ func TestManagerSingleNode_TxRx(t *testing.T) {
func TestManagerTwoNodes_TxRx(t *testing.T) {
mgr, notif, _ := newTestManager(t)
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
mgr.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
r1 := notif.sentRole("aa:00:00:00:00:01")
r2 := notif.sentRole("aa:00:00:00:00:02")
@ -226,9 +226,9 @@ func TestManagerTwoNodes_TxRx(t *testing.T) {
func TestManagerThreeNodes_HalfTx(t *testing.T) {
mgr, notif, _ := newTestManager(t)
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
mgr.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
mgr.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
roles := []string{
notif.sentRole("aa:00:00:00:00:01"),
@ -253,9 +253,9 @@ func TestManagerThreeNodes_HalfTx(t *testing.T) {
func TestManagerNodeDisconnect_Rebalance(t *testing.T) {
mgr, notif, _ := newTestManager(t)
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
mgr.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
mgr.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
// Node 2 goes offline.
mgr.OnNodeDisconnected("aa:00:00:00:00:02")
@ -283,7 +283,7 @@ func TestManagerNodeDisconnect_Rebalance(t *testing.T) {
func TestManagerLastNodeDisconnect_ClearsState(t *testing.T) {
mgr, notif, _ := newTestManager(t)
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
mgr.OnNodeDisconnected("aa:00:00:00:00:01")
mgr.mu.RLock()
@ -299,7 +299,7 @@ func TestManagerLastNodeDisconnect_ClearsState(t *testing.T) {
func TestManagerSelfHeal_RepushesRoles(t *testing.T) {
mgr, notif, _ := newTestManager(t)
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
// Simulate notifier tracking connected nodes.
notif.mu.Lock()
@ -322,7 +322,7 @@ func TestManagerSelfHeal_RepushesRoles(t *testing.T) {
func TestManagerOverrideRole(t *testing.T) {
mgr, notif, bcaster := newTestManager(t)
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
prevCalls := bcaster.broadcastCount()
if err := mgr.OverrideRole("aa:00:00:00:00:01", "rx"); err != nil {
@ -350,7 +350,7 @@ func TestManagerBroadcastOnConnect(t *testing.T) {
mgr, _, bcaster := newTestManager(t)
before := bcaster.broadcastCount()
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
after := bcaster.broadcastCount()
if after <= before {
@ -361,7 +361,7 @@ func TestManagerBroadcastOnConnect(t *testing.T) {
func TestManagerBroadcastOnDisconnect(t *testing.T) {
mgr, _, bcaster := newTestManager(t)
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
mgr.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
before := bcaster.broadcastCount()
mgr.OnNodeDisconnected("aa:00:00:00:00:01")
after := bcaster.broadcastCount()
@ -380,7 +380,7 @@ func TestManagerPersistenceAcrossRestart(t *testing.T) {
mgr1 := NewManager(reg)
n1 := newMockNotifier()
mgr1.SetNotifier(n1)
mgr1.OnNodeConnected("aa:00:00:00:00:01", "v1.2", "ESP32-S3")
mgr1.OnNodeConnected("aa:00:00:00:00:01", "v1.2", "ESP32-S3", nil, nil, nil)
// Second manager with same registry simulates a restart.
mgr2 := NewManager(reg)

View file

@ -1517,7 +1517,7 @@ func TestManagerOnNodeConnectedForwardsPositionToEngine(t *testing.T) {
engine.SetNodePosition(m, x, y, z)
})
mgr.OnNodeConnected(mac, "1.0.0", "ESP32-S3")
mgr.OnNodeConnected(mac, "1.0.0", "ESP32-S3", nil, nil, nil)
p := engineNodePosition(engine, mac)
if p == nil {
@ -1528,6 +1528,105 @@ func TestManagerOnNodeConnectedForwardsPositionToEngine(t *testing.T) {
}
}
// TestManagerOnNodeConnectedPersistsHelloPosition (bf-24xp) verifies the
// acceptance criterion for hello-announced node positions: when a node
// connects announcing its 3D position (spaxel-sim does this with its computed
// corner/perimeter geometry), the manager persists it to the registry so the
// fleet/DB row is NOT left at the schema default (0,0,1) and matches the
// announced coordinates. The announced position must also reach the fusion
// engine through the bf-3p6g connect/register sink.
func TestManagerOnNodeConnectedPersistsHelloPosition(t *testing.T) {
f64 := func(v float64) *float64 { return &v }
tests := []struct {
name string
posX, posY, posZ float64
}{
{name: "corner_geometry", posX: 6.0, posY: 0.0, posZ: 2.5},
{name: "interior_point", posX: 3.25, posY: 2.75, posZ: 1.1},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
const mac = "02:53:AC:00:00:01" // matches spaxel-sim's MAC OUI
reg := newTestRegistry(t)
engine := fusion.NewEngine(nil)
mgr := NewManager(reg)
mgr.SetNodePositionSink(func(m string, x, y, z float64) {
engine.SetNodePosition(m, x, y, z)
})
// Connect announcing the position, exactly as spaxel-sim does in hello.
mgr.OnNodeConnected(mac, "sim-1.0.0", "ESP32-S3", f64(tc.posX), f64(tc.posY), f64(tc.posZ))
// Registry row must hold the announced position, not the (0,0,1) default.
rx, ry, rz, ok := reg.GetNodePosition(mac)
if !ok {
t.Fatal("node row missing after connect")
}
if rx == 0 && ry == 0 && rz == 1 {
t.Errorf("registry position still at schema default (0,0,1); got (%v,%v,%v)", rx, ry, rz)
}
if rx != tc.posX || ry != tc.posY || rz != tc.posZ {
t.Errorf("registry position = (%v,%v,%v), want (%v,%v,%v)", rx, ry, rz, tc.posX, tc.posY, tc.posZ)
}
// The same coordinates must have reached the fusion engine.
p := engineNodePosition(engine, mac)
if p == nil {
t.Fatal("announced position not forwarded to fusion engine")
}
if p.X != tc.posX || p.Y != tc.posY || p.Z != tc.posZ {
t.Errorf("engine position = (%v,%v,%v), want (%v,%v,%v)", p.X, p.Y, p.Z, tc.posX, tc.posY, tc.posZ)
}
})
}
}
// TestManagerOnNodeConnectedWithoutHelloPositionPreservesExisting (bf-24xp)
// verifies the real-ESP32 case: a node that does NOT announce a position (nil
// on all three axes — a real node's position is user-placed in the dashboard)
// leaves any existing registry position untouched, so a connect never clobbers
// a user-placed position with the schema default. A partial announcement
// (only some axes present) is likewise ignored.
func TestManagerOnNodeConnectedWithoutHelloPositionPreservesExisting(t *testing.T) {
const mac = "AA:BB:CC:DD:EE:FF"
reg := newTestRegistry(t)
reg.UpsertNode(mac, "1.0.0", "ESP32-S3")
reg.SetNodePosition(mac, 4.0, 5.0, 6.0) // user-placed while offline
engine := fusion.NewEngine(nil)
mgr := NewManager(reg)
mgr.SetNodePositionSink(func(m string, x, y, z float64) {
engine.SetNodePosition(m, x, y, z)
})
// Connect with no announced position; the existing (4,5,6) must survive.
mgr.OnNodeConnected(mac, "1.0.0", "ESP32-S3", nil, nil, nil)
rx, ry, rz, ok := reg.GetNodePosition(mac)
if !ok {
t.Fatal("node row missing after connect")
}
if rx != 4.0 || ry != 5.0 || rz != 6.0 {
t.Errorf("existing position overwritten by nil announce; got (%v,%v,%v), want (4,5,6)", rx, ry, rz)
}
// A partial announcement (only X present) must also be ignored.
onlyX := 9.0
reg.SetNodePosition(mac, 4.0, 5.0, 6.0) // reset
mgr.OnNodeConnected(mac, "1.0.0", "ESP32-S3", &onlyX, nil, nil)
rx, ry, rz, ok = reg.GetNodePosition(mac)
if !ok {
t.Fatal("node row missing after partial-announce connect")
}
if rx != 4.0 || ry != 5.0 || rz != 6.0 {
t.Errorf("existing position overwritten by partial announce; got (%v,%v,%v), want (4,5,6)", rx, ry, rz)
}
}
// ─── Fleet page specific tests ─────────────────────────────────────────────────────
// TestFleetTableRendering verifies that the fleet table renders correctly with 4 nodes

View file

@ -115,8 +115,13 @@ func (fh *FleetHealer) SetBroadcaster(b RegistryBroadcaster) {
fh.mu.Unlock()
}
// OnNodeConnected handles a node connection event
func (fh *FleetHealer) OnNodeConnected(mac, firmware, chip string) {
// OnNodeConnected handles a node connection event. posX/posY/posZ are the
// hello-announced position (bf-24xp); the healer reads geometry from the
// registry, so they are ignored here.
func (fh *FleetHealer) OnNodeConnected(mac, firmware, chip string, posX, posY, posZ *float64) {
_ = posX
_ = posY
_ = posZ
fh.mu.Lock()
fh.online[mac] = struct{}{}
wasDegraded := fh.degradedMode

View file

@ -83,7 +83,7 @@ func TestFleetHealer_SingleNode(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{MinOnlineNodes: 2})
fh.SetNotifier(newMockNotifier())
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
roles := fh.GetOptimalRoles()
if roles["aa:00:00:00:00:01"] != "tx_rx" {
@ -103,8 +103,8 @@ func TestFleetHealer_TwoNodes(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{MinOnlineNodes: 2})
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
roles := fh.GetOptimalRoles()
if len(roles) != 2 {
@ -139,9 +139,9 @@ func TestFleetHealer_ThreeNodes_OptimalRoles(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{MinOnlineNodes: 2})
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
roles := fh.GetOptimalRoles()
if len(roles) != 3 {
@ -168,9 +168,9 @@ func TestFleetHealer_NodeDisconnect(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{MinOnlineNodes: 2})
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
if len(fh.GetOnlineNodes()) != 3 {
t.Fatalf("expected 3 online nodes, got %d", len(fh.GetOnlineNodes()))
@ -200,10 +200,10 @@ func TestFleetHealer_DegradedMode(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{MinOnlineNodes: 4})
// Connect all 4 - should not be degraded
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3", nil, nil, nil)
if fh.IsDegraded() {
t.Error("should not be degraded with 4/4 nodes")
@ -225,7 +225,7 @@ func TestFleetHealer_Coverage(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{})
fh.SetGDOPCalculator(newMockGDOPCalculator(1.5, 20, 20))
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.UpdateNodePosition("aa:00:00:00:00:01", 2.0, 2.0)
coverage := fh.GetCoverage()
@ -251,7 +251,7 @@ func TestFleetHealer_CoverageHistory(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{MaxHistorySize: 5})
fh.SetGDOPCalculator(newMockGDOPCalculator(1.5, 20, 20))
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
// Trigger multiple coverage computations
for i := 0; i < 10; i++ {
@ -278,10 +278,10 @@ func TestFleetHealer_GDOPBasedOptimization(t *testing.T) {
mockCalc := newMockGDOPCalculator(2.0, 20, 20)
fh.SetGDOPCalculator(mockCalc)
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3", nil, nil, nil)
fh.UpdateNodePosition("aa:00:00:00:00:01", 0.0, 0.0)
fh.UpdateNodePosition("aa:00:00:00:00:02", 4.0, 0.0)
@ -310,8 +310,8 @@ func TestFleetHealer_WorstCoverageZone(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{})
fh.SetGDOPCalculator(newMockGDOPCalculator(3.0, 20, 20))
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
fh.UpdateNodePosition("aa:00:00:00:00:01", 1.0, 2.0)
fh.UpdateNodePosition("aa:00:00:00:00:02", 3.0, 2.0)
@ -336,7 +336,7 @@ func TestFleetHealer_SuggestNodePosition(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{})
fh.SetGDOPCalculator(newMockGDOPCalculator(3.0, 20, 20))
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.UpdateNodePosition("aa:00:00:00:00:01", 0.5, 0.5)
// Should suggest a position away from the existing node
@ -354,8 +354,8 @@ func TestFleetHealer_NoGDOPCalculator(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{})
// Without GDOP calculator, should fall back to simple assignment
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
roles := fh.GetOptimalRoles()
if len(roles) != 2 {
@ -439,10 +439,10 @@ func TestFleetHealer_RecoveryFromDegraded(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{MinOnlineNodes: 4})
// Connect all 4
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
fh.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3", nil, nil, nil)
if fh.IsDegraded() {
t.Fatal("should not start degraded")
@ -455,7 +455,7 @@ func TestFleetHealer_RecoveryFromDegraded(t *testing.T) {
}
// Reconnect - recovered
fh.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3", nil, nil, nil)
if fh.IsDegraded() {
t.Error("should recover after reconnect")
}
@ -466,7 +466,7 @@ func TestFleetHealer_ComputeCoverage_NoGDOPCalculator(t *testing.T) {
reg.UpsertNode("aa:00:00:00:00:01", "v1", "S3")
fh := NewFleetHealer(reg, FleetHealerConfig{})
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
coverage := fh.computeCoverage()
if coverage == nil {
@ -485,7 +485,7 @@ func TestFleetHealer_CoverageScore(t *testing.T) {
fh := NewFleetHealer(reg, FleetHealerConfig{})
fh.SetGDOPCalculator(newMockGDOPCalculator(1.5, 20, 20))
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
fh.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
fh.UpdateNodePosition("aa:00:00:00:00:01", 2.0, 2.0)
coverage := fh.computeCoverage()

View file

@ -181,11 +181,30 @@ func (m *Manager) ForwardNodePosition(mac string, x, y, z float64) {
// OnNodeConnected is called when a node completes its hello handshake.
// It persists the node, assigns a role, and broadcasts updated state.
func (m *Manager) OnNodeConnected(mac, firmware, chip string) {
//
// posX/posY/posZ carry the node's hello-announced 3D position (nil on all
// three when the node did not announce one). When present they are persisted
// to the registry so the fleet/DB row matches the announced location instead
// of the schema default — e.g. a spaxel-sim node announcing its corner
// geometry (bf-24xp). When absent (a real ESP32, whose position the user sets
// in the dashboard) any existing position is preserved. The position is then
// forwarded to the fusion engine below via the existing GetNodePosition +
// nodePositionSink path wired in bf-3p6g.
func (m *Manager) OnNodeConnected(mac, firmware, chip string, posX, posY, posZ *float64) {
if err := m.registry.UpsertNode(mac, firmware, chip); err != nil {
log.Printf("[WARN] fleet: upsert node %s: %v", mac, err)
}
// bf-24xp: a node may announce its 3D position in the hello handshake.
// Persist it now that the row exists (UpsertNode above), so the
// fleet/DB row is not left at the schema default. Only write when all
// three axes are present — a partially-announced position is ignored.
if posX != nil && posY != nil && posZ != nil {
if err := m.registry.SetNodePosition(mac, *posX, *posY, *posZ); err != nil {
log.Printf("[WARN] fleet: set hello position %s: %v", mac, err)
}
}
m.mu.Lock()
m.online[mac] = struct{}{}
m.mu.Unlock()

View file

@ -128,8 +128,13 @@ func (shm *SelfHealManager) SetGDOPCalculator(calc GDOPCalculator) {
shm.mu.Unlock()
}
// OnNodeConnected handles a node connection event
func (shm *SelfHealManager) OnNodeConnected(mac, firmware, chip string) {
// OnNodeConnected handles a node connection event. posX/posY/posZ are the
// hello-announced position (bf-24xp); the self-healer persists the node row
// itself and ignores the announced position (the fleet Manager owns that).
func (shm *SelfHealManager) OnNodeConnected(mac, firmware, chip string, posX, posY, posZ *float64) {
_ = posX
_ = posY
_ = posZ
now := time.Now()
shm.mu.Lock()

View file

@ -34,7 +34,7 @@ func TestSelfHealManager_SingleNode(t *testing.T) {
shm := NewSelfHealManager(reg, optimiser, cfg)
shm.SetNotifier(newMockNotifier())
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
roles := shm.GetCurrentRoles()
if roles["aa:00:00:00:00:01"] != RoleTXRX {
@ -52,8 +52,8 @@ func TestSelfHealManager_TwoNodes(t *testing.T) {
cfg := DefaultSelfHealConfig()
shm := NewSelfHealManager(reg, optimiser, cfg)
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
roles := shm.GetCurrentRoles()
if len(roles) != 2 {
@ -89,9 +89,9 @@ func TestSelfHealManager_ReconnectWithinGracePeriod(t *testing.T) {
shm.SetNotifier(notifier)
// Connect 3 nodes
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
// Record initial role
rolesBefore := shm.GetCurrentRoles()
@ -110,7 +110,7 @@ func TestSelfHealManager_ReconnectWithinGracePeriod(t *testing.T) {
time.Sleep(100 * time.Millisecond)
// Reconnect within grace period
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
// Verify role was restored
rolesAfter := shm.GetCurrentRoles()
@ -140,9 +140,9 @@ func TestSelfHealManager_ReconnectAfterGracePeriod(t *testing.T) {
shm.SetNotifier(notifier)
// Connect 3 nodes
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
// Disconnect one node
shm.OnNodeDisconnected("aa:00:00:00:00:02")
@ -151,7 +151,7 @@ func TestSelfHealManager_ReconnectAfterGracePeriod(t *testing.T) {
time.Sleep(150 * time.Millisecond)
// Reconnect after grace period - should trigger re-optimisation, not restore
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
// Verify all 3 nodes have valid roles
roles := shm.GetCurrentRoles()
@ -179,7 +179,7 @@ func TestSelfHealManager_GracePeriodExpiration(t *testing.T) {
shm.SetNotifier(newMockNotifier())
// Connect a node
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
// Disconnect it
shm.OnNodeDisconnected("aa:00:00:00:00:01")
@ -237,10 +237,10 @@ func TestSelfHealManager_GDOPComparison(t *testing.T) {
shm.SetGDOPCalculator(mockGDOP)
// Connect all 4
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:03", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:04", "v1", "S3", nil, nil, nil)
// Record initial coverage
coverageBefore := shm.GetCoverageScore()
@ -295,8 +295,8 @@ func TestSelfHealManager_FleetChangeEventContainsGDOP(t *testing.T) {
reg.SetNodePosition("aa:00:00:00:00:02", 3, 0, 3)
bcaster.reset()
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3")
shm.OnNodeConnected("aa:00:00:00:00:01", "v1", "S3", nil, nil, nil)
shm.OnNodeConnected("aa:00:00:00:00:02", "v1", "S3", nil, nil, nil)
// Disconnect to trigger event with GDOP data
bcaster.reset()

View file

@ -1,9 +1,12 @@
package fusion
import (
"fmt"
"math"
"testing"
"time"
"github.com/spaxel/mothership/internal/simulator"
)
// ---- Grid3D unit tests ----
@ -381,6 +384,146 @@ func TestEngine_SeedNodePositions(t *testing.T) {
}
}
// TestEngine_DefaultPlacementProducesPeaks is the bf-18yn acceptance test and
// closes the bf-4q5w symptom (the 3D fusion engine emitting no / degenerate
// peaks). It seeds the engine using ONLY the default node placement —
// simulator.DefaultNodePositions, the spread geometry a freshly-onboarded
// virtual/sim fleet receives with no manual positioning (bf-3fr6, bf-xrej) —
// then drives a synthetic walker through the room centre and asserts the
// accumulation grid produces non-zero fusion peaks (len(blobs) > 0).
//
// This is the fleet->engine counterpart to TestEngine_SeedNodePositions
// (bf-6s3d): that test locks in the seeding invariant (distinct, non-(0,0,1)
// positions); this one locks in the downstream consequence the seeding exists
// to deliver — that spread nodes actually let Fuse form blobs.
func TestEngine_DefaultPlacementProducesPeaks(t *testing.T) {
space := simulator.DefaultSpace()
minX, minY, minZ, maxX, maxY, maxZ := space.Bounds()
// 2+ nodes are required to form any link; count == 1 has no links and
// cannot produce peaks (that is a valid empty state, not a regression).
for _, count := range []int{2, 3, 4, 6} {
t.Run(fmt.Sprintf("nodes=%d", count), func(t *testing.T) {
// DEFAULT placement — no manual positioning. Distinct, room-spanning
// points distributed across the space's bounding box.
pts := simulator.DefaultNodePositions(space, count)
// Regression guard (bf-4q5w): if default placement ever collapses
// back to the co-located DB origin, every link goes degenerate and
// Fuse can only emit zero peaks. Fail loudly with a clear reason
// rather than letting the regression pass silently.
assertPlacementNotCollapsed(t, pts)
// Seed the engine exactly as main.go does at startup: one
// SetNodePosition per node, reading the (default) registry
// positions. The grid is sized to the same bounding box so every
// default-placed node lands in-bounds.
e := NewEngine(&Config{
Width: maxX - minX,
Height: maxY - minY,
Depth: maxZ - minZ,
OriginX: minX,
OriginY: minY,
OriginZ: minZ,
CellSize: 0.2,
MinDeltaRMS: 0.01,
MaxBlobs: 6,
BlobThreshold: 0.1,
})
nodes := make([]NodePosition, len(pts))
for i, p := range pts {
mac := fmt.Sprintf("SN:%02d", i+1)
nodes[i] = NodePosition{MAC: mac, X: p.X, Y: p.Y, Z: p.Z}
e.SetNodePosition(mac, p.X, p.Y, p.Z) // mirrors main.go seeding loop
}
// A walker at the room centre perturbs every link whose first
// Fresnel zone crosses it — the same synthetic-motion model the
// position-accuracy tests use (buildSyntheticLinks).
links := buildSyntheticLinks(nodes,
(minX+maxX)/2, (minY+maxY)/2, (minZ+maxZ)/2)
if len(links) == 0 {
t.Fatalf("expected at least one motion link crossing the room centre, got 0")
}
r := e.Fuse(links)
// Acceptance criterion (bf-18yn / bf-4q5w): the default placement
// must yield non-zero fusion peaks — either extracted blobs
// (len > 0) OR an accumulation grid whose max rises above the peak
// threshold. The OR covers small fleets whose links paint a flat
// ridge that the strict local-maximum extractor may not promote to
// a blob, even though the grid is plainly non-zero. Co-located
// (0,0,1) nodes paint nothing, so their grid max stays at 0 — the
// condition is non-trivial (see TestEngine_CoLocatedOriginYieldsNoPeaks).
gridMax := gridMaxValue(e.GetGridSnapshot().Data)
if len(r.Blobs) == 0 && gridMax <= e.blobThresh {
t.Fatalf("default placement of %d nodes produced no peaks: "+
"0 blobs and gridMax=%.4f <= threshold %.4f (activeLinks=%d) — "+
"bf-4q5w regression: spread nodes must let the grid accumulate",
count, gridMax, e.blobThresh, r.ActiveLinks)
}
})
}
}
// TestEngine_CoLocatedOriginYieldsNoPeaks is the counter-example that pins the
// bf-4q5w symptom: nodes left collapsed at the (0,0,1) DB schema default are
// co-located, so every link is degenerate (length < 0.1 m), the accumulation
// grid stays at zero, and Fuse emits no blobs. This is exactly the failure the
// default placement (tested above) exists to prevent — a fleet seeded this way
// could never localize. It documents why the non-zero-peak assertion is
// meaningful rather than trivially satisfiable.
func TestEngine_CoLocatedOriginYieldsNoPeaks(t *testing.T) {
e := NewEngine(&Config{
Width: 6, Height: 5, Depth: 2.5,
CellSize: 0.2, MinDeltaRMS: 0.01, MaxBlobs: 6, BlobThreshold: 0.1,
})
// Four nodes all at the DB default — the pre-spread state.
for i := 0; i < 4; i++ {
e.SetNodePosition(fmt.Sprintf("CN:%02d", i+1), 0, 0, 1)
}
links := []LinkMotion{
{NodeMAC: "CN:01", PeerMAC: "CN:02", DeltaRMS: 1.0, Motion: true},
{NodeMAC: "CN:03", PeerMAC: "CN:04", DeltaRMS: 1.0, Motion: true},
}
r := e.Fuse(links)
if len(r.Blobs) != 0 {
t.Fatalf("co-located (0,0,1) nodes must produce 0 blobs, got %d — "+
"degenerate links should leave the grid at zero", len(r.Blobs))
}
}
// assertPlacementNotCollapsed fails the test if any node sits at the co-located
// (0,0,1) DB default or if any two nodes share a position — i.e. the placement
// has collapsed instead of spreading across the room (bf-4q5w root cause).
func assertPlacementNotCollapsed(t *testing.T, pts []simulator.Point) {
t.Helper()
seen := make(map[simulator.Point]bool, len(pts))
for i, p := range pts {
if p.X == 0 && p.Y == 0 && p.Z == 1 {
t.Fatalf("default-placed node %d collapsed to DB origin (0,0,1) — "+
"bf-4q5w regression: positions must be spread, not co-located", i)
}
if seen[p] {
t.Fatalf("default-placed nodes co-located at %v — bf-4q5w regression: "+
"positions must be distinct", p)
}
seen[p] = true
}
}
// gridMaxValue returns the maximum voxel value in a flat grid snapshot.
func gridMaxValue(data []float64) float64 {
max := 0.0
for _, v := range data {
if v > max {
max = v
}
}
return max
}
// TestEngine_HealthWeight verifies that links with lower health scores contribute less to fusion.
// Per spec: "each link's contribution to the 3D occupancy grid is multiplied by its health_score"
func TestEngine_HealthWeight(t *testing.T) {

View file

@ -28,8 +28,13 @@ func NewFleetNotifier(mgr *Manager, getNodeLastSeen func(mac string) time.Time)
}
}
// OnNodeConnected is called when a node connects.
func (n *FleetNotifier) OnNodeConnected(mac, firmware, chip string) {
// OnNodeConnected is called when a node connects. posX/posY/posZ are the
// hello-announced position (bf-24xp); guided troubleshooting only tracks
// connect/disconnect, so they are ignored.
func (n *FleetNotifier) OnNodeConnected(mac, firmware, chip string, posX, posY, posZ *float64) {
_ = posX
_ = posY
_ = posZ
n.mu.Lock()
defer n.mu.Unlock()

View file

@ -20,6 +20,17 @@ type HelloMessage struct {
APBSSID string `json:"ap_bssid,omitempty"`
APChannel int `json:"ap_channel,omitempty"`
Token string `json:"token,omitempty"`
// PosX/PosY/PosZ carry the node's announced 3D world position, in meters.
// Pointers (not plain float64) so an *absent* position is distinguishable
// from a genuine (0,0,0): a real ESP32 omits these in hello (the user places
// the node in the dashboard), whereas spaxel-sim announces its computed
// corner geometry so the mothership persists it instead of leaving the
// fleet/DB row at the schema default (bf-24xp). nil on all three axes =
// "not announced"; the mothership then preserves any existing position.
PosX *float64 `json:"pos_x,omitempty"`
PosY *float64 `json:"pos_y,omitempty"`
PosZ *float64 `json:"pos_z,omitempty"`
}
// HealthMessage is sent every 10 seconds

View file

@ -24,8 +24,15 @@ type CSIBroadcaster interface {
}
// FleetNotifier receives node lifecycle events for fleet management.
//
// OnNodeConnected's trailing posX/posY/posZ carry the node's hello-announced
// 3D position (nil on all three when the node did not announce one — e.g. a
// real ESP32; the mothership then preserves any user-placed position). They
// flow to the fleet Manager, which persists them via the registry so the
// node DB row is not left at the schema default, and then onward to the
// fusion engine through the bf-3p6g connect/register sink (bf-24xp).
type FleetNotifier interface {
OnNodeConnected(mac, firmware, chip string)
OnNodeConnected(mac, firmware, chip string, posX, posY, posZ *float64)
OnNodeDisconnected(mac string)
}
@ -548,7 +555,7 @@ func (s *Server) HandleNodeWS(w http.ResponseWriter, r *http.Request) {
}
if fleetFn != nil {
fleetFn.OnNodeConnected(hello.MAC, hello.FirmwareVersion, hello.Chip)
fleetFn.OnNodeConnected(hello.MAC, hello.FirmwareVersion, hello.Chip, hello.PosX, hello.PosY, hello.PosZ)
} else {
s.sendRole(nc, "rx", "")
s.sendConfig(nc, RateIdle, 0, DefaultVarianceThreshold, "")

View file

@ -3,8 +3,22 @@ package simulator
import (
"fmt"
"sort"
)
// DefaultNodeOrigin is the degenerate position virtual nodes carry when they
// are created without an explicit placement — it matches the fleet registry's
// nodes-table schema defaults (pos_x=0, pos_y=0, pos_z=1). Nodes still sitting
// at this origin are treated as "unset": the bridge reassigns them spread-out
// geometry at sync time so the registry and downstream fusion engine never see
// co-located / all-at-origin virtual nodes (core symptom in bf-18yn / bf-4q5w).
var DefaultNodeOrigin = Point{X: 0, Y: 0, Z: 1}
// isDefaultOrigin reports whether p is the unset DB-default origin position.
func isDefaultOrigin(p Point) bool {
return p.X == DefaultNodeOrigin.X && p.Y == DefaultNodeOrigin.Y && p.Z == DefaultNodeOrigin.Z
}
// FleetRegistryBridge integrates virtual nodes with the fleet registry.
// This allows virtual nodes to participate in coverage planning and role assignment.
type FleetRegistryBridge struct {
@ -12,6 +26,78 @@ type FleetRegistryBridge struct {
registryKey string // Prefix for MAC addresses in registry
}
// space returns the geometry the bridge spreads default nodes across. It
// prefers the store's configured space and falls back to DefaultSpace() so the
// bridge is safe even when the store has no space attached.
func (b *FleetRegistryBridge) space() *Space {
if b.store != nil {
if s := b.store.GetSpace(); s != nil {
return s
}
}
return DefaultSpace()
}
// effectivePositions returns the registry position to sync for each node ID.
// Nodes still at the default DB origin (DefaultNodeOrigin) are reassigned
// distinct, spread-out geometry from DefaultNodePositions — sized to the full
// node count so even when every node is unset the geometry spans the room —
// while explicitly-placed (non-origin) nodes keep their position. Default-origin
// nodes are assigned successive spread slots in sorted-ID order, so the mapping
// is deterministic regardless of map iteration order and a single-node sync
// produces the same position as a full sync. The result therefore has no two
// co-located points and is never entirely at the origin.
func (b *FleetRegistryBridge) effectivePositions(nodes []*VirtualNodeState) map[string]Point {
effective := make(map[string]Point, len(nodes))
var defaults []*VirtualNodeState
for _, n := range nodes {
if isDefaultOrigin(n.Position) {
defaults = append(defaults, n)
} else {
effective[n.ID] = n.Position
}
}
if len(defaults) == 0 {
return effective
}
// Deterministic slot assignment independent of map iteration order.
sort.Slice(defaults, func(i, j int) bool { return defaults[i].ID < defaults[j].ID })
space := b.space()
spread := DefaultNodePositions(space, len(nodes))
// Track positions already in use so default-origin nodes never collide
// with each other or with an explicitly-placed node.
taken := make(map[Point]bool, len(effective)+len(defaults))
for _, p := range effective {
taken[p] = true
}
si := 0
for _, n := range defaults {
for si < len(spread) && taken[spread[si]] {
si++
}
var pos Point
if si < len(spread) {
pos = spread[si]
si++
} else {
// Spread set exhausted: more default nodes than slots. The set is
// sized to len(nodes) so this is effectively unreachable; fall back
// to the room center, which is still distinct from the origin for
// any non-degenerate room.
minX, minY, minZ, maxX, maxY, maxZ := space.Bounds()
pos = Point{X: (minX + maxX) / 2, Y: (minY + maxY) / 2, Z: (minZ + maxZ) / 2}
}
effective[n.ID] = pos
taken[pos] = true
}
return effective
}
// NewFleetRegistryBridge creates a new bridge between virtual nodes and fleet registry
func NewFleetRegistryBridge(store *VirtualNodeStore) *FleetRegistryBridge {
return &FleetRegistryBridge{
@ -42,16 +128,24 @@ type NodeRecord struct {
Enabled bool
}
// SyncToRegistry synchronizes all virtual nodes to the fleet registry
// SyncToRegistry synchronizes all virtual nodes to the fleet registry.
//
// Positions are resolved through effectivePositions: any node still at the
// default DB origin (DefaultNodeOrigin) is reassigned distinct, spread-out
// geometry across the store's space so the registry — and the fusion engine
// fed from it via the existing wiring — never observes co-located or
// all-at-origin virtual nodes.
func (b *FleetRegistryBridge) SyncToRegistry(registry RegistryNodeAdapter) error {
if registry == nil {
return fmt.Errorf("registry is nil")
}
nodes := b.store.ListNodes()
positions := b.effectivePositions(nodes)
for _, node := range nodes {
mac := b.virtualMAC(node.ID)
pos := positions[node.ID]
// Check if node exists in registry
existing, err := registry.GetNode(mac)
@ -60,21 +154,21 @@ func (b *FleetRegistryBridge) SyncToRegistry(registry RegistryNodeAdapter) error
if err := registry.AddVirtualNode(
mac,
node.Name,
node.Position.X,
node.Position.Y,
node.Position.Z,
pos.X,
pos.Y,
pos.Z,
); err != nil {
return fmt.Errorf("add virtual node %s: %w", node.ID, err)
}
} else {
// Node exists, update position/role if changed
if existing.PosX != node.Position.X ||
existing.PosY != node.Position.Y ||
existing.PosZ != node.Position.Z {
if existing.PosX != pos.X ||
existing.PosY != pos.Y ||
existing.PosZ != pos.Z {
if err := registry.SetNodePosition(mac,
node.Position.X,
node.Position.Y,
node.Position.Z,
pos.X,
pos.Y,
pos.Z,
); err != nil {
return fmt.Errorf("update position for %s: %w", node.ID, err)
}
@ -94,7 +188,12 @@ func (b *FleetRegistryBridge) SyncToRegistry(registry RegistryNodeAdapter) error
return nil
}
// SyncOneNode syncs a single virtual node to the registry
// SyncOneNode syncs a single virtual node to the registry.
//
// The effective position is resolved over the full node set (not just this
// node) so a single-node sync produces the same spread geometry a full sync
// would write: default-origin nodes receive deterministic spread slots keyed
// by their rank among the unset nodes.
func (b *FleetRegistryBridge) SyncOneNode(registry RegistryNodeAdapter, nodeID string) error {
if registry == nil {
return fmt.Errorf("registry is nil")
@ -105,6 +204,10 @@ func (b *FleetRegistryBridge) SyncOneNode(registry RegistryNodeAdapter, nodeID s
return fmt.Errorf("get node %s: %w", nodeID, err)
}
// Resolve over the full set so the slot assignment matches SyncToRegistry.
positions := b.effectivePositions(b.store.ListNodes())
pos := positions[nodeID]
mac := b.virtualMAC(nodeID)
existing, err := registry.GetNode(mac)
@ -113,20 +216,20 @@ func (b *FleetRegistryBridge) SyncOneNode(registry RegistryNodeAdapter, nodeID s
return registry.AddVirtualNode(
mac,
node.Name,
node.Position.X,
node.Position.Y,
node.Position.Z,
pos.X,
pos.Y,
pos.Z,
)
}
// Update existing node
if existing.PosX != node.Position.X ||
existing.PosY != node.Position.Y ||
existing.PosZ != node.Position.Z {
if existing.PosX != pos.X ||
existing.PosY != pos.Y ||
existing.PosZ != pos.Z {
if err := registry.SetNodePosition(mac,
node.Position.X,
node.Position.Y,
node.Position.Z,
pos.X,
pos.Y,
pos.Z,
); err != nil {
return fmt.Errorf("update position: %w", err)
}
@ -180,19 +283,23 @@ func (b *FleetRegistryBridge) VirtualNodeID(mac string) (string, bool) {
return "", true // TODO: implement reverse mapping
}
// ToRegistryRecords converts virtual nodes to fleet registry records
// ToRegistryRecords converts virtual nodes to fleet registry records. Positions
// are resolved through effectivePositions so the records reflect exactly what
// SyncToRegistry would write (spread geometry for default-origin nodes).
func (b *FleetRegistryBridge) ToRegistryRecords() []NodeRecord {
nodes := b.store.ListNodes()
positions := b.effectivePositions(nodes)
records := make([]NodeRecord, 0, len(nodes))
for _, node := range nodes {
pos := positions[node.ID]
records = append(records, NodeRecord{
MAC: b.virtualMAC(node.ID),
Name: node.Name,
Role: string(node.Role),
PosX: node.Position.X,
PosY: node.Position.Y,
PosZ: node.Position.Z,
PosX: pos.X,
PosY: pos.Y,
PosZ: pos.Z,
Virtual: true,
Enabled: node.Enabled,
})

View file

@ -0,0 +1,337 @@
package simulator
import (
"fmt"
"testing"
)
// fakeRegistry is an in-memory RegistryNodeAdapter that records every mutation
// so bridge tests can assert what was written to the registry (and in what order).
type fakeRegistry struct {
nodes map[string]NodeRecord
addCalls []NodeRecord
setPosCalls []NodeRecord
setRoleCalls []macRole
}
type macRole struct {
MAC, Role string
}
func newFakeRegistry() *fakeRegistry {
return &fakeRegistry{nodes: make(map[string]NodeRecord)}
}
func (f *fakeRegistry) AddVirtualNode(mac, name string, x, y, z float64) error {
rec := NodeRecord{MAC: mac, Name: name, PosX: x, PosY: y, PosZ: z, Virtual: true, Enabled: true}
f.nodes[mac] = rec
f.addCalls = append(f.addCalls, rec)
return nil
}
func (f *fakeRegistry) SetNodePosition(mac string, x, y, z float64) error {
rec := f.nodes[mac]
rec.PosX, rec.PosY, rec.PosZ = x, y, z
f.nodes[mac] = rec
f.setPosCalls = append(f.setPosCalls, NodeRecord{MAC: mac, PosX: x, PosY: y, PosZ: z})
return nil
}
func (f *fakeRegistry) SetNodeRole(mac, role string) error {
rec := f.nodes[mac]
rec.Role = role
f.nodes[mac] = rec
f.setRoleCalls = append(f.setRoleCalls, macRole{MAC: mac, Role: role})
return nil
}
func (f *fakeRegistry) DeleteNode(mac string) error {
delete(f.nodes, mac)
return nil
}
func (f *fakeRegistry) GetNode(mac string) (*NodeRecord, error) {
rec, ok := f.nodes[mac]
if !ok {
return nil, fmt.Errorf("not found: %s", mac)
}
return &rec, nil
}
func (f *fakeRegistry) GetAllNodes() ([]NodeRecord, error) {
out := make([]NodeRecord, 0, len(f.nodes))
for _, r := range f.nodes {
out = append(out, r)
}
return out, nil
}
// storeWithNodes builds a fresh store (DefaultSpace) holding one virtual node
// per entry. Each entry gives (id, position). Positions at the default origin
// are created as-is (the origin is in-bounds for DefaultSpace).
func storeWithNodes(t *testing.T, nodes ...struct{ ID string; Pos Point }) *VirtualNodeStore {
t.Helper()
store, _ := tempStore(t)
for _, n := range nodes {
if _, err := store.CreateVirtualNode(n.ID, n.ID, n.Pos); err != nil {
t.Fatalf("create %s: %v", n.ID, err)
}
}
return store
}
// originStore builds a store with count nodes all at the default DB origin.
func originStore(t *testing.T, count int) *VirtualNodeStore {
t.Helper()
entries := make([]struct{ ID string; Pos Point }, 0, count)
for i := 0; i < count; i++ {
entries = append(entries, struct{ ID string; Pos Point }{
ID: fmt.Sprintf("node-%d", i+1),
Pos: DefaultNodeOrigin,
})
}
return storeWithNodes(t, entries...)
}
// assertDistinctNonOrigin fails the test if any record is still at the default
// origin or if any two records share the same position.
func assertDistinctNonOrigin(t *testing.T, recs []NodeRecord) {
t.Helper()
seen := make(map[Point]bool, len(recs))
for _, r := range recs {
p := Point{X: r.PosX, Y: r.PosY, Z: r.PosZ}
if isDefaultOrigin(p) {
t.Errorf("node %s still at default origin %v", r.MAC, p)
}
if seen[p] {
t.Errorf("co-located registry nodes at %v (mac %s)", p, r.MAC)
}
seen[p] = true
}
}
func TestIsDefaultOrigin(t *testing.T) {
cases := []struct {
name string
p Point
want bool
}{
{"db default origin", DefaultNodeOrigin, true},
{"explicit zero", Point{X: 0, Y: 0, Z: 0}, false},
{"origin wrong Z", Point{X: 0, Y: 0, Z: 1.5}, false},
{"origin shifted X", Point{X: 1, Y: 0, Z: 1}, false},
{"placed node", Point{X: 3, Y: 2, Z: 1.5}, false},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
if got := isDefaultOrigin(c.p); got != c.want {
t.Errorf("isDefaultOrigin(%v) = %v, want %v", c.p, got, c.want)
}
})
}
}
// TestSyncToRegistry_OriginNodesGetSpreadGeometry is the core acceptance check:
// a fleet of nodes created at the default origin must be written to the
// registry at distinct, non-degenerate positions (spread across the room).
func TestSyncToRegistry_OriginNodesGetSpreadGeometry(t *testing.T) {
for _, count := range []int{1, 2, 3, 4, 6, 9} {
t.Run(fmt.Sprintf("count=%d", count), func(t *testing.T) {
store := originStore(t, count)
bridge := NewFleetRegistryBridge(store)
reg := newFakeRegistry()
if err := bridge.SyncToRegistry(reg); err != nil {
t.Fatalf("SyncToRegistry: %v", err)
}
if len(reg.nodes) != count {
t.Fatalf("expected %d registry nodes, got %d", count, len(reg.nodes))
}
if len(reg.addCalls) != count {
t.Fatalf("expected %d AddVirtualNode calls, got %d", count, len(reg.addCalls))
}
recs := make([]NodeRecord, 0, len(reg.nodes))
for _, r := range reg.nodes {
recs = append(recs, r)
}
assertDistinctNonOrigin(t, recs)
})
}
}
// TestSyncToRegistry_ExplicitPositionsPreserved verifies nodes placed at a real
// (non-origin) position are written through unchanged — the bridge only
// substitutes geometry for unset nodes.
func TestSyncToRegistry_ExplicitPositionsPreserved(t *testing.T) {
a := Point{X: 1, Y: 1, Z: 1.5}
b := Point{X: 4, Y: 4, Z: 2}
store := storeWithNodes(t,
struct{ ID string; Pos Point }{"a", a},
struct{ ID string; Pos Point }{"b", b},
)
bridge := NewFleetRegistryBridge(store)
reg := newFakeRegistry()
if err := bridge.SyncToRegistry(reg); err != nil {
t.Fatalf("SyncToRegistry: %v", err)
}
for id, want := range map[string]Point{"a": a, "b": b} {
rec, ok := reg.nodes[bridge.virtualMAC(id)]
if !ok {
t.Fatalf("node %s missing from registry", id)
}
got := Point{X: rec.PosX, Y: rec.PosY, Z: rec.PosZ}
if got != want {
t.Errorf("node %s: expected explicit %v, got %v", id, want, got)
}
}
}
// TestSyncToRegistry_MixedNodes keeps explicit placements while spreading the
// unset ones, and asserts the final set is distinct and non-origin.
func TestSyncToRegistry_MixedNodes(t *testing.T) {
store := storeWithNodes(t,
struct{ ID string; Pos Point }{"explicit", Point{X: 5, Y: 4, Z: 2}},
struct{ ID string; Pos Point }{"o1", DefaultNodeOrigin},
struct{ ID string; Pos Point }{"o2", DefaultNodeOrigin},
struct{ ID string; Pos Point }{"o3", DefaultNodeOrigin},
)
bridge := NewFleetRegistryBridge(store)
reg := newFakeRegistry()
if err := bridge.SyncToRegistry(reg); err != nil {
t.Fatalf("SyncToRegistry: %v", err)
}
// Explicit node keeps its position.
rec := reg.nodes[bridge.virtualMAC("explicit")]
if (Point{X: rec.PosX, Y: rec.PosY, Z: rec.PosZ}) != (Point{X: 5, Y: 4, Z: 2}) {
t.Errorf("explicit node moved: got %v", rec)
}
recs := make([]NodeRecord, 0, len(reg.nodes))
for _, r := range reg.nodes {
recs = append(recs, r)
}
assertDistinctNonOrigin(t, recs)
}
// TestSyncToRegistry_Idempotent asserts a second sync is a no-op: because the
// spread slot assignment is deterministic, the registry already holds the
// effective positions and no further writes occur.
func TestSyncToRegistry_Idempotent(t *testing.T) {
store := originStore(t, 4)
bridge := NewFleetRegistryBridge(store)
reg := newFakeRegistry()
if err := bridge.SyncToRegistry(reg); err != nil {
t.Fatalf("first SyncToRegistry: %v", err)
}
firstAdds, firstSets := len(reg.addCalls), len(reg.setPosCalls)
if err := bridge.SyncToRegistry(reg); err != nil {
t.Fatalf("second SyncToRegistry: %v", err)
}
if len(reg.addCalls) != firstAdds {
t.Errorf("second sync added nodes again: %d != %d", len(reg.addCalls), firstAdds)
}
if len(reg.setPosCalls) != firstSets {
t.Errorf("second sync rewrote positions: %d != %d", len(reg.setPosCalls), firstSets)
}
}
// TestSyncOneNode_MatchesFullSync asserts a single-node sync writes the same
// effective position that a full sync would — including spread geometry for a
// default-origin node.
func TestSyncOneNode_MatchesFullSync(t *testing.T) {
store := originStore(t, 3)
fullReg := newFakeRegistry()
bridge := NewFleetRegistryBridge(store)
if err := bridge.SyncToRegistry(fullReg); err != nil {
t.Fatalf("SyncToRegistry: %v", err)
}
target := "node-2"
mac := bridge.virtualMAC(target)
wantRec, ok := fullReg.nodes[mac]
if !ok {
t.Fatalf("node %s missing from full sync", target)
}
oneReg := newFakeRegistry()
if err := bridge.SyncOneNode(oneReg, target); err != nil {
t.Fatalf("SyncOneNode: %v", err)
}
gotRec, ok := oneReg.nodes[mac]
if !ok {
t.Fatalf("node %s not written by SyncOneNode", target)
}
want := Point{X: wantRec.PosX, Y: wantRec.PosY, Z: wantRec.PosZ}
got := Point{X: gotRec.PosX, Y: gotRec.PosY, Z: gotRec.PosZ}
if got != want {
t.Errorf("SyncOneNode position %v != SyncToRegistry position %v", got, want)
}
if isDefaultOrigin(got) {
t.Errorf("SyncOneNode left default-origin node at the origin: %v", got)
}
}
// TestSyncOneNode_ExplicitPreserved asserts SyncOneNode passes an explicit
// (non-origin) position through unchanged.
func TestSyncOneNode_ExplicitPreserved(t *testing.T) {
store := storeWithNodes(t, struct{ ID string; Pos Point }{"a", Point{X: 2, Y: 3, Z: 1.2}})
bridge := NewFleetRegistryBridge(store)
reg := newFakeRegistry()
if err := bridge.SyncOneNode(reg, "a"); err != nil {
t.Fatalf("SyncOneNode: %v", err)
}
rec := reg.nodes[bridge.virtualMAC("a")]
got := Point{X: rec.PosX, Y: rec.PosY, Z: rec.PosZ}
if got != (Point{X: 2, Y: 3, Z: 1.2}) {
t.Errorf("expected explicit position preserved, got %v", got)
}
}
// TestSyncToRegistry_NilRegistry guards the nil-adapter guard.
func TestSyncToRegistry_NilRegistry(t *testing.T) {
store := originStore(t, 2)
bridge := NewFleetRegistryBridge(store)
if err := bridge.SyncToRegistry(nil); err == nil {
t.Fatal("expected error syncing to nil registry")
}
if err := bridge.SyncOneNode(nil, "node-1"); err == nil {
t.Fatal("expected error syncing one node to nil registry")
}
}
// TestToRegistryRecords_UsesEffectivePositions ensures the record view matches
// what SyncToRegistry writes (spread geometry for origin nodes).
func TestToRegistryRecords_UsesEffectivePositions(t *testing.T) {
store := originStore(t, 3)
bridge := NewFleetRegistryBridge(store)
records := bridge.ToRegistryRecords()
if len(records) != 3 {
t.Fatalf("expected 3 records, got %d", len(records))
}
assertDistinctNonOrigin(t, records)
// Cross-check against an actual sync.
reg := newFakeRegistry()
if err := bridge.SyncToRegistry(reg); err != nil {
t.Fatalf("SyncToRegistry: %v", err)
}
for _, r := range records {
synced := reg.nodes[r.MAC]
if synced.PosX != r.PosX || synced.PosY != r.PosY || synced.PosZ != r.PosZ {
t.Errorf("record %s %v != synced %v", r.MAC, r, synced)
}
}
}