From 8ad4bd613244efea87b84d76ab92cae410952385 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 07:55:53 +0000 Subject: [PATCH 1/6] feat(sdk,webapp): add tags.delete() to remove tags from a run Adds a per-run tag removal path, mirroring the existing tags.add() flow end to end: - `tags.delete(tag | tags)` in the SDK, plus JSDoc on both `add` and `delete` - `DELETE /api/v1/runs/:runId/tags` as a method branch on the existing route - `RemoveTagsRequestBody` and `ApiClient#removeTags` in core - `RunStore#removeTags`, a single-statement UPDATE so there is no read-modify-write window racing a concurrent pushTags - a `remove_tags` snapshot patch for runs still in the buffer Removing a tag only affects the run it's called from. Removing a tag the run doesn't have, or an empty list, is an idempotent success. Also fixes the `add` span name, which was hardcoded to "tags.set()", and an OpenAPI code sample advertising a `runs.addTags(...)` function that doesn't exist. Co-Authored-By: Claude --- .changeset/tags-delete.md | 7 + .../app/routes/api.v1.runs.$runId.tags.ts | 160 +++++++++++- .../performTaskRunAlertsStoreRouting.test.ts | 3 + .../updateMetadataStoreRoutingHetero.test.ts | 3 + docs/docs.json | 1 + docs/management/runs/remove-tags.mdx | 4 + docs/tags.mdx | 34 ++- docs/v3-openapi.yaml | 85 +++++- .../run-store/src/PostgresRunStore.test.ts | 246 ++++++++++++++++++ .../run-store/src/PostgresRunStore.ts | 54 ++++ .../run-store/src/runOpsStore.ts | 9 + internal-packages/run-store/src/types.ts | 10 + packages/core/src/v3/apiClient/index.ts | 14 + packages/core/src/v3/schemas/api.ts | 6 + .../redis-worker/src/mollifier/buffer.test.ts | 216 +++++++++++++++ packages/redis-worker/src/mollifier/buffer.ts | 32 +++ packages/trigger-sdk/src/v3/tags.ts | 92 ++++++- 17 files changed, 959 insertions(+), 17 deletions(-) create mode 100644 .changeset/tags-delete.md create mode 100644 docs/management/runs/remove-tags.mdx diff --git a/.changeset/tags-delete.md b/.changeset/tags-delete.md new file mode 100644 index 00000000000..03ea12d3232 --- /dev/null +++ b/.changeset/tags-delete.md @@ -0,0 +1,7 @@ +--- +"@trigger.dev/sdk": patch +"@trigger.dev/core": patch +"@trigger.dev/redis-worker": patch +--- + +You can now remove tags from a run while it's running with `tags.delete("my-tag")` (or an array of tags). It only affects that run — every other run keeps the tag, and the tag stays available to filter by. diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts index be717efa355..5c468ba5db9 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts @@ -1,10 +1,10 @@ import { type ActionFunctionArgs, json } from "@remix-run/server-runtime"; -import { AddTagsRequestBody } from "@trigger.dev/core/v3"; +import { AddTagsRequestBody, RemoveTagsRequestBody } from "@trigger.dev/core/v3"; import type { BufferEntry } from "@trigger.dev/redis-worker"; import { z } from "zod"; import { prisma } from "~/db.server"; import { MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server"; -import { authenticateApiRequest } from "~/services/apiAuth.server"; +import { type AuthenticatedEnvironment, authenticateApiRequest } from "~/services/apiAuth.server"; import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server"; import { logger } from "~/services/logger.server"; import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server"; @@ -31,24 +31,60 @@ const ParamsSchema = z.object({ runId: z.string(), }); -export async function action({ request, params }: ActionFunctionArgs) { - if (request.method.toUpperCase() !== "POST") { - return { status: 405, body: "Method Not Allowed" }; - } +type ResolvedRequest = + | { kind: "ok"; environment: AuthenticatedEnvironment; runId: string } + | { kind: "error"; response: Response }; +// Auth + params, shared by both methods. Secret-key auth only, deliberately: adding +// an RBAC `authorization` gate here would change behaviour for narrowly-scoped JWTs. +async function resolveRequest( + request: Request, + params: ActionFunctionArgs["params"] +): Promise { const authenticationResult = await authenticateApiRequest(request); if (!authenticationResult) { - return json({ error: "Invalid or Missing API Key" }, { status: 401 }); + return { + kind: "error", + response: json({ error: "Invalid or Missing API Key" }, { status: 401 }), + }; } const parsedParams = ParamsSchema.safeParse(params); if (!parsedParams.success) { - return json( - { error: "Invalid request parameters", issues: parsedParams.error.issues }, - { status: 400 } - ); + return { + kind: "error", + response: json( + { error: "Invalid request parameters", issues: parsedParams.error.issues }, + { status: 400 } + ), + }; } + return { + kind: "ok", + environment: authenticationResult.environment, + runId: parsedParams.data.runId, + }; +} + +export async function action({ request, params }: ActionFunctionArgs) { + switch (request.method.toUpperCase()) { + case "POST": + return addRunTags(request, params); + case "DELETE": + return removeRunTags(request, params); + default: + return json({ error: "Method Not Allowed" }, { status: 405 }); + } +} + +async function addRunTags(request: Request, params: ActionFunctionArgs["params"]) { + const resolved = await resolveRequest(request, params); + if (resolved.kind === "error") { + return resolved.response; + } + const { environment: env, runId } = resolved; + try { const anyBody = await request.json(); const body = AddTagsRequestBody.safeParse(anyBody); @@ -62,9 +98,8 @@ export async function action({ request, params }: ActionFunctionArgs) { return json({ message: "No new tags to add" }, { status: 200 }); } - const env = authenticationResult.environment; const outcome = await mutateWithFallback({ - runId: parsedParams.data.runId, + runId, environmentId: env.id, organizationId: env.organizationId, bufferPatch: { type: "append_tags", tags: nonEmptyTags, maxTags: MAX_TAGS_PER_RUN }, @@ -141,3 +176,102 @@ export async function action({ request, params }: ActionFunctionArgs) { return json({ error: "Something went wrong, please try again." }, { status: 500 }); } } + +// Removes tags from THIS run only. Tags aren't a shared entity — they're strings in +// the run's own `runTags` array — so there is nothing org-wide to delete, and other +// runs carrying the same tag are untouched. +async function removeRunTags(request: Request, params: ActionFunctionArgs["params"]) { + const resolved = await resolveRequest(request, params); + if (resolved.kind === "error") { + return resolved.response; + } + const { environment: env, runId } = resolved; + + try { + const anyBody = await request.json(); + const body = RemoveTagsRequestBody.safeParse(anyBody); + if (!body.success) { + return json({ error: "Invalid request body", issues: body.error.issues }, { status: 400 }); + } + const bodyTags = typeof body.data.tags === "string" ? [body.data.tags] : body.data.tags; + const nonEmptyTags = bodyTags.filter((t) => t.trim().length > 0); + + // Nothing asked for. No MAX_TAGS_PER_RUN check applies to a removal — it can + // only ever shrink the list. + if (nonEmptyTags.length === 0) { + return json({ message: "No tags to remove" }, { status: 200 }); + } + + const outcome = await mutateWithFallback({ + runId, + environmentId: env.id, + organizationId: env.organizationId, + bufferPatch: { type: "remove_tags", tags: nonEmptyTags }, + pgMutation: async (taskRun) => { + const existing = taskRun.runTags ?? []; + const doomed = new Set(nonEmptyTags); + const remaining = existing.filter((t) => !doomed.has(t)); + const removedCount = existing.length - remaining.length; + + // Removing tags the run doesn't have is an idempotent success, not a 404. + // Skip the write entirely so we don't bump `updatedAt`, replicate a row that + // didn't change, or publish a no-op realtime record. + if (removedCount === 0) { + return json({ message: "Successfully removed 0 tags." }, { status: 200 }); + } + + const updated = await runStore.removeTags( + taskRun.id, + nonEmptyTags, + { runtimeEnvironmentId: env.id }, + prisma + ); + + // The run vanished (or moved environment) between the read and this write. + if (!updated) { + return json({ error: "Run not found" }, { status: 404 }); + } + + // Publish a run-changed record with the REMAINING tag set so tag feeds + // reindex (no-op unless enabled). updatedAt is the read-your-writes + // watermark. Note this record no longer carries the removed tag, so a feed + // subscribed to that tag simply stops receiving this run — there is no + // "tag removed" un-subscribe signal. + publishChangeRecord({ + runId: taskRun.id, + envId: env.id, + tags: remaining, + batchId: taskRun.batchId, + updatedAtMs: updated.updatedAt.getTime(), + }); + + return json({ message: `Successfully removed ${removedCount} tags.` }, { status: 200 }); + }, + // Buffer-applied patch path. The Lua removed the tags from the snapshot + // atomically. Count off the pre-mutation entry (already fetched by + // mutateWithFallback's env-auth pre-check, so no extra Redis read) so the + // message reports the same number the PG path would for the same input. + synthesisedResponse: ({ bufferEntry }) => { + const existing = parseSnapshotTags(bufferEntry); + const removedCount = existing + ? existing.filter((t) => nonEmptyTags.includes(t)).length + : nonEmptyTags.length; + return json({ message: `Successfully removed ${removedCount} tags.` }, { status: 200 }); + }, + // No `rejectedResponse`: a `remove_tags` patch carries no cap, so the buffer + // never reports `limit_exceeded` for it. + abortSignal: getRequestAbortSignal(), + }); + + if (outcome.kind === "not_found") { + return json({ error: "Run not found" }, { status: 404 }); + } + if (outcome.kind === "timed_out") { + return json({ error: "Run materialisation timed out" }, { status: 503 }); + } + return outcome.response; + } catch (error) { + logger.error("Failed to remove run tags", { error }); + return json({ error: "Something went wrong, please try again." }, { status: 500 }); + } +} diff --git a/apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts b/apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts index a7e0520df4f..ae3cc0ad1f0 100644 --- a/apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts +++ b/apps/webapp/test/performTaskRunAlertsStoreRouting.test.ts @@ -140,6 +140,9 @@ class RoutingRunStore implements RunStore { pushTags(runId: string, ...a: any[]): any { return (this.#resolveById(runId).pushTags as any)(runId, ...a); } + removeTags(runId: string, ...a: any[]): any { + return (this.#resolveById(runId).removeTags as any)(runId, ...a); + } pushRealtimeStream(runId: string, ...a: any[]): any { return (this.#resolveById(runId).pushRealtimeStream as any)(runId, ...a); } diff --git a/apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts b/apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts index 764ff6f1ae4..019eeb7bdca 100644 --- a/apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts +++ b/apps/webapp/test/updateMetadataStoreRoutingHetero.test.ts @@ -211,6 +211,9 @@ class RoutingRunStore implements RunStore { pushTags(runId: string, tags: string[], where: any, _tx?: unknown): any { return this.#resolveById(runId).pushTags(runId, tags, where); } + removeTags(runId: string, tags: string[], where: any, _tx?: unknown): any { + return this.#resolveById(runId).removeTags(runId, tags, where); + } pushRealtimeStream(runId: string, streamId: string, _tx?: unknown): any { return this.#resolveById(runId).pushRealtimeStream(runId, streamId); } diff --git a/docs/docs.json b/docs/docs.json index 7a3cfa46722..6a7d1ee8af8 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -343,6 +343,7 @@ "management/runs/reschedule", "management/runs/update-metadata", "management/runs/add-tags", + "management/runs/remove-tags", "management/runs/retrieve-events", "management/runs/retrieve-trace", "management/runs/retrieve-result" diff --git a/docs/management/runs/remove-tags.mdx b/docs/management/runs/remove-tags.mdx new file mode 100644 index 00000000000..13b3e8cfc0d --- /dev/null +++ b/docs/management/runs/remove-tags.mdx @@ -0,0 +1,4 @@ +--- +title: "Remove tags from a run" +openapi: "v3-openapi DELETE /api/v1/runs/{runId}/tags" +--- diff --git a/docs/tags.mdx b/docs/tags.mdx index 33029929bb1..a4280bcf39d 100644 --- a/docs/tags.mdx +++ b/docs/tags.mdx @@ -18,11 +18,13 @@ We don't enforce prefixes but if you use them you'll find it easier to filter an ## How to add tags -There are two ways to add tags to a run: +You can add tags to a run in two ways: 1. When triggering the run. 2. Inside the `run` function, using `tags.add()`. +You can also [remove tags](#removing-tags) from a run while it's running. + ### 1. Adding tags when triggering the run You can add tags when triggering a run using the `tags` option. All the different [trigger](/triggering) methods support this. @@ -96,6 +98,36 @@ export const myTask = task({ }); ``` +## Removing tags + +Use the `tags.delete()` function to remove tags from inside the `run` function. It takes a single string or an array of strings, just like `tags.add()`: + +```ts +import { tags, task } from "@trigger.dev/sdk"; + +export const myTask = task({ + id: "my-task", + run: async (payload: { message: string }) => { + await tags.add(["status_processing", "user_123456"]); + + // ...do the work... + + // Remove a single tag, or an array of tags + await tags.delete("status_processing"); + await tags.add("status_done"); + }, +}); +``` + +This only affects the run you call it from. Other runs keep the same tag, and the tag stays available to filter by in the dashboard. + + + Removing a tag the run doesn't have does nothing and won't throw, so you don't need to + check the run's current tags first. + + +Tags can only be removed from inside the `run` function — there's no equivalent option when triggering. + ## Filtering runs by tags You can filter runs by tags in the dashboard and in the SDK. diff --git a/docs/v3-openapi.yaml b/docs/v3-openapi.yaml index bb42f7af2ca..09a5339dde9 100644 --- a/docs/v3-openapi.yaml +++ b/docs/v3-openapi.yaml @@ -427,9 +427,10 @@ paths: - lang: typescript label: SDK source: |- - import { runs } from "@trigger.dev/sdk"; + import { tags } from "@trigger.dev/sdk"; - await runs.addTags("run_1234", ["tag-1", "tag-2"]); + // Called from inside a run — applies to the current run + await tags.add(["tag-1", "tag-2"]); - lang: typescript label: Fetch source: |- @@ -441,6 +442,86 @@ paths: }, body: JSON.stringify({ tags: ["tag-1", "tag-2"] }), }); + delete: + operationId: remove_run_tags_v1 + summary: Remove tags from a run + description: Removes one or more tags from a run. Only this run is affected — the tag is left on any other run that has it. Tags the run doesn't have are ignored, so the request is idempotent. + requestBody: + required: true + content: + application/json: + schema: + type: object + required: + - tags + properties: + tags: + $ref: "#/components/schemas/RunTags" + responses: + "200": + description: Successful request + content: + application/json: + schema: + type: object + properties: + message: + type: string + example: "Successfully removed 2 tags." + "400": + description: Invalid request + content: + application/json: + schema: + type: object + properties: + error: + type: string + "401": + description: Unauthorized request + content: + application/json: + schema: + type: object + properties: + error: + type: string + enum: + - Invalid or Missing API Key + "404": + description: Run not found + content: + application/json: + schema: + type: object + properties: + error: + type: string + enum: + - Run not found + tags: + - runs + security: + - secretKey: [] + x-codeSamples: + - lang: typescript + label: SDK + source: |- + import { tags } from "@trigger.dev/sdk"; + + // Called from inside a run — applies to the current run + await tags.delete(["tag-1", "tag-2"]); + - lang: typescript + label: Fetch + source: |- + await fetch("https://api.trigger.dev/api/v1/runs/run_1234/tags", { + method: "DELETE", + headers: { + "Authorization": `Bearer ${process.env.TRIGGER_SECRET_KEY}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ tags: ["tag-1", "tag-2"] }), + }); "/api/v1/runs/{runId}/trace": parameters: diff --git a/internal-packages/run-store/src/PostgresRunStore.test.ts b/internal-packages/run-store/src/PostgresRunStore.test.ts index b0423f24bdb..c15e27a3a36 100644 --- a/internal-packages/run-store/src/PostgresRunStore.test.ts +++ b/internal-packages/run-store/src/PostgresRunStore.test.ts @@ -1686,6 +1686,252 @@ describe("PostgresRunStore — delayed / debounce / metadata / idempotency / arr } ); + // --------------------------------------------------------------------------- + // removeTags + // --------------------------------------------------------------------------- + + postgresTest( + "removeTags removes a single tag (seed [a,b,c], remove [b] → [a,c])", + async ({ prisma }) => { + const { organization, project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const runId = "run_remove_tags_1"; + + await seedRun(prisma, { + runId, + friendlyId: "run_remove_tags_friendly_1", + organizationId: organization.id, + projectId: project.id, + runtimeEnvironmentId: environment.id, + runTags: ["a", "b", "c"], + }); + + const result = await store.removeTags(runId, ["b"], { + runtimeEnvironmentId: environment.id, + }); + + expect(result?.updatedAt).toBeInstanceOf(Date); + + const row = await prisma.taskRun.findFirst({ + where: { id: runId }, + select: { runTags: true }, + }); + expect(row?.runTags).toEqual(["a", "c"]); + } + ); + + postgresTest( + "removeTags removes several tags at once (seed [a,b,c,d], remove [b,d] → [a,c])", + async ({ prisma }) => { + const { organization, project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const runId = "run_remove_tags_2"; + + await seedRun(prisma, { + runId, + friendlyId: "run_remove_tags_friendly_2", + organizationId: organization.id, + projectId: project.id, + runtimeEnvironmentId: environment.id, + runTags: ["a", "b", "c", "d"], + }); + + await store.removeTags(runId, ["b", "d"], { runtimeEnvironmentId: environment.id }); + + const row = await prisma.taskRun.findFirst({ + where: { id: runId }, + select: { runTags: true }, + }); + expect(row?.runTags).toEqual(["a", "c"]); + } + ); + + postgresTest( + "removeTags is a no-op success when the tag isn't on the run", + async ({ prisma }) => { + const { organization, project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const runId = "run_remove_tags_3"; + + await seedRun(prisma, { + runId, + friendlyId: "run_remove_tags_friendly_3", + organizationId: organization.id, + projectId: project.id, + runtimeEnvironmentId: environment.id, + runTags: ["a", "b"], + }); + + const result = await store.removeTags(runId, ["nope"], { + runtimeEnvironmentId: environment.id, + }); + + // Matched the run, so we get an updatedAt back — just nothing changed. + expect(result?.updatedAt).toBeInstanceOf(Date); + + const row = await prisma.taskRun.findFirst({ + where: { id: runId }, + select: { runTags: true }, + }); + expect(row?.runTags).toEqual(["a", "b"]); + } + ); + + postgresTest("removeTags removing every tag leaves an empty array", async ({ prisma }) => { + const { organization, project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const runId = "run_remove_tags_4"; + + await seedRun(prisma, { + runId, + friendlyId: "run_remove_tags_friendly_4", + organizationId: organization.id, + projectId: project.id, + runtimeEnvironmentId: environment.id, + runTags: ["a", "b"], + }); + + await store.removeTags(runId, ["a", "b"], { runtimeEnvironmentId: environment.id }); + + const row = await prisma.taskRun.findFirst({ + where: { id: runId }, + select: { runTags: true }, + }); + expect(row?.runTags).toEqual([]); + }); + + postgresTest("removeTags preserves the order of the surviving tags", async ({ prisma }) => { + const { organization, project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const runId = "run_remove_tags_5"; + + await seedRun(prisma, { + runId, + friendlyId: "run_remove_tags_friendly_5", + organizationId: organization.id, + projectId: project.id, + runtimeEnvironmentId: environment.id, + runTags: ["z", "a", "m", "b", "c"], + }); + + await store.removeTags(runId, ["a", "b"], { runtimeEnvironmentId: environment.id }); + + const row = await prisma.taskRun.findFirst({ + where: { id: runId }, + select: { runTags: true }, + }); + // Insertion order, NOT sorted — an EXCEPT-based rewrite would reorder these. + expect(row?.runTags).toEqual(["z", "m", "c"]); + }); + + postgresTest( + "removeTags preserves duplicates among the survivors (no dedupe)", + async ({ prisma }) => { + const { organization, project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const runId = "run_remove_tags_6"; + + await seedRun(prisma, { + runId, + friendlyId: "run_remove_tags_friendly_6", + organizationId: organization.id, + projectId: project.id, + runtimeEnvironmentId: environment.id, + runTags: ["a", "b", "a"], + }); + + await store.removeTags(runId, ["b"], { runtimeEnvironmentId: environment.id }); + + const row = await prisma.taskRun.findFirst({ + where: { id: runId }, + select: { runTags: true }, + }); + // An EXCEPT-based rewrite would collapse this to ["a"]. + expect(row?.runTags).toEqual(["a", "a"]); + } + ); + + postgresTest( + "removeTags does not touch a run in a different runtimeEnvironmentId", + async ({ prisma }) => { + const { organization, project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const runId = "run_remove_tags_7"; + + const otherEnvironment = await prisma.runtimeEnvironment.create({ + data: { + type: "PREVIEW", + slug: "other-env", + projectId: project.id, + organizationId: organization.id, + apiKey: "tr_other_apikey", + pkApiKey: "pk_other_apikey", + shortcode: "other_short_code", + }, + }); + + await seedRun(prisma, { + runId, + friendlyId: "run_remove_tags_friendly_7", + organizationId: organization.id, + projectId: project.id, + runtimeEnvironmentId: environment.id, + runTags: ["a", "b"], + }); + + const result = await store.removeTags(runId, ["a"], { + runtimeEnvironmentId: otherEnvironment.id, + }); + + // No row matched, so the caller can 404 rather than report a phantom success. + expect(result).toBeNull(); + + const row = await prisma.taskRun.findFirst({ + where: { id: runId }, + select: { runTags: true }, + }); + expect(row?.runTags).toEqual(["a", "b"]); + } + ); + + postgresTest( + "removeTags returns a fresh updatedAt that matches what was stored", + async ({ prisma }) => { + const { organization, project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const runId = "run_remove_tags_8"; + + await seedRun(prisma, { + runId, + friendlyId: "run_remove_tags_friendly_8", + organizationId: organization.id, + projectId: project.id, + runtimeEnvironmentId: environment.id, + runTags: ["a", "b"], + }); + + // Force updatedAt well into the past so the freshness assertion below can't + // flake on two statements landing in the same millisecond. + const stale = new Date("2020-01-01T00:00:00.000Z"); + await prisma.$executeRaw`UPDATE "TaskRun" SET "updatedAt" = ${stale} WHERE "id" = ${runId}`; + + const result = await store.removeTags(runId, ["a"], { + runtimeEnvironmentId: environment.id, + }); + + const row = await prisma.taskRun.findFirst({ + where: { id: runId }, + select: { updatedAt: true }, + }); + + // Raw SQL doesn't fire Prisma's @updatedAt, so the statement sets it explicitly. + expect(result?.updatedAt.getTime()).toBeGreaterThan(stale.getTime()); + // The returned value is the read-your-writes watermark, so it must be exactly + // what landed on the row. + expect(result?.updatedAt.getTime()).toBe(row?.updatedAt.getTime()); + } + ); + // --------------------------------------------------------------------------- // pushRealtimeStream // --------------------------------------------------------------------------- diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index 1b0d2f89cee..df330087742 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -1464,6 +1464,60 @@ export class PostgresRunStore implements RunStore { }); } + /** + * Removes `tags` from a run's `runTags`. Deliberately a SINGLE statement: Prisma has no + * array-pull operator, and a read-modify-write `{ set }` would open a window where a + * concurrent `pushTags` is lost. Rebuilding the array with `unnest` preserves the order + * (and any duplicates) of the surviving tags — `EXCEPT` would dedupe and reorder them. + * + * Raw SQL does not fire Prisma's `@updatedAt`, so `"updatedAt"` is set explicitly and + * returned: realtime uses it as the read-your-writes watermark. + * + * Resolves to `null` when no run matched the id + environment. + */ + async removeTags( + runId: string, + tags: string[], + where: { runtimeEnvironmentId: string }, + tx?: PrismaClientOrTransaction + ): Promise<{ updatedAt: Date } | null> { + const prisma = tx ?? this.prisma; + + // Nothing to remove. Don't touch the row: bumping "updatedAt" would emit a pointless + // replication event, and Prisma.join would build an invalid `IN ()` clause below. + if (tags.length === 0) { + return prisma.taskRun.findFirst({ + where: { id: runId, runtimeEnvironmentId: where.runtimeEnvironmentId }, + select: { updatedAt: true }, + }); + } + + // Dedicated: the run-ops generated client binds a bare value array ambiguously (jsonb), so we + // pass the tag list as a single `text[]` param and match with `= ANY`, mirroring expireRunsBatch. + const rows = + this.schemaVariant === "dedicated" + ? await prisma.$queryRaw<{ updatedAt: Date }[]>` + UPDATE "TaskRun" + SET "runTags" = ARRAY( + SELECT t FROM unnest("runTags") AS t WHERE NOT (t = ANY(${tags}::text[])) + ), + "updatedAt" = NOW() + WHERE "id" = ${runId} AND "runtimeEnvironmentId" = ${where.runtimeEnvironmentId} + RETURNING "updatedAt" + ` + : await prisma.$queryRaw<{ updatedAt: Date }[]>` + UPDATE "TaskRun" + SET "runTags" = ARRAY( + SELECT t FROM unnest("runTags") AS t WHERE t NOT IN (${Prisma.join(tags)}) + ), + "updatedAt" = NOW() + WHERE "id" = ${runId} AND "runtimeEnvironmentId" = ${where.runtimeEnvironmentId} + RETURNING "updatedAt" + `; + + return rows[0] ?? null; + } + async pushRealtimeStream( runId: string, streamId: string, diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 467df81df26..8ceb3a7d949 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -707,6 +707,15 @@ export class RoutingRunStore implements RunStore { return (await this.#routeForWrite(runId)).pushTags(runId, tags, where); } + async removeTags( + runId: string, + tags: string[], + where: { runtimeEnvironmentId: string }, + tx?: PrismaClientOrTransaction + ): Promise<{ updatedAt: Date } | null> { + return (await this.#routeForWrite(runId)).removeTags(runId, tags, where); + } + async pushRealtimeStream( runId: string, streamId: string, diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts index 374a550e419..608cfc3476c 100644 --- a/internal-packages/run-store/src/types.ts +++ b/internal-packages/run-store/src/types.ts @@ -537,6 +537,16 @@ export interface RunStore { where: { runtimeEnvironmentId: string }, tx?: PrismaClientOrTransaction ): Promise<{ updatedAt: Date }>; + /** + * Removes `tags` from a run's `runTags`. Resolves to `null` when no run matched the + * id + environment, so a caller can 404 rather than report a phantom success. + */ + removeTags( + runId: string, + tags: string[], + where: { runtimeEnvironmentId: string }, + tx?: PrismaClientOrTransaction + ): Promise<{ updatedAt: Date } | null>; pushRealtimeStream( runId: string, streamId: string, diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts index 23f13f37573..e04d6b82663 100644 --- a/packages/core/src/v3/apiClient/index.ts +++ b/packages/core/src/v3/apiClient/index.ts @@ -26,6 +26,7 @@ import { type PromotePromptVersionRequestBody, type QueueTypeName, type ReactivatePromptOverrideRequestBody, + type RemoveTagsRequestBody, type RescheduleRunRequestBody, type ResolvePromptRequestBody, type RetrieveQueueParam, @@ -901,6 +902,19 @@ export class ApiClient { ); } + removeTags(runId: string, body: RemoveTagsRequestBody, requestOptions?: ZodFetchOptions) { + return zodfetch( + z.object({ message: z.string() }), + `${this.baseUrl}/api/v1/runs/${runId}/tags`, + { + method: "DELETE", + headers: this.#getHeaders(false), + body: JSON.stringify(body), + }, + mergeRequestOptions(this.defaultRequestOptions, requestOptions) + ); + } + createSchedule(options: CreateScheduleOptions, requestOptions?: ZodFetchOptions) { return zodfetch( ScheduleObject, diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index 3b57395b297..6a7723c4c88 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -535,6 +535,12 @@ export const AddTagsRequestBody = z.object({ export type AddTagsRequestBody = z.infer; +export const RemoveTagsRequestBody = z.object({ + tags: RunTags, +}); + +export type RemoveTagsRequestBody = z.infer; + export const RescheduleRunRequestBody = z.object({ delay: z.string().or(z.coerce.date()), }); diff --git a/packages/redis-worker/src/mollifier/buffer.test.ts b/packages/redis-worker/src/mollifier/buffer.test.ts index 3b34e32fe73..e0f0d9d239b 100644 --- a/packages/redis-worker/src/mollifier/buffer.test.ts +++ b/packages/redis-worker/src/mollifier/buffer.test.ts @@ -2048,6 +2048,222 @@ describe("MollifierBuffer.mutateSnapshot", () => { } ); + redisTest( + "remove_tags removes the named tags and preserves the order of the survivors", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + logger: new Logger("test", "log"), + }); + try { + await buffer.accept({ + runId: "r_rm1", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ tags: ["z", "a", "m", "b", "c"] }), + }); + + const result = await buffer.mutateSnapshot("r_rm1", { + type: "remove_tags", + tags: ["a", "b"], + }); + expect(result).toBe("applied_to_snapshot"); + + const entry = await buffer.getEntry("r_rm1"); + const payload = JSON.parse(entry!.payload) as { tags: string[] }; + expect(payload.tags).toEqual(["z", "m", "c"]); + } finally { + await buffer.close(); + } + } + ); + + redisTest( + "remove_tags is an applied no-op for tags the snapshot doesn't carry", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + logger: new Logger("test", "log"), + }); + try { + await buffer.accept({ + runId: "r_rm2", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ tags: ["a", "b"] }), + }); + + // Idempotent: unknown tags don't reject, they just change nothing. + const result = await buffer.mutateSnapshot("r_rm2", { + type: "remove_tags", + tags: ["nope"], + }); + expect(result).toBe("applied_to_snapshot"); + + const entry = await buffer.getEntry("r_rm2"); + const payload = JSON.parse(entry!.payload) as { tags: string[] }; + expect(payload.tags).toEqual(["a", "b"]); + } finally { + await buffer.close(); + } + } + ); + + redisTest( + "remove_tags on a snapshot with no tags field is an applied no-op", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + logger: new Logger("test", "log"), + }); + try { + await buffer.accept({ + runId: "r_rm3", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ taskId: "t" }), + }); + + const result = await buffer.mutateSnapshot("r_rm3", { + type: "remove_tags", + tags: ["a"], + }); + expect(result).toBe("applied_to_snapshot"); + + const entry = await buffer.getEntry("r_rm3"); + const payload = JSON.parse(entry!.payload) as { taskId: string; tags?: unknown }; + expect(payload.taskId).toBe("t"); + expect(payload.tags).toBeUndefined(); + } finally { + await buffer.close(); + } + } + ); + + redisTest( + "remove_tags removing the LAST tag never writes a JSON object for tags", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + logger: new Logger("test", "log"), + }); + try { + await buffer.accept({ + runId: "r_rm4", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ taskId: "t", tags: ["only"] }), + }); + + const result = await buffer.mutateSnapshot("r_rm4", { + type: "remove_tags", + tags: ["only"], + }); + expect(result).toBe("applied_to_snapshot"); + + const entry = await buffer.getEntry("r_rm4"); + + // The hazard this guards: cjson encodes an EMPTY Lua table as `{}` (a JSON + // object), not `[]`. Readers do `Array.isArray(tags)` / spread the value, so a + // `{"tags":{}}` payload would either silently drop the field or throw. The Lua + // drops the field entirely instead. + expect(entry!.payload).not.toContain('"tags":{}'); + + const payload = JSON.parse(entry!.payload) as { taskId: string; tags?: unknown }; + expect(payload.taskId).toBe("t"); + // Either absent or a real array — never an object. + expect(payload.tags === undefined || Array.isArray(payload.tags)).toBe(true); + // And in both shapes a reader normalises it to the empty list. + expect(payload.tags ?? []).toEqual([]); + } finally { + await buffer.close(); + } + } + ); + + redisTest( + "remove_tags then append_tags rebuilds a dense array after the list was emptied", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + logger: new Logger("test", "log"), + }); + try { + await buffer.accept({ + runId: "r_rm5", + envId: "env_m", + orgId: "org_1", + payload: serialiseSnapshot({ tags: ["a", "b"] }), + }); + + expect( + await buffer.mutateSnapshot("r_rm5", { type: "remove_tags", tags: ["a", "b"] }) + ).toBe("applied_to_snapshot"); + + expect(await buffer.mutateSnapshot("r_rm5", { type: "append_tags", tags: ["c"] })).toBe( + "applied_to_snapshot" + ); + + const entry = await buffer.getEntry("r_rm5"); + const payload = JSON.parse(entry!.payload) as { tags: string[] }; + expect(payload.tags).toEqual(["c"]); + } finally { + await buffer.close(); + } + } + ); + + redisTest( + "remove_tags returns not_found when no entry exists for the runId", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + logger: new Logger("test", "log"), + }); + try { + const result = await buffer.mutateSnapshot("nope_rm", { + type: "remove_tags", + tags: ["x"], + }); + // Critically NOT 'busy' — an unhandled patch type falls into the terminal + // else branch, which the API turns into a 2s stall then a 503. + expect(result).toBe("not_found"); + } finally { + await buffer.close(); + } + } + ); + redisTest( "set_metadata replaces metadata + metadataType (last-write-wins)", { timeout: 20_000 }, diff --git a/packages/redis-worker/src/mollifier/buffer.ts b/packages/redis-worker/src/mollifier/buffer.ts index 179de94eca3..4aa17d7805c 100644 --- a/packages/redis-worker/src/mollifier/buffer.ts +++ b/packages/redis-worker/src/mollifier/buffer.ts @@ -70,6 +70,10 @@ export type SnapshotPatch = // PG-path MAX_TAGS_PER_RUN check so a buffered run can't accumulate more // tags than the trigger validator would have allowed at creation. | { type: "append_tags"; tags: string[]; maxTags?: number } + // Removes `tags` from the snapshot, per-run only. No cap applies — a removal can + // never push a run over MAX_TAGS_PER_RUN. Removing a tag the snapshot doesn't + // carry is an applied no-op, so the API stays idempotent. + | { type: "remove_tags"; tags: string[] } | { type: "set_metadata"; metadata: string; metadataType: string } | { type: "set_delay"; delayUntil: string } | { type: "mark_cancelled"; cancelledAt: string; cancelReason?: string }; @@ -1023,6 +1027,34 @@ export class MollifierBuffer { return 'limit_exceeded' end payload.tags = merged + elseif patch.type == 'remove_tags' then + -- Per-run removal only: a tag is just a string in this run's list, never a + -- shared entity, so dropping it here cannot affect any other run. Order and + -- any duplicates among the surviving tags are preserved. Removing a tag the + -- snapshot doesn't carry is an applied no-op, keeping the API idempotent. + local existing = payload.tags or {} + local doomed = {} + for _, t in ipairs(patch.tags or {}) do + doomed[t] = true + end + local kept = {} + for _, t in ipairs(existing) do + if not doomed[t] then + table.insert(kept, t) + end + end + if #kept == 0 then + -- cjson encodes an EMPTY Lua table as '{}' (a JSON *object*), not '[]', and + -- Redis's bundled cjson has no empty-array sentinel to force the array form. + -- Removing a run's last tag hits this case constantly, and a '{}' here would + -- put a JSON object where every reader expects an array. So drop the field + -- instead: an ABSENT 'tags' is already the shape a run triggered without tags + -- has, and every reader normalises it to an empty list via the same + -- Array.isArray guards that exist for the '{}' hazard. + payload.tags = nil + else + payload.tags = kept + end elseif patch.type == 'set_metadata' then payload.metadata = patch.metadata payload.metadataType = patch.metadataType diff --git a/packages/trigger-sdk/src/v3/tags.ts b/packages/trigger-sdk/src/v3/tags.ts index f8f6b6feb74..3b7e8ebe557 100644 --- a/packages/trigger-sdk/src/v3/tags.ts +++ b/packages/trigger-sdk/src/v3/tags.ts @@ -9,10 +9,39 @@ import { } from "@trigger.dev/core/v3"; import { tracer } from "./tracer.js"; +/** + * Provides access to the tags of the current run. + * @namespace + * @property {Function} add - Add one or more tags to the current run. + * @property {Function} delete - Remove one or more tags from the current run. + */ export const tags = { add: addTags, + delete: deleteTags, }; +/** + * Add one or more tags to the current run. Existing tags are kept, and tags the run + * already has are ignored. + * + * A run can have at most 10 tags. If the call would take the run over that limit an + * error is logged and the new tags are not added. + * + * @param {RunTags} tags - A single tag, or an array of tags, to add to the run. + * @param {ApiRequestOptions} [requestOptions] - Optional request options. + * @returns {Promise} Resolves once the tags have been added. + * + * @example + * import { tags, task } from "@trigger.dev/sdk"; + * + * export const myTask = task({ + * id: "my-task", + * run: async (payload: { userId: string }) => { + * await tags.add(`user_${payload.userId}`); + * await tags.add(["product_1234567", "org_abcdefg"]); + * }, + * }); + */ async function addTags(tags: RunTags, requestOptions?: ApiRequestOptions) { const apiClient = apiClientManager.clientOrThrow(); @@ -26,7 +55,7 @@ async function addTags(tags: RunTags, requestOptions?: ApiRequestOptions) { const $requestOptions = mergeRequestOptions( { tracer, - name: "tags.set()", + name: "tags.add()", icon: "tag", attributes: { ...accessoryAttributes({ @@ -59,3 +88,64 @@ async function addTags(tags: RunTags, requestOptions?: ApiRequestOptions) { throw error; } } + +/** + * Remove one or more tags from the current run. Only this run is affected — the tag + * remains on any other run that has it, and stays available for filtering. + * + * Removing a tag the run doesn't have is a no-op, so it's safe to call this without + * checking the run's current tags first. + * + * @param {RunTags} tags - A single tag, or an array of tags, to remove from the run. + * @param {ApiRequestOptions} [requestOptions] - Optional request options. + * @returns {Promise} Resolves once the tags have been removed. + * + * @example + * import { tags, task } from "@trigger.dev/sdk"; + * + * export const myTask = task({ + * id: "my-task", + * run: async () => { + * await tags.add("status_processing"); + * // ... + * await tags.delete("status_processing"); + * await tags.add("status_done"); + * }, + * }); + */ +async function deleteTags(tags: RunTags, requestOptions?: ApiRequestOptions): Promise { + const apiClient = apiClientManager.clientOrThrow(); + + const run = taskContext.ctx?.run; + if (!run) { + throw new Error("Can't delete tags outside of a run."); + } + + const $requestOptions = mergeRequestOptions( + { + tracer, + name: "tags.delete()", + icon: "tag", + attributes: { + ...accessoryAttributes({ + items: [ + { + text: typeof tags === "string" ? tags : tags.join(", "), + variant: "normal", + }, + ], + style: "codepath", + }), + }, + }, + requestOptions + ); + + try { + await apiClient.removeTags(run.id, { tags }, $requestOptions); + } catch (error) { + logger.error("Failed to delete tags", { error }); + + throw error; + } +} From b05a7e72401dc629db4641456077e111d0a591a9 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 28 Aug 2026 10:55:12 +0000 Subject: [PATCH 2/6] fix(webapp,run-store): confirm removals against the primary and bind updatedAt off the app clock - Removal now re-reads the run from the primary when the replica shows none of the requested tags, so an add-then-delete in the same run is not silently skipped. - A DELETE with an unparseable body answers 400 rather than falling through to 500. - The buffered-run synthesised response no longer reports removals that did not happen when the snapshot carries no tags array. - removeTags binds "updatedAt" from the app clock instead of NOW(), matching expireRunsBatch, so the read-your-writes watermark is comparable to Date.now(). Co-Authored-By: Claude --- .../app/routes/api.v1.runs.$runId.tags.ts | 55 ++++++++++++++++--- .../run-store/src/PostgresRunStore.test.ts | 8 +++ .../run-store/src/PostgresRunStore.ts | 12 +++- 3 files changed, 65 insertions(+), 10 deletions(-) diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts index 5c468ba5db9..010cf0ffb33 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts @@ -188,7 +188,16 @@ async function removeRunTags(request: Request, params: ActionFunctionArgs["param const { environment: env, runId } = resolved; try { - const anyBody = await request.json(); + // A DELETE with no body at all is the natural thing to try by hand, and + // `request.json()` throws on it. That's a malformed request, not a server fault, so + // map it to 400 instead of letting it fall through to the catch-all 500 + error log. + let anyBody: unknown; + try { + anyBody = await request.json(); + } catch { + return json({ error: "Invalid request body" }, { status: 400 }); + } + const body = RemoveTagsRequestBody.safeParse(anyBody); if (!body.success) { return json({ error: "Invalid request body", issues: body.error.issues }, { status: 400 }); @@ -208,9 +217,38 @@ async function removeRunTags(request: Request, params: ActionFunctionArgs["param organizationId: env.organizationId, bufferPatch: { type: "remove_tags", tags: nonEmptyTags }, pgMutation: async (taskRun) => { - const existing = taskRun.runTags ?? []; const doomed = new Set(nonEmptyTags); - const remaining = existing.filter((t) => !doomed.has(t)); + let existing = taskRun.runTags ?? []; + let remaining = existing.filter((t) => !doomed.has(t)); + + if (existing.length === remaining.length) { + // `taskRun` normally comes from the READ REPLICA, so "the run has none of + // these tags" can just be replication lag rather than the truth — and + // `tags.add("x")` immediately followed by `tags.delete("x")` is a completely + // ordinary pattern. Confirm against the primary before concluding there's + // nothing to do, the same disambiguation mutateWithFallback does for a + // replica-lag 404. Passing the writer as the read client makes the routing + // store read the OWNING store's primary, so this is read-your-writes for a + // run on either database. + // + // The add path's equivalent skip needs no such check because a stale read is + // safe in that direction: it can only make the write bigger, never skip a + // real one. Skipping a real removal, by contrast, would silently drop the + // customer's mutation and still answer 200. + const fresh = await runStore.findRun( + { id: taskRun.id, runtimeEnvironmentId: env.id }, + { select: { runTags: true } }, + prisma + ); + + if (!fresh) { + return json({ error: "Run not found" }, { status: 404 }); + } + + existing = fresh.runTags ?? []; + remaining = existing.filter((t) => !doomed.has(t)); + } + const removedCount = existing.length - remaining.length; // Removing tags the run doesn't have is an idempotent success, not a 404. @@ -252,10 +290,13 @@ async function removeRunTags(request: Request, params: ActionFunctionArgs["param // mutateWithFallback's env-auth pre-check, so no extra Redis read) so the // message reports the same number the PG path would for the same input. synthesisedResponse: ({ bufferEntry }) => { - const existing = parseSnapshotTags(bufferEntry); - const removedCount = existing - ? existing.filter((t) => nonEmptyTags.includes(t)).length - : nonEmptyTags.length; + // `null` here means the snapshot carried no readable `tags` array. For a + // buffered run that is simply the shape of a run triggered without tags, so + // there was nothing to remove — unlike the add path, whose unknown-existing + // fallback is the request count, falling back to the request count here would + // report removals that never happened on the most common buffered shape. + const existing = parseSnapshotTags(bufferEntry) ?? []; + const removedCount = existing.filter((t) => nonEmptyTags.includes(t)).length; return json({ message: `Successfully removed ${removedCount} tags.` }, { status: 200 }); }, // No `rejectedResponse`: a `remove_tags` patch carries no cap, so the buffer diff --git a/internal-packages/run-store/src/PostgresRunStore.test.ts b/internal-packages/run-store/src/PostgresRunStore.test.ts index c15e27a3a36..6f767015901 100644 --- a/internal-packages/run-store/src/PostgresRunStore.test.ts +++ b/internal-packages/run-store/src/PostgresRunStore.test.ts @@ -1915,9 +1915,11 @@ describe("PostgresRunStore — delayed / debounce / metadata / idempotency / arr const stale = new Date("2020-01-01T00:00:00.000Z"); await prisma.$executeRaw`UPDATE "TaskRun" SET "updatedAt" = ${stale} WHERE "id" = ${runId}`; + const before = Date.now(); const result = await store.removeTags(runId, ["a"], { runtimeEnvironmentId: environment.id, }); + const after = Date.now(); const row = await prisma.taskRun.findFirst({ where: { id: runId }, @@ -1929,6 +1931,12 @@ describe("PostgresRunStore — delayed / debounce / metadata / idempotency / arr // The returned value is the read-your-writes watermark, so it must be exactly // what landed on the row. expect(result?.updatedAt.getTime()).toBe(row?.updatedAt.getTime()); + // ...and it must come from the APP clock, not the database's. The change router + // compares this watermark against `Date.now()` to estimate replica lag, so a + // `NOW()`-sourced value would drift by the database's clock skew and, because + // `updatedAt` is `timestamp(3)`, by its session `TimeZone` too. + expect(result!.updatedAt.getTime()).toBeGreaterThanOrEqual(before); + expect(result!.updatedAt.getTime()).toBeLessThanOrEqual(after); } ); diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index df330087742..f70a09c5a77 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -1471,7 +1471,12 @@ export class PostgresRunStore implements RunStore { * (and any duplicates) of the surviving tags — `EXCEPT` would dedupe and reorder them. * * Raw SQL does not fire Prisma's `@updatedAt`, so `"updatedAt"` is set explicitly and - * returned: realtime uses it as the read-your-writes watermark. + * returned: realtime uses it as the read-your-writes watermark. It is bound as a + * parameter off the app clock rather than `NOW()` — the watermark is compared against + * `Date.now()` when the change router estimates replica lag, and `NOW()` would source it + * from the database's clock and `TimeZone` instead (`TaskRun.updatedAt` is + * `timestamp(3)`, so `now()` is cast through the session time zone). This mirrors + * `expireRunsBatch`, which binds its timestamp the same way. * * Resolves to `null` when no run matched the id + environment. */ @@ -1482,6 +1487,7 @@ export class PostgresRunStore implements RunStore { tx?: PrismaClientOrTransaction ): Promise<{ updatedAt: Date } | null> { const prisma = tx ?? this.prisma; + const now = new Date(); // Nothing to remove. Don't touch the row: bumping "updatedAt" would emit a pointless // replication event, and Prisma.join would build an invalid `IN ()` clause below. @@ -1501,7 +1507,7 @@ export class PostgresRunStore implements RunStore { SET "runTags" = ARRAY( SELECT t FROM unnest("runTags") AS t WHERE NOT (t = ANY(${tags}::text[])) ), - "updatedAt" = NOW() + "updatedAt" = ${now} WHERE "id" = ${runId} AND "runtimeEnvironmentId" = ${where.runtimeEnvironmentId} RETURNING "updatedAt" ` @@ -1510,7 +1516,7 @@ export class PostgresRunStore implements RunStore { SET "runTags" = ARRAY( SELECT t FROM unnest("runTags") AS t WHERE t NOT IN (${Prisma.join(tags)}) ), - "updatedAt" = NOW() + "updatedAt" = ${now} WHERE "id" = ${runId} AND "runtimeEnvironmentId" = ${where.runtimeEnvironmentId} RETURNING "updatedAt" `; From c227fea4f746a458762b2d0816949c081cfd4b04 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 28 Aug 2026 11:08:31 +0000 Subject: [PATCH 3/6] fix(run-store): teach the new RunStore decorators about removeTags main gained DelegatingRunStore and the TaskRunExecutionSnapshotStore decorator that extends it, plus the RUN_STORE_METHOD_NAMES registry whose compile-time parity assertions cover every interface member. None of them knew about the removeTags method this branch adds to RunStore, so the merged tree did not typecheck. Add the forwarder and the registry entry, both next to pushTags to match the interface order. Co-Authored-By: Claude --- internal-packages/run-store/src/delegatingRunStore.ts | 9 +++++++++ internal-packages/run-store/src/runStoreMethodNames.ts | 1 + 2 files changed, 10 insertions(+) diff --git a/internal-packages/run-store/src/delegatingRunStore.ts b/internal-packages/run-store/src/delegatingRunStore.ts index 6735a742822..554e789f078 100644 --- a/internal-packages/run-store/src/delegatingRunStore.ts +++ b/internal-packages/run-store/src/delegatingRunStore.ts @@ -308,6 +308,15 @@ export class DelegatingRunStore implements RunStore { return this.delegate.pushTags(runId, tags, where, tx); } + removeTags( + runId: string, + tags: string[], + where: { runtimeEnvironmentId: string }, + tx?: PrismaClientOrTransaction + ): Promise<{ updatedAt: Date } | null> { + return this.delegate.removeTags(runId, tags, where, tx); + } + pushRealtimeStream( runId: string, streamId: string, diff --git a/internal-packages/run-store/src/runStoreMethodNames.ts b/internal-packages/run-store/src/runStoreMethodNames.ts index 2b10bd766d0..27e270623d7 100644 --- a/internal-packages/run-store/src/runStoreMethodNames.ts +++ b/internal-packages/run-store/src/runStoreMethodNames.ts @@ -36,6 +36,7 @@ export const RUN_STORE_METHOD_NAMES = [ "updateMetadata", "clearIdempotencyKey", "pushTags", + "removeTags", "pushRealtimeStream", "findRun", "findRunOrThrow", From b0d71062065fe64d332154c37e44356f9ef7e77f Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 28 Aug 2026 11:30:25 +0000 Subject: [PATCH 4/6] fix(run-store): classify removeTags in the placement catalog placement.proof.test.ts parses the RunStore interface out of types.ts and fails until every method is catalogued as a read or a write. removeTags was added to the interface but not the catalog, so the census reported it as uncatalogued and CI's internal shard 3/12 went red. It belongs in ROUTES_BY_GIVEN_RUN_ID next to pushTags: runOpsStore.removeTags routes through #routeForWrite(runId), which is exactly what the catalog's GIVEN_RUN_ID_ROUTE check asserts. Co-Authored-By: Claude --- internal-packages/run-store/src/placementCatalog.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/internal-packages/run-store/src/placementCatalog.ts b/internal-packages/run-store/src/placementCatalog.ts index 4768c389075..bf9cf7000f6 100644 --- a/internal-packages/run-store/src/placementCatalog.ts +++ b/internal-packages/run-store/src/placementCatalog.ts @@ -38,6 +38,7 @@ export const ROUTES_BY_GIVEN_RUN_ID: readonly string[] = [ "enqueueDelayedRun", "rewriteDebouncedRun", "pushTags", + "removeTags", "pushRealtimeStream", ]; From 2ad0a9869c4d5e2f86ed39a2b236b681c7e4d58a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 28 Aug 2026 16:35:57 +0000 Subject: [PATCH 5/6] chore(changeset): drop redis-worker from the tags.delete changeset The changeset is a user-facing release note for `tags.delete()`. The redis-worker change behind it is internal plumbing, and CONTRIBUTING.md and AGENTS.md both name `@trigger.dev/redis-worker` as a package to leave out of changesets, since it is not consumed independently and a version bump means nothing to a user. Listing it also copied the `tags.delete()` note verbatim into the redis-worker changelog. All `@trigger.dev/*` packages are a `fixed` version group in the changeset config, so redis-worker still gets the same version bump either way; only the stray changelog entry goes away. Co-Authored-By: Claude --- .changeset/tags-delete.md | 1 - 1 file changed, 1 deletion(-) diff --git a/.changeset/tags-delete.md b/.changeset/tags-delete.md index 03ea12d3232..77b48b25524 100644 --- a/.changeset/tags-delete.md +++ b/.changeset/tags-delete.md @@ -1,7 +1,6 @@ --- "@trigger.dev/sdk": patch "@trigger.dev/core": patch -"@trigger.dev/redis-worker": patch --- You can now remove tags from a run while it's running with `tags.delete("my-tag")` (or an array of tags). It only affects that run — every other run keeps the tag, and the tag stays available to filter by. From 1c1bae589bd1dd5852ab6717c5c592e5f0c71c1d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 29 Aug 2026 16:50:43 +0000 Subject: [PATCH 6/6] fix(webapp): confirm apparent duplicate tags against the primary before skipping an add `tags.add()` dedups the requested tags against the run row `mutateWithFallback` hands its `pgMutation`, and that row normally comes from the read replica. While tags were add-only a stale replica could only ever make an add larger than necessary, so deduping against it was safe. Now that `tags.delete()` exists the replica can be stale in the other direction too -- still carrying a tag the primary has already dropped -- and `tags.delete("x")` immediately followed by `tags.add("x")` is an ordinary pattern. Deduping against that row swallowed the add and still answered 200, leaving the run untagged. The add path now re-reads the run from the owning store's primary whenever any requested tag looks already-present, mirroring the confirmation the removal path already performs for an apparent no-op, and recomputes the tag set from the fresh row. Only an apparent duplicate pays for the extra primary read; when the tags genuinely are present the fresh read agrees and the write is still skipped. The comment on the removal path claiming the add path needed no such check was the reasoning that left this open, so it is corrected rather than left to mislead the next reader. Co-Authored-By: Claude --- .../app/routes/api.v1.runs.$runId.tags.ts | 40 +++- .../test/tagsRouteReplicaLag.guard.test.ts | 211 ++++++++++++++++++ 2 files changed, 246 insertions(+), 5 deletions(-) create mode 100644 apps/webapp/test/tagsRouteReplicaLag.guard.test.ts diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts index 010cf0ffb33..38d628d1787 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts @@ -104,8 +104,38 @@ async function addRunTags(request: Request, params: ActionFunctionArgs["params"] organizationId: env.organizationId, bufferPatch: { type: "append_tags", tags: nonEmptyTags, maxTags: MAX_TAGS_PER_RUN }, pgMutation: async (taskRun) => { - const existing = taskRun.runTags ?? []; - const newTags = nonEmptyTags.filter((t) => !existing.includes(t)); + let existing = taskRun.runTags ?? []; + let newTags = nonEmptyTags.filter((t) => !existing.includes(t)); + + if (newTags.length < nonEmptyTags.length) { + // At least one requested tag looks like it's already on the run. But `taskRun` + // normally comes from the READ REPLICA, so "the run already has this tag" can + // be replication lag rather than the truth -- and now that `tags.delete()` + // exists the replica can be stale in the direction that LOSES a write: + // `tags.delete("x")` immediately followed by `tags.add("x")` is an ordinary + // pattern, and a replica still carrying the deleted "x" would make us dedup it + // away and answer 200 with the run left untagged. Confirm against the primary + // before skipping any tag, mirroring the removal path's no-op confirmation. + // + // Passing the writer as the read client makes the routing store read the + // OWNING store's primary, so this is read-your-writes for a run on either + // database. Only an apparent duplicate pays for the extra primary read: a + // plain add of genuinely-new tags takes the replica row straight to the write. + // When the tags really are present the fresh read agrees and we still skip. + const fresh = await runStore.findRun( + { id: taskRun.id, runtimeEnvironmentId: env.id }, + { select: { runTags: true } }, + prisma + ); + + // The run vanished (or moved environment) between the replica read and here. + if (!fresh) { + return json({ error: "Run not found" }, { status: 404 }); + } + + existing = fresh.runTags ?? []; + newTags = nonEmptyTags.filter((t) => !existing.includes(t)); + } if (existing.length + newTags.length > MAX_TAGS_PER_RUN) { return json( @@ -231,9 +261,9 @@ async function removeRunTags(request: Request, params: ActionFunctionArgs["param // store read the OWNING store's primary, so this is read-your-writes for a // run on either database. // - // The add path's equivalent skip needs no such check because a stale read is - // safe in that direction: it can only make the write bigger, never skip a - // real one. Skipping a real removal, by contrast, would silently drop the + // The add path performs the mirror-image confirmation, for the same reason: + // once removals exist a stale replica can equally make an add dedup away a + // tag that is no longer on the run. Either direction would silently drop the // customer's mutation and still answer 200. const fresh = await runStore.findRun( { id: taskRun.id, runtimeEnvironmentId: env.id }, diff --git a/apps/webapp/test/tagsRouteReplicaLag.guard.test.ts b/apps/webapp/test/tagsRouteReplicaLag.guard.test.ts new file mode 100644 index 00000000000..27ea58f446a --- /dev/null +++ b/apps/webapp/test/tagsRouteReplicaLag.guard.test.ts @@ -0,0 +1,211 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +// Regression guard for the delete-then-add sequence. `tags.add()` dedups the +// requested tags against the run row handed to it by `mutateWithFallback`, and that +// row normally comes from the READ REPLICA. Before `tags.delete()` existed a stale +// replica could only ever make an add bigger than necessary, so the dedup was safe. +// Now a replica can still be carrying a tag the primary has already dropped, and +// deduping against it would silently swallow the add. The add path therefore +// confirms an apparent duplicate against the primary before skipping the write. + +const { mocks } = vi.hoisted(() => ({ + mocks: { + writer: { __client: "writer" }, + replica: { __client: "replica" }, + environment: { id: "env_tags", organizationId: "org_tags" }, + authenticateApiRequest: vi.fn(), + findRun: vi.fn(), + pushTags: vi.fn(), + removeTags: vi.fn(), + publishChangeRecord: vi.fn(), + mutateWithFallback: vi.fn(), + }, +})); + +vi.mock("~/db.server", () => ({ prisma: mocks.writer, $replica: mocks.replica })); +vi.mock("~/models/taskRunTag.server", () => ({ MAX_TAGS_PER_RUN: 10 })); +vi.mock("~/services/apiAuth.server", () => ({ + authenticateApiRequest: mocks.authenticateApiRequest, +})); +vi.mock("~/services/httpAsyncStorage.server", () => ({ + getRequestAbortSignal: () => undefined, +})); +vi.mock("~/services/logger.server", () => ({ + logger: { warn: vi.fn(), info: vi.fn(), error: vi.fn(), debug: vi.fn() }, +})); +vi.mock("~/services/realtime/runChangeNotifierInstance.server", () => ({ + publishChangeRecord: mocks.publishChangeRecord, +})); +vi.mock("~/v3/mollifier/mutateWithFallback.server", () => ({ + mutateWithFallback: mocks.mutateWithFallback, +})); +vi.mock("~/v3/runStore.server", () => ({ + runStore: { + findRun: mocks.findRun, + pushTags: mocks.pushTags, + removeTags: mocks.removeTags, + }, +})); + +import { action } from "~/routes/api.v1.runs.$runId.tags"; + +const runId = "run_tags_lag"; + +// Drive the route straight down the PG path with a caller-supplied row, standing in +// for what `mutateWithFallback` reads off the replica. +function replicaRowIs(runTags: string[]) { + mocks.mutateWithFallback.mockImplementation(async (input: any) => ({ + kind: "pg", + response: await input.pgMutation({ id: runId, runTags, batchId: null }), + })); +} + +async function callAction(method: "POST" | "DELETE", tags: string | string[]) { + return (await action({ + request: new Request(`https://example.com/api/v1/runs/${runId}/tags`, { + method, + headers: { Authorization: "Bearer tr_dev_tags", "content-type": "application/json" }, + body: JSON.stringify({ tags }), + }), + params: { runId }, + context: {} as never, + })) as Response; +} + +beforeEach(() => { + vi.clearAllMocks(); + mocks.authenticateApiRequest.mockResolvedValue({ environment: mocks.environment }); + mocks.pushTags.mockResolvedValue({ updatedAt: new Date("2026-08-29T12:00:00Z") }); + mocks.removeTags.mockResolvedValue({ updatedAt: new Date("2026-08-29T12:00:00Z") }); + mocks.findRun.mockResolvedValue(null); +}); + +describe("tags POST under replica lag after a delete", () => { + it("re-adds a tag the replica still shows but the primary has already dropped", async () => { + // The awaited DELETE removed "status_processing" on the primary; the replica + // hasn't caught up and still lists it. + replicaRowIs(["status_processing"]); + mocks.findRun.mockResolvedValue({ runTags: [] }); + + const response = await callAction("POST", "status_processing"); + + // The apparent duplicate was confirmed against the OWNING store's primary by + // passing the writer as the read client. + expect(mocks.findRun).toHaveBeenCalledWith( + { id: runId, runtimeEnvironmentId: mocks.environment.id }, + { select: { runTags: true } }, + mocks.writer + ); + // ...and the write actually happened rather than being deduped away. + expect(mocks.pushTags).toHaveBeenCalledWith( + runId, + ["status_processing"], + { runtimeEnvironmentId: mocks.environment.id }, + mocks.writer + ); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ + message: "Successfully set 1 new tags.", + }); + }); + + it("still skips the write when the primary agrees the tag is present", async () => { + replicaRowIs(["status_processing"]); + mocks.findRun.mockResolvedValue({ runTags: ["status_processing"] }); + + const response = await callAction("POST", "status_processing"); + + expect(mocks.findRun).toHaveBeenCalledTimes(1); + expect(mocks.pushTags).not.toHaveBeenCalled(); + expect(mocks.publishChangeRecord).not.toHaveBeenCalled(); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ message: "No new tags to add" }); + }); + + it("recovers the deleted tag when only part of the request looks duplicated", async () => { + replicaRowIs(["a"]); + mocks.findRun.mockResolvedValue({ runTags: [] }); + + const response = await callAction("POST", ["a", "b"]); + + expect(mocks.pushTags).toHaveBeenCalledWith( + runId, + ["a", "b"], + { runtimeEnvironmentId: mocks.environment.id }, + mocks.writer + ); + await expect(response.json()).resolves.toEqual({ + message: "Successfully set 2 new tags.", + }); + }); + + it("publishes the change record with the confirmed tag set", async () => { + replicaRowIs(["a"]); + mocks.findRun.mockResolvedValue({ runTags: ["b"] }); + + await callAction("POST", ["a"]); + + expect(mocks.publishChangeRecord).toHaveBeenCalledWith( + expect.objectContaining({ runId, envId: mocks.environment.id, tags: ["b", "a"] }) + ); + }); + + it("does not touch the primary when no requested tag looks duplicated", async () => { + replicaRowIs(["something_else"]); + + const response = await callAction("POST", "brand_new"); + + expect(mocks.findRun).not.toHaveBeenCalled(); + expect(mocks.pushTags).toHaveBeenCalledWith( + runId, + ["brand_new"], + { runtimeEnvironmentId: mocks.environment.id }, + mocks.writer + ); + expect(response.status).toBe(200); + }); + + it("404s when the confirmation read finds the run gone", async () => { + replicaRowIs(["status_processing"]); + mocks.findRun.mockResolvedValue(null); + + const response = await callAction("POST", "status_processing"); + + expect(mocks.pushTags).not.toHaveBeenCalled(); + expect(response.status).toBe(404); + await expect(response.json()).resolves.toEqual({ error: "Run not found" }); + }); +}); + +describe("tags DELETE keeps its own primary confirmation", () => { + it("removes a tag the replica has not caught up on yet", async () => { + // The replica predates the add, so it shows none of the requested tags. + replicaRowIs([]); + mocks.findRun.mockResolvedValue({ runTags: ["status_processing"] }); + + const response = await callAction("DELETE", "status_processing"); + + expect(mocks.removeTags).toHaveBeenCalledWith( + runId, + ["status_processing"], + { runtimeEnvironmentId: mocks.environment.id }, + mocks.writer + ); + await expect(response.json()).resolves.toEqual({ + message: "Successfully removed 1 tags.", + }); + }); + + it("reports a genuine no-op without writing", async () => { + replicaRowIs([]); + mocks.findRun.mockResolvedValue({ runTags: [] }); + + const response = await callAction("DELETE", "never_had_it"); + + expect(mocks.removeTags).not.toHaveBeenCalled(); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ + message: "Successfully removed 0 tags.", + }); + }); +});