fix(stack): include recent service output in startup failure errors

packages/stack/src/services/ProcessRecipe.integration.test.ts

@@ -0,0 +1,160 @@
+import { NodeHttpClient, NodeServices } from "@effect/platform-node";
+import { describe, expect, it } from "@effect/vitest";
+import { Crypto, Deferred, Effect, Fiber, FileSystem, Layer, Path, Sink, Stream } from "effect";
+import { TestClock } from "effect/testing";
+import { HttpClient } from "effect/unstable/http";
+import { ChildProcessSpawner } from "effect/unstable/process";
+import type { ContainerRuntime } from "../runtime/Container.ts";
+import { makeService } from "../Service.ts";
+import { makeProcessRecipe } from "./ProcessRecipe.ts";
+import * as Realtime from "./Realtime.ts";
+
+const encode = (text: string) => new TextEncoder().encode(text);
+
+const poolTimeout =
+  "** (DBConnection.ConnectionError) connection not available and request was dropped from queue";
+
+const postgrexFailure =
+  '[error] Postgrex.Protocol ("db_conn_1") failed to connect: ** (DBConnection.ConnectionError) tcp connect (host.docker.internal:54322): network is unreachable - :enetunreach';
+
+const startupContainer = (tool: {
+  readonly stdout: Stream.Stream<string>;
+  readonly stderr: Stream.Stream<string>;
+  readonly exitCode: Effect.Effect<number>;
+}): ContainerRuntime => ({
+  prepare: () => Effect.void,
+  prepareImage: (image) => Effect.succeed(image),
+  launch: () => Effect.die("the main process must not launch after a failed startup"),
+  launchTool: () =>
+    Effect.succeed({
+      id: "startup-tool",
+      ports: {},
+      stdout: tool.stdout.pipe(Stream.map(encode)),
+      stderr: tool.stderr.pipe(Stream.map(encode)),
+      exitCode: tool.exitCode,
+      stdin: Sink.drain,
+      stop: Effect.void,
+      discard: Effect.void,
+      kill: Effect.void,
+      remove: Effect.void,
+    }),
+});
+
+const realtimeService = Effect.fn(function* (container: ContainerRuntime) {
+  const creation: Realtime.Creation = {
+    service: "realtime",
+    config: { databaseUrl: "postgresql://postgres:postgres@host.docker.internal:54322/postgres" },
+  };
+  const recipe = yield* makeProcessRecipe(
+    creation,
+    {
+      stackId: "process-recipe-test",
+      instanceId: "instance",
+      root: "/unused",
+      cacheRoot: "/unused/cache",
+      runtime: "docker",
+    },
+    {
+      fs: yield* FileSystem.FileSystem,
+      path: yield* Path.Path,
+      crypto: yield* Crypto.Crypto,
+      client: yield* HttpClient.HttpClient,
+      spawner: yield* ChildProcessSpawner.ChildProcessSpawner,
+      container,
+    },
+    Realtime.makeSpec(),
+  );
+  return yield* makeService(recipe.definition, { id: "realtime", config: creation });
+});
+
+const platform = Layer.merge(NodeServices.layer, NodeHttpClient.layerNodeHttp);
+
+describe("process recipe startup", () => {
+  it.effect("reports the startup process's recent stdout and stderr when it exits non-zero", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const realtime = yield* realtimeService(
+          startupContainer({
+            stdout: Stream.make(
+              "[info] Running migrations\n",
+              postgrexFailure.slice(0, 60),
+              `${postgrexFailure.slice(60)}\n`,
+            ),
+            stderr: Stream.make(`${poolTimeout}\n`),
+            exitCode: Effect.succeed(1),
+          }),
+        );
+
+        const error = yield* Effect.flip(realtime.start);
+
+        expect(error.message).toContain("realtime startup exited with 1");
+        expect(error.message).toContain(postgrexFailure);
+        expect(error.message).toContain(poolTimeout);
+      }),
+    ).pipe(Effect.provide(platform)),
+  );
+
+  it.effect("keeps the stderr error when later stdout exceeds the tail", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const stderrWritten = yield* Deferred.make<void>();
+        const noise = Array.from({ length: 30 }, (_, index) => `[info] shutdown step ${index}\n`);
+        const realtime = yield* realtimeService(
+          startupContainer({
+            stdout: Stream.fromEffectDrain(Deferred.await(stderrWritten)).pipe(
+              Stream.concat(Stream.fromIterable(noise)),
+            ),
+            stderr: Stream.make(`${poolTimeout}\n`).pipe(
+              Stream.concat(Stream.fromEffectDrain(Deferred.succeed(stderrWritten, undefined))),
+            ),
+            exitCode: Effect.succeed(1),
+          }),
+        );
+
+        const error = yield* Effect.flip(realtime.start);
+
+        expect(error.message).toContain(poolTimeout);
+        expect(error.message).toContain("[info] shutdown step 29");
+        expect(error.message).not.toContain("[info] shutdown step 9\n");
+      }),
+    ).pipe(Effect.provide(platform)),
+  );
+
+  it.effect(
+    "reports the startup process's recent output, including an unterminated line, when it never exits",
+    () =>
+      Effect.scoped(
+        Effect.gen(function* () {
+          const running = yield* Deferred.make<void>();
+          const realtime = yield* realtimeService(
+            startupContainer({
+              stdout: Stream.make(
+                `${postgrexFailure}\n`,
+                "[info] ",
+                "😀".repeat(2_500),
+                " Retrying database connection",
+              ).pipe(
+                Stream.concat(Stream.fromEffectDrain(Deferred.succeed(running, undefined))),
+                Stream.concat(Stream.never),
+              ),
+              stderr: Stream.never,
+              exitCode: Effect.never,
+            }),
+          );
+
+          const failure = yield* realtime.start.pipe(Effect.flip, Effect.forkChild);
+          yield* Deferred.await(running);
+          yield* Effect.yieldNow;
+          yield* TestClock.adjust("60 seconds");
+          const error = yield* Fiber.join(failure);
+
+          expect(error.message).toContain("realtime startup timed out after 60 seconds");
+          expect(error.message).toContain(postgrexFailure);
+          expect(error.message).toContain("😀 Retrying database connection");
+          expect(error.message).not.toContain("[info] 😀");
+          expect(error.message).not.toMatch(/…[\uDC00-\uDFFF]/);
+          expect(error.message.length).toBeLessThan(2_000);
+        }),
+      ).pipe(Effect.provide(platform)),
+  );
+});

packages/stack/src/services/ProcessRecipe.ts

@@ -1,9 +1,11 @@
 import {
   Cause,
   Crypto,
+  Duration,
   Effect,
   Exit,
   FileSystem,
+  Option,
   Path,
   PubSub,
   Ref,
@@ -201,53 +203,100 @@ const publishLogs = Effect.fn("ProcessRecipe.publishLogs")((
   );
 });
 
+const startupTimeoutSeconds = 60;
+const startupOutputTailLines = 20;
+const startupOutputLineChars = 1_000;
+
+type StartupOutput = Readonly<Record<CatalogLog["stream"], ReadonlyArray<string>>>;
+
+const clipLine = (line: string) =>
+  line.length > startupOutputLineChars
+    ? `…${line.slice(-startupOutputLineChars).replace(/^[\uDC00-\uDFFF]/, "")}`
+    : line;
+
+const withRecentOutput = (summary: string, output: StartupOutput) =>
+  [
+    summary,
+    ...(["stdout", "stderr"] as const)
+      .filter((name) => output[name].length > 0)
+      .map((name) => `Recent ${name}:\n${output[name].join("\n")}`),
+  ].join("\n");
+
 const awaitStartup = Effect.fn("ProcessRecipe.awaitStartup")(
   (
+    service: ServiceKind,
     process: {
       readonly stdout: Stream.Stream<Uint8Array, NativeProcessError | ContainerError>;
       readonly stderr: Stream.Stream<Uint8Array, NativeProcessError | ContainerError>;
       readonly exitCode: Effect.Effect<number, NativeProcessError | ContainerError>;
     },
     logs: PubSub.PubSub<CatalogLog>,
-  ): Effect.Effect<Readonly<{ readonly code: number; readonly stderr: string }>, ServiceError> =>
+  ): Effect.Effect<
+    Readonly<{ readonly code: number; readonly output: StartupOutput }>,
+    ServiceError
+  > =>
     Effect.gen(function* () {
-      const [stdout, stderr, exitCode] = yield* Effect.all(
-        [
-          process.stdout.pipe(
-            Stream.decodeText,
-            Stream.runFold(
-              () => "",
-              (text, chunk) => text + chunk,
-            ),
-          ),
-          process.stderr.pipe(
-            Stream.decodeText,
-            Stream.runFold(
-              () => "",
-              (text, chunk) => text + chunk,
+      const collect = Effect.fnUntraced(function* (
+        stream: Stream.Stream<Uint8Array, NativeProcessError | ContainerError>,
+        name: CatalogLog["stream"],
+        tail: Ref.Ref<ReadonlyArray<string>>,
+      ) {
+        const appendLines = (lines: ReadonlyArray<string>) =>
+          Ref.update(tail, (current) =>
+            [...current, ...lines.filter((line) => line.trim().length > 0).map(clipLine)].slice(
+              -startupOutputTailLines,
             ),
+          );
+        const partial = yield* Ref.make("");
+        // The unterminated last line is flushed on interruption so a timeout still reports it.
+        yield* stream.pipe(
+          Stream.tap((bytes) => PubSub.publish(logs, { stream: name, bytes })),
+          Stream.decodeText,
+          Stream.runForEach((text) =>
+            Ref.modify(partial, (rest): [ReadonlyArray<string>, string] => {
+              const lines = `${rest}${text}`.split(/\r?\n/);
+              const next = lines.pop() ?? "";
+              return [lines, next.slice(-(startupOutputLineChars + 1))];
+            }).pipe(Effect.flatMap(appendLines)),
           ),
+          Effect.ensuring(Ref.get(partial).pipe(Effect.flatMap((rest) => appendLines([rest])))),
+        );
+      });
+      const stdout = yield* Ref.make<ReadonlyArray<string>>([]);
+      const stderr = yield* Ref.make<ReadonlyArray<string>>([]);
+      const completed = yield* Effect.all(
+        [
+          collect(process.stdout, "stdout", stdout),
+          collect(process.stderr, "stderr", stderr),
           process.exitCode,
         ],
         { concurrency: "unbounded" },
-      ).pipe(Effect.mapError((cause) => serviceError("launch", cause)));
-      if (stdout.length > 0)
-        yield* PubSub.publish(logs, {
-          stream: "stdout",
-          bytes: new TextEncoder().encode(stdout),
-        });
-      if (stderr.length > 0)
-        yield* PubSub.publish(logs, {
-          stream: "stderr",
-          bytes: new TextEncoder().encode(stderr),
-        });
-      return { code: Number(exitCode), stderr };
-    }).pipe(
-      Effect.timeout("60 seconds"),
-      Effect.mapError((cause) => serviceError("launch", cause)),
-    ),
+      ).pipe(
+        Effect.mapError((cause) => serviceError("launch", cause)),
+        Effect.timeoutOption(Duration.seconds(startupTimeoutSeconds)),
+      );
+      const output = { stdout: yield* Ref.get(stdout), stderr: yield* Ref.get(stderr) };
+      if (Option.isNone(completed))
+        return yield* serviceError(
+          "launch",
+          withRecentOutput(
+            `${service} startup timed out after ${startupTimeoutSeconds} seconds`,
+            output,
+          ),
+        );
+      return { code: Number(completed.value[2]), output };
+    }),
 );
 
+const startupFailure = (
+  service: ServiceKind,
+  result: { readonly code: number; readonly output: StartupOutput },
+): ServiceError =>
+  serviceError(
+    "launch",
+    withRecentOutput(`${service} startup exited with ${result.code}`, result.output),
+  );
+
 const readiness = Effect.fn("ProcessRecipe.readiness")(
   (
     client: HttpClient.HttpClient,
@@ -355,12 +404,8 @@ export const makeProcessRecipe = <C extends RecipeCreation<ServiceKind, unknown>
             Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, deps.spawner),
             Effect.mapError((cause) => serviceError("launch", cause)),
           );
-          const result = yield* awaitStartup(startupProcess, logs);
-          if (result.code !== 0)
-            return yield* serviceError(
-              "launch",
-              `${context.config.service} startup exited with ${result.code}: ${result.stderr.trim()}`,
-            );
+          const result = yield* awaitStartup(context.config.service, startupProcess, logs);
+          if (result.code !== 0) return yield* startupFailure(context.config.service, result);
         }
         const native: NativeProcess = yield* spawnNativeProcess(
           {
@@ -420,15 +465,11 @@ export const makeProcessRecipe = <C extends RecipeCreation<ServiceKind, unknown>
             Effect.mapError((cause) => serviceError("launch", cause)),
             Scope.provide(context.scope),
           );
-        const result = yield* awaitStartup(startupProcess, logs);
+        const result = yield* awaitStartup(context.config.service, startupProcess, logs);
         yield* startupProcess.remove.pipe(
           Effect.mapError((cause) => serviceError("launch", cause)),
         );
-        if (result.code !== 0)
-          return yield* serviceError(
-            "launch",
-            `${context.config.service} startup exited with ${result.code}: ${result.stderr.trim()}`,
-          );
+        if (result.code !== 0) return yield* startupFailure(context.config.service, result);
       }
       const launched = yield* deps.container
         .launch({