diff --git a/apps/webapp/app/v3/mollifier/readFallback.server.ts b/apps/webapp/app/v3/mollifier/readFallback.server.ts index 497bf96a3..abe1c87fb 100644 --- a/apps/webapp/app/v3/mollifier/readFallback.server.ts +++ b/apps/webapp/app/v3/mollifier/readFallback.server.ts @@ -1,7 +1,7 @@ import type { MollifierBuffer } from "@trigger.dev/redis-worker"; import { logger } from "~/services/logger.server"; -import { getMollifierBuffer } from "./mollifierBuffer.server"; import { deserialiseMollifierSnapshot } from "./mollifierSnapshot.server"; +import { getMollifierBuffer } from "./mollifierBuffer.server"; export type ReadFallbackInput = { runId: string; @@ -14,7 +14,28 @@ export type SyntheticRun = { status: "QUEUED" | "FAILED"; taskIdentifier: string | undefined; createdAt: Date; + payload: unknown; + payloadType: string | undefined; + metadata: unknown; + metadataType: string | undefined; + + idempotencyKey: string | undefined; + idempotencyKeyOptions: string[] | undefined; + isTest: boolean; + depth: number; + ttl: string | undefined; + tags: string[]; + lockedToVersion: string | undefined; + resumeParentOnCompletion: boolean; + parentTaskRunId: string | undefined; + + // Allocated at gate-accept time and embedded in the snapshot so the run's + // trace is continuous from QUEUED-in-buffer through executing post-drain. + traceId: string | undefined; + spanId: string | undefined; + parentSpanId: string | undefined; + error?: { code: string; message: string }; }; @@ -22,6 +43,14 @@ export type ReadFallbackDeps = { getBuffer?: () => MollifierBuffer | null; }; +function asString(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +function asStringArray(value: unknown): string[] { + return Array.isArray(value) && value.every((v) => typeof v === "string") ? (value as string[]) : []; +} + export async function findRunByIdWithMollifierFallback( input: ReadFallbackInput, deps: ReadFallbackDeps = {}, @@ -43,15 +72,36 @@ export async function findRunByIdWithMollifierFallback( } const snapshot = deserialiseMollifierSnapshot(entry.payload); - const taskIdentifier = - typeof snapshot.taskIdentifier === "string" ? snapshot.taskIdentifier : undefined; + const idempotencyKeyOptionsRaw = snapshot.idempotencyKeyOptions; + const idempotencyKeyOptions = Array.isArray(idempotencyKeyOptionsRaw) + ? asStringArray(idempotencyKeyOptionsRaw) + : undefined; return { friendlyId: entry.runId, status: entry.status === "FAILED" ? "FAILED" : "QUEUED", - taskIdentifier, + taskIdentifier: asString(snapshot.taskIdentifier), createdAt: entry.createdAt, - payload: snapshot, + + payload: snapshot.payload, + payloadType: asString(snapshot.payloadType), + metadata: snapshot.metadata, + metadataType: asString(snapshot.metadataType), + + idempotencyKey: asString(snapshot.idempotencyKey), + idempotencyKeyOptions, + isTest: snapshot.isTest === true, + depth: typeof snapshot.depth === "number" ? snapshot.depth : 0, + ttl: asString(snapshot.ttl), + tags: asStringArray(snapshot.tags), + lockedToVersion: asString(snapshot.lockToVersion), + resumeParentOnCompletion: snapshot.resumeParentOnCompletion === true, + parentTaskRunId: asString(snapshot.parentTaskRunId), + + traceId: asString(snapshot.traceId), + spanId: asString(snapshot.spanId), + parentSpanId: asString(snapshot.parentSpanId), + error: entry.lastError, }; } catch (err) { diff --git a/apps/webapp/test/mollifierReadFallback.test.ts b/apps/webapp/test/mollifierReadFallback.test.ts index ae8c55c4d..6a9a21254 100644 --- a/apps/webapp/test/mollifierReadFallback.test.ts +++ b/apps/webapp/test/mollifierReadFallback.test.ts @@ -123,4 +123,97 @@ describe("findRunByIdWithMollifierFallback", () => { expect(result!.status).toBe("FAILED"); expect(result!.error).toEqual({ code: "VALIDATION", message: "task not found" }); }); + + it("extracts snapshot-derived fields from the buffered payload", async () => { + const entry: BufferEntry = { + runId: "run_1", + envId: "env_a", + orgId: "org_1", + payload: JSON.stringify({ + taskIdentifier: "my-task", + payload: '{"foo":"bar"}', + payloadType: "application/json", + metadata: '{"customer":"acme"}', + metadataType: "application/json", + idempotencyKey: "client-abc", + idempotencyKeyOptions: ["payload"], + isTest: true, + depth: 2, + ttl: "1h", + tags: ["tag-a", "tag-b"], + lockToVersion: "20260511.1", + resumeParentOnCompletion: false, + parentTaskRunId: "run_parent", + }), + status: "QUEUED", + attempts: 0, + createdAt: NOW, + }; + const result = await findRunByIdWithMollifierFallback( + { runId: "run_1", environmentId: "env_a", organizationId: "org_1" }, + { getBuffer: () => fakeBuffer(entry) }, + ); + expect(result).not.toBeNull(); + expect(result!.payloadType).toBe("application/json"); + expect(result!.metadata).toBe('{"customer":"acme"}'); + expect(result!.metadataType).toBe("application/json"); + expect(result!.idempotencyKey).toBe("client-abc"); + expect(result!.idempotencyKeyOptions).toEqual(["payload"]); + expect(result!.isTest).toBe(true); + expect(result!.depth).toBe(2); + expect(result!.ttl).toBe("1h"); + expect(result!.tags).toEqual(["tag-a", "tag-b"]); + expect(result!.lockedToVersion).toBe("20260511.1"); + expect(result!.resumeParentOnCompletion).toBe(false); + expect(result!.parentTaskRunId).toBe("run_parent"); + }); + + it("extracts gate-allocated trace context from the snapshot", async () => { + const entry: BufferEntry = { + runId: "run_1", + envId: "env_a", + orgId: "org_1", + payload: JSON.stringify({ + taskIdentifier: "t", + traceId: "trace_abc", + spanId: "span_xyz", + parentSpanId: "span_parent", + }), + status: "QUEUED", + attempts: 0, + createdAt: NOW, + }; + const result = await findRunByIdWithMollifierFallback( + { runId: "run_1", environmentId: "env_a", organizationId: "org_1" }, + { getBuffer: () => fakeBuffer(entry) }, + ); + expect(result!.traceId).toBe("trace_abc"); + expect(result!.spanId).toBe("span_xyz"); + expect(result!.parentSpanId).toBe("span_parent"); + }); + + it("defaults snapshot-derived fields to safe values when absent", async () => { + const entry: BufferEntry = { + runId: "run_1", + envId: "env_a", + orgId: "org_1", + payload: JSON.stringify({ taskIdentifier: "t" }), + status: "QUEUED", + attempts: 0, + createdAt: NOW, + }; + const result = await findRunByIdWithMollifierFallback( + { runId: "run_1", environmentId: "env_a", organizationId: "org_1" }, + { getBuffer: () => fakeBuffer(entry) }, + ); + expect(result!.payloadType).toBeUndefined(); + expect(result!.metadata).toBeUndefined(); + expect(result!.idempotencyKey).toBeUndefined(); + expect(result!.isTest).toBe(false); + expect(result!.depth).toBe(0); + expect(result!.tags).toEqual([]); + expect(result!.resumeParentOnCompletion).toBe(false); + expect(result!.traceId).toBeUndefined(); + expect(result!.spanId).toBeUndefined(); + }); });