fix(webapp): catch change-routing failures instead of crashing the process

The realtime change router invoked its async routing tick
fire-and-forget without a catch, so a transient database error during
hydration became an unhandled promise rejection, which exits the
process. Failures are now caught and logged; affected feeds fall back
to their full-resolve backstop.

Author: ericallam

apps/webapp/app/services/realtime/envChangeRouter.server.ts

@@ -1,5 +1,6 @@
 import { type ChangeRecord } from "./runChangeNotifier.server";
 import { type RealtimeRunRow, serializeRunRow } from "./electricStreamProtocol.server";
+import { logger } from "~/services/logger.server";
 
 /**
  * EnvChangeRouter — the per-instance routing layer that turns "feeds as predicates over
@@ -182,9 +183,15 @@ export class EnvChangeRouter {
     };
     this.#envs.set(environmentId, env);
     env.unsubscribe = this.options.source.subscribeToEnv(environmentId, (records) => {
-      // Fire-and-forget; the notifier doesn't await us. Errors fall through to the feeds'
-      // backstop (a hydrate failure leaves waiters to time out into a full resolve).
-      void this.#onBatch(environmentId, env, records);
+      // Fire-and-forget; the notifier doesn't await us. A hydrate failure must be caught
+      // here (an unhandled rejection exits the process); the matched feeds' waiters stay
+      // armed and time out into the full-resolve backstop.
+      this.#onBatch(environmentId, env, records).catch((error) => {
+        logger.error("[envChangeRouter] failed to route a change batch", {
+          environmentId,
+          error,
+        });
+      });
     });
     return env;
   }

apps/webapp/test/realtime/envChangeRouter.test.ts

@@ -98,6 +98,27 @@ describe("EnvChangeRouter", () => {
     regs.forEach((r) => r.close());
   });
 
+  it("a hydrate failure doesn't reject out of the source callback; the feed times out", async () => {
+    const src = fakeSource();
+    const hydrateSpy = vi.fn<RowHydrator["hydrateByIds"]>(async () => {
+      throw new Error("replica down");
+    });
+    const router = new EnvChangeRouter({
+      source: src.source,
+      hydrator: { hydrateByIds: hydrateSpy },
+    });
+    const reg = router.register("env_1", { kind: "run", runId: "r1" }, []);
+    const wait = reg.waitForMatch(undefined, 50);
+
+    // Would be an unhandled rejection (process exit) if #onBatch's promise were unguarded.
+    src.push("env_1", [record("r1")]);
+
+    const result = await wait;
+    expect(result.reason).toBe("timeout");
+    expect(hydrateSpy).toHaveBeenCalledTimes(1);
+    reg.close();
+  });
+
   it("routes a run feed by exact runId", async () => {
     const rows = new Map([["r1", row("r1")]]);
     const { router, src } = makeRouter(rows);