root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-4813): complete Step 4 — wire handoff into scheduler and lease recovery

Browse Source 
Fusion-Task-Id: FN-4813
Fusion-Task-Lineage: 846893a5-2afa-4817-8f64-8c444d2fd713
This commit is contained in:
Fusion (runfusion.ai)
2026-05-16 16:52:30 -07:00
committed by gsxdsm
parent b2ca02f743
commit 92b4899ee0
1 changed files with 50 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
packages/engine/src/scheduler.ts
View File

@@ -1038,21 +1038,60 @@ export class Scheduler {
policy: settings.unavailableNodePolicy,
});
if (!decision.allowed) {
if (!this.wasNodeBlocked.has(task.id)) {
this.wasNodeBlocked.add(task.id);
schedulerLog.log(`Task ${task.id} dispatch blocked — ${decision.reason}`);
await this.store.logEntry(task.id, decision.reason);
if (freshTask.checkoutNodeId) {
const ownerNodeHealth = this.options.nodeHealthMonitor.getNodeHealth(freshTask.checkoutNodeId);
if (ownerNodeHealth === "offline" || ownerNodeHealth === "error") {
const handoffDecision = decideOwningNodeHandoff({
task: freshTask,
ownerNodeId: freshTask.checkoutNodeId,
ownerNodeHealth,
localNodeId: settings.defaultNodeId ?? "local",
handoffPolicy: settings.owningNodeHandoffPolicy,
});
if (handoffDecision.action === "park") {
if (!this.wasNodeBlocked.has(task.id)) {
this.wasNodeBlocked.add(task.id);
schedulerLog.log(`Task ${task.id} dispatch blocked — ${handoffDecision.reason}`);
await this.store.logEntry(task.id, handoffDecision.reason);
}
continue;
}
if (handoffDecision.action === "reassign-local") {
effectiveNode = { nodeId: undefined, source: "local" };
await this.store.logEntry(task.id, `Owner handoff: ${handoffDecision.reason}`);
} else if (handoffDecision.action === "reassign-any") {
skipUnavailableNodePolicy = true;
await this.store.logEntry(task.id, `Owner handoff: ${handoffDecision.reason}`);
}
}
continue;
}
this.wasNodeBlocked.delete(task.id);
if (!skipUnavailableNodePolicy) {
const nodeHealth = this.options.nodeHealthMonitor.getNodeHealth(effectiveNode.nodeId);
const decision = applyUnavailableNodePolicy({
effectiveNode,
nodeHealth,
policy: settings.unavailableNodePolicy,
});
if (decision.fallbackToLocal) {
schedulerLog.log(`Task ${task.id} falling back to local — ${decision.reason}`);
await this.store.logEntry(task.id, decision.reason);
effectiveNode = { nodeId: undefined, source: "local" };
if (!decision.allowed) {
if (!this.wasNodeBlocked.has(task.id)) {
this.wasNodeBlocked.add(task.id);
schedulerLog.log(`Task ${task.id} dispatch blocked — ${decision.reason}`);
await this.store.logEntry(task.id, decision.reason);
}
continue;
}
this.wasNodeBlocked.delete(task.id);
if (decision.fallbackToLocal) {
schedulerLog.log(`Task ${task.id} falling back to local — ${decision.reason}`);
await this.store.logEntry(task.id, decision.reason);
effectiveNode = { nodeId: undefined, source: "local" };
}
}
}