fix(FN-6084): honor worktree limit for hold releases
Workflow-column hold releases now reserve against the active worktree count before moving tasks into in-progress, preventing restart bursts from exceeding maxWorktrees. Fusion-Task-Id: FN-6084
This commit is contained in:
5
.changeset/worktree-capacity-hold-release.md
Normal file
5
.changeset/worktree-capacity-hold-release.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@runfusion/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Honor the worktree execution limit when workflow-column hold releases dispatch tasks.
|
||||||
@@ -630,6 +630,55 @@ describe("Scheduler", () => {
|
|||||||
expect(vi.mocked(store.listTasks).mock.calls.some(([options]) => options?.startupMemo === false)).toBe(true);
|
expect(vi.mocked(store.listTasks).mock.calls.some(([options]) => options?.startupMemo === false)).toBe(true);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("holds workflow-column releases when maxWorktrees is exhausted", async () => {
|
||||||
|
vi.mocked(existsSync).mockReturnValue(true);
|
||||||
|
vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
|
||||||
|
|
||||||
|
const tasks = new Map<string, Task>([
|
||||||
|
["FN-001", createMockTask({ id: "FN-001", column: "in-progress", dependencies: [] })],
|
||||||
|
["FN-002", createMockTask({ id: "FN-002", column: "in-progress", dependencies: [] })],
|
||||||
|
["FN-003", createMockTask({ id: "FN-003", column: "todo", dependencies: [] })],
|
||||||
|
["FN-004", createMockTask({ id: "FN-004", column: "todo", dependencies: [] })],
|
||||||
|
["FN-005", createMockTask({ id: "FN-005", column: "todo", dependencies: [] })],
|
||||||
|
]);
|
||||||
|
const movedListeners = new Set<(data: { task: object; to: string }) => void>();
|
||||||
|
const moveTask = vi.fn(async (taskId: string, column: Task["column"]) => {
|
||||||
|
const current = tasks.get(taskId);
|
||||||
|
if (!current) throw new Error(`missing task ${taskId}`);
|
||||||
|
const updated = { ...current, column } as Task;
|
||||||
|
tasks.set(taskId, updated);
|
||||||
|
for (const listener of movedListeners) {
|
||||||
|
listener({ task: updated, to: column });
|
||||||
|
}
|
||||||
|
return updated;
|
||||||
|
});
|
||||||
|
const store = createMockStore({
|
||||||
|
listTasks: vi.fn(async () => [...tasks.values()]),
|
||||||
|
getTask: vi.fn(async (taskId: string) => tasks.get(taskId) ?? null),
|
||||||
|
getSettings: vi.fn().mockResolvedValue({
|
||||||
|
maxConcurrent: 15,
|
||||||
|
maxWorktrees: 3,
|
||||||
|
experimentalFeatures: { workflowColumns: true },
|
||||||
|
}),
|
||||||
|
moveTask,
|
||||||
|
on: vi.fn((event: string, listener: (data: { task: object; to: string }) => void) => {
|
||||||
|
if (event === "task:moved") movedListeners.add(listener);
|
||||||
|
}),
|
||||||
|
off: vi.fn((event: string, listener: (data: { task: object; to: string }) => void) => {
|
||||||
|
if (event === "task:moved") movedListeners.delete(listener);
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
|
||||||
|
const scheduler = new Scheduler(store);
|
||||||
|
(scheduler as unknown as { running: boolean }).running = true;
|
||||||
|
await scheduler.schedule();
|
||||||
|
|
||||||
|
const inProgress = [...tasks.values()].filter((task) => task.column === "in-progress");
|
||||||
|
expect(inProgress.map((task) => task.id)).toEqual(["FN-001", "FN-002", "FN-003"]);
|
||||||
|
expect(moveTask.mock.calls.filter((call) => call[1] === "in-progress").map((call) => call[0])).toEqual(["FN-003"]);
|
||||||
|
expect(schedulerLog.log).toHaveBeenCalledWith(expect.stringContaining("no reservable slot"));
|
||||||
|
});
|
||||||
|
|
||||||
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 () => {
|
||||||
const off = setupTodoStore(false);
|
const off = setupTodoStore(false);
|
||||||
await off.scheduler.schedule();
|
await off.scheduler.schedule();
|
||||||
|
|||||||
@@ -1273,7 +1273,7 @@ export class Scheduler {
|
|||||||
// still drives todo→in-progress pickup (parity); the sweep adds custom-
|
// still drives todo→in-progress pickup (parity); the sweep adds custom-
|
||||||
// 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, settings);
|
||||||
tasks = await this.store.listTasks({ slim: true, includeArchived: false, startupMemo: false });
|
tasks = await this.store.listTasks({ slim: true, includeArchived: false, startupMemo: false });
|
||||||
settings = await this.store.getSettings();
|
settings = await this.store.getSettings();
|
||||||
}
|
}
|
||||||
@@ -2122,24 +2122,31 @@ export class Scheduler {
|
|||||||
* worktree allocation into the reservation-first ordering (KTD-10). Failures
|
* worktree allocation into the reservation-first ordering (KTD-10). Failures
|
||||||
* are isolated so a sweep error never breaks the scheduling pass.
|
* are isolated so a sweep error never breaks the scheduling pass.
|
||||||
*/
|
*/
|
||||||
private async runHoldReleaseSweepPass(): Promise<void> {
|
private async runHoldReleaseSweepPass(tasks: Task[], settings: Settings): Promise<void> {
|
||||||
try {
|
try {
|
||||||
|
const maxWorktrees = settings.maxWorktrees ?? this.options.maxWorktrees ?? 4;
|
||||||
|
let reservedWorktreeSlots = tasks.filter((task) => task.column === "in-progress").length;
|
||||||
await runHoldReleaseSweep(this.store, {
|
await runHoldReleaseSweep(this.store, {
|
||||||
now: () => Date.now(),
|
now: () => Date.now(),
|
||||||
reserveSlot: this.options.semaphore
|
reserveSlot: (): SlotReservation | null => {
|
||||||
? (): SlotReservation | null => {
|
if (Number.isFinite(maxWorktrees) && reservedWorktreeSlots >= maxWorktrees) {
|
||||||
const sem = this.options.semaphore!;
|
return null;
|
||||||
if (!sem.tryAcquire()) return null;
|
|
||||||
let released = false;
|
|
||||||
return {
|
|
||||||
release: () => {
|
|
||||||
if (released) return;
|
|
||||||
released = true;
|
|
||||||
sem.release();
|
|
||||||
},
|
|
||||||
};
|
|
||||||
}
|
}
|
||||||
: undefined,
|
|
||||||
|
const sem = this.options.semaphore;
|
||||||
|
if (sem && !sem.tryAcquire()) return null;
|
||||||
|
|
||||||
|
reservedWorktreeSlots += 1;
|
||||||
|
let released = false;
|
||||||
|
return {
|
||||||
|
release: () => {
|
||||||
|
if (released) return;
|
||||||
|
released = true;
|
||||||
|
reservedWorktreeSlots = Math.max(0, reservedWorktreeSlots - 1);
|
||||||
|
sem?.release();
|
||||||
|
},
|
||||||
|
};
|
||||||
|
},
|
||||||
allocateWorktree: (task, reservedNames) =>
|
allocateWorktree: (task, reservedNames) =>
|
||||||
planTaskWorktreePath(task, this.store.getRootDir(), undefined, reservedNames, {}),
|
planTaskWorktreePath(task, this.store.getRootDir(), undefined, reservedNames, {}),
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user