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>
This commit is contained in:
Matt Aitken
@@ -0,0 +1,5 @@
|
||||
---
|
||||
"@trigger.dev/core": patch
|
||||
---
|
||||
|
||||
Large run outputs can use the new API which allows switching object storage providers.
|
||||
+22
-3
@@ -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
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
---
|
||||
area: webapp
|
||||
type: feature
|
||||
---
|
||||
|
||||
Multi-provider object storage with protocol-based routing for zero-downtime migration
|
||||
@@ -0,0 +1,6 @@
|
||||
---
|
||||
area: webapp
|
||||
type: feature
|
||||
---
|
||||
|
||||
Add IAM role-based auth support for object stores (no access keys required).
|
||||
@@ -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(),
|
||||
|
||||
@@ -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";
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
+10
-9
@@ -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({
|
||||
|
||||
@@ -2,7 +2,6 @@ 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";
|
||||
@@ -10,6 +9,7 @@ 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({
|
||||
|
||||
@@ -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 });
|
||||
}
|
||||
+24
-9
@@ -2,13 +2,7 @@ import { env } from "~/env.server";
|
||||
import { parse } from "@conform-to/zod";
|
||||
import { Form, useLocation, useNavigation, useSubmit } from "@remix-run/react";
|
||||
import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
|
||||
import {
|
||||
conditionallyExportPacket,
|
||||
IOPacket,
|
||||
stringifyIO,
|
||||
timeoutError,
|
||||
WaitpointTokenStatus,
|
||||
} from "@trigger.dev/core/v3";
|
||||
import { stringifyIO, timeoutError, WaitpointTokenStatus } from "@trigger.dev/core/v3";
|
||||
import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
|
||||
import type { Waitpoint } from "@trigger.dev/database";
|
||||
import { useCallback, useRef } from "react";
|
||||
@@ -24,6 +18,8 @@ import { $replica } from "~/db.server";
|
||||
import { useOrganization } from "~/hooks/useOrganizations";
|
||||
import { useProject } from "~/hooks/useProject";
|
||||
import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
|
||||
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
|
||||
import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import { requireUserId } from "~/services/session.server";
|
||||
import { EnvironmentParamSchema, ProjectParamSchema, v3RunsPath } from "~/utils/pathBuilder";
|
||||
@@ -86,6 +82,7 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
|
||||
const waitpoint = await $replica.waitpoint.findFirst({
|
||||
select: {
|
||||
projectId: true,
|
||||
environmentId: true,
|
||||
},
|
||||
where: {
|
||||
id: waitpointId,
|
||||
@@ -150,11 +147,29 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
|
||||
);
|
||||
}
|
||||
|
||||
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
|
||||
if (!environment) {
|
||||
return redirectWithErrorMessage(
|
||||
submission.value.failureRedirect,
|
||||
request,
|
||||
"Environment not found"
|
||||
);
|
||||
}
|
||||
|
||||
if (environment.id !== waitpoint.environmentId) {
|
||||
return redirectWithErrorMessage(
|
||||
submission.value.failureRedirect,
|
||||
request,
|
||||
"No waitpoint found"
|
||||
);
|
||||
}
|
||||
|
||||
const data = submission.value.payload ? JSON.parse(submission.value.payload) : {};
|
||||
const stringifiedData = await stringifyIO(data);
|
||||
const finalData = await conditionallyExportPacket(
|
||||
const finalData = await processWaitpointCompletionPacket(
|
||||
stringifiedData,
|
||||
`${waitpointId}/waitpoint/token`
|
||||
environment,
|
||||
`${WaitpointId.toFriendlyId(waitpointId)}/token`
|
||||
);
|
||||
|
||||
const result = await engine.completeWaitpoint({
|
||||
|
||||
@@ -3,7 +3,7 @@ import { basename } from "node:path";
|
||||
import { z } from "zod";
|
||||
import { prisma } from "~/db.server";
|
||||
import { requireUserId } from "~/services/session.server";
|
||||
import { generatePresignedRequest } from "~/v3/r2.server";
|
||||
import { generatePresignedRequest } from "~/v3/objectStore.server";
|
||||
|
||||
const ParamSchema = z.object({
|
||||
environmentId: z.string(),
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import { IOPacket, packetRequiresOffloading, tryCatch } from "@trigger.dev/core/v3";
|
||||
import { type IOPacket, packetRequiresOffloading, tryCatch } from "@trigger.dev/core/v3";
|
||||
import { env } from "~/env.server";
|
||||
import { startActiveSpan } from "~/v3/tracer.server";
|
||||
import { uploadPacketToObjectStore, r2 } from "~/v3/r2.server";
|
||||
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import { hasObjectStoreClient, uploadPacketToObjectStore } from "~/v3/objectStore.server";
|
||||
import { startActiveSpan } from "~/v3/tracer.server";
|
||||
|
||||
export type BatchPayloadProcessResult = {
|
||||
/** The processed payload - either the original or an R2 path */
|
||||
@@ -31,7 +31,7 @@ export class BatchPayloadProcessor {
|
||||
* If not available, large payloads will be stored inline (which may fail for very large payloads).
|
||||
*/
|
||||
isObjectStoreAvailable(): boolean {
|
||||
return r2 !== undefined && env.OBJECT_STORE_BASE_URL !== undefined;
|
||||
return hasObjectStoreClient();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -103,11 +103,17 @@ export class BatchPayloadProcessor {
|
||||
};
|
||||
}
|
||||
|
||||
// Upload to R2
|
||||
// Upload to object store
|
||||
const filename = `batch_${batchId}/item_${itemIndex}/payload.json`;
|
||||
|
||||
const [uploadError] = await tryCatch(
|
||||
uploadPacketToObjectStore(filename, packet.data, packet.dataType, environment)
|
||||
const [uploadError, uploadedFilename] = await tryCatch(
|
||||
uploadPacketToObjectStore(
|
||||
filename,
|
||||
packet.data,
|
||||
packet.dataType,
|
||||
environment,
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL
|
||||
)
|
||||
);
|
||||
|
||||
if (uploadError) {
|
||||
@@ -125,18 +131,18 @@ export class BatchPayloadProcessor {
|
||||
);
|
||||
}
|
||||
|
||||
logger.debug("Batch item payload offloaded to R2", {
|
||||
logger.debug("Batch item payload offloaded to object store", {
|
||||
batchId,
|
||||
itemIndex,
|
||||
filename,
|
||||
filename: uploadedFilename,
|
||||
size,
|
||||
});
|
||||
|
||||
span.setAttribute("wasOffloaded", true);
|
||||
span.setAttribute("offloadPath", filename);
|
||||
span.setAttribute("offloadPath", uploadedFilename);
|
||||
|
||||
return {
|
||||
payload: filename,
|
||||
payload: uploadedFilename!,
|
||||
payloadType: "application/store",
|
||||
wasOffloaded: true,
|
||||
size,
|
||||
|
||||
@@ -2,7 +2,7 @@ import { IOPacket, packetRequiresOffloading, tryCatch } from "@trigger.dev/core/
|
||||
import { PayloadProcessor, TriggerTaskRequest } from "../types";
|
||||
import { env } from "~/env.server";
|
||||
import { startActiveSpan } from "~/v3/tracer.server";
|
||||
import { uploadPacketToObjectStore } from "~/v3/r2.server";
|
||||
import { uploadPacketToObjectStore } from "~/v3/objectStore.server";
|
||||
import { ServiceValidationError } from "~/v3/services/common.server";
|
||||
|
||||
export class DefaultPayloadProcessor implements PayloadProcessor {
|
||||
@@ -31,8 +31,8 @@ export class DefaultPayloadProcessor implements PayloadProcessor {
|
||||
|
||||
const filename = `${request.friendlyId}/payload.json`;
|
||||
|
||||
const [uploadError] = await tryCatch(
|
||||
uploadPacketToObjectStore(filename, packet.data, packet.dataType, request.environment)
|
||||
const [uploadError, uploadedFilename] = await tryCatch(
|
||||
uploadPacketToObjectStore(filename, packet.data, packet.dataType, request.environment, env.OBJECT_STORE_DEFAULT_PROTOCOL)
|
||||
);
|
||||
|
||||
if (uploadError) {
|
||||
@@ -40,7 +40,7 @@ export class DefaultPayloadProcessor implements PayloadProcessor {
|
||||
}
|
||||
|
||||
return {
|
||||
data: filename,
|
||||
data: uploadedFilename!,
|
||||
dataType: "application/store",
|
||||
};
|
||||
});
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import { type IOPacket, packetRequiresOffloading, tryCatch } from "@trigger.dev/core/v3";
|
||||
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
||||
import { env } from "~/env.server";
|
||||
import { uploadPacketToObjectStore } from "~/v3/objectStore.server";
|
||||
import { ServiceValidationError } from "~/v3/services/common.server";
|
||||
|
||||
function packetExtensionForDataType(dataType: string): string {
|
||||
switch (dataType) {
|
||||
case "application/json":
|
||||
case "application/super+json":
|
||||
return "json";
|
||||
case "text/plain":
|
||||
return "txt";
|
||||
default:
|
||||
return "txt";
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Offloads large waitpoint completion payloads to object store (same threshold and
|
||||
* upload path pattern as DefaultPayloadProcessor). Object key prefix should use the
|
||||
* waitpoint friendly id folder, e.g. `${WaitpointId.toFriendlyId(internalId)}/token`.
|
||||
* Replaces no-op conditionallyExportPacket usage in webapp routes where apiClientManager is unset.
|
||||
*/
|
||||
export async function processWaitpointCompletionPacket(
|
||||
packet: IOPacket,
|
||||
environment: AuthenticatedEnvironment,
|
||||
pathPrefix: string
|
||||
): Promise<IOPacket> {
|
||||
if (!packet.data) {
|
||||
return packet;
|
||||
}
|
||||
|
||||
const { needsOffloading, size } = packetRequiresOffloading(
|
||||
packet,
|
||||
env.TASK_PAYLOAD_OFFLOAD_THRESHOLD
|
||||
);
|
||||
|
||||
if (!needsOffloading) {
|
||||
return packet;
|
||||
}
|
||||
|
||||
const filename = `${pathPrefix}.${packetExtensionForDataType(packet.dataType)}`;
|
||||
|
||||
const [uploadError, uploadedFilename] = await tryCatch(
|
||||
uploadPacketToObjectStore(
|
||||
filename,
|
||||
packet.data,
|
||||
packet.dataType,
|
||||
environment,
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL
|
||||
)
|
||||
);
|
||||
|
||||
if (uploadError) {
|
||||
throw new ServiceValidationError("Failed to upload large waitpoint to object store", 500);
|
||||
}
|
||||
|
||||
return {
|
||||
data: uploadedFilename!,
|
||||
dataType: "application/store",
|
||||
};
|
||||
}
|
||||
@@ -16,7 +16,7 @@ import { env } from "~/env.server";
|
||||
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import { batchTriggerWorker } from "~/v3/batchTriggerWorker.server";
|
||||
import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../../v3/r2.server";
|
||||
import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../../v3/objectStore.server";
|
||||
import { ServiceValidationError, WithRunEngine } from "../../v3/services/baseService.server";
|
||||
import { TriggerTaskService } from "../../v3/services/triggerTask.server";
|
||||
import { startActiveSpan } from "../../v3/tracer.server";
|
||||
@@ -716,10 +716,10 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
|
||||
|
||||
const filename = `${pathPrefix}/payload.json`;
|
||||
|
||||
await uploadPacketToObjectStore(filename, packet.data, packet.dataType, environment);
|
||||
const uploadedFilename = await uploadPacketToObjectStore(filename, packet.data, packet.dataType, environment);
|
||||
|
||||
return {
|
||||
data: filename,
|
||||
data: uploadedFilename,
|
||||
dataType: "application/store",
|
||||
};
|
||||
});
|
||||
|
||||
@@ -0,0 +1,317 @@
|
||||
import { type IOPacket } from "@trigger.dev/core/v3";
|
||||
import { env } from "~/env.server";
|
||||
import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import { singleton } from "~/utils/singleton";
|
||||
import { ObjectStoreClient, type ObjectStoreClientConfig } from "./objectStoreClient.server";
|
||||
|
||||
/**
|
||||
* Parsed storage URI with optional protocol prefix
|
||||
* @example { protocol: "s3", path: "run_abc/payload.json" }
|
||||
* @example { protocol: undefined, path: "batch_123/item_0/payload.json" } // legacy, uses default
|
||||
*/
|
||||
export type ParsedStorageUri = {
|
||||
protocol?: string;
|
||||
path: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Parse a storage URI into protocol and path components
|
||||
* @param uri Storage URI, optionally prefixed with protocol (e.g., "s3://path" or "path")
|
||||
* @returns Parsed components { protocol?, path }
|
||||
*/
|
||||
export function parseStorageUri(uri: string): ParsedStorageUri {
|
||||
const match = uri.match(/^([a-z0-9]+):\/\/(.+)$/);
|
||||
if (match) {
|
||||
return {
|
||||
protocol: match[1],
|
||||
path: match[2],
|
||||
};
|
||||
}
|
||||
return {
|
||||
protocol: undefined,
|
||||
path: uri,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Format a storage URI with optional protocol prefix
|
||||
* @param path Storage path
|
||||
* @param protocol Optional protocol to prefix (e.g., "s3", "r2")
|
||||
* @returns Formatted URI (e.g., "s3://path" or "path")
|
||||
*/
|
||||
export function formatStorageUri(path: string, protocol?: string): string {
|
||||
if (protocol) {
|
||||
return `${protocol}://${path}`;
|
||||
}
|
||||
return path;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get object storage configuration for a given protocol.
|
||||
* Returns a config if baseUrl is set, even without explicit credentials —
|
||||
* in that case the AWS credential chain (ECS task role, EC2 IMDS, etc.) is used,
|
||||
* and OBJECT_STORE_BUCKET must also be set.
|
||||
*/
|
||||
function getObjectStoreConfig(protocol?: string): ObjectStoreClientConfig | undefined {
|
||||
if (protocol) {
|
||||
// Named provider (e.g., OBJECT_STORE_S3_*)
|
||||
const prefix = `OBJECT_STORE_${protocol.toUpperCase()}_`;
|
||||
const baseUrl = process.env[`${prefix}BASE_URL`];
|
||||
if (!baseUrl) return undefined;
|
||||
|
||||
return {
|
||||
baseUrl,
|
||||
bucket: process.env[`${prefix}BUCKET`] || undefined,
|
||||
accessKeyId: process.env[`${prefix}ACCESS_KEY_ID`] || undefined,
|
||||
secretAccessKey: process.env[`${prefix}SECRET_ACCESS_KEY`] || undefined,
|
||||
region: process.env[`${prefix}REGION`] || undefined,
|
||||
service: process.env[`${prefix}SERVICE`] || undefined,
|
||||
};
|
||||
}
|
||||
|
||||
// Default provider (backward compatible)
|
||||
if (!env.OBJECT_STORE_BASE_URL) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return {
|
||||
baseUrl: env.OBJECT_STORE_BASE_URL,
|
||||
bucket: env.OBJECT_STORE_BUCKET || undefined,
|
||||
accessKeyId: env.OBJECT_STORE_ACCESS_KEY_ID || undefined,
|
||||
secretAccessKey: env.OBJECT_STORE_SECRET_ACCESS_KEY || undefined,
|
||||
region: env.OBJECT_STORE_REGION || undefined,
|
||||
service: env.OBJECT_STORE_SERVICE || undefined,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Object storage client registry. Maps protocol name to ObjectStoreClient singleton.
|
||||
* ObjectStoreClient internally uses either aws4fetch (static credentials) or the
|
||||
* AWS SDK S3Client (IAM credential chain), selected at creation time.
|
||||
*/
|
||||
const objectStoreClients = singleton(
|
||||
"objectStoreClients",
|
||||
() => new Map<string, ObjectStoreClient>()
|
||||
);
|
||||
|
||||
function getObjectStoreClient(protocol?: string): ObjectStoreClient | undefined {
|
||||
const config = getObjectStoreConfig(protocol);
|
||||
if (!config) return undefined;
|
||||
|
||||
// Key includes baseUrl so that config changes (e.g. different containers in tests)
|
||||
// always produce a fresh client while production usage (stable env) is effectively
|
||||
// a per-protocol singleton.
|
||||
const cacheKey = `${protocol ?? "default"}:${config.baseUrl}`;
|
||||
if (objectStoreClients.has(cacheKey)) {
|
||||
return objectStoreClients.get(cacheKey);
|
||||
}
|
||||
|
||||
const client = ObjectStoreClient.create(config);
|
||||
objectStoreClients.set(cacheKey, client);
|
||||
return client;
|
||||
}
|
||||
|
||||
export function hasObjectStoreClient(): boolean {
|
||||
const defaultConfig = getObjectStoreConfig();
|
||||
const protocolConfig = env.OBJECT_STORE_DEFAULT_PROTOCOL
|
||||
? getObjectStoreConfig(env.OBJECT_STORE_DEFAULT_PROTOCOL)
|
||||
: undefined;
|
||||
return !!(defaultConfig || protocolConfig);
|
||||
}
|
||||
|
||||
export async function uploadPacketToObjectStore(
|
||||
filename: string,
|
||||
data: ReadableStream | string,
|
||||
contentType: string,
|
||||
environment: AuthenticatedEnvironment,
|
||||
storageProtocol?: string
|
||||
): Promise<string> {
|
||||
const protocol = storageProtocol || env.OBJECT_STORE_DEFAULT_PROTOCOL;
|
||||
const client = getObjectStoreClient(protocol);
|
||||
|
||||
if (!client) {
|
||||
throw new Error(`Object store is not configured for protocol: ${protocol || "default"}`);
|
||||
}
|
||||
|
||||
const key = `packets/${environment.project.externalRef}/${environment.slug}/${filename}`;
|
||||
|
||||
logger.debug("Uploading to object store", { key, protocol: protocol || "default" });
|
||||
|
||||
await client.putObject(key, data, contentType);
|
||||
|
||||
// Return filename with protocol prefix if specified
|
||||
return formatStorageUri(filename, protocol);
|
||||
}
|
||||
|
||||
export async function downloadPacketFromObjectStore(
|
||||
packet: IOPacket,
|
||||
environment: AuthenticatedEnvironment
|
||||
): Promise<IOPacket> {
|
||||
if (packet.dataType !== "application/store") {
|
||||
return packet;
|
||||
}
|
||||
|
||||
// There shouldn't be an offloaded packet with undefined data…
|
||||
if (!packet.data) {
|
||||
logger.error("Object store packet has undefined data", { packet, environment });
|
||||
return {
|
||||
dataType: "application/json",
|
||||
data: undefined,
|
||||
};
|
||||
}
|
||||
|
||||
const { protocol, path } = parseStorageUri(packet.data);
|
||||
const client = getObjectStoreClient(protocol);
|
||||
|
||||
if (!client) {
|
||||
throw new Error(`Object store is not configured for protocol: ${protocol || "default"}`);
|
||||
}
|
||||
|
||||
const key = `packets/${environment.project.externalRef}/${environment.slug}/${path}`;
|
||||
|
||||
logger.debug("Downloading from object store", { key, protocol: protocol || "default" });
|
||||
|
||||
const data = await client.getObject(key);
|
||||
|
||||
return { data, dataType: "application/json" };
|
||||
}
|
||||
|
||||
export type GeneratePacketPresignOptions = {
|
||||
/**
|
||||
* When true (v1 packet PUT only), unprefixed keys use the legacy default object store only.
|
||||
* When false/omitted (v2 packet PUT), unprefixed keys also use OBJECT_STORE_DEFAULT_PROTOCOL.
|
||||
* Ignored for GET — reads never infer protocol from env for unprefixed keys.
|
||||
*/
|
||||
forceNoPrefix?: boolean;
|
||||
};
|
||||
|
||||
/**
|
||||
* Resolve object-store protocol for packet presigns.
|
||||
* GET: never apply OBJECT_STORE_DEFAULT_PROTOCOL to unprefixed keys.
|
||||
* PUT: optional forceNoPrefix for v1 legacy upload behavior.
|
||||
*/
|
||||
export function resolveStoreProtocolForPacketPresign(
|
||||
filename: string,
|
||||
method: "PUT" | "GET",
|
||||
forceNoPrefix?: boolean
|
||||
): { path: string; storeProtocol: string | undefined } {
|
||||
const { protocol: explicitProtocol, path } = parseStorageUri(filename);
|
||||
|
||||
if (method === "GET") {
|
||||
return { path, storeProtocol: explicitProtocol };
|
||||
}
|
||||
|
||||
if (explicitProtocol !== undefined) {
|
||||
return { path, storeProtocol: explicitProtocol };
|
||||
}
|
||||
|
||||
if (forceNoPrefix) {
|
||||
return { path, storeProtocol: undefined };
|
||||
}
|
||||
|
||||
return { path, storeProtocol: env.OBJECT_STORE_DEFAULT_PROTOCOL };
|
||||
}
|
||||
|
||||
export async function generatePresignedRequest(
|
||||
projectRef: string,
|
||||
envSlug: string,
|
||||
filename: string,
|
||||
method: "PUT" | "GET" = "PUT",
|
||||
options?: GeneratePacketPresignOptions
|
||||
): Promise<
|
||||
| {
|
||||
success: false;
|
||||
error: string;
|
||||
}
|
||||
| {
|
||||
success: true;
|
||||
request: Request;
|
||||
/** Canonical pointer for IOPacket.data (PUT only). */
|
||||
storagePath?: string;
|
||||
}
|
||||
> {
|
||||
const { path, storeProtocol } = resolveStoreProtocolForPacketPresign(
|
||||
filename,
|
||||
method,
|
||||
options?.forceNoPrefix
|
||||
);
|
||||
|
||||
const config = getObjectStoreConfig(storeProtocol);
|
||||
if (!config?.baseUrl) {
|
||||
return {
|
||||
success: false,
|
||||
error: `Object store is not configured for protocol: ${storeProtocol || "default"}`,
|
||||
};
|
||||
}
|
||||
|
||||
const client = getObjectStoreClient(storeProtocol);
|
||||
if (!client) {
|
||||
return {
|
||||
success: false,
|
||||
error: `Object store is not configured for protocol: ${storeProtocol || "default"}`,
|
||||
};
|
||||
}
|
||||
|
||||
const key = `packets/${projectRef}/${envSlug}/${path}`;
|
||||
|
||||
try {
|
||||
const url = await client.presign(key, method, 300); // 5 minutes
|
||||
|
||||
logger.debug("Generated presigned URL", {
|
||||
url,
|
||||
projectRef,
|
||||
envSlug,
|
||||
filename,
|
||||
protocol: storeProtocol || "default",
|
||||
});
|
||||
|
||||
const storagePath = method === "PUT" ? formatStorageUri(path, storeProtocol) : undefined;
|
||||
|
||||
return {
|
||||
success: true,
|
||||
request: new Request(url, { method }),
|
||||
storagePath,
|
||||
};
|
||||
} catch (error) {
|
||||
return {
|
||||
success: false,
|
||||
error: `Failed to generate presigned URL: ${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export async function generatePresignedUrl(
|
||||
projectRef: string,
|
||||
envSlug: string,
|
||||
filename: string,
|
||||
method: "PUT" | "GET" = "PUT",
|
||||
options?: GeneratePacketPresignOptions
|
||||
): Promise<
|
||||
| {
|
||||
success: false;
|
||||
error: string;
|
||||
}
|
||||
| {
|
||||
success: true;
|
||||
url: string;
|
||||
storagePath?: string;
|
||||
}
|
||||
> {
|
||||
const signed = await generatePresignedRequest(projectRef, envSlug, filename, method, options);
|
||||
|
||||
if (!signed.success) {
|
||||
return {
|
||||
success: false,
|
||||
error: signed.error,
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
success: true,
|
||||
url: signed.request.url,
|
||||
storagePath: signed.storagePath,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,200 @@
|
||||
import { AwsClient } from "aws4fetch";
|
||||
import { GetObjectCommand, PutObjectCommand, S3Client } from "@aws-sdk/client-s3";
|
||||
import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
|
||||
|
||||
interface IObjectStoreClient {
|
||||
putObject(key: string, body: ReadableStream | string, contentType: string): Promise<string>;
|
||||
getObject(key: string): Promise<string>;
|
||||
presign(key: string, method: "PUT" | "GET", expiresIn: number): Promise<string>;
|
||||
}
|
||||
|
||||
type Aws4FetchConfig = {
|
||||
baseUrl: string;
|
||||
accessKeyId: string;
|
||||
secretAccessKey: string;
|
||||
region?: string;
|
||||
service?: string;
|
||||
};
|
||||
|
||||
class Aws4FetchClient implements IObjectStoreClient {
|
||||
private readonly awsClient: AwsClient;
|
||||
|
||||
constructor(private readonly config: Aws4FetchConfig) {
|
||||
this.awsClient = new AwsClient({
|
||||
accessKeyId: config.accessKeyId,
|
||||
secretAccessKey: config.secretAccessKey,
|
||||
region: config.region,
|
||||
// We set the default value to "s3" in the schema to enhance interoperability with various
|
||||
// S3-compatible services. Setting this env var to an empty string restores the old behaviour.
|
||||
service: config.service || undefined,
|
||||
});
|
||||
}
|
||||
|
||||
private buildUrl(key: string): string {
|
||||
const url = new URL(this.config.baseUrl);
|
||||
url.pathname = `/${key}`;
|
||||
return url.toString();
|
||||
}
|
||||
|
||||
async putObject(key: string, body: ReadableStream | string, contentType: string): Promise<string> {
|
||||
const objectUrl = this.buildUrl(key);
|
||||
const response = await this.awsClient.fetch(objectUrl, {
|
||||
method: "PUT",
|
||||
headers: { "Content-Type": contentType },
|
||||
body,
|
||||
});
|
||||
if (!response.ok) {
|
||||
throw new Error(`Failed to upload to object store: ${response.statusText}`);
|
||||
}
|
||||
return objectUrl;
|
||||
}
|
||||
|
||||
async getObject(key: string): Promise<string> {
|
||||
const response = await this.awsClient.fetch(this.buildUrl(key));
|
||||
if (!response.ok) {
|
||||
throw new Error(`Failed to download from object store: ${response.statusText}`);
|
||||
}
|
||||
return response.text();
|
||||
}
|
||||
|
||||
async presign(key: string, method: "PUT" | "GET", expiresIn: number): Promise<string> {
|
||||
const url = new URL(this.config.baseUrl);
|
||||
url.pathname = `/${key}`;
|
||||
url.searchParams.set("X-Amz-Expires", String(expiresIn));
|
||||
|
||||
const signed = await this.awsClient.sign(new Request(url, { method }), {
|
||||
aws: { signQuery: true },
|
||||
});
|
||||
return signed.url;
|
||||
}
|
||||
}
|
||||
|
||||
type AwsSdkConfig = {
|
||||
bucket: string;
|
||||
baseUrl: string;
|
||||
region?: string;
|
||||
};
|
||||
|
||||
class AwsSdkClient implements IObjectStoreClient {
|
||||
private readonly s3Client: S3Client;
|
||||
|
||||
constructor(private readonly config: AwsSdkConfig) {
|
||||
this.s3Client = new S3Client({
|
||||
endpoint: config.baseUrl,
|
||||
forcePathStyle: true,
|
||||
...(config.region ? { region: config.region } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Callers use a single logical key (same as aws4fetch path: `bucket/object/...`).
|
||||
* S3 APIs take Bucket + Key where Key must not repeat the bucket name.
|
||||
*/
|
||||
private toS3ObjectKey(logicalKey: string): string {
|
||||
const prefix = `${this.config.bucket}/`;
|
||||
if (logicalKey.startsWith(prefix)) {
|
||||
return logicalKey.slice(prefix.length);
|
||||
}
|
||||
return logicalKey;
|
||||
}
|
||||
|
||||
private logicalObjectUrl(logicalKey: string): string {
|
||||
const url = new URL(this.config.baseUrl);
|
||||
url.pathname = `/${logicalKey}`;
|
||||
return url.href;
|
||||
}
|
||||
|
||||
async putObject(key: string, body: ReadableStream | string, contentType: string): Promise<string> {
|
||||
const s3Key = this.toS3ObjectKey(key);
|
||||
await this.s3Client.send(
|
||||
new PutObjectCommand({
|
||||
Bucket: this.config.bucket,
|
||||
Key: s3Key,
|
||||
Body: body,
|
||||
ContentType: contentType,
|
||||
})
|
||||
);
|
||||
return this.logicalObjectUrl(key);
|
||||
}
|
||||
|
||||
async getObject(key: string): Promise<string> {
|
||||
const s3Key = this.toS3ObjectKey(key);
|
||||
const response = await this.s3Client.send(
|
||||
new GetObjectCommand({ Bucket: this.config.bucket, Key: s3Key })
|
||||
);
|
||||
if (!response.Body) {
|
||||
throw new Error(`Empty response body from object store for key: ${key}`);
|
||||
}
|
||||
return response.Body.transformToString();
|
||||
}
|
||||
|
||||
async presign(key: string, method: "PUT" | "GET", expiresIn: number): Promise<string> {
|
||||
const s3Key = this.toS3ObjectKey(key);
|
||||
const command =
|
||||
method === "PUT"
|
||||
? new PutObjectCommand({ Bucket: this.config.bucket, Key: s3Key })
|
||||
: new GetObjectCommand({ Bucket: this.config.bucket, Key: s3Key });
|
||||
|
||||
return getSignedUrl(this.s3Client, command, { expiresIn });
|
||||
}
|
||||
}
|
||||
|
||||
export type ObjectStoreClientConfig = {
|
||||
baseUrl: string;
|
||||
bucket?: string;
|
||||
accessKeyId?: string;
|
||||
secretAccessKey?: string;
|
||||
region?: string;
|
||||
service?: string;
|
||||
};
|
||||
|
||||
export class ObjectStoreClient implements IObjectStoreClient {
|
||||
private constructor(
|
||||
private readonly impl: IObjectStoreClient,
|
||||
/** When set, logical keys may start with `${bucket}/…`; AwsSdkClient strips that prefix for S3 APIs. */
|
||||
readonly bucket: string | undefined
|
||||
) {}
|
||||
|
||||
static create(config: ObjectStoreClientConfig): ObjectStoreClient {
|
||||
if (config.accessKeyId && config.secretAccessKey) {
|
||||
return new ObjectStoreClient(
|
||||
new Aws4FetchClient({
|
||||
baseUrl: config.baseUrl,
|
||||
accessKeyId: config.accessKeyId,
|
||||
secretAccessKey: config.secretAccessKey,
|
||||
region: config.region,
|
||||
service: config.service,
|
||||
}),
|
||||
config.bucket
|
||||
);
|
||||
}
|
||||
|
||||
// IAM credential chain — AWS SDK S3Client handles credential refresh automatically
|
||||
if (!config.bucket) {
|
||||
throw new Error(
|
||||
"OBJECT_STORE_BUCKET is required when not using access key credentials (IAM mode)"
|
||||
);
|
||||
}
|
||||
|
||||
return new ObjectStoreClient(
|
||||
new AwsSdkClient({
|
||||
bucket: config.bucket,
|
||||
baseUrl: config.baseUrl,
|
||||
region: config.region,
|
||||
}),
|
||||
config.bucket
|
||||
);
|
||||
}
|
||||
|
||||
putObject(key: string, body: ReadableStream | string, contentType: string): Promise<string> {
|
||||
return this.impl.putObject(key, body, contentType);
|
||||
}
|
||||
|
||||
getObject(key: string): Promise<string> {
|
||||
return this.impl.getObject(key);
|
||||
}
|
||||
|
||||
presign(key: string, method: "PUT" | "GET", expiresIn: number): Promise<string> {
|
||||
return this.impl.presign(key, method, expiresIn);
|
||||
}
|
||||
}
|
||||
@@ -1,240 +0,0 @@
|
||||
import { AwsClient } from "aws4fetch";
|
||||
import { env } from "~/env.server";
|
||||
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import { singleton } from "~/utils/singleton";
|
||||
import { startActiveSpan } from "./tracer.server";
|
||||
import { IOPacket } from "@trigger.dev/core/v3";
|
||||
|
||||
export const r2 = singleton("r2", initializeR2);
|
||||
|
||||
function initializeR2() {
|
||||
if (!env.OBJECT_STORE_ACCESS_KEY_ID || !env.OBJECT_STORE_SECRET_ACCESS_KEY) {
|
||||
return;
|
||||
}
|
||||
|
||||
return new AwsClient({
|
||||
accessKeyId: env.OBJECT_STORE_ACCESS_KEY_ID,
|
||||
secretAccessKey: env.OBJECT_STORE_SECRET_ACCESS_KEY,
|
||||
region: env.OBJECT_STORE_REGION,
|
||||
// We now set the default value to "s3" in the schema to enhance interoperability with various S3-compatible services.
|
||||
// Setting this env var to an empty string will restore the previous behavior of not setting a service.
|
||||
service: env.OBJECT_STORE_SERVICE ? env.OBJECT_STORE_SERVICE : undefined,
|
||||
});
|
||||
}
|
||||
|
||||
export async function uploadPacketToObjectStore(
|
||||
filename: string,
|
||||
data: ReadableStream | string,
|
||||
contentType: string,
|
||||
environment: AuthenticatedEnvironment
|
||||
): Promise<string> {
|
||||
return await startActiveSpan("uploadPacketToObjectStore()", async (span) => {
|
||||
if (!r2) {
|
||||
throw new Error("Object store credentials are not set");
|
||||
}
|
||||
|
||||
if (!env.OBJECT_STORE_BASE_URL) {
|
||||
throw new Error("Object store base URL is not set");
|
||||
}
|
||||
|
||||
span.setAttributes({
|
||||
projectRef: environment.project.externalRef,
|
||||
environmentSlug: environment.slug,
|
||||
filename: filename,
|
||||
});
|
||||
|
||||
const url = new URL(env.OBJECT_STORE_BASE_URL);
|
||||
url.pathname = `/packets/${environment.project.externalRef}/${environment.slug}/${filename}`;
|
||||
|
||||
logger.debug("Uploading to object store", { url: url.href });
|
||||
|
||||
const response = await r2.fetch(url.toString(), {
|
||||
method: "PUT",
|
||||
headers: {
|
||||
"Content-Type": contentType,
|
||||
},
|
||||
body: data,
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Failed to upload output to ${url}: ${response.statusText}`);
|
||||
}
|
||||
|
||||
return url.href;
|
||||
});
|
||||
}
|
||||
|
||||
export async function downloadPacketFromObjectStore(
|
||||
packet: IOPacket,
|
||||
environment: AuthenticatedEnvironment
|
||||
): Promise<IOPacket> {
|
||||
if (packet.dataType !== "application/store") {
|
||||
return packet;
|
||||
}
|
||||
|
||||
return await startActiveSpan("downloadPacketFromObjectStore()", async (span) => {
|
||||
if (!r2) {
|
||||
throw new Error("Object store credentials are not set");
|
||||
}
|
||||
|
||||
if (!env.OBJECT_STORE_BASE_URL) {
|
||||
throw new Error("Object store base URL is not set");
|
||||
}
|
||||
|
||||
span.setAttributes({
|
||||
projectRef: environment.project.externalRef,
|
||||
environmentSlug: environment.slug,
|
||||
filename: packet.data,
|
||||
});
|
||||
|
||||
const url = new URL(env.OBJECT_STORE_BASE_URL);
|
||||
url.pathname = `/packets/${environment.project.externalRef}/${environment.slug}/${packet.data}`;
|
||||
|
||||
logger.debug("Downloading from object store", { url: url.href });
|
||||
|
||||
const response = await r2.fetch(url.toString());
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Failed to download input from ${url}: ${response.statusText}`);
|
||||
}
|
||||
|
||||
const data = await response.text();
|
||||
|
||||
const rawPacket = {
|
||||
data,
|
||||
dataType: "application/json",
|
||||
};
|
||||
|
||||
return rawPacket;
|
||||
});
|
||||
}
|
||||
|
||||
export async function uploadDataToObjectStore(
|
||||
filename: string,
|
||||
data: string,
|
||||
contentType: string,
|
||||
prefix?: string
|
||||
): Promise<string> {
|
||||
return await startActiveSpan("uploadDataToObjectStore()", async (span) => {
|
||||
if (!r2) {
|
||||
throw new Error("Object store credentials are not set");
|
||||
}
|
||||
|
||||
if (!env.OBJECT_STORE_BASE_URL) {
|
||||
throw new Error("Object store base URL is not set");
|
||||
}
|
||||
|
||||
span.setAttributes({
|
||||
prefix,
|
||||
filename,
|
||||
});
|
||||
|
||||
const url = new URL(env.OBJECT_STORE_BASE_URL);
|
||||
url.pathname = `${prefix}/${filename}`;
|
||||
|
||||
logger.debug("Uploading to object store", { url: url.href });
|
||||
|
||||
const response = await r2.fetch(url.toString(), {
|
||||
method: "PUT",
|
||||
headers: {
|
||||
"Content-Type": contentType,
|
||||
},
|
||||
body: data,
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Failed to upload data to ${url}: ${response.statusText}`);
|
||||
}
|
||||
|
||||
return url.href;
|
||||
});
|
||||
}
|
||||
|
||||
export async function generatePresignedRequest(
|
||||
projectRef: string,
|
||||
envSlug: string,
|
||||
filename: string,
|
||||
method: "PUT" | "GET" = "PUT"
|
||||
): Promise<
|
||||
| {
|
||||
success: false;
|
||||
error: string;
|
||||
}
|
||||
| {
|
||||
success: true;
|
||||
request: Request;
|
||||
}
|
||||
> {
|
||||
if (!env.OBJECT_STORE_BASE_URL) {
|
||||
return {
|
||||
success: false,
|
||||
error: "Object store base URL is not set",
|
||||
};
|
||||
}
|
||||
|
||||
if (!r2) {
|
||||
return {
|
||||
success: false,
|
||||
error: "Object store client is not initialized",
|
||||
};
|
||||
}
|
||||
|
||||
const url = new URL(env.OBJECT_STORE_BASE_URL);
|
||||
url.pathname = `/packets/${projectRef}/${envSlug}/${filename}`;
|
||||
url.searchParams.set("X-Amz-Expires", "300"); // 5 minutes
|
||||
|
||||
const signed = await r2.sign(
|
||||
new Request(url, {
|
||||
method,
|
||||
}),
|
||||
{
|
||||
aws: { signQuery: true },
|
||||
}
|
||||
);
|
||||
|
||||
logger.debug("Generated presigned URL", {
|
||||
url: signed.url,
|
||||
headers: Object.fromEntries(signed.headers),
|
||||
projectRef,
|
||||
envSlug,
|
||||
filename,
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
request: signed,
|
||||
};
|
||||
}
|
||||
|
||||
export async function generatePresignedUrl(
|
||||
projectRef: string,
|
||||
envSlug: string,
|
||||
filename: string,
|
||||
method: "PUT" | "GET" = "PUT"
|
||||
): Promise<
|
||||
| {
|
||||
success: false;
|
||||
error: string;
|
||||
}
|
||||
| {
|
||||
success: true;
|
||||
url: string;
|
||||
}
|
||||
> {
|
||||
const signed = await generatePresignedRequest(projectRef, envSlug, filename, method);
|
||||
|
||||
if (!signed.success) {
|
||||
return {
|
||||
success: false,
|
||||
error: signed.error,
|
||||
};
|
||||
}
|
||||
|
||||
signed;
|
||||
|
||||
return {
|
||||
success: true,
|
||||
url: signed.request.url,
|
||||
};
|
||||
}
|
||||
@@ -25,7 +25,7 @@ import { generateFriendlyId } from "../friendlyIdentifiers";
|
||||
import { legacyRunEngineWorker } from "../legacyRunEngineWorker.server";
|
||||
import { marqs } from "../marqs/index.server";
|
||||
import { guardQueueSizeLimitsForEnv } from "../queueSizeLimits.server";
|
||||
import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../r2.server";
|
||||
import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../objectStore.server";
|
||||
import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus";
|
||||
import { startActiveSpan } from "../tracer.server";
|
||||
import { BaseService, ServiceValidationError } from "./baseService.server";
|
||||
@@ -930,10 +930,10 @@ export class BatchTriggerV3Service extends BaseService {
|
||||
|
||||
const filename = `${pathPrefix}/payload.json`;
|
||||
|
||||
await uploadPacketToObjectStore(filename, packet.data, packet.dataType, environment);
|
||||
const uploadedFilename = await uploadPacketToObjectStore(filename, packet.data, packet.dataType, environment);
|
||||
|
||||
return {
|
||||
data: filename,
|
||||
data: uploadedFilename,
|
||||
dataType: "application/store",
|
||||
};
|
||||
});
|
||||
|
||||
@@ -26,7 +26,7 @@ import { getV3EventRepository } from "../eventRepository/index.server";
|
||||
import { generateFriendlyId } from "../friendlyIdentifiers";
|
||||
import { findCurrentWorkerFromEnvironment } from "../models/workerDeployment.server";
|
||||
import { guardQueueSizeLimitsForEnv } from "../queueSizeLimits.server";
|
||||
import { uploadPacketToObjectStore } from "../r2.server";
|
||||
import { uploadPacketToObjectStore } from "../objectStore.server";
|
||||
import { removeQueueConcurrencyLimits, updateQueueConcurrencyLimits } from "../runQueue.server";
|
||||
import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus";
|
||||
import { startActiveSpan } from "../tracer.server";
|
||||
@@ -756,10 +756,10 @@ export class TriggerTaskServiceV1 extends BaseService {
|
||||
|
||||
const filename = `${pathPrefix}/payload.json`;
|
||||
|
||||
await uploadPacketToObjectStore(filename, packet.data, packet.dataType, environment);
|
||||
const uploadedFilename = await uploadPacketToObjectStore(filename, packet.data, packet.dataType, environment);
|
||||
|
||||
return {
|
||||
data: filename,
|
||||
data: uploadedFilename,
|
||||
dataType: "application/store",
|
||||
};
|
||||
});
|
||||
|
||||
@@ -0,0 +1,649 @@
|
||||
import { postgresAndMinioTest } from "@internal/testcontainers";
|
||||
import { type IOPacket } from "@trigger.dev/core/v3";
|
||||
import { type PrismaClient } from "@trigger.dev/database";
|
||||
import { afterAll, describe, expect, it, vi } from "vitest";
|
||||
import { env } from "~/env.server";
|
||||
import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
|
||||
import {
|
||||
downloadPacketFromObjectStore,
|
||||
formatStorageUri,
|
||||
generatePresignedRequest,
|
||||
generatePresignedUrl,
|
||||
hasObjectStoreClient,
|
||||
parseStorageUri,
|
||||
resolveStoreProtocolForPacketPresign,
|
||||
uploadPacketToObjectStore,
|
||||
} from "~/v3/objectStore.server";
|
||||
|
||||
// Extend the timeout for container tests
|
||||
vi.setConfig({ testTimeout: 60_000 });
|
||||
|
||||
// Helper to create a test environment
|
||||
async function createTestEnvironment(prisma: PrismaClient) {
|
||||
const suffix = Date.now().toString(36);
|
||||
|
||||
const org = await prisma.organization.create({
|
||||
data: {
|
||||
title: `Test Org ${suffix}`,
|
||||
slug: `test-org-${suffix}`,
|
||||
},
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: `Test Project ${suffix}`,
|
||||
slug: `test-project-${suffix}`,
|
||||
externalRef: `proj_test${suffix}`,
|
||||
organizationId: org.id,
|
||||
},
|
||||
});
|
||||
|
||||
const environment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "dev",
|
||||
type: "DEVELOPMENT",
|
||||
organizationId: org.id,
|
||||
projectId: project.id,
|
||||
apiKey: `test_key_${suffix}`,
|
||||
pkApiKey: `test_pk_key_${suffix}`,
|
||||
shortcode: suffix.slice(0, 4),
|
||||
},
|
||||
include: {
|
||||
project: true,
|
||||
organization: true,
|
||||
},
|
||||
});
|
||||
|
||||
return environment;
|
||||
}
|
||||
|
||||
// Save original env values for restoration in afterAll
|
||||
const originalEnv = process.env;
|
||||
const originalEnvObj = {
|
||||
OBJECT_STORE_BASE_URL: env.OBJECT_STORE_BASE_URL,
|
||||
OBJECT_STORE_BUCKET: env.OBJECT_STORE_BUCKET,
|
||||
OBJECT_STORE_ACCESS_KEY_ID: env.OBJECT_STORE_ACCESS_KEY_ID,
|
||||
OBJECT_STORE_SECRET_ACCESS_KEY: env.OBJECT_STORE_SECRET_ACCESS_KEY,
|
||||
OBJECT_STORE_REGION: env.OBJECT_STORE_REGION,
|
||||
OBJECT_STORE_DEFAULT_PROTOCOL: env.OBJECT_STORE_DEFAULT_PROTOCOL,
|
||||
TASK_PAYLOAD_OFFLOAD_THRESHOLD: env.TASK_PAYLOAD_OFFLOAD_THRESHOLD,
|
||||
};
|
||||
|
||||
describe("Object Storage", () => {
|
||||
describe("URI parsing functions", () => {
|
||||
it("should parse URI with protocol", () => {
|
||||
const result = parseStorageUri("s3://run_abc123/payload.json");
|
||||
expect(result).toEqual({
|
||||
protocol: "s3",
|
||||
path: "run_abc123/payload.json",
|
||||
});
|
||||
});
|
||||
|
||||
it("should parse URI with R2 protocol", () => {
|
||||
const result = parseStorageUri("r2://batch_xyz/item_0/payload.json");
|
||||
expect(result).toEqual({
|
||||
protocol: "r2",
|
||||
path: "batch_xyz/item_0/payload.json",
|
||||
});
|
||||
});
|
||||
|
||||
it("should parse legacy URI without protocol", () => {
|
||||
const result = parseStorageUri("run_abc123/payload.json");
|
||||
expect(result).toEqual({
|
||||
protocol: undefined,
|
||||
path: "run_abc123/payload.json",
|
||||
});
|
||||
});
|
||||
|
||||
it("should format URI with protocol", () => {
|
||||
const result = formatStorageUri("run_abc123/payload.json", "s3");
|
||||
expect(result).toBe("s3://run_abc123/payload.json");
|
||||
});
|
||||
|
||||
it("should format URI without protocol", () => {
|
||||
const result = formatStorageUri("run_abc123/payload.json");
|
||||
expect(result).toBe("run_abc123/payload.json");
|
||||
});
|
||||
});
|
||||
|
||||
describe("resolveStoreProtocolForPacketPresign", () => {
|
||||
afterEach(() => {
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = originalEnvObj.OBJECT_STORE_DEFAULT_PROTOCOL;
|
||||
});
|
||||
|
||||
it("GET uses legacy default for unprefixed keys even when OBJECT_STORE_DEFAULT_PROTOCOL is set", () => {
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = "s3";
|
||||
expect(resolveStoreProtocolForPacketPresign("a/b.json", "GET").storeProtocol).toBeUndefined();
|
||||
});
|
||||
|
||||
it("PUT without forceNoPrefix uses OBJECT_STORE_DEFAULT_PROTOCOL for unprefixed keys", () => {
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = "s3";
|
||||
expect(resolveStoreProtocolForPacketPresign("a/b.json", "PUT", false).storeProtocol).toBe(
|
||||
"s3"
|
||||
);
|
||||
});
|
||||
|
||||
it("PUT with forceNoPrefix skips OBJECT_STORE_DEFAULT_PROTOCOL for unprefixed keys", () => {
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = "s3";
|
||||
expect(resolveStoreProtocolForPacketPresign("a/b.json", "PUT", true).storeProtocol).toBeUndefined();
|
||||
});
|
||||
|
||||
it("explicit protocol in key wins for PUT with forceNoPrefix", () => {
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = "r2";
|
||||
expect(resolveStoreProtocolForPacketPresign("s3://x/y.json", "PUT", true).storeProtocol).toBe(
|
||||
"s3"
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
postgresAndMinioTest(
|
||||
"should upload and download data without protocol (legacy)",
|
||||
async ({ minioConfig, prisma }) => {
|
||||
// Override env directly for the default provider
|
||||
env.OBJECT_STORE_BASE_URL = minioConfig.baseUrl;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
env.OBJECT_STORE_REGION = minioConfig.region;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
|
||||
const environment = await createTestEnvironment(prisma);
|
||||
|
||||
const testData = JSON.stringify({ test: "data", value: 123 });
|
||||
const filename = "test_run/payload.json";
|
||||
|
||||
// Upload
|
||||
const uploadedFilename = await uploadPacketToObjectStore(
|
||||
filename,
|
||||
testData,
|
||||
"application/json",
|
||||
environment as any
|
||||
);
|
||||
|
||||
// Should return filename without protocol (legacy)
|
||||
expect(uploadedFilename).toBe(filename);
|
||||
|
||||
// Download
|
||||
const packet: IOPacket = {
|
||||
data: uploadedFilename,
|
||||
dataType: "application/store",
|
||||
};
|
||||
|
||||
const downloadedPacket = await downloadPacketFromObjectStore(packet, environment as any);
|
||||
|
||||
expect(downloadedPacket.dataType).toBe("application/json");
|
||||
expect(downloadedPacket.data).toBe(testData);
|
||||
|
||||
// Cleanup
|
||||
await prisma.runtimeEnvironment.delete({ where: { id: environment.id } });
|
||||
await prisma.project.delete({ where: { id: environment.projectId } });
|
||||
await prisma.organization.delete({ where: { id: environment.organizationId } });
|
||||
}
|
||||
);
|
||||
|
||||
postgresAndMinioTest(
|
||||
"should upload and download data with protocol prefix",
|
||||
async ({ minioConfig, prisma }) => {
|
||||
// Named provider — controlled via process.env (read dynamically)
|
||||
process.env.OBJECT_STORE_S3_BASE_URL = minioConfig.baseUrl;
|
||||
process.env.OBJECT_STORE_S3_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
process.env.OBJECT_STORE_S3_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
process.env.OBJECT_STORE_S3_REGION = minioConfig.region;
|
||||
process.env.OBJECT_STORE_S3_SERVICE = "s3";
|
||||
|
||||
const environment = await createTestEnvironment(prisma);
|
||||
|
||||
const testData = JSON.stringify({ test: "protocol-data", value: 456 });
|
||||
const filename = "test_run2/payload.json";
|
||||
|
||||
// Upload with explicit protocol — bypasses env.OBJECT_STORE_DEFAULT_PROTOCOL
|
||||
const uploadedFilename = await uploadPacketToObjectStore(
|
||||
filename,
|
||||
testData,
|
||||
"application/json",
|
||||
environment as any,
|
||||
"s3"
|
||||
);
|
||||
|
||||
// Should return filename with s3:// protocol
|
||||
expect(uploadedFilename).toBe("s3://test_run2/payload.json");
|
||||
|
||||
// Download
|
||||
const packet: IOPacket = {
|
||||
data: uploadedFilename,
|
||||
dataType: "application/store",
|
||||
};
|
||||
|
||||
const downloadedPacket = await downloadPacketFromObjectStore(packet, environment as any);
|
||||
|
||||
expect(downloadedPacket.dataType).toBe("application/json");
|
||||
expect(downloadedPacket.data).toBe(testData);
|
||||
|
||||
// Cleanup
|
||||
await prisma.runtimeEnvironment.delete({ where: { id: environment.id } });
|
||||
await prisma.project.delete({ where: { id: environment.projectId } });
|
||||
await prisma.organization.delete({ where: { id: environment.organizationId } });
|
||||
}
|
||||
);
|
||||
|
||||
postgresAndMinioTest(
|
||||
"should support migration from default provider to named provider",
|
||||
async ({ minioConfig, prisma }) => {
|
||||
const environment = await createTestEnvironment(prisma);
|
||||
|
||||
// Step 1: Upload old data without protocol (using default provider)
|
||||
env.OBJECT_STORE_BASE_URL = minioConfig.baseUrl;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
env.OBJECT_STORE_REGION = minioConfig.region;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
|
||||
const oldData = JSON.stringify({ legacy: true });
|
||||
const oldFilename = "old_run/payload.json";
|
||||
|
||||
const uploadedOldFilename = await uploadPacketToObjectStore(
|
||||
oldFilename,
|
||||
oldData,
|
||||
"application/json",
|
||||
environment as any
|
||||
);
|
||||
|
||||
expect(uploadedOldFilename).toBe(oldFilename); // No protocol
|
||||
|
||||
// Step 2: Configure new provider (S3) and set default protocol
|
||||
process.env.OBJECT_STORE_S3_BASE_URL = minioConfig.baseUrl; // Same MinIO for testing
|
||||
process.env.OBJECT_STORE_S3_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
process.env.OBJECT_STORE_S3_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
process.env.OBJECT_STORE_S3_REGION = minioConfig.region;
|
||||
process.env.OBJECT_STORE_S3_SERVICE = "s3";
|
||||
|
||||
// Step 3: Upload new data with explicit protocol
|
||||
const newData = JSON.stringify({ new: true });
|
||||
const newFilename = "new_run/payload.json";
|
||||
|
||||
const uploadedNewFilename = await uploadPacketToObjectStore(
|
||||
newFilename,
|
||||
newData,
|
||||
"application/json",
|
||||
environment as any,
|
||||
"s3"
|
||||
);
|
||||
|
||||
expect(uploadedNewFilename).toBe("s3://new_run/payload.json"); // Has protocol
|
||||
|
||||
// Step 4: Verify both can be downloaded
|
||||
// Old data (no protocol, uses default provider)
|
||||
const oldPacket: IOPacket = {
|
||||
data: uploadedOldFilename,
|
||||
dataType: "application/store",
|
||||
};
|
||||
const downloadedOld = await downloadPacketFromObjectStore(oldPacket, environment as any);
|
||||
expect(downloadedOld.data).toBe(oldData);
|
||||
|
||||
// New data (with protocol, uses named provider)
|
||||
const newPacket: IOPacket = {
|
||||
data: uploadedNewFilename,
|
||||
dataType: "application/store",
|
||||
};
|
||||
const downloadedNew = await downloadPacketFromObjectStore(newPacket, environment as any);
|
||||
expect(downloadedNew.data).toBe(newData);
|
||||
|
||||
// Cleanup
|
||||
await prisma.runtimeEnvironment.delete({ where: { id: environment.id } });
|
||||
await prisma.project.delete({ where: { id: environment.projectId } });
|
||||
await prisma.organization.delete({ where: { id: environment.organizationId } });
|
||||
}
|
||||
);
|
||||
|
||||
postgresAndMinioTest(
|
||||
"should upload and download using IAM credential chain (AWS SDK path)",
|
||||
async ({ minioConfig, prisma }) => {
|
||||
// IAM mode: override env with bucket but no access keys.
|
||||
// We put the credentials in AWS_* env vars so the S3Client credential
|
||||
// chain picks them up (same as it would from an ECS task role in production).
|
||||
env.OBJECT_STORE_BASE_URL = minioConfig.baseUrl;
|
||||
env.OBJECT_STORE_BUCKET = "packets";
|
||||
env.OBJECT_STORE_REGION = minioConfig.region;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = undefined;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = undefined;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
|
||||
process.env.AWS_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
process.env.AWS_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
process.env.AWS_REGION = minioConfig.region;
|
||||
|
||||
const environment = await createTestEnvironment(prisma);
|
||||
|
||||
const testData = JSON.stringify({ iam: true, value: 789 });
|
||||
const filename = "iam_test_run/payload.json";
|
||||
|
||||
const uploadedFilename = await uploadPacketToObjectStore(
|
||||
filename,
|
||||
testData,
|
||||
"application/json",
|
||||
environment as any
|
||||
);
|
||||
|
||||
expect(uploadedFilename).toBe(filename);
|
||||
|
||||
const packet: IOPacket = {
|
||||
data: uploadedFilename,
|
||||
dataType: "application/store",
|
||||
};
|
||||
|
||||
const downloadedPacket = await downloadPacketFromObjectStore(packet, environment as any);
|
||||
|
||||
expect(downloadedPacket.dataType).toBe("application/json");
|
||||
expect(downloadedPacket.data).toBe(testData);
|
||||
|
||||
// Cleanup
|
||||
delete process.env.AWS_ACCESS_KEY_ID;
|
||||
delete process.env.AWS_SECRET_ACCESS_KEY;
|
||||
delete process.env.AWS_REGION;
|
||||
|
||||
await prisma.runtimeEnvironment.delete({ where: { id: environment.id } });
|
||||
await prisma.project.delete({ where: { id: environment.projectId } });
|
||||
await prisma.organization.delete({ where: { id: environment.organizationId } });
|
||||
}
|
||||
);
|
||||
|
||||
postgresAndMinioTest(
|
||||
"processWaitpointCompletionPacket offloads above threshold and round-trips download",
|
||||
async ({ minioConfig, prisma }) => {
|
||||
env.OBJECT_STORE_BASE_URL = minioConfig.baseUrl;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
env.OBJECT_STORE_REGION = minioConfig.region;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
|
||||
const savedThreshold = env.TASK_PAYLOAD_OFFLOAD_THRESHOLD;
|
||||
env.TASK_PAYLOAD_OFFLOAD_THRESHOLD = 256;
|
||||
|
||||
const environment = await createTestEnvironment(prisma);
|
||||
const pathPrefix = `waitpoint_completiontest/token`;
|
||||
|
||||
try {
|
||||
const smallPacket: IOPacket = {
|
||||
data: JSON.stringify({ ok: true }),
|
||||
dataType: "application/json",
|
||||
};
|
||||
const smallResult = await processWaitpointCompletionPacket(
|
||||
smallPacket,
|
||||
environment as any,
|
||||
pathPrefix
|
||||
);
|
||||
expect(smallResult).toEqual(smallPacket);
|
||||
|
||||
const largeBody = "x".repeat(400);
|
||||
const largePacket: IOPacket = {
|
||||
data: largeBody,
|
||||
dataType: "text/plain",
|
||||
};
|
||||
const largeResult = await processWaitpointCompletionPacket(
|
||||
largePacket,
|
||||
environment as any,
|
||||
pathPrefix
|
||||
);
|
||||
|
||||
expect(largeResult.dataType).toBe("application/store");
|
||||
expect(largeResult.data).toBe(`${pathPrefix}.txt`);
|
||||
|
||||
const downloadedPacket = await downloadPacketFromObjectStore(
|
||||
{
|
||||
data: largeResult.data,
|
||||
dataType: "application/store",
|
||||
},
|
||||
environment as any
|
||||
);
|
||||
|
||||
expect(downloadedPacket.data).toBe(largeBody);
|
||||
} finally {
|
||||
env.TASK_PAYLOAD_OFFLOAD_THRESHOLD = savedThreshold;
|
||||
|
||||
await prisma.runtimeEnvironment.delete({ where: { id: environment.id } });
|
||||
await prisma.project.delete({ where: { id: environment.projectId } });
|
||||
await prisma.organization.delete({ where: { id: environment.organizationId } });
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
describe("hasObjectStoreClient", () => {
|
||||
it("returns false when no store is configured", () => {
|
||||
env.OBJECT_STORE_BASE_URL = undefined;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
delete process.env.OBJECT_STORE_NOTCONFIGURED_BASE_URL;
|
||||
expect(hasObjectStoreClient()).toBe(false);
|
||||
});
|
||||
|
||||
it("returns true when default provider base URL is set", () => {
|
||||
env.OBJECT_STORE_BASE_URL = "http://localhost:9000";
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
expect(hasObjectStoreClient()).toBe(true);
|
||||
env.OBJECT_STORE_BASE_URL = undefined;
|
||||
});
|
||||
|
||||
it("returns true when named protocol base URL is set", () => {
|
||||
env.OBJECT_STORE_BASE_URL = undefined;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = "s3";
|
||||
process.env.OBJECT_STORE_S3_BASE_URL = "http://localhost:9000";
|
||||
expect(hasObjectStoreClient()).toBe(true);
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
delete process.env.OBJECT_STORE_S3_BASE_URL;
|
||||
});
|
||||
});
|
||||
|
||||
postgresAndMinioTest(
|
||||
"generatePresignedUrl - PUT then GET round-trip (static credentials / aws4fetch path)",
|
||||
async ({ minioConfig }) => {
|
||||
env.OBJECT_STORE_BASE_URL = minioConfig.baseUrl;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
env.OBJECT_STORE_REGION = minioConfig.region;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
|
||||
const projectRef = "proj_presign_test";
|
||||
const envSlug = "dev";
|
||||
const filename = "presigned-static/payload.json";
|
||||
const data = JSON.stringify({ presigned: "static" });
|
||||
|
||||
// Upload via presigned PUT
|
||||
const putResult = await generatePresignedUrl(projectRef, envSlug, filename, "PUT");
|
||||
expect(putResult.success).toBe(true);
|
||||
if (!putResult.success) throw new Error(putResult.error);
|
||||
expect(putResult.storagePath).toBe(filename);
|
||||
|
||||
const putResponse = await fetch(putResult.url, {
|
||||
method: "PUT",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: data,
|
||||
});
|
||||
expect(putResponse.ok).toBe(true);
|
||||
|
||||
// Download via presigned GET
|
||||
const getResult = await generatePresignedUrl(projectRef, envSlug, filename, "GET");
|
||||
expect(getResult.success).toBe(true);
|
||||
if (!getResult.success) throw new Error(getResult.error);
|
||||
expect(getResult.storagePath).toBeUndefined();
|
||||
|
||||
const getResponse = await fetch(getResult.url);
|
||||
expect(getResponse.ok).toBe(true);
|
||||
expect(await getResponse.text()).toBe(data);
|
||||
}
|
||||
);
|
||||
|
||||
postgresAndMinioTest(
|
||||
"generatePresignedUrl - PUT unprefixed with OBJECT_STORE_DEFAULT_PROTOCOL returns s3 storagePath",
|
||||
async ({ minioConfig }) => {
|
||||
env.OBJECT_STORE_BASE_URL = undefined;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = undefined;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = undefined;
|
||||
env.OBJECT_STORE_REGION = undefined;
|
||||
|
||||
process.env.OBJECT_STORE_S3_BASE_URL = minioConfig.baseUrl;
|
||||
process.env.OBJECT_STORE_S3_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
process.env.OBJECT_STORE_S3_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
process.env.OBJECT_STORE_S3_REGION = minioConfig.region;
|
||||
process.env.OBJECT_STORE_S3_SERVICE = "s3";
|
||||
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = "s3";
|
||||
|
||||
const projectRef = "proj_presign_v2_style";
|
||||
const envSlug = "dev";
|
||||
const filename = "v2-style/payload.json";
|
||||
const data = JSON.stringify({ v2: true });
|
||||
|
||||
const putResult = await generatePresignedUrl(projectRef, envSlug, filename, "PUT");
|
||||
expect(putResult.success).toBe(true);
|
||||
if (!putResult.success) throw new Error(putResult.error);
|
||||
expect(putResult.storagePath).toBe(`s3://${filename}`);
|
||||
|
||||
const putResponse = await fetch(putResult.url, {
|
||||
method: "PUT",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: data,
|
||||
});
|
||||
expect(putResponse.ok).toBe(true);
|
||||
|
||||
const getResult = await generatePresignedUrl(projectRef, envSlug, putResult.storagePath!, "GET");
|
||||
expect(getResult.success).toBe(true);
|
||||
if (!getResult.success) throw new Error(getResult.error);
|
||||
|
||||
const getResponse = await fetch(getResult.url);
|
||||
expect(getResponse.ok).toBe(true);
|
||||
expect(await getResponse.text()).toBe(data);
|
||||
|
||||
delete process.env.OBJECT_STORE_S3_BASE_URL;
|
||||
delete process.env.OBJECT_STORE_S3_ACCESS_KEY_ID;
|
||||
delete process.env.OBJECT_STORE_S3_SECRET_ACCESS_KEY;
|
||||
delete process.env.OBJECT_STORE_S3_REGION;
|
||||
delete process.env.OBJECT_STORE_S3_SERVICE;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
}
|
||||
);
|
||||
|
||||
postgresAndMinioTest(
|
||||
"generatePresignedUrl - forceNoPrefix PUT fails without legacy default when only S3 named",
|
||||
async ({ minioConfig }) => {
|
||||
env.OBJECT_STORE_BASE_URL = undefined;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = undefined;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = undefined;
|
||||
env.OBJECT_STORE_REGION = undefined;
|
||||
|
||||
process.env.OBJECT_STORE_S3_BASE_URL = minioConfig.baseUrl;
|
||||
process.env.OBJECT_STORE_S3_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
process.env.OBJECT_STORE_S3_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
process.env.OBJECT_STORE_S3_REGION = minioConfig.region;
|
||||
process.env.OBJECT_STORE_S3_SERVICE = "s3";
|
||||
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = "s3";
|
||||
|
||||
const putLegacy = await generatePresignedUrl(
|
||||
"proj_force_noprefix",
|
||||
"dev",
|
||||
"only-legacy/payload.json",
|
||||
"PUT",
|
||||
{ forceNoPrefix: true }
|
||||
);
|
||||
expect(putLegacy.success).toBe(false);
|
||||
|
||||
const getUnprefixed = await generatePresignedUrl(
|
||||
"proj_force_noprefix",
|
||||
"dev",
|
||||
"any.json",
|
||||
"GET"
|
||||
);
|
||||
expect(getUnprefixed.success).toBe(false);
|
||||
|
||||
delete process.env.OBJECT_STORE_S3_BASE_URL;
|
||||
delete process.env.OBJECT_STORE_S3_ACCESS_KEY_ID;
|
||||
delete process.env.OBJECT_STORE_S3_SECRET_ACCESS_KEY;
|
||||
delete process.env.OBJECT_STORE_S3_REGION;
|
||||
delete process.env.OBJECT_STORE_S3_SERVICE;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
}
|
||||
);
|
||||
|
||||
postgresAndMinioTest(
|
||||
"generatePresignedUrl - PUT then GET round-trip (IAM credential chain / AWS SDK path)",
|
||||
async ({ minioConfig }) => {
|
||||
env.OBJECT_STORE_BASE_URL = minioConfig.baseUrl;
|
||||
env.OBJECT_STORE_BUCKET = "packets";
|
||||
env.OBJECT_STORE_REGION = minioConfig.region;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = undefined;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = undefined;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
|
||||
process.env.AWS_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
process.env.AWS_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
process.env.AWS_REGION = minioConfig.region;
|
||||
|
||||
const projectRef = "proj_presign_iam";
|
||||
const envSlug = "dev";
|
||||
const filename = "presigned-iam/payload.json";
|
||||
const data = JSON.stringify({ presigned: "iam" });
|
||||
|
||||
// Upload via presigned PUT
|
||||
const putResult = await generatePresignedUrl(projectRef, envSlug, filename, "PUT");
|
||||
expect(putResult.success).toBe(true);
|
||||
if (!putResult.success) throw new Error(putResult.error);
|
||||
expect(putResult.storagePath).toBe(filename);
|
||||
|
||||
const putResponse = await fetch(putResult.url, {
|
||||
method: "PUT",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: data,
|
||||
});
|
||||
expect(putResponse.ok).toBe(true);
|
||||
|
||||
// Download via presigned GET
|
||||
const getResult = await generatePresignedUrl(projectRef, envSlug, filename, "GET");
|
||||
expect(getResult.success).toBe(true);
|
||||
if (!getResult.success) throw new Error(getResult.error);
|
||||
|
||||
const getResponse = await fetch(getResult.url);
|
||||
expect(getResponse.ok).toBe(true);
|
||||
expect(await getResponse.text()).toBe(data);
|
||||
|
||||
delete process.env.AWS_ACCESS_KEY_ID;
|
||||
delete process.env.AWS_SECRET_ACCESS_KEY;
|
||||
delete process.env.AWS_REGION;
|
||||
}
|
||||
);
|
||||
|
||||
postgresAndMinioTest(
|
||||
"generatePresignedRequest - returns a signed Request object",
|
||||
async ({ minioConfig }) => {
|
||||
env.OBJECT_STORE_BASE_URL = minioConfig.baseUrl;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = minioConfig.accessKeyId;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = minioConfig.secretAccessKey;
|
||||
env.OBJECT_STORE_REGION = minioConfig.region;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = undefined;
|
||||
|
||||
const result = await generatePresignedRequest(
|
||||
"proj_req_test",
|
||||
"dev",
|
||||
"req-test/file.json",
|
||||
"GET"
|
||||
);
|
||||
|
||||
expect(result.success).toBe(true);
|
||||
if (!result.success) throw new Error(result.error);
|
||||
|
||||
// URL should point at the right key and contain SigV4 query params
|
||||
expect(result.request.url).toContain("packets/proj_req_test/dev/req-test/file.json");
|
||||
expect(result.request.url).toContain("X-Amz-");
|
||||
expect(result.request.method).toBe("GET");
|
||||
}
|
||||
);
|
||||
|
||||
// Restore env after all tests
|
||||
afterAll(() => {
|
||||
process.env = originalEnv;
|
||||
env.OBJECT_STORE_BASE_URL = originalEnvObj.OBJECT_STORE_BASE_URL;
|
||||
env.OBJECT_STORE_BUCKET = originalEnvObj.OBJECT_STORE_BUCKET;
|
||||
env.OBJECT_STORE_ACCESS_KEY_ID = originalEnvObj.OBJECT_STORE_ACCESS_KEY_ID;
|
||||
env.OBJECT_STORE_SECRET_ACCESS_KEY = originalEnvObj.OBJECT_STORE_SECRET_ACCESS_KEY;
|
||||
env.OBJECT_STORE_REGION = originalEnvObj.OBJECT_STORE_REGION;
|
||||
env.OBJECT_STORE_DEFAULT_PROTOCOL = originalEnvObj.OBJECT_STORE_DEFAULT_PROTOCOL;
|
||||
env.TASK_PAYLOAD_OFFLOAD_THRESHOLD = originalEnvObj.TASK_PAYLOAD_OFFLOAD_THRESHOLD;
|
||||
});
|
||||
});
|
||||
@@ -4,6 +4,7 @@ volumes:
|
||||
database-data:
|
||||
database-data-alt:
|
||||
redis-data:
|
||||
minio-data:
|
||||
clickhouse-data:
|
||||
clickhouse-logs:
|
||||
prometheus-data:
|
||||
@@ -49,6 +50,45 @@ services:
|
||||
ports:
|
||||
- 6379:6379
|
||||
|
||||
# S3-compatible API for local object store (large payloads / packet offload).
|
||||
# Host :9005 = S3 API, :9006 = web console (ClickHouse uses host :9000).
|
||||
minio:
|
||||
container_name: minio
|
||||
image: minio/minio:latest
|
||||
restart: always
|
||||
command: server /data --console-address ":9001"
|
||||
environment:
|
||||
MINIO_ROOT_USER: minioadmin
|
||||
MINIO_ROOT_PASSWORD: minioadmin
|
||||
volumes:
|
||||
- minio-data:/data
|
||||
ports:
|
||||
- "9005:9000"
|
||||
- "9006:9001"
|
||||
networks:
|
||||
- app_network
|
||||
healthcheck:
|
||||
test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
|
||||
interval: 5s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 5s
|
||||
|
||||
minio-init:
|
||||
image: minio/mc:latest
|
||||
depends_on:
|
||||
minio:
|
||||
condition: service_healthy
|
||||
networks:
|
||||
- app_network
|
||||
entrypoint: /bin/sh
|
||||
command:
|
||||
- -c
|
||||
- |
|
||||
mc alias set local http://minio:9000 minioadmin minioadmin
|
||||
mc mb -p local/packets || true
|
||||
restart: "no"
|
||||
|
||||
electric:
|
||||
container_name: electric
|
||||
image: electricsql/electric:1.2.4@sha256:20da3d0b0e74926c5623392db67fd56698b9e374c4aeb6cb5cadeb8fea171c36
|
||||
|
||||
Vendored
+247
-141
@@ -5,145 +5,251 @@ sidebarTitle: "Webapp"
|
||||
mode: "wide"
|
||||
---
|
||||
|
||||
| Name | Required | Default | Description |
|
||||
| :----------------------------------------------- | :------- | :-------------------- | :-------------------------------------------------------------------------------------------------------- |
|
||||
| **Secrets** | | | |
|
||||
| `SESSION_SECRET` | Yes | — | Session encryption secret. Run: `openssl rand -hex 16` |
|
||||
| `MAGIC_LINK_SECRET` | Yes | — | Magic link encryption secret. Run: `openssl rand -hex 16` |
|
||||
| `ENCRYPTION_KEY` | Yes | — | Secret store encryption key. Run: `openssl rand -hex 16` |
|
||||
| `MANAGED_WORKER_SECRET` | No | managed-secret | Managed worker secret. Should be changed and match supervisor. |
|
||||
| **Domains & ports** | | | |
|
||||
| `REMIX_APP_PORT` | No | 3030 | Remix app port. |
|
||||
| `APP_ORIGIN` | Yes | http://localhost:3030 | App origin URL. |
|
||||
| `LOGIN_ORIGIN` | Yes | http://localhost:3030 | Login origin URL. Most likely the same as `APP_ORIGIN`. |
|
||||
| `API_ORIGIN` | No | `APP_ORIGIN` | API origin URL. |
|
||||
| `STREAM_ORIGIN` | No | `APP_ORIGIN` | Realtime stream origin URL. |
|
||||
| `ELECTRIC_ORIGIN` | No | http://localhost:3060 | Electric origin URL. |
|
||||
| **Postgres** | | | |
|
||||
| `DATABASE_URL` | Yes | — | PostgreSQL connection string. |
|
||||
| `DIRECT_URL` | Yes | — | Direct DB connection string used for migrations etc. |
|
||||
| `DATABASE_CONNECTION_LIMIT` | No | 10 | Max DB connections. |
|
||||
| `DATABASE_POOL_TIMEOUT` | No | 60 | DB pool timeout (s). |
|
||||
| `DATABASE_CONNECTION_TIMEOUT` | No | 20 | DB connect timeout (s). |
|
||||
| `DATABASE_READ_REPLICA_URL` | No | `DATABASE_URL` | Read-replica DB string. |
|
||||
| **Redis** | | | |
|
||||
| `REDIS_HOST` | Yes | — | Redis host. |
|
||||
| `REDIS_PORT` | Yes | — | Redis port. |
|
||||
| `REDIS_READER_HOST` | No | `REDIS_HOST` | Redis reader host. |
|
||||
| `REDIS_READER_PORT` | No | `REDIS_PORT` | Redis reader port. |
|
||||
| `REDIS_USERNAME` | No | — | Redis username. |
|
||||
| `REDIS_PASSWORD` | No | — | Redis password. |
|
||||
| `REDIS_TLS_DISABLED` | No | — | Disable Redis TLS. |
|
||||
| **Auth** | | | |
|
||||
| `WHITELISTED_EMAILS` | No | — | Whitelisted emails regex. |
|
||||
| `AUTH_GITHUB_CLIENT_ID` | No | — | GitHub client ID. |
|
||||
| `AUTH_GITHUB_CLIENT_SECRET` | No | — | GitHub client secret. |
|
||||
| **Email** | | | |
|
||||
| `EMAIL_TRANSPORT` | No | — | Email transport type. One of `resend`, `smtp`, `aws-ses`. |
|
||||
| `FROM_EMAIL` | No | — | From email address. |
|
||||
| `REPLY_TO_EMAIL` | No | — | Reply-to email address. |
|
||||
| `RESEND_API_KEY` | No | — | Resend API key. |
|
||||
| `SMTP_HOST` | No | — | SMTP host. |
|
||||
| `SMTP_PORT` | No | — | SMTP port. |
|
||||
| `SMTP_SECURE` | No | — | SMTP secure flag. |
|
||||
| `SMTP_USER` | No | — | SMTP user. |
|
||||
| `SMTP_PASSWORD` | No | — | SMTP password. |
|
||||
| `AWS_REGION` | No | — | AWS region for SES. |
|
||||
| `AWS_ACCESS_KEY_ID` | No | — | AWS access key ID for SES. |
|
||||
| `AWS_SECRET_ACCESS_KEY` | No | — | AWS secret access key for SES. |
|
||||
| **Graphile & Redis worker** | | | |
|
||||
| `WORKER_CONCURRENCY` | No | 10 | Redis worker concurrency. |
|
||||
| `WORKER_POLL_INTERVAL` | No | 1000 | Redis worker poll interval (ms). |
|
||||
| `WORKER_SCHEMA` | No | graphile_worker | Graphile worker schema. |
|
||||
| `GRACEFUL_SHUTDOWN_TIMEOUT` | No | 60000 (1m) | Graphile graceful shutdown timeout (ms). Affects shutdown time. |
|
||||
| **Concurrency limits** | | | |
|
||||
| `DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT` | No | 100 | Default env execution concurrency. |
|
||||
| `DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT` | No | 300 | Default org execution concurrency, needs to be 3x env concurrency. |
|
||||
| **Dev** | | | |
|
||||
| `DEV_MAX_CONCURRENT_RUNS` | No | 25 | Sets the max concurrency for dev runs via the CLI. |
|
||||
| `DEV_OTEL_EXPORTER_OTLP_ENDPOINT` | No | `APP_ORIGIN/otel` | OTel endpoint for dev runs. |
|
||||
| **Rate limiting** | | | |
|
||||
| `API_RATE_LIMIT_REFILL_INTERVAL` | No | 10s | API rate limit refill interval. |
|
||||
| `API_RATE_LIMIT_MAX` | No | 750 | API rate limit max. |
|
||||
| `API_RATE_LIMIT_REFILL_RATE` | No | 250 | API rate limit refill rate. |
|
||||
| `API_RATE_LIMIT_REQUEST_LOGS_ENABLED` | No | 0 | API rate limit request logs. |
|
||||
| `API_RATE_LIMIT_REJECTION_LOGS_ENABLED` | No | 1 | API rate limit rejection logs. |
|
||||
| `API_RATE_LIMIT_LIMITER_LOGS_ENABLED` | No | 0 | API rate limit limiter logs. |
|
||||
| `API_RATE_LIMIT_JWT_WINDOW` | No | 1m | API rate limit JWT window. |
|
||||
| `API_RATE_LIMIT_JWT_TOKENS` | No | 60 | API rate limit JWT tokens. |
|
||||
| **Deploy & Registry** | | | |
|
||||
| `DEPLOY_REGISTRY_HOST` | Yes | — | Deploy registry host. |
|
||||
| `DEPLOY_REGISTRY_USERNAME` | No | — | Deploy registry username. |
|
||||
| `DEPLOY_REGISTRY_PASSWORD` | No | — | Deploy registry password. |
|
||||
| `DEPLOY_REGISTRY_NAMESPACE` | No | trigger | Deploy registry namespace. |
|
||||
| `DEPLOY_IMAGE_PLATFORM` | No | linux/amd64 | Deploy image platform, same values as docker `--platform` flag. |
|
||||
| `DEPLOY_TIMEOUT_MS` | No | 480000 (8m) | Deploy timeout (ms). |
|
||||
| **Object store (S3)** | | | |
|
||||
| `OBJECT_STORE_BASE_URL` | No | — | Object store base URL. |
|
||||
| `OBJECT_STORE_ACCESS_KEY_ID` | No | — | Object store access key. |
|
||||
| `OBJECT_STORE_SECRET_ACCESS_KEY` | No | — | Object store secret key. |
|
||||
| `OBJECT_STORE_REGION` | No | — | Object store region. |
|
||||
| `OBJECT_STORE_SERVICE` | No | s3 | Object store service. |
|
||||
| **Alerts** | | | |
|
||||
| `ORG_SLACK_INTEGRATION_CLIENT_ID` | No | — | Slack client ID. Required for Slack alerts. |
|
||||
| `ORG_SLACK_INTEGRATION_CLIENT_SECRET` | No | — | Slack client secret. Required for Slack alerts. |
|
||||
| `ALERT_EMAIL_TRANSPORT` | No | — | Alert email transport. |
|
||||
| `ALERT_FROM_EMAIL` | No | — | Alert from email. |
|
||||
| `ALERT_REPLY_TO_EMAIL` | No | — | Alert reply-to email. |
|
||||
| `ALERT_RESEND_API_KEY` | No | — | Alert Resend API key. |
|
||||
| `ALERT_SMTP_HOST` | No | — | Alert SMTP host. |
|
||||
| `ALERT_SMTP_PORT` | No | — | Alert SMTP port. |
|
||||
| `ALERT_SMTP_SECURE` | No | — | Alert SMTP secure. |
|
||||
| `ALERT_SMTP_USER` | No | — | Alert SMTP user. |
|
||||
| `ALERT_SMTP_PASSWORD` | No | — | Alert SMTP password. |
|
||||
| **Limits** | | | |
|
||||
| `TASK_PAYLOAD_OFFLOAD_THRESHOLD` | No | 524288 (512KB) | Max task payload size before offloading to S3. |
|
||||
| `TASK_PAYLOAD_MAXIMUM_SIZE` | No | 3145728 (3MB) | Max task payload size. |
|
||||
| `BATCH_TASK_PAYLOAD_MAXIMUM_SIZE` | No | 1000000 (1MB) | Max batch payload size. |
|
||||
| `TASK_RUN_METADATA_MAXIMUM_SIZE` | No | 262144 (256KB) | Max metadata size. |
|
||||
| `MAX_BATCH_V2_TRIGGER_ITEMS` | No | 500 | Max batch size (legacy v2 API). |
|
||||
| `STREAMING_BATCH_MAX_ITEMS` | No | 1000 | Max items in streaming batch (v3 API, requires SDK 4.3.1+). |
|
||||
| `STREAMING_BATCH_ITEM_MAXIMUM_SIZE` | No | 3145728 (3MB) | Max size per item in streaming batch. |
|
||||
| `MAXIMUM_DEV_QUEUE_SIZE` | No | — | Max dev queue size. |
|
||||
| `MAXIMUM_DEPLOYED_QUEUE_SIZE` | No | — | Max deployed queue size. |
|
||||
| **OTel limits** | | | |
|
||||
| `TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT` | No | 1024 | OTel span attribute count limit. |
|
||||
| `TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT` | No | 1024 | OTel log attribute count limit. |
|
||||
| `TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT` | No | 131072 | OTel span attribute value length limit. |
|
||||
| `TRIGGER_OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT` | No | 131072 | OTel log attribute value length limit. |
|
||||
| `TRIGGER_OTEL_SPAN_EVENT_COUNT_LIMIT` | No | 10 | OTel span event count limit. |
|
||||
| `TRIGGER_OTEL_LINK_COUNT_LIMIT` | No | 2 | OTel link count limit. |
|
||||
| `TRIGGER_OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT` | No | 10 | OTel attribute per link count limit. |
|
||||
| `TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT` | No | 10 | OTel attribute per event count limit. |
|
||||
| `SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT` | No | 8192 | OTel span attribute value length limit. |
|
||||
| **Realtime** | | | |
|
||||
| `REALTIME_STREAM_MAX_LENGTH` | No | 1000 | Realtime stream max length. |
|
||||
| `REALTIME_STREAM_TTL` | No | 86400 (1d) | Realtime stream TTL (s). |
|
||||
| **Bootstrap** | | | |
|
||||
| `TRIGGER_BOOTSTRAP_ENABLED` | No | 0 | Trigger bootstrap enabled. |
|
||||
| `TRIGGER_BOOTSTRAP_WORKER_GROUP_NAME` | No | — | Trigger bootstrap worker group name. |
|
||||
| `TRIGGER_BOOTSTRAP_WORKER_TOKEN_PATH` | No | — | Trigger bootstrap worker token path. |
|
||||
| **Run engine** | | | |
|
||||
| `RUN_ENGINE_WORKER_COUNT` | No | 4 | Run engine worker count. |
|
||||
| `RUN_ENGINE_TASKS_PER_WORKER` | No | 10 | Run engine tasks per worker. |
|
||||
| `RUN_ENGINE_WORKER_CONCURRENCY_LIMIT` | No | 10 | Run engine worker concurrency limit. |
|
||||
| `RUN_ENGINE_WORKER_POLL_INTERVAL` | No | 100 | Run engine worker poll interval (ms). |
|
||||
| `RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL` | No | 100 | Run engine worker immediate poll interval (ms). |
|
||||
| `RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS` | No | 60000 (1m) | Run engine worker shutdown timeout (ms). |
|
||||
| `RUN_ENGINE_RATE_LIMIT_REFILL_INTERVAL` | No | 10s | Run engine rate limit refill interval. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_MAX` | No | 1200 | Run engine rate limit max. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_REFILL_RATE` | No | 400 | Run engine rate limit refill rate. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_REQUEST_LOGS_ENABLED` | No | 0 | Run engine rate limit request logs. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_REJECTION_LOGS_ENABLED` | No | 1 | Run engine rate limit rejection logs. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_LIMITER_LOGS_ENABLED` | No | 0 | Run engine rate limit limiter logs. |
|
||||
| Name | Required | Default | Description |
|
||||
| :----------------------------------------------- | :------- | :-------------------- | :----------------------------------------------------------------------------------------------------------------- |
|
||||
| **Secrets** | | | |
|
||||
| `SESSION_SECRET` | Yes | — | Session encryption secret. Run: `openssl rand -hex 16` |
|
||||
| `MAGIC_LINK_SECRET` | Yes | — | Magic link encryption secret. Run: `openssl rand -hex 16` |
|
||||
| `ENCRYPTION_KEY` | Yes | — | Secret store encryption key. Run: `openssl rand -hex 16` |
|
||||
| `MANAGED_WORKER_SECRET` | No | managed-secret | Managed worker secret. Should be changed and match supervisor. |
|
||||
| **Domains & ports** | | | |
|
||||
| `REMIX_APP_PORT` | No | 3030 | Remix app port. |
|
||||
| `APP_ORIGIN` | Yes | http://localhost:3030 | App origin URL. |
|
||||
| `LOGIN_ORIGIN` | Yes | http://localhost:3030 | Login origin URL. Most likely the same as `APP_ORIGIN`. |
|
||||
| `API_ORIGIN` | No | `APP_ORIGIN` | API origin URL. |
|
||||
| `STREAM_ORIGIN` | No | `APP_ORIGIN` | Realtime stream origin URL. |
|
||||
| `ELECTRIC_ORIGIN` | No | http://localhost:3060 | Electric origin URL. |
|
||||
| **Postgres** | | | |
|
||||
| `DATABASE_URL` | Yes | — | PostgreSQL connection string. |
|
||||
| `DIRECT_URL` | Yes | — | Direct DB connection string used for migrations etc. |
|
||||
| `DATABASE_CONNECTION_LIMIT` | No | 10 | Max DB connections. |
|
||||
| `DATABASE_POOL_TIMEOUT` | No | 60 | DB pool timeout (s). |
|
||||
| `DATABASE_CONNECTION_TIMEOUT` | No | 20 | DB connect timeout (s). |
|
||||
| `DATABASE_READ_REPLICA_URL` | No | `DATABASE_URL` | Read-replica DB string. |
|
||||
| **Redis** | | | |
|
||||
| `REDIS_HOST` | Yes | — | Redis host. |
|
||||
| `REDIS_PORT` | Yes | — | Redis port. |
|
||||
| `REDIS_READER_HOST` | No | `REDIS_HOST` | Redis reader host. |
|
||||
| `REDIS_READER_PORT` | No | `REDIS_PORT` | Redis reader port. |
|
||||
| `REDIS_USERNAME` | No | — | Redis username. |
|
||||
| `REDIS_PASSWORD` | No | — | Redis password. |
|
||||
| `REDIS_TLS_DISABLED` | No | — | Disable Redis TLS. |
|
||||
| **Auth** | | | |
|
||||
| `WHITELISTED_EMAILS` | No | — | Whitelisted emails regex. |
|
||||
| `AUTH_GITHUB_CLIENT_ID` | No | — | GitHub client ID. |
|
||||
| `AUTH_GITHUB_CLIENT_SECRET` | No | — | GitHub client secret. |
|
||||
| **Email** | | | |
|
||||
| `EMAIL_TRANSPORT` | No | — | Email transport type. One of `resend`, `smtp`, `aws-ses`. |
|
||||
| `FROM_EMAIL` | No | — | From email address. |
|
||||
| `REPLY_TO_EMAIL` | No | — | Reply-to email address. |
|
||||
| `RESEND_API_KEY` | No | — | Resend API key. |
|
||||
| `SMTP_HOST` | No | — | SMTP host. |
|
||||
| `SMTP_PORT` | No | — | SMTP port. |
|
||||
| `SMTP_SECURE` | No | — | SMTP secure flag. |
|
||||
| `SMTP_USER` | No | — | SMTP user. |
|
||||
| `SMTP_PASSWORD` | No | — | SMTP password. |
|
||||
| `AWS_REGION` | No | — | AWS region for SES. |
|
||||
| `AWS_ACCESS_KEY_ID` | No | — | AWS access key ID for SES. |
|
||||
| `AWS_SECRET_ACCESS_KEY` | No | — | AWS secret access key for SES. |
|
||||
| **Graphile & Redis worker** | | | |
|
||||
| `WORKER_CONCURRENCY` | No | 10 | Redis worker concurrency. |
|
||||
| `WORKER_POLL_INTERVAL` | No | 1000 | Redis worker poll interval (ms). |
|
||||
| `WORKER_SCHEMA` | No | graphile_worker | Graphile worker schema. |
|
||||
| `GRACEFUL_SHUTDOWN_TIMEOUT` | No | 60000 (1m) | Graphile graceful shutdown timeout (ms). Affects shutdown time. |
|
||||
| **Concurrency limits** | | | |
|
||||
| `DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT` | No | 100 | Default env execution concurrency. |
|
||||
| `DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT` | No | 300 | Default org execution concurrency, needs to be 3x env concurrency. |
|
||||
| **Dev** | | | |
|
||||
| `DEV_MAX_CONCURRENT_RUNS` | No | 25 | Sets the max concurrency for dev runs via the CLI. |
|
||||
| `DEV_OTEL_EXPORTER_OTLP_ENDPOINT` | No | `APP_ORIGIN/otel` | OTel endpoint for dev runs. |
|
||||
| **Rate limiting** | | | |
|
||||
| `API_RATE_LIMIT_REFILL_INTERVAL` | No | 10s | API rate limit refill interval. |
|
||||
| `API_RATE_LIMIT_MAX` | No | 750 | API rate limit max. |
|
||||
| `API_RATE_LIMIT_REFILL_RATE` | No | 250 | API rate limit refill rate. |
|
||||
| `API_RATE_LIMIT_REQUEST_LOGS_ENABLED` | No | 0 | API rate limit request logs. |
|
||||
| `API_RATE_LIMIT_REJECTION_LOGS_ENABLED` | No | 1 | API rate limit rejection logs. |
|
||||
| `API_RATE_LIMIT_LIMITER_LOGS_ENABLED` | No | 0 | API rate limit limiter logs. |
|
||||
| `API_RATE_LIMIT_JWT_WINDOW` | No | 1m | API rate limit JWT window. |
|
||||
| `API_RATE_LIMIT_JWT_TOKENS` | No | 60 | API rate limit JWT tokens. |
|
||||
| **Deploy & Registry** | | | |
|
||||
| `DEPLOY_REGISTRY_HOST` | Yes | — | Deploy registry host. |
|
||||
| `DEPLOY_REGISTRY_USERNAME` | No | — | Deploy registry username. |
|
||||
| `DEPLOY_REGISTRY_PASSWORD` | No | — | Deploy registry password. |
|
||||
| `DEPLOY_REGISTRY_NAMESPACE` | No | trigger | Deploy registry namespace. |
|
||||
| `DEPLOY_IMAGE_PLATFORM` | No | linux/amd64 | Deploy image platform, same values as docker `--platform` flag. |
|
||||
| `DEPLOY_TIMEOUT_MS` | No | 480000 (8m) | Deploy timeout (ms). |
|
||||
| **Object store (S3)** | | | |
|
||||
| `OBJECT_STORE_BASE_URL` | No | — | Object store base URL (default provider). |
|
||||
| `OBJECT_STORE_ACCESS_KEY_ID` | No | — | Object store access key (default provider). |
|
||||
| `OBJECT_STORE_SECRET_ACCESS_KEY` | No | — | Object store secret key (default provider). |
|
||||
| `OBJECT_STORE_REGION` | No | — | Object store region (default provider). |
|
||||
| `OBJECT_STORE_SERVICE` | No | s3 | Object store service (default provider). |
|
||||
| `OBJECT_STORE_DEFAULT_PROTOCOL` | No | — | Protocol to use for new uploads (e.g., `s3`, `r2`). Enables protocol-prefixed storage. See migration guide below. |
|
||||
| `OBJECT_STORE_{PROTOCOL}_BASE_URL` | No | — | Named provider base URL (replace `{PROTOCOL}` with protocol name, e.g., `OBJECT_STORE_S3_BASE_URL`). |
|
||||
| `OBJECT_STORE_{PROTOCOL}_ACCESS_KEY_ID` | No | — | Named provider access key. |
|
||||
| `OBJECT_STORE_{PROTOCOL}_SECRET_ACCESS_KEY` | No | — | Named provider secret key. |
|
||||
| `OBJECT_STORE_{PROTOCOL}_REGION` | No | — | Named provider region. |
|
||||
| `OBJECT_STORE_{PROTOCOL}_SERVICE` | No | — | Named provider service. |
|
||||
| **Alerts** | | | |
|
||||
| `ORG_SLACK_INTEGRATION_CLIENT_ID` | No | — | Slack client ID. Required for Slack alerts. |
|
||||
| `ORG_SLACK_INTEGRATION_CLIENT_SECRET` | No | — | Slack client secret. Required for Slack alerts. |
|
||||
| `ALERT_EMAIL_TRANSPORT` | No | — | Alert email transport. |
|
||||
| `ALERT_FROM_EMAIL` | No | — | Alert from email. |
|
||||
| `ALERT_REPLY_TO_EMAIL` | No | — | Alert reply-to email. |
|
||||
| `ALERT_RESEND_API_KEY` | No | — | Alert Resend API key. |
|
||||
| `ALERT_SMTP_HOST` | No | — | Alert SMTP host. |
|
||||
| `ALERT_SMTP_PORT` | No | — | Alert SMTP port. |
|
||||
| `ALERT_SMTP_SECURE` | No | — | Alert SMTP secure. |
|
||||
| `ALERT_SMTP_USER` | No | — | Alert SMTP user. |
|
||||
| `ALERT_SMTP_PASSWORD` | No | — | Alert SMTP password. |
|
||||
| **Limits** | | | |
|
||||
| `TASK_PAYLOAD_OFFLOAD_THRESHOLD` | No | 524288 (512KB) | Max task payload size before offloading to S3. |
|
||||
| `TASK_PAYLOAD_MAXIMUM_SIZE` | No | 3145728 (3MB) | Max task payload size. |
|
||||
| `BATCH_TASK_PAYLOAD_MAXIMUM_SIZE` | No | 1000000 (1MB) | Max batch payload size. |
|
||||
| `TASK_RUN_METADATA_MAXIMUM_SIZE` | No | 262144 (256KB) | Max metadata size. |
|
||||
| `MAX_BATCH_V2_TRIGGER_ITEMS` | No | 500 | Max batch size (legacy v2 API). |
|
||||
| `STREAMING_BATCH_MAX_ITEMS` | No | 1000 | Max items in streaming batch (v3 API, requires SDK 4.3.1+). |
|
||||
| `STREAMING_BATCH_ITEM_MAXIMUM_SIZE` | No | 3145728 (3MB) | Max size per item in streaming batch. |
|
||||
| `MAXIMUM_DEV_QUEUE_SIZE` | No | — | Max dev queue size. |
|
||||
| `MAXIMUM_DEPLOYED_QUEUE_SIZE` | No | — | Max deployed queue size. |
|
||||
| **OTel limits** | | | |
|
||||
| `TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT` | No | 1024 | OTel span attribute count limit. |
|
||||
| `TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT` | No | 1024 | OTel log attribute count limit. |
|
||||
| `TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT` | No | 131072 | OTel span attribute value length limit. |
|
||||
| `TRIGGER_OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT` | No | 131072 | OTel log attribute value length limit. |
|
||||
| `TRIGGER_OTEL_SPAN_EVENT_COUNT_LIMIT` | No | 10 | OTel span event count limit. |
|
||||
| `TRIGGER_OTEL_LINK_COUNT_LIMIT` | No | 2 | OTel link count limit. |
|
||||
| `TRIGGER_OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT` | No | 10 | OTel attribute per link count limit. |
|
||||
| `TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT` | No | 10 | OTel attribute per event count limit. |
|
||||
| `SERVER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT` | No | 8192 | OTel span attribute value length limit. |
|
||||
| **Realtime** | | | |
|
||||
| `REALTIME_STREAM_MAX_LENGTH` | No | 1000 | Realtime stream max length. |
|
||||
| `REALTIME_STREAM_TTL` | No | 86400 (1d) | Realtime stream TTL (s). |
|
||||
| **Bootstrap** | | | |
|
||||
| `TRIGGER_BOOTSTRAP_ENABLED` | No | 0 | Trigger bootstrap enabled. |
|
||||
| `TRIGGER_BOOTSTRAP_WORKER_GROUP_NAME` | No | — | Trigger bootstrap worker group name. |
|
||||
| `TRIGGER_BOOTSTRAP_WORKER_TOKEN_PATH` | No | — | Trigger bootstrap worker token path. |
|
||||
| **Run engine** | | | |
|
||||
| `RUN_ENGINE_WORKER_COUNT` | No | 4 | Run engine worker count. |
|
||||
| `RUN_ENGINE_TASKS_PER_WORKER` | No | 10 | Run engine tasks per worker. |
|
||||
| `RUN_ENGINE_WORKER_CONCURRENCY_LIMIT` | No | 10 | Run engine worker concurrency limit. |
|
||||
| `RUN_ENGINE_WORKER_POLL_INTERVAL` | No | 100 | Run engine worker poll interval (ms). |
|
||||
| `RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL` | No | 100 | Run engine worker immediate poll interval (ms). |
|
||||
| `RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS` | No | 60000 (1m) | Run engine worker shutdown timeout (ms). |
|
||||
| `RUN_ENGINE_RATE_LIMIT_REFILL_INTERVAL` | No | 10s | Run engine rate limit refill interval. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_MAX` | No | 1200 | Run engine rate limit max. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_REFILL_RATE` | No | 400 | Run engine rate limit refill rate. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_REQUEST_LOGS_ENABLED` | No | 0 | Run engine rate limit request logs. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_REJECTION_LOGS_ENABLED` | No | 1 | Run engine rate limit rejection logs. |
|
||||
| `RUN_ENGINE_RATE_LIMIT_LIMITER_LOGS_ENABLED` | No | 0 | Run engine rate limit limiter logs. |
|
||||
| `RUN_ENGINE_DEFAULT_MAX_TTL` | No | — | Maximum TTL for all runs (e.g. "14d"). Runs without a TTL use this as default; runs with a larger TTL are clamped. |
|
||||
| `MAXIMUM_DEV_QUEUE_SIZE` | No | — | Maximum queued runs per queue in development environments. |
|
||||
| `MAXIMUM_DEPLOYED_QUEUE_SIZE` | No | — | Maximum queued runs per queue in deployed (staging/prod) environments. |
|
||||
| **Misc** | | | |
|
||||
| `TRIGGER_TELEMETRY_DISABLED` | No | — | Disable telemetry. |
|
||||
| `NODE_MAX_OLD_SPACE_SIZE` | No | 8192 | Maximum memory allocation for Node.js heap in MiB (e.g. "4096" for 4GB). |
|
||||
| `OPENAI_API_KEY` | No | — | OpenAI API key. |
|
||||
| `MACHINE_PRESETS_OVERRIDE_PATH` | No | — | Path to machine presets override file. See [machine overrides](/self-hosting/overview#machine-overrides). |
|
||||
| `APP_ENV` | No | `NODE_ENV` | App environment. Used for things like the title tag. |
|
||||
| `ADMIN_EMAILS` | No | — | Regex of user emails to automatically promote to admin on signup. Does not apply to existing users. |
|
||||
| `EVENT_LOOP_MONITOR_ENABLED` | No | 1 | Node.js event loop lag monitor. |
|
||||
| `MAXIMUM_DEV_QUEUE_SIZE` | No | — | Maximum queued runs per queue in development environments. |
|
||||
| `MAXIMUM_DEPLOYED_QUEUE_SIZE` | No | — | Maximum queued runs per queue in deployed (staging/prod) environments. |
|
||||
| **Misc** | | | |
|
||||
| `TRIGGER_TELEMETRY_DISABLED` | No | — | Disable telemetry. |
|
||||
| `NODE_MAX_OLD_SPACE_SIZE` | No | 8192 | Maximum memory allocation for Node.js heap in MiB (e.g. "4096" for 4GB). |
|
||||
| `OPENAI_API_KEY` | No | — | OpenAI API key. |
|
||||
| `MACHINE_PRESETS_OVERRIDE_PATH` | No | — | Path to machine presets override file. See [machine overrides](/self-hosting/overview#machine-overrides). |
|
||||
| `APP_ENV` | No | `NODE_ENV` | App environment. Used for things like the title tag. |
|
||||
| `ADMIN_EMAILS` | No | — | Regex of user emails to automatically promote to admin on signup. Does not apply to existing users. |
|
||||
| `EVENT_LOOP_MONITOR_ENABLED` | No | 1 | Node.js event loop lag monitor. |
|
||||
|
||||
## Multi-Provider Object Storage
|
||||
|
||||
The object storage system supports multiple S3-compatible providers (R2, S3, GCS, MinIO, etc.) using protocol prefixes. This enables migrating between providers without breaking existing runs.
|
||||
|
||||
### How It Works
|
||||
|
||||
When data exceeds the configured threshold (`TASK_PAYLOAD_OFFLOAD_THRESHOLD`), it's uploaded to object storage. The storage location is saved in the database with an optional protocol prefix:
|
||||
|
||||
- **With protocol**: `s3://run_abc/payload.json` or `r2://batch_123/item_0/payload.json`
|
||||
- **Without protocol** (legacy): `batch_123/item_0/payload.json` (uses default provider)
|
||||
|
||||
### Configuration
|
||||
|
||||
#### Default Provider (Backward Compatible)
|
||||
|
||||
The default provider is used for data without a protocol prefix:
|
||||
|
||||
```bash
|
||||
# Default provider (backward compatible - no protocol prefix)
|
||||
OBJECT_STORE_BASE_URL=https://r2.example.com
|
||||
OBJECT_STORE_ACCESS_KEY_ID=...
|
||||
OBJECT_STORE_SECRET_ACCESS_KEY=...
|
||||
OBJECT_STORE_REGION=auto
|
||||
OBJECT_STORE_SERVICE=s3
|
||||
```
|
||||
|
||||
#### Named Providers
|
||||
|
||||
Named providers are accessed via protocol-prefixed URIs. Configure them using `OBJECT_STORE_{PROTOCOL}_*` variables:
|
||||
|
||||
```bash
|
||||
# S3 provider (accessed via s3:// prefix)
|
||||
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
|
||||
|
||||
# R2 provider (accessed via r2:// prefix)
|
||||
OBJECT_STORE_R2_BASE_URL=https://...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
|
||||
```
|
||||
|
||||
#### Default Protocol for New Uploads
|
||||
|
||||
Set `OBJECT_STORE_DEFAULT_PROTOCOL` to specify which provider to use for new uploads:
|
||||
|
||||
```bash
|
||||
# Use S3 for new uploads (old data without prefix still uses default provider)
|
||||
OBJECT_STORE_DEFAULT_PROTOCOL=s3
|
||||
```
|
||||
|
||||
### Migration Guide
|
||||
|
||||
To migrate from R2 to S3 without breaking existing runs:
|
||||
|
||||
<Steps>
|
||||
<Step title="Configure S3 provider">
|
||||
Add S3 credentials as a named provider:
|
||||
|
||||
```bash
|
||||
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
|
||||
```
|
||||
|
||||
Keep your existing `OBJECT_STORE_*` variables (R2) as the default provider.
|
||||
|
||||
</Step>
|
||||
|
||||
<Step title="Test the configuration">
|
||||
Restart the webapp and verify both providers work:
|
||||
|
||||
- Old runs (no prefix) should still access R2
|
||||
- New runs with `s3://` prefix should use S3
|
||||
</Step>
|
||||
|
||||
<Step title="Switch to S3 for new uploads">
|
||||
Set the default protocol to use S3 for new uploads:
|
||||
|
||||
```bash
|
||||
OBJECT_STORE_DEFAULT_PROTOCOL=s3
|
||||
```
|
||||
|
||||
After this change:
|
||||
|
||||
- New data uses `s3://` prefix and goes to S3
|
||||
- Old data (no prefix) still uses R2
|
||||
- Data with explicit protocol uses the corresponding provider
|
||||
</Step>
|
||||
|
||||
<Step title="Optionally decommission R2">
|
||||
Once all active runs using R2 data have completed (check your data retention policies), you can remove the R2 credentials. Keep `OBJECT_STORE_DEFAULT_PROTOCOL=s3` to ensure new data continues using S3.
|
||||
</Step>
|
||||
</Steps>
|
||||
|
||||
@@ -77,12 +77,29 @@ DOCKER_REGISTRY_PASSWORD=very-secure-indeed
|
||||
DOCKER_REGISTRY_NAMESPACE=trigger
|
||||
|
||||
# Object store
|
||||
# - You need to log into the Minio dashboard and create a bucket called "packets"
|
||||
# - You need to log into the Minio dashboard and create a bucket called "packets"
|
||||
# - See the docs for more information: https://trigger.dev/docs/self-hosting/docker#object-storage
|
||||
# Default provider (backward compatible - no protocol prefix)
|
||||
OBJECT_STORE_ACCESS_KEY_ID=admin
|
||||
OBJECT_STORE_SECRET_ACCESS_KEY=very-safe-password
|
||||
# You will have to uncomment and configure this for production
|
||||
# OBJECT_STORE_BASE_URL=http://localhost:9000
|
||||
# OBJECT_STORE_REGION=auto
|
||||
# OBJECT_STORE_SERVICE=s3
|
||||
# OBJECT_STORE_DEFAULT_PROTOCOL=s3 # Optional: protocol to use for new uploads (e.g., "s3", "r2")
|
||||
#
|
||||
# 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
|
||||
# Credentials to access the Minio dashboard at http://localhost:9001
|
||||
# - You should change these credentials and not use them for the `OBJECT_STORE_` env vars above
|
||||
# - Instead, setup a non-root user with access the "packets" bucket
|
||||
|
||||
@@ -645,6 +645,18 @@ s3:
|
||||
existingSecret: "" # Name of existing secret containing S3 credentials
|
||||
existingSecretAccessKeyIdKey: "access-key-id" # Key in existing secret containing access key ID
|
||||
existingSecretSecretAccessKeyKey: "secret-access-key" # Key in existing secret containing secret access key
|
||||
#
|
||||
# Multi-provider support (optional)
|
||||
# To use multiple S3-compatible providers (e.g., S3 + R2), set defaultProtocol and configure named providers
|
||||
# via extraEnvVars. For example:
|
||||
# - OBJECT_STORE_DEFAULT_PROTOCOL=s3 # Protocol to use for new uploads
|
||||
# - OBJECT_STORE_S3_BASE_URL=https://s3.amazonaws.com
|
||||
# - OBJECT_STORE_S3_ACCESS_KEY_ID=...
|
||||
# - OBJECT_STORE_S3_SECRET_ACCESS_KEY=...
|
||||
# - OBJECT_STORE_R2_BASE_URL=https://...r2.cloudflarestorage.com
|
||||
# - OBJECT_STORE_R2_ACCESS_KEY_ID=...
|
||||
# - OBJECT_STORE_R2_SECRET_ACCESS_KEY=...
|
||||
# See documentation for migration guide: https://trigger.dev/docs/self-hosting/env/webapp
|
||||
|
||||
# Docker Registry configuration
|
||||
registry:
|
||||
|
||||
@@ -9,15 +9,18 @@ import {
|
||||
createElectricContainer,
|
||||
createPostgresContainer,
|
||||
createRedisContainer,
|
||||
createMinIOContainer,
|
||||
useContainer,
|
||||
withContainerSetup,
|
||||
} from "./utils";
|
||||
import { getTaskMetadata, logCleanup, logSetup } from "./logs";
|
||||
import { StartedClickHouseContainer } from "./clickhouse";
|
||||
import { StartedMinIOContainer, type MinIOConnectionConfig } from "./minio";
|
||||
import { ClickHouseClient, createClient } from "@clickhouse/client";
|
||||
|
||||
export { assertNonNullable } from "./utils";
|
||||
export { logCleanup };
|
||||
export type { MinIOConnectionConfig };
|
||||
|
||||
type NetworkContext = { network: StartedNetwork };
|
||||
|
||||
@@ -40,11 +43,17 @@ export type PostgresAndRedisContext = NetworkContext & PostgresContext & RedisCo
|
||||
export type ContainerWithElectricAndRedisContext = ContainerContext & ElectricContext;
|
||||
export type ContainerWithElectricContext = NetworkContext & PostgresContext & ElectricContext;
|
||||
|
||||
type MinIOContext = NetworkContext & {
|
||||
minioContainer: StartedMinIOContainer;
|
||||
minioConfig: MinIOConnectionConfig;
|
||||
};
|
||||
|
||||
export type {
|
||||
StartedNetwork,
|
||||
StartedPostgreSqlContainer,
|
||||
StartedRedisContainer,
|
||||
StartedClickHouseContainer,
|
||||
StartedMinIOContainer,
|
||||
};
|
||||
|
||||
type Use<T> = (value: T) => Promise<void>;
|
||||
@@ -257,3 +266,39 @@ export const containerWithElectricAndRedisTest = test.extend<ContainerWithElectr
|
||||
clickhouseContainer,
|
||||
clickhouseClient,
|
||||
});
|
||||
|
||||
const minioContainer = async (
|
||||
{ network, task }: { network: StartedNetwork } & TaskContext,
|
||||
use: Use<StartedMinIOContainer>
|
||||
) => {
|
||||
const { container, metadata } = await withContainerSetup({
|
||||
name: "minioContainer",
|
||||
task,
|
||||
setup: createMinIOContainer(network),
|
||||
});
|
||||
|
||||
await useContainer("minioContainer", { container, task, use: () => use(container) });
|
||||
};
|
||||
|
||||
const minioConfig = async (
|
||||
{ minioContainer }: { minioContainer: StartedMinIOContainer },
|
||||
use: Use<MinIOConnectionConfig>
|
||||
) => {
|
||||
await use(minioContainer.getConnectionConfig());
|
||||
};
|
||||
|
||||
export const minioTest = test.extend<MinIOContext>({
|
||||
network,
|
||||
minioContainer,
|
||||
minioConfig,
|
||||
});
|
||||
|
||||
type PostgresAndMinIOContext = NetworkContext & PostgresContext & MinIOContext;
|
||||
|
||||
export const postgresAndMinioTest = test.extend<PostgresAndMinIOContext>({
|
||||
network,
|
||||
postgresContainer,
|
||||
prisma,
|
||||
minioContainer,
|
||||
minioConfig,
|
||||
});
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
import {
|
||||
AbstractStartedContainer,
|
||||
GenericContainer,
|
||||
StartedTestContainer,
|
||||
Wait,
|
||||
} from "testcontainers";
|
||||
import { x } from "tinyexec";
|
||||
|
||||
const MINIO_PORT = 9000;
|
||||
|
||||
export type MinIOConnectionConfig = {
|
||||
baseUrl: string;
|
||||
accessKeyId: string;
|
||||
secretAccessKey: string;
|
||||
region: string;
|
||||
};
|
||||
|
||||
export class MinIOContainer extends GenericContainer {
|
||||
private accessKeyId = "minioadmin";
|
||||
private secretAccessKey = "minioadmin";
|
||||
private region = "us-east-1";
|
||||
|
||||
constructor(image = "minio/minio:latest") {
|
||||
super(image);
|
||||
this.withExposedPorts(MINIO_PORT);
|
||||
this.withCommand(["server", "/data"]);
|
||||
this.withWaitStrategy(Wait.forLogMessage(/API:/));
|
||||
this.withStartupTimeout(120_000);
|
||||
}
|
||||
|
||||
public withAccessKeyId(accessKeyId: string): this {
|
||||
this.accessKeyId = accessKeyId;
|
||||
return this;
|
||||
}
|
||||
|
||||
public withSecretAccessKey(secretAccessKey: string): this {
|
||||
this.secretAccessKey = secretAccessKey;
|
||||
return this;
|
||||
}
|
||||
|
||||
public withRegion(region: string): this {
|
||||
this.region = region;
|
||||
return this;
|
||||
}
|
||||
|
||||
public override async start(): Promise<StartedMinIOContainer> {
|
||||
this.withEnvironment({
|
||||
MINIO_ROOT_USER: this.accessKeyId,
|
||||
MINIO_ROOT_PASSWORD: this.secretAccessKey,
|
||||
});
|
||||
|
||||
const startedContainer = await super.start();
|
||||
|
||||
// Create the "packets" bucket using MinIO client
|
||||
await x(
|
||||
"docker",
|
||||
[
|
||||
"exec",
|
||||
startedContainer.getId(),
|
||||
"mc",
|
||||
"alias",
|
||||
"set",
|
||||
"local",
|
||||
"http://localhost:9000",
|
||||
this.accessKeyId,
|
||||
this.secretAccessKey,
|
||||
],
|
||||
{ throwOnError: true }
|
||||
);
|
||||
|
||||
await x(
|
||||
"docker",
|
||||
["exec", startedContainer.getId(), "mc", "mb", "local/packets"],
|
||||
{ throwOnError: true }
|
||||
);
|
||||
|
||||
return new StartedMinIOContainer(
|
||||
startedContainer,
|
||||
this.accessKeyId,
|
||||
this.secretAccessKey,
|
||||
this.region
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export class StartedMinIOContainer extends AbstractStartedContainer {
|
||||
constructor(
|
||||
startedTestContainer: StartedTestContainer,
|
||||
private readonly accessKeyId: string,
|
||||
private readonly secretAccessKey: string,
|
||||
private readonly region: string
|
||||
) {
|
||||
super(startedTestContainer);
|
||||
}
|
||||
|
||||
public getPort(): number {
|
||||
return super.getMappedPort(MINIO_PORT);
|
||||
}
|
||||
|
||||
public getAccessKeyId(): string {
|
||||
return this.accessKeyId;
|
||||
}
|
||||
|
||||
public getSecretAccessKey(): string {
|
||||
return this.secretAccessKey;
|
||||
}
|
||||
|
||||
public getRegion(): string {
|
||||
return this.region;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the base URL (protocol, host and mapped port) for the MinIO container.
|
||||
* Example: `http://localhost:32768`
|
||||
*/
|
||||
public getBaseUrl(): string {
|
||||
const protocol = "http";
|
||||
const host = this.getHost();
|
||||
const port = this.getPort();
|
||||
return `${protocol}://${host}:${port}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets connection configuration suitable for object storage clients.
|
||||
*/
|
||||
public getConnectionConfig(): MinIOConnectionConfig {
|
||||
return {
|
||||
baseUrl: this.getBaseUrl(),
|
||||
accessKeyId: this.getAccessKeyId(),
|
||||
secretAccessKey: this.getSecretAccessKey(),
|
||||
region: this.getRegion(),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import { GenericContainer, StartedNetwork, StartedTestContainer, Wait } from "te
|
||||
import { x } from "tinyexec";
|
||||
import { expect, TaskContext } from "vitest";
|
||||
import { ClickHouseContainer, runClickhouseMigrations } from "./clickhouse";
|
||||
import { MinIOContainer } from "./minio";
|
||||
import { getContainerMetadata, getTaskMetadata, logCleanup, logSetup } from "./logs";
|
||||
|
||||
export async function createPostgresContainer(network: StartedNetwork) {
|
||||
@@ -170,6 +171,18 @@ export async function createElectricContainer(
|
||||
};
|
||||
}
|
||||
|
||||
export async function createMinIOContainer(network: StartedNetwork) {
|
||||
const container = await new MinIOContainer()
|
||||
.withNetwork(network)
|
||||
.withNetworkAliases("minio")
|
||||
.start();
|
||||
|
||||
return {
|
||||
container,
|
||||
network,
|
||||
};
|
||||
}
|
||||
|
||||
export function assertNonNullable<T>(value: T): asserts value is NonNullable<T> {
|
||||
expect(value).toBeDefined();
|
||||
expect(value).not.toBeNull();
|
||||
|
||||
@@ -559,9 +559,10 @@ export class ApiClient {
|
||||
}
|
||||
|
||||
createUploadPayloadUrl(filename: string, requestOptions?: ZodFetchOptions) {
|
||||
const encoded = encodeURIComponent(filename);
|
||||
return zodfetch(
|
||||
CreateUploadPayloadUrlResponseBody,
|
||||
`${this.baseUrl}/api/v1/packets/${filename}`,
|
||||
`${this.baseUrl}/api/v2/packets/${encoded}`,
|
||||
{
|
||||
method: "PUT",
|
||||
headers: this.#getHeaders(false),
|
||||
@@ -571,9 +572,10 @@ export class ApiClient {
|
||||
}
|
||||
|
||||
getPayloadUrl(filename: string, requestOptions?: ZodFetchOptions) {
|
||||
const encoded = encodeURIComponent(filename);
|
||||
return zodfetch(
|
||||
CreateUploadPayloadUrlResponseBody,
|
||||
`${this.baseUrl}/api/v1/packets/${filename}`,
|
||||
`${this.baseUrl}/api/v1/packets/${encoded}`,
|
||||
{
|
||||
method: "GET",
|
||||
headers: this.#getHeaders(false),
|
||||
|
||||
@@ -788,6 +788,8 @@ export const DeploymentEventFromString = z
|
||||
|
||||
export const CreateUploadPayloadUrlResponseBody = z.object({
|
||||
presignedUrl: z.string(),
|
||||
/** Present on `/api/v2/packets` PUT (upload handshake); omitted on v1 GET download presign. */
|
||||
storagePath: z.string().optional(),
|
||||
});
|
||||
|
||||
export const WorkersListResponseBody = z
|
||||
|
||||
@@ -169,6 +169,12 @@ async function exportPacket(packet: IOPacket, pathPrefix: string): Promise<IOPac
|
||||
|
||||
const presignedResponse = await apiClientManager.client!.createUploadPayloadUrl(filename);
|
||||
|
||||
if (!presignedResponse.storagePath) {
|
||||
throw new Error(
|
||||
"Packet upload presign response missing storagePath; ensure the server supports /api/v2/packets"
|
||||
);
|
||||
}
|
||||
|
||||
const uploadResponse = await zodfetch(
|
||||
z.any(),
|
||||
presignedResponse.presignedUrl,
|
||||
@@ -191,7 +197,7 @@ async function exportPacket(packet: IOPacket, pathPrefix: string): Promise<IOPac
|
||||
}
|
||||
|
||||
return {
|
||||
data: filename,
|
||||
data: presignedResponse.storagePath,
|
||||
dataType: "application/store",
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user