Skip to content

Commit 9435378

Browse files
authored
fix(core): coordinate credential refreshes across locations and processes (#54023)
1 parent 74556a2 commit 9435378

8 files changed

Lines changed: 213 additions & 10 deletions

File tree

‎packages/core/src/credential.ts‎

Lines changed: 28 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,14 @@
11
export * as Credential from "./credential.js"
22

33
import { asc, desc, eq } from "drizzle-orm"
4-
import { Cause, Context, Effect, Layer, Schema } from "effect"
4+
import { Cause, Context, Effect, Fiber, Layer, Schema } from "effect"
55
import { Credential } from "@opencode/schema/credential"
66
import { CredentialLegacy } from "@opencode/schema/credential-legacy"
77
import { Integration } from "@opencode/schema/integration"
88
import { Database } from "./database/database.js"
99
import { Bus } from "./bus.js"
1010
import { makeGlobalNode } from "@opencode/util/effect/app-node"
11+
import { EffectFlock } from "@opencode/util/effect-flock"
1112
import { CredentialTable } from "./credential/sql.js"
1213
import { ErrorSummary } from "./util/error-summary.js"
1314

@@ -42,6 +43,11 @@ export interface Interface {
4243
readonly list: (integrationID: Integration.ID) => Effect.Effect<Info[]>
4344
/** Returns one stored credential by ID. */
4445
readonly get: (id: ID) => Effect.Effect<Info | undefined>
46+
/** Concurrent callers for the same credential share the running operation's result, including failures. */
47+
readonly lock: (
48+
id: ID,
49+
operation: Effect.Effect<OAuth | undefined, unknown>,
50+
) => Effect.Effect<OAuth | undefined, unknown>
4551
/**
4652
* Creates a credential for an integration and returns the new record. The new credential becomes the
4753
* integration's selection unless `activate` is false and the integration already has a credential.
@@ -68,6 +74,9 @@ const layer = Layer.effect(
6874
Effect.gen(function* () {
6975
const db = (yield* Database.Service).db
7076
const bus = yield* Bus.Service
77+
const flock = yield* EffectFlock.Service
78+
const scope = yield* Effect.scope
79+
const pending = new Map<ID, Fiber.Fiber<OAuth | undefined, unknown>>()
7180
// Rows written before v2.1 may carry the old key credential shape.
7281
const decode = Schema.decodeUnknownSync(CredentialLegacy.Value)
7382
const stored = (row: typeof CredentialTable.$inferSelect) => {
@@ -86,6 +95,23 @@ const layer = Layer.effect(
8695
})
8796

8897
return Service.of({
98+
lock: Effect.fn("Credential.lock")((id, operation) =>
99+
Effect.uninterruptibleMask((restore) =>
100+
Effect.gen(function* () {
101+
const current = pending.get(id)
102+
if (current) return yield* restore(Fiber.join(current))
103+
// The global scope owns the operation, not the first Session or Location waiting for it.
104+
// The file lock extends this to other processes sharing the database.
105+
const fiber = yield* operation.pipe(
106+
flock.withLock(`credential-refresh:${id}`),
107+
Effect.ensuring(Effect.sync(() => pending.delete(id))),
108+
Effect.forkIn(scope),
109+
)
110+
pending.set(id, fiber)
111+
return yield* restore(Fiber.join(fiber))
112+
}),
113+
),
114+
),
89115
all: Effect.fn("Credential.all")(() =>
90116
db
91117
.select()
@@ -327,4 +353,4 @@ const layer = Layer.effect(
327353
}),
328354
)
329355

330-
export const node = makeGlobalNode({ service: Service, layer, deps: [Database.node, Bus.node] })
356+
export const node = makeGlobalNode({ service: Service, layer, deps: [Database.node, Bus.node, EffectFlock.node] })

‎packages/core/src/integration.ts‎

Lines changed: 21 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -731,16 +731,29 @@ const layer = Layer.effect(
731731
const credential = yield* credentials.get(connection.id)
732732
if (!credential) return undefined
733733
if (credential.value.type !== "oauth") return credential.value
734-
const implementation = state
734+
const refresh = state
735735
.get()
736736
.integrations.get(credential.integrationID)
737-
?.implementations.get(credential.value.methodID)
738-
if (!implementation?.refresh) return credential.value
739-
const now = yield* Clock.currentTimeMillis
740-
if (credential.value.expires > now + Duration.toMillis(Duration.minutes(5))) return credential.value
741-
const value = yield* authorize(implementation.refresh(credential.value))
742-
yield* credentials.update(credential.id, { value })
743-
return value
737+
?.implementations.get(credential.value.methodID)?.refresh
738+
const expiring = (value: Credential.OAuth) =>
739+
Clock.currentTimeMillis.pipe(
740+
Effect.map((now) => value.expires <= now + Duration.toMillis(Duration.minutes(5))),
741+
)
742+
if (!refresh || !(yield* expiring(credential.value))) return credential.value
743+
return yield* authorize(
744+
credentials.lock(
745+
credential.id,
746+
Effect.gen(function* () {
747+
// Another Location may have refreshed while this caller waited.
748+
const current = yield* credentials.get(credential.id)
749+
if (!current || current.value.type !== "oauth") return undefined
750+
if (!(yield* expiring(current.value))) return current.value
751+
const value = yield* refresh(current.value).pipe(Effect.timeout("30 seconds"))
752+
yield* credentials.update(current.id, { value })
753+
return value
754+
}),
755+
),
756+
)
744757
}),
745758
key: Effect.fn("Integration.connection.key")(function* (input) {
746759
const method = state

‎packages/core/test/config/config.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -802,6 +802,7 @@ describe("Config", () => {
802802
layer: Layer.succeed(
803803
Credential.Service,
804804
Credential.Service.of({
805+
lock: () => Effect.die("unused Credential.lock"),
805806
all: () => Effect.die("unused Credential.all"),
806807
list: () =>
807808
Effect.succeed([
@@ -1051,6 +1052,7 @@ describe("Config", () => {
10511052
layer: Layer.succeed(
10521053
Credential.Service,
10531054
Credential.Service.of({
1055+
lock: () => Effect.die("unused Credential.lock"),
10541056
all: () => Effect.die("unused Credential.all"),
10551057
list: () =>
10561058
Effect.succeed([
Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
import { expect } from "bun:test"
2+
import { Deferred, Effect, Fiber } from "effect"
3+
import { Credential } from "@opencode/core/credential"
4+
import { Integration } from "@opencode/schema/integration"
5+
import { LayerNode } from "@opencode/util/effect/layer-node"
6+
import { testEffect } from "./lib/effect"
7+
8+
const it = testEffect(LayerNode.compile(Credential.node))
9+
const id = Credential.ID.make("cred_refresh")
10+
const value = Credential.OAuth.make({
11+
type: "oauth",
12+
methodID: Integration.MethodID.make("oauth"),
13+
access: "new-access",
14+
refresh: "new-refresh",
15+
expires: Number.MAX_SAFE_INTEGER,
16+
})
17+
18+
it.effect("shares failures but allows a later refresh", () =>
19+
Effect.gen(function* () {
20+
const credentials = yield* Credential.Service
21+
const entered = yield* Deferred.make<void>()
22+
const release = yield* Deferred.make<void>()
23+
const error = new Error("token endpoint unavailable")
24+
let calls = 0
25+
const operation = Effect.gen(function* () {
26+
calls++
27+
yield* Deferred.succeed(entered, undefined)
28+
yield* Deferred.await(release)
29+
return yield* Effect.fail(error)
30+
})
31+
const first = yield* credentials.lock(id, operation).pipe(Effect.flip, Effect.forkScoped)
32+
yield* Deferred.await(entered)
33+
const second = yield* credentials.lock(id, operation).pipe(Effect.flip, Effect.forkScoped)
34+
yield* Effect.yieldNow
35+
yield* Deferred.succeed(release, undefined)
36+
expect(yield* Fiber.join(first)).toBe(error)
37+
expect(yield* Fiber.join(second)).toBe(error)
38+
expect(calls).toBe(1)
39+
expect(yield* credentials.lock(id, Effect.succeed(value))).toEqual(value)
40+
}),
41+
)

‎packages/core/test/fixture/config-nodes.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ export const emptyCredentialNode = makeGlobalNode({
88
layer: Layer.succeed(
99
Credential.Service,
1010
Credential.Service.of({
11+
lock: () => Effect.die("unused Credential.lock"),
1112
all: () => Effect.succeed([]),
1213
list: () => Effect.succeed([]),
1314
get: () => Effect.undefined,
Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
import { expect } from "bun:test"
2+
import { Context, Deferred, Effect, Fiber, Layer } from "effect"
3+
import { TestClock } from "effect/testing"
4+
import { Node } from "@opencode/util/effect/app-node"
5+
import { LayerNode } from "@opencode/util/effect/layer-node"
6+
import { Credential } from "@opencode/core/credential"
7+
import { Integration } from "@opencode/core/integration"
8+
import { testEffect } from "./lib/effect"
9+
10+
const it = testEffect(Layer.empty)
11+
12+
const locations = Effect.gen(function* () {
13+
const layer = LayerNode.compile(LayerNode.group([Integration.node, Credential.node]), {
14+
shared: Node.tags.values.global,
15+
})
16+
const memo = yield* Layer.makeMemoMap
17+
const scope = yield* Effect.scope
18+
const first = yield* Layer.buildWithMemoMap(layer, memo, scope)
19+
const second = yield* Layer.buildWithMemoMap(layer, memo, scope)
20+
const a = Context.get(first, Integration.Service)
21+
const b = Context.get(second, Integration.Service)
22+
const credentials = Context.get(first, Credential.Service)
23+
expect(a).not.toBe(b)
24+
expect(credentials).toBe(Context.get(second, Credential.Service))
25+
return { a, b, credentials }
26+
})
27+
28+
for (const cancel of [false, true]) {
29+
it.effect(`shares refresh across Locations${cancel ? " after the first caller is interrupted" : ""}`, () =>
30+
Effect.gen(function* () {
31+
const services = yield* locations
32+
const entered = yield* Deferred.make<void>()
33+
const release = yield* Deferred.make<void>()
34+
const integrationID = Integration.ID.make("openai")
35+
const methodID = Integration.MethodID.make("oauth")
36+
let calls = 0
37+
const registration = {
38+
integrationID,
39+
method: { id: methodID, type: "oauth" as const, label: "OAuth" },
40+
authorize: () => Effect.die("unused sign-in"),
41+
refresh: (value: Credential.OAuth) =>
42+
Effect.gen(function* () {
43+
calls++
44+
expect(value.refresh).toBe("old-refresh")
45+
yield* Deferred.succeed(entered, undefined)
46+
yield* Deferred.await(release)
47+
return Credential.OAuth.make({
48+
...value,
49+
access: "new-access",
50+
refresh: "new-refresh",
51+
expires: Number.MAX_SAFE_INTEGER,
52+
})
53+
}),
54+
}
55+
yield* services.a.transform((editor) => editor.method.update(registration))
56+
yield* services.b.transform((editor) => editor.method.update(registration))
57+
const stored = yield* services.credentials.create({
58+
integrationID,
59+
value: Credential.OAuth.make({
60+
type: "oauth",
61+
methodID,
62+
access: "old-access",
63+
refresh: "old-refresh",
64+
expires: 1,
65+
}),
66+
})
67+
const connection = { type: "credential" as const, id: stored.id, label: stored.label, method: "oauth" as const }
68+
const first = yield* services.a.connection.resolve(connection).pipe(Effect.forkScoped)
69+
yield* Deferred.await(entered)
70+
if (cancel) yield* Fiber.interrupt(first)
71+
const second = yield* services.b.connection.resolve(connection).pipe(Effect.forkScoped)
72+
yield* Effect.yieldNow
73+
expect(calls).toBe(1)
74+
yield* Deferred.succeed(release, undefined)
75+
const value = yield* Fiber.join(second)
76+
if (!cancel) expect(yield* Fiber.join(first)).toEqual(value)
77+
expect((yield* services.credentials.get(stored.id))?.value).toEqual(value)
78+
expect(yield* services.b.connection.resolve(connection)).toEqual(value)
79+
expect(calls).toBe(1)
80+
}),
81+
)
82+
}
83+
84+
it.effect("bounds a stuck integration refresh and allows a later attempt", () =>
85+
Effect.gen(function* () {
86+
const services = yield* locations
87+
const entered = yield* Deferred.make<void>()
88+
const integrationID = Integration.ID.make("openai")
89+
const methodID = Integration.MethodID.make("oauth")
90+
const stored = yield* services.credentials.create({
91+
integrationID,
92+
value: Credential.OAuth.make({ type: "oauth", methodID, access: "old", refresh: "refresh", expires: 1 }),
93+
})
94+
yield* services.a.transform((editor) =>
95+
editor.method.update({
96+
integrationID,
97+
method: { id: methodID, type: "oauth", label: "OAuth" },
98+
authorize: () => Effect.die("unused sign-in"),
99+
refresh: () => Deferred.succeed(entered, undefined).pipe(Effect.andThen(Effect.never)),
100+
}),
101+
)
102+
const connection = { type: "credential" as const, id: stored.id, label: stored.label, method: "oauth" as const }
103+
const first = yield* services.a.connection.resolve(connection).pipe(Effect.flip, Effect.forkScoped)
104+
yield* Deferred.await(entered)
105+
yield* TestClock.adjust("30 seconds")
106+
expect(yield* Fiber.join(first)).toBeInstanceOf(Integration.AuthorizationError)
107+
yield* services.a.transform((editor) =>
108+
editor.method.update({
109+
integrationID,
110+
method: { id: methodID, type: "oauth", label: "OAuth" },
111+
authorize: () => Effect.die("unused sign-in"),
112+
refresh: (value) =>
113+
Effect.succeed(Credential.OAuth.make({ ...value, access: "new", expires: Number.MAX_SAFE_INTEGER })),
114+
}),
115+
)
116+
expect(yield* services.a.connection.resolve(connection)).toMatchObject({ access: "new" })
117+
}),
118+
)

‎packages/core/test/integration.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ const failingCredentialNode = makeGlobalNode({
1818
layer: Layer.succeed(
1919
Credential.Service,
2020
Credential.Service.of({
21+
lock: () => Effect.die("unused Credential.lock"),
2122
all: () => Effect.succeed([]),
2223
list: () => Effect.succeed([]),
2324
get: () => Effect.undefined,

‎packages/core/test/mcp-oauth.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ const memoryCredentials = (initial: Credential.Info[]) => {
3737
const rows = new Map(initial.map((row) => [row.id, row]))
3838
const unused = () => Effect.die("unused credential method")
3939
const service = Credential.Service.of({
40+
lock: unused,
4041
all: unused,
4142
create: unused,
4243
activate: unused,

0 commit comments

Comments
 (0)