perf(provider-core): provider instances share one latest-version lookup

packages/provider-core/src/server/ProviderLatestVersions.test.ts

@@ -0,0 +1,136 @@
+import { describe, expect, it } from "@effect/vitest";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
+import * as Fiber from "effect/Fiber";
+
+import * as ProviderLatestVersions from "./ProviderLatestVersions.ts";
+
+describe("ProviderLatestVersions", () => {
+  it.effect("concurrent misses for one key share a single lookup", () =>
+    Effect.gen(function* () {
+      const latestVersions = yield* ProviderLatestVersions.make();
+      const release = yield* Deferred.make<void>();
+      let lookups = 0;
+      const lookup = Effect.suspend(() => {
+        lookups += 1;
+        return Deferred.await(release).pipe(Effect.as("2.0.0"));
+      });
+
+      const callers = yield* Effect.forEach([1, 2, 3], () =>
+        Effect.forkChild(latestVersions.cached("@example/cli", lookup), {
+          startImmediately: true,
+        }),
+      );
+      yield* Deferred.succeed(release, undefined);
+      expect(yield* Effect.forEach(callers, Fiber.join)).toEqual(["2.0.0", "2.0.0", "2.0.0"]);
+      expect(lookups).toBe(1);
+    }),
+  );
+
+  it.effect("a waiting caller looks up again when the shared lookup is interrupted", () =>
+    Effect.gen(function* () {
+      const latestVersions = yield* ProviderLatestVersions.make();
+      let lookups = 0;
+      const lookup = Effect.suspend(() => {
+        lookups += 1;
+        return lookups === 1 ? Effect.never : Effect.succeed("2.0.0");
+      });
+
+      const owner = yield* Effect.forkChild(latestVersions.cached("@example/cli", lookup), {
+        startImmediately: true,
+      });
+      const waiter = yield* Effect.forkChild(latestVersions.cached("@example/cli", lookup), {
+        startImmediately: true,
+      });
+      expect(lookups).toBe(1);
+      yield* Fiber.interrupt(owner);
+      expect(yield* Fiber.join(waiter)).toBe("2.0.0");
+      expect(yield* latestVersions.cached("@example/cli", lookup)).toBe("2.0.0");
+      expect(lookups).toBe(2);
+    }),
+  );
+
+  it.effect("a caller interrupted while it claims the lookup frees the key", () =>
+    Effect.gen(function* () {
+      const latestVersions = yield* ProviderLatestVersions.make();
+      let lookups = 0;
+      const lookup = Effect.sync(() => {
+        lookups += 1;
+        return "2.0.0";
+      });
+      // `cached` reads `fresh` in the same step that claims the key, so the
+      // interrupt arrives after the claim and before the lookup starts.
+      const interruptOnClaim = {
+        get fresh() {
+          Fiber.getCurrent()?.interruptUnsafe();
+          return false;
+        },
+      };
+
+      const owner = yield* Effect.forkChild(
+        latestVersions.cached("@example/cli", lookup, interruptOnClaim),
+        { startImmediately: true },
+      );
+      expect(Exit.hasInterrupts(yield* Fiber.await(owner))).toBe(true);
+      expect(lookups).toBe(0);
+      const next = yield* Effect.forkChild(latestVersions.cached("@example/cli", lookup), {
+        startImmediately: true,
+      });
+      expect(lookups).toBe(1);
+      expect(yield* Fiber.join(next)).toBe("2.0.0");
+    }),
+  );
+  it.effect("a waiting caller retries with its own lookup after the owner fails", () =>
+    Effect.gen(function* () {
+      const latestVersions = yield* ProviderLatestVersions.make();
+      const release = yield* Deferred.make<void>();
+      const owner = yield* Effect.forkChild(
+        latestVersions.cached(
+          "cli",
+          Deferred.await(release).pipe(Effect.andThen(Effect.fail("offline"))),
+        ),
+        { startImmediately: true },
+      );
+      let retries = 0;
+      const waiter = yield* Effect.forkChild(
+        latestVersions.cached(
+          "cli",
+          Effect.sync(() => {
+            retries += 1;
+            return "2.0.0";
+          }),
+        ),
+        { startImmediately: true },
+      );
+      expect(retries).toBe(0);
+      yield* Deferred.succeed(release, undefined);
+      expect(yield* Effect.flip(Fiber.join(owner))).toBe("offline");
+      expect(yield* Fiber.join(waiter)).toBe("2.0.0");
+      expect(retries).toBe(1);
+    }),
+  );
+
+  it.effect("interrupting a waiter leaves the shared lookup running", () =>
+    Effect.gen(function* () {
+      const latestVersions = yield* ProviderLatestVersions.make();
+      const release = yield* Deferred.make<void>();
+      let lookups = 0;
+      const lookup = Effect.suspend(() => {
+        lookups += 1;
+        return Deferred.await(release).pipe(Effect.as("2.0.0"));
+      });
+      const owner = yield* Effect.forkChild(latestVersions.cached("cli", lookup), {
+        startImmediately: true,
+      });
+      const waiter = yield* Effect.forkChild(latestVersions.cached("cli", lookup), {
+        startImmediately: true,
+      });
+      yield* Fiber.interrupt(waiter);
+      yield* Deferred.succeed(release, undefined);
+      expect(yield* Fiber.join(owner)).toBe("2.0.0");
+      expect(yield* latestVersions.cached("cli", lookup)).toBe("2.0.0");
+      expect(lookups).toBe(1);
+    }),
+  );
+});

packages/provider-core/src/server/ProviderLatestVersions.ts

@@ -1,8 +1,11 @@
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
+import * as Deferred from "effect/Deferred";
 import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
 import * as Ref from "effect/Ref";
 
 const FOUND_TTL = Duration.hours(1);
@@ -11,6 +14,7 @@ const MISSING_TTL = Duration.minutes(1);
 interface Entry {
   readonly expiresAt: number;
   readonly version: string | null;
+  readonly inFlight?: Deferred.Deferred<Option.Option<string | null>>;
 }
 
 export class ProviderLatestVersions extends Context.Service<
@@ -33,23 +37,46 @@ export const make = Effect.fn("ProviderLatestVersions.make")(function* (
       Array.from(seed, ([key, version]) => [key, { version, expiresAt: Number.MAX_SAFE_INTEGER }]),
     ),
   );
+  const cached: ProviderLatestVersions["Service"]["cached"] = Effect.fn(
+    "ProviderLatestVersions.cached",
+  )(function* (key, lookup, options) {
+    const now = DateTime.toEpochMillis(yield* DateTime.now);
+    const inFlight = yield* Deferred.make<Option.Option<string | null>>();
+    // Replacing or invalidating this entry retires any lookup still in flight.
+    const pending: Entry = { version: null, expiresAt: 0, inFlight };
+    // Claim and cleanup registration must be atomic with respect to interruption.
+    return yield* Effect.acquireUseRelease(
+      Ref.modify(entries, (current) => {
+        const entry = current.get(key);
+        return !options?.fresh && entry && (entry.expiresAt > now || entry.inFlight)
+          ? [entry, current]
+          : [undefined, new Map(current).set(key, pending)];
+      }),
+      (existing) =>
+        Effect.gen(function* () {
+          if (!existing) return yield* lookup;
+          if (!existing.inFlight) return existing.version;
+          const shared = yield* Deferred.await(existing.inFlight);
+          return Option.isSome(shared) ? shared.value : yield* cached(key, lookup, options);
+        }),
+      (existing, exit) => {
+        if (existing) return Effect.void;
+        const version = Exit.isSuccess(exit) ? Option.some(exit.value) : Option.none();
+        return Ref.update(entries, (current) => {
+          if (current.get(key) !== pending) return current;
+          const next = new Map(current);
+          if (Option.isNone(version)) {
+            next.delete(key);
+            return next;
+          }
+          const ttl = version.value === null ? MISSING_TTL : FOUND_TTL;
+          return next.set(key, { version: version.value, expiresAt: now + Duration.toMillis(ttl) });
+        }).pipe(Effect.andThen(Deferred.succeed(inFlight, version)), Effect.asVoid);
+      },
+    );
+  });
   return ProviderLatestVersions.of({
-    cached: Effect.fn("ProviderLatestVersions.cached")(function* (key, lookup, options) {
-      const now = DateTime.toEpochMillis(yield* DateTime.now);
-      const existing = (yield* Ref.get(entries)).get(key);
-      if (!options?.fresh && existing && existing.expiresAt > now) return existing.version;
-      // Replacing or invalidating this entry retires any lookup still in flight.
-      const pending = { version: null, expiresAt: 0 };
-      yield* Ref.update(entries, (current) => new Map(current).set(key, pending));
-      const version = yield* lookup;
-      const ttl = version === null ? MISSING_TTL : FOUND_TTL;
-      yield* Ref.update(entries, (current) =>
-        current.get(key) === pending
-          ? new Map(current).set(key, { version, expiresAt: now + Duration.toMillis(ttl) })
-          : current,
-      );
-      return version;
-    }),
+    cached,
     invalidate: Effect.fn("ProviderLatestVersions.invalidate")(function (key) {
       return Ref.update(entries, (current) => {
         const next = new Map(current);