feat(stack): retry transient HTTP failures in slim services downloads

packages/stack/src/preparation/Errors.ts

@@ -11,6 +11,7 @@ interface PreparationErrorFields {
   readonly target?: string;
   readonly path?: string;
   readonly key?: string;
+  readonly status?: number;
 }
 
 export class PreparationError extends Data.TaggedError(

packages/stack/src/preparation/SlimServicesSource.ts

@@ -1,6 +1,6 @@
-import { Effect, FileSystem, Path, Schema, Stream } from "effect";
+import { Effect, FileSystem, Path, Schedule, Schema, Stream } from "effect";
 import { NodeStream } from "@effect/platform-node";
-import { HttpClient } from "effect/unstable/http";
+import { HttpClient, HttpClientError } from "effect/unstable/http";
 import { createZstdDecompress } from "node:zlib";
 import { createHash } from "node:crypto";
 import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
@@ -43,7 +43,7 @@ export interface TarBoundary {
 }
 
 /** The system tar boundary is argv-based so archive paths never enter a shell string. */
-export const systemTarBoundary: TarBoundary = {
+const systemTarBoundary: TarBoundary = {
   list: Effect.fn("SlimServicesSource.tarList")(function* (archivePath) {
     const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
     return yield* spawner
@@ -80,16 +80,52 @@ const responseFor = (url: string) =>
     Effect.flatMap((response) =>
       Effect.gen(function* () {
         if (response.status < 200 || response.status >= 300)
-          return yield* new PreparationError({ message: `HTTP ${response.status}` });
+          return yield* new PreparationError({
+            message: `HTTP ${response.status}`,
+            status: response.status,
+          });
         return response;
       }),
     ),
   );
 
-const fetchBytes = Effect.fn("SlimServicesSource.fetchBytes")(function* (url: string) {
+/**
+ * Release hosts answer rate limits, gateway errors, and dropped transfers that a later attempt
+ * resolves. A transfer cut mid-body surfaces as `DecodeError`, so it retries alongside connect
+ * failures, while deterministic request faults fail on the first attempt.
+ */
+const transferFault = (error: unknown): boolean =>
+  error instanceof PreparationError
+    ? error.status !== undefined &&
+      (error.status === 408 || error.status === 429 || error.status >= 500)
+    : HttpClientError.isHttpClientError(error) &&
+      (error.reason._tag === "TransportError" || error.reason._tag === "DecodeError");
+
+/** 4 retries (5 attempts) per request: 500ms exponential, jittered. */
+const TRANSFER_MAX_RETRIES = 4;
+
+const transferBackoff = Schedule.exponential("500 millis").pipe(Schedule.jittered);
+
+const withTransferRetry =
+  (url: string, backoff: Schedule.Schedule<unknown>) =>
+  <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
+    effect.pipe(
+      Effect.tapError((cause) =>
+        transferFault(cause)
+          ? Effect.logWarning(`Retrying slim-services transfer of ${url}`, cause)
+          : Effect.void,
+      ),
+      Effect.retry({ schedule: backoff, times: TRANSFER_MAX_RETRIES, while: transferFault }),
+    );
+
+const fetchBytes = Effect.fn("SlimServicesSource.fetchBytes")(function* (
+  url: string,
+  backoff: Schedule.Schedule<unknown>,
+) {
   return yield* responseFor(url).pipe(
     Effect.flatMap((response) => response.arrayBuffer),
     Effect.map((bytes) => new Uint8Array(bytes)),
+    withTransferRetry(url, backoff),
     Effect.mapError(
       (cause) => new PreparationError({ message: `Unable to download ${url}`, cause }),
     ),
@@ -130,29 +166,34 @@ const downloadToFile = Effect.fn("SlimServicesSource.downloadToFile")(function*
   url: string,
   destination: string,
   expectedSha256: string,
+  backoff: Schedule.Schedule<unknown>,
 ) {
   const fs = yield* FileSystem.FileSystem;
-  const response = yield* responseFor(url);
-  const hash = yield* Effect.try({
-    try: () => createHash("sha256"),
-    catch: (cause) =>
-      new PreparationError({ message: "Unable to initialize archive digest", cause }),
-  });
-  yield* response.stream.pipe(
-    Stream.tap((chunk) =>
-      Effect.try({
-        try: () => {
-          hash.update(chunk);
-        },
-        catch: (cause) => new PreparationError({ message: "Unable to hash archive", cause }),
-      }),
-    ),
-    Stream.run(fs.sink(destination, { mode: 0o600 })),
-  );
-  const actual = yield* Effect.try({
-    try: () => hash.digest("hex"),
-    catch: (cause) => new PreparationError({ message: "Unable to finish archive digest", cause }),
+  // Each attempt reopens the sink in truncating mode, so a retry replaces any partial transfer.
+  const transfer = Effect.gen(function* () {
+    const response = yield* responseFor(url);
+    const hash = yield* Effect.try({
+      try: () => createHash("sha256"),
+      catch: (cause) =>
+        new PreparationError({ message: "Unable to initialize archive digest", cause }),
+    });
+    yield* response.stream.pipe(
+      Stream.tap((chunk) =>
+        Effect.try({
+          try: () => {
+            hash.update(chunk);
+          },
+          catch: (cause) => new PreparationError({ message: "Unable to hash archive", cause }),
+        }),
+      ),
+      Stream.run(fs.sink(destination, { mode: 0o600 })),
+    );
+    return yield* Effect.try({
+      try: () => hash.digest("hex"),
+      catch: (cause) => new PreparationError({ message: "Unable to finish archive digest", cause }),
+    });
   });
+  const actual = yield* transfer.pipe(withTransferRetry(url, backoff));
   if (actual !== expectedSha256.toLowerCase())
     return yield* new PreparationError({
       message: `expected ${expectedSha256}, got ${actual}`,
@@ -167,8 +208,9 @@ const checksumFor = (contents: string, archiveName: string): string | undefined
 
 export const slimServicesChecksum = Effect.fn("SlimServicesSource.checksum")(function* (
   artifact: SlimServicesArtifact,
+  backoff: Schedule.Schedule<unknown> = transferBackoff,
 ) {
-  return yield* fetchBytes(artifact.checksumUrl).pipe(
+  return yield* fetchBytes(artifact.checksumUrl, backoff).pipe(
     Effect.map((bytes) => new TextDecoder().decode(bytes)),
     Effect.flatMap((contents) => {
       const checksum = checksumFor(contents, `${artifact.assetName}.tar.zst`);
@@ -256,9 +298,15 @@ const validateExtractedTree = (
 
 export const makeSlimServicesSource = (
   resolve: (request: ArtifactRequest) => SlimServicesArtifact | undefined,
-  tarBoundary: TarBoundary = systemTarBoundary,
-  decompressor: ZstdDecompressor = nodeZstdDecompressor,
+  overrides: {
+    readonly tarBoundary?: TarBoundary;
+    readonly decompressor?: ZstdDecompressor;
+    readonly backoff?: Schedule.Schedule<unknown>;
+  } = {},
 ): ArtifactSource => {
+  const tarBoundary = overrides.tarBoundary ?? systemTarBoundary;
+  const decompressor = overrides.decompressor ?? nodeZstdDecompressor;
+  const backoff = overrides.backoff ?? transferBackoff;
   const resolveArtifact = Effect.fn("SlimServicesSource.resolveArtifact")(function* (
     request: ArtifactRequest,
   ) {
@@ -269,7 +317,9 @@ export const makeSlimServicesSource = (
   });
   return {
     checksum: (request) =>
-      resolveArtifact(request).pipe(Effect.flatMap((artifact) => slimServicesChecksum(artifact))),
+      resolveArtifact(request).pipe(
+        Effect.flatMap((artifact) => slimServicesChecksum(artifact, backoff)),
+      ),
     materialize: Effect.fn("SlimServicesSource.materialize")(
       function* (request, destination, expectedSha256, onProgress) {
         const fs = yield* FileSystem.FileSystem;
@@ -282,7 +332,7 @@ export const makeSlimServicesSource = (
         ]);
         return yield* Effect.gen(function* () {
           const artifact = yield* resolveArtifact(request);
-          const manifestBytes = yield* fetchBytes(artifact.manifestUrl);
+          const manifestBytes = yield* fetchBytes(artifact.manifestUrl, backoff);
           const manifestText = new TextDecoder().decode(manifestBytes);
           const manifestSchema = Schema.Struct({
             service: Schema.String,
@@ -325,7 +375,9 @@ export const makeSlimServicesSource = (
               version: artifact.version,
             });
           yield* Effect.sync(() => onProgress?.("downloading")).pipe(
-            Effect.andThen(downloadToFile(artifact.downloadUrl, compressedPath, expectedSha256)),
+            Effect.andThen(
+              downloadToFile(artifact.downloadUrl, compressedPath, expectedSha256, backoff),
+            ),
             Effect.mapError(
               (cause) =>
                 new PreparationError({

packages/stack/src/preparation/slim-services.integration.test.ts

@@ -1,6 +1,18 @@
 import { NodeHttpClient, NodeServices } from "@effect/platform-node";
 import { describe, expect, it } from "@effect/vitest";
-import { Cause, Crypto, Deferred, Effect, Exit, Fiber, FileSystem, Layer, Option } from "effect";
+import {
+  Cause,
+  Crypto,
+  Deferred,
+  Duration,
+  Effect,
+  Exit,
+  Fiber,
+  FileSystem,
+  Layer,
+  Option,
+  Schedule,
+} from "effect";
 // oxlint-disable-next-line effecttsgo/node-builtin-import -- the redirect test owns a local native listener.
 import { createServer, type Server } from "node:http";
 import { zstdCompress } from "node:zlib";
@@ -10,7 +22,6 @@ import { digestHex } from "./Integrity.ts";
 import {
   makeSlimServicesSource,
   slimServicesChecksum,
-  systemTarBoundary,
   type SlimServicesArtifact,
   type ZstdDecompressor,
 } from "./SlimServicesSource.ts";
@@ -113,6 +124,9 @@ type FetchLike = (
 const requestUrl = (input: Parameters<typeof fetch>[0]): string =>
   typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url;
 
+/** Keeps retry coverage off the production backoff, whose jittered delays would idle the suite. */
+const immediate = Schedule.spaced(Duration.zero);
+
 const withFetch = <A, E, R>(fetcher: FetchLike, effect: Effect.Effect<A, E, R>) =>
   effect.pipe(
     Effect.provide(FetchHttpClient.layer),
@@ -188,6 +202,75 @@ describe("slim-services artifact source", () => {
     ),
   );
 
+  it.live("retries a gateway error on the checksum and a truncated archive transfer", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const archive = yield* compress(tar("bin/demo", "demo"));
+        const crypto = yield* Crypto.Crypto;
+        const expected = digestHex(yield* crypto.digest("SHA-256", archive));
+        const attempts = new Map<string, number>();
+        const count = (url: string): number => {
+          const next = (attempts.get(url) ?? 0) + 1;
+          attempts.set(url, next);
+          return next;
+        };
+        const truncated = () =>
+          new ReadableStream<Uint8Array>({
+            start: (controller) => {
+              controller.enqueue(archive.slice(0, 4));
+              controller.error("connection reset");
+            },
+          });
+        const flaky: FetchLike = (input) => {
+          const url = requestUrl(input);
+          if (url.endsWith("SHA256SUMS"))
+            return Promise.resolve(
+              count(url) === 1
+                ? new Response("", { status: 504 })
+                : new Response(`${expected}  demo-v1.0.0-linux-amd64.tar.zst\n`),
+            );
+          if (url.endsWith("manifest.json"))
+            return Promise.resolve(
+              new Response(
+                JSON.stringify({ service: "demo", version: "v1.0.0", target: "linux-amd64" }),
+              ),
+            );
+          return Promise.resolve(
+            count(url) === 1 ? new Response(truncated()) : new Response(archive),
+          );
+        };
+        const fs = yield* FileSystem.FileSystem;
+        const destination = yield* fs.makeTempDirectoryScoped({ prefix: "slim-services-retry-" });
+        const source = makeSlimServicesSource(() => artifact, { backoff: immediate });
+        expect(yield* withFetch(flaky, source.checksum(request))).toBe(expected);
+        expect(attempts.get(artifact.checksumUrl)).toBe(2);
+        yield* withFetch(flaky, source.materialize(request, destination, expected));
+        expect(yield* fs.readFileString(`${destination}/bin/demo`)).toBe("demo");
+        expect(attempts.get(artifact.downloadUrl)).toBe(2);
+      }).pipe(Effect.provide(NodeServices.layer)),
+    ),
+  );
+
+  it.live("spends five attempts on a persistent gateway error and one on a missing asset", () =>
+    Effect.gen(function* () {
+      let gateway = 0;
+      const exhausted = yield* withFetch(() => {
+        gateway += 1;
+        return Promise.resolve(new Response("", { status: 504 }));
+      }, slimServicesChecksum(artifact, immediate).pipe(Effect.exit));
+      expect(errorOf(exhausted)).toBeInstanceOf(PreparationError);
+      expect(gateway).toBe(5);
+
+      let missing = 0;
+      const failed = yield* withFetch(() => {
+        missing += 1;
+        return Promise.resolve(new Response("", { status: 404 }));
+      }, slimServicesChecksum(artifact, immediate).pipe(Effect.exit));
+      expect(errorOf(failed)).toBeInstanceOf(PreparationError);
+      expect(missing).toBe(1);
+    }),
+  );
+
   it.live("rejects archive members that escape the artifact root", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -494,7 +577,7 @@ describe("slim-services artifact source", () => {
         const fiber = yield* Effect.forkChild(
           withFetch(
             fetcher,
-            makeSlimServicesSource(() => artifact, systemTarBoundary, decompressor).materialize(
+            makeSlimServicesSource(() => artifact, { decompressor }).materialize(
               request,
               destination,
               expected,