From 6a74b202810cd8cf411dfc6059f7974cef0d0510 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 16:30:33 +0000 Subject: [PATCH 1/4] Bypass Next.js patched fetch for SSE stream connection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Next.js monkey-patches the global fetch with caching layers that clone the response and buffer its entire body — the Data Cache in production, and an HMR cache in `next dev` that buffers even `cache: 'no-store'` requests. An SSE body never ends, so the buffered clone grows until the connection drops (e.g. Cloudflare terminating a long-lived socket), at which point Next.js logs: Failed to set fetch cache TypeError: terminated [cause]: SocketError: other side closed (UND_ERR_SOCKET) Prefer the original, un-patched fetch that Next.js exposes on the patched function (`_nextOriginalFetch`) when connecting the EventSource. A background SSE connection should not participate in render caching or dynamic tracking anyway. Falls back to the global fetch everywhere else. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01XHpkNV7Bi4SHg4cjon8HqP --- src/managers/sync/stream-manager.ts | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/managers/sync/stream-manager.ts b/src/managers/sync/stream-manager.ts index dcecfb7..5429554 100644 --- a/src/managers/sync/stream-manager.ts +++ b/src/managers/sync/stream-manager.ts @@ -11,6 +11,22 @@ 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 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; +} + export const streamManager = ( settings: FsSettings, eventManager: IEventManager, @@ -26,7 +42,7 @@ export const streamManager = ( es = new EventSource(`${urls.sse}/sse/sdk-updates/server`, { withCredentials: true, fetch: (input, init) => - fetch(input, { + getBaseFetch()(input, { ...init, headers: { ...init.headers, From 25d1101313d7daebe3cf63756b4fec4b4acf549c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 17:15:30 +0000 Subject: [PATCH 2/4] Full-set flag sync: replace store, heartbeat watchdog, reconnect resync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flag deletes and creates never took effect in a running SDK: the memory store only spread-merged incoming sets, and the SSE stream only carried partial updates. This wires the SDK up to the full-set sync the backends now provide on every sync path: - memory-manager gains replace(), which drops flags absent from the incoming set — the only way deletes can propagate. A new internal UPDATE_RECEIVED_FULL event routes full sets to it; UPDATE_RECEIVED stays merge-based for legacy partial updates. - stream-manager listens for the named "flags" SSE event (full ruleset, replace semantics) and ignores legacy partial messages once full-set support is observed, so a change is not applied twice. - stream-manager consumes the server's 25s "heartbeat" events to drive a staleness watchdog that restarts half-dead connections no error event would ever surface. - On reconnect, the stream manager refetches /sdk/rules and replaces the store, recovering updates emitted while disconnected (there is no server-side replay). - ws-manager and poll-manager switch to UPDATE_RECEIVED_FULL: Sunrise always pushes the entire ruleset and polling always fetches it, so both were already full sets being incorrectly merged. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01XHpkNV7Bi4SHg4cjon8HqP --- src/logger/messages.ts | 5 + src/managers/event/types.ts | 2 + src/managers/storage/memory-manager.ts | 12 ++ .../storage/storage-manger-factory.ts | 16 ++- src/managers/storage/types.ts | 1 + src/managers/sync/poll-manager.ts | 2 +- src/managers/sync/stream-manager.ts | 129 +++++++++++++++--- src/managers/sync/ws-manager.ts | 7 +- 8 files changed, 152 insertions(+), 22 deletions(-) diff --git a/src/logger/messages.ts b/src/logger/messages.ts index ef61e5f..e06cf04 100644 --- a/src/logger/messages.ts +++ b/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; diff --git a/src/managers/event/types.ts b/src/managers/event/types.ts index 2a94605..1c3d037 100644 --- a/src/managers/event/types.ts +++ b/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; } diff --git a/src/managers/storage/memory-manager.ts b/src/managers/storage/memory-manager.ts index 9d5d78a..1defb35 100644 --- a/src/managers/storage/memory-manager.ts +++ b/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, }; } diff --git a/src/managers/storage/storage-manger-factory.ts b/src/managers/storage/storage-manger-factory.ts index 6edb182..85ff53e 100644 --- a/src/managers/storage/storage-manger-factory.ts +++ b/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; } diff --git a/src/managers/storage/types.ts b/src/managers/storage/types.ts index 51af06a..c1254f5 100644 --- a/src/managers/storage/types.ts +++ b/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; } diff --git a/src/managers/sync/poll-manager.ts b/src/managers/sync/poll-manager.ts index 8f86724..67c71fe 100644 --- a/src/managers/sync/poll-manager.ts +++ b/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) { diff --git a/src/managers/sync/stream-manager.ts b/src/managers/sync/stream-manager.ts index 5429554..f1945d9 100644 --- a/src/managers/sync/stream-manager.ts +++ b/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'; @@ -27,19 +29,75 @@ function getBaseFetch(): 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) => getBaseFetch()(input, { @@ -51,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)); @@ -75,20 +170,20 @@ 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(), ); } @@ -96,6 +191,8 @@ export const streamManager = ( } function kill() { + killed = true; + clearTimeout(watchdog); if (es) { log.debug(formatter(MESSAGE.STREAM_CONN_CLOSING)); es.close(); diff --git a/src/managers/sync/ws-manager.ts b/src/managers/sync/ws-manager.ts index 2d32459..870719d 100644 --- a/src/managers/sync/ws-manager.ts +++ b/src/managers/sync/ws-manager.ts @@ -51,8 +51,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()); From abff62252a1268b5d3d8ac9aa5f1aa08681f7eba Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 7 Jul 2026 00:38:27 +0000 Subject: [PATCH 3/4] Fix WebSocket shutdown leaving the event loop pinned MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit kill() only closed the socket when readyState was OPEN and never cleared a pending reconnect timer. A SIGINT landing while the socket was reconnecting (or mid-handshake) left the timer alive, which then opened a fresh connection after shutdown — pinning the event loop and hanging Ctrl-C until force-killed. kill() now clears the reconnect timer, guards future reconnects with a killed flag, and tears the socket down in any state (terminate() during CONNECTING to avoid a noisy handshake-abort error; close(1000) when OPEN). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01XHpkNV7Bi4SHg4cjon8HqP --- src/managers/sync/ws-manager.ts | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/src/managers/sync/ws-manager.ts b/src/managers/sync/ws-manager.ts index 870719d..c5e0fd0 100644 --- a/src/managers/sync/ws-manager.ts +++ b/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, { @@ -72,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); } @@ -93,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.'); From 8f7ec9dd35f509f01e0f6201b2bd3d97767d17d9 Mon Sep 17 00:00:00 2001 From: Mike Chabot Date: Tue, 7 Jul 2026 02:33:44 +0000 Subject: [PATCH 4/4] 0.8.4-alpha.0 --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 75066f7..be33da0 100644 --- a/package.json +++ b/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",