diff --git a/packages/stack/src/utils/expontential-backoff.ts b/packages/stack/src/utils/expontential-backoff.ts new file mode 100644 index 0000000..03188ce --- /dev/null +++ b/packages/stack/src/utils/expontential-backoff.ts @@ -0,0 +1,155 @@ +/** + * Exponential backoff is a technique used to retry a function after a delay. + * + * The delay increases exponentially with each attempt, up to a maximum delay. + * + * The jitter is a random amount of time added to the delay to prevent thundering herd problems. + * + * The growth rate is the factor by which the delay increases with each attempt. + */ +export class ExponentialBackoff { + /** + * Create a new ExponentialBackoff instance + * + * @param config - The configuration for the exponential backoff + * @returns The ExponentialBackoff instance + */ + static from(config?: Partial): ExponentialBackoff { + const backoff = new ExponentialBackoff(config); + return backoff; + } + + /** + * Run the function with exponential backoff + * + * @param fn - The function to run + * @param onError - The callback to call when an error occurs + * @param options - The configuration for the exponential backoff + * + * @throws The last error if the function fails and we have hit the max attempts + * + * @returns The result of the function + */ + static run( + fn: () => Promise, + onError = (_error: Error) => {}, + options?: Partial, + ): Promise { + const backoff = ExponentialBackoff.from(options); + return backoff.run(fn, onError); + } + + private readonly options: ExponentialBackoffOptions; + + constructor(options?: Partial) { + this.options = { + maxDelay: 10000, + maxAttempts: 10, + baseDelay: 1000, + growthRate: 2, + jitter: 0.1, + ...options, + }; + } + + /** + * Run the function with exponential backoff + * + * If the function fails but we have not hit the max attempts, the error will be passed to the onError callback + * and the function will be retried with an exponential delay + * + * If the function fails and we have hit the max attempts, the last error will be thrown + * + * @param fn - The function to run + * @param onError - The callback to call when an error occurs + * + * @throws The last error if the function fails and we have hit the max attempts + * + * @returns The result of the function + */ + async run( + fn: () => Promise, + onError = (_error: Error) => {}, + ): Promise { + let lastError: Error = new Error('Exponential backoff: Max retries hit'); + + let attempt = 0; + + while ( + attempt < this.options.maxAttempts || + this.options.maxAttempts == 0 + ) { + try { + return await fn(); + } catch (error) { + // Store the error in case we fail every attempt + lastError = error instanceof Error ? error : new Error(`${error}`); + onError(lastError); + + // Wait before going to the next attempt + const delay = this.calculateDelay(attempt); + await new Promise((resolve) => setTimeout(resolve, delay)); + } + + attempt++; + } + + // We completed the loop without ever succeeding. Throw the last error we got + throw lastError; + } + + /** + * Calculate the delay before we should attempt to retry + * + * NOTE: The maximum delay is (maxDelay * (1 + jitter)) + * + * @param attempt + * @returns The time in milliseconds before another attempt should be made + */ + private calculateDelay(attempt: number): number { + // Get the power of the growth rate + const power = Math.pow(this.options.growthRate, attempt); + + // Get the delay before jitter or limit + const rawDelay = this.options.baseDelay * power; + + // Cap the delay to the maximum. Do this before the jitter so jitter does not become larger than delay + const cappedDelay = Math.min(rawDelay, this.options.maxDelay); + + // Get the jitter direction. This will be between -1 and 1 + const jitterDirection = 2 * Math.random() - 1; + + // Calculate the jitter + const jitter = jitterDirection * this.options.jitter * cappedDelay; + + // Add the jitter to the delay + return cappedDelay + jitter; + } +} + +export type ExponentialBackoffOptions = { + /** + * The maximum delay between attempts in milliseconds + */ + maxDelay: number; + + /** + * The maximum number of attempts. Passing 0 will result in infinite attempts. + */ + maxAttempts: number; + + /** + * The base delay between attempts in milliseconds + */ + baseDelay: number; + + /** + * The growth rate of the delay + */ + growthRate: number; + + /** + * The jitter of the delay as a percentage of growthRate + */ + jitter: number; +}; diff --git a/packages/stack/src/utils/index.ts b/packages/stack/src/utils/index.ts index 4deb577..c372534 100644 --- a/packages/stack/src/utils/index.ts +++ b/packages/stack/src/utils/index.ts @@ -1,4 +1,5 @@ export * from './event-emitter.js'; +export * from './expontential-backoff.js'; export * from './ext-error.js'; export * from './inverse-bloom-filter.js'; export * from './memoizer.js'; diff --git a/packages/stack/src/utils/sse-session.ts b/packages/stack/src/utils/sse-session.ts index f4e20ad..2c3d854 100644 --- a/packages/stack/src/utils/sse-session.ts +++ b/packages/stack/src/utils/sse-session.ts @@ -1,3 +1,5 @@ +import { ExponentialBackoff } from './expontential-backoff.js'; + /** * A Server-Sent Events client implementation using fetch API. * Supports custom headers, POST requests, and is non-blocking. @@ -28,9 +30,6 @@ export class SSESession { // Listener for when the tab is hidden or shown. private visibilityChangeHandler: ((event: Event) => void) | null = null; - // Keep track of the current reconnection attempt to use for exponential backoff. - protected currentAttempt: number = 0; - // Text decoders and encoders for parsing the message buffer. private textDecoder: TextDecoder = new TextDecoder(); private textEncoder: TextEncoder = new TextEncoder(); @@ -62,11 +61,14 @@ export class SSESession { ...options, }; this.controller = new AbortController(); - + // Set up visibility change handling if in mobile browser environment if (typeof document !== 'undefined') { this.visibilityChangeHandler = this.handleVisibilityChange.bind(this); - document.addEventListener('visibilitychange', this.visibilityChangeHandler); + document.addEventListener( + 'visibilitychange', + this.visibilityChangeHandler, + ); } } @@ -82,7 +84,7 @@ export class SSESession { // When coming back to foreground, attempt to reconnect if not connected if (document.visibilityState === 'visible' && !this.connected) { - await this.attemptReconnect(); + await this.connect(); } } @@ -93,8 +95,7 @@ export class SSESession { if (this.connected) return; this.connected = true; - const abortController = new AbortController(); - this.controller = abortController; + this.controller = new AbortController(); const { method, headers, body } = this.options; @@ -106,100 +107,87 @@ export class SSESession { cache: 'no-store', }; - // Using a regular Promise executor (not async) - return new Promise((resolve, reject) => { - // Move the async logic inside a separate function - const establishConnection = async () => { - try { - // Fetch from the provided URL using our fetch function. - const response = await this.options.fetch(this.url, fetchOptions); - - if (!response.ok) { - throw new Error(`HTTP error! Status: ${response.status}`); - } - - if (!response.body) { - throw new Error('Response body is null'); - } - - // Reset the current attempt. - this.currentAttempt = 0; - - // Call the onConnected callback. - this.handleCallback(this.options.onConnected, undefined); - - const reader = response.body.getReader(); - - // Start the read here so we can resolve and then start handling the stream - // We will await this at the start of the loop, and redeclare it at the end of each loop iteration - let readPromise = reader.read(); - - // Resolve before starting the loop so that we can start performing other actions - // Keep this out of the loop to avoid flags or `already resolved` errors. - resolve(); - - while (true) { - const { done, value } = await readPromise; - - if (done) { - this.connected = false; - - // Call the onDisconnected callback. - this.handleCallback(this.options.onDisconnected, undefined); - - // If the connection was closed by the server, we want to attempt a reconnect if the connection should be persistent. - if (this.options.persistent) { - await this.attemptReconnect(); - } - - break; - } - - // Decode and parse the chunk - const events = this.parseEvents(value); - - // Handle each complete event - for (const event of events) { - if (this.options.onMessage) { - this.handleCallback(this.options.onMessage, event); - } - } - - // Start reading the next chunk - readPromise = reader.read(); - } - } catch (error) { - this.connected = false; - - // Call the onDisconnected callback. - this.handleCallback(this.options.onDisconnected, error); - - // If the connection was aborted using the controller, we don't need to call onError. - if (abortController.signal.aborted) { - return; - } - - // Call the onError callback. - // NOTE: we dont use the handleCallback here because it would result in 2 error callbacks. - try { - this.options.onError(error); - } catch (error) { - console.log(`SSE Session: onError callback error:`, error); - } - - // Attempt to reconnect if enabled - if (this.options.attemptReconnect) { - return await this.attemptReconnect(); - } - - // Reject the promise - reject(error); - } - }; - - // Execute the async function - establishConnection(); + const exponentialBackoff = ExponentialBackoff.from({ + baseDelay: this.options.retryDelay, + maxDelay: 10000, + maxAttempts: 0, + growthRate: 1.3, + jitter: 0.3, }); + + // Establish the connection and get the reader using the exponential backoff + const reader = await exponentialBackoff.run(async () => { + const res = await this.options.fetch(this.url, fetchOptions); + if (!res.ok) { + throw new Error(`HTTP error! Status: ${res.status}`); + } + + if (!res.body) { + throw new Error('Response body is null'); + } + + return res.body.getReader(); + }); + + // Call the onConnected callback + this.handleCallback(this.options.onConnected, undefined); + + const readStream = async () => { + try { + while (true) { + const { done, value } = await reader.read(); + + if (done) { + this.connected = false; + + // Call the onDisconnected callback. + this.handleCallback(this.options.onDisconnected, undefined); + + // If the connection was closed by the server, we want to attempt a reconnect if the connection should be persistent. + if (this.options.persistent) { + await this.connect(); + } + + break; + } + + const events = this.parseEvents(value); + + for (const event of events) { + if (this.options.onMessage) { + this.handleCallback(this.options.onMessage, event); + } + } + } + } catch (error) { + this.connected = false; + + // Call the onDisconnected callback. + this.handleCallback(this.options.onDisconnected, error); + + // If the connection was aborted using the controller, we don't need to call onError. + if (this.controller.signal.aborted) { + return; + } + + // Call the onError callback. + // NOTE: we dont use the handleCallback here because it would result in 2 error callbacks. + try { + this.options.onError(error); + } catch (error) { + console.log(`SSE Session: onError callback error:`, error); + } + + // Attempt to reconnect if enabled + if (this.options.attemptReconnect) { + await this.connect(); + } + } + }; + + readStream(); + + return; } protected parseEvents(chunk: Uint8Array): SSEvent[] { @@ -263,25 +251,6 @@ export class SSESession { return events; } - /** - * Attempts to reconnect to the SSE endpoint. - * @returns void - */ - protected attemptReconnect(): void { - // Don't attempt to reconnect if it's not enabled or if we are already connected - if (!this.options.attemptReconnect || this.connected) return; - - // Exponential backoff with a maximum of 10 seconds - // TODO: We should update this to use an Exponential Backoff utility when we have one. - this.currentAttempt++; - const delay = Math.min(this.options.retryDelay * Math.pow(1.3, this.currentAttempt - 1), 10000); - console.log(`Attempting to reconnect in ${delay}ms`); - - setTimeout(async () => { - await this.connect(); - }, delay); - } - /** * Override the onMessage callback. * @@ -300,7 +269,10 @@ export class SSESession { // Remove the visibility handler (This is only required on browsers) if (this.visibilityChangeHandler && typeof document !== 'undefined') { - document.removeEventListener('visibilitychange', this.visibilityChangeHandler); + document.removeEventListener( + 'visibilitychange', + this.visibilityChangeHandler, + ); this.visibilityChangeHandler = null; } } @@ -316,7 +288,7 @@ export class SSESession { /** * Will handle thrown errors from the callback and call the onError callback. * This is to avoid the sse-session from disconnecting from errors that are not a result of the sse-session itself. - * + * * @param callback The callback to handle. * @param args The arguments to pass to the callback. */