Object Storage seamless migration (#3275)

This allows seamless migration to different object storage.

Existing runs that have offloaded payloads/outputs will continue to use
the default object store (configured using `OBJECT_STORE_*` env vars).

You can add additional stores by setting new env vars:
- `OBJECT_STORE_DEFAULT_PROTOCOL` this determines where new run large
payloads will get stored.
- If you set that you need to set new env vars for that protocol.
  
Example:

```
OBJECT_STORE_DEFAULT_PROTOCOL=“s3"
OBJECT_STORE_S3_BASE_URL=https://s3.us-east-1.amazonaws.com
OBJECT_STORE_S3_ACCESS_KEY_ID=<val>
OBJECT_STORE_S3_SECRET_ACCESS_KEY=<val>
OBJECT_STORE_S3_REGION=us-east-1
OBJECT_STORE_S3_SERVICE=s3
```

---------

Co-authored-by: nicktrn <55853254+nicktrn@users.noreply.github.com>

Author: matt-aitken

.changeset/packet-v2-packets-api.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Large run outputs can use the new API which allows switching object storage providers.

.env.example

@@ -77,9 +77,28 @@ POSTHOG_PROJECT_KEY=
 # DEPOT_TOKEN=<Depot org token>
 # DEV_OTEL_EXPORTER_OTLP_ENDPOINT="http://0.0.0.0:4318"
 # These are needed for the object store (for handling large payloads/outputs)
-# OBJECT_STORE_BASE_URL="https://{bucket}.{accountId}.r2.cloudflarestorage.com"
-# OBJECT_STORE_ACCESS_KEY_ID=
-# OBJECT_STORE_SECRET_ACCESS_KEY=
+#
+# Default provider
+# OBJECT_STORE_BASE_URL=http://localhost:9005
+# OBJECT_STORE_BUCKET=packets
+# OBJECT_STORE_ACCESS_KEY_ID=minioadmin
+# OBJECT_STORE_SECRET_ACCESS_KEY=minioadmin
+# OBJECT_STORE_REGION=us-east-1
+# OBJECT_STORE_SERVICE=s3
+#
+# OBJECT_STORE_DEFAULT_PROTOCOL=s3  # Only specify this if you're going to migrate object storage and set protocol values below
+# Named providers (protocol-prefixed data) - optional for multi-provider support
+# OBJECT_STORE_S3_BASE_URL=https://s3.amazonaws.com
+# OBJECT_STORE_S3_ACCESS_KEY_ID=
+# OBJECT_STORE_S3_SECRET_ACCESS_KEY=
+# OBJECT_STORE_S3_REGION=us-east-1
+# OBJECT_STORE_S3_SERVICE=s3
+#
+# OBJECT_STORE_R2_BASE_URL=https://{bucket}.{accountId}.r2.cloudflarestorage.com
+# OBJECT_STORE_R2_ACCESS_KEY_ID=
+# OBJECT_STORE_R2_SECRET_ACCESS_KEY=
+# OBJECT_STORE_R2_REGION=auto
+# OBJECT_STORE_R2_SERVICE=s3
 # CHECKPOINT_THRESHOLD_IN_MS=10000
 
 # These control the server-side internal telemetry

.server-changes/multi-provider-object-storage.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Multi-provider object storage with protocol-based routing for zero-downtime migration

.server-changes/object-store-iam-auth.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Add IAM role-based auth support for object stores (no access keys required).

apps/webapp/app/env.server.ts

@@ -349,11 +349,18 @@ const EnvironmentSchema = z
       .default(60 * 1000 * 15), // 15 minutes
 
     OBJECT_STORE_BASE_URL: z.string().optional(),
+    OBJECT_STORE_BUCKET: z.string().optional(),
     OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
     OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
     OBJECT_STORE_REGION: z.string().optional(),
     OBJECT_STORE_SERVICE: z.string().default("s3"),
 
+    // Protocol to use for new uploads (e.g., "s3", "r2"). Data without protocol uses default provider above.
+    // If specified, you must configure the corresponding provider using OBJECT_STORE_{PROTOCOL}_* env vars.
+    // Example: OBJECT_STORE_DEFAULT_PROTOCOL=s3 requires OBJECT_STORE_S3_BASE_URL, OBJECT_STORE_S3_ACCESS_KEY_ID, etc.
+    // Enables zero-downtime migration between providers (old data keeps working, new data uses new provider).
+    OBJECT_STORE_DEFAULT_PROTOCOL: z.string().regex(/^[a-z0-9]+$/).optional(),
+
     ARTIFACTS_OBJECT_STORE_BUCKET: z.string().optional(),
     ARTIFACTS_OBJECT_STORE_BASE_URL: z.string().optional(),
     ARTIFACTS_OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),

apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts

@@ -15,7 +15,7 @@ import assertNever from "assert-never";
 import { API_VERSIONS, CURRENT_API_VERSION, RunStatusUnspecifiedApiVersion } from "~/api/versions";
 import { $replica, prisma } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { generatePresignedUrl } from "~/v3/r2.server";
+import { generatePresignedUrl } from "~/v3/objectStore.server";
 import { tracer } from "~/v3/tracer.server";
 import { startSpanWithEnv } from "~/v3/tracing.server";
 

apps/webapp/app/routes/api.v1.packets.$.ts

@@ -3,7 +3,7 @@ import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { generatePresignedUrl } from "~/v3/r2.server";
+import { generatePresignedUrl } from "~/v3/objectStore.server";
 
 const ParamsSchema = z.object({
   "*": z.string(),
@@ -29,7 +29,8 @@ export async function action({ request, params }: ActionFunctionArgs) {
     authenticationResult.environment.project.externalRef,
     authenticationResult.environment.slug,
     filename,
-    "PUT"
+    "PUT",
+    { forceNoPrefix: true }
   );
 
   if (!signed.success) {

apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts

@@ -1,13 +1,11 @@
 import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
-import {
-  type CompleteWaitpointTokenResponseBody,
-  conditionallyExportPacket,
-  stringifyIO,
-} from "@trigger.dev/core/v3";
+import { type CompleteWaitpointTokenResponseBody, stringifyIO } from "@trigger.dev/core/v3";
 import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { env } from "~/env.server";
+import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { verifyHttpCallbackHash } from "~/services/httpCallback.server";
 import { logger } from "~/services/logger.server";
 import { engine } from "~/v3/runEngine.server";
@@ -41,8 +39,10 @@ export async function action({ request, params }: ActionFunctionArgs) {
       },
       include: {
         environment: {
-          select: {
-            apiKey: true,
+          include: {
+            project: true,
+            organization: true,
+            orgMember: true,
             parentEnvironment: {
               select: {
                 apiKey: true,
@@ -77,9 +77,10 @@ export async function action({ request, params }: ActionFunctionArgs) {
     const body = await request.json().catch(() => ({}));
 
     const stringifiedData = await stringifyIO(body);
-    const finalData = await conditionallyExportPacket(
+    const finalData = await processWaitpointCompletionPacket(
       stringifiedData,
-      `${waitpointId}/waitpoint/http-callback`
+      waitpoint.environment,
+      `${WaitpointId.toFriendlyId(waitpointId)}/http-callback`
     );
 
     const result = await engine.completeWaitpoint({

apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts

@@ -2,14 +2,14 @@ import { json } from "@remix-run/server-runtime";
 import {
   CompleteWaitpointTokenRequestBody,
   type CompleteWaitpointTokenResponseBody,
-  conditionallyExportPacket,
   stringifyIO,
 } from "@trigger.dev/core/v3";
 import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
+import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { engine } from "~/v3/runEngine.server";
 
@@ -52,9 +52,10 @@ const { action, loader } = createActionApiRoute(
       }
 
       const stringifiedData = await stringifyIO(body.data);
-      const finalData = await conditionallyExportPacket(
+      const finalData = await processWaitpointCompletionPacket(
         stringifiedData,
-        `${waitpointId}/waitpoint/token`
+        authentication.environment,
+        `${WaitpointId.toFriendlyId(waitpointId)}/token`
       );
 
       const result = await engine.completeWaitpoint({

apps/webapp/app/routes/api.v2.packets.$.ts

@@ -0,0 +1,45 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { generatePresignedUrl } from "~/v3/objectStore.server";
+
+const ParamsSchema = z.object({
+  "*": z.string(),
+});
+
+/**
+ * PUT-only presign for packet uploads (SDK offload). Uses OBJECT_STORE_DEFAULT_PROTOCOL for
+ * unprefixed keys; returns canonical storagePath for IOPacket.data. GET presigns use v1.
+ */
+export async function action({ request, params }: ActionFunctionArgs) {
+  if (request.method.toUpperCase() !== "PUT") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const parsedParams = ParamsSchema.parse(params);
+  const filename = parsedParams["*"];
+
+  const signed = await generatePresignedUrl(
+    authenticationResult.environment.project.externalRef,
+    authenticationResult.environment.slug,
+    filename,
+    "PUT"
+  );
+
+  if (!signed.success) {
+    return json({ error: `Failed to generate presigned URL: ${signed.error}` }, { status: 500 });
+  }
+
+  if (signed.storagePath === undefined) {
+    return json({ error: "Failed to resolve storage path for packet upload" }, { status: 500 });
+  }
+
+  return json({ presignedUrl: signed.url, storagePath: signed.storagePath });
+}