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

feat(FN-4053): unify task ID allocation with store-owned distributed alloca

Browse Source 
Unifies task ID allocation under a single store-owned distributed allocator, removing redundant ID-generation logic from dashboard route handlers and simplifying the overall flow; Step 1 merges the allocation authority into the store, Step 3 removes the now-unnecessary mixed-path routing, and tests

Fusion-Task-Id: FN-4053
This commit is contained in:
Fusion
2026-05-11 21:38:25 -07:00
committed by gsxdsm
parent ce0e5f4ce6
commit 4404c6103e
12 changed files with 244 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
packages/dashboard/src/__tests__/routes-tasks.test.ts
View File

@@ -717,8 +717,8 @@ describe("POST /tasks", () => {
createIssueSpy.mockRestore();
});
it("uses distributed allocator flow when reserved-id create is available", async () => {
const createTaskWithReservedId = vi.fn().mockResolvedValue({
it("uses store.createTask for local task creation", async () => {
const createTask = vi.fn().mockResolvedValue({
...FAKE_TASK_DETAIL,
id: "FN-7001",
column: "triage",
@@ -726,14 +726,11 @@ describe("POST /tasks", () => {
updatedAt: "2026-05-05T00:00:00.000Z",
nodeId: "node-target",
});
const storeWithReservedCreate = createMockStore({
createTaskWithReservedId,
getTask: vi.fn().mockResolvedValue({ ...FAKE_TASK_DETAIL, prompt: "# FN-7001\n\nBig initiative\n" }),
});
const storeWithCreate = createMockStore({ createTask });
const app = express();
app.use(express.json());
app.use("/api", createApiRoutes(storeWithReservedCreate));
app.use("/api", createApiRoutes(storeWithCreate));
const res = await REQUEST(
app,
@@ -744,11 +741,11 @@ describe("POST /tasks", () => {
);
expect(res.status).toBe(201);
expect(createTaskWithReservedId).toHaveBeenCalledWith(
expect(createTask).toHaveBeenCalledWith(
expect.objectContaining({ description: "Big initiative", nodeId: "node-target" }),
expect.objectContaining({ taskId: "FN-7001" }),
expect.objectContaining({ settings: { autoSummarizeTitles: undefined } }),
);
expect((storeWithReservedCreate.getDistributedTaskIdAllocator as ReturnType<typeof vi.fn>).mock.results[0]?.value.commitDistributedTaskIdReservation).toHaveBeenCalled();
expect((storeWithCreate.getDistributedTaskIdAllocator as ReturnType<typeof vi.fn>)).not.toHaveBeenCalled();
});
it("returns 400 when nodeId is not a string", async () => {
@@ -764,39 +761,13 @@ describe("POST /tasks", () => {
expect(res.body.error).toContain("nodeId must be a string");
});
it("retries reserved-id create when first reservation overlaps an existing task id", async () => {
const reserveDistributedTaskId = vi
.fn()
.mockResolvedValueOnce({ reservationId: "res-1", taskId: "FN-7001" })
.mockResolvedValueOnce({ reservationId: "res-2", taskId: "FN-7002" });
const commitDistributedTaskIdReservation = vi.fn().mockResolvedValue({});
const abortDistributedTaskIdReservation = vi.fn().mockResolvedValue({});
const createTaskWithReservedId = vi
.fn()
.mockRejectedValueOnce(new Error("Task ID already exists: FN-7001"))
.mockResolvedValueOnce({
...FAKE_TASK_DETAIL,
id: "FN-7002",
column: "triage",
createdAt: "2026-05-05T00:00:00.000Z",
updatedAt: "2026-05-05T00:00:00.000Z",
});
const deleteTask = vi.fn().mockResolvedValue(undefined);
const getTask = vi.fn().mockResolvedValue({ ...FAKE_TASK_DETAIL, prompt: "# FN-7002\n\nBig initiative\n" });
const storeWithReservedCreate = createMockStore({
createTaskWithReservedId,
deleteTask,
getTask,
getDistributedTaskIdAllocator: vi.fn().mockReturnValue({
reserveDistributedTaskId,
commitDistributedTaskIdReservation,
abortDistributedTaskIdReservation,
}),
});
it("returns 500 when store.createTask throws", async () => {
const createTask = vi.fn().mockRejectedValue(new Error("Task ID already exists: FN-7001"));
const storeWithCreate = createMockStore({ createTask });
const app = express();
app.use(express.json());
app.use("/api", createApiRoutes(storeWithReservedCreate));
app.use("/api", createApiRoutes(storeWithCreate));
const res = await REQUEST(
app,
@@ -806,68 +777,8 @@ describe("POST /tasks", () => {
{ "Content-Type": "application/json" },
);
expect(res.status).toBe(201);
expect(createTaskWithReservedId).toHaveBeenCalledTimes(2);
expect(createTaskWithReservedId).toHaveBeenNthCalledWith(
1,
expect.objectContaining({ description: "Big initiative" }),
expect.objectContaining({ taskId: "FN-7001" }),
);
expect(createTaskWithReservedId).toHaveBeenNthCalledWith(
2,
expect.objectContaining({ description: "Big initiative" }),
expect.objectContaining({ taskId: "FN-7002" }),
);
expect(abortDistributedTaskIdReservation).toHaveBeenCalledTimes(1);
expect(abortDistributedTaskIdReservation).toHaveBeenCalledWith(expect.objectContaining({ reservationId: "res-1", reason: "failed-create" }));
expect(commitDistributedTaskIdReservation).toHaveBeenCalledWith(expect.objectContaining({ reservationId: "res-2" }));
expect(deleteTask).not.toHaveBeenCalled();
});
it("aborts reservation and deletes local task on replication failure", async () => {
const reserveDistributedTaskId = vi.fn().mockResolvedValue({ reservationId: "res-1", taskId: "FN-7002" });
const commitDistributedTaskIdReservation = vi.fn().mockResolvedValue({});
const abortDistributedTaskIdReservation = vi.fn().mockResolvedValue({});
const createTaskWithReservedId = vi.fn().mockResolvedValue({
...FAKE_TASK_DETAIL,
id: "FN-7002",
column: "triage",
createdAt: "2026-05-05T00:00:00.000Z",
updatedAt: "2026-05-05T00:00:00.000Z",
});
const deleteTask = vi.fn().mockResolvedValue(undefined);
const storeWithReservedCreate = createMockStore({
createTaskWithReservedId,
deleteTask,
getTask: vi.fn().mockResolvedValue({ ...FAKE_TASK_DETAIL, prompt: "# FN-7002\n\nBig initiative\n" }),
getDistributedTaskIdAllocator: vi.fn().mockReturnValue({
reserveDistributedTaskId,
commitDistributedTaskIdReservation,
abortDistributedTaskIdReservation,
}),
});
mockCentralListNodes.mockResolvedValue([{ id: "node-remote", type: "remote", url: "https://remote.example.com", apiKey: "secret" }]);
vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("network down")));
const app = express();
app.use(express.json());
app.use("/api", createApiRoutes(storeWithReservedCreate));
const res = await REQUEST(
app,
"POST",
"/api/tasks",
JSON.stringify({ description: "Big initiative" }),
{ "Content-Type": "application/json" },
);
expect(res.status).toBe(503);
expect(abortDistributedTaskIdReservation).toHaveBeenCalledWith(expect.objectContaining({ reservationId: "res-1", reason: "failed-create" }));
expect(deleteTask).toHaveBeenCalledWith("FN-7002");
expect(commitDistributedTaskIdReservation).not.toHaveBeenCalled();
expect(reserveDistributedTaskId).toHaveBeenCalledTimes(1);
vi.unstubAllGlobals();
mockCentralListNodes.mockResolvedValue([]);
expect(res.status).toBe(500);
expect(res.body.error).toContain("Task ID already exists: FN-7001");
});
it("forwards branch and baseBranch on create", async () => {

95
packages/dashboard/src/routes/register-task-workflow-routes.ts
View File

@@ -4,11 +4,9 @@ import {
COLUMNS,
TASK_PRIORITIES,
VALID_TRANSITIONS,
buildMeshReplicatedTaskCreatePayload,
isTaskPriority,
REPO_OVERRIDE_RE,
resolveTitleSummarizerSettingsModel,
toReplicatedCreateInput,
validateNodeOverrideChange,
canAgentTakeImplementationTaskForExplicitRouting,
formatRoleMismatchReason,
@@ -19,7 +17,6 @@ import { maybeCreateTrackingIssue } from "../github-tracking.js";
import { parseGitHubBadgeUrl } from "./register-git-github.js";
import { planTaskWorktreePath } from "@fusion/engine";
import { ApiError, badRequest, conflict, notFound } from "../api-error.js";
import { fetchFromRemoteNode } from "./register-settings-sync-helpers.js";
import type { ApiRoutesContext } from "./types.js";
import { resolveBranchSelection } from "./branch-selection.js";
@@ -350,91 +347,13 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
...(validatedGithubTracking ? { githubTracking: validatedGithubTracking } : {}),
};
if (typeof scopedStore.createTaskWithReservedId !== "function") {
const task = await scopedStore.createTask(
createInput,
{ onSummarize, settings: { autoSummarizeTitles: settings.autoSummarizeTitles } },
);
await maybeCreateTaskTrackingIssue(scopedStore, task, options?.githubToken);
res.status(201).json(task);
return;
}
const allocator = scopedStore.getDistributedTaskIdAllocator();
const { CentralCore } = await import("@fusion/core");
const central = new CentralCore();
await central.init();
const nodes = await central.listNodes();
const localNode = nodes.find((node) => node.type === "local");
const remoteNodes = nodes.filter((node) => node.type === "remote" && node.url && node.apiKey);
await central.close();
const nodeIdForReservation = localNode?.id ?? "local";
const isOverlapClassFailure = (err: unknown): boolean => {
const message = err instanceof Error ? err.message : String(err);
return (
message.includes("Task ID already exists:") ||
message.includes("Replicated task payload collision for existing task")
);
};
const maxCreateAttempts = 3;
for (let attempt = 1; attempt <= maxCreateAttempts; attempt += 1) {
const reservation = await allocator.reserveDistributedTaskId({
prefix: "FN",
nodeId: nodeIdForReservation,
});
let createdTask: Task | null = null;
try {
createdTask = await scopedStore.createTaskWithReservedId(createInput, {
taskId: reservation.taskId,
});
await maybeCreateTaskTrackingIssue(scopedStore, createdTask, options?.githubToken);
const replicatedPayload = buildMeshReplicatedTaskCreatePayload({
taskId: createdTask.id,
reservationId: reservation.reservationId,
sourceNodeId: nodeIdForReservation,
createdAt: createdTask.createdAt,
updatedAt: createdTask.updatedAt,
prompt: (await scopedStore.getTask(createdTask.id)).prompt,
createInput: toReplicatedCreateInput(createdTask),
});
for (const peer of remoteNodes) {
await fetchFromRemoteNode(peer, "/api/mesh/tasks/create", {
method: "POST",
body: replicatedPayload,
});
}
await allocator.commitDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId: nodeIdForReservation,
});
res.status(201).json(createdTask);
return;
} catch (err: unknown) {
await allocator.abortDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId: nodeIdForReservation,
reason: "failed-create",
}).catch(() => undefined);
if (createdTask) {
await scopedStore.deleteTask(createdTask.id).catch(() => undefined);
}
if (attempt < maxCreateAttempts && isOverlapClassFailure(err)) {
continue;
}
const message = err instanceof Error ? err.message : String(err);
throw new ApiError(503, `Cluster task create failed: ${message}`);
}
}
const task = await scopedStore.createTask(
createInput,
{ onSummarize, settings: { autoSummarizeTitles: settings.autoSummarizeTitles } },
);
await maybeCreateTaskTrackingIssue(scopedStore, task, options?.githubToken);
res.status(201).json(task);
return;
} catch (err: unknown) {
if (err instanceof ApiError) {
throw err;