fix(hub): keep the live app running through a failed check or bad line

apps/hub/src/config/live-app.ts

@@ -5,6 +5,7 @@ export const liveAppWording = {
     "The app is not answering. The Live Preview tab in Builder says why.",
   partStopped: "The app's server stopped.",
   stopped: "The app stopped.",
+  checkFailed: "Checking on the app failed.",
   commandFailed: (command: string) => `\`${command}\` failed.`,
   hostExited: (code: number) =>
     `The process running the app exited with code ${String(code)}.`,

apps/hub/src/live-app/host-report.test.ts

@@ -0,0 +1,50 @@
+import { expect, test } from "bun:test";
+import { liveAppWording } from "../config/live-app.ts";
+import { hostReports, reportState, type HostReport } from "./host-report.ts";
+
+test("a state check that throws reports failed, and the next check still reports", async () => {
+  const reported: HostReport[] = [];
+  const failures: unknown[] = [];
+  const checks = [
+    () => Promise.reject(new Error("executor_failed")),
+    () => Promise.resolve<HostReport>({ state: "running", url: "http://app" }),
+  ];
+  for (const check of checks)
+    await reportState(
+      check,
+      (state) => reported.push(state),
+      (error) => failures.push(error),
+    );
+  expect(reported).toEqual([
+    { state: "failed", error: liveAppWording.checkFailed, log: [] },
+    { state: "running", url: "http://app" },
+  ]);
+  expect(failures).toHaveLength(1);
+});
+
+test("a bad state line between two good ones is skipped, and the state ends on the second", async () => {
+  const lines = [
+    '{"state":"starting","phase":"installing","log":[]}\n',
+    "not json\n",
+    '{"state":"running"}\n',
+    '{"state":"running","url":"http://app"}\n',
+  ];
+  const stdout = new ReadableStream<Uint8Array>({
+    start(controller) {
+      // Split mid-line so a report spanning chunks still parses.
+      const bytes = new TextEncoder().encode(lines.join(""));
+      controller.enqueue(bytes.slice(0, 10));
+      controller.enqueue(bytes.slice(10));
+      controller.close();
+    },
+  });
+  const skipped: string[] = [];
+  const reports: HostReport[] = [];
+  for await (const report of hostReports(stdout, (reason) =>
+    skipped.push(reason),
+  ))
+    reports.push(report);
+  expect(skipped).toEqual(["not JSON", "not a report"]);
+  expect(reports.at(-1)).toEqual({ state: "running", url: "http://app" });
+  expect(reports).toHaveLength(2);
+});

apps/hub/src/live-app/host-report.ts

@@ -1,4 +1,6 @@
 import { type } from "arktype";
+import { liveAppWording } from "../config/live-app.ts";
+import { parseJson } from "../lib/json.ts";
 
 export const HostReport = type({
   state: "'starting'",
@@ -8,3 +10,43 @@ export const HostReport = type({
   .or({ state: "'running'", url: "string" })
   .or({ state: "'failed'", error: "string", log: "string[]" });
 export type HostReport = typeof HostReport.infer;
+
+/** Runs one state check; a check that throws reports a failed state, so one bad check never ends the host. */
+export async function reportState(
+  check: () => Promise<HostReport | undefined>,
+  report: (state: HostReport) => void,
+  failed: (error: unknown) => void,
+) {
+  let state: HostReport | undefined;
+  try {
+    state = await check();
+  } catch (error) {
+    failed(error);
+    state = { state: "failed", error: liveAppWording.checkFailed, log: [] };
+  }
+  if (state) report(state);
+}
+
+/** Each report the host writes to stdout, in order; a line that is not a report is skipped, so one bad line never freezes the state. */
+export async function* hostReports(
+  stdout: ReadableStream<Uint8Array>,
+  skipped: (reason: "not JSON" | "not a report") => void,
+): AsyncGenerator<HostReport> {
+  const decoder = new TextDecoder();
+  let pending = "";
+  for await (const chunk of stdout) {
+    pending += decoder.decode(chunk, { stream: true });
+    for (
+      let end = pending.indexOf("\n");
+      end !== -1;
+      end = pending.indexOf("\n")
+    ) {
+      const json = parseJson(pending.slice(0, end));
+      pending = pending.slice(end + 1);
+      const report = json === undefined ? undefined : HostReport(json);
+      if (report === undefined) skipped("not JSON");
+      else if (report instanceof type.errors) skipped("not a report");
+      else yield report;
+    }
+  }
+}

apps/hub/src/live-app/host.ts

@@ -12,7 +12,7 @@ import { type } from "arktype";
 import { hasCode } from "@intx/types";
 import { liveAppWording } from "../config/live-app.ts";
 import { errorCode } from "../request-error.ts";
-import type { HostReport } from "./host-report.ts";
+import { reportState, type HostReport } from "./host-report.ts";
 import { parseJson } from "../lib/json.ts";
 import { writeBundle } from "../artifacts/design-tools.ts";
 import { bundleFiles, collectApp } from "../builds/app-snapshot.ts";
@@ -271,27 +271,32 @@ async function answers(url: string) {
   }
 }
 
-setInterval(async () => {
-  if (installing)
-    return report({ state: "starting", phase: "installing", log: [] });
-  if (failure) return report({ state: "failed", ...failure });
+async function currentState(): Promise<HostReport | undefined> {
+  if (installing) return { state: "starting", phase: "installing", log: [] };
+  if (failure) return { state: "failed", ...failure };
   const app = executor.appState();
-  if (!app) return report({ state: "starting", phase: "starting", log: [] });
+  if (!app) return { state: "starting", phase: "starting", log: [] };
   if (!app.running || app.stopped !== undefined) {
     if (!restartedSinceChange) {
       restartedSinceChange = true;
       enqueue(async () => void (await executor.app()));
-      return;
+      return undefined;
     }
-    return report({
+    return {
       state: "failed",
       error: app.running ? liveAppWording.partStopped : liveAppWording.stopped,
       log: app.log,
-    });
+    };
   }
-  report(
-    (await answers(app.url))
-      ? { state: "running", url: app.url }
-      : { state: "starting", phase: "starting", log: app.log },
-  );
-}, STATE_MS);
+  return (await answers(app.url))
+    ? { state: "running", url: app.url }
+    : { state: "starting", phase: "starting", log: app.log };
+}
+
+setInterval(
+  () =>
+    void reportState(currentState, report, (error) =>
+      stderrLine("Checking on the app failed", error),
+    ),
+  STATE_MS,
+);

apps/hub/src/live-app/runner.ts

@@ -5,7 +5,7 @@ import { getLogger } from "@intx/log";
 import { errorCode, requestErrorMetadata } from "../request-error.ts";
 import { liveAppWording } from "../config/live-app.ts";
 import type { HostSource } from "./host.ts";
-import { HostReport } from "./host-report.ts";
+import { hostReports, type HostReport } from "./host-report.ts";
 import type { createLiveAppProxy } from "./proxy.ts";
 
 const log = getLogger(["corbits-builder", "hub", "live-app"]);
@@ -57,20 +57,14 @@ export function createLiveAppRunner(options: {
   sweep.unref();
 
   async function follow(projectId: string, run: Run) {
-    const decoder = new TextDecoder();
-    let pending = "";
     try {
-      for await (const chunk of run.child.stdout) {
-        pending += decoder.decode(chunk, { stream: true });
-        for (
-          let end = pending.indexOf("\n");
-          end !== -1;
-          end = pending.indexOf("\n")
-        ) {
-          run.state = HostReport.assert(JSON.parse(pending.slice(0, end)));
-          pending = pending.slice(end + 1);
-        }
-      }
+      for await (const report of hostReports(run.child.stdout, (reason) => {
+        log.warn("Skipped a live app host state line for {projectId}", {
+          projectId,
+          reason,
+        });
+      }))
+        run.state = report;
     } catch (error) {
       log.warn("Reading the live app host's state failed for {projectId}", {
         projectId,
@@ -95,9 +89,19 @@ export function createLiveAppRunner(options: {
     }
   }
 
-  async function failOnExit(projectId: string, run: Run) {
-    const code = await run.child.exited;
-    if (runs.get(projectId) === run && run.state.state !== "failed")
+  async function failOnExit(
+    projectId: string,
+    run: Run,
+    reading: Promise<void>,
+  ) {
+    // Lines the host wrote just before exiting are applied first, so the exit has the last word.
+    const [code] = await Promise.all([run.child.exited, reading]);
+    if (runs.get(projectId) !== run) return;
+    // A failed check is passing; the exit code says more about why the app is gone.
+    if (
+      run.state.state !== "failed" ||
+      run.state.error === liveAppWording.checkFailed
+    )
       run.state = {
         state: "failed",
         error: liveAppWording.hostExited(code),
@@ -135,9 +139,9 @@ export function createLiveAppRunner(options: {
       lastSeen: Date.now(),
     };
     runs.set(projectId, run);
-    void follow(projectId, run);
+    const reading = follow(projectId, run);
     void collect(projectId, run);
-    void failOnExit(projectId, run);
+    void failOnExit(projectId, run, reading);
     return run;
   }