diff --git a/node/CHANGELOG.md b/node/CHANGELOG.md index b50d05b..bf3c1a1 100644 --- a/node/CHANGELOG.md +++ b/node/CHANGELOG.md @@ -12,5 +12,10 @@ All notable changes to the Node.js package are documented here. The format follo serverless caches and node-based replication groups. - `ElastiCacheIAMAuthTokenProvider` for generating fresh tokens when clients connect or reconnect. -- `ToolkitInputError`, `InvalidParameterError`, and `ConfigurationError`. +- `ElastiCacheIAMAuthTokenManager` for caching a token, refreshing it in the background + before it expires, retrying transient refresh failures, and notifying an + `onTokenChanged` callback. Its no-argument `getCredentials()` hook returns the user id + and current token together for client credential-provider integrations. +- `ToolkitInputError`, `InvalidParameterError`, `ConfigurationError`, and + `TokenRefreshError`. - Runnable token-generation and iovalkey integration examples under `examples/`. diff --git a/node/README.md b/node/README.md index ad60e40..bcc7f82 100644 --- a/node/README.md +++ b/node/README.md @@ -1,8 +1,8 @@ # Developer Toolkit for Amazon ElastiCache (Node.js) This package generates IAM authentication tokens for Amazon ElastiCache serverless caches -and node-based replication groups. It provides a typed TypeScript API and an async token -provider. +and node-based replication groups. It provides a typed TypeScript API, an async token +provider, and a caching token manager that refreshes tokens in the background. ## Installation @@ -59,6 +59,9 @@ shared config and credentials files, SSO, assume-role, container credentials, an instance metadata. One-shot calls resolve the region and credentials for each invocation, so refreshed configuration and credentials are used. +Token generation cannot detect expired or otherwise invalid credentials: signing may +succeed, but ElastiCache will reject the connection, for example with `WRONGPASS`. + ### Reconnecting clients `ElastiCacheIAMAuthTokenProvider` is client-agnostic. It resolves and retains the region @@ -84,9 +87,88 @@ const password = await auth.getToken(); ``` Pass `username` and `password` to the client connection configuration. Fetch another token -whenever the client creates a new connection. See -[`examples/connect-iovalkey.mjs`](examples/connect-iovalkey.mjs) for an iovalkey -integration. +whenever the client creates a new connection. + +### Caching and background refresh + +`ElastiCacheIAMAuthTokenProvider` is stateless: every `getToken()` call resolves +credentials and signs a new token. `ElastiCacheIAMAuthTokenManager` adds caching on top of +that: + +| | `ElastiCacheIAMAuthTokenProvider` | `ElastiCacheIAMAuthTokenManager` | +| ---------------------- | --------------------------------- | --------------------------------------------------- | +| Token per `getToken()` | A freshly signed token every call | The cached token while it is valid | +| Refresh | Caller-driven | Background, at `refreshAfterSeconds` after issuance | +| Refresh failures | Surfaced to the caller | Retried, with the current token still served | +| Change notification | None | `onTokenChanged` | +| Cleanup | None needed | `close()` | + +Use the manager when a client reconnects often, or asks for a token more often than the +900 second token lifetime, and use the provider when each call should mint a new token. + +```typescript +import { ElastiCacheIAMAuthTokenManager } from "@aws/developer-toolkit-elasticache"; + +const auth = await ElastiCacheIAMAuthTokenManager.create({ + serverlessCacheName: "my-cache", + userId: "iam-user", + region: "us-east-1", + // Optional; defaults to 300. Must be greater than 0 and less than 880. + refreshAfterSeconds: 300, + onTokenChanged: () => console.log("installed a new IAM auth token"), +}); + +try { + const [username, password] = await auth.getCredentials(); +} finally { + auth.close(); +} +``` + +`getCredentials()` is the no-argument provider hook for integrations that expect the user +id and token together. `getToken()` remains available when the client accepts the username +and password separately. + +The region is resolved when the manager is created, but credentials are not resolved and +no token is signed until the first `getToken()` or `getCredentials()` call. After that, +`getToken()` returns the cached token immediately while it is valid, starting a background +refresh once the token is past its refresh point. Concurrent callers with no valid token +share a single refresh. + +A token is served for 880 seconds, a 20 second margin below the 900 second token lifetime, +so a cached token is never handed to a client right at the edge of its real expiry. +`refreshAfterSeconds` must stay below that 880 second effective lifetime. + +A refresh makes up to 8 attempts, the first immediate and the rest with jittered +exponential backoff capped at 5 seconds. While the current token is still valid, a failed +refresh cycle is invisible to callers: the cached token is returned and another refresh +attempt is scheduled 5 seconds later. Callers that ask for a token during that 5 second +cooldown keep getting the valid cached token without starting a redundant refresh cycle of +their own; a successful refresh clears the cooldown. `getToken()` throws +`TokenRefreshError` only if the cached token has expired before a refresh could replace +it. The first token is different: its failure is reported as the underlying +`ConfigurationError` or `InvalidParameterError`. + +Call `refreshToken()` to force a new token immediately instead of waiting for the next +scheduled background refresh, for example after the client reports an authentication +failure. It invalidates the cached token and mints a replacement, sharing that refresh +with any concurrent `getToken()` or `refreshToken()` caller, including a background +refresh already under way. Like `getToken()`, it throws `TokenRefreshError` if the manager +is closed, and reports a first-mint-style failure if it could not produce a replacement. + +```typescript +// From the client's authentication-failure handling path: +const replacementPassword = await auth.refreshToken(); +// Supply replacementPassword on the next connection attempt. +``` + +The manager's timers are unreferenced, so they will not keep a process alive on their own. +Call `close()` when the manager is no longer needed; afterwards `getToken()` and +`refreshToken()` throw. + +See [`examples/connect-iovalkey.mjs`](examples/connect-iovalkey.mjs) for an iovalkey +integration, and [`examples/connect-node-redis.mjs`](examples/connect-node-redis.mjs) for +a node-redis integration using `credentialsProvider`. ### Examples @@ -98,15 +180,21 @@ node examples/generate-token.mjs ``` The iovalkey example demonstrates integration with a client that does not provide built-in -ElastiCache IAM authentication. It is an optional dependency and is not added to this -package's runtime dependencies. Install it in the application that runs the example, then -run: +ElastiCache IAM authentication, reconnecting manually on each new connection. The +node-redis example demonstrates a client with a native `credentialsProvider` hook instead. +Both clients are optional dependencies and are not added to this package's runtime +dependencies. Install one in the application that runs its example, then run: ```bash npm install iovalkey node examples/connect-iovalkey.mjs ``` +```bash +npm install redis +node examples/connect-node-redis.mjs +``` + ## Errors The public error hierarchy is: @@ -115,6 +203,11 @@ The public error hierarchy is: - `InvalidParameterError`: invalid values or target combinations. - `ConfigurationError`: missing AWS region or credentials. +`TokenRefreshError` is an operational failure rather than an input error, so it extends +`Error` directly. It is only thrown by `ElastiCacheIAMAuthTokenManager`, when a cached +token expired before a refresh could replace it or when the manager is closed. When a +refresh failed, its `cause` is the error from the last refresh attempt. + ## Security An IAM auth token is a short-lived bearer credential. Do not log, persist, or place it in diff --git a/node/examples/connect-iovalkey.mjs b/node/examples/connect-iovalkey.mjs index d68d540..b7d2dee 100644 --- a/node/examples/connect-iovalkey.mjs +++ b/node/examples/connect-iovalkey.mjs @@ -11,32 +11,56 @@ * The cache name used for signing is different from the connection endpoint. * For a node-based deployment, use replicationGroupId instead. * - * iovalkey does not generate ElastiCache IAM authentication tokens. This - * short-lived example disables automatic reconnects because they would reuse - * the initial token. Long-running applications should create a replacement - * client with a fresh token whenever the connection is lost. + * iovalkey does not generate ElastiCache IAM authentication tokens, and it + * cannot re-authenticate an existing connection. This example uses + * ElastiCacheIAMAuthTokenManager, which caches one token and refreshes it in + * the background, so every new client is created with a valid token without + * signing a new one per connection. Automatic reconnects are disabled because + * they would reuse the token the connection was opened with; long-running + * applications should create a replacement client, calling getCredentials() + * again, whenever the connection is lost. */ import { Valkey } from "iovalkey"; -import { ElastiCacheIAMAuthTokenProvider } from "@aws/developer-toolkit-elasticache"; +import { ElastiCacheIAMAuthTokenManager } from "@aws/developer-toolkit-elasticache"; -const auth = await ElastiCacheIAMAuthTokenProvider.create({ +const ENDPOINT = "my-cache-abc123.serverless.use1.cache.amazonaws.com"; + +const auth = await ElastiCacheIAMAuthTokenManager.create({ serverlessCacheName: "my-cache", userId: "my-iam-user", region: "us-east-1", + // Never log the token itself; it is a bearer credential. + onTokenChanged: () => console.log("installed a new IAM auth token"), }); -const client = new Valkey({ - host: "my-cache-abc123.serverless.use1.cache.amazonaws.com", - port: 6379, - username: auth.userId, - password: await auth.getToken(), - tls: {}, - retryStrategy: () => null, -}); +async function connect() { + const [username, password] = await auth.getCredentials(); + return new Valkey({ + host: ENDPOINT, + port: 6379, + username, + password, + tls: {}, + retryStrategy: () => null, + }); +} try { - await client.set("hello", "world"); - console.log(await client.get("hello")); + const client = await connect(); + try { + await client.set("hello", "world"); + console.log(await client.get("hello")); + } finally { + client.disconnect(); + } + + // A second connection reuses the cached token instead of signing another one. + const replacement = await connect(); + try { + console.log(await replacement.get("hello")); + } finally { + replacement.disconnect(); + } } finally { - client.disconnect(); + auth.close(); } diff --git a/node/examples/connect-node-redis.mjs b/node/examples/connect-node-redis.mjs new file mode 100644 index 0000000..c7d1ff4 --- /dev/null +++ b/node/examples/connect-node-redis.mjs @@ -0,0 +1,56 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +/* global console */ + +/** + * Run this example after installing redis: + * + * npm install redis + * + * The cache name used for signing is different from the connection endpoint. + * For a node-based deployment, use replicationGroupId instead. + * + * node-redis calls an AsyncCredentialsProvider's credentials() on every + * connection handshake, including its own automatic reconnects. Backing it + * with ElastiCacheIAMAuthTokenManager.getCredentials() means every handshake + * gets a valid cached token without signing a new one per connection, and + * node-redis's built-in reconnect strategy can stay enabled. + */ +import { createClient } from "redis"; +import { ElastiCacheIAMAuthTokenManager } from "@aws/developer-toolkit-elasticache"; + +const ENDPOINT = "my-cache-abc123.serverless.use1.cache.amazonaws.com"; + +const auth = await ElastiCacheIAMAuthTokenManager.create({ + serverlessCacheName: "my-cache", + userId: "my-iam-user", + region: "us-east-1", + // Never log the token itself; it is a bearer credential. + onTokenChanged: () => console.log("installed a new IAM auth token"), +}); + +const client = createClient({ + socket: { + host: ENDPOINT, + port: 6379, + tls: true, + }, + credentialsProvider: { + type: "async-credentials-provider", + credentials: async () => { + const [username, password] = await auth.getCredentials(); + return { username, password }; + }, + }, +}); +client.on("error", (error) => console.error("node-redis client error", error)); + +try { + await client.connect(); + await client.set("hello", "world"); + console.log(await client.get("hello")); +} finally { + await client.close(); + auth.close(); +} diff --git a/node/src/errors.ts b/node/src/errors.ts index dbbb6b6..da50e61 100644 --- a/node/src/errors.ts +++ b/node/src/errors.ts @@ -31,3 +31,15 @@ export class ConfigurationError extends ToolkitInputError { this.name = "ConfigurationError"; } } + +/** + * A cached token could not be refreshed before it expired, or the token manager + * was closed. This is an operational failure rather than an input error, so it + * does not extend {@link ToolkitInputError}. + */ +export class TokenRefreshError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "TokenRefreshError"; + } +} diff --git a/node/src/index.ts b/node/src/index.ts index e3d7192..bc723ac 100644 --- a/node/src/index.ts +++ b/node/src/index.ts @@ -4,6 +4,7 @@ export { ConfigurationError, InvalidParameterError, + TokenRefreshError, ToolkitInputError, } from "./errors.js"; export type { @@ -14,3 +15,10 @@ export { ElastiCacheIAMAuthTokenProvider, generateIamAuthToken, } from "./token-generator.js"; +export type { + TokenManagerDependencies, + TokenManagerOptions, + TokenManagerScheduler, + TokenManagerTimer, +} from "./token-manager.js"; +export { ElastiCacheIAMAuthTokenManager } from "./token-manager.js"; diff --git a/node/src/token-generator.ts b/node/src/token-generator.ts index b3d6599..1637382 100644 --- a/node/src/token-generator.ts +++ b/node/src/token-generator.ts @@ -14,7 +14,8 @@ import type { import { ConfigurationError, InvalidParameterError } from "./errors.js"; -const TOKEN_TTL_SECONDS = 900; +/** Lifetime of a presigned ElastiCache IAM token, in seconds. */ +export const TOKEN_TTL_SECONDS = 900; const SIGNING_SERVICE = "elasticache"; // The cache name becomes the SigV4 signing host. Keep this validation tight so a @@ -346,11 +347,21 @@ export class ElastiCacheIAMAuthTokenProvider { dependencies: TokenGeneratorDependencies = {}, ): Promise { const provider = new ElastiCacheIAMAuthTokenProvider(options, dependencies); - const regionResolution = await provider.regionResolution; + await provider.ensureRegionResolved(); + return provider; + } + + /** + * Await region resolution, throwing the retained configuration error if the + * region could not be resolved. + * + * @throws {ConfigurationError} when no region can be resolved. + */ + async ensureRegionResolved(): Promise { + const regionResolution = await this.regionResolution; if ("error" in regionResolution) { throw regionResolution.error; } - return provider; } get userId(): string { diff --git a/node/src/token-manager.ts b/node/src/token-manager.ts new file mode 100644 index 0000000..fd422d8 --- /dev/null +++ b/node/src/token-manager.ts @@ -0,0 +1,361 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { InvalidParameterError, TokenRefreshError } from "./errors.js"; +import { ElastiCacheIAMAuthTokenProvider, TOKEN_TTL_SECONDS } from "./token-generator.js"; +import type { + TokenGeneratorDependencies, + TokenGeneratorOptions, +} from "./token-generator.js"; + +const DEFAULT_REFRESH_AFTER_SECONDS = 300; +const MAX_ATTEMPTS = 8; +const BASE_DELAY_MS = 100; +const MAX_DELAY_MS = 5_000; +const JITTER_RATIO = 0.2; + +// ElastiCache tokens are valid for TOKEN_TTL_SECONDS from issuance, but clock +// skew and in-flight requests mean a token accepted at signing time can be +// rejected by the time it reaches the server. Serve tokens for a shorter, +// effective lifetime so a cached token is never handed out right at the edge +// of the real expiry. +const SERVE_MARGIN_SECONDS = 20; +const EFFECTIVE_TOKEN_LIFETIME_SECONDS = TOKEN_TTL_SECONDS - SERVE_MARGIN_SECONDS; + +const CLOSED_MESSAGE = "The token manager is closed."; +const REFRESH_FAILED_MESSAGE = + "Could not refresh the ElastiCache IAM authentication token before the " + + "previous token expired."; + +/** A cancellable timer that can be detached from the event loop. */ +export interface TokenManagerTimer { + unref(): void; +} + +/** Timer source, injectable so tests can advance time deterministically. */ +export interface TokenManagerScheduler { + setTimer(callback: () => void, delayMs: number): TokenManagerTimer; + clearTimer(timer: TokenManagerTimer): void; +} + +export interface TokenManagerOptions extends TokenGeneratorOptions { + /** + * Seconds after issuance at which a token is refreshed in the background. + * Must be greater than 0 and less than the effective token lifetime (the + * 900 second token lifetime minus a 20 second serve margin, i.e. 880). + * Defaults to 300. + */ + refreshAfterSeconds?: number; + /** + * Called with each newly installed token. Synchronous throws and rejected + * callback promises are ignored: they must not discard a token that was + * minted successfully. + */ + onTokenChanged?: (token: string) => void; +} + +/** + * Advanced dependencies for deterministic tests and applications with an + * explicitly managed AWS configuration. + */ +export interface TokenManagerDependencies extends TokenGeneratorDependencies { + /** Monotonic-enough clock in milliseconds. Defaults to `Date.now`. */ + now?: () => number; + /** Jitter source in [0, 1). Defaults to `Math.random`. */ + random?: () => number; + scheduler?: TokenManagerScheduler; +} + +type CachedToken = { + token: string; + issuedAt: number; + refreshAt: number; + expiresAt: number; +}; + +const defaultScheduler: TokenManagerScheduler = { + setTimer(callback, delayMs) { + return setTimeout(callback, delayMs); + }, + clearTimer(timer) { + clearTimeout(timer as ReturnType); + }, +}; + +function resolveRefreshAfterSeconds(value: unknown): number { + const resolved = value ?? DEFAULT_REFRESH_AFTER_SECONDS; + if ( + typeof resolved !== "number" || + !Number.isFinite(resolved) || + resolved <= 0 || + resolved >= EFFECTIVE_TOKEN_LIFETIME_SECONDS + ) { + throw new InvalidParameterError( + "Invalid value for parameter 'refreshAfterSeconds': must be greater than 0 " + + `and less than ${EFFECTIVE_TOKEN_LIFETIME_SECONDS}.`, + ); + } + return resolved; +} + +/** + * Caches an ElastiCache IAM authentication token and refreshes it before expiry. + * + * The first token is minted lazily. Subsequent calls reuse it while valid, and + * concurrent callers share one refresh. Call `close()` when the manager is no + * longer needed. + */ +export class ElastiCacheIAMAuthTokenManager { + private readonly provider: ElastiCacheIAMAuthTokenProvider; + private readonly refreshAfterMs: number; + private readonly onTokenChanged?: (token: string) => void; + private readonly now: () => number; + private readonly random: () => number; + private readonly scheduler: TokenManagerScheduler; + + private cached?: CachedToken; + private inFlight?: Promise; + private refreshTimer?: TokenManagerTimer; + private retryTimer?: TokenManagerTimer; + private cancelSleep?: () => void; + private closed = false; + /** + * When set, `getToken()` serves the cached token without starting another + * refresh cycle until this time: a refresh cycle is already scheduled to + * retry, and starting a second one would duplicate that work. Cleared by a + * successful refresh. + */ + private retryNotBefore?: number; + + constructor(options: TokenManagerOptions, dependencies: TokenManagerDependencies = {}) { + this.provider = new ElastiCacheIAMAuthTokenProvider(options, dependencies); + this.refreshAfterMs = resolveRefreshAfterSeconds(options.refreshAfterSeconds) * 1000; + this.onTokenChanged = options.onTokenChanged; + this.now = dependencies.now ?? Date.now; + this.random = dependencies.random ?? Math.random; + this.scheduler = dependencies.scheduler ?? defaultScheduler; + } + + /** + * Create a manager, resolving the region before returning. + * + * @throws {InvalidParameterError} for an invalid target, user id, or refresh interval. + * @throws {ConfigurationError} when no region can be resolved. + */ + static async create( + options: TokenManagerOptions, + dependencies: TokenManagerDependencies = {}, + ): Promise { + const manager = new ElastiCacheIAMAuthTokenManager(options, dependencies); + await manager.provider.ensureRegionResolved(); + return manager; + } + + get userId(): string { + return this.provider.userId; + } + + /** + * Return the normalized user id and a valid token for credential-provider + * integrations. + */ + async getCredentials(): Promise<[userId: string, token: string]> { + return [this.userId, await this.getToken()]; + } + + /** + * Return a valid token, minting one on the first call. + * + * While a cached token remains valid it is returned immediately. Once its + * refresh point is reached, refresh proceeds in the background. + * + * @throws {TokenRefreshError} when a refresh could not replace a token that has + * since expired, or when the manager is closed. + */ + async getToken(): Promise { + if (this.closed) { + throw new TokenRefreshError(CLOSED_MESSAGE); + } + + const cached = this.cached; + if (cached && this.now() < cached.expiresAt) { + if (this.now() >= cached.refreshAt && !this.inRetryCooldown()) { + this.startBackgroundRefresh(); + } + return cached.token; + } + return this.refresh(); + } + + /** + * Force a new token, invalidating the cached one instead of waiting for the + * next scheduled background refresh. + * + * A refresh already in progress, whether started here or in the background, + * is shared: concurrent callers wait for the same replacement token. + * + * @throws {TokenRefreshError} when the manager is closed. + * @throws {ConfigurationError} when no usable AWS credentials can be resolved. + */ + async refreshToken(): Promise { + if (this.closed) { + throw new TokenRefreshError(CLOSED_MESSAGE); + } + this.cached = undefined; + this.clearRefreshTimer(); + this.retryNotBefore = undefined; + return this.refresh(); + } + + /** + * Stop all background work and release the manager's timers. + * + * Subsequent `getToken()` calls reject; a refresh already in flight stops at + * its next attempt. + */ + close(): void { + this.closed = true; + this.clearRefreshTimer(); + this.clearRetryTimer(); + const cancelSleep = this.cancelSleep; + this.cancelSleep = undefined; + cancelSleep?.(); + } + + private refresh(): Promise { + if (!this.inFlight) { + const cycle = this.runRefreshCycle(); + this.inFlight = cycle; + const clear = () => { + if (this.inFlight === cycle) { + this.inFlight = undefined; + } + }; + cycle.then(clear, clear); // clear no matter the outcome + } + return this.inFlight; + } + + private startBackgroundRefresh(): void { + void this.refresh().catch(() => {}); + } + + private async runRefreshCycle(): Promise { + let lastError: unknown = new TokenRefreshError(CLOSED_MESSAGE); + + for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { + if (this.closed) { + throw new TokenRefreshError(CLOSED_MESSAGE, { cause: lastError }); + } + try { + const issuedAt = this.now(); + const token = await this.provider.getToken(); + if (this.closed) { + throw new TokenRefreshError(CLOSED_MESSAGE); + } + this.install(token, issuedAt); + return token; + } catch (error) { + lastError = error; + if (this.closed) { + throw new TokenRefreshError(CLOSED_MESSAGE, { cause: error }); + } + if (attempt === MAX_ATTEMPTS) { + break; + } + await this.sleep(this.backoffDelay(attempt)); + } + } + + const cached = this.cached; + + // If a cached token is still valid, return it and schedule another refresh. If + // there is no cached token, throw the last error. If there is a cached token + // but it has expired, throw a TokenRefreshError with the last error as its + // cause. + if (cached && this.now() < cached.expiresAt) { + this.retryNotBefore = this.now() + MAX_DELAY_MS; + this.scheduleRefresh(MAX_DELAY_MS); + return cached.token; + } + if (!cached) { + throw lastError; + } + throw new TokenRefreshError(REFRESH_FAILED_MESSAGE, { cause: lastError }); + } + + private install(token: string, issuedAt: number): void { + this.cached = { + token, + issuedAt, + refreshAt: issuedAt + this.refreshAfterMs, + expiresAt: issuedAt + EFFECTIVE_TOKEN_LIFETIME_SECONDS * 1000, + }; + this.retryNotBefore = undefined; + this.scheduleRefresh(this.refreshAfterMs); + this.notify(token); + } + + private inRetryCooldown(): boolean { + return this.retryNotBefore !== undefined && this.now() < this.retryNotBefore; + } + + private notify(token: string): void { + if (!this.onTokenChanged) { + return; + } + try { + void Promise.resolve(this.onTokenChanged(token)).catch(() => {}); + } catch { + // A consumer's callback failure must not invalidate a token that was + // minted successfully, and must not trigger another mint. + } + } + + private scheduleRefresh(delayMs: number): void { + this.clearRefreshTimer(); + if (this.closed) { + return; + } + const timer = this.scheduler.setTimer(() => { + this.refreshTimer = undefined; + this.startBackgroundRefresh(); + }, delayMs); + timer.unref(); + this.refreshTimer = timer; + } + + private sleep(delayMs: number): Promise { + return new Promise((resolve) => { + const finish = () => { + this.retryTimer = undefined; + this.cancelSleep = undefined; + resolve(); + }; + this.cancelSleep = finish; + const timer = this.scheduler.setTimer(finish, delayMs); + timer.unref(); + this.retryTimer = timer; + }); + } + + private backoffDelay(failedAttempt: number): number { + const delay = Math.min(BASE_DELAY_MS * 2 ** (failedAttempt - 1), MAX_DELAY_MS); + const jitter = 1 - JITTER_RATIO + 2 * JITTER_RATIO * this.random(); + return Math.round(delay * jitter); + } + + private clearRefreshTimer(): void { + if (this.refreshTimer) { + this.scheduler.clearTimer(this.refreshTimer); + this.refreshTimer = undefined; + } + } + + private clearRetryTimer(): void { + if (this.retryTimer) { + this.scheduler.clearTimer(this.retryTimer); + this.retryTimer = undefined; + } + } +} diff --git a/node/test/package.test.ts b/node/test/package.test.ts index 0bf233e..130a948 100644 --- a/node/test/package.test.ts +++ b/node/test/package.test.ts @@ -101,11 +101,14 @@ test("packed package installs and exposes its public API", () => { exportsProbePath, [ 'import assert from "node:assert/strict";', - 'import { ConfigurationError, ElastiCacheIAMAuthTokenProvider, InvalidParameterError, ToolkitInputError, generateIamAuthToken } from "@aws/developer-toolkit-elasticache";', + 'import { ConfigurationError, ElastiCacheIAMAuthTokenManager, ElastiCacheIAMAuthTokenProvider, InvalidParameterError, TokenRefreshError, ToolkitInputError, generateIamAuthToken } from "@aws/developer-toolkit-elasticache";', 'assert.equal(typeof generateIamAuthToken, "function");', 'assert.equal(typeof ElastiCacheIAMAuthTokenProvider, "function");', + 'assert.equal(typeof ElastiCacheIAMAuthTokenManager, "function");', "assert.equal(ConfigurationError.prototype instanceof ToolkitInputError, true);", "assert.equal(ConfigurationError.prototype instanceof InvalidParameterError, false);", + "assert.equal(TokenRefreshError.prototype instanceof ToolkitInputError, false);", + "assert.equal(TokenRefreshError.prototype instanceof Error, true);", ].join("\n"), ); const exportsProbe = spawnSync(process.execPath, [exportsProbePath], { @@ -127,7 +130,9 @@ test("package root exports the complete public API", async () => { assert.equal(typeof packageModule.generateIamAuthToken, "function"); assert.equal(typeof packageModule.ElastiCacheIAMAuthTokenProvider, "function"); + assert.equal(typeof packageModule.ElastiCacheIAMAuthTokenManager, "function"); assert.equal(typeof packageModule.ToolkitInputError, "function"); + assert.equal(typeof packageModule.TokenRefreshError, "function"); assert.equal(typeof packageModule.InvalidParameterError, "function"); assert.equal(typeof packageModule.ConfigurationError, "function"); assert.equal( @@ -139,4 +144,8 @@ test("package root exports the complete public API", async () => { packageModule.InvalidParameterError, false, ); + assert.equal( + packageModule.TokenRefreshError.prototype instanceof packageModule.ToolkitInputError, + false, + ); }); diff --git a/node/test/token-manager.test.ts b/node/test/token-manager.test.ts new file mode 100644 index 0000000..b793a90 --- /dev/null +++ b/node/test/token-manager.test.ts @@ -0,0 +1,603 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import { + ElastiCacheIAMAuthTokenManager, + InvalidParameterError, + TokenRefreshError, +} from "../src/index.js"; +import type { TokenManagerScheduler, TokenManagerTimer } from "../src/index.js"; + +const CACHE = "my-cache"; +const USER = "testuser"; +const REGION = "us-east-1"; +const CREDENTIALS = { + accessKeyId: "AKIAIOSFODNN7EXAMPLE", + secretAccessKey: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", +}; +const ROTATED_CREDENTIALS = { + accessKeyId: "AKIAI44QH8DHBEXAMPLE", + secretAccessKey: "other-secret", +}; +const SIGNING_DATE = new Date("2025-01-01T00:00:00Z"); + +function flush(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} + +class FakeTimer implements TokenManagerTimer { + unrefCount = 0; + cancelled = false; + + constructor( + readonly at: number, + readonly delayMs: number, + readonly callback: () => void, + ) {} + + unref(): void { + this.unrefCount++; + } +} + +class TestClock { + time = 0; + created: FakeTimer[] = []; + private pending: FakeTimer[] = []; + + readonly now = (): number => this.time; + + readonly scheduler: TokenManagerScheduler = { + setTimer: (callback, delayMs) => { + const timer = new FakeTimer(this.time + delayMs, delayMs, callback); + this.created.push(timer); + this.pending.push(timer); + return timer; + }, + clearTimer: (timer) => { + const fake = timer as FakeTimer; + fake.cancelled = true; + this.pending = this.pending.filter((candidate) => candidate !== fake); + }, + }; + + async advance(milliseconds: number): Promise { + const target = this.time + milliseconds; + while (true) { + const due = this.pending + .filter((timer) => !timer.cancelled && timer.at <= target) + .sort((left, right) => left.at - right.at)[0]; + if (!due) { + break; + } + this.pending = this.pending.filter((timer) => timer !== due); + this.time = due.at; + due.callback(); + await flush(); + } + this.time = target; + await flush(); + } +} + +type Harness = { + clock: TestClock; + manager: ElastiCacheIAMAuthTokenManager; + credentialCalls: () => number; + changedTokens: string[]; + fail: () => void; + succeed: () => void; +}; + +function harness( + overrides: { + refreshAfterSeconds?: number; + onTokenChanged?: (token: string) => void; + userId?: string; + } = {}, +): Harness { + const clock = new TestClock(); + let calls = 0; + let failing = false; + const changedTokens: string[] = []; + + const manager = new ElastiCacheIAMAuthTokenManager( + { + serverlessCacheName: CACHE, + userId: overrides.userId ?? USER, + region: REGION, + refreshAfterSeconds: overrides.refreshAfterSeconds, + onTokenChanged: overrides.onTokenChanged ?? ((token) => changedTokens.push(token)), + }, + { + credentialProvider: async () => { + calls++; + if (failing) { + throw new Error("credential chain failed"); + } + return calls === 1 ? CREDENTIALS : ROTATED_CREDENTIALS; + }, + signingDate: SIGNING_DATE, + now: clock.now, + random: () => 0.5, + scheduler: clock.scheduler, + }, + ); + + return { + clock, + manager, + credentialCalls: () => calls, + changedTokens, + fail: () => { + failing = true; + }, + succeed: () => { + failing = false; + }, + }; +} + +test("caches a token until its background refresh", async () => { + const { clock, manager, credentialCalls } = harness(); + try { + const first = await manager.getToken(); + const second = await manager.getToken(); + + assert.equal(second, first); + assert.equal(credentialCalls(), 1); + assert.equal(clock.created[0]?.delayMs, 300_000); + assert.equal(clock.created[0]?.unrefCount, 1); + } finally { + manager.close(); + } +}); + +test("refreshes in the background and installs credentials that rotated", async () => { + const { clock, manager, credentialCalls, changedTokens } = harness(); + try { + const first = await manager.getToken(); + await clock.advance(300_000); + const second = await manager.getToken(); + + assert.equal(credentialCalls(), 2); + assert.notEqual(second, first); + assert.deepEqual(changedTokens, [first, second]); + } finally { + manager.close(); + } +}); + +test("honours a custom refresh interval", async () => { + const { clock, manager, credentialCalls } = harness({ refreshAfterSeconds: 30 }); + try { + await manager.getToken(); + await clock.advance(29_999); + assert.equal(credentialCalls(), 1); + await clock.advance(1); + assert.equal(credentialCalls(), 2); + } finally { + manager.close(); + } +}); + +test("shares one initial token mint across concurrent callers", async () => { + const { manager, credentialCalls } = harness(); + try { + const [first, second, third] = await Promise.all([ + manager.getToken(), + manager.getToken(), + manager.getToken(), + ]); + + assert.equal(first, second); + assert.equal(second, third); + assert.equal(credentialCalls(), 1); + } finally { + manager.close(); + } +}); + +test("returns the user id and cached token from the credential-provider hook", async () => { + const { manager, credentialCalls } = harness({ userId: "TestUser" }); + try { + const [userId, token] = await manager.getCredentials(); + + assert.equal(userId, USER); + assert.equal(token, await manager.getToken()); + assert.equal(credentialCalls(), 1); + } finally { + manager.close(); + } +}); + +test("retries refresh failures with capped exponential backoff", async () => { + const { clock, manager, credentialCalls, fail } = harness(); + fail(); + + const result = manager.getToken().then( + () => undefined, + (error: unknown) => error, + ); + await flush(); + await clock.advance(12_000); + + assert.equal((await result) instanceof Error, true); + assert.equal(credentialCalls(), 8); + assert.deepEqual( + clock.created.map((timer) => timer.delayMs), + [100, 200, 400, 800, 1_600, 3_200, 5_000], + ); + assert.equal( + clock.created.every((timer) => timer.unrefCount === 1), + true, + ); + manager.close(); +}); + +test("keeps serving a valid token while refresh credentials are unavailable", async () => { + const { clock, manager, credentialCalls, fail, succeed } = harness(); + try { + const token = await manager.getToken(); + fail(); + await clock.advance(312_000); + + assert.equal(clock.created.at(-1)?.delayMs, 5_000); + assert.equal(await manager.getToken(), token); + // The cooldown scheduled above is still pending, so this call must not + // start a redundant, concurrent refresh cycle of its own. + assert.equal(credentialCalls(), 9); + + succeed(); + await clock.advance(5_000); + assert.notEqual(await manager.getToken(), token); + } finally { + manager.close(); + } +}); + +test("throws TokenRefreshError after the cached token expires", async () => { + const { clock, manager, fail } = harness({ refreshAfterSeconds: 879 }); + try { + await manager.getToken(); + fail(); + await clock.advance(911_000); + + const result = manager.getToken().then( + () => undefined, + (error: unknown) => error, + ); + await flush(); + await clock.advance(12_000); + + const error = await result; + assert.equal(error instanceof TokenRefreshError, true); + assert.equal((error as Error).cause instanceof Error, true); + } finally { + manager.close(); + } +}); + +test("ignores onTokenChanged callback failures", async () => { + const manager = harness({ + onTokenChanged: () => { + throw new Error("consumer callback failed"); + }, + }).manager; + try { + assert.equal(typeof (await manager.getToken()), "string"); + assert.equal(await manager.getToken(), await manager.getToken()); + } finally { + manager.close(); + } +}); + +test("ignores async onTokenChanged callback rejections", async () => { + const manager = harness({ + onTokenChanged: async () => { + throw new Error("async consumer callback failed"); + }, + }).manager; + try { + assert.equal(typeof (await manager.getToken()), "string"); + await flush(); + } finally { + manager.close(); + } +}); + +test("close cancels background work and rejects later requests", async () => { + const { clock, manager, credentialCalls } = harness(); + await manager.getToken(); + manager.close(); + await clock.advance(300_000); + + assert.equal(credentialCalls(), 1); + await assert.rejects(() => manager.getToken(), TokenRefreshError); +}); + +test("close prevents retry when an in-flight mint fails", async () => { + const clock = new TestClock(); + let rejectCredentials!: (reason: Error) => void; + const credentials = new Promise((_resolve, reject) => { + rejectCredentials = reject; + }); + const manager = new ElastiCacheIAMAuthTokenManager( + { + serverlessCacheName: CACHE, + userId: USER, + region: REGION, + }, + { + credentialProvider: () => credentials, + signingDate: SIGNING_DATE, + now: clock.now, + random: () => 0.5, + scheduler: clock.scheduler, + }, + ); + const result = manager.getToken(); + await flush(); + manager.close(); + + const rejection = assert.rejects(result, TokenRefreshError); + rejectCredentials(new Error("credential chain failed")); + await rejection; + assert.equal(clock.created.length, 0); +}); + +test("close discards a successful in-flight mint", async () => { + const clock = new TestClock(); + let resolveCredentials!: (credentials: typeof CREDENTIALS) => void; + const credentials = new Promise((resolve) => { + resolveCredentials = resolve; + }); + const changedTokens: string[] = []; + const manager = new ElastiCacheIAMAuthTokenManager( + { + serverlessCacheName: CACHE, + userId: USER, + region: REGION, + onTokenChanged: (token) => changedTokens.push(token), + }, + { + credentialProvider: () => credentials, + signingDate: SIGNING_DATE, + now: clock.now, + random: () => 0.5, + scheduler: clock.scheduler, + }, + ); + const result = manager.getToken(); + await flush(); + manager.close(); + + const rejection = assert.rejects(result, TokenRefreshError); + resolveCredentials(CREDENTIALS); + await rejection; + assert.deepEqual(changedTokens, []); + assert.equal(clock.created.length, 0); +}); + +for (const refreshAfterSeconds of [ + 0, + -1, + 880, + 899, + 900, + Number.NaN, + Number.POSITIVE_INFINITY, +]) { + test(`rejects invalid refreshAfterSeconds ${String(refreshAfterSeconds)}`, () => { + assert.throws( + () => + new ElastiCacheIAMAuthTokenManager({ + serverlessCacheName: CACHE, + userId: USER, + region: REGION, + refreshAfterSeconds, + }), + InvalidParameterError, + ); + }); +} + +test("accepts a refreshAfterSeconds just below the effective token lifetime", () => { + assert.doesNotThrow(() => + new ElastiCacheIAMAuthTokenManager({ + serverlessCacheName: CACHE, + userId: USER, + region: REGION, + refreshAfterSeconds: 879, + }).close(), + ); +}); + +test("treats a cached token as expired at the 880 second margin, not the raw 900 second lifetime", async () => { + const { clock, manager, fail } = harness({ refreshAfterSeconds: 879 }); + try { + await manager.getToken(); + fail(); + // The background cycle started at refreshAt (879s) exhausts its 8 attempts + // shortly after 890s: past the 880s effective expiry, but short of the raw + // 900s token lifetime. Under the raw lifetime this would still be "valid" + // and the cycle would fall back to serving the stale token instead. + await clock.advance(891_000); + + const result = manager.getToken().then( + () => undefined, + (error: unknown) => error, + ); + await flush(); + await clock.advance(12_000); + + assert.ok((await result) instanceof TokenRefreshError); + } finally { + manager.close(); + } +}); + +test("does not start a second refresh cycle while one is scheduled during cooldown", async () => { + const { clock, manager, credentialCalls, fail, succeed } = harness(); + try { + const token = await manager.getToken(); + fail(); + // Run the first refresh cycle to exhaustion; it falls back to serving the + // still-valid cached token and schedules a retry after MAX_DELAY_MS. + await clock.advance(311_300); + const callsAfterFirstCycle = credentialCalls(); + assert.equal(clock.created.at(-1)?.delayMs, 5_000); + + // Many concurrent reconnect-storm callers hit the cooldown window before + // the scheduled retry fires. None of them should start another cycle. + const results = await Promise.all( + Array.from({ length: 20 }, () => manager.getToken()), + ); + assert.equal( + results.every((result) => result === token), + true, + ); + assert.equal(credentialCalls(), callsAfterFirstCycle); + + succeed(); + await clock.advance(5_000); + assert.notEqual(await manager.getToken(), token); + } finally { + manager.close(); + } +}); + +test("refreshToken forces a new token ahead of the scheduled refresh", async () => { + const { manager, credentialCalls, changedTokens } = harness(); + try { + const first = await manager.getToken(); + const second = await manager.refreshToken(); + + assert.notEqual(second, first); + assert.equal(await manager.getToken(), second); + assert.equal(credentialCalls(), 2); + assert.deepEqual(changedTokens, [first, second]); + } finally { + manager.close(); + } +}); + +test("refreshToken shares one mint across concurrent callers", async () => { + const { manager, credentialCalls } = harness(); + try { + await manager.getToken(); + const [first, second, third] = await Promise.all([ + manager.refreshToken(), + manager.refreshToken(), + manager.refreshToken(), + ]); + + assert.equal(first, second); + assert.equal(second, third); + assert.equal(credentialCalls(), 2); + } finally { + manager.close(); + } +}); + +test("refreshToken joins an already in-flight background refresh", async () => { + const { clock, manager, credentialCalls } = harness(); + try { + const first = await manager.getToken(); + // Cross refreshAt without going through clock.advance()'s timer-firing loop, + // so the background cycle it starts is still in flight below. + clock.time = 300_000; + const background = manager.getToken(); + const forced = manager.refreshToken(); + const [backgroundToken, forcedToken] = await Promise.all([background, forced]); + + assert.equal(backgroundToken, first); + assert.notEqual(forcedToken, first); + assert.equal(credentialCalls(), 2); + assert.equal(await manager.getToken(), forcedToken); + } finally { + manager.close(); + } +}); + +test("refreshToken invalidates the cache while joining a failing background refresh", async () => { + const { clock, manager, credentialCalls, fail } = harness(); + try { + const token = await manager.getToken(); + fail(); + clock.time = 300_000; + + const backgroundToken = manager.getToken(); + const forcedResult = manager.refreshToken().then( + () => undefined, + (error: unknown) => error, + ); + const concurrentResult = manager.getToken().then( + () => undefined, + (error: unknown) => error, + ); + + assert.equal(await backgroundToken, token); + await flush(); + await clock.advance(12_000); + + assert.ok((await forcedResult) instanceof Error); + assert.ok((await concurrentResult) instanceof Error); + assert.equal(credentialCalls(), 9); + } finally { + manager.close(); + } +}); + +test("refreshToken rejects without a cached token when every attempt fails", async () => { + const { clock, manager, fail } = harness(); + fail(); + + const result = manager.refreshToken().then( + () => undefined, + (error: unknown) => error, + ); + await flush(); + await clock.advance(12_000); + + assert.equal((await result) instanceof Error, true); + manager.close(); +}); + +test("refreshToken rejects after close", async () => { + const { manager } = harness(); + manager.close(); + await assert.rejects(() => manager.refreshToken(), TokenRefreshError); +}); + +test("close aborts an in-flight refreshToken call", async () => { + const clock = new TestClock(); + let resolveCredentials!: (credentials: typeof CREDENTIALS) => void; + const credentials = new Promise((resolve) => { + resolveCredentials = resolve; + }); + const manager = new ElastiCacheIAMAuthTokenManager( + { + serverlessCacheName: CACHE, + userId: USER, + region: REGION, + }, + { + credentialProvider: () => credentials, + signingDate: SIGNING_DATE, + now: clock.now, + random: () => 0.5, + scheduler: clock.scheduler, + }, + ); + const result = manager.refreshToken(); + await flush(); + manager.close(); + + const rejection = assert.rejects(result, TokenRefreshError); + resolveCredentials(CREDENTIALS); + await rejection; +});