more o11y

Author: carderne

apps/webapp/app/services/runsReplicationService.server.ts

@@ -1307,6 +1307,7 @@ export class RunsReplicationService {
       run.id, // run_id
       run.updatedAt.getTime(), // updated_at
       run.createdAt.getTime(), // created_at
+      run.queueTimestamp?.getTime() ?? null, // queue_timestamp
       run.status, // status
       environmentType, // environment_type
       run.friendlyId, // friendly_id

apps/webapp/test/runsReplicationService.part1.test.ts

@@ -73,6 +73,7 @@ describe("RunsReplicationService (part 1/7)", () => {
         },
       });
 
+      const queueTimestamp = new Date("2026-08-11T12:34:56.789Z");
       const taskRun = await prisma.taskRun.create({
         data: {
           friendlyId: "run_1234",
@@ -81,6 +82,7 @@ describe("RunsReplicationService (part 1/7)", () => {
           traceId: "1234",
           spanId: "1234",
           queue: "test",
+          queueTimestamp,
           workerQueue: "us-east-1-next",
           region: "us-east-1",
           planType: "free",
@@ -100,7 +102,8 @@ describe("RunsReplicationService (part 1/7)", () => {
 
       const queryRuns = clickhouse.reader.query({
         name: "runs-replication",
-        query: "SELECT * FROM trigger_dev.task_runs_v2",
+        query:
+          "SELECT *, toString(toUnixTimestamp64Milli(queue_timestamp)) AS queue_timestamp_ms FROM trigger_dev.task_runs_v2",
         schema: z.any(),
       });
 
@@ -125,6 +128,7 @@ describe("RunsReplicationService (part 1/7)", () => {
           organization_id: organization.id,
           environment_type: "DEVELOPMENT",
           engine: "V2",
+          queue_timestamp_ms: queueTimestamp.getTime().toString(),
           trigger_source: "api",
           root_trigger_source: "dashboard",
           is_warm_start: 1,

internal-packages/run-engine/src/engine/systems/dequeueSystem.ts

@@ -156,6 +156,10 @@ export class DequeueSystem {
 
         const orgId = message.message.orgId;
         const runId = message.messageId;
+        const queueWaitMs =
+          typeof message.message.eligibleAtMs === "number"
+            ? Math.max(0, Date.now() - message.message.eligibleAtMs)
+            : undefined;
 
         this.$.logger.info("DequeueSystem.dequeueFromWorkerQueue dequeued message", {
           runId,
@@ -174,6 +178,9 @@ export class DequeueSystem {
         span.setAttribute("consumer_id", consumerId);
         span.setAttribute("worker_queue", workerQueue);
         span.setAttribute("blocking_pop", blockingPop ?? true);
+        if (queueWaitMs !== undefined) {
+          span.setAttribute("queue_wait_ms", queueWaitMs);
+        }
 
         //lock the run so nothing else can modify it
         try {

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

@@ -225,18 +225,31 @@ export class ScheduleEngine {
           instance.taskSchedule.timezone,
           nominalAt
         );
-        const { effectiveAt: candidateEffectiveAt } = calculateEffectiveScheduleTime({
+        const {
+          effectiveAt: candidateEffectiveAt,
+          effectiveRangeMs,
+          windowMs,
+          offsetMs: candidateDelayMs,
+          rangeWasClamped,
+        } = calculateEffectiveScheduleTime({
           nominalAt,
           nextNominalAt,
           schedulePhase,
           window: scheduleWindow,
         });
         const effectiveAt = this.options.cronSpreadEnabled ? candidateEffectiveAt : nominalAt;
+        const appliedDelayMs = effectiveAt.getTime() - nominalAt.getTime();
 
         span.setAttribute("cron_spread_enabled", this.options.cronSpreadEnabled);
+        span.setAttribute("schedule_window_type", scheduleWindow?.type ?? "none");
         span.setAttribute("next_scheduled_timestamp", nominalAt.toISOString());
         span.setAttribute("candidate_effective_schedule_time", candidateEffectiveAt.toISOString());
         span.setAttribute("effective_schedule_time", effectiveAt.toISOString());
+        span.setAttribute("candidate_delay_ms", candidateDelayMs);
+        span.setAttribute("applied_delay_ms", appliedDelayMs);
+        span.setAttribute("schedule_window_ms", windowMs);
+        span.setAttribute("effective_range_ms", effectiveRangeMs);
+        span.setAttribute("schedule_range_was_clamped", rangeWasClamped);
 
         const schedulingDelayMs = effectiveAt.getTime() - Date.now();
         span.setAttribute("scheduling_delay_ms", schedulingDelayMs);
@@ -248,6 +261,11 @@ export class ScheduleEngine {
           candidateEffectiveAt: candidateEffectiveAt.toISOString(),
           effectiveAt: effectiveAt.toISOString(),
           cronSpreadEnabled: this.options.cronSpreadEnabled,
+          scheduleWindowType: scheduleWindow?.type ?? "none",
+          candidateDelayMs,
+          appliedDelayMs,
+          effectiveRangeMs,
+          rangeWasClamped,
           schedulingDelayMs,
           generatorExpression: instance.taskSchedule.generatorExpression,
           timezone: instance.taskSchedule.timezone,