Merge branch 'main' into issue/ffmpegv7

Author: Rudra-Sankha-Sinhamahapatra

.changeset/smart-coins-hammer.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+fix: Realtime streams: prevent enqueuing into closed ReadableStream

apps/supervisor/README.md

@@ -8,20 +8,15 @@
 api_url=http://localhost:3030
 wg_name=my-worker
 
-# edit these
+# edit this
 admin_pat=tr_pat_...
-project_id=clsw6q8wz...
 
 curl -sS \
     -X POST \
     "$api_url/admin/api/v1/workers" \
     -H "Authorization: Bearer $admin_pat" \
     -H "Content-Type: application/json" \
-    -d "{
-        \"name\": \"$wg_name\",
-        \"makeDefault\": true,
-        \"projectId\": \"$project_id\"
-    }"
+    -d "{\"name\": \"$wg_name\"}"
 ```
 
 2. Create `.env` and set the worker token
@@ -47,3 +42,26 @@ pnpm exec trigger deploy --self-hosted
 # The additional network flag is required on linux
 pnpm exec trigger deploy --self-hosted --network host
 ```
+
+## Additional worker groups
+
+When adding more worker groups you might also want to make them the default for a specific project. This will allow you to test it without having to change the global default:
+
+```sh
+api_url=http://localhost:3030
+wg_name=my-worker
+
+# edit these
+admin_pat=tr_pat_...
+project_id=clsw6q8wz...
+
+curl -sS \
+    -X POST \
+    "$api_url/admin/api/v1/workers" \
+    -H "Authorization: Bearer $admin_pat" \
+    -H "Content-Type: application/json" \
+    -d "{
+        \"name\": \"$wg_name\",
+        \"makeDefaultForProjectId\": \"$project_id\"
+    }"
+```

apps/supervisor/src/env.ts

@@ -1,14 +1,7 @@
 import { randomUUID } from "crypto";
 import { env as stdEnv } from "std-env";
 import { z } from "zod";
-
-const BoolEnv = z.preprocess((val) => {
-  if (typeof val !== "string") {
-    return val;
-  }
-
-  return ["true", "1"].includes(val.toLowerCase().trim());
-}, z.boolean());
+import { AdditionalEnvVars, BoolEnv } from "./envUtil.js";
 
 const Env = z.object({
   // This will come from `spec.nodeName` in k8s
@@ -26,9 +19,15 @@ const Env = z.object({
     .transform((s) => z.enum(["http", "https"]).parse(s.toLowerCase()))
     .default("http"),
   TRIGGER_WORKLOAD_API_DOMAIN: z.string().optional(), // If unset, will use orchestrator-specific default
+  TRIGGER_WORKLOAD_API_HOST_INTERNAL: z.string().default("0.0.0.0"),
   TRIGGER_WORKLOAD_API_PORT_INTERNAL: z.coerce.number().default(8020), // This is the port the workload API listens on
   TRIGGER_WORKLOAD_API_PORT_EXTERNAL: z.coerce.number().default(8020), // This is the exposed port passed to the run controller
 
+  // Runner settings
+  RUNNER_HEARTBEAT_INTERVAL_SECONDS: z.coerce.number().optional(),
+  RUNNER_SNAPSHOT_POLL_INTERVAL_SECONDS: z.coerce.number().optional(),
+  RUNNER_ADDITIONAL_ENV_VARS: AdditionalEnvVars, // optional (csv)
+
   // Dequeue settings (provider mode)
   TRIGGER_DEQUEUE_ENABLED: BoolEnv.default("true"),
   TRIGGER_DEQUEUE_INTERVAL_MS: z.coerce.number().int().default(1000),
@@ -41,6 +40,7 @@ const Env = z.object({
   DOCKER_NETWORK: z.string().default("host"),
   OTEL_EXPORTER_OTLP_ENDPOINT: z.string().url(),
   ENFORCE_MACHINE_PRESETS: z.coerce.boolean().default(false),
+  KUBERNETES_IMAGE_PULL_SECRETS: z.string().optional(), // csv
 
   // Used by the resource monitor
   OVERRIDE_CPU_TOTAL: z.coerce.number().optional(),
@@ -53,7 +53,10 @@ const Env = z.object({
   EPHEMERAL_STORAGE_SIZE_REQUEST: z.string().default("2Gi"),
 
   // Metrics
+  METRICS_ENABLED: BoolEnv.default(true),
   METRICS_COLLECT_DEFAULTS: BoolEnv.default(true),
+  METRICS_HOST: z.string().default("127.0.0.1"),
+  METRICS_PORT: z.coerce.number().int().default(9090),
 
   // Pod cleaner
   POD_CLEANER_ENABLED: BoolEnv.default(true),
@@ -63,6 +66,9 @@ const Env = z.object({
   // Failed pod handler
   FAILED_POD_HANDLER_ENABLED: BoolEnv.default(true),
   FAILED_POD_HANDLER_RECONNECT_INTERVAL_MS: z.coerce.number().int().default(1000),
+
+  // Debug
+  DEBUG: BoolEnv.default(false),
 });
 
 export const env = Env.parse(stdEnv);

apps/supervisor/src/envUtil.test.ts

@@ -0,0 +1,80 @@
+import { describe, it, expect } from "vitest";
+import { BoolEnv, AdditionalEnvVars } from "./envUtil.js";
+
+describe("BoolEnv", () => {
+  it("should parse string 'true' as true", () => {
+    expect(BoolEnv.parse("true")).toBe(true);
+    expect(BoolEnv.parse("TRUE")).toBe(true);
+    expect(BoolEnv.parse("True")).toBe(true);
+  });
+
+  it("should parse string '1' as true", () => {
+    expect(BoolEnv.parse("1")).toBe(true);
+  });
+
+  it("should parse string 'false' as false", () => {
+    expect(BoolEnv.parse("false")).toBe(false);
+    expect(BoolEnv.parse("FALSE")).toBe(false);
+    expect(BoolEnv.parse("False")).toBe(false);
+  });
+
+  it("should handle whitespace", () => {
+    expect(BoolEnv.parse(" true ")).toBe(true);
+    expect(BoolEnv.parse(" 1 ")).toBe(true);
+  });
+
+  it("should pass through boolean values", () => {
+    expect(BoolEnv.parse(true)).toBe(true);
+    expect(BoolEnv.parse(false)).toBe(false);
+  });
+
+  it("should return false for invalid inputs", () => {
+    expect(BoolEnv.parse("invalid")).toBe(false);
+    expect(BoolEnv.parse("")).toBe(false);
+  });
+});
+
+describe("AdditionalEnvVars", () => {
+  it("should parse single key-value pair", () => {
+    expect(AdditionalEnvVars.parse("FOO=bar")).toEqual({ FOO: "bar" });
+  });
+
+  it("should parse multiple key-value pairs", () => {
+    expect(AdditionalEnvVars.parse("FOO=bar,BAZ=qux")).toEqual({
+      FOO: "bar",
+      BAZ: "qux",
+    });
+  });
+
+  it("should handle whitespace", () => {
+    expect(AdditionalEnvVars.parse(" FOO = bar , BAZ = qux ")).toEqual({
+      FOO: "bar",
+      BAZ: "qux",
+    });
+  });
+
+  it("should return undefined for empty string", () => {
+    expect(AdditionalEnvVars.parse("")).toBeUndefined();
+  });
+
+  it("should return undefined for invalid format", () => {
+    expect(AdditionalEnvVars.parse("invalid")).toBeUndefined();
+  });
+
+  it("should skip invalid pairs but include valid ones", () => {
+    expect(AdditionalEnvVars.parse("FOO=bar,INVALID,BAZ=qux")).toEqual({
+      FOO: "bar",
+      BAZ: "qux",
+    });
+  });
+
+  it("should pass through undefined", () => {
+    expect(AdditionalEnvVars.parse(undefined)).toBeUndefined();
+  });
+
+  it("should handle empty values", () => {
+    expect(AdditionalEnvVars.parse("FOO=,BAR=value")).toEqual({
+      BAR: "value",
+    });
+  });
+});

apps/supervisor/src/envUtil.ts

@@ -0,0 +1,39 @@
+import { z } from "zod";
+
+export const BoolEnv = z.preprocess((val) => {
+  if (typeof val !== "string") {
+    return val;
+  }
+
+  return ["true", "1"].includes(val.toLowerCase().trim());
+}, z.boolean());
+
+export const AdditionalEnvVars = z.preprocess((val) => {
+  if (typeof val !== "string") {
+    return val;
+  }
+
+  if (!val) {
+    return undefined;
+  }
+
+  try {
+    const result = val.split(",").reduce(
+      (acc, pair) => {
+        const [key, value] = pair.split("=");
+        if (!key || !value) {
+          return acc;
+        }
+        acc[key.trim()] = value.trim();
+        return acc;
+      },
+      {} as Record<string, string>
+    );
+
+    // Return undefined if no valid key-value pairs were found
+    return Object.keys(result).length === 0 ? undefined : result;
+  } catch (error) {
+    console.warn("Failed to parse additional env vars", { error, val });
+    return undefined;
+  }
+}, z.record(z.string(), z.string()).optional());