pangolin/server/internal/scheduler/detect/lifecycle.go

// 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
}