Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions infra/task-cache/src/Admission.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<void>();
Expand All @@ -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)),
);
Expand Down
14 changes: 11 additions & 3 deletions infra/task-cache/src/Admission.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,25 +2,33 @@ 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>()(
"AdmissionRejected",
{},
) {}

export class Admission extends Context.Service<
Admission,
{
readonly withPermit: <A, E, R>(
store: boolean,
task: Effect.Effect<A, E, R>,
) => Effect.Effect<A, E, R>;
) => Effect.Effect<A, E | AdmissionRejected, R>;
}
>()("@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();
}),
});
});
Expand Down
2 changes: 2 additions & 0 deletions infra/task-cache/src/Cache.ts
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,8 @@ const make = Effect.gen(function* () {
.pipe(
Effect.timeout(writing ? "2 minutes" : "15 seconds"),
Effect.catchTags({
AdmissionRejected: () =>

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add a focused Cache-level test that occupies both upload permits (or all read permits), sends another authenticated request, and asserts a 503 response with concurrency_limit? The Admission test covers rejection at the service boundary, but the new HTTP error translation is not exercised.

Posted via Macroscope — Effect Service Conventions

Effect.fail(new CacheError({ status: 503, reason: "concurrency_limit" })),
TimeoutError: () => Effect.fail(new CacheError({ status: 503, reason: "deadline" })),
}),
);
Expand Down
Loading