Skip to content

perf(run-engine): optimize waitpoint mapping in executionSnapshotSystem #4467

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
deepshekhardas wants to merge 4 commits into from
Closed

perf(run-engine): optimize waitpoint mapping in executionSnapshotSystem #4467

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6b9289d
fix: explicitly set machineConfig to null when task machine is remove…
Feb 14, 2026
d6dca75
fix: apply consistent ?? null handling to all Json? fields in Backgro…
Feb 14, 2026
152c383
feat(marqs): optimize batch context fetching in sharedQueueConsumer
Feb 14, 2026
4b1f911
perf(run-engine): optimize waitpoint mapping in executionSnapshotSystem
Feb 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 18 additions & 8 deletions apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,7 +262,7 @@ export class SharedQueueConsumer {

console.log("✅ Started the SharedQueueConsumer");

this.#doWork().finally(() => {});
this.#doWork().finally(() => { });

@devin-ai-integration devin-ai-integration Bot Aug 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Several touched files are no longer formatted per the repository's required formatter

Code in the changed files is re-indented into a style the project's mandated formatter does not produce (e.g. () => { } at apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts:265), so the required formatting check is violated.
Impact: The repository's formatting requirement is broken, producing noisy unrelated diffs and failing format checks.

Prettier-enforced style per AGENTS.md

AGENTS.md ("Coding style") states: "Formatting is enforced using Prettier. Run pnpm run format before committing." The PR introduces non-Prettier output in multiple places: .finally(() => { }) at apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts:265 and :420, the re-indented nested ternary at apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts:623-624 and :1763-1767, the return-type object at :1913-1915, and the object-literal re-indentation in internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts:89-104 and :233-245. Running pnpm run format will revert these.

Suggested change
this.#doWork().finally(() => { });
this.#doWork().finally(() => {});
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

}

#endCurrentSpan() {
Expand Down Expand Up @@ -417,7 +417,7 @@ export class SharedQueueConsumer {
span.end();

setTimeout(() => {
this.#doWork().finally(() => {});
this.#doWork().finally(() => { });
}, nextInterval);
}
});
Expand Down Expand Up @@ -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",
});
Expand Down Expand Up @@ -1650,6 +1650,12 @@ export const AttemptForExecutionGetPayload = {
maxDurationInSeconds: true,
tags: true,
taskEventStore: true,
batch: {
select: {
id: true,
friendlyId: true,
},
},
Comment on lines +1653 to +1658

@devin-ai-integration devin-ai-integration Bot Aug 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔍 PR contains three unrelated changes across two subsystems

The stated scope is a waitpoint-mapping optimization in the run engine, but the branch also re-adds batch context to the legacy shared-queue execution payload (apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts:1653-1658, :1763-1767) and changes nullable Json handling in apps/webapp/app/v3/services/createBackgroundWorker.server.ts:277-284. CONTRIBUTING.md explicitly asks for one issue per PR; splitting would also make the batch-relation change (which reverses a deliberate // TODO: Removing this for now until we can do it more efficiently) reviewable on its own merits, including the extra join cost per dequeue.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

},
},
queue: {
Expand Down Expand Up @@ -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,
Comment on lines +1763 to +1767

@devin-ai-integration devin-ai-integration Bot Aug 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔍 Batch id semantics in the restored execution payload match the attempt-creation path

batch.id is populated from attempt.taskRun.batch.friendlyId, which matches the convention used when the execution payload is built at attempt-creation time (apps/webapp/app/v3/services/createTaskRunAttempt.server.ts:244-246 uses batchTaskRun.friendlyId as id). One difference worth noting: that path resolves the batch through batchItems/BatchTaskRunItem, while this one uses the direct TaskRun.batchId relation, so runs whose batch association only exists via BatchTaskRunItem would report batch: undefined here, producing inconsistent ctx.batch between initial execution and resume.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

worker: {
id: attempt.backgroundWorkerId,
contentHash: attempt.backgroundWorker.contentHash,
Expand Down Expand Up @@ -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({
Expand Down
8 changes: 4 additions & 4 deletions apps/webapp/app/v3/services/createBackgroundWorker.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,14 +274,14 @@ async function createWorkerTask(
description: task.description,
filePath: task.filePath,
exportName: task.exportName,
retryConfig: task.retry,
queueConfig: task.queue,
machineConfig: task.machine,
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,
Comment on lines +277 to +284

@devin-ai-integration devin-ai-integration Bot Aug 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 Deployed task registration fails when a task has no retry, queue, machine or schema settings

Task records are created with explicit empty values (task.retry ?? null at apps/webapp/app/v3/services/createBackgroundWorker.server.ts:277-279 and 284) instead of simply leaving those settings out, and the database layer rejects that empty value, so registering such tasks silently fails.
Impact: Tasks that don't declare retry/queue/machine/payload-schema options can fail to be registered during deploy or dev registration, and the failure is only logged, leaving those tasks missing from the deployed worker.

Prisma rejects plain null for nullable Json fields

retryConfig, queueConfig, machineConfig and payloadSchema are Json? columns (internal-packages/database/prisma/schema.prisma:559-570). For nullable Json fields Prisma's create input is NullableJsonNullValueInput | InputJsonValue, i.e. Prisma.DbNull / Prisma.JsonNull — passing a bare null is a type error and, at runtime, produces a PrismaClientValidationError ("Invalid value provided. Expected NullableJsonNullValueInput or Json, provided null"). That error is swallowed by the surrounding try/catch in createWorkerTask (apps/webapp/app/v3/services/createBackgroundWorker.server.ts:287-322), which only logs, so the backgroundWorkerTask row is never created.

Also note this is a create, not an update: omitting the field (the previous undefined behaviour) already stores SQL NULL, so the ?? null change provides no benefit.

Suggested change
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,
retryConfig: task.retry,
queueConfig: task.queue,
machineConfig: task.machine,
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,
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

},
});
} catch (error) {
Expand Down
65 changes: 31 additions & 34 deletions internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,23 +60,20 @@ function enhanceExecutionSnapshotWithWaitpoints(
waitpoints: Waitpoint[],
completedWaitpointOrder: string[]
): EnhancedExecutionSnapshot {
const waitpointIndexMap = new Map<string, number[]>();
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 {
Expand All @@ -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,
Expand Down Expand Up @@ -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,
};
Expand Down