Merge branch 'canary' into 2842-deployment-stuck-with-remote-server

apps/dokploy/package.json

@@ -99,7 +99,7 @@
 		"better-auth": "1.4.18",
 		"bl": "6.0.11",
 		"boxen": "^7.1.1",
-		"bullmq": "5.4.2",
+		"bullmq": "5.67.3",
 		"shell-quote": "^1.8.1",
 		"class-variance-authority": "^0.7.1",
 		"clsx": "^2.1.1",

apps/dokploy/server/api/routers/application.ts

@@ -57,9 +57,11 @@ import {
 	apiUpdateApplication,
 	applications,
 } from "@/server/db/schema";
+import { deploymentWorker } from "@/server/queues/deployments-queue";
 import type { DeploymentJob } from "@/server/queues/queue-types";
 import {
 	cleanQueuesByApplication,
+	getJobsByApplicationId,
 	killDockerBuild,
 	myQueue,
 } from "@/server/queues/queueSetup";
@@ -240,6 +242,15 @@ export const applicationRouter = createTRPCRouter({
 				.where(eq(applications.applicationId, input.applicationId))
 				.returning();
 
+			if (!IS_CLOUD) {
+				const queueJobs = await getJobsByApplicationId(input.applicationId);
+				for (const job of queueJobs) {
+					if (job.id) {
+						deploymentWorker.cancelJob(job.id, "User requested cancellation");
+					}
+				}
+			}
+
 			const cleanupOperations = [
 				async () => await deleteAllMiddlewares(application),
 				async () => await removeDeployments(application),

apps/dokploy/server/api/routers/compose.ts

@@ -58,9 +58,11 @@ import {
 	apiUpdateCompose,
 	compose as composeTable,
 } from "@/server/db/schema";
+import { deploymentWorker } from "@/server/queues/deployments-queue";
 import type { DeploymentJob } from "@/server/queues/queue-types";
 import {
 	cleanQueuesByCompose,
+	getJobsByComposeId,
 	killDockerBuild,
 	myQueue,
 } from "@/server/queues/queueSetup";
@@ -222,6 +224,15 @@ export const composeRouter = createTRPCRouter({
 				.where(eq(composeTable.composeId, input.composeId))
 				.returning();
 
+			if (!IS_CLOUD) {
+				const queueJobs = await getJobsByComposeId(input.composeId);
+				for (const job of queueJobs) {
+					if (job.id) {
+						deploymentWorker.cancelJob(job.id, "User requested cancellation");
+					}
+				}
+			}
+
 			const cleanupOperations = [
 				async () => await removeCompose(composeResult, input.deleteVolumes),
 				async () => await removeDeploymentsByComposeId(composeResult),

apps/dokploy/server/queues/queueSetup.ts

@@ -10,6 +10,16 @@ const myQueue = new Queue("deployments", {
 	connection: redisConfig,
 });
 
+export const getJobsByApplicationId = async (applicationId: string) => {
+	const jobs = await myQueue.getJobs();
+	return jobs.filter((job) => job?.data?.applicationId === applicationId);
+};
+
+export const getJobsByComposeId = async (composeId: string) => {
+	const jobs = await myQueue.getJobs();
+	return jobs.filter((job) => job?.data?.composeId === composeId);
+};
+
 process.on("SIGTERM", () => {
 	myQueue.close();
 	process.exit(0);

apps/schedules/package.json

@@ -11,7 +11,7 @@
 		"@dokploy/server": "workspace:*",
 		"@hono/node-server": "^1.14.3",
 		"@hono/zod-validator": "0.3.0",
-		"bullmq": "5.4.2",
+		"bullmq": "5.67.3",
 		"dotenv": "^16.4.5",
 		"drizzle-orm": "^0.41.0",
 		"hono": "^4.11.7",

apps/schedules/src/index.ts

@@ -47,25 +47,25 @@ app.post("/update-backup", zValidator("json", jobQueueSchema), async (c) => {
 			result = await removeJob({
 				backupId: data.backupId,
 				type: "backup",
-				cronSchedule: job.pattern,
+				cronSchedule: job.pattern || "",
 			});
 		} else if (data.type === "server") {
 			result = await removeJob({
 				serverId: data.serverId,
 				type: "server",
-				cronSchedule: job.pattern,
+				cronSchedule: job.pattern || "",
 			});
 		} else if (data.type === "schedule") {
 			result = await removeJob({
 				scheduleId: data.scheduleId,
 				type: "schedule",
-				cronSchedule: job.pattern,
+				cronSchedule: job.pattern || "",
 			});
 		} else if (data.type === "volume-backup") {
 			result = await removeJob({
 				volumeBackupId: data.volumeBackupId,
 				type: "volume-backup",
-				cronSchedule: job.pattern,
+				cronSchedule: job.pattern || "",
 			});
 		}
 		logger.info({ result }, "Job removed");

apps/schedules/src/queue.ts

@@ -1,13 +1,11 @@
 import { Queue, type RepeatableJob } from "bullmq";
-import IORedis from "ioredis";
 import { logger } from "./logger.js";
 import type { QueueJob } from "./schema.js";
 
-export const connection = new IORedis(process.env.REDIS_URL!, {
-	maxRetriesPerRequest: null,
-});
 export const jobQueue = new Queue("backupQueue", {
-	connection,
+	connection: {
+		url: process.env.REDIS_URL!,
+	},
 	defaultJobOptions: {
 		removeOnComplete: true,
 		removeOnFail: true,

apps/schedules/src/workers.ts

@@ -1,6 +1,5 @@
 import { type Job, Worker } from "bullmq";
 import { logger } from "./logger.js";
-import { connection } from "./queue.js";
 import type { QueueJob } from "./schema.js";
 import { runJobs } from "./utils.js";
 
@@ -12,7 +11,9 @@ export const firstWorker = new Worker(
 	},
 	{
 		concurrency: 100,
-		connection,
+		connection: {
+			url: process.env.REDIS_URL!,
+		},
 	},
 );
 export const secondWorker = new Worker(
@@ -23,7 +24,9 @@ export const secondWorker = new Worker(
 	},
 	{
 		concurrency: 100,
-		connection,
+		connection: {
+			url: process.env.REDIS_URL!,
+		},
 	},
 );
 
@@ -35,6 +38,8 @@ export const thirdWorker = new Worker(
 	},
 	{
 		concurrency: 100,
-		connection,
+		connection: {
+			url: process.env.REDIS_URL!,
+		},
 	},
 );