Refresh: one attempt per identity per process; waiters share its outcome

Callers that arrive while a refresh of one linked identity is in flight in
this process join it and receive its success or failure, instead of queuing
on the row lock and presenting the same refresh token again behind a failed
holder. The entry lives as long as the attempt, so a later request retries.
The store lock stays the cross-process guard; the residual risk (one
presentation per process per burst) is written into the service contract.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/service-contract.md

@@ -490,7 +490,12 @@ Important behavior:
   access/refresh tokens. The refresh runs under a row lock on the linked
   identity (`SELECT ... FOR UPDATE`), so a given refresh token is presented to
   the provider once across concurrent requests and across two Heimdall
-  processes sharing one database. The provider call is bounded at 10 seconds
+  processes sharing one database. Within one process, concurrent refreshes of
+  one identity share a single attempt and its outcome (success or failure), so
+  nobody re-presents a token behind a failed holder in that burst. Residual
+  risk, accepted: the guard is one presentation per process per burst, so N
+  Heimdall processes can present one token N times after a transient failure.
+  The provider call is bounded at 10 seconds
 - a provider whose refresh tokens rotate (Joystick) fails closed: a refresh
   response without a new refresh token revokes the credential instead of
   keeping the old token

src/managed-credential.ts

@@ -60,18 +60,31 @@ function withoutTokens(identity: StoredLinkedIdentity): StoredLinkedIdentity {
   return rest;
 }
 
+/**
+ * Refreshes in flight in this process, per store and linked identity. The row
+ * lock serialises processes; it cannot tell a waiter that queued behind a
+ * failed holder from a caller that arrived after the failure, so each waiter
+ * would present the same refresh token again. Callers that arrive while an
+ * attempt is in flight join it and receive its outcome, success or failure.
+ * The entry lives exactly as long as the attempt: a later request retries.
+ *
+ * Accepted residual risk (docs/service-contract.md): one presentation per
+ * process per burst, so N processes may present one token N times.
+ */
+const refreshesInFlight = new WeakMap<HeimdallStore, Map<string, Promise<ManagedCredentialOutcome>>>();
+
 /**
  * The one refresh policy. `snapshot` is a hint: it may skip the lock when no
  * refresh is due and never decides one. The row's lock owns when a refresh
- * token is presented; the decision is made on the row as read under it.
+ * token is presented across processes; this process makes one attempt per
+ * identity at a time; the decision is made on the row as read under the lock.
  */
 export async function refreshManagedCredential(
   ctx: ManagedCredentialContext,
   snapshot: StoredLinkedIdentity
 ): Promise<ManagedCredentialOutcome> {
   const clock = ctx.clock ?? Date.now;
-  const provider = snapshot.provider;
-  const runtime = ctx.oauthRuntimes[provider];
+  const runtime = ctx.oauthRuntimes[snapshot.provider];
 
   if (!snapshot.accessTokenEncrypted) {
     return { kind: "revoked", identity: snapshot };
@@ -80,6 +93,30 @@ export async function refreshManagedCredential(
     return { kind: "current", identity: snapshot };
   }
 
+  let inFlight = refreshesInFlight.get(ctx.store);
+  if (!inFlight) {
+    inFlight = new Map();
+    refreshesInFlight.set(ctx.store, inFlight);
+  }
+  const joined = inFlight.get(snapshot.id);
+  if (joined) {
+    return joined;
+  }
+  const attempt = refreshUnderLock(ctx, snapshot).finally(() => {
+    inFlight.delete(snapshot.id);
+  });
+  inFlight.set(snapshot.id, attempt);
+  return attempt;
+}
+
+async function refreshUnderLock(
+  ctx: ManagedCredentialContext,
+  snapshot: StoredLinkedIdentity
+): Promise<ManagedCredentialOutcome> {
+  const clock = ctx.clock ?? Date.now;
+  const provider = snapshot.provider;
+  const runtime = ctx.oauthRuntimes[provider];
+
   let revocation: ManagedCredentialRevocationReason | undefined;
   const decide = async (locked: StoredLinkedIdentity): Promise<CredentialDecision> => {
     if (!locked.accessTokenEncrypted || !dueForRefresh(locked, runtime, clock())) {

tests/managed-credential.test.ts

@@ -381,6 +381,31 @@ for (const { kind, live } of rigKinds) describe.skipIf(live && !adminUrl)(`manag
       expect(rig.custody.decrypt(outcome.identity.accessTokenEncrypted!)).toBe("access-1");
     });
 
+    it("single flight: concurrent refreshes against a never-answering endpoint make one provider call per process, and every caller gets the typed failure", timeout, async () => {
+      const rig = await makeRig(provider);
+      const endpoint = installTokenEndpoint(provider, "refresh-0");
+      endpoint.mode = "hang";
+      const snapshot = await rig.row();
+      const ctxFor = (store: HeimdallStore) => ({ config: rig.config, store, tokenCustody: rig.custody, oauthRuntimes: createOAuthRuntimeRegistry(), timeoutMs: 150 });
+
+      const results = await Promise.allSettled(
+        Array.from({ length: 12 }, (_, i) => refreshManagedCredential(ctxFor(rig.stores[i % rig.stores.length]!), snapshot))
+      );
+
+      expect(results.map((r) => r.status)).toEqual(Array(12).fill("rejected"));
+      for (const result of results) {
+        expect((result as PromiseRejectedResult).reason).toBeInstanceOf(ManagedCredentialRefreshFailed);
+      }
+      expect(endpoint.calls).toBe(rig.stores.length);
+      expect(rig.custody.decrypt((await rig.row()).refreshTokenEncrypted!)).toBe("refresh-0");
+
+      // The burst is over: a later independent request tries again.
+      endpoint.mode = "normal";
+      const later = await refreshManagedCredential(ctxFor(rig.stores[0]!), await rig.row());
+      expect(later.kind).toBe("current");
+      expect(endpoint.calls).toBe(rig.stores.length + 1);
+    });
+
     it("a snapshot taken before another caller's refresh does not decide: the locked row does", timeout, async () => {
       const rig = await makeRig(provider);
       const endpoint = installTokenEndpoint(provider, "refresh-0");