Merge remote-tracking branch 'origin/main' into ava-shim

Author: gastonfournier

src/repository/fetcher.ts

@@ -34,4 +34,6 @@ export interface PollingFetchingOptions extends CommonFetchingOptions {
 
 export interface StreamingFetchingOptions extends CommonFetchingOptions {
   eventSource?: EventSource;
+  maxFailuresUntilFailover?: number;
+  failureWindowMs?: number;
 }

src/repository/streaming-fail-over.ts

@@ -0,0 +1,97 @@
+export type FailEvent = NetworkEventError | HttpStatusError | ServerEvent;
+
+type BaseFailEvent = { occurredAt: Date; message: string };
+
+type NetworkEventError = BaseFailEvent & {
+  type: 'network-error';
+};
+
+type HttpStatusError = BaseFailEvent & {
+  statusCode: number;
+  type: 'http-status-error';
+};
+
+type ServerEvent = BaseFailEvent & {
+  event: string;
+  type: 'server-hint';
+};
+
+const FAILOVER_SERVER_HINTS = ['polling'];
+
+// explicitly including 429 here, this is used by Unleash to signal to the SDK
+// that there's already too many streaming clients connected and it should switch to polling
+const HARD_FAILOVER_STATUS_CODES = [401, 403, 404, 429, 501];
+const SOFT_FAILOVER_STATUS_CODES = [408, 500, 502, 503, 504];
+
+export class FailoverStrategy {
+  private failures: FailEvent[] = [];
+
+  constructor(
+    private readonly maxFails: number,
+    private readonly relaxTimeMs: number,
+  ) {}
+
+  shouldFailover(event: FailEvent, now: Date = new Date()): boolean {
+    const nowMs = now.getTime();
+
+    this.pruneOldFailures(nowMs);
+    switch (event.type) {
+      case 'http-status-error':
+        return this.handleHttpStatus(event);
+
+      case 'server-hint':
+        return this.handleServerEvent(event);
+
+      case 'network-error':
+        return this.handleNetwork(event);
+    }
+  }
+
+  private handleServerEvent(event: ServerEvent): boolean {
+    if (FAILOVER_SERVER_HINTS.includes(event.event)) {
+      return true;
+    }
+
+    // things like hard disconnects are triggered by rolling restarts or explicit
+    // shutdown. We expect Unleash to come back after such events so we ignore the
+    //  error here. If Unleash doesn't come back up, it'll be handled by the HTTP Status
+    //  events at some point in the near future
+    return false;
+  }
+
+  // Network shenanigans are basically always going to contribute to failover but
+  // never an immediate failover decision. Kinda impossible to know if things will
+  // get better sooo.. sliding window time!
+  private handleNetwork(event: NetworkEventError): boolean {
+    return this.hasTooManyFails(event);
+  }
+
+  private handleHttpStatus(event: HttpStatusError): boolean {
+    if (HARD_FAILOVER_STATUS_CODES.includes(event.statusCode)) {
+      return true;
+    } else if (SOFT_FAILOVER_STATUS_CODES.includes(event.statusCode)) {
+      return this.hasTooManyFails(event);
+    }
+    return false;
+  }
+
+  private hasTooManyFails(event: FailEvent): boolean {
+    this.failures.push(event);
+    return this.failures.length >= this.maxFails;
+  }
+
+  // Because SSE doesn't have a success event, we only prune on new failures.
+  // So we use this to build ourselves a sliding window of recent failures.
+  // Be cool if we didn't have to do this but I see no meaningful way
+  // to get ourselves an error window otherwise.
+  private pruneOldFailures(nowMs: number): void {
+    const cutoff = nowMs - this.relaxTimeMs;
+    let write = 0;
+    for (let read = 0; read < this.failures.length; read++) {
+      if (this.failures[read].occurredAt.getTime() >= cutoff) {
+        this.failures[write++] = this.failures[read];
+      }
+    }
+    this.failures.length = write;
+  }
+}

src/repository/streaming-fetcher.ts

@@ -5,16 +5,51 @@ import { resolveUrl } from '../url-utils';
 import { UnleashEvents } from '../events';
 import { EventSource } from '../event-source';
 import { FetcherInterface, StreamingFetchingOptions } from './fetcher';
+import { FailEvent, FailoverStrategy } from './streaming-fail-over';
 
 export class StreamingFetcher extends EventEmitter implements FetcherInterface {
   private eventSource: EventSource | undefined;
 
-  private options: StreamingFetchingOptions;
+  private readonly url: string;
 
-  constructor(options: StreamingFetchingOptions) {
+  private readonly appName: string;
+
+  private readonly instanceId: string;
+
+  private readonly headers?: Record<string, string>;
+
+  private readonly connectionId?: string;
+
+  private readonly onSaveDelta: StreamingFetchingOptions['onSaveDelta'];
+
+  private readonly onModeChange?: StreamingFetchingOptions['onModeChange'];
+
+  private readonly failoverStrategy: FailoverStrategy;
+
+  constructor({
+    url,
+    appName,
+    instanceId,
+    headers,
+    connectionId,
+    eventSource,
+    maxFailuresUntilFailover = 5,
+    failureWindowMs = 60_000,
+    onSaveDelta,
+    onModeChange,
+  }: StreamingFetchingOptions) {
     super();
-    this.options = options;
-    this.eventSource = options.eventSource;
+
+    this.url = url;
+    this.appName = appName;
+    this.instanceId = instanceId;
+    this.headers = headers;
+    this.connectionId = connectionId;
+    this.onSaveDelta = onSaveDelta;
+    this.onModeChange = onModeChange;
+
+    this.eventSource = eventSource;
+    this.failoverStrategy = new FailoverStrategy(maxFailuresUntilFailover, failureWindowMs);
   }
 
   private setupEventSource() {
@@ -23,46 +58,89 @@ export class StreamingFetcher extends EventEmitter implements FetcherInterface {
         await this.handleFlagsFromStream(event);
       });
       this.eventSource.addEventListener('unleash-updated', this.handleFlagsFromStream.bind(this));
-      this.eventSource.addEventListener('error', (error: unknown) => {
-        this.emit(UnleashEvents.Warn, error);
-      });
-      this.eventSource.addEventListener('end', (error: unknown) => {
-        this.emit(UnleashEvents.Warn, error);
-      });
+      this.eventSource.addEventListener('error', this.handleErrorEvent.bind(this));
+      this.eventSource.addEventListener('end', this.handleServerDisconnect.bind(this));
       this.eventSource.addEventListener('fetch-mode', this.handleModeChange.bind(this));
     }
   }
 
+  private async handleErrorEvent(error: any): Promise<void> {
+    const now = new Date();
+
+    const failEvent: FailEvent =
+      typeof error?.status === 'number'
+        ? {
+            type: 'http-status-error',
+            message: error.message ?? `Stream failed with http status code ${error.status}`,
+            statusCode: error.status,
+            occurredAt: now,
+          }
+        : {
+            type: 'network-error',
+            message: error.message ?? 'Network error occurred in streaming',
+            occurredAt: now,
+          };
+
+    await this.handleFailoverDecision(failEvent);
+  }
+
+  private async handleServerDisconnect(): Promise<void> {
+    const failEvent: FailEvent = {
+      type: 'network-error',
+      message: 'Server closed the streaming connection',
+      occurredAt: new Date(),
+    };
+
+    await this.handleFailoverDecision(failEvent);
+  }
+
+  private async handleFailoverDecision(event: FailEvent): Promise<void> {
+    const now = new Date();
+    const shouldFailover = this.failoverStrategy.shouldFailover(event, now);
+
+    if (!shouldFailover) {
+      return;
+    }
+
+    this.emit(UnleashEvents.Warn, event.message);
+
+    if (this.onModeChange) {
+      await this.onModeChange('polling');
+    }
+  }
+
   private async handleFlagsFromStream(event: { data: string }) {
     try {
       const data = parseClientFeaturesDelta(JSON.parse(event.data));
-      await this.options.onSaveDelta(data);
+      await this.onSaveDelta(data);
     } catch (err) {
       this.emit(UnleashEvents.Error, err);
     }
   }
 
   private async handleModeChange(event: { data: string }) {
-    try {
-      const newMode = event.data as 'polling' | 'streaming';
-      if (this.options.onModeChange) {
-        await this.options.onModeChange(newMode);
-      }
-    } catch (err) {
-      this.emit(UnleashEvents.Error, new Error(`Failed to handle mode change: ${err}`));
+    const newMode = event.data as 'polling' | 'streaming';
+
+    if (newMode === 'polling') {
+      await this.handleFailoverDecision({
+        type: 'server-hint',
+        event: `polling`,
+        message: 'Server has explicitly requested switching to polling mode',
+        occurredAt: new Date(),
+      });
     }
   }
 
   private createEventSource(): EventSource {
-    return new EventSource(resolveUrl(this.options.url, './client/streaming'), {
+    return new EventSource(resolveUrl(this.url, './client/streaming'), {
       headers: buildHeaders({
-        appName: this.options.appName,
-        instanceId: this.options.instanceId,
+        appName: this.appName,
+        instanceId: this.instanceId,
         etag: undefined,
         contentType: undefined,
-        custom: this.options.headers,
+        custom: this.headers,
         specVersionSupported: '5.2.0',
-        connectionId: this.options.connectionId,
+        connectionId: this.connectionId,
       }),
       readTimeoutMillis: 60000,
       initialRetryDelayMillis: 2000,