// 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 } } }