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

feat(FN-3710): retry overlap-class cluster create failures in distributed t

Browse Source 
Adds retry logic for distributed task ID overlap conflicts across the cluster, hardening the ID allocator with exponential-backoff recovery, wiring the retry into task workflow registration, and documenting the behavior in architecture docs. Minor CSS tokenization on the Mailbox and Plugin Manager m

Fusion-Task-Id: FN-3710
This commit is contained in:
Fusion
2026-05-07 11:44:42 -07:00
committed by gsxdsm
parent a98e5aac1a
commit 24325cd3c1
8 changed files with 264 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -249,53 +249,70 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
const remoteNodes = nodes.filter((node) => node.type === "remote" && node.url && node.apiKey);
await central.close();
const reservation = await allocator.reserveDistributedTaskId({
prefix: "FN",
nodeId: localNode?.id ?? "local",
});
let createdTask: Task | null = null;
try {
createdTask = await scopedStore.createTaskWithReservedId(createInput, {
taskId: reservation.taskId,
});
const replicatedPayload = buildMeshReplicatedTaskCreatePayload({
taskId: createdTask.id,
reservationId: reservation.reservationId,
sourceNodeId: localNode?.id ?? "local",
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: localNode?.id ?? "local",
});
res.status(201).json(createdTask);
} catch (err: unknown) {
await allocator.abortDistributedTaskIdReservation({
reservationId: reservation.reservationId,
nodeId: localNode?.id ?? "local",
reason: "failed-create",
}).catch(() => undefined);
if (createdTask) {
await scopedStore.deleteTask(createdTask.id).catch(() => undefined);
}
const nodeIdForReservation = localNode?.id ?? "local";
const isOverlapClassFailure = (err: unknown): boolean => {
const message = err instanceof Error ? err.message : String(err);
throw new ApiError(503, `Cluster task create failed: ${message}`);
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,
});
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}`);
}
}
} catch (err: unknown) {
if (err instanceof ApiError) {