feat(scheduler): lifecycle 真实接线适配器(2/2) + 装配 BuildRealConfig

把 scheduler 的 stub lifecycle 换成 SQL 后端真实实现,并接入 server:
- SQLLifecycle：MySQL 后端,同时满足 detect.LifecycleService 与
  orchestrate.LifecycleService(两个 thin adapter 包装共享核心)。
  - TransitionStatus：乐观锁 UPDATE nodes SET status WHERE id AND status=from
    → rows affected(1=成功/0=冲突 no-op)+ 写 node_events(按状态映射 enum,
    eventForStatus 纯函数已单测)+ bump directory_version,全在一个事务内。
  - ListNodes/GetNode/SetWeight/BumpVersion/GetLoad(读 LoadCache)/WriteAuditLog。
  - 已知限制:无历史 load 表 → GetLoadHistory 返回当前点(掉量规则安全降级);
    destroy 的 cert 撤销留 TODO(provision 已拆 VM,撤销为纵深防御,待机群再接)。
- BuildRealConfig：用真实 lifecycle + provisionAdapter 装配三循环。
- main.go：SCHED_ENABLED 且有 DB 时走 BuildRealConfig(provision 无厂商凭证则
  CreateNode 优雅失败、替换保持 pending),否则回退 stub。
- 全量 server 23 包测试通过;e2e(判封→drain→换机→置备)待机群+厂商凭证验证。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

server/cmd/server/main.go

@@ -34,6 +34,8 @@ import (
 	"github.com/wangjia/pangolin/server/internal/mtls"
 	"github.com/wangjia/pangolin/server/internal/nodes"
 	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+	"github.com/wangjia/pangolin/server/internal/provision"
+	"github.com/wangjia/pangolin/server/internal/provision/providers"
 	"github.com/wangjia/pangolin/server/internal/redisutil"
 	"github.com/wangjia/pangolin/server/internal/scheduler"
 	"github.com/wangjia/pangolin/server/internal/scheduler/probe"
@@ -164,9 +166,29 @@ func main() {
 			if ps == nil {
 				ps = probe.NewStore(schedRDB)
 			}
-			cfg := scheduler.BuildStubConfig(schedRDB, ps)
-			// TODO(#5): replace stub lifecycle with real LifecycleService
-			// TODO(#14): replace stub provision with real ProvisionService
+			// Prefer real lifecycle (SQL) + provision (#14) wiring when a DB is
+			// available; fall back to no-op stubs otherwise. Without vendor
+			// credentials the provision service's CreateNode fails gracefully
+			// (pending replacements stay pending) while lifecycle reads/transitions
+			// still operate on real node rows.
+			var cfg scheduler.Config
+			if sqlDB != nil {
+				provStore := provision.NewMySQLStore(sqlDB)
+				provSvc, perr := provision.NewService(provision.Config{
+					Store:    provStore,
+					Adapters: providers.NewRegistry(),
+				})
+				if perr != nil {
+					log.Printf("scheduler: provision service init failed (%v) — using stub config", perr)
+					cfg = scheduler.BuildStubConfig(schedRDB, ps)
+				} else {
+					cfg = scheduler.BuildRealConfig(schedRDB, ps, sqlDB,
+						nodes.NewLoadCache(schedRDB), provSvc, provStore)
+					log.Printf("scheduler: real lifecycle + provision wired")
+				}
+			} else {
+				cfg = scheduler.BuildStubConfig(schedRDB, ps)
+			}
 			sched := scheduler.New(cfg)
 
 			// sigCtx cancels on SIGINT/SIGTERM, giving the scheduler up to
@@ -179,7 +201,7 @@ func main() {
 					slog.Error("scheduler: Start error", "error", err)
 				}
 			}()
-			log.Printf("scheduler started (SCHED_ENABLED=true, stub lifecycle/provision)")
+			log.Printf("scheduler started (SCHED_ENABLED=true)")
 		}
 	}
 

server/internal/scheduler/wiring.go

@@ -11,14 +11,63 @@ package scheduler
 
 import (
 	"context"
+	"database/sql"
 
 	"github.com/redis/go-redis/v9"
 
+	"github.com/wangjia/pangolin/server/internal/nodes"
 	"github.com/wangjia/pangolin/server/internal/scheduler/detect"
 	"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
 	"github.com/wangjia/pangolin/server/internal/scheduler/probe"
 )
 
+// BuildRealConfig wires the scheduler against the real lifecycle (SQL-backed)
+// and provision (#14) services, replacing the no-op stubs of BuildStubConfig.
+//
+//   - detect/orchestrate lifecycle → SQLLifecycle over the nodes table + LoadCache.
+//   - provision               → provisionAdapter over the real provision.Service.
+//
+// Without vendor credentials the provision service's CreateNode fails (a pending
+// replacement simply stays pending), but all lifecycle reads/transitions operate
+// on real node rows. Capacity monitoring is left disabled (nil) as in the stub.
+//
+// e2e behaviour (auto-detect → drain → replace → provision) is verifiable only
+// against a real node fleet with vendor provisioning configured.
+func BuildRealConfig(
+	rdb *redis.Client,
+	probeStore *probe.Store,
+	db *sql.DB,
+	loads *nodes.LoadCache,
+	provSvc provisionSvc,
+	provStore provisionResolver,
+) Config {
+	sqlLC := NewSQLLifecycle(db, loads)
+	detectLC := NewDetectLifecycle(sqlLC)
+	orchLC := NewOrchestrateLifecycle(sqlLC)
+	prov := NewProvisionAdapter(provSvc, provStore)
+
+	streaks := detect.NewStreakStore(rdb)
+	engine := detect.NewEngine(probeStore, detectLC, streaks, rdb, nil, nil)
+
+	replacer := orchestrate.NewReplacer(orchestrate.Config{
+		RDB:      rdb,
+		Prov:     prov,
+		LC:       orchLC,
+		Snaps:    probeStore,
+		Breaker:  nil, // StubBreaker default until 15F policy tuned
+		Notifier: nil, // LogNotifier default
+		Clock:    nil,
+	})
+	grayscale := orchestrate.NewGrayscale(rdb, orchLC, nil)
+
+	return Config{
+		RDB:       rdb,
+		Engine:    engine,
+		Replacer:  replacer,
+		Grayscale: grayscale,
+	}
+}
+
 // BuildStubConfig constructs a scheduler Config wired entirely with in-process
 // stubs.  Useful for:
 //   - Starting the scheduler before tasks #5 / #14 are delivered (no-op ticks).
@@ -41,8 +90,8 @@ func BuildStubConfig(rdb *redis.Client, probeStore *probe.Store) Config {
 
 	streaks := detect.NewStreakStore(rdb)
 	engine := detect.NewEngine(
-		probeStore,    // 15A probe snapshot reader
-		stubDetectLC,  // 15D lifecycle (stub until #5)
+		probeStore,   // 15A probe snapshot reader
+		stubDetectLC, // 15D lifecycle (stub until #5)
 		streaks,
 		rdb,
 		nil, // notifier — LogNotifier used by default
@@ -91,8 +140,8 @@ func (stubOrchLC) TransitionStatus(_ context.Context, _, _, _ string, _ map[stri
 	return 1, nil
 }
 
-func (stubOrchLC) SetWeight(_ context.Context, _ string, _ int) error     { return nil }
-func (stubOrchLC) BumpVersion(_ context.Context) error                     { return nil }
+func (stubOrchLC) SetWeight(_ context.Context, _ string, _ int) error       { return nil }
+func (stubOrchLC) BumpVersion(_ context.Context) error                      { return nil }
 func (stubOrchLC) WriteAuditLog(_ context.Context, _, _, _, _ string) error { return nil }
 
 // stubProvision implements orchestrate.ProvisionService as a no-op.

server/internal/scheduler/wiring_lifecycle.go

@@ -0,0 +1,319 @@
+package scheduler
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/nodes"
+	"github.com/wangjia/pangolin/server/internal/scheduler/detect"
+	"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
+)
+
+// SQLLifecycle is a MySQL-backed lifecycle store. The detect and orchestrate
+// LifecycleService interfaces are satisfied by the two thin adapter wrappers at
+// the bottom of this file (detectLifecycle / orchestrateLifecycle), both of
+// which delegate to this shared core.
+//
+// TransitionStatus is the documented optimistic-lock contract:
+//
+//	UPDATE nodes SET status=to WHERE id=nodeID AND status=from
+//
+// returning rows affected (1 = applied, 0 = lock conflict / no-op). On a 1-row
+// transition it best-effort writes a node_events row and bumps the global
+// directory_version so clients re-fetch.
+//
+// NOTE: cert revocation on a destroyed transition is intentionally NOT done here
+// — provision.DestroyNode tears down the VM, and the agent's mTLS cert can be
+// revoked separately. Wire CRL revocation when the real fleet exists (TODO).
+type SQLLifecycle struct {
+	db    *sql.DB
+	loads *nodes.LoadCache
+}
+
+// NewSQLLifecycle wires the store. loads may be nil (load reads then return zero).
+func NewSQLLifecycle(db *sql.DB, loads *nodes.LoadCache) *SQLLifecycle {
+	return &SQLLifecycle{db: db, loads: loads}
+}
+
+const nodeSelectCols = `id, uuid, status, region, tier, role, reality_sni, reality_pbk, hy2_port, provider_id, weight, name_zh, name_en`
+
+type lcRow struct {
+	id         int64
+	uuid       string
+	status     string
+	region     string
+	tier       string
+	role       string
+	sni        string
+	pbk        string
+	hy2Port    int
+	providerID int64
+	weight     int
+	nameZH     string
+	nameEn     string
+}
+
+func scanNode(rows interface{ Scan(...any) error }) (lcRow, error) {
+	var r lcRow
+	var hy2 sql.NullInt32
+	err := rows.Scan(&r.id, &r.uuid, &r.status, &r.region, &r.tier, &r.role,
+		&r.sni, &r.pbk, &hy2, &r.providerID, &r.weight, &r.nameZH, &r.nameEn)
+	if hy2.Valid {
+		r.hy2Port = int(hy2.Int32)
+	}
+	return r, err
+}
+
+func (l *SQLLifecycle) list(ctx context.Context, statuses []string) ([]lcRow, int64, error) {
+	q := `SELECT ` + nodeSelectCols + ` FROM nodes`
+	var args []any
+	if len(statuses) > 0 {
+		ph := make([]string, len(statuses))
+		for i, s := range statuses {
+			ph[i] = "?"
+			args = append(args, s)
+		}
+		q += ` WHERE status IN (` + strings.Join(ph, ",") + `)`
+	}
+	rows, err := l.db.QueryContext(ctx, q, args...)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer rows.Close()
+	var out []lcRow
+	for rows.Next() {
+		r, err := scanNode(rows)
+		if err != nil {
+			return nil, 0, err
+		}
+		out = append(out, r)
+	}
+	return out, l.version(ctx), rows.Err()
+}
+
+func (l *SQLLifecycle) get(ctx context.Context, nodeID string) (*lcRow, error) {
+	row := l.db.QueryRowContext(ctx, `SELECT `+nodeSelectCols+` FROM nodes WHERE id = ?`, nodeID)
+	r, err := scanNode(row)
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &r, nil
+}
+
+func (l *SQLLifecycle) transition(ctx context.Context, nodeID, from, to string, detail map[string]any) (int, error) {
+	tx, err := l.db.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, err
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	res, err := tx.ExecContext(ctx,
+		`UPDATE nodes SET status = ? WHERE id = ? AND status = ?`, to, nodeID, from)
+	if err != nil {
+		return 0, err
+	}
+	n, _ := res.RowsAffected()
+	if n == 0 {
+		return 0, nil // optimistic-lock conflict: node already moved → no-op
+	}
+
+	if ev := eventForStatus(to); ev != "" {
+		var dj any
+		if detail != nil {
+			if b, err := json.Marshal(detail); err == nil {
+				dj = string(b)
+			}
+		}
+		_, _ = tx.ExecContext(ctx,
+			`INSERT INTO node_events (node_id, event, detail) VALUES (?, ?, ?)`, nodeID, ev, dj)
+	}
+	if _, err := tx.ExecContext(ctx,
+		`INSERT INTO directory_version (id, version) VALUES (1, 1)
+		 ON DUPLICATE KEY UPDATE version = version + 1`); err != nil {
+		return 0, err
+	}
+	if err := tx.Commit(); err != nil {
+		return 0, err
+	}
+	return int(n), nil
+}
+
+func (l *SQLLifecycle) setWeight(ctx context.Context, nodeID string, weight int) error {
+	_, err := l.db.ExecContext(ctx, `UPDATE nodes SET weight = ? WHERE id = ?`, weight, nodeID)
+	return err
+}
+
+func (l *SQLLifecycle) bumpVersion(ctx context.Context) error {
+	_, err := l.db.ExecContext(ctx,
+		`INSERT INTO directory_version (id, version) VALUES (1, 1)
+		 ON DUPLICATE KEY UPDATE version = version + 1`)
+	return err
+}
+
+func (l *SQLLifecycle) version(ctx context.Context) int64 {
+	var v int64
+	_ = l.db.QueryRowContext(ctx, `SELECT version FROM directory_version WHERE id = 1`).Scan(&v)
+	return v
+}
+
+// loadFor resolves a node id → uuid → its current load sample (online, Mbps).
+func (l *SQLLifecycle) loadFor(ctx context.Context, nodeID string) (int, float64) {
+	if l.loads == nil {
+		return 0, 0
+	}
+	var uuid string
+	if err := l.db.QueryRowContext(ctx, `SELECT uuid FROM nodes WHERE id = ?`, nodeID).Scan(&uuid); err != nil {
+		return 0, 0
+	}
+	nl, ok, err := l.loads.Get(ctx, uuid)
+	if err != nil || !ok || nl == nil {
+		return 0, 0
+	}
+	mbps := float64(nl.BandwidthUpBps+nl.BandwidthDownBps) / 1_000_000.0
+	return int(nl.OnlineCount), mbps
+}
+
+func (l *SQLLifecycle) writeAudit(ctx context.Context, actor, action, target, meta string) error {
+	var metaArg any
+	if meta != "" {
+		metaArg = meta
+	}
+	_, err := l.db.ExecContext(ctx,
+		`INSERT INTO audit_log (actor, action, target, meta) VALUES (?, ?, ?, ?)`,
+		actor, action, target, metaArg)
+	return err
+}
+
+// eventForStatus maps a target node status to the node_events ENUM value, or ""
+// when the status has no corresponding event (then no event row is written).
+func eventForStatus(status string) string {
+	switch status {
+	case "provisioning", "probing":
+		return "provisioned"
+	case "up":
+		return "marked_up"
+	case "draining":
+		return "draining"
+	case "blocked_suspect":
+		return "blocked_suspect"
+	case "blocked_confirmed":
+		return "blocked_confirmed"
+	case "destroyed":
+		return "destroyed"
+	default:
+		return "" // e.g. "down" has no event enum
+	}
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// detect.LifecycleService adapter
+// ─────────────────────────────────────────────────────────────────────────────
+
+type detectLifecycle struct{ *SQLLifecycle }
+
+// NewDetectLifecycle adapts SQLLifecycle to detect.LifecycleService.
+func NewDetectLifecycle(l *SQLLifecycle) detect.LifecycleService { return detectLifecycle{l} }
+
+func (a detectLifecycle) ListNodes(ctx context.Context, f detect.NodeFilter) ([]detect.NodeInfo, error) {
+	statuses := make([]string, len(f.Statuses))
+	for i, s := range f.Statuses {
+		statuses[i] = string(s)
+	}
+	rows, ver, err := a.list(ctx, statuses)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]detect.NodeInfo, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, detect.NodeInfo{
+			ID:      strconv.FormatInt(r.id, 10),
+			UUID:    r.uuid,
+			Status:  detect.NodeStatus(r.status),
+			Weight:  r.weight,
+			Version: ver,
+		})
+	}
+	return out, nil
+}
+
+func (a detectLifecycle) TransitionStatus(ctx context.Context, nodeID string, from, to detect.NodeStatus, detail map[string]any) (int, error) {
+	return a.transition(ctx, nodeID, string(from), string(to), detail)
+}
+
+func (a detectLifecycle) SetWeight(ctx context.Context, nodeID string, weight int) error {
+	return a.setWeight(ctx, nodeID, weight)
+}
+
+func (a detectLifecycle) BumpVersion(ctx context.Context) error { return a.bumpVersion(ctx) }
+
+func (a detectLifecycle) GetLoad(ctx context.Context, nodeID string) (detect.LoadInfo, error) {
+	online, mbps := a.loadFor(ctx, nodeID)
+	return detect.LoadInfo{Online: online, BandwidthMbps: mbps, Timestamp: time.Now().Unix()}, nil
+}
+
+func (a detectLifecycle) GetLoadHistory(ctx context.Context, nodeID string, _ time.Duration) ([]detect.LoadPoint, error) {
+	// No historical load store yet (only the latest sample lives in Redis), so
+	// return the current point. The drop-percentage rule degrades safely (no fire).
+	online, mbps := a.loadFor(ctx, nodeID)
+	if online == 0 && mbps == 0 {
+		return nil, nil
+	}
+	return []detect.LoadPoint{{Timestamp: time.Now().Unix(), Online: online, BandwidthMbps: mbps}}, nil
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// orchestrate.LifecycleService adapter
+// ─────────────────────────────────────────────────────────────────────────────
+
+type orchestrateLifecycle struct{ *SQLLifecycle }
+
+// NewOrchestrateLifecycle adapts SQLLifecycle to orchestrate.LifecycleService.
+func NewOrchestrateLifecycle(l *SQLLifecycle) orchestrate.LifecycleService {
+	return orchestrateLifecycle{l}
+}
+
+func (a orchestrateLifecycle) GetNode(ctx context.Context, nodeID string) (*orchestrate.NodeInfo, error) {
+	r, err := a.get(ctx, nodeID)
+	if err != nil || r == nil {
+		return nil, err
+	}
+	return &orchestrate.NodeInfo{
+		ID:         strconv.FormatInt(r.id, 10),
+		Tier:       r.tier,
+		Region:     r.region,
+		Role:       r.role,
+		ProviderID: strconv.FormatInt(r.providerID, 10),
+		RealitySNI: r.sni,
+		RealityPBK: r.pbk,
+		HY2Port:    r.hy2Port,
+		NameZH:     r.nameZH,
+		NameEn:     r.nameEn,
+	}, nil
+}
+
+func (a orchestrateLifecycle) TransitionStatus(ctx context.Context, nodeID string, from, to string, detail map[string]any) (int, error) {
+	return a.transition(ctx, nodeID, from, to, detail)
+}
+
+func (a orchestrateLifecycle) SetWeight(ctx context.Context, nodeID string, weight int) error {
+	return a.setWeight(ctx, nodeID, weight)
+}
+
+func (a orchestrateLifecycle) BumpVersion(ctx context.Context) error { return a.bumpVersion(ctx) }
+
+func (a orchestrateLifecycle) WriteAuditLog(ctx context.Context, actor, action, target, meta string) error {
+	return a.writeAudit(ctx, actor, action, target, meta)
+}
+
+// assert at compile time that the adapters satisfy their interfaces.
+var (
+	_ detect.LifecycleService      = detectLifecycle{}
+	_ orchestrate.LifecycleService = orchestrateLifecycle{}
+)

server/internal/scheduler/wiring_lifecycle_test.go

@@ -0,0 +1,22 @@
+package scheduler
+
+import "testing"
+
+func TestEventForStatus(t *testing.T) {
+	cases := map[string]string{
+		"up":                "marked_up",
+		"draining":          "draining",
+		"blocked_suspect":   "blocked_suspect",
+		"blocked_confirmed": "blocked_confirmed",
+		"destroyed":         "destroyed",
+		"probing":           "provisioned",
+		"provisioning":      "provisioned",
+		"down":              "", // no event enum → skip writing an event row
+		"bogus":             "",
+	}
+	for status, want := range cases {
+		if got := eventForStatus(status); got != want {
+			t.Errorf("eventForStatus(%q) = %q, want %q", status, got, want)
+		}
+	}
+}