dag.js (3213B)
1 // src/lib/pipeline/dag.js - dependency graph checks and traversal 2 // 3 // The prototype stored a single integer level per task and treated the lowest 4 // queued level as the next runnable task. That serialized every run and, on 5 // failure, only skipped direct dependents, so a transitive dependent could 6 // still be dispatched with its dependency missing. Both problems come from 7 // throwing the graph away, so the graph is kept here and used directly. 8 9 export class CycleError extends Error { 10 constructor(cycle) { 11 super(`pipeline contains a dependency cycle: ${cycle.join(' -> ')}`); 12 this.name = 'CycleError'; 13 this.cycle = cycle; 14 } 15 } 16 17 // Returns the tasks in an order where every dependency precedes its 18 // dependents. Throws CycleError naming the cycle when none exists. 19 export function topologicalOrder(tasks) { 20 const byName = new Map(tasks.map((j) => [j.name, j])); 21 const state = new Map(); 22 const order = []; 23 const stack = []; 24 25 function visit(name) { 26 const current = state.get(name); 27 if (current === 'done') return; 28 if (current === 'active') { 29 const from = stack.indexOf(name); 30 throw new CycleError([...stack.slice(from), name]); 31 } 32 33 state.set(name, 'active'); 34 stack.push(name); 35 for (const dep of byName.get(name).needs) { 36 if (byName.has(dep)) visit(dep); 37 } 38 stack.pop(); 39 state.set(name, 'done'); 40 order.push(name); 41 } 42 43 // Sorted so that an unchanged pipeline always produces the same order. 44 for (const name of [...byName.keys()].sort()) visit(name); 45 return order.map((name) => byName.get(name)); 46 } 47 48 // Depth of each task, where a task with no dependencies is 0. Used only for 49 // display; scheduling reads the edges, never the depth. 50 export function depths(tasks) { 51 const byName = new Map(tasks.map((j) => [j.name, j])); 52 const out = new Map(); 53 for (const task of topologicalOrder(tasks)) { 54 const deps = task.needs.filter((d) => byName.has(d)); 55 out.set(task.name, deps.length === 0 ? 0 : Math.max(...deps.map((d) => out.get(d))) + 1); 56 } 57 return out; 58 } 59 60 // Every task reachable by following dependents from the given names. This is 61 // what a failure must skip: direct dependents alone leave transitive ones 62 // runnable against a dependency that never produced anything. 63 export function transitiveDependents(tasks, startNames) { 64 const dependents = new Map(tasks.map((j) => [j.name, []])); 65 for (const task of tasks) { 66 for (const dep of task.needs) { 67 if (dependents.has(dep)) dependents.get(dep).push(task.name); 68 } 69 } 70 71 const seen = new Set(); 72 const queue = [...startNames]; 73 while (queue.length > 0) { 74 const name = queue.shift(); 75 for (const child of dependents.get(name) ?? []) { 76 if (seen.has(child)) continue; 77 seen.add(child); 78 queue.push(child); 79 } 80 } 81 return seen; 82 } 83 84 // Tasks whose dependencies have all reached a satisfying state. Passed the 85 // current state of every task by name. 86 export function runnable(tasks, stateByName, { satisfied = ['success'] } = {}) { 87 return tasks.filter((task) => { 88 if (stateByName.get(task.name) !== 'queued') return false; 89 return task.needs.every((dep) => satisfied.includes(stateByName.get(dep))); 90 }); 91 }