diff --git a/server/internal/scheduler/detect/engine.go b/server/internal/scheduler/detect/engine.go new file mode 100644 index 0000000..d8a98d3 --- /dev/null +++ b/server/internal/scheduler/detect/engine.go @@ -0,0 +1,335 @@ +package detect + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/wangjia/pangolin/server/internal/idgen" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Notifier — 15G interface stub +// ───────────────────────────────────────────────────────────────────────────── + +// Notifier is the 15G event sink for fault notifications. +// The real implementation (15G) sends a Telegram/alerting message; the stub +// below writes to the structured logger and is used until 15G is ready. +type Notifier interface { + NotifyFault(ctx context.Context, nodeID, reason string) error +} + +// LogNotifier is a Notifier stub that logs via slog. +// It is used when no real Notifier is wired up. +type LogNotifier struct{} + +// NotifyFault implements Notifier. +func (LogNotifier) NotifyFault(_ context.Context, nodeID, reason string) error { + slog.Warn("node fault detected — manual review required", + "node_id", nodeID, + "reason", reason, + ) + return nil +} + +// ───────────────────────────────────────────────────────────────────────────── +// Redis key constants +// ───────────────────────────────────────────────────────────────────────────── + +const ( + // replaceQueueKey is the Redis list that 15E (Replenisher) drains to + // provision replacement nodes. The queue entry shape is replaceRequest JSON. + replaceQueueKey = "detect:replace:queue" + + // probeFreqKeyPrefix is the per-node Redis key prefix written when a node + // enters "blocked_suspect". The value "60" (seconds) tells 15B to probe + // at 1-minute instead of the default 5-minute interval. + probeFreqKeyPrefix = "probe:freq:" + + // probeFreqTTL keeps the elevated-frequency marker alive for 45 minutes — + // long enough to cover a full suspect → confirmed window (6 × 5 min + buffer). + probeFreqTTL = 45 * time.Minute +) + +// ───────────────────────────────────────────────────────────────────────────── +// Engine +// ───────────────────────────────────────────────────────────────────────────── + +// Engine executes one detection cycle per Tick call. +// +// 15H (DetectLoop) is responsible for: +// - scheduling Ticks at 5-minute intervals, and +// - leader election (only the elected leader should call Tick). +// +// Engine itself holds no goroutines and no in-process state beyond its +// constructor arguments; all persistent state lives in Redis (via StreakStore). +type Engine struct { + probeStore ProbeSnapshotter + lc LifecycleService + streaks *StreakStore + rdb *redis.Client + notifier Notifier + cfg *DetectConfig +} + +// NewEngine creates an Engine. If cfg is nil, DefaultConfig() is used. +// If notifier is nil, LogNotifier is used. +func NewEngine( + probeStore ProbeSnapshotter, + lc LifecycleService, + streaks *StreakStore, + rdb *redis.Client, + notifier Notifier, + cfg *DetectConfig, +) *Engine { + if cfg == nil { + d := DefaultConfig() + cfg = &d + } + if notifier == nil { + notifier = LogNotifier{} + } + return &Engine{ + probeStore: probeStore, + lc: lc, + streaks: streaks, + rdb: rdb, + notifier: notifier, + cfg: cfg, + } +} + +// Tick executes one full detection cycle. +// +// It lists nodes in "up" and "blocked_suspect" states, reads their probe +// snapshots, computes signals, and applies the five classification rules. +// Errors from individual nodes are logged but do not halt the cycle for other +// nodes. +func (e *Engine) Tick(ctx context.Context) error { + nodes, err := e.lc.ListNodes(ctx, NodeFilter{ + Statuses: []NodeStatus{StatusUp, StatusBlockedSuspect}, + }) + if err != nil { + return fmt.Errorf("detect: list nodes: %w", err) + } + + for _, node := range nodes { + if err := e.processNode(ctx, node); err != nil { + slog.Error("detect: process node", + "node_id", node.ID, + "status", node.Status, + "error", err, + ) + // Continue — one node failure must not halt the full cycle. + } + } + return nil +} + +// processNode applies all five rules to a single node. +func (e *Engine) processNode(ctx context.Context, node NodeInfo) error { + snapshots, err := e.probeStore.SnapshotsByNode(ctx, node.ID) + if err != nil { + return fmt.Errorf("snapshots: %w", err) + } + + sig := computeSignals(ctx, node.ID, snapshots, e.lc) + + sk, err := e.streaks.Load(ctx, node.ID) + if err != nil { + return fmt.Errorf("streak load: %w", err) + } + + // ── Rule 5: fault ───────────────────────────────────────────────────────── + // Both domestic AND overseas probes failing → node-level outage, not GFW. + // Do NOT transition status; send an alert event for human review. + // Streaks are left unchanged so that when the node recovers the engine + // resumes from its current position rather than re-triggering immediately. + if isFault(sig) { + reason := fmt.Sprintf("domestic_fail_isps=%d overseas_ok=false", sig.DomesticFailISPs) + if notifyErr := e.notifier.NotifyFault(ctx, node.ID, reason); notifyErr != nil { + slog.Error("detect: notify fault", "node_id", node.ID, "error", notifyErr) + } + return nil // do not persist streak changes + } + + switch node.Status { + case StatusUp: + return e.processUp(ctx, node, sig, sk) + case StatusBlockedSuspect: + return e.processSuspect(ctx, node, sig, sk) + default: + return nil + } +} + +// processUp applies the suspect and traffic-warning rules to an "up" node. +func (e *Engine) processUp(ctx context.Context, node NodeInfo, sig NodeSignal, sk Streak) error { + domesticFailing := e.cfg.isSuspectTriggered(sig) + overseasBarrier := sig.OverseasHasData && sig.OverseasOK + + if domesticFailing && overseasBarrier { + // Domestic probes failing, overseas OK → potential GFW block. + threshold := e.cfg.effectiveSuspectStreakMin(sig) + + if sk.FailStreak+1 >= threshold { + // ── Rule 1: suspect ─────────────────────────────────────────────── + detail := map[string]any{ + "from": "up", + "to": "blocked_suspect", + "domestic_fail_isps": sig.DomesticFailISPs, + "domestic_total": sig.DomesticTotalISPs, + "fail_streak": sk.FailStreak + 1, + } + affected, err := e.lc.TransitionStatus(ctx, node.ID, StatusUp, StatusBlockedSuspect, detail) + if err != nil { + return fmt.Errorf("transition up→suspect: %w", err) + } + if affected == 0 { + // Optimistic lock conflict: another writer changed the state + // concurrently. Leave streak unchanged; the engine is idempotent. + slog.Info("detect: suspect transition skipped (lock conflict)", "node_id", node.ID) + return e.streaks.Save(ctx, node.ID, sk) + } + + // Transition succeeded: apply side-effects. + if err := e.lc.SetWeight(ctx, node.ID, e.cfg.SuspectWeight); err != nil { + slog.Error("detect: set suspect weight", "node_id", node.ID, "error", err) + } + if err := e.writeProbeFreqMark(ctx, node.ID); err != nil { + slog.Error("detect: write probe freq mark", "node_id", node.ID, "error", err) + } + if err := e.lc.BumpVersion(ctx); err != nil { + slog.Error("detect: bump version after suspect", "node_id", node.ID, "error", err) + } + + // Reset fail streak; suspect streak begins on the next call to + // processSuspect. + sk.FailStreak = 0 + sk.SuspectStreak = 0 + sk.RecoverStreak = 0 + } else { + // Below the threshold: increment and wait for next cycle. + sk.FailStreak++ + sk.RecoverStreak = 0 + } + } else { + // Domestic probes OK (or overseas data absent / also failing — fault is + // handled above before this function is called). + sk.FailStreak = 0 + sk.RecoverStreak = 0 + } + + return e.streaks.Save(ctx, node.ID, sk) +} + +// processSuspect applies the confirmed and recover rules to a "blocked_suspect" node. +func (e *Engine) processSuspect(ctx context.Context, node NodeInfo, sig NodeSignal, sk Streak) error { + domesticFailing := e.cfg.isSuspectTriggered(sig) + overseasBarrier := sig.OverseasHasData && sig.OverseasOK + + if domesticFailing && overseasBarrier { + // Still looks like a GFW block: advance the suspect streak. + sk.SuspectStreak++ + sk.RecoverStreak = 0 + sk.FailStreak = 0 + + if sk.SuspectStreak >= e.cfg.ConfirmedStreakMin { + // ── Rule 3: confirmed ───────────────────────────────────────────── + confirmedDetail := map[string]any{ + "from": "blocked_suspect", + "to": "blocked_confirmed", + "suspect_streak": sk.SuspectStreak, + } + n1, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedSuspect, StatusBlockedConfirmed, confirmedDetail) + if err != nil { + return fmt.Errorf("transition suspect→confirmed: %w", err) + } + if n1 == 0 { + slog.Info("detect: confirmed transition skipped (lock conflict)", "node_id", node.ID) + return e.streaks.Save(ctx, node.ID, sk) + } + + // Immediately mark down — skip draining per lifecycle policy. + downDetail := map[string]any{ + "from": "blocked_confirmed", + "to": "down", + "note": "skipped draining (confirmed block policy)", + } + if _, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedConfirmed, StatusDown, downDetail); err != nil { + return fmt.Errorf("transition confirmed→down: %w", err) + } + + // Push to 15E replace queue. + if err := e.pushReplaceQueue(ctx, node.ID); err != nil { + slog.Error("detect: push replace queue", "node_id", node.ID, "error", err) + } + + // Reset all streaks; the node is now down and will not be processed + // again (ListNodes only returns "up" and "blocked_suspect"). + sk = Streak{} + } + } else { + // Domestic probes passing (or overseas data absent — treat as recovery). + // ── Rule 4: recover ─────────────────────────────────────────────────── + sk.RecoverStreak++ + sk.SuspectStreak = 0 + sk.FailStreak = 0 + + if sk.RecoverStreak >= e.cfg.RecoverStreakMin { + recoverDetail := map[string]any{ + "from": "blocked_suspect", + "to": "up", + "recover_streak": sk.RecoverStreak, + } + affected, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedSuspect, StatusUp, recoverDetail) + if err != nil { + return fmt.Errorf("transition suspect→up: %w", err) + } + if affected == 0 { + slog.Info("detect: recover transition skipped (lock conflict)", "node_id", node.ID) + return e.streaks.Save(ctx, node.ID, sk) + } + // Weight recovery is delegated to 15E's warmup/gradual-ramp channel. + // The engine intentionally does NOT call SetWeight here. + sk = Streak{} + } + } + + return e.streaks.Save(ctx, node.ID, sk) +} + +// ───────────────────────────────────────────────────────────────────────────── +// Helpers +// ───────────────────────────────────────────────────────────────────────────── + +// writeProbeFreqMark sets a Redis key that tells 15B to probe this node at +// 1-minute instead of the default 5-minute interval while it is suspect. +func (e *Engine) writeProbeFreqMark(ctx context.Context, nodeID string) error { + key := probeFreqKeyPrefix + nodeID + return e.rdb.Set(ctx, key, "60", probeFreqTTL).Err() +} + +// replaceRequest is the payload pushed to the 15E replace queue. +type replaceRequest struct { + NodeID string `json:"nodeId"` + ReplacementUUID string `json:"replacementUuid"` +} + +// pushReplaceQueue enqueues a replacement request for 15E (Replenisher). +// The replacement UUID is pre-allocated here so 15E can reference it without +// generating its own ID. +func (e *Engine) pushReplaceQueue(ctx context.Context, nodeID string) error { + payload, err := json.Marshal(replaceRequest{ + NodeID: nodeID, + ReplacementUUID: idgen.NewString(), + }) + if err != nil { + return fmt.Errorf("detect: marshal replace request: %w", err) + } + return e.rdb.LPush(ctx, replaceQueueKey, payload).Err() +} diff --git a/server/internal/scheduler/detect/engine_test.go b/server/internal/scheduler/detect/engine_test.go new file mode 100644 index 0000000..c0decc1 --- /dev/null +++ b/server/internal/scheduler/detect/engine_test.go @@ -0,0 +1,653 @@ +package detect_test + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + + "github.com/wangjia/pangolin/server/internal/scheduler/detect" + "github.com/wangjia/pangolin/server/internal/scheduler/probe" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Test helpers +// ───────────────────────────────────────────────────────────────────────────── + +// mockSnapshotter implements detect.ProbeSnapshotter for unit tests. +type mockSnapshotter struct { + data map[string]map[string]probe.ProbeSnapshot +} + +func (m *mockSnapshotter) SnapshotsByNode(_ context.Context, nodeID string) (map[string]probe.ProbeSnapshot, error) { + if snaps, ok := m.data[nodeID]; ok { + return snaps, nil + } + return nil, nil +} + +// recordingNotifier records NotifyFault calls for assertion. +type recordingNotifier struct { + calls []string // nodeID values +} + +func (r *recordingNotifier) NotifyFault(_ context.Context, nodeID, _ string) error { + r.calls = append(r.calls, nodeID) + return nil +} + +// newTestRedis creates an in-process Redis (miniredis) and returns a connected +// client plus a cleanup function. Tests must call cleanup() at the end. +func newTestRedis(t *testing.T) (*redis.Client, *miniredis.Miniredis) { + t.Helper() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + return rdb, mr +} + +// newEngine wires up an Engine with the given mock components. +func newEngine( + t *testing.T, + rdb *redis.Client, + lc *detect.MockLifecycle, + snaps *mockSnapshotter, + notifier detect.Notifier, + cfg *detect.DetectConfig, +) *detect.Engine { + t.Helper() + streaks := detect.NewStreakStore(rdb) + return detect.NewEngine(snaps, lc, streaks, rdb, notifier, cfg) +} + +// snap builds a probe.ProbeSnapshot for one (node, vantage) pair. +func snap(country, isp string, l1OK bool, l3OK *bool) probe.ProbeSnapshot { + r := probe.NodeReport{L1: probe.L1Result{OK: l1OK}} + if l3OK != nil { + r.L3 = &probe.L3Result{OK: *l3OK} + } + return probe.ProbeSnapshot{ + Vantage: probe.VantagePoint{Country: country, ISP: isp}, + Report: r, + } +} + +func boolPtr(b bool) *bool { return &b } + +// cnFail returns a domestic (CN) probe snapshot where L3 is failing. +func cnFail(isp string) probe.ProbeSnapshot { + return snap("CN", isp, true, boolPtr(false)) +} + +// cnOK returns a domestic (CN) probe snapshot where L3 passes. +func cnOK(isp string) probe.ProbeSnapshot { + return snap("CN", isp, true, boolPtr(true)) +} + +// overseasOK returns an overseas probe snapshot where L1 passes. +func overseasOK() probe.ProbeSnapshot { + return snap("SG", "AWS", true, nil) +} + +// overseasFail returns an overseas probe snapshot where L1 fails. +func overseasFail() probe.ProbeSnapshot { + return snap("SG", "AWS", false, nil) +} + +// snapsForNode builds the map that mockSnapshotter.data["nodeID"] expects. +func snapsForNode(snaps ...probe.ProbeSnapshot) map[string]probe.ProbeSnapshot { + m := make(map[string]probe.ProbeSnapshot, len(snaps)) + for i, s := range snaps { + key := s.Vantage.Country + ":" + s.Vantage.ISP + _ = i + m[key] = s + } + return m +} + +// ───────────────────────────────────────────────────────────────────────────── +// Core rule tests (table-driven) +// ───────────────────────────────────────────────────────────────────────────── + +const nodeID = "node-001" + +// TestRules covers the five classification rules and their boundary conditions +// through table-driven subtests. +func TestRules(t *testing.T) { + cfg := detect.DefaultConfig() + ctx := context.Background() + + type tick struct { + // snaps to present on this tick (keyed by node ID) + snaps map[string]probe.ProbeSnapshot + } + + tests := []struct { + name string + // initial node status before any ticks + initialStatus detect.NodeStatus + // ticks is the sequence of probe states to feed to the engine + ticks []tick + // wantStatus is the expected node status after all ticks + wantStatus detect.NodeStatus + // wantEvents are (From, To) pairs expected in the lifecycle event log + wantEvents [][2]detect.NodeStatus + // wantFaultNotified indicates whether NotifyFault should have been called + wantFaultNotified bool + }{ + { + // Exactly 2 out of 3 ISPs fail → meets the ≥2/3 threshold. + // After 2 consecutive failing cycles (10 min) the node becomes suspect. + name: "exactly_2_of_3_ISPs_fail_after_2_cycles", + initialStatus: detect.StatusUp, + ticks: []tick{ + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusBlockedSuspect, + wantEvents: [][2]detect.NodeStatus{ + {detect.StatusUp, detect.StatusBlockedSuspect}, + }, + }, + { + // 1 out of 3 ISPs fail → below the 2/3 threshold → no transition. + name: "1_of_3_ISPs_fail_no_suspect", + initialStatus: detect.StatusUp, + ticks: []tick{ + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnOK("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnOK("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusUp, + wantEvents: nil, + }, + { + // 9 min (= 1 complete 5-min cycle): fail_streak reaches 1, not yet 2. + // No transition expected after exactly one failing tick. + name: "9min_one_cycle_no_transition", + initialStatus: detect.StatusUp, + ticks: []tick{ + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusUp, + wantEvents: nil, + }, + { + // 10 min (= 2 complete 5-min cycles): fail_streak reaches 2 ≥ SuspectStreakMin. + // Transition to blocked_suspect expected. + name: "10min_two_cycles_triggers_suspect", + initialStatus: detect.StatusUp, + ticks: []tick{ + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusBlockedSuspect, + wantEvents: [][2]detect.NodeStatus{ + {detect.StatusUp, detect.StatusBlockedSuspect}, + }, + }, + { + // Traffic warning: drop ≥ 80 % + baseline ≥ 20 → effectiveSuspectStreakMin = 1. + // A single failing tick is enough to enter blocked_suspect. + name: "traffic_warning_relaxes_threshold_to_1_cycle", + initialStatus: detect.StatusUp, + ticks: []tick{ + // The MockLifecycle will have loads set before running this test; + // we pass snaps with 2/3 ISPs failing + overseas OK. + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusBlockedSuspect, + wantEvents: [][2]detect.NodeStatus{ + {detect.StatusUp, detect.StatusBlockedSuspect}, + }, + }, + { + // Confirmed: node spends 6 consecutive cycles in blocked_suspect + // with domestic still failing → transitions to blocked_confirmed → down. + name: "suspect_6_cycles_triggers_confirmed", + initialStatus: detect.StatusBlockedSuspect, + ticks: []tick{ + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusDown, + wantEvents: [][2]detect.NodeStatus{ + {detect.StatusBlockedSuspect, detect.StatusBlockedConfirmed}, + {detect.StatusBlockedConfirmed, detect.StatusDown}, + }, + }, + { + // Fault: domestic + overseas both fail → fault rule fires, no state + // transition, NotifyFault is called exactly once. + name: "domestic_and_overseas_fail_triggers_fault_not_block", + initialStatus: detect.StatusUp, + ticks: []tick{ + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), overseasFail())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), overseasFail())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), overseasFail())}, + }, + wantStatus: detect.StatusUp, // no transition + wantEvents: nil, // no lifecycle transitions + wantFaultNotified: true, + }, + { + // Recover: 2 consecutive cycles with domestic passing while in + // blocked_suspect → transitions back to up. + name: "recover_2_cycles_back_to_up", + initialStatus: detect.StatusBlockedSuspect, + ticks: []tick{ + {snaps: snapsForNode(cnOK("ChinaTelecom"), cnOK("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnOK("ChinaTelecom"), cnOK("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusUp, + wantEvents: [][2]detect.NodeStatus{ + {detect.StatusBlockedSuspect, detect.StatusUp}, + }, + }, + { + // Recovery not yet reached: 1 cycle passing while in blocked_suspect + // → still in blocked_suspect. + name: "recover_1_cycle_not_yet_recovered", + initialStatus: detect.StatusBlockedSuspect, + ticks: []tick{ + {snaps: snapsForNode(cnOK("ChinaTelecom"), cnOK("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusBlockedSuspect, + wantEvents: nil, + }, + { + // Third-party ("3rd-") ISP prefix is stripped before grouping: + // "3rd-ChinaTelecom" and "ChinaTelecom" must count as the same ISP. + // Two ISPs fail out of three → triggers suspect. + name: "third_party_isp_prefix_merged", + initialStatus: detect.StatusUp, + ticks: []tick{ + {snaps: snapsForNode(cnFail("3rd-ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("3rd-ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusBlockedSuspect, + wantEvents: [][2]detect.NodeStatus{ + {detect.StatusUp, detect.StatusBlockedSuspect}, + }, + }, + { + // L3 failure overrides L1/L2: even if L1 is OK the ISP is counted + // as failing when L3 is explicitly false. + name: "l3_failure_overrides_l1_l2", + initialStatus: detect.StatusUp, + ticks: []tick{ + // L1 passes but L3 fails for two ISPs → ≥2/3 threshold met. + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + {snaps: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK())}, + }, + wantStatus: detect.StatusBlockedSuspect, + wantEvents: [][2]detect.NodeStatus{ + {detect.StatusUp, detect.StatusBlockedSuspect}, + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + rdb, _ := newTestRedis(t) + + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: tc.initialStatus, Weight: 100}, + }) + + // For the traffic-warning test, set up load history showing a large drop. + // Older half: 100 connections. Newer half: 20 connections. + // Drop = 80 % ≥ TrafficDropThreshold (80 %). + // Current baseline = 20 ≥ TrafficBaselineMin (20) → warning applies. + if tc.name == "traffic_warning_relaxes_threshold_to_1_cycle" { + now := time.Now().Unix() + lc.SetLoads(nodeID, []detect.LoadPoint{ + {Timestamp: now - 14*60, Online: 100}, + {Timestamp: now - 10*60, Online: 100}, + {Timestamp: now - 5*60, Online: 20}, + {Timestamp: now - 60, Online: 20}, + }) + } + + notifier := &recordingNotifier{} + snapper := &mockSnapshotter{data: make(map[string]map[string]probe.ProbeSnapshot)} + eng := newEngine(t, rdb, lc, snapper, notifier, &cfg) + + for i, tick := range tc.ticks { + snapper.data[nodeID] = tick.snaps + if err := eng.Tick(ctx); err != nil { + t.Fatalf("tick %d: Tick() error: %v", i, err) + } + } + + // Verify final node status. + if got := lc.NodeStatus(nodeID); got != tc.wantStatus { + t.Errorf("node status = %q, want %q", got, tc.wantStatus) + } + + // Verify lifecycle events (from/to pairs only; detail checked separately). + events := lc.Events() + if len(events) != len(tc.wantEvents) { + t.Errorf("got %d events, want %d: %v", len(events), len(tc.wantEvents), events) + } else { + for i, ev := range events { + want := tc.wantEvents[i] + if ev.From != want[0] || ev.To != want[1] { + t.Errorf("event[%d]: from=%q to=%q, want from=%q to=%q", + i, ev.From, ev.To, want[0], want[1]) + } + } + } + + // Verify fault notification. + if tc.wantFaultNotified && len(notifier.calls) == 0 { + t.Error("expected NotifyFault to be called, but it was not") + } + if !tc.wantFaultNotified && len(notifier.calls) > 0 { + t.Errorf("unexpected NotifyFault calls: %v", notifier.calls) + } + }) + } +} + +// TestEventDetailContainsFromTo verifies that every TransitionStatus call +// includes "from" and "to" keys in its detail map (node_events audit trail). +func TestEventDetailContainsFromTo(t *testing.T) { + ctx := context.Background() + cfg := detect.DefaultConfig() + rdb, _ := newTestRedis(t) + + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: detect.StatusUp, Weight: 100}, + }) + snapper := &mockSnapshotter{ + data: map[string]map[string]probe.ProbeSnapshot{ + nodeID: snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK()), + }, + } + eng := newEngine(t, rdb, lc, snapper, nil, &cfg) + + // Two ticks to trigger suspect. + if err := eng.Tick(ctx); err != nil { + t.Fatal(err) + } + if err := eng.Tick(ctx); err != nil { + t.Fatal(err) + } + + events := lc.Events() + if len(events) == 0 { + t.Fatal("no events recorded") + } + for _, ev := range events { + if _, ok := ev.Detail["from"]; !ok { + t.Errorf("event %v→%v detail missing 'from'", ev.From, ev.To) + } + if _, ok := ev.Detail["to"]; !ok { + t.Errorf("event %v→%v detail missing 'to'", ev.From, ev.To) + } + } +} + +// TestOptimisticLockConflict verifies that when TransitionStatus returns 0 +// (lock conflict), the current cycle is abandoned without any side-effects, +// and the engine is idempotent — the next tick re-attempts cleanly. +func TestOptimisticLockConflict(t *testing.T) { + ctx := context.Background() + cfg := detect.DefaultConfig() + rdb, _ := newTestRedis(t) + + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: detect.StatusUp, Weight: 100}, + }) + // Register a conflict for the up→suspect transition. + lc.SetConflict(nodeID, detect.StatusUp, detect.StatusBlockedSuspect) + + failSnaps := snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK()) + snapper := &mockSnapshotter{data: map[string]map[string]probe.ProbeSnapshot{nodeID: failSnaps}} + eng := newEngine(t, rdb, lc, snapper, nil, &cfg) + + // Run two ticks — normally enough to trigger suspect. + if err := eng.Tick(ctx); err != nil { + t.Fatal(err) + } + if err := eng.Tick(ctx); err != nil { + t.Fatal(err) + } + + // Conflict registered: status must remain "up", no events logged. + if got := lc.NodeStatus(nodeID); got != detect.StatusUp { + t.Errorf("status = %q after conflict; want %q", got, detect.StatusUp) + } + if events := lc.Events(); len(events) != 0 { + t.Errorf("expected 0 events after conflict, got %d: %v", len(events), events) + } + + // Clear conflict, run one more tick → should transition now (streak already at 1). + lc.ClearConflict(nodeID, detect.StatusUp, detect.StatusBlockedSuspect) + if err := eng.Tick(ctx); err != nil { + t.Fatal(err) + } + if got := lc.NodeStatus(nodeID); got != detect.StatusBlockedSuspect { + t.Errorf("status = %q after conflict cleared; want %q", got, detect.StatusBlockedSuspect) + } +} + +// TestStreakPersistenceAcrossRestart verifies that streak counters survive a +// simulated process restart: a new Engine reading the same Redis should +// continue from the saved streak rather than starting from zero. +func TestStreakPersistenceAcrossRestart(t *testing.T) { + ctx := context.Background() + cfg := detect.DefaultConfig() // SuspectStreakMin = 2 + + rdb, _ := newTestRedis(t) + + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: detect.StatusUp, Weight: 100}, + }) + failSnaps := snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK()) + snapper := &mockSnapshotter{data: map[string]map[string]probe.ProbeSnapshot{nodeID: failSnaps}} + + // First "process instance": run one tick (fail_streak becomes 1). + eng1 := newEngine(t, rdb, lc, snapper, nil, &cfg) + if err := eng1.Tick(ctx); err != nil { + t.Fatal(err) + } + if got := lc.NodeStatus(nodeID); got != detect.StatusUp { + t.Errorf("after tick 1 status = %q; want up", got) + } + + // "Restart": create a brand-new Engine pointing at the same Redis. + // The streak (fail_streak=1) must still be present. + eng2 := newEngine(t, rdb, lc, snapper, nil, &cfg) + if err := eng2.Tick(ctx); err != nil { + t.Fatal(err) + } + + // The second engine picks up fail_streak=1, increments to 2, and fires + // the suspect transition. + if got := lc.NodeStatus(nodeID); got != detect.StatusBlockedSuspect { + t.Errorf("after restart tick status = %q; want blocked_suspect", got) + } +} + +// TestReplaceQueuePopulated verifies that when a node is confirmed, the +// detect:replace:queue Redis list receives a valid JSON entry. +func TestReplaceQueuePopulated(t *testing.T) { + ctx := context.Background() + cfg := detect.DefaultConfig() + + rdb, _ := newTestRedis(t) + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: detect.StatusBlockedSuspect, Weight: 10}, + }) + failSnaps := snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK()) + snapper := &mockSnapshotter{data: map[string]map[string]probe.ProbeSnapshot{nodeID: failSnaps}} + eng := newEngine(t, rdb, lc, snapper, nil, &cfg) + + // Run ConfirmedStreakMin (6) ticks. + for i := 0; i < cfg.ConfirmedStreakMin; i++ { + if err := eng.Tick(ctx); err != nil { + t.Fatalf("tick %d: %v", i, err) + } + } + + if got := lc.NodeStatus(nodeID); got != detect.StatusDown { + t.Errorf("node status = %q after 6 suspect cycles; want down", got) + } + + // Verify the replace queue has one entry with the correct shape. + qlen, err := rdb.LLen(ctx, "detect:replace:queue").Result() + if err != nil { + t.Fatalf("LLEN: %v", err) + } + if qlen != 1 { + t.Fatalf("replace queue length = %d; want 1", qlen) + } + + raw, err := rdb.LIndex(ctx, "detect:replace:queue", 0).Result() + if err != nil { + t.Fatalf("LINDEX: %v", err) + } + var entry struct { + NodeID string `json:"nodeId"` + ReplacementUUID string `json:"replacementUuid"` + } + if err := json.Unmarshal([]byte(raw), &entry); err != nil { + t.Fatalf("unmarshal replace queue entry: %v", err) + } + if entry.NodeID != nodeID { + t.Errorf("replace queue nodeId = %q; want %q", entry.NodeID, nodeID) + } + if entry.ReplacementUUID == "" { + t.Error("replace queue replacementUuid is empty") + } +} + +// TestProbeFreqMarkWritten verifies that the probe:freq:{nodeID} Redis key is +// set when a node transitions to blocked_suspect. +func TestProbeFreqMarkWritten(t *testing.T) { + ctx := context.Background() + cfg := detect.DefaultConfig() + + rdb, _ := newTestRedis(t) + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: detect.StatusUp, Weight: 100}, + }) + failSnaps := snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK()) + snapper := &mockSnapshotter{data: map[string]map[string]probe.ProbeSnapshot{nodeID: failSnaps}} + eng := newEngine(t, rdb, lc, snapper, nil, &cfg) + + // Two ticks to trigger suspect. + for i := 0; i < 2; i++ { + if err := eng.Tick(ctx); err != nil { + t.Fatalf("tick %d: %v", i, err) + } + } + + if got := lc.NodeStatus(nodeID); got != detect.StatusBlockedSuspect { + t.Errorf("status = %q; want blocked_suspect", got) + } + + freqKey := "probe:freq:" + nodeID + val, err := rdb.Get(ctx, freqKey).Result() + if err != nil { + t.Fatalf("probe freq key not found: %v", err) + } + if val != "60" { + t.Errorf("probe:freq value = %q; want 60", val) + } +} + +// TestFaultNoTransitionNoStreak verifies that the fault rule leaves the node +// status and streaks completely unchanged across multiple ticks, even though +// the domestic probes are failing. +func TestFaultNoTransitionNoStreak(t *testing.T) { + ctx := context.Background() + cfg := detect.DefaultConfig() + + rdb, _ := newTestRedis(t) + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: detect.StatusUp, Weight: 100}, + }) + // Both domestic and overseas fail → fault. + faultSnaps := snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), overseasFail()) + snapper := &mockSnapshotter{data: map[string]map[string]probe.ProbeSnapshot{nodeID: faultSnaps}} + notifier := &recordingNotifier{} + eng := newEngine(t, rdb, lc, snapper, notifier, &cfg) + + // Run many ticks — should never transition. + for i := 0; i < 10; i++ { + if err := eng.Tick(ctx); err != nil { + t.Fatalf("tick %d: %v", i, err) + } + } + + if got := lc.NodeStatus(nodeID); got != detect.StatusUp { + t.Errorf("status = %q; want up (fault must not transition)", got) + } + if events := lc.Events(); len(events) != 0 { + t.Errorf("unexpected events: %v", events) + } + if len(notifier.calls) == 0 { + t.Error("expected NotifyFault to be called at least once") + } +} + +// TestSuspectWeightApplied verifies that when a node enters blocked_suspect, +// its routing weight is reduced to cfg.SuspectWeight (default 10). +func TestSuspectWeightApplied(t *testing.T) { + ctx := context.Background() + cfg := detect.DefaultConfig() + + rdb, _ := newTestRedis(t) + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: detect.StatusUp, Weight: 100}, + }) + failSnaps := snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile"), overseasOK()) + snapper := &mockSnapshotter{data: map[string]map[string]probe.ProbeSnapshot{nodeID: failSnaps}} + eng := newEngine(t, rdb, lc, snapper, nil, &cfg) + + for i := 0; i < 2; i++ { + if err := eng.Tick(ctx); err != nil { + t.Fatalf("tick %d: %v", i, err) + } + } + + if got := lc.NodeWeight(nodeID); got != cfg.SuspectWeight { + t.Errorf("weight after suspect = %d; want %d", got, cfg.SuspectWeight) + } +} + +// TestNoOverseasDataNoSuspect verifies that when no overseas probe data is +// available, the engine does not trigger the suspect rule (we cannot confirm +// the node is globally reachable, so we cannot attribute a domestic failure +// to GFW censorship). +func TestNoOverseasDataNoSuspect(t *testing.T) { + ctx := context.Background() + cfg := detect.DefaultConfig() + + rdb, _ := newTestRedis(t) + lc := detect.NewMockLifecycle([]detect.NodeInfo{ + {ID: nodeID, Status: detect.StatusUp, Weight: 100}, + }) + // Only domestic probes, no overseas vantage. + domesticOnlySnaps := snapsForNode(cnFail("ChinaTelecom"), cnFail("ChinaUnicom"), cnOK("ChinaMobile")) + snapper := &mockSnapshotter{data: map[string]map[string]probe.ProbeSnapshot{nodeID: domesticOnlySnaps}} + eng := newEngine(t, rdb, lc, snapper, nil, &cfg) + + for i := 0; i < 5; i++ { + if err := eng.Tick(ctx); err != nil { + t.Fatalf("tick %d: %v", i, err) + } + } + + if got := lc.NodeStatus(nodeID); got != detect.StatusUp { + t.Errorf("status = %q; want up (no overseas data → no suspect)", got) + } +} diff --git a/server/internal/scheduler/detect/lifecycle.go b/server/internal/scheduler/detect/lifecycle.go new file mode 100644 index 0000000..e92260d --- /dev/null +++ b/server/internal/scheduler/detect/lifecycle.go @@ -0,0 +1,288 @@ +// 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 +} diff --git a/server/internal/scheduler/detect/rules.go b/server/internal/scheduler/detect/rules.go new file mode 100644 index 0000000..ca31565 --- /dev/null +++ b/server/internal/scheduler/detect/rules.go @@ -0,0 +1,101 @@ +package detect + +// DetectConfig holds all tunable thresholds for the detection engine. +// All fields are filled with production defaults by DefaultConfig(). +// 15F (hot-reload) may overwrite fields at runtime; the Engine reads via a +// pointer so updates take effect on the next Tick without a restart. +type DetectConfig struct { + // DomesticFailNumerator and DomesticFailDenominator define the ISP-failure + // fraction that triggers the suspect rule. + // Default: 2/3 (two out of every three domestic ISPs must fail). + DomesticFailNumerator int // 2 + DomesticFailDenominator int // 3 + + // SuspectStreakMin is the number of consecutive failing cycles required to + // transition a node from "up" to "blocked_suspect". + // Default: 2 cycles (10 min at the 5-min tick period). + SuspectStreakMin int + + // TrafficDropThreshold is the minimum percentage drop in online-connection + // count over the 15-minute window that activates the traffic-warning rule. + // Default: 80.0 %. + TrafficDropThreshold float64 + + // TrafficBaselineMin is the minimum current online-connection count for + // the traffic-warning relaxation to apply. Below this threshold the node + // may simply be idle rather than experiencing user flight. + // Default: 20 connections. + TrafficBaselineMin int + + // ConfirmedStreakMin is the number of consecutive cycles that a node must + // spend in "blocked_suspect" before being promoted to "blocked_confirmed". + // Default: 6 cycles (30 min). + ConfirmedStreakMin int + + // RecoverStreakMin is the number of consecutive passing cycles while in + // "blocked_suspect" required to recover the node back to "up". + // Default: 2 cycles (10 min). + RecoverStreakMin int + + // SuspectWeight is the routing weight applied when a node first enters + // "blocked_suspect", reducing traffic directed to it. + // Default: 10. + SuspectWeight int +} + +// DefaultConfig returns a DetectConfig pre-filled with production defaults. +func DefaultConfig() DetectConfig { + return DetectConfig{ + DomesticFailNumerator: 2, + DomesticFailDenominator: 3, + SuspectStreakMin: 2, + TrafficDropThreshold: 80.0, + TrafficBaselineMin: 20, + ConfirmedStreakMin: 6, + RecoverStreakMin: 2, + SuspectWeight: 10, + } +} + +// isSuspectTriggered reports whether the domestic ISP-failure rate meets or +// exceeds the configured fraction (default ≥ 2/3). +// +// Integer arithmetic avoids floating-point rounding: +// +// fail/total >= num/den ↔ fail × den >= total × num +func (cfg *DetectConfig) isSuspectTriggered(sig NodeSignal) bool { + if sig.DomesticTotalISPs == 0 { + return false // no domestic probe data this cycle; cannot judge + } + return sig.DomesticFailISPs*cfg.DomesticFailDenominator >= + sig.DomesticTotalISPs*cfg.DomesticFailNumerator +} + +// isFault reports whether the node appears to have a local outage rather than +// a domestic censorship event. +// +// Condition: domestic probes failing AND overseas probes also failing. +// When both sides are down the most likely cause is a node-level failure +// (hardware, network, crashed process) rather than GFW interference. +// +// The fault rule takes precedence over all other rules and must be evaluated +// first in the processing loop. +func isFault(sig NodeSignal) bool { + return sig.DomesticFailISPs > 0 && sig.OverseasHasData && !sig.OverseasOK +} + +// effectiveSuspectStreakMin returns the cycle threshold required to enter +// "blocked_suspect", relaxed to 1 when the traffic-warning rule is active. +func (cfg *DetectConfig) effectiveSuspectStreakMin(sig NodeSignal) int { + if cfg.isTrafficWarning(sig) { + return 1 + } + return cfg.SuspectStreakMin +} + +// isTrafficWarning reports whether the 15-minute traffic-drop signal exceeds +// the threshold and the baseline is large enough to be meaningful. +func (cfg *DetectConfig) isTrafficWarning(sig NodeSignal) bool { + return sig.TrafficDropPct >= cfg.TrafficDropThreshold && + sig.TrafficBaseline >= cfg.TrafficBaselineMin +} diff --git a/server/internal/scheduler/detect/signals.go b/server/internal/scheduler/detect/signals.go new file mode 100644 index 0000000..40f22a0 --- /dev/null +++ b/server/internal/scheduler/detect/signals.go @@ -0,0 +1,177 @@ +package detect + +import ( + "context" + "strings" + "time" + + "github.com/wangjia/pangolin/server/internal/scheduler/probe" +) + +// NodeSignal is the normalised set of detection signals for a single node +// produced by one Tick. +type NodeSignal struct { + NodeID string + + // DomesticTotalISPs is the count of distinct domestic (CN) ISPs for which + // probe data was available this cycle. Vantages with no data are excluded + // from both numerator and denominator per spec. + DomesticTotalISPs int + + // DomesticFailISPs is the count of domestic ISPs judged to be failing. + // An ISP is failing if ANY vantage from that ISP indicates a failure at + // the highest available layer (L3 > L2 > L1; see isReportFailed). + DomesticFailISPs int + + // OverseasHasData is true when at least one non-CN vantage reported this cycle. + OverseasHasData bool + + // OverseasOK is true when every overseas vantage passed L1 this cycle. + // Only meaningful when OverseasHasData is true. + OverseasOK bool + + // TrafficDropPct is the percentage drop in online-connection count computed + // by comparing the first half vs the second half of the 15-minute + // GetLoadHistory window. Zero when insufficient history is available. + TrafficDropPct float64 + + // TrafficBaseline is the online-connection count from the most recent load + // sample. Used by the traffic-warning rule's minimum-baseline guard. + TrafficBaseline int +} + +// ProbeSnapshotter lets the Engine read probe snapshots without depending on +// the concrete *probe.Store (facilitates mocking in unit tests). +type ProbeSnapshotter interface { + SnapshotsByNode(ctx context.Context, nodeID string) (map[string]probe.ProbeSnapshot, error) +} + +// computeSignals builds a NodeSignal for nodeID from the current probe +// snapshots and the 15-minute load history. +// +// ISP normalisation: third-party probes (15C) use a "3rd-" prefix on their +// ISP names (e.g. "3rd-ChinaTelecom"). This prefix is stripped before +// grouping so that first-party (15B) and third-party vantages from the same +// ISP are merged, per spec. +func computeSignals( + ctx context.Context, + nodeID string, + snapshots map[string]probe.ProbeSnapshot, + lc LifecycleService, +) NodeSignal { + sig := NodeSignal{NodeID: nodeID} + + // domesticISPs: normalised ISP name → whether any vantage from that ISP + // shows a failure this cycle. + domesticISPs := make(map[string]bool) + overseasTotal := 0 + overseasOKCount := 0 + + for _, snap := range snapshots { + v := snap.Vantage + r := snap.Report + + switch { + case v.Country == "CN": + // Domestic vantage. Normalise ISP name and merge failures: + // any failing vantage from the same ISP marks that ISP as failed. + isp := normaliseISP(v.ISP) + failed := isReportFailed(r) + if prev, seen := domesticISPs[isp]; seen { + domesticISPs[isp] = prev || failed + } else { + domesticISPs[isp] = failed + } + + case v.Country != "": + // Overseas vantage (any country other than CN). + overseasTotal++ + if r.L1.OK { + overseasOKCount++ + } + // Empty Country: vantage identity unknown – excluded from both pools. + } + } + + sig.DomesticTotalISPs = len(domesticISPs) + for _, failed := range domesticISPs { + if failed { + sig.DomesticFailISPs++ + } + } + sig.OverseasHasData = overseasTotal > 0 + sig.OverseasOK = overseasTotal > 0 && overseasOKCount == overseasTotal + + // Traffic drop: compare the older half of the 15-minute window to the + // newer half. Zero is returned when there is too little history. + history, err := lc.GetLoadHistory(ctx, nodeID, 15*time.Minute) + if err == nil && len(history) > 0 { + sig.TrafficBaseline = history[len(history)-1].Online + if len(history) >= 2 { + sig.TrafficDropPct = trafficDropPct(history) + } + } + + return sig +} + +// isReportFailed determines whether a single NodeReport indicates a failure +// at the highest available measurement layer. +// +// Priority (highest to lowest): +// - L3 present → L3.OK is the authoritative verdict. +// A failing L3 is recorded even when L1/L2 passed (e.g. DPI/SNI reset +// at the application layer after a successful TCP/TLS handshake). +// - L3 absent → check L1 then L2. +func isReportFailed(r probe.NodeReport) bool { + if r.L3 != nil { + return !r.L3.OK + } + if !r.L1.OK { + return true + } + if r.L2 != nil && !r.L2.OK { + return true + } + return false +} + +// normaliseISP strips the "3rd-" prefix that AliyunSyntheticAgent (15C) adds +// to distinguish third-party vantages, yielding the canonical ISP name used +// for grouping. +func normaliseISP(isp string) string { + return strings.TrimPrefix(isp, "3rd-") +} + +// trafficDropPct computes the percentage drop between the first and second +// halves of the load-history slice (oldest → newest). +// Returns 0 when the baseline (first-half average) is zero or when the +// second half average is higher (traffic increased). +func trafficDropPct(history []LoadPoint) float64 { + n := len(history) + if n < 2 { + return 0 + } + half := n / 2 + older := avgOnline(history[:half]) + newer := avgOnline(history[half:]) + if older == 0 { + return 0 + } + drop := older - newer + if drop <= 0 { + return 0 + } + return float64(drop) / float64(older) * 100 +} + +func avgOnline(pts []LoadPoint) int { + if len(pts) == 0 { + return 0 + } + sum := 0 + for _, p := range pts { + sum += p.Online + } + return sum / len(pts) +} diff --git a/server/internal/scheduler/detect/streak.go b/server/internal/scheduler/detect/streak.go new file mode 100644 index 0000000..9664874 --- /dev/null +++ b/server/internal/scheduler/detect/streak.go @@ -0,0 +1,101 @@ +package detect + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + // streakTTL is the Redis TTL for a per-node streak key. + // + // Must be comfortably wider than the longest streak window, which is the + // confirmed threshold: 6 cycles × 5 min = 30 min. Two hours provides + // enough margin to survive a process restart or a short leader-election gap + // without zeroing out a valid streak mid-observation. + streakTTL = 2 * time.Hour +) + +// Streak holds the consecutive-cycle counters for a single node. +// All three counters are stored together under one Redis key to make the +// read-modify-write loop atomic with a single GET + SET round-trip. +type Streak struct { + // FailStreak is the number of consecutive detection cycles in which the + // node's domestic probes failed while the node was in "up" state. + // It is reset to zero once the suspect threshold is reached (transition + // attempted) or when domestic probes recover. + FailStreak int `json:"fail_streak"` + + // SuspectStreak counts consecutive cycles spent in "blocked_suspect" where + // the GFW-block condition (domestic fail + overseas OK) still holds. + // Reaching ConfirmedStreakMin triggers the confirmed transition. + SuspectStreak int `json:"suspect_streak"` + + // RecoverStreak counts consecutive cycles in "blocked_suspect" where + // domestic probes pass. Reaching RecoverStreakMin triggers recovery. + RecoverStreak int `json:"recover_streak"` +} + +// StreakStore persists per-node Streak values in Redis. +// +// Key format: detect:streak:{nodeID} (JSON-encoded Streak, TTL 2 h) +// +// Missing-key semantics: a missing key means "no streak data" and is treated +// as a zero Streak. After a process restart, the first Tick loads whichever +// streaks survived in Redis and continues from that point — streaks are NOT +// re-derived from probe history on restart, which keeps the restart path +// simple at the cost of a potential single-cycle window of uncertainty. +type StreakStore struct { + rdb *redis.Client +} + +// NewStreakStore creates a StreakStore backed by the given Redis client. +func NewStreakStore(rdb *redis.Client) *StreakStore { + return &StreakStore{rdb: rdb} +} + +// streakKey returns the Redis key for nodeID's streak data. +func streakKey(nodeID string) string { + return "detect:streak:" + nodeID +} + +// Load reads the current Streak for nodeID. +// A missing or corrupted key returns a zero Streak without error. +func (s *StreakStore) Load(ctx context.Context, nodeID string) (Streak, error) { + val, err := s.rdb.Get(ctx, streakKey(nodeID)).Result() + if err == redis.Nil { + return Streak{}, nil + } + if err != nil { + return Streak{}, fmt.Errorf("detect: streak get %s: %w", nodeID, err) + } + var sk Streak + if err := json.Unmarshal([]byte(val), &sk); err != nil { + // Corrupted entry: treat as zero (non-fatal; we overwrite on the next Save). + return Streak{}, nil + } + return sk, nil +} + +// Save persists sk for nodeID with the standard streakTTL. +func (s *StreakStore) Save(ctx context.Context, nodeID string, sk Streak) error { + data, err := json.Marshal(sk) + if err != nil { + return fmt.Errorf("detect: streak marshal %s: %w", nodeID, err) + } + if err := s.rdb.Set(ctx, streakKey(nodeID), data, streakTTL).Err(); err != nil { + return fmt.Errorf("detect: streak set %s: %w", nodeID, err) + } + return nil +} + +// Reset deletes the streak key for nodeID (e.g. after a confirmed transition). +func (s *StreakStore) Reset(ctx context.Context, nodeID string) error { + if err := s.rdb.Del(ctx, streakKey(nodeID)).Err(); err != nil { + return fmt.Errorf("detect: streak del %s: %w", nodeID, err) + } + return nil +} diff --git a/server/migrations/000010_node_blocked_statuses.down.sql b/server/migrations/000010_node_blocked_statuses.down.sql new file mode 100644 index 0000000..a6c8b44 --- /dev/null +++ b/server/migrations/000010_node_blocked_statuses.down.sql @@ -0,0 +1,7 @@ +-- Revert nodes.status ENUM to the original six values. +-- WARNING: any rows currently in 'blocked_suspect' or 'blocked_confirmed' +-- will be rejected; run only after verifying no nodes are in those states. +ALTER TABLE nodes + MODIFY COLUMN status + ENUM('provisioning','probing','up','draining','down','destroyed') + NOT NULL DEFAULT 'provisioning'; diff --git a/server/migrations/000010_node_blocked_statuses.up.sql b/server/migrations/000010_node_blocked_statuses.up.sql new file mode 100644 index 0000000..b4b414f --- /dev/null +++ b/server/migrations/000010_node_blocked_statuses.up.sql @@ -0,0 +1,13 @@ +-- Extend nodes.status ENUM to include the two intermediate blocking states +-- added by the detection engine (task 15D). +-- +-- blocked_suspect : domestic probes failing + overseas OK for ≥ 2 cycles +-- blocked_confirmed: suspect held for ≥ 6 cycles → immediately taken off-directory +-- +-- Note: MySQL ENUM modification rebuilds the table in-place for InnoDB; +-- on production, run during a low-traffic window. +ALTER TABLE nodes + MODIFY COLUMN status + ENUM('provisioning','probing','up','draining','down','destroyed', + 'blocked_suspect','blocked_confirmed') + NOT NULL DEFAULT 'provisioning';