diff --git a/server/internal/nodes/lifecycle.go b/server/internal/nodes/lifecycle.go new file mode 100644 index 0000000..1a0f4a4 --- /dev/null +++ b/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= WHERE uuid=? AND status= (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 +} diff --git a/server/internal/nodes/lifecycle_test.go b/server/internal/nodes/lifecycle_test.go new file mode 100644 index 0000000..adf2f93 --- /dev/null +++ b/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=). +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) + } +} diff --git a/server/internal/nodes/testmain_test.go b/server/internal/nodes/testmain_test.go new file mode 100644 index 0000000..191498f --- /dev/null +++ b/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()) +} diff --git a/server/run_mysql_test.sh b/server/run_mysql_test.sh new file mode 100644 index 0000000..0bad607 --- /dev/null +++ b/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"