merge: 判定引擎 signals+rules+streak [tsk_OYEiDCzM9_0Y]

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()
+}

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)
+	}
+}

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

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

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)
+}

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

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';

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';