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", 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 dcecfb7..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'; @@ -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 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,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(), ); } @@ -80,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..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, { @@ -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.');