Claude/great hawking al4kaa

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@flagsync/node-sdk",
-  "version": "0.8.3",
+  "version": "0.8.4-alpha.0",
   "description": "FlagSync SDK for Node.js",
   "author": "Mike Chabot",
   "license": "Apache-2.0",

src/logger/messages.ts

@@ -5,6 +5,7 @@ const KILL_MANAGER_MESSAGE = {
 
 const STORAGE_MANAGER_MESSAGE = {
   STORAGE_SET_FLAG_RULES: 'storing flag rules',
+  STORAGE_REPLACE_FLAG_RULES: 'replacing flag rules',
   STORAGE_GET_FLAG_RULES: 'getting flag rules',
 } as const;
 
@@ -16,10 +17,14 @@ const SERVICE_MANAGER_MESSAGE = {
 const STREAM_MANAGER_MESSAGE = {
   STREAM_CONNECTED: 'connection established',
   STREAM_MESSAGE_RECEIVED: 'message received',
+  STREAM_FULL_SET_RECEIVED: 'full flag set received',
   STREAM_CONN_OPEN: 'connection is open',
   STREAM_CONN_CLOSE: 'ungraceful connection close',
   STREAM_CONN_CLOSING: 'gracefully closing event stream',
   STREAM_RECONNECT: 'reestablishing connection',
+  STREAM_STALE: 'no events or heartbeats received, restarting connection',
+  STREAM_RESYNC_SUCCESS: 'flag rules resynced after reconnect',
+  STREAM_RESYNC_FAILED: 'flag rules resync failed',
   STREAM_MALFORMED_EVENT: 'malformed message event',
   STREAM_UNKNOWN_EVENT_STATE: 'unknown error state',
 } as const;

src/managers/event/types.ts

@@ -43,6 +43,7 @@ export type FsEventType = (typeof FsEvent)[keyof typeof FsEvent];
 
 export const FsIntervalEvent = {
   UPDATE_RECEIVED: 'state::update-received',
+  UPDATE_RECEIVED_FULL: 'state::update-received-full',
 } as const;
 
 export type FsIntervalEventType =
@@ -61,4 +62,5 @@ export interface FsEventTypePayload {
 
 export interface FsInternalEventTypePayload {
   [FsIntervalEvent.UPDATE_RECEIVED]: EventFlagSetPayload;
+  [FsIntervalEvent.UPDATE_RECEIVED_FULL]: EventFlagSetPayload;
 }

src/managers/storage/memory-manager.ts

@@ -21,6 +21,17 @@ export function memoryManager(params: FsSettings): IStoreManager {
     };
   }
 
+  /**
+   * Replace the entire flag set. Unlike the merge in `set`, this drops flags
+   * absent from the incoming set — the only way deletes can take effect.
+   */
+  function replace(incoming: FsFlagSet) {
+    log.debug(formatter(MESSAGE.STORAGE_REPLACE_FLAG_RULES));
+    flagSet = {
+      ...incoming,
+    };
+  }
+
   function get(): FsFlagSet {
     log.debug(formatter(MESSAGE.STORAGE_GET_FLAG_RULES));
     return {
@@ -30,6 +41,7 @@ export function memoryManager(params: FsSettings): IStoreManager {
 
   return {
     set,
+    replace,
     get,
   };
 }

src/managers/storage/storage-manger-factory.ts

@@ -13,10 +13,10 @@ export function storageManagerFactory(
   const manager = memoryManager(params);
 
   /**
-   * The sync managers emit an internal event when an update is received, either
-   * by stream or poll. Streaming updates only include the changed flags, while
-   * poll updates include the entire flag set. The storage manager spreads
-   * the update, partial or full.
+   * The sync managers emit internal events when updates are received.
+   * UPDATE_RECEIVED carries a partial set (legacy SSE updates) and is merged;
+   * UPDATE_RECEIVED_FULL carries the entire flag set (poll, WebSocket, and
+   * full-sync SSE) and replaces the store, so deleted flags drop out.
    */
   eventManager.internal.on(
     FsIntervalEvent.UPDATE_RECEIVED,
@@ -26,5 +26,13 @@ export function storageManagerFactory(
     },
   );
 
+  eventManager.internal.on(
+    FsIntervalEvent.UPDATE_RECEIVED_FULL,
+    (flagSet: FsFlagSet) => {
+      manager.replace(flagSet);
+      eventManager.emit(FsEvent.SDK_UPDATE);
+    },
+  );
+
   return manager;
 }

src/managers/storage/types.ts

@@ -2,5 +2,6 @@ import { FsFlagSet } from '~config/types';
 
 export interface IStoreManager {
   set: (flagSet: FsFlagSet) => void;
+  replace: (flagSet: FsFlagSet) => void;
   get: () => FsFlagSet;
 }

src/managers/sync/poll-manager.ts

@@ -27,7 +27,7 @@ export function pollManager(
       const res = await sdk.sdkControllerGetFlagRules();
       log.debug(formatter(MESSAGE.POLL_SUCCESS));
       eventManager.internal.emit(
-        FsIntervalEvent.UPDATE_RECEIVED,
+        FsIntervalEvent.UPDATE_RECEIVED_FULL,
         res?.flags ?? {},
       );
     } catch (e) {

src/managers/sync/stream-manager.ts

@@ -3,6 +3,8 @@ import { EventSource } from 'eventsource';
 import { FsFlagSet } from '~config/types';
 import { FsSettings } from '~config/types.internal';
 
+import { apiClientFactory } from '~api/api-client-factory';
+
 import { FsIntervalEvent, IEventManager } from '~managers/event/types';
 import { ISyncManager } from '~managers/sync/types';
 
@@ -11,22 +13,94 @@ import { formatMsg } from '~logger/utils';
 
 const formatter = formatMsg.bind(null, 'stream-manager');
 
+/**
+ * Next.js patches the global fetch with caching layers (the Data Cache, and
+ * in `next dev` an HMR cache) that clone the response and buffer its entire
+ * body. An SSE body never ends, so buffering it leaks memory and logs
+ * "Failed to set fetch cache <url> TypeError: terminated" once the connection
+ * drops. `eventsource` already sends `cache: 'no-store'`, but the dev-time
+ * HMR cache buffers even `no-store` requests, so prefer the original,
+ * un-patched fetch that Next.js exposes on the patched function.
+ */
+function getBaseFetch(): typeof fetch {
+  const patched = globalThis.fetch as typeof fetch & {
+    _nextOriginalFetch?: typeof fetch;
+  };
+  return patched._nextOriginalFetch ?? patched;
+}
+
+/**
+ * The server emits a heartbeat every 25s to keep intermediaries (Cloudflare)
+ * from closing the connection as idle. If nothing at all arrives for several
+ * heartbeat periods, the connection is presumed half-dead and restarted.
+ */
+const STALE_CONNECTION_TIMEOUT_MS = 80_000;
+
 export const streamManager = (
   settings: FsSettings,
   eventManager: IEventManager,
 ): ISyncManager => {
   const { urls, log, sdkContext } = settings;
+  const { sdk } = apiClientFactory(settings);
 
-  let es: EventSource;
+  let es: EventSource | undefined;
+  let watchdog: NodeJS.Timeout | undefined;
+  let killed = false;
+  let hasConnected = false;
+
+  /**
+   * Servers that support full-set sync send the entire ruleset as named
+   * "flags" events. Once one is seen, legacy partial "message" events are
+   * ignored to avoid applying the same change twice.
+   */
+  let serverSupportsFullSync = false;
+
+  /**
+   * Restart the connection if no event, heartbeat, or open arrives within
+   * the stale window. Catches half-dead sockets that emit no error, which
+   * `EventSource` would otherwise never recover from.
+   */
+  function resetWatchdog() {
+    clearTimeout(watchdog);
+    watchdog = setTimeout(() => {
+      log.warn(formatter(MESSAGE.STREAM_STALE));
+      es?.close();
+      start();
+    }, STALE_CONNECTION_TIMEOUT_MS);
+    watchdog.unref?.();
+  }
+
+  /**
+   * Updates emitted while the connection was down are lost — there is no
+   * replay on reconnect — so fetch the full ruleset to converge the store.
+   */
+  async function resyncOnReconnect() {
+    try {
+      const res = await sdk.sdkControllerGetFlagRules();
+      eventManager.internal.emit(
+        FsIntervalEvent.UPDATE_RECEIVED_FULL,
+        res?.flags ?? {},
+      );
+      log.debug(formatter(MESSAGE.STREAM_RESYNC_SUCCESS));
+    } catch (error) {
+      log.error(formatter(MESSAGE.STREAM_RESYNC_FAILED), error?.toString());
+    }
+  }
 
   function start() {
+    if (killed) {
+      return;
+    }
+
     /**
      * Create a new EventSource instance and listen for incoming flag updates.
+     * Handlers close over `source` rather than the reassignable `es`, so a
+     * lingering handler from a restarted connection can't act on the new one.
      */
-    es = new EventSource(`${urls.sse}/sse/sdk-updates/server`, {
+    const source = new EventSource(`${urls.sse}/sse/sdk-updates/server`, {
       withCredentials: true,
       fetch: (input, init) =>
-        fetch(input, {
+        getBaseFetch()(input, {
           ...init,
           headers: {
             ...init.headers,
@@ -35,21 +109,58 @@ export const streamManager = (
           },
         }),
     });
+    es = source;
 
-    /**
-     * For debug only
-     */
-    es.onopen = () => {
+    resetWatchdog();
+
+    source.onopen = () => {
       log.debug(formatter(MESSAGE.STREAM_CONNECTED));
+      resetWatchdog();
+      if (hasConnected) {
+        resyncOnReconnect();
+      }
+      hasConnected = true;
     };
 
     /**
-     * When a message is received, parse the JSON and emit an event
-     * to the event manager. This is only a partial update, that is,
-     * the flag that changed.
+     * Full-set sync: the server sends the entire ruleset, which replaces
+     * the store. This is how flag creates and deletes take effect.
      * @param event
      */
-    es.onmessage = (event) => {
+    source.addEventListener('flags', (event) => {
+      resetWatchdog();
+      try {
+        const flagSet = JSON.parse(event.data) as FsFlagSet;
+        log.debug(formatter(MESSAGE.STREAM_FULL_SET_RECEIVED));
+        serverSupportsFullSync = true;
+        eventManager.internal.emit(
+          FsIntervalEvent.UPDATE_RECEIVED_FULL,
+          flagSet,
+        );
+      } catch (error) {
+        log.error(formatter(MESSAGE.STREAM_MALFORMED_EVENT), error?.toString());
+      }
+    });
+
+    /**
+     * Heartbeats carry no data; they keep intermediaries from closing the
+     * connection as idle and feed the staleness watchdog.
+     */
+    source.addEventListener('heartbeat', () => {
+      resetWatchdog();
+    });
+
+    /**
+     * Legacy partial update: only the changed flag, merged into the store.
+     * Servers that send full-set "flags" events also send these for older
+     * SDKs; ignore them once full-set support has been observed.
+     * @param event
+     */
+    source.onmessage = (event) => {
+      resetWatchdog();
+      if (serverSupportsFullSync) {
+        return;
+      }
       try {
         const flagRule = JSON.parse(event.data) as FsFlagSet;
         log.debug(formatter(MESSAGE.STREAM_MESSAGE_RECEIVED));
@@ -59,27 +170,29 @@ export const streamManager = (
       }
     };
 
-    es.onerror = (event: Event) => {
-      switch (es.readyState) {
-        case es.CONNECTING:
+    source.onerror = (event: Event) => {
+      switch (source.readyState) {
+        case source.CONNECTING:
           log.debug(formatter(MESSAGE.STREAM_RECONNECT));
           break;
-        case es.OPEN:
+        case source.OPEN:
           log.debug(formatter(MESSAGE.STREAM_CONN_OPEN));
           break;
-        case es.CLOSED:
+        case source.CLOSED:
           log.debug(formatter(MESSAGE.STREAM_CONN_CLOSE));
           break;
         default:
           log.debug(
-            `${formatter(MESSAGE.STREAM_UNKNOWN_EVENT_STATE)}: "${es.readyState}"`,
+            `${formatter(MESSAGE.STREAM_UNKNOWN_EVENT_STATE)}: "${source.readyState}"`,
             event.toString(),
           );
       }
     };
   }
 
   function kill() {
+    killed = true;
+    clearTimeout(watchdog);
     if (es) {
       log.debug(formatter(MESSAGE.STREAM_CONN_CLOSING));
       es.close();

src/managers/sync/ws-manager.ts

@@ -19,9 +19,13 @@ export const wsManager = (
 
   let ws: WebSocket;
   let reconnectTimeout: NodeJS.Timeout | null = null;
+  let killed = false;
   const RECONNECT_DELAY = 5000;
 
   function connect() {
+    if (killed) {
+      return;
+    }
     const wsUrl = `${urls.ws.replace('https', 'wss')}/sdk/connect`;
 
     ws = new WebSocket(wsUrl, {
@@ -51,8 +55,13 @@ export const wsManager = (
         const data = JSON.parse(event.data.toString());
         log.debug(formatter(MESSAGE.STREAM_MESSAGE_RECEIVED));
         if (data.type === 'flagUpdate') {
+          // Sunrise always pushes the entire ruleset, so replace the store
+          // rather than merge — this is how deletes propagate.
           const ruleset = data.flags as FsFlagSet;
-          eventManager.internal.emit(FsIntervalEvent.UPDATE_RECEIVED, ruleset);
+          eventManager.internal.emit(
+            FsIntervalEvent.UPDATE_RECEIVED_FULL,
+            ruleset,
+          );
         }
       } catch (error) {
         log.error(formatter(MESSAGE.STREAM_MALFORMED_EVENT), error?.toString());
@@ -67,7 +76,7 @@ export const wsManager = (
         formatter(MESSAGE.STREAM_CONN_CLOSE),
         `Code: ${event.code}, Reason: ${event.reason}`,
       );
-      if (event.code !== 1000) {
+      if (!killed && event.code !== 1000) {
         log.debug(formatter(MESSAGE.STREAM_RECONNECT));
         reconnectTimeout = setTimeout(connect, RECONNECT_DELAY);
       }
@@ -88,8 +97,22 @@ export const wsManager = (
     connect();
   }
 
+  /**
+   * A kill must tear down whatever state the connection cycle is in: a
+   * pending reconnect timer or a CONNECTING socket would otherwise keep the
+   * event loop alive (and reconnect after shutdown), hanging SIGINT.
+   */
   function kill() {
-    if (ws && ws.readyState === ws.OPEN) {
+    killed = true;
+    if (reconnectTimeout) {
+      clearTimeout(reconnectTimeout);
+      reconnectTimeout = null;
+    }
+    if (ws && ws.readyState === ws.CONNECTING) {
+      // close() mid-handshake surfaces an abort error; terminate() doesn't.
+      log.debug(formatter(MESSAGE.STREAM_CONN_CLOSING));
+      ws.terminate();
+    } else if (ws && ws.readyState === ws.OPEN) {
       log.debug(formatter(MESSAGE.STREAM_CONN_CLOSING));
       // Use code 1000 for normal closure
       ws.close(1000, 'SDK shutting down.');