From edd9d1f5264d7f8b5d3ced1aa4c013ce2d1e3fec Mon Sep 17 00:00:00 2001 From: Yash Kumar Date: Sat, 12 Sep 2026 00:08:46 +0530 Subject: [PATCH] fix: close 4 race conditions in queued deployment system - Atomic WHERE status = 'queued' claim prevents cancelled jobs from running - initCancelDeployments before server.listen prevents startup cancelling live requests - getActiveDeploymentStatus helper preserves queued/running status when deployments remain - Per-service mutex prevents enqueue/cancel interleaving Extracted shared helpers to eliminate 40+ lines of duplication across services. --- .../deploy/application.command.test.ts | 4 + .../__test__/deploy/application.real.test.ts | 4 + .../dokploy/server/api/routers/application.ts | 13 +- apps/dokploy/server/api/routers/compose.ts | 13 +- apps/dokploy/server/queues/queueSetup.ts | 126 ++++++++++-------- apps/dokploy/server/queues/service-lock.ts | 21 +++ packages/server/src/services/application.ts | 15 ++- packages/server/src/services/compose.ts | 12 ++ packages/server/src/services/deployment.ts | 76 ++++++----- .../server/src/services/preview-deployment.ts | 17 ++- packages/server/src/services/transfer.ts | 3 +- 11 files changed, 188 insertions(+), 116 deletions(-) create mode 100644 apps/dokploy/server/queues/service-lock.ts diff --git a/apps/dokploy/__test__/deploy/application.command.test.ts b/apps/dokploy/__test__/deploy/application.command.test.ts index 6223c0ee1..ecb600f65 100644 --- a/apps/dokploy/__test__/deploy/application.command.test.ts +++ b/apps/dokploy/__test__/deploy/application.command.test.ts @@ -33,6 +33,9 @@ vi.mock("@dokploy/server/db", () => { applications: { findFirst: vi.fn(), }, + deployments: { + findFirst: vi.fn().mockResolvedValue(null), + }, patch: { findMany: vi.fn().mockResolvedValue([]), }, @@ -62,6 +65,7 @@ vi.mock("@dokploy/server/services/admin", () => ({ vi.mock("@dokploy/server/services/deployment", () => ({ createDeployment: vi.fn(), resolveQueuedDeployment: vi.fn(), + getActiveDeploymentStatus: vi.fn().mockResolvedValue(null), updateDeploymentStatus: vi.fn(), updateDeployment: vi.fn(), })); diff --git a/apps/dokploy/__test__/deploy/application.real.test.ts b/apps/dokploy/__test__/deploy/application.real.test.ts index 258523404..9ea049253 100644 --- a/apps/dokploy/__test__/deploy/application.real.test.ts +++ b/apps/dokploy/__test__/deploy/application.real.test.ts @@ -34,6 +34,9 @@ vi.mock("@dokploy/server/db", () => { applications: { findFirst: vi.fn(), }, + deployments: { + findFirst: vi.fn().mockResolvedValue(null), + }, patch: { findMany: vi.fn().mockResolvedValue([]), }, @@ -63,6 +66,7 @@ vi.mock("@dokploy/server/services/admin", () => ({ vi.mock("@dokploy/server/services/deployment", () => ({ createDeployment: vi.fn(), resolveQueuedDeployment: vi.fn(), + getActiveDeploymentStatus: vi.fn().mockResolvedValue(null), updateDeploymentStatus: vi.fn(), updateDeployment: vi.fn(), })); diff --git a/apps/dokploy/server/api/routers/application.ts b/apps/dokploy/server/api/routers/application.ts index fc5b00290..18d5a7df9 100644 --- a/apps/dokploy/server/api/routers/application.ts +++ b/apps/dokploy/server/api/routers/application.ts @@ -1,5 +1,4 @@ import { - cancelAllQueuedDeploymentsByApplicationId, clearOldDeployments, createApplication, createDomain, @@ -70,7 +69,6 @@ import { apiSaveGitProvider, apiUpdateApplication, applications, - deployments, environments, projects, } from "@/server/db/schema"; @@ -845,17 +843,8 @@ export const applicationRouter = createTRPCRouter({ await checkServicePermissionAndAccess(ctx, input.applicationId, { deployment: ["cancel"], }); - await cancelAllQueuedDeploymentsByApplicationId(input.applicationId); await cleanQueuesByApplication(input.applicationId); - const hasRunning = await db.query.deployments.findFirst({ - where: and( - eq(deployments.applicationId, input.applicationId), - eq(deployments.status, "running"), - ), - }); - if (!hasRunning) { - await updateApplicationStatus(input.applicationId, "idle"); - } + await updateApplicationStatus(input.applicationId, "idle"); }), clearDeployments: protectedProcedure .input(apiFindOneApplication) diff --git a/apps/dokploy/server/api/routers/compose.ts b/apps/dokploy/server/api/routers/compose.ts index ebf85f190..484dbe7e5 100644 --- a/apps/dokploy/server/api/routers/compose.ts +++ b/apps/dokploy/server/api/routers/compose.ts @@ -1,7 +1,6 @@ import { join } from "node:path"; import { addDomainToCompose, - cancelAllQueuedDeploymentsByComposeId, clearOldDeployments, cloneCompose, createCommand, @@ -69,7 +68,6 @@ import { apiSaveEnvironmentVariablesCompose, apiUpdateCompose, compose as composeTable, - deployments, environments, projects, } from "@/server/db/schema"; @@ -285,17 +283,8 @@ export const composeRouter = createTRPCRouter({ await checkServicePermissionAndAccess(ctx, input.composeId, { deployment: ["create"], }); - await cancelAllQueuedDeploymentsByComposeId(input.composeId); await cleanQueuesByCompose(input.composeId); - const hasRunning = await db.query.deployments.findFirst({ - where: and( - eq(deployments.composeId, input.composeId), - eq(deployments.status, "running"), - ), - }); - if (!hasRunning) { - await updateCompose(input.composeId, { composeStatus: "idle" }); - } + await updateCompose(input.composeId, { composeStatus: "idle" }); return { success: true, message: "Queues cleaned successfully" }; }), clearDeployments: protectedProcedure diff --git a/apps/dokploy/server/queues/queueSetup.ts b/apps/dokploy/server/queues/queueSetup.ts index 5632c948d..c682f1bd6 100644 --- a/apps/dokploy/server/queues/queueSetup.ts +++ b/apps/dokploy/server/queues/queueSetup.ts @@ -1,4 +1,5 @@ import { + cancelAllQueuedDeployments, createDeployment, createDeploymentCompose, createDeploymentPreview, @@ -16,6 +17,7 @@ import { resolveBuildsConcurrency } from "./concurrency"; import { processDeploymentJob } from "./deployments-queue"; import { type InMemoryJob, InMemoryQueue } from "./in-memory-queue"; import type { DeploymentJob } from "./queue-types"; +import { withServiceLock } from "./service-lock"; /** * Deployment queue. @@ -111,26 +113,32 @@ if (!IS_CLOUD) { } export const cleanQueuesByApplication = async (applicationId: string) => { - const removed = myQueue.removeWaiting( - (data) => - data.applicationType === "application" && - data.applicationId === applicationId, - ); - if (removed > 0) { - console.log( - `Removed ${removed} waiting job(s) for application ${applicationId}`, + return withServiceLock(`application:${applicationId}`, async () => { + await cancelAllQueuedDeployments("applicationId", applicationId); + const removed = myQueue.removeWaiting( + (data) => + data.applicationType === "application" && + data.applicationId === applicationId, ); - } + if (removed > 0) { + console.log( + `Removed ${removed} waiting job(s) for application ${applicationId}`, + ); + } + }); }; export const cleanQueuesByCompose = async (composeId: string) => { - const removed = myQueue.removeWaiting( - (data) => - data.applicationType === "compose" && data.composeId === composeId, - ); - if (removed > 0) { - console.log(`Removed ${removed} waiting job(s) for compose ${composeId}`); - } + return withServiceLock(`compose:${composeId}`, async () => { + await cancelAllQueuedDeployments("composeId", composeId); + const removed = myQueue.removeWaiting( + (data) => + data.applicationType === "compose" && data.composeId === composeId, + ); + if (removed > 0) { + console.log(`Removed ${removed} waiting job(s) for compose ${composeId}`); + } + }); }; export const cleanAllDeploymentQueue = async () => { @@ -191,55 +199,61 @@ const enqueueDeployment = async ( export const enqueueApplicationDeployment = async (jobData: DeploymentJob) => { if (jobData.applicationType !== "application") return; - await updateApplicationStatus(jobData.applicationId, "queued"); - return enqueueDeployment( - jobData, - () => - createDeployment({ - applicationId: jobData.applicationId, - title: jobData.titleLog, - description: jobData.descriptionLog, - status: "queued", - }), - () => updateApplicationStatus(jobData.applicationId, "idle"), - ); + return withServiceLock(`application:${jobData.applicationId}`, async () => { + await updateApplicationStatus(jobData.applicationId, "queued"); + return enqueueDeployment( + jobData, + () => + createDeployment({ + applicationId: jobData.applicationId, + title: jobData.titleLog, + description: jobData.descriptionLog, + status: "queued", + }), + () => updateApplicationStatus(jobData.applicationId, "idle"), + ); + }); }; export const enqueueComposeDeployment = async (jobData: DeploymentJob) => { if (jobData.applicationType !== "compose") return; - await updateCompose(jobData.composeId, { composeStatus: "queued" }); - return enqueueDeployment( - jobData, - () => - createDeploymentCompose({ - composeId: jobData.composeId, - title: jobData.titleLog, - description: jobData.descriptionLog, - status: "queued", - }), - () => updateCompose(jobData.composeId, { composeStatus: "idle" }), - ); + return withServiceLock(`compose:${jobData.composeId}`, async () => { + await updateCompose(jobData.composeId, { composeStatus: "queued" }); + return enqueueDeployment( + jobData, + () => + createDeploymentCompose({ + composeId: jobData.composeId, + title: jobData.titleLog, + description: jobData.descriptionLog, + status: "queued", + }), + () => updateCompose(jobData.composeId, { composeStatus: "idle" }), + ); + }); }; export const enqueuePreviewDeployment = async (jobData: DeploymentJob) => { if (jobData.applicationType !== "application-preview") return; - await updatePreviewDeployment(jobData.previewDeploymentId, { - previewStatus: "queued", + return withServiceLock(`preview:${jobData.previewDeploymentId}`, async () => { + await updatePreviewDeployment(jobData.previewDeploymentId, { + previewStatus: "queued", + }); + return enqueueDeployment( + jobData, + () => + createDeploymentPreview({ + previewDeploymentId: jobData.previewDeploymentId, + title: jobData.titleLog, + description: jobData.descriptionLog, + status: "queued", + }), + () => + updatePreviewDeployment(jobData.previewDeploymentId, { + previewStatus: "idle", + }), + ); }); - return enqueueDeployment( - jobData, - () => - createDeploymentPreview({ - previewDeploymentId: jobData.previewDeploymentId, - title: jobData.titleLog, - description: jobData.descriptionLog, - status: "queued", - }), - () => - updatePreviewDeployment(jobData.previewDeploymentId, { - previewStatus: "idle", - }), - ); }; export { myQueue }; diff --git a/apps/dokploy/server/queues/service-lock.ts b/apps/dokploy/server/queues/service-lock.ts new file mode 100644 index 000000000..bc1da0aff --- /dev/null +++ b/apps/dokploy/server/queues/service-lock.ts @@ -0,0 +1,21 @@ +const locks = new Map>(); + +export const withServiceLock = async ( + serviceKey: string, + fn: () => Promise, +): Promise => { + while (locks.has(serviceKey)) { + await locks.get(serviceKey); + } + let resolve!: () => void; + const p = new Promise((r) => { + resolve = r; + }); + locks.set(serviceKey, p); + try { + return await fn(); + } finally { + locks.delete(serviceKey); + resolve(); + } +}; diff --git a/packages/server/src/services/application.ts b/packages/server/src/services/application.ts index 745870d28..b0507aa03 100644 --- a/packages/server/src/services/application.ts +++ b/packages/server/src/services/application.ts @@ -35,6 +35,7 @@ import { getDokployUrl } from "./admin"; import { createDeployment, createDeploymentPreview, + getActiveDeploymentStatus, resolveQueuedDeployment, updateDeployment, updateDeploymentStatus, @@ -165,10 +166,22 @@ export const updateApplicationStatus = async ( applicationId: string, applicationStatus: Application["applicationStatus"], ) => { + let effectiveStatus = applicationStatus; + + if ( + applicationStatus === "done" || + applicationStatus === "error" || + applicationStatus === "idle" + ) { + effectiveStatus = + (await getActiveDeploymentStatus("applicationId", applicationId)) ?? + applicationStatus; + } + const application = await db .update(applications) .set({ - applicationStatus: applicationStatus, + applicationStatus: effectiveStatus, }) .where(eq(applications.applicationId, applicationId)) .returning(); diff --git a/packages/server/src/services/compose.ts b/packages/server/src/services/compose.ts index 74b346d9e..4c2bb0501 100644 --- a/packages/server/src/services/compose.ts +++ b/packages/server/src/services/compose.ts @@ -39,6 +39,7 @@ import { encodeBase64 } from "../utils/docker/utils"; import { getDokployUrl } from "./admin"; import { createDeploymentCompose, + getActiveDeploymentStatus, resolveQueuedDeployment, updateDeployment, updateDeploymentStatus, @@ -215,6 +216,17 @@ export const updateCompose = async ( composeData: Partial, ) => { const { appName, ...rest } = composeData; + + if ( + rest.composeStatus === "done" || + rest.composeStatus === "error" || + rest.composeStatus === "idle" + ) { + rest.composeStatus = + (await getActiveDeploymentStatus("composeId", composeId)) ?? + rest.composeStatus; + } + const composeResult = await db .update(compose) .set({ diff --git a/packages/server/src/services/deployment.ts b/packages/server/src/services/deployment.ts index ef9d0ebab..c0eeb37ba 100644 --- a/packages/server/src/services/deployment.ts +++ b/packages/server/src/services/deployment.ts @@ -122,6 +122,45 @@ export const findDeploymentByApplicationId = async (applicationId: string) => { export const QUEUED_LOG_MESSAGE = "Waiting for worker to pick job..."; +type DeploymentServiceColumn = + | "applicationId" + | "composeId" + | "previewDeploymentId"; + +export const getActiveDeploymentStatus = async ( + serviceColumn: DeploymentServiceColumn, + serviceId: string, +): Promise<"queued" | "running" | null> => { + const remaining = await db.query.deployments.findFirst({ + where: and( + eq(deployments[serviceColumn], serviceId), + inArray(deployments.status, ["queued", "running"]), + ), + columns: { status: true }, + }); + if (!remaining) return null; + return remaining.status === "running" ? "running" : "queued"; +}; + +export const cancelAllQueuedDeployments = async ( + serviceColumn: DeploymentServiceColumn, + serviceId: string, +) => { + return db + .update(deployments) + .set({ + status: "cancelled", + finishedAt: new Date().toISOString(), + }) + .where( + and( + eq(deployments[serviceColumn], serviceId), + eq(deployments.status, "queued"), + ), + ) + .returning(); +}; + export const resolveQueuedDeployment = async < T extends { deploymentId: string }, >( @@ -1043,41 +1082,12 @@ export const updateDeploymentStatus = async ( return application; }; -export const cancelAllQueuedDeploymentsByApplicationId = async ( +export const cancelAllQueuedDeploymentsByApplicationId = ( applicationId: string, -) => { - return db - .update(deployments) - .set({ - status: "cancelled", - finishedAt: new Date().toISOString(), - }) - .where( - and( - eq(deployments.applicationId, applicationId), - eq(deployments.status, "queued"), - ), - ) - .returning(); -}; +) => cancelAllQueuedDeployments("applicationId", applicationId); -export const cancelAllQueuedDeploymentsByComposeId = async ( - composeId: string, -) => { - return db - .update(deployments) - .set({ - status: "cancelled", - finishedAt: new Date().toISOString(), - }) - .where( - and( - eq(deployments.composeId, composeId), - eq(deployments.status, "queued"), - ), - ) - .returning(); -}; +export const cancelAllQueuedDeploymentsByComposeId = (composeId: string) => + cancelAllQueuedDeployments("composeId", composeId); export const createServerDeployment = async ( deployment: Omit< diff --git a/packages/server/src/services/preview-deployment.ts b/packages/server/src/services/preview-deployment.ts index cc3d3dc02..3cec4043d 100644 --- a/packages/server/src/services/preview-deployment.ts +++ b/packages/server/src/services/preview-deployment.ts @@ -15,7 +15,10 @@ import { authGithub } from "../utils/providers/github"; import { removeTraefikConfig } from "../utils/traefik/application"; import { manageDomain } from "../utils/traefik/domain"; import { findApplicationById } from "./application"; -import { removeDeploymentsByPreviewDeploymentId } from "./deployment"; +import { + getActiveDeploymentStatus, + removeDeploymentsByPreviewDeploymentId, +} from "./deployment"; import { createDomain } from "./domain"; import { findGithubById, getIssueComment } from "./github"; import { getWebServerSettings } from "./web-server-settings"; @@ -99,6 +102,18 @@ export const updatePreviewDeployment = async ( previewDeploymentId: string, previewDeploymentData: Partial, ) => { + if ( + previewDeploymentData.previewStatus === "done" || + previewDeploymentData.previewStatus === "error" || + previewDeploymentData.previewStatus === "idle" + ) { + previewDeploymentData.previewStatus = + (await getActiveDeploymentStatus( + "previewDeploymentId", + previewDeploymentId, + )) ?? previewDeploymentData.previewStatus; + } + const application = await db .update(previewDeployments) .set({ diff --git a/packages/server/src/services/transfer.ts b/packages/server/src/services/transfer.ts index 8466fe32c..10ca25836 100644 --- a/packages/server/src/services/transfer.ts +++ b/packages/server/src/services/transfer.ts @@ -40,6 +40,7 @@ import { deployApplication, findApplicationById, updateApplication, + updateApplicationStatus, } from "./application"; import { deployCompose, @@ -133,7 +134,7 @@ const updateStatus = async ( ) => { switch (serviceType) { case "application": - return await updateApplication(serviceId, { applicationStatus: status }); + return await updateApplicationStatus(serviceId, status); case "compose": return await updateCompose(serviceId, { composeStatus: status }); case "postgres":