Skip to content

Commit 6294582

Browse files
authored
Merge pull request #367 from RhysSullivan/rs/sdk-connection-refresh-dedup
sdk: dedup concurrent connection refreshes + surface reauth-required
2 parents 66887df + b94faa5 commit 6294582

9 files changed

Lines changed: 641 additions & 49 deletions

File tree

7.26 KB
Binary file not shown.

‎packages/core/sdk/src/connections.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,15 @@ export class ConnectionRefreshError extends Data.TaggedError(
100100
)<{
101101
readonly connectionId: ConnectionId;
102102
readonly message: string;
103+
/**
104+
* Set by providers when the refresh failed in a way that the stored
105+
* refresh token cannot recover from (RFC 6749 §5.2 `invalid_grant`
106+
* — the AS has revoked the grant, the user changed their password,
107+
* the refresh token rotated out from under us, ...). The SDK
108+
* translates this into a `ConnectionReauthRequiredError` so callers
109+
* can prompt the user to sign in again instead of silently retrying.
110+
*/
111+
readonly reauthRequired?: boolean;
103112
readonly cause?: unknown;
104113
}> {}
105114

‎packages/core/sdk/src/errors.ts‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,24 @@ export class ConnectionRefreshNotSupportedError extends Schema.TaggedError<Conne
111111
},
112112
) {}
113113

114+
/**
115+
* Raised by `connections.accessToken(id)` when the provider's refresh
116+
* handler reported that the stored refresh token is permanently
117+
* invalid (RFC 6749 §5.2 `invalid_grant` and friends). The caller —
118+
* typically a tool invocation — surfaces this so the UI can prompt the
119+
* user to sign in again. Distinct from `ConnectionRefreshError` so
120+
* "the network flaked, retry later" and "the grant is dead, re-auth"
121+
* don't collapse into one error tag at the plugin boundary.
122+
*/
123+
export class ConnectionReauthRequiredError extends Schema.TaggedError<ConnectionReauthRequiredError>()(
124+
"ConnectionReauthRequiredError",
125+
{
126+
connectionId: ConnectionId,
127+
provider: Schema.String,
128+
message: Schema.String,
129+
},
130+
) {}
131+
114132
// ---------------------------------------------------------------------------
115133
// Union type for convenience in signatures.
116134
// ---------------------------------------------------------------------------
@@ -127,4 +145,5 @@ export type ExecutorError =
127145
| SecretOwnedByConnectionError
128146
| ConnectionNotFoundError
129147
| ConnectionProviderNotRegisteredError
130-
| ConnectionRefreshNotSupportedError;
148+
| ConnectionRefreshNotSupportedError
149+
| ConnectionReauthRequiredError;

‎packages/core/sdk/src/executor.ts‎

Lines changed: 148 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { Effect, FiberRef, Option, Schema } from "effect";
1+
import { Deferred, Effect, FiberRef, Option, Schema } from "effect";
22
import {
33
StorageError,
44
typedAdapter,
@@ -41,6 +41,7 @@ import {
4141
import {
4242
ConnectionNotFoundError,
4343
ConnectionProviderNotRegisteredError,
44+
ConnectionReauthRequiredError,
4445
ConnectionRefreshNotSupportedError,
4546
NoHandlerError,
4647
PluginNotLoadedError,
@@ -219,6 +220,7 @@ export type Executor<TPlugins extends readonly AnyPlugin[] = []> = {
219220
| ConnectionNotFoundError
220221
| ConnectionProviderNotRegisteredError
221222
| ConnectionRefreshNotSupportedError
223+
| ConnectionReauthRequiredError
222224
| ConnectionRefreshError
223225
| StorageFailure
224226
>;
@@ -603,6 +605,25 @@ export const createExecutor = <
603605
// Connection providers keyed by `provider.key` — drive the refresh
604606
// lifecycle for connection-owned tokens.
605607
const connectionProviders = new Map<string, ConnectionProvider>();
608+
// In-flight refresh dedup. `connectionsAccessToken` stamps a
609+
// `Deferred` here before calling the provider's `refresh`; parallel
610+
// callers that walk in while a refresh is still running observe
611+
// the same Deferred and await its resolution instead of hitting
612+
// the AS a second time. The map is mutated under a semaphore so
613+
// check-or-register is atomic under fiber interleavings.
614+
const refreshInFlight = new Map<
615+
string,
616+
Deferred.Deferred<
617+
string,
618+
| ConnectionNotFoundError
619+
| ConnectionProviderNotRegisteredError
620+
| ConnectionRefreshNotSupportedError
621+
| ConnectionReauthRequiredError
622+
| ConnectionRefreshError
623+
| StorageFailure
624+
>
625+
>();
626+
const refreshInFlightLock = Effect.unsafeMakeSemaphore(1);
606627
const extensions: Record<string, object> = {};
607628

608629
// ------------------------------------------------------------------
@@ -1389,46 +1410,26 @@ export const createExecutor = <
13891410
);
13901411
});
13911412

1392-
// accessToken(id) — the single surface plugins use at invoke time.
1393-
// Resolves the backing secret, checks expiry, calls the provider's
1394-
// refresh handler if we're inside the skew window. New tokens are
1395-
// written back through the same provider and the connection row is
1396-
// patched with the new expiry.
1397-
const connectionsAccessToken = (
1398-
id: string,
1399-
): Effect.Effect<
1400-
string,
1413+
// Typed error union that `connectionsAccessToken` and every helper
1414+
// that participates in a refresh returns. Pulled out into a type
1415+
// alias because it has to match the Deferred's channel exactly —
1416+
// otherwise concurrent waiters and the leader diverge on the error
1417+
// type.
1418+
type AccessTokenError =
14011419
| ConnectionNotFoundError
14021420
| ConnectionProviderNotRegisteredError
14031421
| ConnectionRefreshNotSupportedError
1422+
| ConnectionReauthRequiredError
14041423
| ConnectionRefreshError
1405-
| StorageFailure
1406-
> =>
1424+
| StorageFailure;
1425+
1426+
// The actual work of a single refresh cycle, factored out so the
1427+
// concurrency gate (`connectionsAccessToken`) stays readable. Runs
1428+
// for the fiber that wins the `refreshInFlight` race.
1429+
const performRefresh = (
1430+
ref: ConnectionRef,
1431+
): Effect.Effect<string, AccessTokenError> =>
14071432
Effect.gen(function* () {
1408-
const row = yield* findInnermostConnectionRow(id);
1409-
if (!row) {
1410-
return yield* Effect.fail(
1411-
new ConnectionNotFoundError({
1412-
connectionId: ConnectionId.make(id),
1413-
}),
1414-
);
1415-
}
1416-
const ref = rowToConnection(row);
1417-
const now = Date.now();
1418-
const needsRefresh =
1419-
ref.expiresAt !== null &&
1420-
ref.expiresAt - CONNECTION_REFRESH_SKEW_MS <= now;
1421-
1422-
if (!needsRefresh) {
1423-
const current = yield* secretsGet(
1424-
ref.accessTokenSecretId as unknown as string,
1425-
);
1426-
if (current !== null) return current;
1427-
// Fall through to refresh if the stored token vanished — a
1428-
// genuinely-missing secret with no way to refresh is a
1429-
// hard-failure, same behavior as if `expires_at` had passed.
1430-
}
1431-
14321433
const provider = connectionProviders.get(ref.provider);
14331434
if (!provider) {
14341435
return yield* Effect.fail(
@@ -1451,15 +1452,36 @@ export const createExecutor = <
14511452
? yield* secretsGet(ref.refreshTokenSecretId as unknown as string)
14521453
: null;
14531454

1454-
const result = yield* provider.refresh({
1455-
connectionId: ref.id,
1456-
scopeId: ref.scopeId,
1457-
kind: ref.kind,
1458-
identityLabel: ref.identityLabel,
1459-
refreshToken: refreshTokenValue,
1460-
providerState: ref.providerState,
1461-
oauthScope: ref.oauthScope,
1462-
});
1455+
// RFC 6749 §5.2 `invalid_grant` (and anything else the
1456+
// provider tags with `reauthRequired`) is terminal — the
1457+
// stored refresh token can't recover. Translate into the
1458+
// caller-visible "re-authenticate" error so the UI can
1459+
// prompt sign-in instead of silently retrying.
1460+
const rawResult = yield* Effect.either(
1461+
provider.refresh({
1462+
connectionId: ref.id,
1463+
scopeId: ref.scopeId,
1464+
kind: ref.kind,
1465+
identityLabel: ref.identityLabel,
1466+
refreshToken: refreshTokenValue,
1467+
providerState: ref.providerState,
1468+
oauthScope: ref.oauthScope,
1469+
}),
1470+
);
1471+
if (rawResult._tag === "Left") {
1472+
const err = rawResult.left;
1473+
if (err.reauthRequired) {
1474+
return yield* Effect.fail(
1475+
new ConnectionReauthRequiredError({
1476+
connectionId: err.connectionId,
1477+
provider: ref.provider,
1478+
message: err.message,
1479+
}),
1480+
);
1481+
}
1482+
return yield* Effect.fail(err);
1483+
}
1484+
const result = rawResult.right;
14631485

14641486
yield* connectionsUpdateTokens({
14651487
id: ref.id,
@@ -1473,6 +1495,87 @@ export const createExecutor = <
14731495
return result.accessToken;
14741496
});
14751497

1498+
// accessToken(id) — the single surface plugins use at invoke time.
1499+
// Resolves the backing secret, checks expiry, calls the provider's
1500+
// refresh handler if we're inside the skew window. New tokens are
1501+
// written back through the same provider and the connection row is
1502+
// patched with the new expiry.
1503+
//
1504+
// Concurrent invokes on an expired token all share one refresh.
1505+
// The fiber that wins the `refreshInFlightLock` race registers a
1506+
// Deferred and performs the refresh; every other concurrent caller
1507+
// observes the Deferred and awaits its completion. The Deferred is
1508+
// pulled out of the map before the refresh result resolves so
1509+
// later invokes don't reuse a completed slot.
1510+
const connectionsAccessToken = (
1511+
id: string,
1512+
): Effect.Effect<string, AccessTokenError> =>
1513+
Effect.gen(function* () {
1514+
const row = yield* findInnermostConnectionRow(id);
1515+
if (!row) {
1516+
return yield* Effect.fail(
1517+
new ConnectionNotFoundError({
1518+
connectionId: ConnectionId.make(id),
1519+
}),
1520+
);
1521+
}
1522+
const ref = rowToConnection(row);
1523+
const now = Date.now();
1524+
const needsRefresh =
1525+
ref.expiresAt !== null &&
1526+
ref.expiresAt - CONNECTION_REFRESH_SKEW_MS <= now;
1527+
1528+
if (!needsRefresh) {
1529+
const current = yield* secretsGet(
1530+
ref.accessTokenSecretId as unknown as string,
1531+
);
1532+
if (current !== null) return current;
1533+
// Fall through to refresh if the stored token vanished — a
1534+
// genuinely-missing secret with no way to refresh is a
1535+
// hard-failure, same behavior as if `expires_at` had passed.
1536+
}
1537+
1538+
// Concurrency gate. `action` either returns the fresh access
1539+
// token (this fiber did the refresh) or the already-running
1540+
// Deferred that another fiber stamped into the map (this fiber
1541+
// piggybacks on their refresh).
1542+
const action = yield* refreshInFlightLock.withPermits(1)(
1543+
Effect.gen(function* () {
1544+
const existing = refreshInFlight.get(id);
1545+
if (existing) {
1546+
return {
1547+
kind: "await" as const,
1548+
deferred: existing,
1549+
};
1550+
}
1551+
const deferred = yield* Deferred.make<string, AccessTokenError>();
1552+
refreshInFlight.set(id, deferred);
1553+
return { kind: "lead" as const, deferred };
1554+
}),
1555+
);
1556+
1557+
if (action.kind === "await") {
1558+
return yield* action.deferred;
1559+
}
1560+
1561+
// Leader path: run the refresh, pipe the outcome into the
1562+
// Deferred (so waiters wake up), and then clear the map slot
1563+
// regardless of success or failure. Keeping clear+complete in
1564+
// the same `onExit` avoids a window where a late waiter sees
1565+
// the Deferred after it's been completed but before it's been
1566+
// evicted.
1567+
return yield* performRefresh(ref).pipe(
1568+
Effect.onExit((exit) =>
1569+
refreshInFlightLock.withPermits(1)(
1570+
Effect.gen(function* () {
1571+
refreshInFlight.delete(id);
1572+
yield* Deferred.done(action.deferred, exit);
1573+
}),
1574+
),
1575+
),
1576+
);
1577+
});
1578+
14761579
const connectionsListForCtx = () => connectionsList();
14771580

14781581
// ------------------------------------------------------------------

‎packages/core/sdk/src/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ export {
4242
ConnectionNotFoundError,
4343
ConnectionProviderNotRegisteredError,
4444
ConnectionRefreshNotSupportedError,
45+
ConnectionReauthRequiredError,
4546
type ExecutorError,
4647
} from "./errors";
4748

‎packages/core/sdk/src/plugin.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import type {
3030
import type {
3131
ConnectionNotFoundError,
3232
ConnectionProviderNotRegisteredError,
33+
ConnectionReauthRequiredError,
3334
ConnectionRefreshNotSupportedError,
3435
SecretOwnedByConnectionError,
3536
} from "./errors";
@@ -193,6 +194,7 @@ export interface PluginCtx<TStore = unknown> {
193194
| ConnectionNotFoundError
194195
| ConnectionProviderNotRegisteredError
195196
| ConnectionRefreshNotSupportedError
197+
| ConnectionReauthRequiredError
196198
| ConnectionRefreshError
197199
| StorageFailure
198200
>;

‎packages/plugins/oauth2/src/index.ts‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,13 @@ import { Data, Effect, ParseResult, Schema } from "effect";
2222

2323
export class OAuth2Error extends Data.TaggedError("OAuth2Error")<{
2424
readonly message: string;
25+
/**
26+
* RFC 6749 §5.2 error code, when the token endpoint returned one
27+
* (`invalid_grant`, `invalid_client`, `unauthorized_client`, ...).
28+
* Callers use this to distinguish terminal failures (a refresh token
29+
* the AS no longer honours → re-auth required) from transient ones.
30+
*/
31+
readonly error?: string;
2532
readonly cause?: unknown;
2633
}> {}
2734

@@ -104,8 +111,11 @@ export const buildAuthorizationUrl = (input: BuildAuthorizationUrlInput): string
104111
// Token endpoint response parsing
105112
// ---------------------------------------------------------------------------
106113

107-
const oauth2Error = (message: string, cause?: unknown): OAuth2Error =>
108-
new OAuth2Error({ message, cause });
114+
const oauth2Error = (
115+
message: string,
116+
cause?: unknown,
117+
error?: string,
118+
): OAuth2Error => new OAuth2Error({ message, cause, error });
109119

110120
// ---------------------------------------------------------------------------
111121
// Schemas
@@ -213,7 +223,13 @@ export const decodeTokenResponse = (
213223
);
214224
const description =
215225
envelope.error_description ?? envelope.error ?? `status ${response.status}`;
216-
return yield* Effect.fail(oauth2Error(`OAuth token exchange failed: ${description}`));
226+
return yield* Effect.fail(
227+
oauth2Error(
228+
`OAuth token exchange failed: ${description}`,
229+
undefined,
230+
envelope.error,
231+
),
232+
);
217233
}
218234

219235
return yield* decodeSuccessBody(record).pipe(

0 commit comments

Comments
 (0)