feat(supervisor): add RedisBackpressureSignalSource

Reads the backpressure verdict from a Redis key (written by the cluster-side
aggregator). Malformed or wrong-shaped values are treated as unknown so the
monitor fails open. Adds @internal/redis + @internal/testcontainers deps.

Author: nicktrn

apps/supervisor/package.json

@@ -15,6 +15,7 @@
   "dependencies": {
     "@aws-sdk/client-ecr": "^3.839.0",
     "@internal/compute": "workspace:*",
+    "@internal/redis": "workspace:*",
     "@kubernetes/client-node": "^1.0.0",
     "@trigger.dev/core": "workspace:*",
     "dockerode": "^4.0.6",
@@ -25,6 +26,7 @@
     "zod": "3.25.76"
   },
   "devDependencies": {
+    "@internal/testcontainers": "workspace:*",
     "@types/dockerode": "^3.3.33"
   }
 }

apps/supervisor/src/backpressure/redisBackpressureSignalSource.test.ts

@@ -0,0 +1,62 @@
+import { redisTest } from "@internal/testcontainers";
+import { Redis } from "@internal/redis";
+import { describe, expect } from "vitest";
+import { RedisBackpressureSignalSource } from "./redisBackpressureSignalSource.js";
+
+const KEY = "backpressure:test";
+
+describe("RedisBackpressureSignalSource", () => {
+  redisTest("returns null when the key is absent", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+    try {
+      const source = new RedisBackpressureSignalSource(redis, KEY);
+      expect(await source.read()).toBeNull();
+    } finally {
+      await redis.quit();
+    }
+  });
+
+  redisTest("parses a valid engaged verdict", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+    try {
+      await redis.set(KEY, JSON.stringify({ engaged: true, ts: 1_700_000_000_000 }));
+      const source = new RedisBackpressureSignalSource(redis, KEY);
+      expect(await source.read()).toEqual({ engaged: true, ts: 1_700_000_000_000 });
+    } finally {
+      await redis.quit();
+    }
+  });
+
+  redisTest("parses a clear verdict", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+    try {
+      await redis.set(KEY, JSON.stringify({ engaged: false }));
+      const source = new RedisBackpressureSignalSource(redis, KEY);
+      expect(await source.read()).toEqual({ engaged: false });
+    } finally {
+      await redis.quit();
+    }
+  });
+
+  redisTest("returns null for malformed JSON (fail-open)", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+    try {
+      await redis.set(KEY, "not json {");
+      const source = new RedisBackpressureSignalSource(redis, KEY);
+      expect(await source.read()).toBeNull();
+    } finally {
+      await redis.quit();
+    }
+  });
+
+  redisTest("returns null for valid JSON of the wrong shape (fail-open)", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+    try {
+      await redis.set(KEY, JSON.stringify({ foo: "bar" }));
+      const source = new RedisBackpressureSignalSource(redis, KEY);
+      expect(await source.read()).toBeNull();
+    } finally {
+      await redis.quit();
+    }
+  });
+});

apps/supervisor/src/backpressure/redisBackpressureSignalSource.ts

@@ -0,0 +1,35 @@
+import type { Redis } from "@internal/redis";
+import { z } from "zod";
+import type { BackpressureSignalSource, BackpressureVerdict } from "./backpressureMonitor.js";
+
+const VerdictSchema = z.object({
+  engaged: z.boolean(),
+  ts: z.number().optional(),
+});
+
+/** Reads the backpressure verdict from a Redis key written by the cluster-side aggregator. */
+export class RedisBackpressureSignalSource implements BackpressureSignalSource {
+  constructor(
+    private readonly redis: Redis,
+    private readonly key: string
+  ) {}
+
+  async read(): Promise<BackpressureVerdict | null> {
+    const raw = await this.redis.get(this.key);
+    if (raw === null) {
+      return null;
+    }
+
+    // A malformed or wrong-shaped value is treated as unknown (null) so the
+    // monitor fails open rather than acting on garbage.
+    let json: unknown;
+    try {
+      json = JSON.parse(raw);
+    } catch {
+      return null;
+    }
+
+    const parsed = VerdictSchema.safeParse(json);
+    return parsed.success ? parsed.data : null;
+  }
+}