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.

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(),
 }));

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(),
 }));

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)

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

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 };

apps/dokploy/server/queues/service-lock.ts

@@ -0,0 +1,21 @@
+const locks = new Map<string, Promise<void>>();
+
+export const withServiceLock = async <T>(
+	serviceKey: string,
+	fn: () => Promise<T>,
+): Promise<T> => {
+	while (locks.has(serviceKey)) {
+		await locks.get(serviceKey);
+	}
+	let resolve!: () => void;
+	const p = new Promise<void>((r) => {
+		resolve = r;
+	});
+	locks.set(serviceKey, p);
+	try {
+		return await fn();
+	} finally {
+		locks.delete(serviceKey);
+		resolve();
+	}
+};

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();

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<Compose>,
 ) => {
 	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({

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<

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<PreviewDeployment>,
 ) => {
+	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({

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":