feat(FN-3866): add changeset for clearing stale blocked-by relationships
Merges the changeset for a patch release of `@runfusion/fusion` related to clearing stale blocked-by relationships (FN-3864). Fusion-Task-Id: FN-3866
This commit is contained in:
5
.changeset/fn-3864-clear-stale-blocked-by.md
Normal file
5
.changeset/fn-3864-clear-stale-blocked-by.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@runfusion/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
SelfHealingManager now includes a `clearStaleBlockedBy()` recovery sweep that clears `blockedBy` (and transient `status`) on todo tasks when their blocker is missing, done, archived, paused in-review, or failed in-review with merge retries exhausted. This lets the scheduler re-evaluate those tasks cleanly on subsequent ticks instead of leaving them permanently queued behind stale blockers.
|
||||||
@@ -3277,6 +3277,185 @@ describe("SelfHealingManager", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("clearStaleBlockedBy", () => {
|
||||||
|
function createRunningStore() {
|
||||||
|
return createMockStore({
|
||||||
|
getSettings: vi.fn().mockResolvedValue({
|
||||||
|
autoUnpauseEnabled: false,
|
||||||
|
maintenanceIntervalMs: 0,
|
||||||
|
globalPause: false,
|
||||||
|
enginePaused: false,
|
||||||
|
} as unknown as Settings),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
function createTask(id: string, overrides: Record<string, unknown> = {}) {
|
||||||
|
return {
|
||||||
|
id,
|
||||||
|
column: "todo",
|
||||||
|
paused: false,
|
||||||
|
blockedBy: null,
|
||||||
|
mergeRetries: 0,
|
||||||
|
...overrides,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
it("clears stale blockedBy when blocker is missing", async () => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
const taskA = createTask("A", { blockedBy: "FN-MISSING" });
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([taskA]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const recovered = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(recovered).toBe(1);
|
||||||
|
expect(store.updateTask).toHaveBeenCalledWith("A", { blockedBy: null, status: null });
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining("FN-MISSING"));
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining("missing"));
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it.each(["done", "archived"] as const)("clears stale blockedBy when blocker is %s", async (column) => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
const blockerId = "FN-100";
|
||||||
|
const taskA = createTask("A", { blockedBy: blockerId });
|
||||||
|
const taskB = createTask(blockerId, { column });
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([taskA, taskB]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const recovered = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(recovered).toBe(1);
|
||||||
|
expect(store.updateTask).toHaveBeenCalledWith("A", { blockedBy: null, status: null });
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining(blockerId));
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining(column));
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("clears stale blockedBy when blocker is in-review and paused", async () => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
const blockerId = "FN-200";
|
||||||
|
const taskA = createTask("A", { blockedBy: blockerId });
|
||||||
|
const taskB = createTask(blockerId, { column: "in-review", paused: true });
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([taskA, taskB]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const recovered = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(recovered).toBe(1);
|
||||||
|
expect(store.updateTask).toHaveBeenCalledWith("A", { blockedBy: null, status: null });
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining(blockerId));
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining("in-review + paused"));
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("clears stale blockedBy when blocker is in-review failed with exhausted retries", async () => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
const blockerId = "FN-300";
|
||||||
|
const taskA = createTask("A", { blockedBy: blockerId });
|
||||||
|
const taskB = createTask(blockerId, { column: "in-review", status: "failed", mergeRetries: 3 });
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([taskA, taskB]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const recovered = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(recovered).toBe(1);
|
||||||
|
expect(store.updateTask).toHaveBeenCalledWith("A", { blockedBy: null, status: null });
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining(blockerId));
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining("mergeRetries 3/3"));
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not clear blockedBy when blocker is in-progress", async () => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
const taskA = createTask("A", { blockedBy: "FN-400" });
|
||||||
|
const taskB = createTask("FN-400", { column: "in-progress" });
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([taskA, taskB]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const recovered = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(recovered).toBe(0);
|
||||||
|
expect(store.updateTask).not.toHaveBeenCalled();
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not clear blockedBy when blocker is in-review and not paused/failed", async () => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
const taskA = createTask("A", { blockedBy: "FN-500" });
|
||||||
|
const taskB = createTask("FN-500", { column: "in-review", paused: false, mergeRetries: 0 });
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([taskA, taskB]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const recovered = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(recovered).toBe(0);
|
||||||
|
expect(store.updateTask).not.toHaveBeenCalled();
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not clear blockedBy when blocker failed retries are below threshold", async () => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
const taskA = createTask("A", { blockedBy: "FN-600" });
|
||||||
|
const taskB = createTask("FN-600", { column: "in-review", status: "failed", mergeRetries: 1 });
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([taskA, taskB]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const recovered = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(recovered).toBe(0);
|
||||||
|
expect(store.updateTask).not.toHaveBeenCalled();
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it.each([
|
||||||
|
{ settings: { globalPause: true }, label: "globalPause" },
|
||||||
|
{ settings: { enginePaused: true }, label: "enginePaused" },
|
||||||
|
])("returns 0 when $label is active", async ({ settings }) => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
|
||||||
|
autoUnpauseEnabled: false,
|
||||||
|
maintenanceIntervalMs: 0,
|
||||||
|
globalPause: false,
|
||||||
|
enginePaused: false,
|
||||||
|
...settings,
|
||||||
|
} as unknown as Settings);
|
||||||
|
const taskA = createTask("A", { blockedBy: "FN-700" });
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([taskA]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const recovered = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(recovered).toBe(0);
|
||||||
|
expect(store.updateTask).not.toHaveBeenCalled();
|
||||||
|
expect(store.logEntry).not.toHaveBeenCalled();
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("is idempotent after first stale blockedBy recovery", async () => {
|
||||||
|
const store = createRunningStore();
|
||||||
|
const blockerId = "FN-800";
|
||||||
|
const blocked = createTask("A", { blockedBy: blockerId });
|
||||||
|
const blocker = createTask(blockerId, { column: "done" });
|
||||||
|
const recoveredState = createTask("A", { blockedBy: null });
|
||||||
|
|
||||||
|
(store.listTasks as ReturnType<typeof vi.fn>)
|
||||||
|
.mockResolvedValueOnce([blocked, blocker])
|
||||||
|
.mockResolvedValueOnce([blocked, blocker])
|
||||||
|
.mockResolvedValueOnce([recoveredState, blocker]);
|
||||||
|
|
||||||
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||||
|
const first = await manager.clearStaleBlockedBy();
|
||||||
|
const second = await manager.clearStaleBlockedBy();
|
||||||
|
|
||||||
|
expect(first).toBe(1);
|
||||||
|
expect(second).toBe(0);
|
||||||
|
expect(store.updateTask).toHaveBeenCalledTimes(1);
|
||||||
|
expect(store.logEntry).toHaveBeenCalledTimes(1);
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("stale triage processing eviction before recovery", () => {
|
describe("stale triage processing eviction before recovery", () => {
|
||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
vi.useFakeTimers();
|
vi.useFakeTimers();
|
||||||
@@ -3740,13 +3919,14 @@ describe("maintenance cycle concurrency", () => {
|
|||||||
makeSlow("recoverGhostReviewTasks");
|
makeSlow("recoverGhostReviewTasks");
|
||||||
makeSlow("recoverOrphanedAgents");
|
makeSlow("recoverOrphanedAgents");
|
||||||
makeSlow("recoverStaleHeartbeatRuns");
|
makeSlow("recoverStaleHeartbeatRuns");
|
||||||
|
makeSlow("clearStaleBlockedBy");
|
||||||
|
|
||||||
await (manager as any).runMaintenance();
|
await (manager as any).runMaintenance();
|
||||||
|
|
||||||
// Operations run sequentially (one at a time), not in parallel.
|
// Operations run sequentially (one at a time), not in parallel.
|
||||||
expect(maxConcurrent).toBe(1);
|
expect(maxConcurrent).toBe(1);
|
||||||
// All operations should have run (including last one)
|
// All operations should have run (including last one)
|
||||||
expect(executionOrder[executionOrder.length - 1]).toBe("recoverStaleHeartbeatRuns");
|
expect(executionOrder[executionOrder.length - 1]).toBe("clearStaleBlockedBy");
|
||||||
});
|
});
|
||||||
|
|
||||||
it("one failing batch 2 operation does not abort the batch", async () => {
|
it("one failing batch 2 operation does not abort the batch", async () => {
|
||||||
@@ -3766,6 +3946,7 @@ describe("maintenance cycle concurrency", () => {
|
|||||||
"recoverGhostReviewTasks",
|
"recoverGhostReviewTasks",
|
||||||
"recoverOrphanedAgents",
|
"recoverOrphanedAgents",
|
||||||
"recoverStaleHeartbeatRuns",
|
"recoverStaleHeartbeatRuns",
|
||||||
|
"clearStaleBlockedBy",
|
||||||
] as const;
|
] as const;
|
||||||
|
|
||||||
// Make one operation fail
|
// Make one operation fail
|
||||||
|
|||||||
@@ -842,6 +842,7 @@ export class SelfHealingManager {
|
|||||||
{ name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() },
|
{ name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() },
|
||||||
{ name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents() },
|
{ name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents() },
|
||||||
{ name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns() },
|
{ name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns() },
|
||||||
|
{ name: "clear-stale-blocked-by", fn: () => this.clearStaleBlockedBy() },
|
||||||
];
|
];
|
||||||
for (const fn of batch2Fns) {
|
for (const fn of batch2Fns) {
|
||||||
try {
|
try {
|
||||||
@@ -1063,6 +1064,78 @@ export class SelfHealingManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Clear `blockedBy` on todo tasks whose blocker has reached a terminal or
|
||||||
|
* stuck state.
|
||||||
|
*
|
||||||
|
* Stale-blocker conditions (clear if ANY apply):
|
||||||
|
* 1. Blocker task does not exist (id missing entirely)
|
||||||
|
* 2. Blocker `column === "done"` or `column === "archived"`
|
||||||
|
* 3. Blocker `column === "in-review"` and `paused === true`
|
||||||
|
* 4. Blocker `column === "in-review"` and `status === "failed"`
|
||||||
|
* and `(mergeRetries ?? 0) >= MAX_AUTO_MERGE_RETRIES`
|
||||||
|
*
|
||||||
|
* @returns Number of tasks unblocked
|
||||||
|
*/
|
||||||
|
async clearStaleBlockedBy(): Promise<number> {
|
||||||
|
try {
|
||||||
|
const settings = await this.store.getSettings();
|
||||||
|
if (settings.globalPause || settings.enginePaused) return 0;
|
||||||
|
|
||||||
|
const todoTasks = await this.store.listTasks({ column: "todo", slim: true });
|
||||||
|
const blockedTasks = todoTasks.filter(
|
||||||
|
(task) => typeof task.blockedBy === "string" && task.blockedBy.trim().length > 0,
|
||||||
|
);
|
||||||
|
|
||||||
|
if (blockedTasks.length === 0) return 0;
|
||||||
|
|
||||||
|
const allTasks = await this.store.listTasks({ slim: true, includeArchived: true });
|
||||||
|
const taskById = new Map(allTasks.map((task) => [task.id, task]));
|
||||||
|
|
||||||
|
let recovered = 0;
|
||||||
|
for (const task of blockedTasks) {
|
||||||
|
const blockerId = task.blockedBy;
|
||||||
|
if (!blockerId) continue;
|
||||||
|
|
||||||
|
const blocker = taskById.get(blockerId);
|
||||||
|
let reason: string | null = null;
|
||||||
|
|
||||||
|
if (!blocker) {
|
||||||
|
reason = `blocker ${blockerId} missing`;
|
||||||
|
} else if (blocker.column === "done") {
|
||||||
|
reason = `blocker ${blockerId} is done`;
|
||||||
|
} else if (blocker.column === "archived") {
|
||||||
|
reason = `blocker ${blockerId} is archived`;
|
||||||
|
} else if (blocker.column === "in-review" && blocker.paused) {
|
||||||
|
reason = `blocker ${blockerId} in-review + paused`;
|
||||||
|
} else if (
|
||||||
|
blocker.column === "in-review" &&
|
||||||
|
blocker.status === "failed" &&
|
||||||
|
(blocker.mergeRetries ?? 0) >= MAX_AUTO_MERGE_RETRIES
|
||||||
|
) {
|
||||||
|
reason = `blocker ${blockerId} in-review + failed (mergeRetries ${blocker.mergeRetries ?? 0}/${MAX_AUTO_MERGE_RETRIES})`;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!reason) continue;
|
||||||
|
|
||||||
|
try {
|
||||||
|
await this.store.updateTask(task.id, { blockedBy: null, status: null });
|
||||||
|
await this.store.logEntry(task.id, `Auto-recovered: cleared stale blockedBy — ${reason}`);
|
||||||
|
recovered++;
|
||||||
|
} catch (err: unknown) {
|
||||||
|
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||||
|
log.error(`Failed to clear stale blockedBy for ${task.id}: ${errorMessage}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return recovered;
|
||||||
|
} catch (err: unknown) {
|
||||||
|
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||||
|
log.error(`Stale blockedBy sweep failed: ${errorMessage}`);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Recover `in-review` tasks that are fully mergeable but never had
|
* Recover `in-review` tasks that are fully mergeable but never had
|
||||||
* `mergeTask()` invoked.
|
* `mergeTask()` invoked.
|
||||||
|
|||||||
Reference in New Issue
Block a user