fix(aggregator): copilot review fixes

Four findings on PR #972; addressing all.

1. `consecutiveFailures` semantic mismatch (jetstream-ingestor.ts +
   records-do.ts). Docstring claimed "0 means the most recent attempt
   produced events", but the run loop unconditionally incremented after
   each connectAndConsume return. So the counter was always ≥ 1 after
   any disconnect, including ones that successfully streamed events.
   Fix: increment only when no progress was made; reset to 0 (without
   increment) when progress was. Also added a regression test that
   asserts the counter stays 0 across three connect-disconnect cycles
   that all produced events.

2. `/_admin/start` returned the DO's status body (cursor + failure
   count). Even an idempotent admin endpoint shouldn't leak operational
   data to anonymous callers. Fix: route now fires the DO fetch via
   `ctx.waitUntil` and returns a fixed 204 — caller learns nothing
   about whether the DO was already running, just woke up, or is
   mid-startup. The DO's fetch handler still returns the status body
   (used internally by the cron liveness pump, which doesn't proxy it
   either).

3. Unhandled rejection in `wrapAtcuteSubscription.close()`. `void
   inner?.return?.()` suppressed the value but did NOT catch
   rejections. If the inner iterator's cleanup ever rejects (today it
   shouldn't, but a future EventIterator change could), workerd would
   surface an unhandled-promise warning. Fix: chain `.catch(() => {})`.

Tests: 14 (was 13; added counter-semantics regression). 0 lint, 0
typecheck.

Author: ascorbic

apps/aggregator/src/index.ts

@@ -23,22 +23,27 @@ export { RecordsJetstreamDO } from "./records-do.js";
 /**
  * Operational bootstrap route. Hitting `/_admin/start` once after deploy
  * spins up the Records DO, which opens its outbound WebSocket and starts
- * ingesting. The DO's WebSocket keeps it alive thereafter; this route is
- * idempotent — calling it on an already-running DO just returns its current
- * status. Recommended deploy hook:
+ * ingesting. The DO's WebSocket keeps it alive thereafter. The route is
+ * unauthenticated but returns no operational detail — just a fixed 204 —
+ * so a probing caller learns nothing useful. The action is idempotent on
+ * an already-running DO. Recommended deploy hook:
  *
- *   wrangler deploy && curl https://api.emdashcms.com/_admin/start
+ *   wrangler deploy && curl -X POST https://api.emdashcms.com/_admin/start
  */
 const BOOTSTRAP_PATH = "/_admin/start";
 
 export default {
 	async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
 		const url = new URL(request.url);
 		if (url.pathname === BOOTSTRAP_PATH) {
-			return bootstrapRecordsDo(env);
+			const id = env.RECORDS_DO.idFromName(RECORDS_DO_NAME);
+			const stub = env.RECORDS_DO.get(id);
+			// Fire-and-forget so the response shape doesn't depend on the
+			// DO's status output. Caller gets the same 204 whether the DO
+			// was already running, just woke up, or is mid-startup.
+			ctx.waitUntil(stub.fetch("https://do.internal/bootstrap"));
+			return new Response(null, { status: 204 });
 		}
-		// Suppress unused-arg lint until the XRPC routes land.
-		void ctx;
 		return new Response("emdash-aggregator: not yet implemented", {
 			status: 503,
 			headers: { "content-type": "text/plain" },
@@ -61,9 +66,3 @@ export default {
 		ctx.waitUntil(stub.fetch("https://do.internal/liveness"));
 	},
 };
-
-async function bootstrapRecordsDo(env: Env): Promise<Response> {
-	const id = env.RECORDS_DO.idFromName(RECORDS_DO_NAME);
-	const stub = env.RECORDS_DO.get(id);
-	return stub.fetch("https://do.internal/bootstrap");
-}

apps/aggregator/src/jetstream-client.ts

@@ -121,7 +121,11 @@ export function wrapAtcuteSubscription<E extends { kind: string }>(
 		},
 		close: () => {
 			fireClosed();
-			void inner?.return?.();
+			// `.catch` swallows rejections from the inner iterator's cleanup
+			// (an EventIterator's `return()` shouldn't reject, but a future
+			// implementation could). Without this, a rejection here would
+			// surface as an unhandled-promise warning in workerd.
+			inner?.return?.()?.catch(() => {});
 		},
 		[Symbol.asyncIterator](): AsyncIterator<JetstreamCommitEvent> {
 			inner ??= sub[Symbol.asyncIterator]();

apps/aggregator/src/jetstream-ingestor.ts

@@ -135,16 +135,15 @@ export class JetstreamIngestor {
 		while (!this.stopped) {
 			try {
 				await this.connectAndConsume();
-				// Subscription ended cleanly (Jetstream closed the socket
-				// without error). Treat as a soft failure for backoff
-				// purposes — but if we successfully consumed events during
-				// the connection, reset the counter first so the backoff
-				// reflects the latest streak, not historical failures.
+				// Subscription ended cleanly. If we consumed at least one
+				// event, the connection was healthy — reset the counter and
+				// reconnect with the floor delay. Otherwise treat as a soft
+				// failure and grow the backoff.
 				if (this.madeProgress) this._consecutiveFailures = 0;
-				this._consecutiveFailures += 1;
+				else this._consecutiveFailures += 1;
 			} catch (err) {
 				if (this.madeProgress) this._consecutiveFailures = 0;
-				this._consecutiveFailures += 1;
+				else this._consecutiveFailures += 1;
 				this.logger.warn?.("jetstream subscription failed", {
 					error: err instanceof Error ? err.message : String(err),
 					consecutiveFailures: this._consecutiveFailures,

apps/aggregator/src/records-do.ts

@@ -45,12 +45,17 @@ export class RecordsJetstreamDO extends DurableObject<Env> {
 	}
 
 	/**
-	 * Status surface used by `/_admin/start` (post-deploy bootstrap) and the
-	 * 5-minute cron liveness pump. Idempotent — calling it on an
-	 * already-running DO just reports the current cursor and consecutive
-	 * failure count, which is the real liveness signal: 0 means the most
-	 * recent connection attempt produced events; a high value indicates
-	 * Jetstream is unreachable or the wantedCollections filter is wrong.
+	 * Status surface for the `/_admin/start` bootstrap and the 5-minute cron
+	 * liveness pump. Idempotent — calling it on an already-running DO just
+	 * reports the current cursor and consecutive-failure count. `0` means
+	 * the most recent connection attempt produced at least one event; a
+	 * non-zero value indicates the latest reconnect cycle hasn't yet
+	 * delivered an event (Jetstream unreachable, wantedCollections
+	 * mismatch, or queue backpressure).
+	 *
+	 * The bootstrap route in the worker doesn't proxy this body — it
+	 * fires-and-forgets the DO fetch and returns 204 — so this surface is
+	 * effectively internal to the DO + cron pump.
 	 */
 	override async fetch(_request: Request): Promise<Response> {
 		return Response.json({

apps/aggregator/test/jetstream-ingestor.test.ts

@@ -256,6 +256,39 @@ describe("JetstreamIngestor", () => {
 		await expect(h.runPromise).resolves.toBeUndefined();
 	});
 
+	it("consecutiveFailures stays 0 across disconnect-with-events cycles", async () => {
+		// Per the documented contract: 0 means the most recent connection
+		// attempt produced at least one event. A connect → consume → close
+		// cycle must NOT bump the counter to 1.
+		const stream = new MockJetstream();
+		const queue = new InMemoryQueue();
+		const storage = new MapStorage();
+		const ingestor = new JetstreamIngestor({
+			client: new MockJetstreamClient(stream),
+			queue,
+			storage,
+			wantedCollections: [PROFILE_NSID],
+			backoff: { initialDelayMs: 1, maxDelayMs: 5, multiplier: 2, jitter: 0 },
+			sleep: () => Promise.resolve(),
+		});
+		const runPromise = ingestor.run();
+
+		// Three full cycles of: connect → emit → close. After each, the
+		// counter should still be 0 because each attempt made progress.
+		for (let i = 0; i < 3; i++) {
+			stream.emitCommit({ did: TEST_DID, collection: PROFILE_NSID, rkey: `r${i}` });
+			await waitFor(() => queue.jobs.length === i + 1, `event ${i}`);
+			stream.closeAll();
+			// Yield enough microtasks for the run loop to process the close
+			// and complete its bookkeeping before we inspect.
+			await new Promise<void>((r) => setTimeout(r, 5));
+			expect(ingestor.consecutiveFailures).toBe(0);
+		}
+
+		ingestor.stop();
+		await runPromise;
+	});
+
 	it("resets backoff after a successful event, even across reconnects", async () => {
 		// Without a reset, a subscription that disconnects → reconnects →
 		// processes an event → disconnects again would back off as if the