Repository navigation
Expand file tree
/
Copy pathrecover-stuck-runs.ts
More file actions
executable file
·374 lines (337 loc) · 14.1 KB
/
Copy pathrecover-stuck-runs.ts
File metadata and controls
executable file
·374 lines (337 loc) · 14.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
#!/usr/bin/env tsx
/**
* Recovery script for runs stuck in currentConcurrency with QUEUED execution status
*
* PROBLEM:
* During high database load, runs can get dequeued from Redis (added to currentConcurrency)
* but fail to update their execution status in the database. This leaves them stuck in an
* inconsistent state where they won't be re-dequeued because they're marked as "in progress"
* in Redis, but their database state still shows QUEUED.
*
* SOLUTION:
* This script identifies and recovers these stuck runs by:
* 1. Reading from the environment currentConcurrency Redis set
* 2. Checking which runs have QUEUED execution status (inconsistent state)
* 3. Re-adding them to their specific queue sorted sets
* 4. Removing them from the queue-specific currentConcurrency sets
* 5. Removing them from the environment-level currentConcurrency set
*
* SAFETY:
* - Dry-run mode when no write Redis URL is provided (read-only, no writes)
* - Uses separate Redis connections for reads and writes
* - Write connection only created when redisWriteUrl is provided
*
* ARGUMENTS:
* <environmentId> The Trigger.dev environment ID (e.g., env_abc123)
* <postgresUrl> PostgreSQL connection string
* <redisReadUrl> Redis connection string for reads (redis:// or rediss://)
* [redisWriteUrl] Optional Redis connection string for writes (omit for dry-run)
*
* USAGE:
* tsx scripts/recover-stuck-runs.ts <environmentId> <postgresUrl> <redisReadUrl> [redisWriteUrl]
*
* EXAMPLES:
*
* Dry-run mode (safe, no writes):
* tsx scripts/recover-stuck-runs.ts env_1234567890 \
* "postgresql://user:pass@localhost:5432/triggerdev" \
* "redis://readonly.example.com:6379"
*
* Execute mode (makes actual changes):
* tsx scripts/recover-stuck-runs.ts env_1234567890 \
* "postgresql://user:pass@localhost:5432/triggerdev" \
* "redis://readonly.example.com:6379" \
* "redis://writeonly.example.com:6379"
*/
import type { TaskRunExecutionStatus } from "@trigger.dev/database";
import { PrismaClient } from "@trigger.dev/database";
import { createRedisClient } from "@internal/redis";
interface StuckRun {
runId: string;
orgId: string;
projectId: string;
environmentId: string;
queue: string;
concurrencyKey: string | null;
executionStatus: TaskRunExecutionStatus;
snapshotCreatedAt: Date;
taskIdentifier: string;
}
interface RedisOperation {
type: "ZADD" | "SREM";
key: string;
args: (string | number)[];
description: string;
}
async function main() {
const [environmentId, postgresUrl, redisReadUrl, redisWriteUrl] = process.argv.slice(2);
if (!environmentId || !postgresUrl || !redisReadUrl) {
console.error(
"Usage: tsx scripts/recover-stuck-runs.ts <environmentId> <postgresUrl> <redisReadUrl> [redisWriteUrl]"
);
console.error("");
console.error("Dry-run mode when no redisWriteUrl is provided (read-only).");
console.error("Execute mode when redisWriteUrl is provided (makes actual changes).");
console.error("");
console.error("Example (dry-run):");
console.error(" tsx scripts/recover-stuck-runs.ts env_1234567890 \\");
console.error(' "postgresql://user:pass@localhost:5432/triggerdev" \\');
console.error(' "redis://readonly.example.com:6379"');
console.error("");
console.error("Example (execute):");
console.error(" tsx scripts/recover-stuck-runs.ts env_1234567890 \\");
console.error(' "postgresql://user:pass@localhost:5432/triggerdev" \\');
console.error(' "redis://readonly.example.com:6379" \\');
console.error(' "redis://writeonly.example.com:6379"');
process.exit(1);
}
const executeMode = !!redisWriteUrl;
if (executeMode) {
console.log("⚠️ EXECUTE MODE - Changes will be made to Redis\n");
} else {
console.log("🔍 DRY RUN MODE - No changes will be made to Redis\n");
}
console.log(`🔍 Scanning for stuck runs in environment: ${environmentId}`);
// Create Prisma client with the provided connection URL
const prisma = new PrismaClient({
datasources: {
db: {
url: postgresUrl,
},
},
});
try {
// Get environment details
const environment = await prisma.runtimeEnvironment.findUnique({
where: { id: environmentId },
include: {
organization: true,
project: true,
},
});
if (!environment) {
console.error(`❌ Environment not found: ${environmentId}`);
process.exit(1);
}
console.log(`📍 Environment: ${environment.slug} (${environment.type})`);
console.log(`📍 Organization: ${environment.organization.slug}`);
console.log(`📍 Project: ${environment.project.slug}`);
// Parse Redis read URL
const redisReadUrlObj = new URL(redisReadUrl);
const redisReadOptions = {
host: redisReadUrlObj.hostname,
port: parseInt(redisReadUrlObj.port || "6379"),
username: redisReadUrlObj.username || undefined,
password: redisReadUrlObj.password || undefined,
enableAutoPipelining: false,
...(redisReadUrlObj.protocol === "rediss:"
? {
tls: {
// If connecting via localhost tunnel to a remote Redis, disable cert verification
rejectUnauthorized: redisReadUrlObj.hostname !== "localhost",
},
}
: {}),
};
// Create Redis read client
const redisRead = createRedisClient(redisReadOptions);
// Create Redis write client if redisWriteUrl is provided
let redisWrite = null;
if (redisWriteUrl) {
const redisWriteUrlObj = new URL(redisWriteUrl);
const redisWriteOptions = {
host: redisWriteUrlObj.hostname,
port: parseInt(redisWriteUrlObj.port || "6379"),
username: redisWriteUrlObj.username || undefined,
password: redisWriteUrlObj.password || undefined,
enableAutoPipelining: false,
...(redisWriteUrlObj.protocol === "rediss:"
? {
tls: {
// If connecting via localhost tunnel to a remote Redis, disable cert verification
rejectUnauthorized: redisWriteUrlObj.hostname !== "localhost",
},
}
: {}),
};
redisWrite = createRedisClient(redisWriteOptions);
}
try {
// Build the Redis key for environment-level currentConcurrency set
// Format: engine:runqueue:{org:X}:proj:Y:env:Z:currentConcurrency
const envConcurrencyKey = `engine:runqueue:{org:${environment.organizationId}}:proj:${environment.projectId}:env:${environmentId}:currentConcurrency`;
console.log(`\n🔑 Checking Redis key: ${envConcurrencyKey}`);
// Get all run IDs in the environment's currentConcurrency set
const runIds = await redisRead.smembers(envConcurrencyKey);
if (runIds.length === 0) {
console.log(`✅ No runs in currentConcurrency set`);
return;
}
console.log(`📊 Found ${runIds.length} runs in currentConcurrency set`);
// Query database for latest snapshots and queue info of these runs.
// NOTE: raw join of TaskRunExecutionSnapshot to TaskRun, the one TaskRun read not behind
// RunStore (a join, not a by-id read, in an ops script). Revisit at table cutover.
const runInfo = await prisma.$queryRaw<
Array<{
runId: string;
executionStatus: TaskRunExecutionStatus;
snapshotCreatedAt: Date;
organizationId: string;
projectId: string;
environmentId: string;
taskIdentifier: string;
queue: string;
concurrencyKey: string | null;
}>
>`
SELECT DISTINCT ON (s."runId")
s."runId",
s."executionStatus",
s."createdAt" as "snapshotCreatedAt",
r."organizationId",
r."projectId",
r."runtimeEnvironmentId" as "environmentId",
r."taskIdentifier",
r."queue",
r."concurrencyKey"
FROM "TaskRunExecutionSnapshot" s
INNER JOIN "TaskRun" r ON r.id = s."runId"
WHERE s."runId" = ANY(${runIds})
AND s."isValid" = true
ORDER BY s."runId", s."createdAt" DESC
`;
const stuckRuns: StuckRun[] = [];
// Find runs with QUEUED execution status (inconsistent state)
for (const info of runInfo) {
if (info.executionStatus === "QUEUED") {
stuckRuns.push({
runId: info.runId,
orgId: info.organizationId,
projectId: info.projectId,
environmentId: info.environmentId,
queue: info.queue,
concurrencyKey: info.concurrencyKey,
executionStatus: info.executionStatus,
snapshotCreatedAt: info.snapshotCreatedAt,
taskIdentifier: info.taskIdentifier,
});
}
}
if (stuckRuns.length === 0) {
console.log(`✅ No stuck runs found (all runs have progressed beyond QUEUED state)`);
return;
}
console.log(`\n⚠️ Found ${stuckRuns.length} stuck runs in QUEUED state:`);
console.log(`════════════════════════════════════════════════════════════════`);
for (const run of stuckRuns) {
const age = Date.now() - run.snapshotCreatedAt.getTime();
const ageMinutes = Math.floor(age / 1000 / 60);
console.log(` • Run: ${run.runId}`);
console.log(` Task: ${run.taskIdentifier}`);
console.log(` Queue: ${run.queue}`);
console.log(` Concurrency Key: ${run.concurrencyKey || "(none)"}`);
console.log(` Status: ${run.executionStatus}`);
console.log(` Stuck for: ${ageMinutes} minutes`);
console.log(` Snapshot created: ${run.snapshotCreatedAt.toISOString()}`);
console.log();
}
// Prepare recovery operations
console.log(
`\n⚡ ${executeMode ? "Executing" : "Planning"} recovery for ${stuckRuns.length} stuck runs`
);
console.log(`This will:`);
console.log(` 1. Add each run back to its specific queue sorted set`);
console.log(` 2. Remove each run from the queue-specific currentConcurrency set`);
console.log(` 3. Remove each run from the env-level currentConcurrency set`);
console.log();
let successCount = 0;
let failureCount = 0;
const currentTimestamp = Date.now();
for (const run of stuckRuns) {
try {
// Build queue key: engine:runqueue:{org:X}:proj:Y:env:Z:queue:QUEUENAME
// Build queue currentConcurrency key: engine:runqueue:{org:X}:proj:Y:env:Z:queue:QUEUENAME:currentConcurrency
const queueKey = run.concurrencyKey
? `engine:runqueue:{org:${run.orgId}}:proj:${run.projectId}:env:${run.environmentId}:queue:${run.queue}:ck:${run.concurrencyKey}`
: `engine:runqueue:{org:${run.orgId}}:proj:${run.projectId}:env:${run.environmentId}:queue:${run.queue}`;
const queueConcurrencyKey = `${queueKey}:currentConcurrency`;
const operations: RedisOperation[] = [
{
type: "ZADD",
key: queueKey,
args: [currentTimestamp, run.runId],
description: `Add run to queue sorted set with score ${currentTimestamp}`,
},
{
type: "SREM",
key: queueConcurrencyKey,
args: [run.runId],
description: `Remove run from queue currentConcurrency set`,
},
{
type: "SREM",
key: envConcurrencyKey,
args: [run.runId],
description: `Remove run from env currentConcurrency set`,
},
];
if (executeMode && redisWrite) {
// Execute operations using the write client
await redisWrite.zadd(queueKey, currentTimestamp, run.runId);
const removedFromQueue = await redisWrite.srem(queueConcurrencyKey, run.runId);
const removedFromEnv = await redisWrite.srem(envConcurrencyKey, run.runId);
console.log(` ✓ Recovered run ${run.runId} (${run.taskIdentifier})`);
if (removedFromQueue === 0) {
console.log(` ⚠ Run was not in queue currentConcurrency set`);
}
if (removedFromEnv === 0) {
console.log(` ⚠ Run was not in env currentConcurrency set`);
}
successCount++;
} else {
// Dry run - just show what would be done
console.log(` 📝 Would recover run ${run.runId} (${run.taskIdentifier}):`);
for (const op of operations) {
console.log(` ${op.type} ${op.key}`);
console.log(` Args: ${JSON.stringify(op.args)}`);
console.log(` (${op.description})`);
}
successCount++;
}
} catch (error) {
console.error(` ✗ Failed to recover run ${run.runId}:`, error);
failureCount++;
}
}
console.log(`\n═══════════════════════════════════════════════════════════════`);
if (executeMode) {
console.log(`✅ Recovery complete!`);
console.log(` Recovered: ${successCount}`);
console.log(` Failed: ${failureCount}`);
console.log();
console.log(`ℹ️ Note: The recovered runs should be automatically dequeued`);
console.log(` by the master queue consumers within a few seconds.`);
} else {
console.log(`📋 Dry run complete - no changes were made`);
console.log(` Would recover: ${successCount}`);
console.log(` Would fail: ${failureCount}`);
console.log();
console.log(`💡 To execute these changes, run again with a redisWriteUrl argument`);
}
} finally {
await redisRead.quit();
if (redisWrite) {
await redisWrite.quit();
}
}
} catch (error) {
console.error("❌ Error during recovery:", error);
throw error;
} finally {
await prisma.$disconnect();
}
}
main().catch((error) => {
console.error("Fatal error:", error);
process.exit(1);
});