Harvmaster/xo-stack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'exponential-backoff-sse' into 'development'

Browse Source 
Feat: Add Exponential Backoff Utility and use it for the SSE-Session Utility

See merge request GeneralProtocols/xo/stack!97
This commit is contained in:
Jim Hamill
2025-05-27 00:38:38 +00:00
parent 2124bedb99 f2a824aa2e
commit 25b58853a3
3 changed files with 250 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stack/src/utils/expontential-backoff.ts
+155
View File
@@ -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<ExponentialBackoffOptions>): 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<T>(
fn: () => Promise<T>,
onError = (_error: Error) => {},
options?: Partial<ExponentialBackoffOptions>,
): Promise<T> {
const backoff = ExponentialBackoff.from(options);
return backoff.run(fn, onError);
}
private readonly options: ExponentialBackoffOptions;
constructor(options?: Partial<ExponentialBackoffOptions>) {
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<T>(
fn: () => Promise<T>,
onError = (_error: Error) => {},
): Promise<T> {
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;
};
packages/stack/src/utils/index.ts
+1
View File
@@ -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';
packages/stack/src/utils/sse-session.ts
+94 -122
View File
@@ -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.
*/