merge: maestro/tsk_aG4yOS90inF_ into main (bulk integration)

server/internal/nodes/lifecycle.go

@@ -0,0 +1,285 @@
+package nodes
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/wangjia/pangolin/server/internal/mtls"
+)
+
+// ErrInvalidTransition is returned when the requested status transition is not
+// permitted by the lifecycle state machine, or when the node's current state
+// does not match any valid "from" state (optimistic-lock conflict or illegal move).
+var ErrInvalidTransition = errors.New("nodes: invalid lifecycle transition")
+
+// nodeEvent mirrors the nine ENUM values in the node_events.event column (migration 6).
+type nodeEvent string
+
+const (
+	eventProvisioned      nodeEvent = "provisioned"
+	eventMarkedUp         nodeEvent = "marked_up"
+	eventDraining         nodeEvent = "draining"
+	eventProbeFail        nodeEvent = "probe_fail"
+	eventBlockedSuspect   nodeEvent = "blocked_suspect"
+	eventBlockedConfirmed nodeEvent = "blocked_confirmed"
+	eventDestroyed        nodeEvent = "destroyed"
+)
+
+// transitionSpec describes one allowed status move for a Lifecycle method.
+// When a method accepts multiple "from" states, specs are tried in order.
+type transitionSpec struct {
+	from  string
+	to    string
+	event nodeEvent
+}
+
+// Lifecycle drives the node lifecycle state machine.
+//
+// State machine (doc/04 §3):
+//
+//	provisioning → probing → up → draining → down → destroyed
+//	                     ↘ destroyed (bad IP)
+//	                                  ↘ down (skip draining, confirmed block)
+//
+// Each method executes a single MySQL transaction that:
+//  1. UPDATE nodes SET status=<to> WHERE uuid=? AND status=<from>  (optimistic lock)
+//  2. INSERT node_events(node_id, event, detail JSON, at)
+//  3. BumpVersion(ctx, tx)  — increments the directory_version singleton
+//
+// MarkDestroyed additionally (after commit) calls mtls.CRL.Revoke and deletes
+// the Redis keys node:load:{uuid} and node:cmdq:{uuid}.
+//
+// Called by #15 (block-detection / drain scheduler) and #14 (provisioning).
+type Lifecycle struct {
+	db  *sql.DB
+	crl *mtls.CRL    // for MarkDestroyed post-commit hook
+	rdb redis.Cmdable // for MarkDestroyed post-commit cleanup
+}
+
+// NewLifecycle constructs a Lifecycle.
+//   - db:  MySQL connection pool (from store.Open)
+//   - crl: mTLS revocation manager (task 5b, mtls.NewCRL)
+//   - rdb: Redis client or Cmdable (for post-destroy key cleanup)
+func NewLifecycle(db *sql.DB, crl *mtls.CRL, rdb redis.Cmdable) *Lifecycle {
+	return &Lifecycle{db: db, crl: crl, rdb: rdb}
+}
+
+// BumpVersion atomically increments the directory_version singleton within tx.
+//
+// Signature matches the one defined by task 5d (BumpVersion(ctx, tx)); if 5d
+// has merged, remove this copy and update callers to use the 5d version.
+func BumpVersion(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx,
+		`INSERT INTO directory_version (id, version) VALUES (1, 1)
+		 ON DUPLICATE KEY UPDATE version = version + 1`,
+	)
+	if err != nil {
+		return fmt.Errorf("nodes.BumpVersion: %w", err)
+	}
+	return nil
+}
+
+// MarkProbing transitions a node from provisioning → probing.
+// Writes a "provisioned" event.
+// Called by #14 after the VM passes initial reachability checks.
+func (l *Lifecycle) MarkProbing(ctx context.Context, nodeUUID string, detail map[string]any) error {
+	return l.transact(ctx, nodeUUID, []transitionSpec{
+		{from: "provisioning", to: "probing", event: eventProvisioned},
+	}, detail)
+}
+
+// MarkUp transitions a node from probing → up, admitting it to the directory.
+// Writes a "marked_up" event.
+// Called by #15 after initial probes confirm the node is reachable.
+func (l *Lifecycle) MarkUp(ctx context.Context, nodeUUID string, detail map[string]any) error {
+	return l.transact(ctx, nodeUUID, []transitionSpec{
+		{from: "probing", to: "up", event: eventMarkedUp},
+	}, detail)
+}
+
+// MarkDraining transitions a node from up → draining, removing it from the
+// live directory while letting existing connections finish.
+// Writes a "draining" event.
+// The subsequent draining→down transition is driven by the #15 scheduler timer.
+// Called by #15 on sustained suspect signals.
+func (l *Lifecycle) MarkDraining(ctx context.Context, nodeUUID string, detail map[string]any) error {
+	return l.transact(ctx, nodeUUID, []transitionSpec{
+		{from: "up", to: "draining", event: eventDraining},
+	}, detail)
+}
+
+// MarkDown transitions a node from draining → down or (fast-path) up → down.
+// Both paths write a "blocked_confirmed" event.
+// The up → down fast-path skips draining when block is immediately confirmed (#15).
+// The draining → down path is driven by the drain timer in #15 scheduler.
+func (l *Lifecycle) MarkDown(ctx context.Context, nodeUUID string, detail map[string]any) error {
+	return l.transact(ctx, nodeUUID, []transitionSpec{
+		{from: "draining", to: "down", event: eventBlockedConfirmed},
+		{from: "up", to: "down", event: eventBlockedConfirmed},
+	}, detail)
+}
+
+// MarkDestroyed permanently retires a node.
+// Valid from-states:
+//   - probing  → destroyed ("probe_fail" event):  bad IP, never entered the pool.
+//   - down     → destroyed ("destroyed" event):    normal retirement after draining.
+//
+// After the DB transaction commits, two post-commit hooks run:
+//  1. mtls.CRL.Revoke(nodeUUID)       — marks the node's cert as revoked in Redis.
+//  2. Redis DEL node:load:{uuid} and node:cmdq:{uuid} — evicts runtime state.
+//
+// Called by #14 (failed initial probe) and #15 (retirement scheduler).
+func (l *Lifecycle) MarkDestroyed(ctx context.Context, nodeUUID string, detail map[string]any) error {
+	if err := l.transact(ctx, nodeUUID, []transitionSpec{
+		{from: "probing", to: "destroyed", event: eventProbeFail},
+		{from: "down", to: "destroyed", event: eventDestroyed},
+	}, detail); err != nil {
+		return err
+	}
+
+	// Post-commit: revoke mTLS certificate.
+	if l.crl != nil {
+		if err := l.crl.Revoke(ctx, nodeUUID); err != nil {
+			return fmt.Errorf("lifecycle.MarkDestroyed: revoke cert: %w", err)
+		}
+	}
+
+	// Post-commit: delete runtime Redis keys.
+	if l.rdb != nil {
+		if err := l.rdb.Del(ctx,
+			nodeLoadKeyPrefix+nodeUUID,
+			cmdQueuePrefix+nodeUUID,
+		).Err(); err != nil {
+			return fmt.Errorf("lifecycle.MarkDestroyed: redis del: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// MarkBlockedSuspect does NOT change the node's status.
+// It lowers the routing weight to 10, writes a "blocked_suspect" event, and
+// bumps the directory version so clients receive a re-weighted catalogue.
+//
+// Called by #15 on the first domestic-probe failure streak ("疑似" judgement).
+func (l *Lifecycle) MarkBlockedSuspect(ctx context.Context, nodeUUID string, detail map[string]any) error {
+	tx, err := l.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("lifecycle.MarkBlockedSuspect: begin: %w", err)
+	}
+	defer tx.Rollback() //nolint:errcheck // no-op after Commit
+
+	// Resolve the node's internal ID — fail fast if UUID is unknown.
+	var nodeID int64
+	if err := tx.QueryRowContext(ctx,
+		`SELECT id FROM nodes WHERE uuid = ?`, nodeUUID,
+	).Scan(&nodeID); err == sql.ErrNoRows {
+		return ErrInvalidTransition
+	} else if err != nil {
+		return fmt.Errorf("lifecycle.MarkBlockedSuspect: resolve id: %w", err)
+	}
+
+	// Lower routing weight to 10 (idempotent if already ≤ 10).
+	if _, err := tx.ExecContext(ctx,
+		`UPDATE nodes SET weight = 10 WHERE uuid = ?`, nodeUUID,
+	); err != nil {
+		return fmt.Errorf("lifecycle.MarkBlockedSuspect: update weight: %w", err)
+	}
+
+	// Write the blocked_suspect event.
+	detailJSON, err := json.Marshal(detail)
+	if err != nil {
+		return fmt.Errorf("lifecycle.MarkBlockedSuspect: marshal detail: %w", err)
+	}
+	if _, err := tx.ExecContext(ctx,
+		`INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, UTC_TIMESTAMP(6))`,
+		nodeID, string(eventBlockedSuspect), detailJSON,
+	); err != nil {
+		return fmt.Errorf("lifecycle.MarkBlockedSuspect: insert event: %w", err)
+	}
+
+	// Bump directory version.
+	if err := BumpVersion(ctx, tx); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
+// transact attempts each spec in order, all within a single transaction.
+//
+// The first spec whose UPDATE affects ≥ 1 row "wins": the corresponding event
+// is inserted, the directory version is bumped, and the transaction is committed.
+//
+// If no spec matches (all return 0 rows affected), the transaction is rolled back
+// and ErrInvalidTransition is returned.  A 0-row result means either:
+//   - The node is in a state that is not a valid "from" for this method (illegal move).
+//   - A concurrent writer already advanced the node past the expected "from" state
+//     (optimistic-lock conflict: the first writer won; the second sees ErrInvalidTransition).
+func (l *Lifecycle) transact(
+	ctx context.Context,
+	nodeUUID string,
+	specs []transitionSpec,
+	detail map[string]any,
+) error {
+	detailJSON, err := json.Marshal(detail)
+	if err != nil {
+		return fmt.Errorf("lifecycle.transact: marshal detail: %w", err)
+	}
+
+	tx, err := l.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("lifecycle.transact: begin: %w", err)
+	}
+	defer tx.Rollback() //nolint:errcheck // no-op after Commit
+
+	for _, spec := range specs {
+		res, err := tx.ExecContext(ctx,
+			`UPDATE nodes SET status = ? WHERE uuid = ? AND status = ?`,
+			spec.to, nodeUUID, spec.from,
+		)
+		if err != nil {
+			return fmt.Errorf("lifecycle.transact [%s→%s]: %w", spec.from, spec.to, err)
+		}
+		n, err := res.RowsAffected()
+		if err != nil {
+			return fmt.Errorf("lifecycle.transact RowsAffected: %w", err)
+		}
+		if n == 0 {
+			continue // from-state mismatch; try next spec
+		}
+
+		// Transition succeeded — resolve the internal node_id for the event row.
+		var nodeID int64
+		if err := tx.QueryRowContext(ctx,
+			`SELECT id FROM nodes WHERE uuid = ?`, nodeUUID,
+		).Scan(&nodeID); err != nil {
+			return fmt.Errorf("lifecycle.transact: resolve node_id after [%s→%s]: %w",
+				spec.from, spec.to, err)
+		}
+
+		// Insert the event record.
+		if _, err := tx.ExecContext(ctx,
+			`INSERT INTO node_events (node_id, event, detail, at) VALUES (?, ?, ?, UTC_TIMESTAMP(6))`,
+			nodeID, string(spec.event), detailJSON,
+		); err != nil {
+			return fmt.Errorf("lifecycle.transact: insert event [%s→%s]: %w",
+				spec.from, spec.to, err)
+		}
+
+		// Bump the global directory version.
+		if err := BumpVersion(ctx, tx); err != nil {
+			return err
+		}
+
+		return tx.Commit()
+	}
+
+	// No spec matched → illegal transition from the node's current state.
+	return ErrInvalidTransition
+}

server/internal/nodes/lifecycle_test.go

@@ -0,0 +1,551 @@
+//go:build integration
+
+// Integration tests for the Lifecycle state machine.
+//
+// Required: a running MySQL 8 instance with an empty database.
+// Set PANGOLIN_TEST_DSN to its DSN, e.g.:
+//
+//	export PANGOLIN_TEST_DSN="root:secret@tcp(127.0.0.1:3306)/pangolin_test?parseTime=true"
+//	go test -tags integration ./internal/nodes/... -run TestLifecycle -v
+//
+// Tests use miniredis in-process for the CRL — no separate Redis is needed.
+package nodes_test
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"os"
+	"sync"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+
+	"github.com/wangjia/pangolin/server/internal/config"
+	"github.com/wangjia/pangolin/server/internal/mtls"
+	"github.com/wangjia/pangolin/server/internal/nodes"
+	"github.com/wangjia/pangolin/server/internal/store"
+)
+
+// ─── Test infrastructure ─────────────────────────────────────────────────────
+
+// lifecycleFixture holds all resources for one lifecycle integration test.
+type lifecycleFixture struct {
+	lc  *nodes.Lifecycle
+	db  *sql.DB
+	mr  *miniredis.Miniredis
+	rdb *redis.Client // same underlying server as mr
+}
+
+// requireDSN reads PANGOLIN_TEST_DSN from the environment.
+// If not set, the test is skipped with an informative message.
+// The DSN must include multiStatements=true (set automatically by TestMain
+// when it starts its own container; callers providing an external DSN should
+// include it too).
+func requireDSN(t *testing.T) string {
+	t.Helper()
+	dsn := os.Getenv("PANGOLIN_TEST_DSN")
+	if dsn == "" {
+		t.Skip("set PANGOLIN_TEST_DSN to run lifecycle integration tests " +
+			"(e.g. root:secret@tcp(127.0.0.1:3306)/pangolin_test?multiStatements=true)")
+	}
+	return dsn
+}
+
+// setupLifecycle opens the MySQL database identified by PANGOLIN_TEST_DSN,
+// applies all migrations, seeds a provider row, and wires up a Lifecycle
+// backed by miniredis for the CRL.
+//
+// All resources are cleaned up via t.Cleanup.
+//
+// Implementation note: golang-migrate's MySQL driver calls (*sql.DB).Close()
+// when its own Close() is invoked (via m.Close() in newMigrator's cleanup).
+// To avoid having the migration step close the connection that the test will
+// use, we open a dedicated short-lived *sql.DB just for MigrateUp, close it
+// immediately after, and then open a fresh *sql.DB for test assertions.
+func setupLifecycle(t *testing.T) *lifecycleFixture {
+	t.Helper()
+	dsn := requireDSN(t)
+	ctx := context.Background()
+
+	// 1. Run migrations on a dedicated connection that golang-migrate will close.
+	migDB, err := store.Open(&config.Config{DSN: dsn})
+	if err != nil {
+		t.Fatalf("store.Open (migrate): %v", err)
+	}
+	if err := store.MigrateUp(migDB); err != nil {
+		_ = migDB.Close()
+		t.Fatalf("MigrateUp: %v", err)
+	}
+	_ = migDB.Close() // golang-migrate already closed it; this is a no-op but explicit.
+
+	// 2. Open a fresh connection for the actual test operations.
+	db, err := store.Open(&config.Config{DSN: dsn})
+	if err != nil {
+		t.Fatalf("store.Open: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+
+	// Clean up nodes/events from any previous run so tests are idempotent.
+	if _, err := db.ExecContext(ctx,
+		`DELETE FROM node_events`); err != nil {
+		t.Fatalf("truncate node_events: %v", err)
+	}
+	if _, err := db.ExecContext(ctx,
+		`DELETE FROM nodes`); err != nil {
+		t.Fatalf("truncate nodes: %v", err)
+	}
+
+	// Seed a provider row — nodes.provider_id is a FK.
+	if _, err := db.ExecContext(ctx,
+		`INSERT INTO providers (name, api_kind, regions, pool)
+		 VALUES ('test-provider', 'vultr', '["HK"]', 'consumable')
+		 ON DUPLICATE KEY UPDATE name = name`,
+	); err != nil {
+		t.Fatalf("seed provider: %v", err)
+	}
+
+	mr := miniredis.RunT(t)
+	rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { _ = rdb.Close() })
+
+	crl := mtls.NewCRL(rdb, nil) // nil db — only Redis revocation needed in tests
+	lc := nodes.NewLifecycle(db, crl, rdb)
+
+	return &lifecycleFixture{lc: lc, db: db, mr: mr, rdb: rdb}
+}
+
+// insertNode inserts a test node with the given uuid and initial status.
+func insertNode(t *testing.T, db *sql.DB, uuid, status string) {
+	t.Helper()
+	ctx := context.Background()
+	var providerID int64
+	if err := db.QueryRowContext(ctx,
+		`SELECT id FROM providers LIMIT 1`,
+	).Scan(&providerID); err != nil {
+		t.Fatalf("get provider_id: %v", err)
+	}
+	_, err := db.ExecContext(ctx, `
+		INSERT INTO nodes
+		  (uuid, region, name_zh, name_en, tier, endpoint, reality_pbk, reality_sni, provider_id, status)
+		VALUES (?, 'HK', '香港测试', 'HK Test', 'free', '1.2.3.4:443', 'fake-pbk', 'sni.example.com', ?, ?)
+		ON DUPLICATE KEY UPDATE status = VALUES(status), weight = 100
+	`, uuid, providerID, status)
+	if err != nil {
+		t.Fatalf("insertNode(%s, %s): %v", uuid, status, err)
+	}
+}
+
+// nodeStatus reads the current status of a node by UUID.
+func nodeStatus(t *testing.T, db *sql.DB, uuid string) string {
+	t.Helper()
+	var status string
+	if err := db.QueryRowContext(context.Background(),
+		`SELECT status FROM nodes WHERE uuid = ?`, uuid,
+	).Scan(&status); err != nil {
+		t.Fatalf("nodeStatus(%s): %v", uuid, err)
+	}
+	return status
+}
+
+// nodeWeight reads the current weight of a node by UUID.
+func nodeWeight(t *testing.T, db *sql.DB, uuid string) int {
+	t.Helper()
+	var w int
+	if err := db.QueryRowContext(context.Background(),
+		`SELECT weight FROM nodes WHERE uuid = ?`, uuid,
+	).Scan(&w); err != nil {
+		t.Fatalf("nodeWeight(%s): %v", uuid, err)
+	}
+	return w
+}
+
+// eventCount returns the number of node_events rows for the node with the given UUID.
+func eventCount(t *testing.T, db *sql.DB, uuid string) int {
+	t.Helper()
+	var count int
+	if err := db.QueryRowContext(context.Background(), `
+		SELECT COUNT(*) FROM node_events ne
+		JOIN nodes n ON n.id = ne.node_id
+		WHERE n.uuid = ?
+	`, uuid).Scan(&count); err != nil {
+		t.Fatalf("eventCount(%s): %v", uuid, err)
+	}
+	return count
+}
+
+// dirVersion reads the current directory_version.
+func dirVersion(t *testing.T, db *sql.DB) int64 {
+	t.Helper()
+	var v int64
+	if err := db.QueryRowContext(context.Background(),
+		`SELECT version FROM directory_version WHERE id = 1`,
+	).Scan(&v); err != nil {
+		t.Fatalf("dirVersion: %v", err)
+	}
+	return v
+}
+
+// ─── Category 1: all valid transitions succeed ───────────────────────────────
+
+// TestLifecycle_AllValidTransitions verifies that every legal transition:
+//   - Updates the node's status to the target value.
+//   - Inserts exactly one node_events row.
+//   - Increments the directory_version by 1.
+func TestLifecycle_AllValidTransitions(t *testing.T) {
+	f := setupLifecycle(t)
+	ctx := context.Background()
+	detail := map[string]any{"source": "integration-test"}
+
+	cases := []struct {
+		name       string
+		uuid       string
+		start      string
+		fn         func(string) error
+		wantStatus string
+	}{
+		{
+			name:       "provisioning→probing",
+			uuid:       "lc-probing",
+			start:      "provisioning",
+			fn:         func(u string) error { return f.lc.MarkProbing(ctx, u, detail) },
+			wantStatus: "probing",
+		},
+		{
+			name:       "probing→up",
+			uuid:       "lc-up",
+			start:      "probing",
+			fn:         func(u string) error { return f.lc.MarkUp(ctx, u, detail) },
+			wantStatus: "up",
+		},
+		{
+			name:       "up→draining",
+			uuid:       "lc-draining",
+			start:      "up",
+			fn:         func(u string) error { return f.lc.MarkDraining(ctx, u, detail) },
+			wantStatus: "draining",
+		},
+		{
+			name:       "draining→down",
+			uuid:       "lc-down-from-drain",
+			start:      "draining",
+			fn:         func(u string) error { return f.lc.MarkDown(ctx, u, detail) },
+			wantStatus: "down",
+		},
+		{
+			name:       "up→down (skip draining, blocked confirmed)",
+			uuid:       "lc-down-from-up",
+			start:      "up",
+			fn:         func(u string) error { return f.lc.MarkDown(ctx, u, detail) },
+			wantStatus: "down",
+		},
+		{
+			name:       "probing→destroyed (bad IP, never pooled)",
+			uuid:       "lc-destroyed-probe",
+			start:      "probing",
+			fn:         func(u string) error { return f.lc.MarkDestroyed(ctx, u, detail) },
+			wantStatus: "destroyed",
+		},
+		{
+			name:       "down→destroyed (permanent retirement)",
+			uuid:       "lc-destroyed-down",
+			start:      "down",
+			fn:         func(u string) error { return f.lc.MarkDestroyed(ctx, u, detail) },
+			wantStatus: "destroyed",
+		},
+	}
+
+	v0 := dirVersion(t, f.db)
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			insertNode(t, f.db, tc.uuid, tc.start)
+			vBefore := dirVersion(t, f.db)
+
+			if err := tc.fn(tc.uuid); err != nil {
+				t.Fatalf("transition: %v", err)
+			}
+
+			// Status must have reached the target state.
+			if got := nodeStatus(t, f.db, tc.uuid); got != tc.wantStatus {
+				t.Errorf("status = %q, want %q", got, tc.wantStatus)
+			}
+
+			// Exactly one event row per transition.
+			if n := eventCount(t, f.db, tc.uuid); n != 1 {
+				t.Errorf("event_count = %d, want 1", n)
+			}
+
+			// Directory version incremented by exactly 1.
+			vAfter := dirVersion(t, f.db)
+			if vAfter != vBefore+1 {
+				t.Errorf("version delta = %d, want 1 (before=%d after=%d)",
+					vAfter-vBefore, vBefore, vAfter)
+			}
+		})
+	}
+
+	// Total version bumps must equal the number of valid transitions exercised.
+	if got := dirVersion(t, f.db); got != v0+int64(len(cases)) {
+		t.Errorf("total version = %d, want %d (seed=%d + %d transitions)",
+			got, v0+int64(len(cases)), v0, len(cases))
+	}
+}
+
+// ─── Category 2: invalid transitions return ErrInvalidTransition ─────────────
+
+// TestLifecycle_InvalidTransitions verifies that illegal moves return
+// ErrInvalidTransition and leave the node entirely unchanged (no event, no bump).
+func TestLifecycle_InvalidTransitions(t *testing.T) {
+	f := setupLifecycle(t)
+	ctx := context.Background()
+	detail := map[string]any{}
+
+	cases := []struct {
+		name  string
+		uuid  string
+		start string
+		fn    func(string) error
+	}{
+		{
+			name:  "down→up",
+			uuid:  "lc-inv-down-up",
+			start: "down",
+			fn:    func(u string) error { return f.lc.MarkUp(ctx, u, detail) },
+		},
+		{
+			name:  "destroyed→probing",
+			uuid:  "lc-inv-dest-probing",
+			start: "destroyed",
+			fn:    func(u string) error { return f.lc.MarkProbing(ctx, u, detail) },
+		},
+		{
+			name:  "destroyed→up",
+			uuid:  "lc-inv-dest-up",
+			start: "destroyed",
+			fn:    func(u string) error { return f.lc.MarkUp(ctx, u, detail) },
+		},
+		{
+			name:  "destroyed→draining",
+			uuid:  "lc-inv-dest-draining",
+			start: "destroyed",
+			fn:    func(u string) error { return f.lc.MarkDraining(ctx, u, detail) },
+		},
+		{
+			name:  "destroyed→down",
+			uuid:  "lc-inv-dest-down",
+			start: "destroyed",
+			fn:    func(u string) error { return f.lc.MarkDown(ctx, u, detail) },
+		},
+		{
+			name:  "provisioning→up",
+			uuid:  "lc-inv-prov-up",
+			start: "provisioning",
+			fn:    func(u string) error { return f.lc.MarkUp(ctx, u, detail) },
+		},
+		{
+			name:  "up→probing (backward)",
+			uuid:  "lc-inv-up-probing",
+			start: "up",
+			fn:    func(u string) error { return f.lc.MarkProbing(ctx, u, detail) },
+		},
+		{
+			name:  "down→draining (backward)",
+			uuid:  "lc-inv-down-draining",
+			start: "down",
+			fn:    func(u string) error { return f.lc.MarkDraining(ctx, u, detail) },
+		},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			insertNode(t, f.db, tc.uuid, tc.start)
+			vBefore := dirVersion(t, f.db)
+
+			err := tc.fn(tc.uuid)
+			if !errors.Is(err, nodes.ErrInvalidTransition) {
+				t.Errorf("got %v, want ErrInvalidTransition", err)
+			}
+
+			// Status must be unchanged.
+			if got := nodeStatus(t, f.db, tc.uuid); got != tc.start {
+				t.Errorf("status changed to %q after invalid transition; want %q",
+					got, tc.start)
+			}
+
+			// No event must have been written.
+			if n := eventCount(t, f.db, tc.uuid); n != 0 {
+				t.Errorf("event_count = %d after invalid transition; want 0", n)
+			}
+
+			// Directory version must be unchanged.
+			if vAfter := dirVersion(t, f.db); vAfter != vBefore {
+				t.Errorf("directory_version changed from %d to %d after invalid transition",
+					vBefore, vAfter)
+			}
+		})
+	}
+}
+
+// ─── Category 3: concurrent write — exactly one succeeds ──────────────────────
+
+// TestLifecycle_ConcurrentWrite launches two goroutines that both attempt the
+// same transition simultaneously and verifies that exactly one succeeds
+// (optimistic-lock guarantee via UPDATE … WHERE status=<from>).
+func TestLifecycle_ConcurrentWrite(t *testing.T) {
+	f := setupLifecycle(t)
+	ctx := context.Background()
+	detail := map[string]any{"concurrent": true}
+
+	const uuid = "lc-concurrent-destroy"
+	insertNode(t, f.db, uuid, "down")
+
+	var wg sync.WaitGroup
+	errs := make([]error, 2)
+
+	for i := range errs {
+		wg.Add(1)
+		i := i
+		go func() {
+			defer wg.Done()
+			errs[i] = f.lc.MarkDestroyed(ctx, uuid, detail)
+		}()
+	}
+	wg.Wait()
+
+	successes := 0
+	for _, err := range errs {
+		if err == nil {
+			successes++
+		}
+	}
+	if successes != 1 {
+		t.Errorf("concurrent MarkDestroyed: %d successes, want exactly 1 (errs: %v | %v)",
+			successes, errs[0], errs[1])
+	}
+
+	// Final state: destroyed, one event.
+	if got := nodeStatus(t, f.db, uuid); got != "destroyed" {
+		t.Errorf("final status = %q, want destroyed", got)
+	}
+	if n := eventCount(t, f.db, uuid); n != 1 {
+		t.Errorf("event_count = %d after concurrent destroy; want 1", n)
+	}
+}
+
+// ─── Category 4: MarkDestroyed post-commit hooks ──────────────────────────────
+
+// TestLifecycle_MarkDestroyed_PostHooks verifies that after a successful
+// MarkDestroyed call:
+//   - mtls.CRL.IsRevoked(nodeUUID) returns true.
+//   - Redis keys node:load:{uuid} and node:cmdq:{uuid} are deleted.
+func TestLifecycle_MarkDestroyed_PostHooks(t *testing.T) {
+	f := setupLifecycle(t)
+	ctx := context.Background()
+
+	const uuid = "lc-destroyed-hooks"
+	insertNode(t, f.db, uuid, "down")
+
+	// Pre-populate the Redis keys that MarkDestroyed should clean up.
+	loadKey := "node:load:" + uuid
+	cmdqKey := "node:cmdq:" + uuid
+	if err := f.rdb.Set(ctx, loadKey, "dummy-load", 0).Err(); err != nil {
+		t.Fatalf("pre-seed load key: %v", err)
+	}
+	if err := f.rdb.ZAdd(ctx, cmdqKey, redis.Z{Score: 1, Member: "dummy-cmd"}).Err(); err != nil {
+		t.Fatalf("pre-seed cmdq key: %v", err)
+	}
+
+	// Sanity-check: both keys exist before the call.
+	if n, err := f.rdb.Exists(ctx, loadKey, cmdqKey).Result(); err != nil || n != 2 {
+		t.Fatalf("pre-seed check: exists=%d err=%v (want 2 keys)", n, err)
+	}
+
+	if err := f.lc.MarkDestroyed(ctx, uuid, nil); err != nil {
+		t.Fatalf("MarkDestroyed: %v", err)
+	}
+
+	// Certificate must be revoked: verify via an independent CRL view backed
+	// by the same miniredis instance.
+	crl2 := mtls.NewCRL(f.rdb, nil)
+	if !crl2.IsRevoked(uuid) {
+		t.Error("IsRevoked = false after MarkDestroyed; want true")
+	}
+
+	// Both Redis keys must be gone.
+	if n, err := f.rdb.Exists(ctx, loadKey, cmdqKey).Result(); err != nil {
+		t.Fatalf("redis Exists after destroy: %v", err)
+	} else if n != 0 {
+		t.Errorf("redis keys still exist after MarkDestroyed: %d key(s) remain", n)
+	}
+}
+
+// ─── Category 5: MarkBlockedSuspect — weight=10, status unchanged ─────────────
+
+// TestLifecycle_MarkBlockedSuspect verifies that MarkBlockedSuspect:
+//   - Does NOT change the node's status.
+//   - Sets the node's weight to 10.
+//   - Writes exactly one node_events row with event=blocked_suspect.
+//   - Bumps the directory version by 1.
+func TestLifecycle_MarkBlockedSuspect(t *testing.T) {
+	f := setupLifecycle(t)
+	ctx := context.Background()
+
+	const uuid = "lc-suspect"
+	insertNode(t, f.db, uuid, "up")
+
+	vBefore := dirVersion(t, f.db)
+
+	if err := f.lc.MarkBlockedSuspect(ctx, uuid, map[string]any{
+		"reason":         "domestic probes failing",
+		"failure_streak": 3,
+	}); err != nil {
+		t.Fatalf("MarkBlockedSuspect: %v", err)
+	}
+
+	// Status must remain "up".
+	if got := nodeStatus(t, f.db, uuid); got != "up" {
+		t.Errorf("status = %q after MarkBlockedSuspect; want up", got)
+	}
+
+	// Weight must be 10.
+	if w := nodeWeight(t, f.db, uuid); w != 10 {
+		t.Errorf("weight = %d after MarkBlockedSuspect; want 10", w)
+	}
+
+	// Exactly one event row.
+	if n := eventCount(t, f.db, uuid); n != 1 {
+		t.Errorf("event_count = %d after MarkBlockedSuspect; want 1", n)
+	}
+
+	// Directory version must have been bumped by 1.
+	if vAfter := dirVersion(t, f.db); vAfter != vBefore+1 {
+		t.Errorf("directory_version: want %d, got %d", vBefore+1, vAfter)
+	}
+}
+
+// TestLifecycle_MarkBlockedSuspect_Idempotent verifies that calling
+// MarkBlockedSuspect twice keeps weight at 10 and writes a second event.
+func TestLifecycle_MarkBlockedSuspect_Idempotent(t *testing.T) {
+	f := setupLifecycle(t)
+	ctx := context.Background()
+
+	const uuid = "lc-suspect-idem"
+	insertNode(t, f.db, uuid, "up")
+
+	for i := 1; i <= 2; i++ {
+		if err := f.lc.MarkBlockedSuspect(ctx, uuid, map[string]any{"streak": i}); err != nil {
+			t.Fatalf("call %d: MarkBlockedSuspect: %v", i, err)
+		}
+	}
+
+	if w := nodeWeight(t, f.db, uuid); w != 10 {
+		t.Errorf("weight = %d after two calls; want 10", w)
+	}
+	if n := eventCount(t, f.db, uuid); n != 2 {
+		t.Errorf("event_count = %d; want 2 (one per call)", n)
+	}
+}

server/internal/nodes/testmain_test.go

@@ -0,0 +1,86 @@
+//go:build integration
+
+package nodes_test
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"testing"
+	"time"
+)
+
+// TestMain starts a throw-away MySQL 8 Docker container when PANGOLIN_TEST_DSN
+// is not already set, runs all tests, and tears the container down on exit.
+//
+// If Docker is unavailable the tests are skipped gracefully (requireDSN in
+// lifecycle_test.go handles the skip).
+func TestMain(m *testing.M) {
+	if os.Getenv("PANGOLIN_TEST_DSN") != "" {
+		// Caller already provided a DSN — run directly.
+		os.Exit(m.Run())
+	}
+
+	const (
+		containerName = "pangolin-lc-test-mysql"
+		hostPort      = "13399"
+		dbName        = "pangolin_test"
+		rootPwd       = "secret"
+	)
+
+	ctx := context.Background()
+
+	// Remove any leftover container from a previous crashed run.
+	_ = exec.CommandContext(ctx, "docker", "rm", "-f", containerName).Run()
+
+	// Start MySQL 8.
+	startCmd := exec.CommandContext(ctx, "docker", "run", "-d",
+		"--name", containerName,
+		"-e", "MYSQL_ROOT_PASSWORD="+rootPwd,
+		"-e", "MYSQL_DATABASE="+dbName,
+		"-p", hostPort+":3306",
+		"mysql:8",
+	)
+	if out, err := startCmd.CombinedOutput(); err != nil {
+		fmt.Fprintf(os.Stderr, "docker run failed: %v\n%s\n", err, out)
+		fmt.Fprintln(os.Stderr, "skipping lifecycle integration tests (no MySQL)")
+		os.Exit(0) // graceful skip
+	}
+
+	// Tear down on exit regardless of test outcome.
+	defer func() {
+		_ = exec.Command("docker", "stop", containerName).Run()
+		_ = exec.Command("docker", "rm", containerName).Run()
+	}()
+
+	// multiStatements=true is required by golang-migrate's MySQL driver so it
+	// can execute migration files containing multiple SQL statements in one call.
+	// store.Open / buildDSN preserves unrecognised params, so this flag
+	// survives the DSN round-trip through buildDSN.
+	dsn := fmt.Sprintf("root:%s@tcp(127.0.0.1:%s)/%s?multiStatements=true",
+		rootPwd, hostPort, dbName)
+
+	// Wait until MySQL accepts connections (up to 60 s).
+	deadline := time.Now().Add(60 * time.Second)
+	for time.Now().Before(deadline) {
+		ping := exec.CommandContext(ctx, "docker", "exec", containerName,
+			"mysqladmin", "ping", "--silent")
+		if err := ping.Run(); err == nil {
+			break
+		}
+		time.Sleep(2 * time.Second)
+	}
+
+	// Verify connectivity.
+	check := exec.CommandContext(ctx, "docker", "exec", containerName,
+		"mysqladmin", "ping", "--silent")
+	if err := check.Run(); err != nil {
+		fmt.Fprintln(os.Stderr, "MySQL did not become ready within 60 s; skipping tests")
+		os.Exit(0)
+	}
+
+	os.Setenv("PANGOLIN_TEST_DSN", dsn)
+
+	os.Exit(m.Run())
+}

server/run_mysql_test.sh

@@ -0,0 +1,41 @@
+#!/bin/bash
+# Start a MySQL 8 container, wait for it to be ready, run lifecycle integration tests, and clean up.
+set -e
+
+CONTAINER_NAME="lc-mysql-$(date +%s)"
+PORT=13307
+
+echo "Starting MySQL container: $CONTAINER_NAME on port $PORT"
+docker run -d \
+  --name "$CONTAINER_NAME" \
+  -e MYSQL_ROOT_PASSWORD=secret \
+  -e MYSQL_DATABASE=pangolin_test \
+  -p "$PORT:3306" \
+  mysql:8 >/dev/null
+
+cleanup() {
+  echo "Stopping container $CONTAINER_NAME"
+  docker stop "$CONTAINER_NAME" >/dev/null 2>&1 || true
+  docker rm   "$CONTAINER_NAME" >/dev/null 2>&1 || true
+}
+trap cleanup EXIT
+
+# Wait for MySQL to be ready (max 60 s)
+echo "Waiting for MySQL to be ready..."
+for i in $(seq 1 30); do
+  if docker exec "$CONTAINER_NAME" mysqladmin ping --silent 2>/dev/null; then
+    echo "MySQL ready after ${i}*2 seconds"
+    break
+  fi
+  sleep 2
+done
+
+DSN="root:secret@tcp(127.0.0.1:${PORT})/pangolin_test"
+echo "Running lifecycle integration tests with DSN=$DSN"
+
+PANGOLIN_TEST_DSN="$DSN" go test -tags integration \
+  ./internal/nodes/... \
+  -run "TestLifecycle" \
+  -v -count=1 -timeout 120s
+
+echo "Tests complete"