Skip to content

Commit a515874

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 add2047 commit a515874

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
@@ -1164,7 +1164,7 @@ export class RunEngine {
11641164
}
11651165

11661166
this.eventBus.emit("executionSnapshotCreated", {
1167-
time: taskRun.createdAt,
1167+
time: new Date(),
11681168
run: {
11691169
id: taskRun.id,
11701170
},

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)