Add ability to update parent and root run metadata from children

.changeset/giant-mice-cheer.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Increase the number of active streams from 2 to 5, total streams from 5 to 10

.changeset/slow-deers-collect.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/react-hooks": patch
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Adding ability to update parent run metadata from child runs/tasks

apps/webapp/app/env.server.ts

@@ -253,13 +253,14 @@ const EnvironmentSchema = z.object({
   TASK_PAYLOAD_OFFLOAD_THRESHOLD: z.coerce.number().int().default(524_288), // 512KB
   TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(3_145_728), // 3MB
   BATCH_TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(1_000_000), // 1MB
-  TASK_RUN_METADATA_MAXIMUM_SIZE: z.coerce.number().int().default(4_096), // 4KB
+  TASK_RUN_METADATA_MAXIMUM_SIZE: z.coerce.number().int().default(262_144), // 256KB
 
   MAXIMUM_DEV_QUEUE_SIZE: z.coerce.number().int().optional(),
   MAXIMUM_DEPLOYED_QUEUE_SIZE: z.coerce.number().int().optional(),
   MAX_BATCH_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
 
   REALTIME_STREAM_VERSION: z.enum(["v1", "v2"]).default("v1"),
+  BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts

@@ -1,94 +1,29 @@
-import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
-import { parsePacket, UpdateMetadataRequestBody } from "@trigger.dev/core/v3";
+import { json } from "@remix-run/server-runtime";
+import { UpdateMetadataRequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
-import { prisma } from "~/db.server";
-import { authenticateApiRequest } from "~/services/apiAuth.server";
-import { handleMetadataPacket } from "~/utils/packets";
-import { ServiceValidationError } from "~/v3/services/baseService.server";
-import { isFinalRunStatus } from "~/v3/taskStatus";
+import { updateMetadataService } from "~/services/metadata/updateMetadata.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
 });
 
-export async function action({ request, params }: ActionFunctionArgs) {
-  // Ensure this is a PUT request
-  if (request.method.toUpperCase() !== "PUT") {
-    return json({ error: "Method not allowed" }, { status: 405, headers: { Allow: "PUT" } });
-  }
-
-  // Authenticate the request
-  const authenticationResult = await authenticateApiRequest(request);
-  if (!authenticationResult) {
-    return 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 }
-    );
-  }
-
-  try {
-    const anyBody = await request.json();
-
-    const body = UpdateMetadataRequestBody.safeParse(anyBody);
-
-    if (!body.success) {
-      return json({ error: "Invalid request body", issues: body.error.issues }, { status: 400 });
-    }
-
-    const metadataPacket = handleMetadataPacket(
-      body.data.metadata,
-      body.data.metadataType ?? "application/json"
-    );
-
-    if (!metadataPacket) {
-      return json({ error: "Invalid metadata" }, { status: 400 });
-    }
-
-    const taskRun = await prisma.taskRun.findFirst({
-      where: {
-        friendlyId: parsedParams.data.runId,
-        runtimeEnvironmentId: authenticationResult.environment.id,
-      },
-      select: {
-        status: true,
-      },
-    });
-
-    if (!taskRun) {
+const { action } = createActionApiRoute(
+  {
+    params: ParamsSchema,
+    body: UpdateMetadataRequestBody,
+    maxContentLength: 1024 * 1024, // 1MB
+    method: "PUT",
+  },
+  async ({ authentication, body, params }) => {
+    const result = await updateMetadataService.call(authentication.environment, params.runId, body);
+
+    if (!result) {
       return json({ error: "Task Run not found" }, { status: 404 });
     }
 
-    if (isFinalRunStatus(taskRun.status)) {
-      return json({ error: "Cannot update metadata for a completed run" }, { status: 400 });
-    }
-
-    await prisma.taskRun.update({
-      where: {
-        friendlyId: parsedParams.data.runId,
-        runtimeEnvironmentId: authenticationResult.environment.id,
-      },
-      data: {
-        metadata: metadataPacket?.data,
-        metadataType: metadataPacket?.dataType,
-      },
-    });
-
-    const parsedPacket = await parsePacket(metadataPacket);
-
-    return json({ metadata: parsedPacket }, { status: 200 });
-  } catch (error) {
-    if (error instanceof ServiceValidationError) {
-      return json({ error: error.message }, { status: error.status ?? 422 });
-    } else {
-      return json(
-        { error: error instanceof Error ? error.message : "Internal Server Error" },
-        { status: 500 }
-      );
-    }
+    return json(result, { status: 200 });
   }
-}
+);
+
+export { action };

apps/webapp/app/routes/api.v1.tasks.batch.ts

@@ -1,23 +1,21 @@
 import { json } from "@remix-run/server-runtime";
 import {
-  BatchTriggerTaskResponse,
   BatchTriggerTaskV2RequestBody,
   BatchTriggerTaskV2Response,
   generateJWT,
 } from "@trigger.dev/core/v3";
 import { env } from "~/env.server";
+import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
 import {
   BatchProcessingStrategy,
   BatchTriggerV2Service,
 } from "~/v3/services/batchTriggerV2.server";
-import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
-import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
-import { logger } from "~/services/logger.server";
-import { z } from "zod";
+import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 
 const { action, loader } = createActionApiRoute(
   {

apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts

@@ -2,7 +2,6 @@ import { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { relayRealtimeStreams } from "~/services/realtime/relayRealtimeStreams.server";
-import { v1RealtimeStreams } from "~/services/realtime/v1StreamsGlobal.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 
 const ParamsSchema = z.object({

apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts

@@ -0,0 +1,61 @@
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { relayRealtimeStreams } from "~/services/realtime/relayRealtimeStreams.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+const ParamsSchema = z.object({
+  runId: z.string(),
+  target: z.enum(["self", "parent", "root"]),
+  streamId: z.string(),
+});
+
+const { action } = createActionApiRoute(
+  {
+    params: ParamsSchema,
+  },
+  async ({ request, params, authentication }) => {
+    if (!request.body) {
+      return new Response("No body provided", { status: 400 });
+    }
+
+    const run = await $replica.taskRun.findFirst({
+      where: {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: authentication.environment.id,
+      },
+      select: {
+        id: true,
+        friendlyId: true,
+        parentTaskRun: {
+          select: {
+            friendlyId: true,
+          },
+        },
+        rootTaskRun: {
+          select: {
+            friendlyId: true,
+          },
+        },
+      },
+    });
+
+    if (!run) {
+      return new Response("Run not found", { status: 404 });
+    }
+
+    const targetId =
+      params.target === "self"
+        ? run.friendlyId
+        : params.target === "parent"
+        ? run.parentTaskRun?.friendlyId
+        : run.rootTaskRun?.friendlyId;
+
+    if (!targetId) {
+      return new Response("Target not found", { status: 404 });
+    }
+
+    return relayRealtimeStreams.ingestData(request.body, targetId, params.streamId);
+  }
+);
+
+export { action };