diff --git a/.changeset/batch-trigger-debounce.md b/.changeset/batch-trigger-debounce.md new file mode 100644 index 00000000000..ccda1f78987 --- /dev/null +++ b/.changeset/batch-trigger-debounce.md @@ -0,0 +1,15 @@ +--- +"@trigger.dev/sdk": patch +"@trigger.dev/react-hooks": patch +--- + +`debounce` now works when you pass an array of items to `batchTrigger` or `batchTriggerAndWait`, and when you trigger from `useTaskTrigger`. Previously the option was accepted by the types and dropped before the request was sent, so every trigger created its own run instead of collapsing onto the debounce key. + +```ts +await myTask.batchTrigger([ + { payload: { id: "a" }, options: { debounce: { key: "same-key", delay: "30s" } } }, + { payload: { id: "b" }, options: { debounce: { key: "same-key", delay: "30s" } } }, +]); +``` + +The streaming (async iterable) forms of the batch calls were already forwarding `debounce` correctly. diff --git a/packages/react-hooks/src/hooks/useTaskTrigger.ts b/packages/react-hooks/src/hooks/useTaskTrigger.ts index a26fc84c9c4..979cd87983b 100644 --- a/packages/react-hooks/src/hooks/useTaskTrigger.ts +++ b/packages/react-hooks/src/hooks/useTaskTrigger.ts @@ -87,6 +87,7 @@ export function useTaskTrigger( metadata: options?.metadata, maxDuration: options?.maxDuration, lockToVersion: options?.version, + debounce: options?.debounce, }, }); diff --git a/packages/trigger-sdk/src/v3/batchDebounce.test.ts b/packages/trigger-sdk/src/v3/batchDebounce.test.ts new file mode 100644 index 00000000000..9b53699d055 --- /dev/null +++ b/packages/trigger-sdk/src/v3/batchDebounce.test.ts @@ -0,0 +1,242 @@ +import { apiClientManager } from "@trigger.dev/core/v3"; +import { runInMockTaskContext } from "@trigger.dev/core/v3/test"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { batch } from "./batch.js"; +import { createTask } from "./shared.js"; +import { tasks } from "./tasks.js"; + +const debounceFor = (i: number) => ({ + key: `warm-conn-notify:${i}`, + delay: "12h", + maxDelay: "24h", + mode: "trailing" as const, +}); + +const EXPECTED = [debounceFor(0), debounceFor(1)]; + +type Payload = { i: number }; + +const taskA = createTask({ + id: "task-a", + run: async (_payload: Payload) => ({ ok: true }), +}); + +const taskB = createTask({ + id: "task-b", + run: async (_payload: Payload) => ({ ok: true }), +}); + +type SentItem = { + index: number; + task: string; + options?: { debounce?: { key: string; delay: string; mode?: string; maxDelay?: string } }; +}; + +/** + * Captures the NDJSON item stream the SDK sends in phase 2 of a batch trigger, + * answering the only question these tests care about: what actually reached the + * wire. Phase 1 (create) and the item stream get canned success responses. + */ +function installBatchCapture() { + const sent: SentItem[] = []; + const originalFetch = globalThis.fetch; + + globalThis.fetch = (async (input: any, init?: RequestInit) => { + const url = typeof input === "string" ? input : (input?.url ?? String(input)); + + if (url.endsWith("/api/v3/batches")) { + const body = JSON.parse(String(init?.body)); + return Response.json({ id: "batch_test", runCount: body.runCount, isCached: false }); + } + + if (url.includes("/api/v3/batches/") && url.endsWith("/items")) { + const ndjson = await new Response(init?.body as any).text(); + const lines = ndjson.split("\n").filter((line) => line.trim().length > 0); + sent.push(...lines.map((line) => JSON.parse(line) as SentItem)); + + return Response.json({ + id: "batch_test", + itemsAccepted: lines.length, + itemsDeduplicated: 0, + sealed: true, + }); + } + + throw new Error(`Unexpected request during batch trigger: ${url}`); + }) as typeof fetch; + + return { + debounceOptions: () => + [...sent].sort((a, b) => a.index - b.index).map((item) => item.options?.debounce), + restore: () => { + globalThis.fetch = originalFetch; + }, + }; +} + +async function* asAsyncIterable(items: T[]): AsyncIterable { + for (const item of items) { + yield item; + } +} + +describe("batch trigger debounce forwarding", () => { + let capture: ReturnType; + + beforeEach(() => { + apiClientManager.setGlobalAPIClientConfiguration({ + baseURL: "http://localhost:3030", + accessToken: "tr_dev_test", + }); + capture = installBatchCapture(); + }); + + afterEach(() => { + capture.restore(); + apiClientManager.disable(); + }); + + const surfaces: Array<{ name: string; call: () => Promise }> = [ + { + name: "task.batchTrigger(array)", + call: () => + taskA.batchTrigger([ + { payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]), + }, + { + name: "task.batchTrigger(asyncIterable)", + call: () => + taskA.batchTrigger( + asAsyncIterable([ + { payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]) + ), + }, + { + name: "tasks.batchTrigger(array)", + call: () => + tasks.batchTrigger("task-a", [ + { payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]), + }, + { + name: "batch.trigger(array)", + call: () => + batch.trigger([ + { id: "task-a", payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { id: "task-b", payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]), + }, + { + name: "batch.trigger(asyncIterable)", + call: () => + batch.trigger( + asAsyncIterable([ + { id: "task-a" as const, payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { id: "task-b" as const, payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]) + ), + }, + { + name: "batch.triggerByTask(array)", + call: () => + batch.triggerByTask([ + { task: taskA, payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { task: taskB, payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]), + }, + { + name: "batch.triggerByTask(asyncIterable)", + call: () => + batch.triggerByTask( + asAsyncIterable([ + { task: taskA, payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { task: taskB, payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]) + ), + }, + ]; + + it.each(surfaces)("$name forwards debounce for every item", async ({ call }) => { + await call(); + + expect(capture.debounceOptions()).toEqual(EXPECTED); + }); + + const waitSurfaces: Array<{ name: string; call: () => Promise }> = [ + { + name: "task.batchTriggerAndWait(array)", + call: () => + taskA.batchTriggerAndWait([ + { payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]), + }, + { + name: "task.batchTriggerAndWait(asyncIterable)", + call: () => + taskA.batchTriggerAndWait( + asAsyncIterable([ + { payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]) + ), + }, + { + name: "tasks.batchTriggerAndWait(array)", + call: () => + tasks.batchTriggerAndWait("task-a", [ + { payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]), + }, + { + name: "batch.triggerAndWait(array)", + call: () => + batch.triggerAndWait([ + { id: "task-a", payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { id: "task-b", payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]), + }, + { + name: "batch.triggerAndWait(asyncIterable)", + call: () => + batch.triggerAndWait( + asAsyncIterable([ + { id: "task-a" as const, payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { id: "task-b" as const, payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]) + ), + }, + { + name: "batch.triggerByTaskAndWait(array)", + call: () => + batch.triggerByTaskAndWait([ + { task: taskA, payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { task: taskB, payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]), + }, + { + name: "batch.triggerByTaskAndWait(asyncIterable)", + call: () => + batch.triggerByTaskAndWait( + asAsyncIterable([ + { task: taskA, payload: { i: 0 }, options: { debounce: debounceFor(0) } }, + { task: taskB, payload: { i: 1 }, options: { debounce: debounceFor(1) } }, + ]) + ), + }, + ]; + + it.each(waitSurfaces)("$name forwards debounce for every item", async ({ call }) => { + await runInMockTaskContext(async () => { + await call(); + }); + + expect(capture.debounceOptions()).toEqual(EXPECTED); + }); +}); diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts index 50ab312eb39..d06e82ae1bd 100644 --- a/packages/trigger-sdk/src/v3/shared.ts +++ b/packages/trigger-sdk/src/v3/shared.ts @@ -749,7 +749,7 @@ export async function batchTriggerById( lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"), debounce: item.options?.debounce, }, - }; + } satisfies BatchItemNDJSON; }) ); @@ -1005,7 +1005,7 @@ export async function batchTriggerByIdAndWait( region: item.options?.region, debounce: item.options?.debounce, }, - }; + } satisfies BatchItemNDJSON; }) ); @@ -1271,7 +1271,7 @@ export async function batchTriggerTasks( lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"), debounce: item.options?.debounce, }, - }; + } satisfies BatchItemNDJSON; }) ); @@ -1532,7 +1532,7 @@ export async function batchTriggerAndWaitTasks( lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"), debounce: item.options?.debounce, }, - }; + } satisfies BatchItemNDJSON; } } @@ -2071,7 +2071,7 @@ async function* transformBatchItemsStreamForWait( region: item.options?.region, debounce: item.options?.debounce, }, - }; + } satisfies BatchItemNDJSON; } } @@ -2122,7 +2122,7 @@ async function* transformBatchByTaskItemsStream( lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"), debounce: item.options?.debounce, }, - }; + } satisfies BatchItemNDJSON; } } @@ -2286,7 +2286,7 @@ async function* transformSingleTaskBatchItemsStreamForWait( region: item.options?.region, debounce: item.options?.debounce, }, - }; + } satisfies BatchItemNDJSON; } } @@ -2423,8 +2423,9 @@ async function batchTrigger_internal( priority: item.options?.priority, region: item.options?.region, lockToVersion: item.options?.version ?? scopedEnvVar("TRIGGER_VERSION"), + debounce: item.options?.debounce, }, - }; + } satisfies BatchItemNDJSON; }) ); @@ -2854,8 +2855,9 @@ async function batchTriggerAndWait_internal