|
| 1 | +import { json } from "@remix-run/server-runtime"; |
| 2 | +import { Ratelimit } from "@upstash/ratelimit"; |
| 3 | +import { tryCatch } from "@trigger.dev/core"; |
| 4 | +import { DevDisconnectRequestBody } from "@trigger.dev/core/v3"; |
| 5 | +import { BulkActionId, RunId } from "@trigger.dev/core/v3/isomorphic"; |
| 6 | +import { BulkActionNotificationType, BulkActionType } from "@trigger.dev/database"; |
| 7 | +import { prisma } from "~/db.server"; |
| 8 | +import { logger } from "~/services/logger.server"; |
| 9 | +import { RateLimiter } from "~/services/rateLimiter.server"; |
| 10 | +import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server"; |
| 11 | +import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server"; |
| 12 | +import { commonWorker } from "~/v3/commonWorker.server"; |
| 13 | +import pMap from "p-map"; |
| 14 | + |
| 15 | +const CANCEL_REASON = "Dev session ended (CLI exited)"; |
| 16 | + |
| 17 | +// Below this threshold, cancel runs inline with pMap. |
| 18 | +// Above it, create a bulk action and process asynchronously. |
| 19 | +const BULK_ACTION_THRESHOLD = 25; |
| 20 | + |
| 21 | +// Maximum number of runs that can be cancelled in a single disconnect call. |
| 22 | +const MAX_RUNS = 500; |
| 23 | + |
| 24 | +// Rate limit: 5 calls per minute per environment |
| 25 | +const disconnectRateLimiter = new RateLimiter({ |
| 26 | + keyPrefix: "dev-disconnect", |
| 27 | + limiter: Ratelimit.fixedWindow(5, "1 m"), |
| 28 | + logFailure: true, |
| 29 | +}); |
| 30 | + |
| 31 | +const { action } = createActionApiRoute( |
| 32 | + { |
| 33 | + body: DevDisconnectRequestBody, |
| 34 | + maxContentLength: 1024 * 256, // 256KB |
| 35 | + method: "POST", |
| 36 | + }, |
| 37 | + async ({ authentication, body }) => { |
| 38 | + // Only allow dev environments — this endpoint uses finalizeRun which |
| 39 | + // skips PENDING_CANCEL and immediately finalizes executing runs. |
| 40 | + if (authentication.environment.type !== "DEVELOPMENT") { |
| 41 | + return json({ error: "This endpoint is only available for dev environments" }, { status: 403 }); |
| 42 | + } |
| 43 | + |
| 44 | + const environmentId = authentication.environment.id; |
| 45 | + |
| 46 | + // Rate limit per environment |
| 47 | + const rateLimitResult = await disconnectRateLimiter.limit(environmentId); |
| 48 | + if (!rateLimitResult.success) { |
| 49 | + return json( |
| 50 | + { error: "Rate limit exceeded", retryAfter: Math.ceil((rateLimitResult.reset - Date.now()) / 1000) }, |
| 51 | + { status: 429 } |
| 52 | + ); |
| 53 | + } |
| 54 | + |
| 55 | + if (body.runFriendlyIds.length > MAX_RUNS) { |
| 56 | + return json( |
| 57 | + { error: `A maximum of ${MAX_RUNS} runs can be cancelled per request` }, |
| 58 | + { status: 400 } |
| 59 | + ); |
| 60 | + } |
| 61 | + |
| 62 | + const { runFriendlyIds } = body; |
| 63 | + |
| 64 | + if (runFriendlyIds.length === 0) { |
| 65 | + return json({ cancelled: 0 }, { status: 200 }); |
| 66 | + } |
| 67 | + |
| 68 | + logger.info("Dev disconnect: cancelling runs", { |
| 69 | + environmentId, |
| 70 | + runCount: runFriendlyIds.length, |
| 71 | + }); |
| 72 | + |
| 73 | + // For small numbers of runs, cancel inline |
| 74 | + if (runFriendlyIds.length <= BULK_ACTION_THRESHOLD) { |
| 75 | + const cancelled = await cancelRunsInline(runFriendlyIds, environmentId); |
| 76 | + return json({ cancelled }, { status: 200 }); |
| 77 | + } |
| 78 | + |
| 79 | + // For large numbers, create a bulk action to process asynchronously |
| 80 | + const bulkActionId = await createBulkCancelAction( |
| 81 | + runFriendlyIds, |
| 82 | + authentication.environment.project.id, |
| 83 | + environmentId |
| 84 | + ); |
| 85 | + |
| 86 | + logger.info("Dev disconnect: created bulk action for large run set", { |
| 87 | + environmentId, |
| 88 | + bulkActionId, |
| 89 | + runCount: runFriendlyIds.length, |
| 90 | + }); |
| 91 | + |
| 92 | + return json({ cancelled: 0, bulkActionId }, { status: 200 }); |
| 93 | + } |
| 94 | +); |
| 95 | + |
| 96 | +async function cancelRunsInline( |
| 97 | + runFriendlyIds: string[], |
| 98 | + environmentId: string |
| 99 | +): Promise<number> { |
| 100 | + const runIds = runFriendlyIds.map((fid) => RunId.toId(fid)); |
| 101 | + |
| 102 | + const runs = await prisma.taskRun.findMany({ |
| 103 | + where: { |
| 104 | + id: { in: runIds }, |
| 105 | + runtimeEnvironmentId: environmentId, |
| 106 | + }, |
| 107 | + select: { |
| 108 | + id: true, |
| 109 | + engine: true, |
| 110 | + friendlyId: true, |
| 111 | + status: true, |
| 112 | + createdAt: true, |
| 113 | + completedAt: true, |
| 114 | + taskEventStore: true, |
| 115 | + }, |
| 116 | + }); |
| 117 | + |
| 118 | + let cancelled = 0; |
| 119 | + const cancelService = new CancelTaskRunService(prisma); |
| 120 | + |
| 121 | + await pMap( |
| 122 | + runs, |
| 123 | + async (run) => { |
| 124 | + const [error, result] = await tryCatch( |
| 125 | + cancelService.call(run, { reason: CANCEL_REASON, finalizeRun: true }) |
| 126 | + ); |
| 127 | + |
| 128 | + if (error) { |
| 129 | + logger.error("Dev disconnect: failed to cancel run", { |
| 130 | + runId: run.id, |
| 131 | + error, |
| 132 | + }); |
| 133 | + } else if (result && !result.alreadyFinished) { |
| 134 | + cancelled++; |
| 135 | + } |
| 136 | + }, |
| 137 | + { concurrency: 10 } |
| 138 | + ); |
| 139 | + |
| 140 | + logger.info("Dev disconnect: completed inline cancellation", { |
| 141 | + environmentId, |
| 142 | + cancelled, |
| 143 | + total: runFriendlyIds.length, |
| 144 | + }); |
| 145 | + |
| 146 | + return cancelled; |
| 147 | +} |
| 148 | + |
| 149 | +async function createBulkCancelAction( |
| 150 | + runFriendlyIds: string[], |
| 151 | + projectId: string, |
| 152 | + environmentId: string |
| 153 | +): Promise<string> { |
| 154 | + const { id, friendlyId } = BulkActionId.generate(); |
| 155 | + |
| 156 | + await prisma.bulkActionGroup.create({ |
| 157 | + data: { |
| 158 | + id, |
| 159 | + friendlyId, |
| 160 | + projectId, |
| 161 | + environmentId, |
| 162 | + name: "Dev session disconnect", |
| 163 | + type: BulkActionType.CANCEL, |
| 164 | + params: { runId: runFriendlyIds, finalizeRun: true }, |
| 165 | + queryName: "bulk_action_v1", |
| 166 | + totalCount: runFriendlyIds.length, |
| 167 | + completionNotification: BulkActionNotificationType.NONE, |
| 168 | + }, |
| 169 | + }); |
| 170 | + |
| 171 | + await commonWorker.enqueue({ |
| 172 | + id: `processBulkAction-${id}`, |
| 173 | + job: "processBulkAction", |
| 174 | + payload: { bulkActionId: id }, |
| 175 | + }); |
| 176 | + |
| 177 | + return friendlyId; |
| 178 | +} |
| 179 | + |
| 180 | +export { action }; |
0 commit comments