diff --git a/package.json b/package.json index ac08121..80850ac 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@flagsync/js-sdk", - "version": "0.7.1", + "version": "0.7.2-alpha.0", "description": "FlagSync SDK for JavaScript", "author": "Mike Chabot", "license": "Apache-2.0", diff --git a/src/logger/messages.ts b/src/logger/messages.ts index 5cf2053..6d048eb 100644 --- a/src/logger/messages.ts +++ b/src/logger/messages.ts @@ -10,6 +10,7 @@ const STORAGE_MANAGER_FACTORY_MESSAGE = { const STORAGE_MANAGER_MESSAGE = { STORAGE_SET_FLAGS: 'storing flags', + STORAGE_REPLACE_FLAGS: 'replacing flags', STORAGE_GET_FLAGS: 'getting flags', STORAGE_PARSE_FAIL: 'failed to parse flags from storage', } as const; @@ -22,10 +23,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: 'flags resynced after reconnect', + STREAM_RESYNC_FAILED: 'flags 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 046de46..46e5d7a 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/localstorage-manager.ts b/src/managers/storage/localstorage-manager.ts index bf872f3..1b12243 100644 --- a/src/managers/storage/localstorage-manager.ts +++ b/src/managers/storage/localstorage-manager.ts @@ -28,6 +28,19 @@ export function localStorageManager(settings: FsSettings): IStoreManager { localStorage.setItem(buildKey(), JSON.stringify(flagSet)); } + /** + * Replace the entire flag set, in memory and in localStorage. Unlike the + * merge in `set`, this drops flags absent from the incoming set — the only + * way deletes can take effect. + */ + function replace(incoming: FsFlagSet) { + flagSet = { + ...incoming, + }; + log.debug(formatter(MESSAGE.STORAGE_REPLACE_FLAGS)); + localStorage.setItem(buildKey(), JSON.stringify(flagSet)); + } + function get(): FsFlagSet { log.debug(formatter(MESSAGE.STORAGE_GET_FLAGS)); const cached = localStorage.getItem(buildKey()); @@ -55,6 +68,7 @@ export function localStorageManager(settings: FsSettings): IStoreManager { return { set, + replace, get, }; } diff --git a/src/managers/storage/memory-manager.ts b/src/managers/storage/memory-manager.ts index c6435eb..195eba2 100644 --- a/src/managers/storage/memory-manager.ts +++ b/src/managers/storage/memory-manager.ts @@ -23,6 +23,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_FLAGS)); + flagSet = { + ...incoming, + }; + } + function get(): FsFlagSet { log.debug(formatter(MESSAGE.STORAGE_GET_FLAGS)); return { @@ -32,6 +43,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 ffe94aa..b77a482 100644 --- a/src/managers/storage/storage-manger-factory.ts +++ b/src/managers/storage/storage-manger-factory.ts @@ -35,10 +35,10 @@ export function storageManagerFactory( } /** - * 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, @@ -48,5 +48,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 d5cc233..ea691f0 100644 --- a/src/managers/storage/types.ts +++ b/src/managers/storage/types.ts @@ -2,5 +2,6 @@ import type { 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 95615c1..e78155e 100644 --- a/src/managers/sync/poll-manager.ts +++ b/src/managers/sync/poll-manager.ts @@ -29,7 +29,7 @@ export function pollManager( }); 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 3601916..51c0bf0 100644 --- a/src/managers/sync/stream-manager.ts +++ b/src/managers/sync/stream-manager.ts @@ -3,6 +3,8 @@ import { EventSource } from 'extended-eventsource'; import type { FsFlagSet } from '~config/types'; import type { FsSettings } from '~config/types.internal'; +import { apiClientFactory } from '~api/api-client-factory'; + import { FsIntervalEvent, IEventManager } from '~managers/event/types'; import type { ISyncManager } from '~managers/sync/types'; @@ -11,19 +13,76 @@ import { formatMsg } from '~logger/utils'; const formatter = formatMsg.bind(null, 'stream-manager'); +/** + * 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, context } = settings; + const { sdk } = apiClientFactory(settings); - let es: EventSource; + let es: EventSource | undefined; + let watchdog: ReturnType | undefined; + let killed = false; + let hasConnected = false; + + /** + * Servers that support full-set sync send the entire evaluated value map + * 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); + } + + /** + * Updates emitted while the connection was down are lost — there is no + * replay on reconnect — so fetch the full value map to converge the store. + */ + async function resyncOnReconnect() { + try { + const res = await sdk.sdkControllerGetFlags({ + context, + }); + 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( + const source = new EventSource( `${urls.sse}/sse/sdk-updates/client?timestamp=${new Date().getTime()}`, { withCredentials: true, @@ -34,21 +93,57 @@ 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 evaluated value map, which + * replaces the store. This is how flag creates and deletes take effect. + */ + source.addEventListener('flags', (event: MessageEvent) => { + 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, 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 */ - es.onmessage = (event: MessageEvent) => { + source.onmessage = (event: MessageEvent) => { + resetWatchdog(); + if (serverSupportsFullSync) { + return; + } try { const flagSet = JSON.parse(event.data) as FsFlagSet; log.debug(formatter(MESSAGE.STREAM_MESSAGE_RECEIVED)); @@ -58,20 +153,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(), ); } @@ -79,6 +174,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 fb07e3b..655375a 100644 --- a/src/managers/sync/ws-manager.ts +++ b/src/managers/sync/ws-manager.ts @@ -17,6 +17,7 @@ export const wsManager = ( let ws: WebSocket; let reconnectTimeout: NodeJS.Timeout | null = null; + let killed = false; const RECONNECT_DELAY = 5000; const url = new URL(`${urls.ws.replace('https', 'wss')}/sdk/connect`); @@ -24,6 +25,9 @@ export const wsManager = ( url.searchParams.append('x-ridgeline-user-ctx', JSON.stringify(context)); function connect() { + if (killed) { + return; + } ws = new WebSocket(url.toString()); /** @@ -46,8 +50,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 evaluated value map, 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()); @@ -62,7 +71,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); } @@ -83,8 +92,21 @@ 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 + * reconnect after shutdown. + */ function kill() { - if (ws && ws.readyState === ws.OPEN) { + killed = true; + if (reconnectTimeout) { + clearTimeout(reconnectTimeout); + reconnectTimeout = null; + } + if ( + ws && + (ws.readyState === ws.OPEN || ws.readyState === ws.CONNECTING) + ) { log.debug(formatter(MESSAGE.STREAM_CONN_CLOSING)); // Use code 1000 for normal closure ws.close(1000, 'SDK shutting down.');