// Package detect implements the detection engine that consumes probe snapshots, // normalises them into per-node signals, applies five classification rules // (suspect / traffic-warning / confirmed / recover / fault), and drives // lifecycle-state transitions via the LifecycleService interface. // // This package only exposes Engine.Tick(ctx); the caller (15H DetectLoop) is // responsible for the 5-minute schedule and for leader election. package detect import ( "context" "fmt" "sync" "time" ) // NodeStatus is the lifecycle state of a node as used by this package. type NodeStatus string const ( StatusProvisioning NodeStatus = "provisioning" StatusProbing NodeStatus = "probing" StatusUp NodeStatus = "up" StatusDraining NodeStatus = "draining" StatusDown NodeStatus = "down" StatusDestroyed NodeStatus = "destroyed" StatusBlockedSuspect NodeStatus = "blocked_suspect" StatusBlockedConfirmed NodeStatus = "blocked_confirmed" ) // NodeFilter restricts which nodes ListNodes returns. // An empty Statuses slice means "all statuses". type NodeFilter struct { Statuses []NodeStatus } // NodeInfo is the minimal node descriptor required by the detection engine. type NodeInfo struct { ID string UUID string Status NodeStatus Weight int Version int64 } // LoadInfo is the most recent load sample for a node. type LoadInfo struct { Online int BandwidthMbps float64 Timestamp int64 } // LoadPoint is a single load sample in a historical series. type LoadPoint struct { Timestamp int64 Online int BandwidthMbps float64 } // LifecycleService is the interface the detection engine uses to read and mutate // node lifecycle state. // // The real implementation (#5 LifecycleService) executes all mutations inside // MySQL transactions with optimistic locks. MockLifecycle is the in-memory // stub used in tests. // // TransitionStatus convention: the underlying store executes // // UPDATE nodes SET status=to, version=version+1 // WHERE id=nodeID AND status=from // // and returns the number of rows affected. A return value of 0 means the node // was already in a different state (concurrent change); the caller must treat // this as a no-op for the current tick (idempotent, safe to retry next cycle). type LifecycleService interface { // ListNodes returns nodes matching the filter. ListNodes(ctx context.Context, filter NodeFilter) ([]NodeInfo, error) // TransitionStatus attempts an optimistic-lock status transition. // On success it writes a node_events record with from/to/detail and bumps // the node version. Returns (1, nil) on success, (0, nil) on lock conflict. TransitionStatus(ctx context.Context, nodeID string, from, to NodeStatus, detail map[string]any) (int, error) // SetWeight updates the routing weight for a node. SetWeight(ctx context.Context, nodeID string, weight int) error // BumpVersion increments the global directory version so clients refetch. BumpVersion(ctx context.Context) error // GetLoad returns the latest load sample for a node. GetLoad(ctx context.Context, nodeID string) (LoadInfo, error) // GetLoadHistory returns load samples recorded within the given window // (oldest to newest), enabling a drop-percentage computation. GetLoadHistory(ctx context.Context, nodeID string, window time.Duration) ([]LoadPoint, error) } // ───────────────────────────────────────────────────────────────────────────── // MockLifecycle — in-memory stub for unit tests // ───────────────────────────────────────────────────────────────────────────── // MockEvent records a single TransitionStatus call for assertion in tests. type MockEvent struct { NodeID string From NodeStatus To NodeStatus Detail map[string]any } // MockLifecycle is an in-memory LifecycleService implementation used in tests. // All fields are safe for concurrent use. type MockLifecycle struct { mu sync.Mutex nodes map[string]*NodeInfo // nodeID → node events []MockEvent // recorded TransitionStatus calls version int64 // global directory version loads map[string][]LoadPoint // nodeID → ordered load samples // conflictKeys is a set of "nodeID:from:to" strings that should simulate // an optimistic-lock conflict (TransitionStatus returns 0 rows). conflictKeys map[string]bool } // NewMockLifecycle creates a MockLifecycle pre-populated with the given nodes. func NewMockLifecycle(nodes []NodeInfo) *MockLifecycle { m := &MockLifecycle{ nodes: make(map[string]*NodeInfo, len(nodes)), loads: make(map[string][]LoadPoint), conflictKeys: make(map[string]bool), } for _, n := range nodes { nn := n m.nodes[n.ID] = &nn } return m } // SetConflict registers a (nodeID, from, to) tuple that should simulate a // concurrent-modification conflict on the next matching TransitionStatus call. func (m *MockLifecycle) SetConflict(nodeID string, from, to NodeStatus) { m.mu.Lock() defer m.mu.Unlock() m.conflictKeys[conflictKey(nodeID, from, to)] = true } // ClearConflict removes a previously registered conflict. func (m *MockLifecycle) ClearConflict(nodeID string, from, to NodeStatus) { m.mu.Lock() defer m.mu.Unlock() delete(m.conflictKeys, conflictKey(nodeID, from, to)) } func conflictKey(nodeID string, from, to NodeStatus) string { return fmt.Sprintf("%s:%s:%s", nodeID, from, to) } // SetLoads sets the load history for a node (oldest to newest). func (m *MockLifecycle) SetLoads(nodeID string, points []LoadPoint) { m.mu.Lock() defer m.mu.Unlock() m.loads[nodeID] = append([]LoadPoint{}, points...) } // NodeStatus returns the current status of a node (test helper). func (m *MockLifecycle) NodeStatus(nodeID string) NodeStatus { m.mu.Lock() defer m.mu.Unlock() if n, ok := m.nodes[nodeID]; ok { return n.Status } return "" } // NodeWeight returns the current weight of a node (test helper). func (m *MockLifecycle) NodeWeight(nodeID string) int { m.mu.Lock() defer m.mu.Unlock() if n, ok := m.nodes[nodeID]; ok { return n.Weight } return 0 } // Events returns a copy of all recorded TransitionStatus calls. func (m *MockLifecycle) Events() []MockEvent { m.mu.Lock() defer m.mu.Unlock() return append([]MockEvent{}, m.events...) } // Version returns the current global directory version (test helper). func (m *MockLifecycle) Version() int64 { m.mu.Lock() defer m.mu.Unlock() return m.version } // ListNodes implements LifecycleService. func (m *MockLifecycle) ListNodes(_ context.Context, filter NodeFilter) ([]NodeInfo, error) { m.mu.Lock() defer m.mu.Unlock() statusSet := make(map[NodeStatus]bool, len(filter.Statuses)) for _, s := range filter.Statuses { statusSet[s] = true } var result []NodeInfo for _, n := range m.nodes { if len(statusSet) == 0 || statusSet[n.Status] { result = append(result, *n) } } return result, nil } // TransitionStatus implements LifecycleService. func (m *MockLifecycle) TransitionStatus(_ context.Context, nodeID string, from, to NodeStatus, detail map[string]any) (int, error) { m.mu.Lock() defer m.mu.Unlock() if m.conflictKeys[conflictKey(nodeID, from, to)] { return 0, nil } n, ok := m.nodes[nodeID] if !ok { return 0, fmt.Errorf("mock: node %s not found", nodeID) } if n.Status != from { // Optimistic lock failed: node is in a different state. return 0, nil } n.Status = to n.Version++ m.events = append(m.events, MockEvent{ NodeID: nodeID, From: from, To: to, Detail: detail, }) return 1, nil } // SetWeight implements LifecycleService. func (m *MockLifecycle) SetWeight(_ context.Context, nodeID string, weight int) error { m.mu.Lock() defer m.mu.Unlock() n, ok := m.nodes[nodeID] if !ok { return fmt.Errorf("mock: node %s not found", nodeID) } n.Weight = weight return nil } // BumpVersion implements LifecycleService. func (m *MockLifecycle) BumpVersion(_ context.Context) error { m.mu.Lock() defer m.mu.Unlock() m.version++ return nil } // GetLoad implements LifecycleService. func (m *MockLifecycle) GetLoad(_ context.Context, nodeID string) (LoadInfo, error) { m.mu.Lock() defer m.mu.Unlock() pts := m.loads[nodeID] if len(pts) == 0 { return LoadInfo{}, nil } p := pts[len(pts)-1] return LoadInfo{Online: p.Online, BandwidthMbps: p.BandwidthMbps, Timestamp: p.Timestamp}, nil } // GetLoadHistory implements LifecycleService. func (m *MockLifecycle) GetLoadHistory(_ context.Context, nodeID string, window time.Duration) ([]LoadPoint, error) { m.mu.Lock() defer m.mu.Unlock() cutoff := time.Now().Add(-window).Unix() var result []LoadPoint for _, p := range m.loads[nodeID] { if p.Timestamp >= cutoff { result = append(result, p) } } return result, nil }