From a17bdd2c8c9f4dc31aaecfbe47b96ab9b3b29a5b Mon Sep 17 00:00:00 2001 From: jedarden Date: Fri, 3 Jul 2026 17:31:40 -0400 Subject: [PATCH] feat(fusion): propagate runtime node positions to the fusion engine (bf-3p6g) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Forward node 3D position changes to the blob-producing fusion engine so a node moved at runtime does not keep a stale engine position. The fleet manager gains a nodePositionSink — the write-side mirror of the read-side SetNodePositionAccessor used by diagnostics and weather.go — wired to fusionEngine.SetNodePosition at startup. It fires from both position paths: PATCH /api/nodes/{mac}/position (handler.updateNodePosition → ForwardNodePosition) and node connect/register (OnNodeConnected forwards the registry position). Tests assert the engine's stored position for a MAC changes after a PATCH with new coordinates and is seeded on connect. Co-Authored-By: Claude --- mothership/cmd/mothership/main.go | 12 +++ mothership/internal/fleet/handler.go | 4 + mothership/internal/fleet/handler_test.go | 103 ++++++++++++++++++++++ mothership/internal/fleet/manager.go | 46 ++++++++++ 4 files changed, 165 insertions(+) diff --git a/mothership/cmd/mothership/main.go b/mothership/cmd/mothership/main.go index 621dc1b..b2774dd 100644 --- a/mothership/cmd/mothership/main.go +++ b/mothership/cmd/mothership/main.go @@ -1410,6 +1410,18 @@ func main() { // Legacy fleet manager (kept for basic operations) fleetMgr := fleet.NewManager(fleetReg) + // bf-3p6g: Forward runtime node position changes to the blob-producing 3D + // fusion engine so a node moved at runtime does not stay at its stale engine + // position. The sink fires from both PATCH /api/nodes/{mac}/position + // (fleet.Handler.updateNodePosition → ForwardNodePosition) and node + // connect/register (Manager.OnNodeConnected), converging the two position + // paths on one accessor — the write-side mirror of the read-side + // SetNodePositionAccessor used by diagnostics and weather.go. fusionEngine + // is constructed above (Phase 6, bf-3f6q) and captured by this closure. + fleetMgr.SetNodePositionSink(func(mac string, x, y, z float64) { + fusionEngine.SetNodePosition(mac, x, y, z) + }) + // Phase 5: Multi-notifier broadcasts node events to legacy manager, healer, and self-heal manager multiNotify := newMultiNotifier(fleetMgr, fleetHealer, selfHealManager) ingestSrv.SetFleetNotifier(multiNotify) diff --git a/mothership/internal/fleet/handler.go b/mothership/internal/fleet/handler.go index 8355261..7220e31 100644 --- a/mothership/internal/fleet/handler.go +++ b/mothership/internal/fleet/handler.go @@ -393,6 +393,10 @@ func (h *Handler) updateNodePosition(w http.ResponseWriter, r *http.Request) { return } h.mgr.BroadcastRegistry() + // bf-3p6g: Forward the new position to the blob-producing fusion engine so + // a node moved at runtime does not stay at its stale engine position (the + // engine's node registry is the geometry Fuse localizes against). + h.mgr.ForwardNodePosition(mac, req.X, req.Y, req.Z) w.WriteHeader(http.StatusNoContent) } diff --git a/mothership/internal/fleet/handler_test.go b/mothership/internal/fleet/handler_test.go index 46c9d40..cd9bbc3 100644 --- a/mothership/internal/fleet/handler_test.go +++ b/mothership/internal/fleet/handler_test.go @@ -11,6 +11,7 @@ import ( "time" "github.com/go-chi/chi/v5" + "github.com/spaxel/mothership/internal/fusion" ) // mockNodeIdentifier is a mock implementation of NodeIdentifier for testing. @@ -1425,6 +1426,108 @@ func TestHandlerUpdateNodePosition(t *testing.T) { } } +// engineNodePosition returns the fusion engine's stored position for mac, or +// nil if the engine holds no position for it. The engine exposes its node +// registry only via NodePositions() (a snapshot slice), so the helper scans. +func engineNodePosition(e *fusion.Engine, mac string) *fusion.NodePosition { + for _, p := range e.NodePositions() { + if p.MAC == mac { + cp := p + return &cp + } + } + return nil +} + +// patchNodePosition issues a PUT /api/nodes/{mac}/position through the handler +// and fails the test if the handler does not return 204 No Content. +func patchNodePosition(t *testing.T, h *Handler, mac, body string) { + t.Helper() + req := httptest.NewRequest("PUT", "/api/nodes/"+mac+"/position", bytes.NewBufferString(body)) + req.Header.Set("Content-Type", "application/json") + rctx := chi.NewRouteContext() + rctx.URLParams.Add("mac", mac) + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, rctx)) + w := httptest.NewRecorder() + h.updateNodePosition(w, req) + if w.Code != http.StatusNoContent { + t.Fatalf("updateNodePosition status = %v, want %d (body=%q)", w.Code, http.StatusNoContent, w.Body.String()) + } +} + +// TestHandlerUpdateNodePositionForwardsToEngine (bf-3p6g) verifies that a +// PATCH /api/nodes/{mac}/position flows through the handler and the manager's +// node position sink into the blob-producing fusion engine, and that the +// position the engine stores for that MAC changes to the new coordinates. +func TestHandlerUpdateNodePositionForwardsToEngine(t *testing.T) { + const mac = "AA:BB:CC:DD:EE:FF" + reg := newTestRegistry(t) + reg.UpsertNode(mac, "1.0.0", "ESP32-S3") + reg.SetNodeRole(mac, "rx") + reg.SetNodePosition(mac, 0, 0, 0) + + engine := fusion.NewEngine(nil) + mgr := NewManager(reg) + mgr.SetNodePositionSink(func(m string, x, y, z float64) { + engine.SetNodePosition(m, x, y, z) + }) + h := &Handler{mgr: mgr} + + // Sanity: engine holds no position for this MAC before any PATCH. + if p := engineNodePosition(engine, mac); p != nil { + t.Fatalf("engine should hold no position before PATCH, got (%v,%v,%v)", p.X, p.Y, p.Z) + } + + // First PATCH seeds the engine position. + patchNodePosition(t, h, mac, `{"x": 1.5, "y": 2.5, "z": 3.5}`) + p := engineNodePosition(engine, mac) + if p == nil { + t.Fatal("PATCH did not forward node position to fusion engine") + } + if p.X != 1.5 || p.Y != 2.5 || p.Z != 3.5 { + t.Errorf("engine position after PATCH = (%v,%v,%v), want (1.5,2.5,3.5)", p.X, p.Y, p.Z) + } + + // Second PATCH with new coordinates: the stored engine position must change. + patchNodePosition(t, h, mac, `{"x": 7.0, "y": 8.0, "z": 9.0}`) + p = engineNodePosition(engine, mac) + if p == nil { + t.Fatal("engine lost position after second PATCH") + } + if p.X != 7.0 || p.Y != 8.0 || p.Z != 9.0 { + t.Errorf("engine position after second PATCH = (%v,%v,%v), want (7,8,9)", p.X, p.Y, p.Z) + } +} + +// TestManagerOnNodeConnectedForwardsPositionToEngine (bf-3p6g) verifies that a +// node connect/register event pushes the node's current registry position into +// the fusion engine, so a node positioned (or moved) while offline does not +// keep a stale engine position when it reconnects. +func TestManagerOnNodeConnectedForwardsPositionToEngine(t *testing.T) { + const mac = "AA:BB:CC:DD:EE:FF" + reg := newTestRegistry(t) + // Pre-position the node in the registry (e.g. placed while offline), then + // connect it; the manager must forward the registry position to the engine. + reg.UpsertNode(mac, "1.0.0", "ESP32-S3") + reg.SetNodePosition(mac, 4.0, 5.0, 6.0) + + engine := fusion.NewEngine(nil) + mgr := NewManager(reg) + mgr.SetNodePositionSink(func(m string, x, y, z float64) { + engine.SetNodePosition(m, x, y, z) + }) + + mgr.OnNodeConnected(mac, "1.0.0", "ESP32-S3") + + p := engineNodePosition(engine, mac) + if p == nil { + t.Fatal("OnNodeConnected did not forward node position to fusion engine") + } + if p.X != 4.0 || p.Y != 5.0 || p.Z != 6.0 { + t.Errorf("engine position after connect = (%v,%v,%v), want (4,5,6)", p.X, p.Y, p.Z) + } +} + // ─── Fleet page specific tests ───────────────────────────────────────────────────── // TestFleetTableRendering verifies that the fleet table renders correctly with 4 nodes diff --git a/mothership/internal/fleet/manager.go b/mothership/internal/fleet/manager.go index 8cb5107..0af5037 100644 --- a/mothership/internal/fleet/manager.go +++ b/mothership/internal/fleet/manager.go @@ -108,6 +108,16 @@ type Manager struct { // Collision detection and adaptive re-stagger collisionDetector *CollisionDetector collisionCheckTick time.Duration + + // nodePositionSink, when set, forwards a node's 3D world position to the + // blob-producing fusion engine (internal/fusion.Engine) whenever a node is + // positioned at runtime: from a PATCH /api/nodes/{mac}/position request + // (Handler.updateNodePosition → ForwardNodePosition) or a node + // connect/register event (OnNodeConnected). This is the write-side mirror + // of the read-side SetNodePositionAccessor pattern used by diagnostics and + // weather.go. Without it, a node moved at runtime keeps its stale engine + // position and Fuse localizes against the wrong geometry (bf-3p6g). + nodePositionSink func(mac string, x, y, z float64) } // NewManager creates a fleet manager backed by registry. @@ -144,6 +154,31 @@ func (m *Manager) SetBroadcaster(b RegistryBroadcaster) { m.mu.Unlock() } +// SetNodePositionSink wires the sink that receives node position updates to +// forward to the blob-producing fusion engine. Injected at startup (main.go) +// — see nodePositionSink. Mirrors the read-side SetNodePositionAccessor used +// by diagnostics/weather.go. +func (m *Manager) SetNodePositionSink(fn func(mac string, x, y, z float64)) { + m.mu.Lock() + m.nodePositionSink = fn + m.mu.Unlock() +} + +// ForwardNodePosition pushes a node's position to the registered sink (if any). +// Called by the REST handler after the registry has accepted a new position +// (PATCH /api/nodes/{mac}/position), and by OnNodeConnected on node +// connect/register, so the fusion engine's node registry stays in sync with +// the fleet registry and a node moved at runtime does not keep a stale engine +// position (bf-3p6g). +func (m *Manager) ForwardNodePosition(mac string, x, y, z float64) { + m.mu.RLock() + sink := m.nodePositionSink + m.mu.RUnlock() + if sink != nil { + sink(mac, x, y, z) + } +} + // 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) { @@ -162,6 +197,17 @@ func (m *Manager) OnNodeConnected(mac, firmware, chip string) { m.applyRoleAndConfig(mac, role) m.broadcastRegistry() + + // bf-3p6g: Push the node's current registry position to the fusion engine + // on connect/register so a node positioned (or moved) while offline does + // not keep a stale engine position. The fleet registry is the source of + // truth for positions; the engine holds a mirror that Fuse reads each + // tick. On first connect the position is the schema default until the user + // places the node, which is still forwarded (it matches the registry). + if x, y, z, ok := m.registry.GetNodePosition(mac); ok { + m.ForwardNodePosition(mac, x, y, z) + } + log.Printf("[INFO] fleet: node %s joined as %s", mac, role) }