github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Added more logging to TaskRunConcurrencyTracker and some more try/catches

Browse Source
This commit is contained in:
Matt Aitken
2024-08-13 14:54:50 +01:00
parent 5cf90da722
commit 937b61913d
1 changed files with 79 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/v3/services/taskRunConcurrencyTracker.server.ts
+79 -41
View File
@@ -26,6 +26,11 @@ class TaskRunConcurrencyTracker implements MessageQueueSubscriber {
async messageEnqueued(message: MessagePayload): Promise<void> {}
async messageDequeued(message: MessagePayload): Promise<void> {
logger.debug("TaskRunConcurrencyTracker.messageDequeued()", {
data: message.data,
messageId: message.messageId,
});
const data = this.getMessageData(message);
if (!data) {
logger.info(
@@ -45,6 +50,11 @@ class TaskRunConcurrencyTracker implements MessageQueueSubscriber {
}
async messageAcked(message: MessagePayload): Promise<void> {
logger.debug("TaskRunConcurrencyTracker.messageAcked()", {
data: message.data,
messageId: message.messageId,
});
const data = this.getMessageData(message);
if (!data) {
logger.info(
@@ -64,6 +74,11 @@ class TaskRunConcurrencyTracker implements MessageQueueSubscriber {
}
async messageNacked(message: MessagePayload): Promise<void> {
logger.debug("TaskRunConcurrencyTracker.messageNacked()", {
data: message.data,
messageId: message.messageId,
});
const data = this.getMessageData(message);
if (!data) {
logger.info(
@@ -103,14 +118,18 @@ class TaskRunConcurrencyTracker implements MessageQueueSubscriber {
environmentId: string;
deployed: boolean;
}): Promise<void> {
const pipeline = this.redis.pipeline();
try {
const pipeline = this.redis.pipeline();
pipeline.sadd(this.getTaskKey(projectId, taskId), runId);
pipeline.sadd(this.getTaskEnvironmentKey(projectId, taskId, environmentId), runId);
pipeline.sadd(this.getEnvironmentKey(projectId, environmentId), runId);
pipeline.sadd(this.getGlobalKey(deployed), runId);
pipeline.sadd(this.getTaskKey(projectId, taskId), runId);
pipeline.sadd(this.getTaskEnvironmentKey(projectId, taskId, environmentId), runId);
pipeline.sadd(this.getEnvironmentKey(projectId, environmentId), runId);
pipeline.sadd(this.getGlobalKey(deployed), runId);
await pipeline.exec();
await pipeline.exec();
} catch (error) {
logger.error("TaskRunConcurrencyTracker.executionStarted() error", { error });
}
}
private async executionFinished({
@@ -126,14 +145,18 @@ class TaskRunConcurrencyTracker implements MessageQueueSubscriber {
environmentId: string;
deployed: boolean;
}): Promise<void> {
const pipeline = this.redis.pipeline();
try {
const pipeline = this.redis.pipeline();
pipeline.srem(this.getTaskKey(projectId, taskId), runId);
pipeline.srem(this.getTaskEnvironmentKey(projectId, taskId, environmentId), runId);
pipeline.srem(this.getEnvironmentKey(projectId, environmentId), runId);
pipeline.srem(this.getGlobalKey(deployed), runId);
pipeline.srem(this.getTaskKey(projectId, taskId), runId);
pipeline.srem(this.getTaskEnvironmentKey(projectId, taskId, environmentId), runId);
pipeline.srem(this.getEnvironmentKey(projectId, environmentId), runId);
pipeline.srem(this.getGlobalKey(deployed), runId);
await pipeline.exec();
await pipeline.exec();
} catch (error) {
logger.error("TaskRunConcurrencyTracker.executionFinished() error", { error });
}
}
async taskConcurrentRunCount(projectId: string, taskId: string): Promise<number> {
@@ -149,21 +172,26 @@ class TaskRunConcurrencyTracker implements MessageQueueSubscriber {
}
private async getTaskCounts(projectId: string, taskIds: string[]): Promise<number[]> {
const pipeline = this.redis.pipeline();
taskIds.forEach((taskId) => {
pipeline.scard(this.getTaskKey(projectId, taskId));
});
const results = await pipeline.exec();
if (!results) {
try {
const pipeline = this.redis.pipeline();
taskIds.forEach((taskId) => {
pipeline.scard(this.getTaskKey(projectId, taskId));
});
const results = await pipeline.exec();
if (!results) {
return [];
}
return results.map(([err, count]) => {
if (err) {
console.error("Error in getTaskCounts:", err);
return 0;
}
return count as number;
});
} catch (error) {
logger.error("TaskRunConcurrencyTracker.getTaskCounts() error", { error });
return [];
}
return results.map(([err, count]) => {
if (err) {
console.error("Error in getTaskCounts:", err);
return 0;
}
return count as number;
});
}
async projectTotalConcurrentRunCount(projectId: string, taskIds: string[]): Promise<number> {
@@ -177,7 +205,7 @@ class TaskRunConcurrencyTracker implements MessageQueueSubscriber {
): Promise<Record<string, number>> {
const counts = await this.getTaskCounts(projectId, taskIds);
return taskIds.reduce((acc, taskId, index) => {
acc[taskId] = counts[index];
acc[taskId] = counts[index] ?? 0;
return acc;
}, {} as Record<string, number>);
}
@@ -186,23 +214,28 @@ class TaskRunConcurrencyTracker implements MessageQueueSubscriber {
projectId: string,
environmentIds: string[]
): Promise<Record<string, number>> {
const pipeline = this.redis.pipeline();
environmentIds.forEach((environmentId) => {
pipeline.scard(this.getEnvironmentKey(projectId, environmentId));
});
const results = await pipeline.exec();
if (!results) {
try {
const pipeline = this.redis.pipeline();
environmentIds.forEach((environmentId) => {
pipeline.scard(this.getEnvironmentKey(projectId, environmentId));
});
const results = await pipeline.exec();
if (!results) {
return Object.fromEntries(environmentIds.map((id) => [id, 0]));
}
return results.reduce((acc, [err, count], index) => {
if (err) {
console.error("Error in environmentConcurrentRunCounts:", err);
return acc;
}
acc[environmentIds[index]] = count as number;
return acc;
}, {} as Record<string, number>);
} catch (error) {
logger.error("TaskRunConcurrencyTracker.environmentConcurrentRunCounts() error", { error });
return Object.fromEntries(environmentIds.map((id) => [id, 0]));
}
return results.reduce((acc, [err, count], index) => {
if (err) {
console.error("Error in environmentConcurrentRunCounts:", err);
return acc;
}
acc[environmentIds[index]] = count as number;
return acc;
}, {} as Record<string, number>);
}
private getTaskKey(projectId: string, taskId: string): string {
@@ -231,6 +264,11 @@ function getTracker() {
);
}
logger.debug("Initializing TaskRunConcurrencyTracker", {
redisHost: env.REDIS_HOST,
redisPort: env.REDIS_PORT,
});
return new TaskRunConcurrencyTracker({
redis: {
keyPrefix: "concurrencytracker:",