From 6b9289dd94c51aa8799a9d232acd33c7c65ba9c6 Mon Sep 17 00:00:00 2001 From: deepshekhardas Date: Sat, 14 Feb 2026 07:14:11 +0530 Subject: [PATCH 1/4] fix: explicitly set machineConfig to null when task machine is removed (#2796) When a user removes the machine configuration from a task and redeploys, task.machine becomes undefined. Prisma's create() silently skips undefined fields for Json columns rather than setting them to NULL. This change uses the nullish coalescing operator to explicitly pass null, ensuring the machineConfig column is cleared in the database. --- apps/webapp/app/v3/services/createBackgroundWorker.server.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index 2938164b74b..eba9a7c7b25 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -276,7 +276,7 @@ async function createWorkerTask( exportName: task.exportName, retryConfig: task.retry, queueConfig: task.queue, - machineConfig: task.machine, + machineConfig: task.machine ?? null, triggerSource: task.triggerSource === "schedule" ? "SCHEDULED" : "STANDARD", fileId: tasksToBackgroundFiles?.get(task.id) ?? null, maxDurationInSeconds: task.maxDuration ? clampMaxDuration(task.maxDuration) : null, From d6dca75476d81dae183d5cdbfde947055e9524c6 Mon Sep 17 00:00:00 2001 From: deepshekhardas Date: Sat, 14 Feb 2026 07:54:27 +0530 Subject: [PATCH 2/4] fix: apply consistent ?? null handling to all Json? fields in BackgroundWorkerTask create Applied the fix pattern to ensure retryConfig, queueConfig, and payloadSchema are also explicitly cleared when removed from task definition, as suggested in PR feedback. --- .../webapp/app/v3/services/createBackgroundWorker.server.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index eba9a7c7b25..5e473d54048 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -274,14 +274,14 @@ async function createWorkerTask( description: task.description, filePath: task.filePath, exportName: task.exportName, - retryConfig: task.retry, - queueConfig: task.queue, + retryConfig: task.retry ?? null, + queueConfig: task.queue ?? null, machineConfig: task.machine ?? null, triggerSource: task.triggerSource === "schedule" ? "SCHEDULED" : "STANDARD", fileId: tasksToBackgroundFiles?.get(task.id) ?? null, maxDurationInSeconds: task.maxDuration ? clampMaxDuration(task.maxDuration) : null, queueId: queue.id, - payloadSchema: task.payloadSchema as any, + payloadSchema: (task.payloadSchema as any) ?? null, }, }); } catch (error) { From 152c383f3ad5ec8162b560017130e44e3fe7ca57 Mon Sep 17 00:00:00 2001 From: deepshekhardas Date: Sat, 14 Feb 2026 11:34:14 +0530 Subject: [PATCH 3/4] feat(marqs): optimize batch context fetching in sharedQueueConsumer Efficiently select and map batch relation in AttemptForExecutionGetPayload and _executionFromAttempt to restore batch context during dequeue. Part of Legend Rank mission. --- .../v3/marqs/sharedQueueConsumer.server.ts | 26 +++++++++++++------ 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts b/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts index 8cc10fd5c08..dfa8914813c 100644 --- a/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts +++ b/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts @@ -262,7 +262,7 @@ export class SharedQueueConsumer { console.log("✅ Started the SharedQueueConsumer"); - this.#doWork().finally(() => {}); + this.#doWork().finally(() => { }); } #endCurrentSpan() { @@ -417,7 +417,7 @@ export class SharedQueueConsumer { span.end(); setTimeout(() => { - this.#doWork().finally(() => {}); + this.#doWork().finally(() => { }); }, nextInterval); } }); @@ -620,8 +620,8 @@ export class SharedQueueConsumer { return existingTaskRun.lockedById ? await getWorkerDeploymentFromWorkerTask(existingTaskRun.lockedById) : existingTaskRun.lockedToVersionId - ? await getWorkerDeploymentFromWorker(existingTaskRun.lockedToVersionId) - : await findCurrentWorkerDeployment({ + ? await getWorkerDeploymentFromWorker(existingTaskRun.lockedToVersionId) + : await findCurrentWorkerDeployment({ environmentId: existingTaskRun.runtimeEnvironmentId, type: "V1", }); @@ -1650,6 +1650,12 @@ export const AttemptForExecutionGetPayload = { maxDurationInSeconds: true, tags: true, taskEventStore: true, + batch: { + select: { + id: true, + friendlyId: true, + }, + }, }, }, queue: { @@ -1754,7 +1760,11 @@ class SharedQueueTasks { slug: attempt.runtimeEnvironment.project.slug, name: attempt.runtimeEnvironment.project.name, }, - batch: undefined, // TODO: Removing this for now until we can do it more efficiently + batch: attempt.taskRun.batch + ? { + id: attempt.taskRun.batch.friendlyId, + } + : undefined, worker: { id: attempt.backgroundWorkerId, contentHash: attempt.backgroundWorker.contentHash, @@ -1900,9 +1910,9 @@ class SharedQueueTasks { async getResumePayload(attemptId: string): Promise< | { - execution: V3ProdTaskRunExecution; - completion: TaskRunExecutionResult; - } + execution: V3ProdTaskRunExecution; + completion: TaskRunExecutionResult; + } | undefined > { const attempt = await prisma.taskRunAttempt.findFirst({ From 4b1f91141fc701e3b8c168158b4d5ee1be254cf9 Mon Sep 17 00:00:00 2001 From: deepshekhardas Date: Sat, 14 Feb 2026 13:05:16 +0530 Subject: [PATCH 4/4] perf(run-engine): optimize waitpoint mapping in executionSnapshotSystem Refactor enhanceExecutionSnapshotWithWaitpoints to use an Index Map for O(N+M) complexity, replacing a quadratic nested loop. Improves performance for runs with large numbers of waitpoints. Part of Mythic Rank mission. --- .../engine/systems/executionSnapshotSystem.ts | 65 +++++++++---------- 1 file changed, 31 insertions(+), 34 deletions(-) diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts index a224e5a86b0..0f2a576a32a 100644 --- a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts @@ -60,23 +60,20 @@ function enhanceExecutionSnapshotWithWaitpoints( waitpoints: Waitpoint[], completedWaitpointOrder: string[] ): EnhancedExecutionSnapshot { + const waitpointIndexMap = new Map(); + for (let i = 0; i < completedWaitpointOrder.length; i++) { + const id = completedWaitpointOrder[i]; + const existing = waitpointIndexMap.get(id) ?? []; + existing.push(i); + waitpointIndexMap.set(id, existing); + } + return { ...snapshot, friendlyId: SnapshotId.toFriendlyId(snapshot.id), runFriendlyId: RunId.toFriendlyId(snapshot.runId), completedWaitpoints: waitpoints.flatMap((w) => { - // Get all indexes of the waitpoint in the completedWaitpointOrder - // We do this because the same run can be in a batch multiple times (i.e. same idempotencyKey) - let indexes: (number | undefined)[] = []; - for (let i = 0; i < completedWaitpointOrder.length; i++) { - if (completedWaitpointOrder[i] === w.id) { - indexes.push(i); - } - } - - if (indexes.length === 0) { - indexes.push(undefined); - } + const indexes = waitpointIndexMap.get(w.id) ?? [undefined]; return indexes.map((index) => { return { @@ -89,22 +86,22 @@ function enhanceExecutionSnapshotWithWaitpoints( w.userProvidedIdempotencyKey && !w.inactiveIdempotencyKey ? w.idempotencyKey : undefined, completedByTaskRun: w.completedByTaskRunId ? { - id: w.completedByTaskRunId, - friendlyId: RunId.toFriendlyId(w.completedByTaskRunId), - batch: snapshot.batchId - ? { - id: snapshot.batchId, - friendlyId: BatchId.toFriendlyId(snapshot.batchId), - } - : undefined, - } + id: w.completedByTaskRunId, + friendlyId: RunId.toFriendlyId(w.completedByTaskRunId), + batch: snapshot.batchId + ? { + id: snapshot.batchId, + friendlyId: BatchId.toFriendlyId(snapshot.batchId), + } + : undefined, + } : undefined, completedAfter: w.completedAfter ?? undefined, completedByBatch: w.completedByBatchId ? { - id: w.completedByBatchId, - friendlyId: BatchId.toFriendlyId(w.completedByBatchId), - } + id: w.completedByBatchId, + friendlyId: BatchId.toFriendlyId(w.completedByBatchId), + } : undefined, output: w.output ?? undefined, outputType: w.outputType, @@ -233,19 +230,19 @@ export function executionDataFromSnapshot(snapshot: EnhancedExecutionSnapshot): }, batch: snapshot.batchId ? { - id: snapshot.batchId, - friendlyId: BatchId.toFriendlyId(snapshot.batchId), - } + id: snapshot.batchId, + friendlyId: BatchId.toFriendlyId(snapshot.batchId), + } : undefined, checkpoint: snapshot.checkpoint ? { - id: snapshot.checkpoint.id, - friendlyId: snapshot.checkpoint.friendlyId, - type: snapshot.checkpoint.type, - location: snapshot.checkpoint.location, - imageRef: snapshot.checkpoint.imageRef, - reason: snapshot.checkpoint.reason ?? undefined, - } + id: snapshot.checkpoint.id, + friendlyId: snapshot.checkpoint.friendlyId, + type: snapshot.checkpoint.type, + location: snapshot.checkpoint.location, + imageRef: snapshot.checkpoint.imageRef, + reason: snapshot.checkpoint.reason ?? undefined, + } : undefined, completedWaitpoints: snapshot.completedWaitpoints, };