From bc70651f6288fdb5f8a5bf39e94b21bcd816320e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=F0=9D=93=9B=F0=9D=93=B2=F0=9D=93=BD=F0=9D=93=BD?= =?UTF-8?q?=F0=9D=93=B5=F0=9D=93=AE=20=F0=9D=93=95=F0=9D=93=BB=F0=9D=93=AA?= =?UTF-8?q?=F0=9D=93=B7=F0=9D=93=B4?= Date: Fri, 26 Jun 2026 19:59:08 +0000 Subject: [PATCH] fix(core): dedupe credential refreshes --- packages/core/src/integration.ts | 24 +++++- packages/core/test/integration.test.ts | 114 ++++++++++++++++++++++++- 2 files changed, 134 insertions(+), 4 deletions(-) diff --git a/packages/core/src/integration.ts b/packages/core/src/integration.ts index af096df2fd..6d54cb2f49 100644 --- a/packages/core/src/integration.ts +++ b/packages/core/src/integration.ts @@ -1,6 +1,7 @@ export * as Integration from "./integration" import { + Cache, Cause, Clock, Context, @@ -310,6 +311,25 @@ export const locationLayer = Layer.effect( const authorize = (effect: Effect.Effect) => effect.pipe(Effect.mapError((cause) => new AuthorizationError({ cause }))) + const refreshes = yield* Cache.make({ + capacity: Number.POSITIVE_INFINITY, + timeToLive: Duration.zero, + lookup: Effect.fnUntraced(function* (credentialID) { + const credential = yield* credentials.get(credentialID) + if (!credential || credential.value.type === "key") return credential?.value + const implementation = state + .get() + .integrations.get(credential.integrationID) + ?.implementations.get(credential.value.methodID) + if (!implementation?.refresh) return credential.value + const now = yield* Clock.currentTimeMillis + if (credential.value.expires > now + Duration.toMillis(Duration.minutes(5))) return credential.value + const value = yield* authorize(implementation.refresh(credential.value)) + yield* credentials.update(credentialID, { value }) + return value + }), + }) + const close = (attemptScope: Scope.Closeable) => Scope.close(attemptScope, Exit.void).pipe(Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid) @@ -396,9 +416,7 @@ export const locationLayer = Layer.effect( if (!implementation?.refresh) return credential.value const now = yield* Clock.currentTimeMillis if (credential.value.expires > now + Duration.toMillis(Duration.minutes(5))) return credential.value - const value = yield* authorize(implementation.refresh(credential.value)) - yield* credentials.update(credential.id, { value }) - return value + return yield* Cache.get(refreshes, credential.id) }), key: Effect.fn("Integration.connection.key")(function* (input) { const method = state diff --git a/packages/core/test/integration.test.ts b/packages/core/test/integration.test.ts index 9dd6959f72..f7edfd7bc9 100644 --- a/packages/core/test/integration.test.ts +++ b/packages/core/test/integration.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "bun:test" -import { Duration, Effect, Exit, Fiber, Layer, Scope, Stream } from "effect" +import { Deferred, Duration, Effect, Exit, Fiber, Layer, Scope, Stream } from "effect" import * as TestClock from "effect/testing/TestClock" import { Integration } from "@opencode-ai/core/integration" import { Credential } from "@opencode-ai/core/credential" @@ -346,4 +346,116 @@ describe("Integration", () => { }), ) }) + + it.effect("shares concurrent OAuth credential refreshes", () => + Effect.gen(function* () { + const integrations = yield* Integration.Service + const credentials = yield* Credential.Service + const integrationID = Integration.ID.make("openai") + const methodID = Integration.MethodID.make("chatgpt") + const started = yield* Deferred.make() + const release = yield* Deferred.make() + let refreshes = 0 + const value = Credential.OAuth.make({ + type: "oauth", + methodID, + access: "refreshed", + refresh: "refresh-2", + expires: Duration.toMillis(Duration.hours(1)), + }) + yield* integrations.transform((editor) => + editor.method.update({ + integrationID, + method: { id: methodID, type: "oauth", label: "ChatGPT" }, + authorize: () => Effect.die("unexpected authorization"), + refresh: () => + Effect.sync(() => refreshes++).pipe( + Effect.andThen(Deferred.succeed(started, undefined)), + Effect.andThen(Deferred.await(release)), + Effect.as(value), + ), + }), + ) + const credential = yield* credentials.create({ + integrationID, + value: Credential.OAuth.make({ + type: "oauth", + methodID, + access: "expired", + refresh: "refresh-1", + expires: 0, + }), + }) + const connection = { type: "credential" as const, id: credential.id, label: credential.label } + + const first = yield* integrations.connection.resolve(connection).pipe(Effect.forkChild) + yield* Deferred.await(started) + const second = yield* integrations.connection.resolve(connection).pipe(Effect.forkChild) + yield* Effect.yieldNow + expect(refreshes).toBe(1) + + yield* Deferred.succeed(release, undefined) + expect(yield* Effect.all([Fiber.join(first), Fiber.join(second)], { concurrency: "unbounded" })).toEqual([ + value, + value, + ]) + expect(refreshes).toBe(1) + expect((yield* credentials.get(credential.id))?.value).toEqual(value) + }), + ) + + it.effect("shares concurrent refresh failures and retries later", () => + Effect.gen(function* () { + const integrations = yield* Integration.Service + const credentials = yield* Credential.Service + const integrationID = Integration.ID.make("openai") + const methodID = Integration.MethodID.make("chatgpt") + const started = yield* Deferred.make() + const release = yield* Deferred.make() + const failure = new Error("refresh failed") + let refreshes = 0 + yield* integrations.transform((editor) => + editor.method.update({ + integrationID, + method: { id: methodID, type: "oauth", label: "ChatGPT" }, + authorize: () => Effect.die("unexpected authorization"), + refresh: () => + Effect.sync(() => refreshes++).pipe( + Effect.andThen(Deferred.succeed(started, undefined)), + Effect.andThen(Deferred.await(release)), + Effect.andThen(Effect.fail(failure)), + ), + }), + ) + const credential = yield* credentials.create({ + integrationID, + value: Credential.OAuth.make({ + type: "oauth", + methodID, + access: "expired", + refresh: "refresh", + expires: 0, + }), + }) + const connection = { type: "credential" as const, id: credential.id, label: credential.label } + + const first = yield* integrations.connection.resolve(connection).pipe(Effect.flip, Effect.forkChild) + yield* Deferred.await(started) + const second = yield* integrations.connection.resolve(connection).pipe(Effect.flip, Effect.forkChild) + yield* Effect.yieldNow + expect(refreshes).toBe(1) + + yield* Deferred.succeed(release, undefined) + const results = yield* Effect.all([Fiber.join(first), Fiber.join(second)], { concurrency: "unbounded" }) + expect(results).toEqual([ + new Integration.AuthorizationError({ cause: failure }), + new Integration.AuthorizationError({ cause: failure }), + ]) + + expect(yield* integrations.connection.resolve(connection).pipe(Effect.flip)).toEqual( + new Integration.AuthorizationError({ cause: failure }), + ) + expect(refreshes).toBe(2) + }), + ) })