From eab998bcd42f3d3ff82a7101e3ba08fd2f03b3d8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 25 Oct 2026 22:55:20 -0700 Subject: [PATCH 1/2] fix(task-cache): avoid cross-request admission waiters --- infra/task-cache/src/Admission.test.ts | 12 +++++++----- infra/task-cache/src/Admission.ts | 14 +++++++++++--- infra/task-cache/src/Cache.ts | 2 ++ 3 files changed, 20 insertions(+), 8 deletions(-) diff --git a/infra/task-cache/src/Admission.test.ts b/infra/task-cache/src/Admission.test.ts index c1fef5204796c..db9d01cb45cda 100644 --- a/infra/task-cache/src/Admission.test.ts +++ b/infra/task-cache/src/Admission.test.ts @@ -2,9 +2,10 @@ import { expect, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as Exit from "effect/Exit"; import * as Admission from "./Admission.ts"; -it.effect("queues uploads and releases the permit when an upload is interrupted", () => +it.effect("rejects excess uploads immediately and releases interrupted uploads", () => Effect.gen(function* () { const admission = yield* Admission.Admission; const entered = yield* Deferred.make(); @@ -20,11 +21,12 @@ it.effect("queues uploads and releases the permit when an upload is interrupted" ) .pipe(Effect.forkChild); yield* Deferred.await(secondEntered); - const third = yield* admission - .withPermit(true, Effect.succeed("restored")) - .pipe(Effect.forkChild); + const rejected = yield* admission + .withPermit(true, Effect.die("must not run")) + .pipe(Effect.exit); + expect(rejected).toEqual(Exit.fail(new Admission.AdmissionRejected())); yield* Fiber.interrupt(first); - expect(yield* Fiber.join(third)).toBe("restored"); + expect(yield* admission.withPermit(true, Effect.succeed("restored"))).toBe("restored"); yield* Fiber.interrupt(second); }).pipe(Effect.provide(Admission.layer)), ); diff --git a/infra/task-cache/src/Admission.ts b/infra/task-cache/src/Admission.ts index 4f69041e8b491..ba81aa978f930 100644 --- a/infra/task-cache/src/Admission.ts +++ b/infra/task-cache/src/Admission.ts @@ -2,6 +2,13 @@ import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Semaphore from "effect/Semaphore"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; + +export class AdmissionRejected extends Schema.TaggedError()( + "AdmissionRejected", + {}, +) {} export class Admission extends Context.Service< Admission, @@ -9,18 +16,19 @@ export class Admission extends Context.Service< readonly withPermit: ( store: boolean, task: Effect.Effect, - ) => Effect.Effect; + ) => Effect.Effect; } >()("@t3tools/task-cache/Admission") {} const make = Effect.gen(function* () { - // Bound simultaneous upload buffers; Effect cancels queued waiters and releases permits. + // Never queue fibers across Worker requests: their async contexts cannot cross request boundaries. const stores = yield* Semaphore.make(2); const reads = yield* Semaphore.make(4); return Admission.of({ // The caller owns the request span; admission adds no second span per request. withPermit: Effect.fnUntraced(function* (store, task) { - return yield* (store ? stores : reads).withPermits(1)(task); + const result = yield* (store ? stores : reads).withPermitsIfAvailable(1)(task); + return Option.isSome(result) ? result.value : yield* new AdmissionRejected(); }), }); }); diff --git a/infra/task-cache/src/Cache.ts b/infra/task-cache/src/Cache.ts index ce9759b498a26..4677de8cd5b2e 100644 --- a/infra/task-cache/src/Cache.ts +++ b/infra/task-cache/src/Cache.ts @@ -220,6 +220,8 @@ const make = Effect.gen(function* () { .pipe( Effect.timeout(writing ? "2 minutes" : "15 seconds"), Effect.catchTags({ + AdmissionRejected: () => + Effect.fail(new CacheError({ status: 503, reason: "concurrency_limit" })), TimeoutError: () => Effect.fail(new CacheError({ status: 503, reason: "deadline" })), }), ); From cf92fbb9509ee2d74c1bead923595c97c1b2bbb4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 25 Oct 2026 23:29:38 -0700 Subject: [PATCH 2/2] test(task-cache): cover HTTP admission rejection --- infra/task-cache/src/Cache.test.ts | 61 ++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/infra/task-cache/src/Cache.test.ts b/infra/task-cache/src/Cache.test.ts index 7ff2a8fb76878..4ba37b9784088 100644 --- a/infra/task-cache/src/Cache.test.ts +++ b/infra/task-cache/src/Cache.test.ts @@ -4,6 +4,7 @@ import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import { Deadline } from "../vendor/vite-plus-remote-cache-cloudflare/src/streams.ts"; import * as Layer from "effect/Layer"; +import * as Logger from "effect/Logger"; import { vi } from "vite-plus/test"; import { Observations } from "../vendor/vite-plus-remote-cache-cloudflare/src/observations.ts"; import * as Admission from "./Admission.ts"; @@ -62,6 +63,66 @@ it.effect("includes successful blob downloads in exported R2 usage", () => }), ); +it.effect("returns 503 immediately when authenticated requests fill all read permits", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const logs: unknown[] = []; + const get = vi.fn(() => { + if (get.mock.calls.length === 4) Deferred.doneUnsafe(entered, Effect.void); + return new Promise(() => {}); + }); + const bindings = { + developerKeys: JSON.stringify({ + julius: "210eb3d2c67a7207923e1999751300a87a2dd22839b258b80c3d805a6fe64984", + }), + ARTIFACTS: { get }, + readLimit: { limit: async () => ({ success: true }) }, + } as unknown as Cache.Bindings["Service"]; + yield* Effect.gen(function* () { + const cache = yield* Cache.Cache; + const request = () => + cache.handle( + new Request( + "https://cache.example/projects/t3code/blob/12345678-1234-1234-1234-123456789abc", + { + headers: { + authorization: `Basic ${btoa("julius:fixture-token-012345678901234567890123456789")}`, + }, + }, + ), + { waitUntil: () => {} }, + ); + const pending = yield* Effect.all( + Array.from({ length: 4 }, () => request().pipe(Effect.forkChild)), + ); + yield* Deferred.await(entered); + const response = yield* request(); + expect(response.status).toBe(503); + expect(yield* Effect.promise(() => response.text())).toBe("Service unavailable"); + expect(logs).toContainEqual(["Cache request failed", { reason: "concurrency_limit" }]); + expect(get).toHaveBeenCalledTimes(4); + yield* Effect.forEach(pending, Fiber.interrupt); + }).pipe( + Effect.provide([ + Cache.layer.pipe( + Layer.provide([ + Layer.succeed(Cache.Bindings, bindings), + Admission.layer, + Layer.succeed(Savings.Savings, { + record: () => Effect.void, + emitTotals: () => Effect.void, + cleanup: Effect.void, + }), + ]), + ), + Logger.layer([Logger.make(({ message }) => logs.push(message))], { + mergeWithExisting: false, + }), + ]), + ); + }), +); + it.effect("aborts the upstream deadline before disposing it on interruption", () => Effect.gen(function* () { const entered = yield* Deferred.make();