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..faddbd0e3f 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"; @@ -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) => + (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 }), ), @@ -130,29 +166,34 @@ const downloadToFile = Effect.fn("SlimServicesSource.downloadToFile")(function* url: string, destination: string, expectedSha256: string, + backoff: Schedule.Schedule, ) { 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 = 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; + } = {}, ): 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({ diff --git a/packages/stack/src/preparation/slim-services.integration.test.ts b/packages/stack/src/preparation/slim-services.integration.test.ts index 9b7d882013..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,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(); + const count = (url: string): number => { + const next = (attempts.get(url) ?? 0) + 1; + 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")) + 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,