|
1 | 1 | import type { LoaderFunctionArgs } from "@remix-run/server-runtime"; |
2 | 2 | import { json } from "@remix-run/server-runtime"; |
3 | 3 | import { tryCatch } from "@trigger.dev/core/utils"; |
4 | | -import type { RunMetadataChangeOperation } from "@trigger.dev/core/v3/schemas"; |
5 | 4 | import { UpdateMetadataRequestBody } from "@trigger.dev/core/v3"; |
6 | 5 | import { z } from "zod"; |
7 | 6 | import { $replica } from "~/db.server"; |
8 | | -// Aliased to avoid shadowing the local `env: AuthenticatedEnvironment` |
9 | | -// parameter the route handler and `routeOperationsToRun` use. |
| 7 | +// Aliased to avoid shadowing the local `env` parameter in the handler. |
10 | 8 | import { env as appEnv } from "~/env.server"; |
11 | | -import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; |
12 | 9 | import { authenticateApiRequest } from "~/services/apiAuth.server"; |
13 | | -import { logger } from "~/services/logger.server"; |
14 | 10 | import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server"; |
15 | 11 | import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server"; |
16 | 12 | import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server"; |
17 | 13 | import { ServiceValidationError } from "~/v3/services/common.server"; |
18 | 14 | import { applyMetadataMutationToBufferedRun } from "~/v3/mollifier/applyMetadataMutation.server"; |
| 15 | +import { routeOperationsToRun } from "~/v3/mollifier/routeOperationsToRun.server"; |
19 | 16 | import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server"; |
20 | 17 | import { runStore } from "~/v3/runStore.server"; |
21 | 18 |
|
@@ -67,114 +64,6 @@ export async function loader({ request, params }: LoaderFunctionArgs) { |
67 | 64 | return json({ error: "Run not found" }, { status: 404 }); |
68 | 65 | } |
69 | 66 |
|
70 | | -// Route parent/root operations to the existing PG service by directly |
71 | | -// invoking it against the parent/root runId. The service ingests via |
72 | | -// its batching worker, which targets PG by id. If the parent/root is |
73 | | -// itself buffered we recurse through our buffered-mutation helper. |
74 | | -// `_ingestion_only` flag: a synthetic body that has the operations |
75 | | -// promoted to top-level `operations` so the service applies them to |
76 | | -// `targetRunId` directly. |
77 | | -// Exported so the silent-failure logging behaviour can be unit-tested. |
78 | | -// The route handler itself isn't an attractive test target (createActionApiRoute |
79 | | -// wraps it in auth + body parsing + error-handler middleware), but the |
80 | | -// fan-out helper carries the load-bearing logic — including the ops- |
81 | | -// visibility branch this change adds. |
82 | | -async function routeOperationsToRun( |
83 | | - targetRunId: string | undefined, |
84 | | - operations: RunMetadataChangeOperation[] | undefined, |
85 | | - env: AuthenticatedEnvironment |
86 | | -): Promise<void> { |
87 | | - if (!targetRunId || !operations || operations.length === 0) return; |
88 | | - |
89 | | - // Try PG first via the existing service (this is how parent/root |
90 | | - // operations have always landed; preserve that). Accepts the full |
91 | | - // AuthenticatedEnvironment so we don't have to recover the unsafe |
92 | | - // `as unknown` cast that the previous narrowed `{ id, organizationId }` |
93 | | - // signature forced on us. |
94 | | - // |
95 | | - // Two non-success outcomes from `call`: |
96 | | - // * throws — PG threw (e.g. "Cannot update metadata for a completed |
97 | | - // run", or a transient PG outage). |
98 | | - // * resolves with undefined — PG row didn't exist (the target may be |
99 | | - // buffered, not yet materialised). |
100 | | - // Either way we want to try the buffer fallback below; treating the |
101 | | - // undefined-return as success would make the fallback unreachable. |
102 | | - const [error, result] = await tryCatch( |
103 | | - updateMetadataService.call(targetRunId, { operations }, env) |
104 | | - ); |
105 | | - if (!error && result !== undefined) { |
106 | | - // The parent/root run changed too — wake its live feeds (only when something was |
107 | | - // actually written here; buffered writes publish from the flusher). |
108 | | - if (result.updatedAtMs !== undefined) { |
109 | | - publishChangeRecord({ |
110 | | - runId: result.runId, |
111 | | - envId: env.id, |
112 | | - tags: result.runTags, |
113 | | - batchId: result.batchId, |
114 | | - updatedAtMs: result.updatedAtMs, |
115 | | - }); |
116 | | - } |
117 | | - return; |
118 | | - } |
119 | | - |
120 | | - if (error) { |
121 | | - // PG threw — auxiliary op, stay best-effort and don't surface this |
122 | | - // to the caller (the caller's primary mutation already landed). But |
123 | | - // warn so a genuine PG outage on these ops isn't invisible. |
124 | | - logger.warn("metadata route: parent/root PG op failed", { |
125 | | - targetRunId, |
126 | | - error: error instanceof Error ? error.message : String(error), |
127 | | - }); |
128 | | - } |
129 | | - |
130 | | - // Buffer fallback only makes sense for friendlyId-keyed entries. The |
131 | | - // PG-side parent/root IDs are internal cuids; the buffer keys entries |
132 | | - // by friendlyId, so passing the internal id would silently no-op. |
133 | | - // Skip explicitly — a buffered child's parent is always materialised |
134 | | - // in PG already (a buffered run hasn't executed, so it can't have |
135 | | - // triggered the child), so the buffered-parent branch isn't actually |
136 | | - // reachable. Treating the no-op as intentional rather than incidental. |
137 | | - if (!targetRunId.startsWith("run_")) return; |
138 | | - |
139 | | - // Best-effort buffer fallback. Wrap so a transient Redis throw on |
140 | | - // this auxiliary op can't 500 the request after the primary mutation |
141 | | - // already succeeded. |
142 | | - const [bufferError, bufferOutcome] = await tryCatch( |
143 | | - applyMetadataMutationToBufferedRun({ |
144 | | - runId: targetRunId, |
145 | | - environmentId: env.id, |
146 | | - organizationId: env.organizationId, |
147 | | - maximumSize: appEnv.TASK_RUN_METADATA_MAXIMUM_SIZE, |
148 | | - maxRetries: appEnv.TRIGGER_MOLLIFIER_METADATA_MAX_RETRIES, |
149 | | - backoffBaseMs: appEnv.TRIGGER_MOLLIFIER_METADATA_BACKOFF_BASE_MS, |
150 | | - backoffStepMs: appEnv.TRIGGER_MOLLIFIER_METADATA_BACKOFF_STEP_MS, |
151 | | - body: { operations }, |
152 | | - }) |
153 | | - ); |
154 | | - if (bufferError) { |
155 | | - logger.warn("metadata route: buffer fallback for parent/root op failed", { |
156 | | - targetRunId, |
157 | | - error: bufferError instanceof Error ? bufferError.message : String(bufferError), |
158 | | - }); |
159 | | - return; |
160 | | - } |
161 | | - // `applyMetadataMutationToBufferedRun` reports non-throw failures via |
162 | | - // its returned outcome kind: `not_found`, `busy`, `version_exhausted`, |
163 | | - // `metadata_too_large`. Without inspecting `.kind`, the parent/root |
164 | | - // operation can silently disappear — no PG row landed it (handled |
165 | | - // above) and the buffer rejected it for one of these reasons but the |
166 | | - // helper returned cleanly. Surface a warn log per non-success branch |
167 | | - // so ops can trace why a parent/root op went missing. The customer's |
168 | | - // primary mutation has already succeeded by this point; this remains |
169 | | - // best-effort, so we still don't bubble these to the response. |
170 | | - if (bufferOutcome && bufferOutcome.kind !== "applied") { |
171 | | - logger.warn("metadata route: parent/root buffer op did not apply", { |
172 | | - targetRunId, |
173 | | - kind: bufferOutcome.kind, |
174 | | - }); |
175 | | - } |
176 | | -} |
177 | | - |
178 | 67 | const { action } = createActionApiRoute( |
179 | 68 | { |
180 | 69 | params: ParamsSchema, |
|
0 commit comments