package nodes import ( "context" "database/sql" "encoding/json" "errors" "fmt" "time" "github.com/redis/go-redis/v9" dbx "github.com/wangjia/pangolin/server/internal/db" "github.com/wangjia/pangolin/server/internal/mtls" "github.com/wangjia/pangolin/server/internal/store" ) // 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 dialect dbx.Dialect 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, dialect: dbx.DialectForDB(db), crl: crl, rdb: rdb} } // BumpVersion atomically increments the directory_version singleton within tx. // Delegates to the canonical store.BumpDirectoryVersion (dialect-aware). func BumpVersion(ctx context.Context, tx *sql.Tx, d dbx.Dialect) error { return store.BumpDirectoryVersion(ctx, tx, d) } // 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 (?, ?, ?, ?)`, nodeID, string(eventBlockedSuspect), detailJSON, time.Now().UTC(), ); err != nil { return fmt.Errorf("lifecycle.MarkBlockedSuspect: insert event: %w", err) } // Bump directory version. if err := BumpVersion(ctx, tx, l.dialect); 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 (?, ?, ?, ?)`, nodeID, string(spec.event), detailJSON, time.Now().UTC(), ); 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, l.dialect); err != nil { return err } return tx.Commit() } // No spec matched → illegal transition from the node's current state. return ErrInvalidTransition }