diff --git a/apps/server/src/cloud/pinnedRuntime.test.ts b/apps/server/src/cloud/pinnedRuntime.test.ts index 0c016a504c9b4..9fd493d15690b 100644 --- a/apps/server/src/cloud/pinnedRuntime.test.ts +++ b/apps/server/src/cloud/pinnedRuntime.test.ts @@ -285,6 +285,80 @@ it.layer(NodeServices.layer)("ensurePinnedRuntimeInstalled", (it) => { }), ); + it.effect("retries a release archive whose body stalls after the host answered", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pinned-retry-stall-" }); + const checksums = yield* validChecksums; + let archiveRequests = 0; + const archiveRequested = yield* Deferred.make(); + const install = yield* ensurePinnedRuntimeInstalled({ + baseDir, + version, + fs, + path, + platform: "linux", + arch: "x64", + httpClient: HttpClient.make((request) => { + const isChecksums = request.url.endsWith("/SHA256SUMS"); + const response = HttpClientResponse.fromWeb( + request, + new Response(isChecksums ? checksums : archiveBytes), + ); + if (isChecksums || ++archiveRequests > 1) return Effect.succeed(response); + const stalledBody = Stream.concat(Stream.make(archiveBytes.slice(0, 4)), Stream.never); + Deferred.doneUnsafe(archiveRequested, Effect.void); + return Effect.sleep("1 second").pipe( + Effect.as( + Object.create(response, { stream: { value: stalledBody } }), + ), + ); + }), + runner: extractingRunner(fs, path), + validate: () => Effect.void, + }).pipe(Effect.forkScoped); + yield* Deferred.await(archiveRequested); + // 1 s to answer, the 1 min stall timeout, then the 2 s retry wait. + yield* TestClock.adjust(Duration.seconds(63)); + const paths = yield* Fiber.join(install); + assert.equal(yield* fs.readFileString(paths.sentinelPath), `${version}\n`); + assert.equal(archiveRequests, 2); + }), + ); + + it.effect("fails at once when the release host never answers", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pinned-no-answer-" }); + let requests = 0; + const requested = yield* Deferred.make(); + const install = yield* ensurePinnedRuntimeInstalled({ + baseDir, + version, + fs, + path, + platform: "linux", + arch: "x64", + httpClient: HttpClient.make(() => { + requests += 1; + Deferred.doneUnsafe(requested, Effect.void); + return Effect.never; + }), + runner: extractingRunner(fs, path), + validate: () => Effect.die("must not validate without a download"), + }).pipe(Effect.forkScoped); + yield* Deferred.await(requested); + // Long enough for a retry to have started had the no-answer timeout been retried. + yield* TestClock.adjust(Duration.minutes(2)); + assert.equal(requests, 1); + const error = yield* Fiber.join(install).pipe(Effect.flip); + assert.instanceOf(error, PinnedRuntimeInstallError); + assert.equal(error.step, "downloading the t3 release checksums (timed out)"); + }), + ); + it.effect("fails at once when the release host cannot be reached", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; diff --git a/apps/server/src/cloud/pinnedRuntime.ts b/apps/server/src/cloud/pinnedRuntime.ts index 5f7e170530d79..e1f7514e5b5cd 100644 --- a/apps/server/src/cloud/pinnedRuntime.ts +++ b/apps/server/src/cloud/pinnedRuntime.ts @@ -123,11 +123,18 @@ interface PinnedRuntimeInstallInput { // GitHub release downloads intermittently answer 5xx during incidents; a retry usually lands. const DOWNLOAD_RETRY = { times: 3, schedule: Schedule.exponential("2 seconds") } as const; +/** The release host answered, then sent no body bytes for the stall timeout. */ +class ReleaseDownloadStalledError extends Schema.TaggedError()( + "ReleaseDownloadStalledError", + {}, +) {} + const isTransientDownloadFailure = ( - error: HttpClientError.HttpClientError | Cause.TimeoutError, + error: HttpClientError.HttpClientError | Cause.TimeoutError | ReleaseDownloadStalledError, ): boolean => { - // No answer or a stall means the host cannot reach the release source; retrying only delays that. + // No answer means the host cannot reach the release source; retrying only delays that. if (error._tag === "TimeoutError") return false; + if (error._tag === "ReleaseDownloadStalledError") return true; // Body bytes are read raw, so a DecodeError is a body that failed after the host answered. if (error.reason._tag === "DecodeError") return true; const status = error.response?.status; @@ -154,7 +161,7 @@ const fetchReleaseAsset = Effect.fn("cloud.pinned_runtime.fetch_release_asset")( const chunks = yield* response.stream.pipe( Stream.timeoutOrElse({ duration: PINNED_RUNTIME_DOWNLOAD_STALL_TIMEOUT, - orElse: () => Stream.fail(new Cause.TimeoutError()), + orElse: () => Stream.fail(new ReleaseDownloadStalledError()), }), Stream.tap((chunk) => Effect.sync(() => { @@ -180,7 +187,10 @@ const fetchReleaseAsset = Effect.fn("cloud.pinned_runtime.fetch_release_asset")( Effect.mapError( (cause) => new PinnedRuntimeInstallError({ - step: cause._tag === "TimeoutError" ? `${step} (timed out)` : step, + step: + cause._tag === "TimeoutError" || cause._tag === "ReleaseDownloadStalledError" + ? `${step} (timed out)` + : step, httpStatus: HttpClientError.isHttpClientError(cause) && cause.reason._tag === "StatusCodeError" ? cause.reason.response.status