Skip to content

Commit f2eed68

Browse files
committed
fix(run-engine): stamp the queued timeline event at write time
The run's createdAt is caller-supplied and the scheduler sets it to the exact schedule time, so using it for the emitted event backdated the queued entry on every scheduled run's timeline. Use the write moment, as the other nested-create paths do.
1 parent 8c6bf23 commit f2eed68

2 files changed

Lines changed: 82 additions & 1 deletion

File tree

internal-packages/run-engine/src/engine/index.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1165,7 +1165,7 @@ export class RunEngine {
11651165
}
11661166

11671167
this.eventBus.emit("executionSnapshotCreated", {
1168-
time: taskRun.createdAt,
1168+
time: new Date(),
11691169
run: {
11701170
id: taskRun.id,
11711171
},

internal-packages/run-engine/src/engine/tests/triggerSnapshotCollapse.test.ts

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,87 @@ describe("RunEngine trigger() execution snapshots", () => {
100100
}
101101
);
102102

103+
containerTest(
104+
"the QUEUED snapshot event is stamped at write time, not at an overridden run createdAt",
105+
async ({ prisma, redisOptions }) => {
106+
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
107+
108+
const engine = new RunEngine({
109+
prisma,
110+
worker: {
111+
redis: redisOptions,
112+
workers: 1,
113+
tasksPerWorker: 10,
114+
pollIntervalMs: 100,
115+
},
116+
queue: {
117+
redis: redisOptions,
118+
masterQueueConsumersDisabled: true,
119+
processWorkerQueueDebounceMs: 50,
120+
},
121+
runLock: {
122+
redis: redisOptions,
123+
},
124+
machines: {
125+
defaultMachine: "small-1x",
126+
machines: {
127+
"small-1x": {
128+
name: "small-1x" as const,
129+
cpu: 0.5,
130+
memory: 0.5,
131+
centsPerMs: 0.0001,
132+
},
133+
},
134+
baseCostInCents: 0.0001,
135+
},
136+
tracer: trace.getTracer("test", "0.0.0"),
137+
});
138+
139+
try {
140+
const taskIdentifier = "test-task";
141+
142+
await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
143+
144+
const stampedTimes: Date[] = [];
145+
engine.eventBus.on("executionSnapshotCreated", ({ time }) => {
146+
stampedTimes.push(time);
147+
});
148+
149+
const backdatedCreatedAt = new Date(Date.now() - 60 * 60 * 1000);
150+
const triggeredAt = Date.now();
151+
152+
const run = await engine.trigger(
153+
{
154+
number: 1,
155+
friendlyId: "run_1236",
156+
environment: authenticatedEnvironment,
157+
taskIdentifier,
158+
payload: "{}",
159+
payloadType: "application/json",
160+
context: {},
161+
traceContext: {},
162+
traceId: "t_collapse_3",
163+
spanId: "s_collapse_3",
164+
workerQueue: "main",
165+
queue: "task/test-task",
166+
isTest: false,
167+
tags: [],
168+
createdAt: backdatedCreatedAt,
169+
},
170+
prisma
171+
);
172+
173+
const storedRun = await prisma.taskRun.findUnique({ where: { id: run.id } });
174+
expect(storedRun?.createdAt.getTime()).toBe(backdatedCreatedAt.getTime());
175+
176+
expect(stampedTimes.length).toBe(1);
177+
expect(stampedTimes[0].getTime()).toBeGreaterThanOrEqual(triggeredAt);
178+
} finally {
179+
await engine.quit();
180+
}
181+
}
182+
);
183+
103184
containerTest(
104185
"a delayed run keeps DELAYED and QUEUED as separate snapshots",
105186
async ({ prisma, redisOptions }) => {

0 commit comments

Comments
 (0)