sorcery-go/pkg/cluster/scheduler.go

// HPC Grid-Casting: distributed compilation across the Coven.
//
// When a spell is too heavy for a single Sanctum (think glibc, llvm, rust),
// the Master shards the build across every Worker that has spare "Mana".
// Workers compile their assigned translation units in isolated namespaces
// and stream the object files back. The Master links them and produces a
// single signed Essence.
//
// With Fester integration, the Scheduler delegates build dispatch and node
// selection to the Fester master's weighted/thermal/cache-aware scheduler.
// The Coven's Scheduler becomes a thin proxy that translates between
// sorcery-go's Shard model and Fester's build API.
package cluster

import (
        "context"
        "fmt"
        "sync"
)

// Shard is one slice of a Grid-Cast.
type Shard struct {
        ID       string
        Spell    string
        WorkerID string // arch hint or specific node name
        Cmd      string // build command to execute
        Dir      string // working directory
        Status   string // "queued", "compiling", "uploaded", "failed", "completed"
}

// Scheduler routes shards to Workers based on available ComputePower.
// When a Fester client is available, all scheduling is delegated to Fester.
type Scheduler struct {
        Coven *Coven
        mu    sync.Mutex
        queue []Shard
}

// NewScheduler wraps a Coven.
func NewScheduler(c *Coven) *Scheduler {
        return &Scheduler{Coven: c}
}

// Thermal derating constants.
const (
        thermalCritical float64 = 80 // °C — half-power threshold
        thermalHot      float64 = 70 // °C — 75% power threshold
        derateHalf      int    = 2  // effectivePower * 2 / 4 = 50%
        derateThreeQtr  int    = 3  // effectivePower * 3 / 4 = 75%
        thermalDivisor  int    = 4  // denominator for derate calculation
)

// thermalThreshold defines a temperature threshold and its derate factor.
// The array is ordered by threshold ascending; the first match wins.
var thermalThresholds = []struct {
        temp   float64
        derate int // effectivePower * derate / thermalDivisor
}{
        {thermalCritical, derateHalf},    // > 80°C  → half power
        {thermalHot, derateThreeQtr},     // > 70°C  → 75% power
}

// thermalDerate applies temperature-based derating to effective power.
func thermalDerate(power int, temp float64) int {
        for _, t := range thermalThresholds {
                if temp > t.temp {
                        return power * t.derate / thermalDivisor
                }
        }
        return power
}

// blockedPolicies is the set of scheduling policies that exclude a node.
// Used by both Fester-mode and standalone-mode best-worker selection.
var blockedPolicies = map[string]bool{
        "avoid":    true,
        "offline":  true,
        "draining": true,
}

// bestNodeByPower selects the node with the highest effective power,
// filtered by arch (if non-empty) and scheduling policy.
func bestNodeByPower(nodes []*Node, arch string) *Node {
        var best *Node
        bestPower := -1
        for _, n := range nodes {
                if arch != "" && n.Arch != arch {
                        continue
                }
                if blockedPolicies[n.Policy] || blockedPolicies[n.Status] {
                        continue
                }
                effectivePower := n.MaxJobs - n.ActiveBuilds
                effectivePower = thermalDerate(effectivePower, n.Temperature)
                if effectivePower > bestPower {
                        best = n
                        bestPower = effectivePower
                }
        }
        return best
}

// bestPeerByComputePower selects the best peer from the local Coven map
// for standalone mode. This is the simplified path without thermal data.
func bestPeerByComputePower(peers map[string]*Node, arch string) *Node {
        var best *Node
        bestPower := -1
        for _, p := range peers {
                if p.Arch != arch {
                        continue
                }
                if p.ComputePower > bestPower {
                        best = p
                        bestPower = p.ComputePower
                }
        }
        return best
}

// GetBestWorker picks the Worker with the most spare ComputePower for the
// requested arch. In Fester mode, this queries Fester's node list and
// picks the best candidate locally (a quick heuristic). The actual
// scheduling decision is made by Fester's optimizer when Dispatch() is
// called.
// Returns nil if no Worker matches.
func (s *Scheduler) GetBestWorker(arch string) *Node {
        if s.Coven.Fester != nil {
                nodes, err := s.Coven.Fester.GetNodes()
                if err == nil {
                        return bestNodeByPower(nodes, arch)
                }
                // Fallback to local state on error.
        }
        s.mu.Lock()
        defer s.mu.Unlock()
        return bestPeerByComputePower(s.Coven.Peers, arch)
}

// Dispatch sends a shard to the chosen Worker for execution.
//
// In Fester mode: first checks the shared CAS for a cached artifact.
// If the artifact exists (any runtime, any node), the shard is marked
// "completed" immediately without dispatching to Fester. This is the
// cross-runtime deduplication layer — an artifact built inside LXC on
// Node A is instantly available to a Firecracker microVM on Node B.
//
// If not cached, submits the shard as a Fester build via POST /api/build.
// Fester's scheduler picks the best node based on CPU, thermal, cache, and
// policy constraints. The build ID is tracked in the shard's ID field.
//
// In standalone mode: queues the shard in-memory (no actual execution).
//
// The actionHash parameter is the expected SHA-256 of the build output.
// If empty, the CAS check is skipped and the shard is always dispatched.
func (s *Scheduler) Dispatch(ctx context.Context, shard Shard, actionHash string) error {
        // Fester mode: check shared CAS first.
        if s.Coven.Fester != nil && actionHash != "" {
                entry, err := s.Coven.Fester.CAS.CheckArtifact(ctx, actionHash)
                if err == nil && entry != nil {
                        shard.ID = fmt.Sprintf("cas:%s", actionHash[:12])
                        shard.Status = "completed"
                        s.mu.Lock()
                        s.queue = append(s.queue, shard)
                        s.mu.Unlock()
                        return nil
                }
        }

        // Fester mode: delegate to Fester's build API.
        if s.Coven.Fester != nil {
                target := shard.Spell
                if shard.WorkerID != "" {
                        target = shard.WorkerID
                }
                cmd := shard.Cmd
                if cmd == "" {
                        cmd = fmt.Sprintf("make -j$(nproc) %s", shard.Spell)
                }

                build, err := s.Coven.Fester.SubmitBuild(ctx, target, cmd, shard.Dir, "")
                if err != nil {
                        return fmt.Errorf("scheduler: fester dispatch failed: %w", err)
                }

                shard.ID = build.ID
                shard.Status = "compiling"
                shard.WorkerID = build.Node

                s.mu.Lock()
                s.queue = append(s.queue, shard)
                s.mu.Unlock()
                return nil
        }

        // Standalone mode: pick best worker locally and queue.
        worker := s.GetBestWorker(shard.WorkerID)
        if worker == nil {
                return fmt.Errorf("scheduler: no worker available for %s", shard.WorkerID)
        }
        shard.WorkerID = worker.ID
        shard.Status = "compiling"
        s.mu.Lock()
        s.queue = append(s.queue, shard)
        s.mu.Unlock()
        return nil
}

// PulseQueue returns a copy of every active shard.
// Used by the Cockpit "Pulse" tab.
func (s *Scheduler) PulseQueue() []Shard {
        s.mu.Lock()
        defer s.mu.Unlock()
        return append([]Shard(nil), s.queue...)
}

// CancelShard cancels a dispatched shard. In Fester mode, this cancels the
// corresponding Fester build.
func (s *Scheduler) CancelShard(ctx context.Context, shardID string) error {
        if s.Coven.Fester != nil {
                return s.Coven.Fester.CancelBuild(shardID)
        }

        s.mu.Lock()
        defer s.mu.Unlock()
        // Linear scan — s.queue is not sorted by shard ID, so sort.Search
        // would give incorrect results. Linear is acceptable because the
        // queue is typically small (tens of shards, not millions).
        for i := range s.queue {
                if s.queue[i].ID == shardID {
                        s.queue[i].Status = "cancelled"
                        return nil
                }
        }
        return fmt.Errorf("scheduler: shard %s not found", shardID)
}

// SyncFromFester pulls the latest build statuses from Fester and updates
// the local shard queue. Call this periodically to keep the local state
// in sync with Fester's ground truth.
func (s *Scheduler) SyncFromFester(ctx context.Context) error {
        if s.Coven.Fester == nil {
                return nil
        }

        builds, err := s.Coven.Fester.ListBuilds()
        if err != nil {
                return fmt.Errorf("scheduler: sync from fester: %w", err)
        }

        festerStatus := make(map[string]string, len(builds))
        festerNode := make(map[string]string, len(builds))
        for _, b := range builds {
                festerStatus[b.ID] = b.Status
                festerNode[b.ID] = b.Node
        }

        s.mu.Lock()
        defer s.mu.Unlock()
        for i, sh := range s.queue {
                if status, ok := festerStatus[sh.ID]; ok {
                        s.queue[i].Status = status
                        if node, ok := festerNode[sh.ID]; ok && node != "" {
                                s.queue[i].WorkerID = node
                        }
                }
        }
        return nil
}

// WatchFesterEvents starts a background goroutine that listens to Fester's
// WebSocket event stream and updates shard status in real time.
// Returns a cancel function to stop the watcher.
func (s *Scheduler) WatchFesterEvents(ctx context.Context) (cancel func(), err error) {
        if s.Coven.Fester == nil {
                return func() {}, nil
        }

        wsCtx, wsCancel := context.WithCancel(ctx)

        go func() {
                s.Coven.Fester.WatchEvents(wsCtx, func(event FesterEvent) {
                        s.handleFesterEvent(event)
                })
        }()

        return wsCancel, nil
}

// festerEventStatus maps Fester event types to shard status updates.
// Events not in this map (node_offline, node_draining, etc.) are ignored
// by the handler — Fester handles their scheduling impact.
var festerEventStatus = map[string]string{
        "build_started":   "compiling",
        "build_completed": "completed",
        "build_failed":    "failed",
}

// handleFesterEvent updates the local shard queue based on Fester events.
func (s *Scheduler) handleFesterEvent(event FesterEvent) {
        newStatus, ok := festerEventStatus[event.Type]
        if !ok {
                return
        }

        s.mu.Lock()
        defer s.mu.Unlock()
        for i, sh := range s.queue {
                if sh.ID == event.BuildID {
                        s.queue[i].Status = newStatus
                        if event.Node != "" {
                                s.queue[i].WorkerID = event.Node
                        }
                        return
                }
        }
}