FN-6063: enforce workflow scheduling capacity after hold release
Keep scheduler capacity checks aligned with fresh task state during workflow dispatch. - bypass the startup slim-list memo when scheduling and after workflow hold release sweeps - clear the startup slim-list memo when task.json writes move tasks between columns - treat capacity-exhausted moveTask rejections as queued dispatches, log the reason, and cover the behavior with scheduler/store tests - add a patch changeset for the published CLI package Files changed: .changeset/workflow-capacity-scheduler.md | 5 +++ packages/core/src/__tests__/store-movement.test.ts | 15 ++++++++ packages/core/src/store.ts | 1 + packages/engine/src/__tests__/scheduler.test.ts | 40 ++++++++++++++++++++-- packages/engine/src/scheduler.ts | 25 ++++++++++---- 5 files changed, 77 insertions(+), 9 deletions(-) Fusion-Task-Id: FN-6063 Fusion-Task-Lineage: 80ddc0e7-f78d-483c-886a-27c4c07bde7a
This commit is contained in:
5
.changeset/workflow-capacity-scheduler.md
Normal file
5
.changeset/workflow-capacity-scheduler.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@runfusion/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Fix workflow scheduling so in-progress column limits are enforced from fresh task state after hold-advancing sweep dispatches.
|
||||||
@@ -33,6 +33,21 @@ describe("TaskStore", () => {
|
|||||||
const createSourceIssueFixture = () => harness.createSourceIssueFixture();
|
const createSourceIssueFixture = () => harness.createSourceIssueFixture();
|
||||||
const insertLogEntryWithTimestamp = (...args: any[]) => (harness as any).insertLogEntryWithTimestamp(...args);
|
const insertLogEntryWithTimestamp = (...args: any[]) => (harness as any).insertLogEntryWithTimestamp(...args);
|
||||||
|
|
||||||
|
describe("listTasks startup memo invalidation", () => {
|
||||||
|
it("returns fresh slim task state after moveTask writes task json", async () => {
|
||||||
|
const task = await store.createTask({ description: "memo invalidation after move" });
|
||||||
|
await store.moveTask(task.id, "todo");
|
||||||
|
|
||||||
|
const beforeMove = await store.listTasks({ slim: true, includeArchived: false, startupMemo: true });
|
||||||
|
expect(beforeMove.find((listed) => listed.id === task.id)?.column).toBe("todo");
|
||||||
|
|
||||||
|
await store.moveTask(task.id, "in-progress");
|
||||||
|
|
||||||
|
const afterMove = await store.listTasks({ slim: true, includeArchived: false, startupMemo: true });
|
||||||
|
expect(afterMove.find((listed) => listed.id === task.id)?.column).toBe("in-progress");
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("moveTask — in-progress to triage", () => {
|
describe("moveTask — in-progress to triage", () => {
|
||||||
it("allows moving an in-progress task to triage", async () => {
|
it("allows moving an in-progress task to triage", async () => {
|
||||||
const task = await store.createTask({ description: "test in-progress to triage" });
|
const task = await store.createTask({ description: "test in-progress to triage" });
|
||||||
|
|||||||
@@ -3131,6 +3131,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
|
|||||||
}
|
}
|
||||||
|
|
||||||
private async writeTaskJsonFile(dir: string, task: Task): Promise<void> {
|
private async writeTaskJsonFile(dir: string, task: Task): Promise<void> {
|
||||||
|
this.clearStartupSlimListMemo();
|
||||||
const taskJsonPath = join(dir, "task.json");
|
const taskJsonPath = join(dir, "task.json");
|
||||||
// Use a unique tmp filename per write so concurrent writers to the same task
|
// Use a unique tmp filename per write so concurrent writers to the same task
|
||||||
// don't race on a shared `task.json.tmp` (one rename consumes it, the other
|
// don't race on a shared `task.json.tmp` (one rename consumes it, the other
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ import {
|
|||||||
getUnmetSchedulingDependencies,
|
getUnmetSchedulingDependencies,
|
||||||
} from "../scheduler.js";
|
} from "../scheduler.js";
|
||||||
import { AgentSemaphore } from "../concurrency.js";
|
import { AgentSemaphore } from "../concurrency.js";
|
||||||
import type { TaskStore, Task, TaskDetail } from "@fusion/core";
|
import { makeTransitionRejection, TransitionRejectionError, type TaskStore, type Task, type TaskDetail } from "@fusion/core";
|
||||||
import { existsSync } from "node:fs";
|
import { existsSync } from "node:fs";
|
||||||
import { readFile } from "node:fs/promises";
|
import { readFile } from "node:fs/promises";
|
||||||
import { schedulerLog } from "../logger.js";
|
import { schedulerLog } from "../logger.js";
|
||||||
@@ -599,7 +599,10 @@ describe("Scheduler", () => {
|
|||||||
if (column === "in-progress") {
|
if (column === "in-progress") {
|
||||||
const inProgressCount = [...tasks.values()].filter((task) => task.column === "in-progress").length;
|
const inProgressCount = [...tasks.values()].filter((task) => task.column === "in-progress").length;
|
||||||
if (inProgressCount >= 3) {
|
if (inProgressCount >= 3) {
|
||||||
throw new Error("capacity-exhausted");
|
throw new TransitionRejectionError(
|
||||||
|
makeTransitionRejection("capacity-exhausted", "transition.rejected.capacityExhausted", true),
|
||||||
|
"queued — in-progress column at capacity",
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
const updated = { ...current, column } as Task;
|
const updated = { ...current, column } as Task;
|
||||||
@@ -624,6 +627,7 @@ describe("Scheduler", () => {
|
|||||||
expect([...tasks.values()].filter((task) => task.column === "in-progress")).toHaveLength(3);
|
expect([...tasks.values()].filter((task) => task.column === "in-progress")).toHaveLength(3);
|
||||||
expect(moveTask.mock.calls.filter((call) => call[1] === "in-progress")).toHaveLength(6);
|
expect(moveTask.mock.calls.filter((call) => call[1] === "in-progress")).toHaveLength(6);
|
||||||
expect(vi.mocked(store.listTasks).mock.calls.length).toBeGreaterThanOrEqual(2);
|
expect(vi.mocked(store.listTasks).mock.calls.length).toBeGreaterThanOrEqual(2);
|
||||||
|
expect(vi.mocked(store.listTasks).mock.calls.some(([options]) => options?.startupMemo === false)).toBe(true);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("flag-OFF: todo dispatch is tagged as scheduler-sourced for redispatch guards", async () => {
|
it("flag-OFF: todo dispatch is tagged as scheduler-sourced for redispatch guards", async () => {
|
||||||
@@ -1481,6 +1485,38 @@ describe("Scheduler", () => {
|
|||||||
expect(store.moveTask).not.toHaveBeenCalled();
|
expect(store.moveTask).not.toHaveBeenCalled();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("queues capacity-exhausted dispatches and continues to later candidates", async () => {
|
||||||
|
vi.mocked(existsSync).mockReturnValue(true);
|
||||||
|
vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
|
||||||
|
|
||||||
|
const tasks = [
|
||||||
|
createMockTask({ id: "FN-001", column: "todo", dependencies: [] }),
|
||||||
|
createMockTask({ id: "FN-002", column: "todo", dependencies: [] }),
|
||||||
|
];
|
||||||
|
const store = createMockStore({
|
||||||
|
listTasks: vi.fn().mockResolvedValue(tasks),
|
||||||
|
getTask: vi.fn(async (taskId: string) => tasks.find((task) => task.id === taskId) ?? null),
|
||||||
|
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 5, maxWorktrees: 10 }),
|
||||||
|
moveTask: vi.fn().mockRejectedValue(
|
||||||
|
new TransitionRejectionError(
|
||||||
|
makeTransitionRejection("capacity-exhausted", "transition.rejected.capacityExhausted", true),
|
||||||
|
"queued — in-progress column at capacity",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
});
|
||||||
|
|
||||||
|
const scheduler = new Scheduler(store);
|
||||||
|
(scheduler as unknown as { running: boolean }).running = true;
|
||||||
|
await scheduler.schedule();
|
||||||
|
|
||||||
|
expect(store.moveTask).toHaveBeenCalledTimes(2);
|
||||||
|
expect(vi.mocked(store.updateTask).mock.calls.filter((call) => call[1]?.status === "queued").map((call) => call[0])).toEqual([
|
||||||
|
"FN-001",
|
||||||
|
"FN-002",
|
||||||
|
]);
|
||||||
|
expect(store.logEntry).toHaveBeenCalledWith("FN-001", expect.stringContaining("queued — in-progress column at capacity"));
|
||||||
|
});
|
||||||
|
|
||||||
it("respects maxWorktrees limit", async () => {
|
it("respects maxWorktrees limit", async () => {
|
||||||
const tasks = [
|
const tasks = [
|
||||||
createMockTask({ id: "FN-001", column: "in-progress" }),
|
createMockTask({ id: "FN-001", column: "in-progress" }),
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import {
|
|||||||
type PrInfo,
|
type PrInfo,
|
||||||
type AgentStore,
|
type AgentStore,
|
||||||
type Settings,
|
type Settings,
|
||||||
|
TransitionRejectionError,
|
||||||
} from "@fusion/core";
|
} from "@fusion/core";
|
||||||
import { existsSync } from "node:fs";
|
import { existsSync } from "node:fs";
|
||||||
import { readFile } from "node:fs/promises";
|
import { readFile } from "node:fs/promises";
|
||||||
@@ -1225,7 +1226,7 @@ export class Scheduler {
|
|||||||
this.scheduling = true;
|
this.scheduling = true;
|
||||||
|
|
||||||
try {
|
try {
|
||||||
let tasks = await this.store.listTasks({ slim: true, includeArchived: false });
|
let tasks = await this.store.listTasks({ slim: true, includeArchived: false, startupMemo: false });
|
||||||
let settings = await this.store.getSettings();
|
let settings = await this.store.getSettings();
|
||||||
this.idleSemaphoreLeakCandidateSince = recoverIdleSemaphoreLeak(
|
this.idleSemaphoreLeakCandidateSince = recoverIdleSemaphoreLeak(
|
||||||
this.options.semaphore,
|
this.options.semaphore,
|
||||||
@@ -1273,7 +1274,7 @@ export class Scheduler {
|
|||||||
// workflow hold handling and the generalized capacity-release path.
|
// workflow hold handling and the generalized capacity-release path.
|
||||||
if (isWorkflowColumnsEnabled(settings)) {
|
if (isWorkflowColumnsEnabled(settings)) {
|
||||||
await this.runHoldReleaseSweepPass();
|
await this.runHoldReleaseSweepPass();
|
||||||
tasks = await this.store.listTasks({ slim: true, includeArchived: false });
|
tasks = await this.store.listTasks({ slim: true, includeArchived: false, startupMemo: false });
|
||||||
settings = await this.store.getSettings();
|
settings = await this.store.getSettings();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2039,11 +2040,21 @@ export class Scheduler {
|
|||||||
effectiveNodeSource: effectiveNode.source,
|
effectiveNodeSource: effectiveNode.source,
|
||||||
mergeRetries: 0,
|
mergeRetries: 0,
|
||||||
});
|
});
|
||||||
await this.store.moveTask(task.id, "in-progress", {
|
try {
|
||||||
moveSource: "scheduler",
|
await this.store.moveTask(task.id, "in-progress", {
|
||||||
allocateWorktree: (reservedNames) =>
|
moveSource: "scheduler",
|
||||||
this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings),
|
allocateWorktree: (reservedNames) =>
|
||||||
});
|
this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings),
|
||||||
|
});
|
||||||
|
} catch (error) {
|
||||||
|
if (error instanceof TransitionRejectionError && error.rejection.code === "capacity-exhausted") {
|
||||||
|
await this.store.updateTask(task.id, { status: "queued" });
|
||||||
|
const reason = error.message || "queued — in-progress column at capacity";
|
||||||
|
await this.logDispatchQueuedReason(task.id, reason, `capacity-exhausted:${reason}`);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
await this.store.updateTask(task.id, {
|
await this.store.updateTask(task.id, {
|
||||||
dispatchStormCount: nextDispatchStormCount,
|
dispatchStormCount: nextDispatchStormCount,
|
||||||
lastDispatchAt: dispatchTimestamp,
|
lastDispatchAt: dispatchTimestamp,
|
||||||
|
|||||||
Reference in New Issue
Block a user