perspt_sdk/
scheduler.rs

1//! Dependency-aware parallel ready-queue scheduler (PSP-8 System 4).
2//!
3//! The scheduler runs independent work in parallel and serializes only work that
4//! conflicts through dependencies, leases, or non-commuting durable effects. Two
5//! durable turns commute — and need no relative order — only when their semantic
6//! read and write footprints are disjoint:
7//!
8//! ```text
9//! writes(e) ∩ reads(e') = ∅  and  writes(e') ∩ reads(e) = ∅.
10//! ```
11//!
12//! The conflict footprint includes durable platform state, not only workspace
13//! files: the capability table, risk budgets, fresh-identifier allocation, and
14//! the ledger root are modeled as read/write resources.
15
16use std::collections::{BTreeSet, HashMap, HashSet};
17
18use serde::{Deserialize, Serialize};
19
20use crate::certificate::ResidualCertificate;
21use crate::workgraph::{WorkGraphRevision, WorkNode, WorkNodeState};
22
23/// A durable resource that turns read or write. Overlapping footprints force
24/// serialization (PSP-8 System 4 / Theorem 8).
25#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
26pub enum Resource {
27    File(String),
28    Interface(String),
29    Manifest(String),
30    Lockfile(String),
31    Migration(String),
32    TestFixture(String),
33    Toolchain(String),
34    /// An extension-defined resource family keyed by validated arguments.
35    Scoped {
36        family: String,
37        key: String,
38    },
39    /// A specific capability in the capability table `Γ`.
40    Capability(String),
41    /// A named risk budget.
42    RiskBudget(String),
43    /// The fresh-identifier allocator.
44    FreshIdAllocator,
45    /// The ledger root (Merkle head).
46    LedgerRoot,
47    /// The mutable work graph (PSP-9 system 15).
48    WorkGraph,
49    /// One provider's rate limit — a *throughput* resource, not a component
50    /// of the semantic durable state. Contention on it is a scheduling delay
51    /// and is never recorded as a required serialization (Gate P).
52    ProviderRateLimit(String),
53    /// Fallback for tools whose touched state cannot be described: conflicts
54    /// with everything, so an undeclared footprint serializes rather than
55    /// races (PSP-9 system 5).
56    OpaqueWorkspace,
57}
58
59/// The read/write footprint of a turn.
60#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
61pub struct Footprint {
62    pub reads: BTreeSet<Resource>,
63    pub writes: BTreeSet<Resource>,
64}
65
66impl Footprint {
67    pub fn new() -> Self {
68        Self::default()
69    }
70
71    pub fn read(mut self, r: Resource) -> Self {
72        self.reads.insert(r);
73        self
74    }
75
76    pub fn write(mut self, r: Resource) -> Self {
77        self.writes.insert(r);
78        self
79    }
80
81    /// Whether this turn commutes with `other`: write/read and write/write
82    /// footprints must be disjoint in both directions.
83    pub fn commutes_with(&self, other: &Footprint) -> bool {
84        // An opaque workspace footprint conflicts with everything mutating:
85        // an undeclared footprint serializes rather than races (PSP-9).
86        let opaque = |f: &Footprint| f.writes.contains(&Resource::OpaqueWorkspace);
87        let touches = |f: &Footprint| !f.writes.is_empty();
88        if (opaque(self) && touches(other)) || (opaque(other) && touches(self)) {
89            return false;
90        }
91        self.writes.is_disjoint(&other.reads)
92            && other.writes.is_disjoint(&self.reads)
93            && self.writes.is_disjoint(&other.writes)
94    }
95
96    pub fn conflicts_with(&self, other: &Footprint) -> bool {
97        !self.commutes_with(other)
98    }
99}
100
101/// What a lease protects.
102#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
103#[serde(rename_all = "snake_case")]
104pub enum LeaseKind {
105    /// Serializes all graph revisions against each other.
106    GraphWrite,
107    /// Package-manager / dependency mutation.
108    Toolchain,
109    /// Exclusive access to a workspace resource.
110    Resource,
111}
112
113/// A held lease.
114#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
115pub struct ExecutionLease {
116    pub lease_id: String,
117    pub holder_work_id: String,
118    pub kind: LeaseKind,
119    pub scope: Resource,
120}
121
122/// The lease table, tracking exclusive grants.
123#[derive(Debug, Clone, Default)]
124pub struct LeaseTable {
125    held: Vec<ExecutionLease>,
126}
127
128impl LeaseTable {
129    pub fn new() -> Self {
130        Self::default()
131    }
132
133    /// Whether a lease over `scope` is available (not already held by another).
134    pub fn is_available(&self, scope: &Resource) -> bool {
135        !self.held.iter().any(|l| &l.scope == scope)
136    }
137
138    /// Acquire a lease, returning its id, or `None` if it is unavailable.
139    pub fn acquire(&mut self, holder: &str, kind: LeaseKind, scope: Resource) -> Option<String> {
140        if !self.is_available(&scope) {
141            return None;
142        }
143        let lease_id = uuid::Uuid::new_v4().to_string();
144        self.held.push(ExecutionLease {
145            lease_id: lease_id.clone(),
146            holder_work_id: holder.to_string(),
147            kind,
148            scope,
149        });
150        Some(lease_id)
151    }
152
153    pub fn release(&mut self, lease_id: &str) {
154        self.held.retain(|l| l.lease_id != lease_id);
155    }
156
157    pub fn held_count(&self) -> usize {
158        self.held.len()
159    }
160}
161
162/// A repair outcome that becomes durable scheduler work (PSP-8 System 4).
163#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
164#[serde(tag = "action", rename_all = "snake_case")]
165pub enum RepairAction {
166    RetryNode {
167        node_id: String,
168        generation: u32,
169    },
170    ExpandScope {
171        node_id: String,
172        generation: u32,
173        added_paths: Vec<String>,
174    },
175    SplitNode {
176        node_id: String,
177        generation: u32,
178        child_goals: Vec<String>,
179    },
180    InsertInterfaceNode {
181        boundary: String,
182    },
183    AddNode {
184        goal: String,
185        reason: String,
186    },
187    RetireNode {
188        node_id: String,
189        generation: u32,
190        reason: String,
191    },
192    ReplanSubgraph {
193        root: String,
194        affected: Vec<String>,
195    },
196    StopNode {
197        node_id: String,
198        generation: u32,
199        certificate_id: String,
200    },
201}
202
203/// A durable scheduler command consumed by the ready-queue loop (PSP-8
204/// `SchedulerEffect`).
205#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
206#[serde(tag = "effect", rename_all = "snake_case")]
207pub enum SchedulerEffect {
208    CommitNode {
209        node_id: String,
210        generation: u32,
211    },
212    RequeueNode {
213        node_id: String,
214        generation: u32,
215        reason: String,
216    },
217    ApplyGraphRevision {
218        revision_id: String,
219    },
220    SpawnWork {
221        work_id: String,
222    },
223    CancelWork {
224        work_id: String,
225        reason: String,
226    },
227    RequestApproval {
228        proposal_id: String,
229    },
230    StopWithCertificate {
231        certificate_id: String,
232    },
233}
234
235/// Convert a repair action into scheduler effects. A local repair that produces
236/// new executable work SHALL NOT escalate; it returns effects the ready-queue
237/// loop consumes (PSP-8 System 4).
238pub fn repair_to_effects(action: &RepairAction) -> Vec<SchedulerEffect> {
239    match action {
240        RepairAction::RetryNode {
241            node_id,
242            generation,
243        } => vec![SchedulerEffect::RequeueNode {
244            node_id: node_id.clone(),
245            generation: *generation,
246            reason: "retry".into(),
247        }],
248        RepairAction::ExpandScope {
249            node_id,
250            generation,
251            ..
252        } => vec![SchedulerEffect::RequeueNode {
253            node_id: node_id.clone(),
254            generation: generation + 1,
255            reason: "scope expanded".into(),
256        }],
257        RepairAction::SplitNode { child_goals, .. } => child_goals
258            .iter()
259            .map(|_| SchedulerEffect::SpawnWork {
260                work_id: uuid::Uuid::new_v4().to_string(),
261            })
262            .chain(std::iter::once(SchedulerEffect::ApplyGraphRevision {
263                revision_id: uuid::Uuid::new_v4().to_string(),
264            }))
265            .collect(),
266        RepairAction::InsertInterfaceNode { .. } | RepairAction::AddNode { .. } => {
267            vec![
268                SchedulerEffect::SpawnWork {
269                    work_id: uuid::Uuid::new_v4().to_string(),
270                },
271                SchedulerEffect::ApplyGraphRevision {
272                    revision_id: uuid::Uuid::new_v4().to_string(),
273                },
274            ]
275        }
276        RepairAction::RetireNode { .. } | RepairAction::ReplanSubgraph { .. } => {
277            vec![SchedulerEffect::ApplyGraphRevision {
278                revision_id: uuid::Uuid::new_v4().to_string(),
279            }]
280        }
281        RepairAction::StopNode { certificate_id, .. } => {
282            vec![SchedulerEffect::StopWithCertificate {
283                certificate_id: certificate_id.clone(),
284            }]
285        }
286    }
287}
288
289/// A node currently executing, with the footprint it occupies.
290#[derive(Debug, Clone)]
291pub struct RunningTask {
292    pub node_id: String,
293    pub generation: u32,
294    pub footprint: Footprint,
295}
296
297/// The mutable parallel scheduler.
298#[derive(Debug)]
299pub struct Scheduler {
300    max_parallel: usize,
301    running: Vec<RunningTask>,
302    pub leases: LeaseTable,
303}
304
305impl Scheduler {
306    pub fn new(max_parallel: usize) -> Self {
307        Self {
308            max_parallel: max_parallel.max(1),
309            running: Vec::new(),
310            leases: LeaseTable::new(),
311        }
312    }
313
314    pub fn running_count(&self) -> usize {
315        self.running.len()
316    }
317
318    /// Compute the ready set: pending nodes whose dependencies are all accepted,
319    /// whose required sensors are resolved, and whose footprints do not conflict
320    /// with currently-running tasks. `footprint_of` supplies each node's
321    /// footprint (domain-derived from output targets and edges).
322    pub fn ready_nodes<'a, F>(
323        &self,
324        revision: &'a WorkGraphRevision,
325        footprint_of: F,
326    ) -> Vec<&'a WorkNode>
327    where
328        F: Fn(&WorkNode) -> Footprint,
329    {
330        let accepted: HashSet<&str> = revision
331            .nodes
332            .iter()
333            .filter(|n| n.is_accepted())
334            .map(|n| n.node_id.as_str())
335            .collect();
336
337        let mut ready = Vec::new();
338        // Footprints occupied by running tasks plus already-selected ready nodes,
339        // so two conflicting ready nodes are not dispatched together.
340        let mut occupied: Vec<Footprint> =
341            self.running.iter().map(|t| t.footprint.clone()).collect();
342        let slots = self.max_parallel.saturating_sub(self.running.len());
343
344        for node in &revision.nodes {
345            if ready.len() >= slots {
346                break;
347            }
348            if !matches!(node.state, WorkNodeState::Pending | WorkNodeState::Ready) {
349                continue;
350            }
351            // Required sensors resolved?
352            if !node.required_sensors.is_empty() {
353                continue;
354            }
355            // All dependencies accepted?
356            let deps = revision.dependencies_of(&node.node_id);
357            if !deps.iter().all(|d| accepted.contains(d)) {
358                continue;
359            }
360            // Footprint free?
361            let fp = footprint_of(node);
362            if occupied.iter().any(|o| o.conflicts_with(&fp)) {
363                continue;
364            }
365            occupied.push(fp);
366            ready.push(node);
367        }
368        ready
369    }
370
371    /// Mark a node as running, occupying its footprint.
372    pub fn start(&mut self, node: &WorkNode, footprint: Footprint) {
373        self.running.push(RunningTask {
374            node_id: node.node_id.clone(),
375            generation: node.generation,
376            footprint,
377        });
378    }
379
380    /// Mark a running node as finished, freeing its footprint.
381    pub fn finish(&mut self, node_id: &str, generation: u32) {
382        self.running
383            .retain(|t| !(t.node_id == node_id && t.generation == generation));
384    }
385}
386
387/// A terminal classification for a node generation, ensuring recovery is total:
388/// every node ends accepted, certified-stopped, or escalated (PSP-8 System 4).
389#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
390#[serde(tag = "outcome", rename_all = "snake_case")]
391pub enum NodeOutcome {
392    Committed { node_id: String, generation: u32 },
393    Stopped { certificate: ResidualCertificate },
394    Escalated { node_id: String, reason: String },
395}
396
397/// Validate that a set of node outcomes covers every node exactly once with a
398/// terminal classification (recovery totality check).
399pub fn recovery_is_total(revision: &WorkGraphRevision, outcomes: &[NodeOutcome]) -> bool {
400    let classified: HashMap<&str, usize> = {
401        let mut m: HashMap<&str, usize> = HashMap::new();
402        for o in outcomes {
403            let id = match o {
404                NodeOutcome::Committed { node_id, .. } => node_id.as_str(),
405                NodeOutcome::Stopped { certificate } => certificate.node_id.as_str(),
406                NodeOutcome::Escalated { node_id, .. } => node_id.as_str(),
407            };
408            *m.entry(id).or_default() += 1;
409        }
410        m
411    };
412    revision
413        .nodes
414        .iter()
415        .all(|n| classified.get(n.node_id.as_str()) == Some(&1))
416}
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421    use crate::workgraph::{EdgeKind, GraphRevisionReason, NodeClass, WorkEdge};
422
423    fn node(id: &str) -> WorkNode {
424        WorkNode::new(id, format!("goal {id}"), NodeClass::Implement)
425    }
426
427    fn rev(nodes: Vec<WorkNode>, edges: Vec<WorkEdge>) -> WorkGraphRevision {
428        WorkGraphRevision::build(0, None, GraphRevisionReason::InitialPlan, nodes, edges).unwrap()
429    }
430
431    #[test]
432    fn disjoint_footprints_commute() {
433        let a = Footprint::new().write(Resource::File("a.rs".into()));
434        let b = Footprint::new().write(Resource::File("b.rs".into()));
435        assert!(a.commutes_with(&b));
436    }
437
438    #[test]
439    fn write_read_overlap_does_not_commute() {
440        let a = Footprint::new().write(Resource::File("shared.rs".into()));
441        let b = Footprint::new().read(Resource::File("shared.rs".into()));
442        assert!(a.conflicts_with(&b));
443    }
444
445    #[test]
446    fn manifest_writes_serialize() {
447        let a = Footprint::new().write(Resource::Manifest("Cargo.toml".into()));
448        let b = Footprint::new().write(Resource::Manifest("Cargo.toml".into()));
449        assert!(a.conflicts_with(&b));
450    }
451
452    #[test]
453    fn capability_table_is_a_conflict_resource() {
454        // A capability grant conflicts with a commit that reads that capability.
455        let grant = Footprint::new().write(Resource::Capability("write-src".into()));
456        let commit = Footprint::new().read(Resource::Capability("write-src".into()));
457        assert!(grant.conflicts_with(&commit));
458    }
459
460    #[test]
461    fn ledger_root_and_fresh_id_serialize() {
462        let a = Footprint::new().write(Resource::LedgerRoot);
463        let b = Footprint::new().write(Resource::LedgerRoot);
464        assert!(a.conflicts_with(&b));
465        let c = Footprint::new().write(Resource::FreshIdAllocator);
466        let d = Footprint::new().write(Resource::FreshIdAllocator);
467        assert!(c.conflicts_with(&d));
468    }
469
470    #[test]
471    fn independent_nodes_are_ready_in_parallel() {
472        let nodes = vec![node("a"), node("b")];
473        let revision = rev(nodes, vec![]);
474        let sched = Scheduler::new(4);
475        let fp = |n: &WorkNode| Footprint::new().write(Resource::File(format!("{}.rs", n.node_id)));
476        let ready = sched.ready_nodes(&revision, fp);
477        assert_eq!(ready.len(), 2);
478    }
479
480    #[test]
481    fn dependent_node_waits_for_its_predecessor() {
482        let nodes = vec![node("a"), node("b")];
483        let edges = vec![WorkEdge::new("a", "b", EdgeKind::RequiresArtifact)];
484        let revision = rev(nodes, edges);
485        let sched = Scheduler::new(4);
486        let fp = |n: &WorkNode| Footprint::new().write(Resource::File(format!("{}.rs", n.node_id)));
487        let ready = sched.ready_nodes(&revision, fp);
488        // Only `a` is ready; `b` depends on (not-yet-accepted) `a`.
489        assert_eq!(ready.len(), 1);
490        assert_eq!(ready[0].node_id, "a");
491    }
492
493    #[test]
494    fn conflicting_footprints_do_not_dispatch_together() {
495        let nodes = vec![node("a"), node("b")];
496        let revision = rev(nodes, vec![]);
497        let sched = Scheduler::new(4);
498        // Both touch the same manifest -> only one is ready at a time.
499        let fp = |_n: &WorkNode| Footprint::new().write(Resource::Manifest("Cargo.toml".into()));
500        let ready = sched.ready_nodes(&revision, fp);
501        assert_eq!(ready.len(), 1);
502    }
503
504    #[test]
505    fn inserted_node_becomes_ready_after_revision() {
506        // The static-snapshot bug: a node inserted by a repair must be executed.
507        let mut nodes = vec![node("a")];
508        nodes[0].state = WorkNodeState::Stable;
509        let revision = rev(nodes, vec![]);
510        // Repair inserts node "b".
511        let mut nodes2 = revision.nodes.clone();
512        nodes2.push(node("b"));
513        let revision2 = WorkGraphRevision::build(
514            1,
515            Some(revision.revision_id.clone()),
516            GraphRevisionReason::LocalRepair,
517            nodes2,
518            vec![],
519        )
520        .unwrap();
521        let sched = Scheduler::new(4);
522        let fp = |n: &WorkNode| Footprint::new().write(Resource::File(format!("{}.rs", n.node_id)));
523        let ready = sched.ready_nodes(&revision2, fp);
524        assert!(
525            ready.iter().any(|n| n.node_id == "b"),
526            "inserted node must be ready"
527        );
528    }
529
530    #[test]
531    fn leases_are_exclusive() {
532        let mut table = LeaseTable::new();
533        let scope = Resource::Toolchain("cargo".into());
534        let l1 = table.acquire("w1", LeaseKind::Toolchain, scope.clone());
535        assert!(l1.is_some());
536        assert!(table
537            .acquire("w2", LeaseKind::Toolchain, scope.clone())
538            .is_none());
539        table.release(&l1.unwrap());
540        assert!(table.acquire("w2", LeaseKind::Toolchain, scope).is_some());
541    }
542
543    #[test]
544    fn repair_retry_becomes_requeue_effect() {
545        let effects = repair_to_effects(&RepairAction::RetryNode {
546            node_id: "a".into(),
547            generation: 0,
548        });
549        assert_eq!(
550            effects,
551            vec![SchedulerEffect::RequeueNode {
552                node_id: "a".into(),
553                generation: 0,
554                reason: "retry".into()
555            }]
556        );
557    }
558
559    #[test]
560    fn split_produces_spawn_and_revision_effects() {
561        let effects = repair_to_effects(&RepairAction::SplitNode {
562            node_id: "a".into(),
563            generation: 0,
564            child_goals: vec!["x".into(), "y".into()],
565        });
566        let spawns = effects
567            .iter()
568            .filter(|e| matches!(e, SchedulerEffect::SpawnWork { .. }))
569            .count();
570        let revs = effects
571            .iter()
572            .filter(|e| matches!(e, SchedulerEffect::ApplyGraphRevision { .. }))
573            .count();
574        assert_eq!(spawns, 2);
575        assert_eq!(revs, 1);
576    }
577
578    #[test]
579    fn recovery_totality_requires_every_node_classified() {
580        let nodes = vec![node("a"), node("b")];
581        let revision = rev(nodes, vec![]);
582        let outcomes = vec![NodeOutcome::Committed {
583            node_id: "a".into(),
584            generation: 0,
585        }];
586        assert!(!recovery_is_total(&revision, &outcomes));
587        let outcomes = vec![
588            NodeOutcome::Committed {
589                node_id: "a".into(),
590                generation: 0,
591            },
592            NodeOutcome::Escalated {
593                node_id: "b".into(),
594                reason: "blocked".into(),
595            },
596        ];
597        assert!(recovery_is_total(&revision, &outcomes));
598    }
599}
perspt_sdk/scheduler.rs

perspt_sdk/
scheduler.rs
