From f06daa8a3bbb8d784a955ac1cccab81a0a7e092c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 21 Sep 2026 15:39:16 +0000 Subject: [PATCH 1/2] fix(stack): retry transient artifact transfer failures Native artifact preparation failed on the first non-2xx response, so a single gateway error from the release host aborted a whole stack start. Retry rate limits, gateway errors, and dropped transfers with jittered backoff, and keep checksum mismatches and missing assets terminal. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019zn4vRJ2rkx5UXwNS1ndnj --- packages/stack/src/preparation/Errors.ts | 1 + .../src/preparation/SlimServicesSource.ts | 77 ++++++++++++------- .../slim-services.integration.test.ts | 51 ++++++++++++ 3 files changed, 103 insertions(+), 26 deletions(-) diff --git a/packages/stack/src/preparation/Errors.ts b/packages/stack/src/preparation/Errors.ts index ce5272c419..2d59b1afe2 100644 --- a/packages/stack/src/preparation/Errors.ts +++ b/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( diff --git a/packages/stack/src/preparation/SlimServicesSource.ts b/packages/stack/src/preparation/SlimServicesSource.ts index 8e9a7a48e4..e4caf2c761 100644 --- a/packages/stack/src/preparation/SlimServicesSource.ts +++ b/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"; @@ -80,16 +80,37 @@ 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; }), ), ); +/** Release hosts answer rate limits, gateway errors, and dropped transfers that a later attempt resolves. */ +const transferFault = (error: unknown): boolean => + error instanceof PreparationError + ? error.status !== undefined && + (error.status === 408 || error.status === 429 || error.status >= 500) + : HttpClientError.isHttpClientError(error); + +const transferSchedule = Schedule.exponential("500 millis").pipe( + Schedule.jittered, + Schedule.upTo({ times: 4 }), +); + +const withTransferRetry = (effect: Effect.Effect): Effect.Effect => + Effect.retry(effect, { schedule: transferSchedule, while: transferFault }); + const fetchBytes = Effect.fn("SlimServicesSource.fetchBytes")(function* (url: string) { - return yield* responseFor(url).pipe( - Effect.flatMap((response) => response.arrayBuffer), - Effect.map((bytes) => new Uint8Array(bytes)), + return yield* withTransferRetry( + responseFor(url).pipe( + Effect.flatMap((response) => response.arrayBuffer), + Effect.map((bytes) => new Uint8Array(bytes)), + ), + ).pipe( Effect.mapError( (cause) => new PreparationError({ message: `Unable to download ${url}`, cause }), ), @@ -132,27 +153,31 @@ const downloadToFile = Effect.fn("SlimServicesSource.downloadToFile")(function* expectedSha256: string, ) { 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* withTransferRetry(transfer); if (actual !== expectedSha256.toLowerCase()) return yield* new PreparationError({ message: `expected ${expectedSha256}, got ${actual}`, diff --git a/packages/stack/src/preparation/slim-services.integration.test.ts b/packages/stack/src/preparation/slim-services.integration.test.ts index 9b7d882013..0c982ebd2a 100644 --- a/packages/stack/src/preparation/slim-services.integration.test.ts +++ b/packages/stack/src/preparation/slim-services.integration.test.ts @@ -188,6 +188,57 @@ describe("slim-services artifact source", () => { ), ); + it.live("retries gateway failures and stops on a missing asset", () => + 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(); + const count = (url: string): number => { + const next = (attempts.get(url) ?? 0) + 1; + attempts.set(url, next); + return next; + }; + 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("", { status: 503 }) : new Response(archive), + ); + }; + const fs = yield* FileSystem.FileSystem; + const destination = yield* fs.makeTempDirectoryScoped({ prefix: "slim-services-retry-" }); + yield* withFetch( + flaky, + makeSlimServicesSource(() => artifact).materialize(request, destination, expected), + ); + expect(yield* fs.readFileString(`${destination}/bin/demo`)).toBe("demo"); + expect(attempts.get(artifact.downloadUrl)).toBe(2); + + const missing: FetchLike = () => Promise.resolve(new Response("", { status: 404 })); + let requests = 0; + const failed = yield* withFetch((input, init) => { + requests += 1; + return missing(input, init); + }, slimServicesChecksum(artifact).pipe(Effect.exit)); + expect(errorOf(failed)).toBeInstanceOf(PreparationError); + expect(requests).toBe(1); + }).pipe(Effect.provide(NodeServices.layer)), + ), + ); + it.live("rejects archive members that escape the artifact root", () => Effect.scoped( Effect.gen(function* () { From 764cf4803bda82e208ef76cff2de3462331ba66b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 21 Sep 2026 16:15:15 +0000 Subject: [PATCH 2/2] fix(stack): narrow artifact transfer retries to transient faults Retry only transport failures and transfers cut mid-body, so deterministic request faults no longer consume the schedule, and cap attempts through Effect.retry's times option. Each retried transfer logs its URL and cause. The backoff is injectable, so coverage pins the retry budget, the checksum and truncated-archive recovery paths, and single-attempt 404s without waiting on jittered delays. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019zn4vRJ2rkx5UXwNS1ndnj --- .../src/preparation/SlimServicesSource.ts | 73 +++++++++++++------ .../slim-services.integration.test.ts | 68 ++++++++++++----- 2 files changed, 100 insertions(+), 41 deletions(-) diff --git a/packages/stack/src/preparation/SlimServicesSource.ts b/packages/stack/src/preparation/SlimServicesSource.ts index e4caf2c761..faddbd0e3f 100644 --- a/packages/stack/src/preparation/SlimServicesSource.ts +++ b/packages/stack/src/preparation/SlimServicesSource.ts @@ -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 @@ -89,28 +89,43 @@ const responseFor = (url: string) => ), ); -/** Release hosts answer rate limits, gateway errors, and dropped transfers that a later attempt resolves. */ +/** + * 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); + : HttpClientError.isHttpClientError(error) && + (error.reason._tag === "TransportError" || error.reason._tag === "DecodeError"); -const transferSchedule = Schedule.exponential("500 millis").pipe( - Schedule.jittered, - Schedule.upTo({ times: 4 }), -); +/** 4 retries (5 attempts) per request: 500ms exponential, jittered. */ +const TRANSFER_MAX_RETRIES = 4; -const withTransferRetry = (effect: Effect.Effect): Effect.Effect => - Effect.retry(effect, { schedule: transferSchedule, while: transferFault }); +const transferBackoff = Schedule.exponential("500 millis").pipe(Schedule.jittered); -const fetchBytes = Effect.fn("SlimServicesSource.fetchBytes")(function* (url: string) { - return yield* withTransferRetry( - responseFor(url).pipe( - Effect.flatMap((response) => response.arrayBuffer), - Effect.map((bytes) => new Uint8Array(bytes)), - ), - ).pipe( +const withTransferRetry = + (url: string, backoff: Schedule.Schedule) => + (effect: Effect.Effect): Effect.Effect => + 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, +) { + 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 }), ), @@ -151,6 +166,7 @@ const downloadToFile = Effect.fn("SlimServicesSource.downloadToFile")(function* url: string, destination: string, expectedSha256: string, + backoff: Schedule.Schedule, ) { const fs = yield* FileSystem.FileSystem; // Each attempt reopens the sink in truncating mode, so a retry replaces any partial transfer. @@ -177,7 +193,7 @@ const downloadToFile = Effect.fn("SlimServicesSource.downloadToFile")(function* catch: (cause) => new PreparationError({ message: "Unable to finish archive digest", cause }), }); }); - const actual = yield* withTransferRetry(transfer); + const actual = yield* transfer.pipe(withTransferRetry(url, backoff)); if (actual !== expectedSha256.toLowerCase()) return yield* new PreparationError({ message: `expected ${expectedSha256}, got ${actual}`, @@ -192,8 +208,9 @@ const checksumFor = (contents: string, archiveName: string): string | undefined export const slimServicesChecksum = Effect.fn("SlimServicesSource.checksum")(function* ( artifact: SlimServicesArtifact, + backoff: Schedule.Schedule = 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`); @@ -281,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; + } = {}, ): 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, ) { @@ -294,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; @@ -307,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, @@ -350,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({ diff --git a/packages/stack/src/preparation/slim-services.integration.test.ts b/packages/stack/src/preparation/slim-services.integration.test.ts index 0c982ebd2a..919588ebba 100644 --- a/packages/stack/src/preparation/slim-services.integration.test.ts +++ b/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[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 = (fetcher: FetchLike, effect: Effect.Effect) => effect.pipe( Effect.provide(FetchHttpClient.layer), @@ -188,7 +202,7 @@ describe("slim-services artifact source", () => { ), ); - it.live("retries gateway failures and stops on a missing asset", () => + 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")); @@ -200,6 +214,13 @@ describe("slim-services artifact source", () => { attempts.set(url, next); return next; }; + const truncated = () => + new ReadableStream({ + start: (controller) => { + controller.enqueue(archive.slice(0, 4)); + controller.error("connection reset"); + }, + }); const flaky: FetchLike = (input) => { const url = requestUrl(input); if (url.endsWith("SHA256SUMS")) @@ -215,30 +236,41 @@ describe("slim-services artifact source", () => { ), ); return Promise.resolve( - count(url) === 1 ? new Response("", { status: 503 }) : new Response(archive), + count(url) === 1 ? new Response(truncated()) : new Response(archive), ); }; const fs = yield* FileSystem.FileSystem; const destination = yield* fs.makeTempDirectoryScoped({ prefix: "slim-services-retry-" }); - yield* withFetch( - flaky, - makeSlimServicesSource(() => artifact).materialize(request, destination, expected), - ); + 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); - - const missing: FetchLike = () => Promise.resolve(new Response("", { status: 404 })); - let requests = 0; - const failed = yield* withFetch((input, init) => { - requests += 1; - return missing(input, init); - }, slimServicesChecksum(artifact).pipe(Effect.exit)); - expect(errorOf(failed)).toBeInstanceOf(PreparationError); - expect(requests).toBe(1); }).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* () { @@ -545,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,