feat: support SSE for update evaluation - #333
Draft
duyhungtnn wants to merge 55 commits into
Draft
Conversation
There was a problem hiding this comment.
Pull request overview
Adds Server-Sent Events (SSE)–based streaming evaluations to the JavaScript/TypeScript client SDK, allowing the scheduler to use a persistent stream (with backoff, watchdog, and optional polling fallback) instead of periodic polling.
Changes:
- Introduces an internal streaming stack (
FetchEventSource,StreamConnection,StreamingTask) with backoff/watchdog handling and optional fallback-to-polling behavior. - Extends public config (
enableStreaming,streamingFallbackToPolling, optionaleventSource) and switchesTaskSchedulerto select streaming vs polling. - Refactors
EvaluationInteractorto exposeapplyEvaluationsResponse()and adds Vitest coverage for backoff / reset behavior.
Reviewed changes
Copilot reviewed 17 out of 18 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
| test/internal/streaming/StreamConnection.spec.ts | Adds regression tests for backoff reset timer behavior. |
| test/internal/streaming/Backoff.spec.ts | Adds unit tests for backoff delay growth, jitter, and reset. |
| test/BKTConfig.spec.ts | Updates config expectations to include new streaming defaults. |
| src/internal/streaming/StreamingTask.ts | New scheduled task to manage SSE streaming + fallback polling recovery. |
| src/internal/streaming/StreamConnection.ts | New connection manager with watchdog, retry backoff, and fallback signaling. |
| src/internal/streaming/httpStatus.ts | New helper to classify recoverable vs non-recoverable HTTP statuses. |
| src/internal/streaming/FetchEventSource.ts | New fetch/ReadableStream-based SSE transport implementation. |
| src/internal/streaming/EventSourceLike.ts | Defines injectable EventSource-like interface/shape for transports. |
| src/internal/streaming/Backoff.ts | New jittered exponential backoff helper. |
| src/internal/scheduler/TaskScheduler.ts | Selects StreamingTask vs EvaluationTask; adds reconnect hook for attribute updates. |
| src/internal/remote/fetch.ts | Extends fetch response typing to optionally include body stream. |
| src/internal/model/request/StreamEvaluationsRequest.ts | Adds request model for streaming POST body. |
| src/internal/evaluation/EvaluationInteractor.ts | Extracts reusable applyEvaluationsResponse() for streaming + polling. |
| src/BKTConfig.ts | Adds streaming configuration knobs and optional eventSource. |
| src/BKTClient.ts | Triggers streaming reconnect on user attribute updates. |
| eslint.config.cjs | Excludes refs/ from ESLint scanning. |
| CLAUDE.md | Adds contributor guidance (commands, style, architecture notes). |
| .gitignore | Ignores /refs and /REPORTS. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Introduce EventSourceErrorLike with an explicit `terminal` flag to distinguish auth failures (401/403) from transient errors, so the watchdog can stop retrying when the API key is known-bad. Add isTerminalStatus() helper alongside the existing isRecoverableStatus() to centralise that classification. Update InternalConfig test fixture with the new enableStreaming and streamingFallbackToPolling fields.
Three correctness issues motivated this rewrite: 1. The old lastOpenAt tracking only started the fallback timer after the first successful open, so a connection that never opened (e.g. a hanging request) could retry indefinitely. unhealthySince now starts on any first failure, whether or not the stream ever opened. 2. A replaced EventSource could still fire callbacks into the new connection. Every handler now guards with `if (this.es !== es) return` so stale instances are fully ignored. 3. reconnect() racing a scheduled backoff retry produced two concurrent streams. openConnection() now unconditionally clears the pending timer and closes any live EventSource before opening a new one. Also threads terminal error info (401/403 or terminal:true) through the onError callback so callers can suppress recovery scheduling. Expands the test suite with a dedicated health-model suite covering all the above scenarios.
Four correctness issues fixed: 1. WHATWG SSE allows \r\n, \n, and \r line endings but the parser only handled \n. normalizeLineEndings() covers all three, and a pending-CR buffer prevents a \r\n split across chunks from being doubled. 2. The no-streaming guard now also checks for getReader(), so React Native environments without a WHATWG fetch polyfill (and injected fetch impls that return a non-WHATWG body like node-fetch) receive a terminal error instead of silently spinning. 3. Header merging switched to Object.assign so caller headers (e.g. Authorization) win over the Content-Type/Accept defaults. Spread after defaults is forbidden by the no-spread-after-defaults lint rule and would re-introduce undefined when callers omit optional keys. 4. reader.cancel() is now guarded: it rejects when the stream already errored, which would otherwise surface as an unhandled rejection.
…ders Terminal errors (auth failures, streaming unsupported) were still scheduling a streaming recovery timer, causing indefinite retries that can never succeed. Recovery is now skipped when info.terminal is true. The full request header profile (Authorization, Content-Type, Accept) is assembled in the request builder so injected EventSourceLike implementations receive a complete request without depending on FetchEventSource's internal defaults. startFallback() now owns the streamingFallbackToPolling guard and scheduleRecovery() is extracted as its own method, making the terminal/non-terminal branching in handleError() explicit.
… stream data Streamed evaluations can arrive after updateUserAttributes() sets the flag but before a fetch() carrying the new attributes has been sent. Clearing the flag inside applyEvaluationsResponse() would cause the next fetch() to skip sending updated attributes entirely. Move clearUserAttributesUpdated() into fetch() so only the path that actually transmitted the updated attributes is allowed to clear the flag. applyEvaluationsResponse() remains flag-agnostic for safe use by the streaming path.
Consumers who inject a custom SSE transport via the eventSource config option need typed contracts to implement against. Without these exports they would have to copy the interface definitions or use `any`. Also adds BKTConfig tests for the enableStreaming + eventSource injection paths.
The SSE event handlers called handleData() without catching the
returned promise. A storage failure (or any other rejection inside
handleData) would surface as an unhandled promise rejection instead
of being silently dropped.
Explicit .catch(() => {}) calls ensure rejections are swallowed at
the boundary where the events map expects void-returning handlers.
…odes Only 401 and 403 were treated as terminal, but several other 4xx codes are equally permanent when the request shape (method, headers, body, URL) never changes between retries: 404, 405, 406, 410, 413, 414, 415, 422, 431, 451. Unlike the polling path (post.ts) which treats 499 as a rollout signal, the streaming path has no such convention, so these codes get no benefit of the doubt.
Status 499 (client closed request) is a deployment-related response that the polling path already retries via ClientClosedRequestException in post.ts. The streaming path treated it as non-recoverable, so a backend rollout that polling survives could still kill the SSE stream. Add 499 to the recoverable set so streaming self-heals the same way polling does.
A named event (`event: foo`) with no registered listener fell back to onmessage, so an unrecognized event name could be misrouted and misinterpreted as an evaluation payload. Native EventSource only falls back to onmessage for genuinely unnamed events, so match that. As defense in depth, handleData() also blind-cast parsed JSON to GetEvaluationsResponse before handing it to storage writes. Add a minimal shape check so any other misrouted or malformed payload is dropped instead of reaching deleteAllAndInsert()/update() with garbage.
A concurrent writer with an older evaluatedAt (e.g. the initial REST fetch racing the stream's first snapshot) could overwrite newer state already applied by another writer. Add a staleness check to update()/deleteAllAndInsert() that skips strictly-older writes and reports no change, so callers don't notify listeners for a write that never happened.
openConnection() runs from setTimeout callbacks (backoff/watchdog/ recovery) as well as from start()/reconnect(). If requestBuilder() or an injected EventSourceLike constructor throws synchronously, the throw was uncaught and would crash a Node process. Route it through the same non-terminal onError path as a runtime error, so the caller falls back to polling and schedules recovery instead of crashing.
Reconnect attempts (from reconnect() while on fallback, and from the 5-minute recovery timer) stopped the polling fallback as soon as they began, before the new connection had actually opened. That left a polling gap for however long the reopen attempt took to succeed or fail. Keep the poller running until onOpen confirms the new stream is live, and make openStream() defensively tear down any existing connection/timer at entry, since callers no longer do that up front.
…rsts updateUserAttributes() calls reconnectStreaming() to pick up fresh attributes on the next connection. Several attributes set in a row (e.g. at login) each triggered their own reconnect(), reopening the stream multiple times in quick succession. Debounce the call by 200ms so a burst collapses into a single reconnect(), and clear the pending timer in stop() so destroy doesn't leave one behind.
Each incoming chunk re-normalized the entire accumulated buffer and re-split it on '\n\n', so parse cost grew quadratically with a long- lived stream's total buffered size instead of the size of each chunk. Normalize only the newly-decoded chunk before appending, and scan for the block separator from a saved offset instead of rescanning bytes already searched, so each chunk is handled in work proportional to its own size.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 31 out of 32 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (2)
src/internal/evaluation/EvaluationStorage.ts:283
- setUserAttributesUpdated() always writes to the underlying storage even when userAttributesUpdated is already true. Since the only required change in that case is bumping updateSequence, this extra storage.set() is redundant and can add unnecessary IO on repeated attribute updates.
async setUserAttributesUpdated(): Promise<void> {
await runWithMutex(this.mutex, async () => {
const entity = this.getCachedEvaluationEntity()
this.updateSequence++
await this.saveAsync({
src/internal/evaluation/EvaluationStorage.ts:307
- clearUserAttributesUpdated() always persists a write even when the flag is already false. Adding a fast-path return avoids redundant storage writes (while preserving the updateSequence staleness guard).
async clearUserAttributesUpdated(state: UserAttributesState): Promise<void> {
await runWithMutex(this.mutex, async () => {
const entity = this.getCachedEvaluationEntity()
if (this.updateSequence !== state.updateSequence) {
// A setUserAttributesUpdated() call landed after state was captured
// — this clear belongs to a now-stale request that didn't carry the
// latest attributes. Leave the flag set for the next request.
return
}
await this.saveAsync({
The caller reads the user object in the same synchronous expression that invokes fetch(), so any attribute update landing after that point is not carried by the request. Reading the attributes state after the first awaited storage read left a window where such an update would be stamped with the in-flight request's sequence, and the success-path clear would then wipe a flag whose attributes were never sent, losing the update until the next unrelated change. Capturing the state as the first statement makes the user snapshot and the attributes snapshot atomic, so a concurrent update is correctly detected as newer and survives for the following request.
The response to a userAttributesUpdated:true request carries the re-evaluation that the flag asked for. Clearing the flag before persisting that response meant a rejected write (browser storage quota, for example) lost the re-evaluation for good: the next poll would send userAttributesUpdated:false and the backend would have no reason to re-evaluate. Persisting first lets a failed write propagate with the flag still set, so the next poll retries. The clear still happens before listeners are notified, so a listener that triggers a nested fetch does not re-send the flag and pull back a redundant forceUpdate snapshot. Splitting the write and the notification apart also removes the need for an options object on applyEvaluationsResponse — the one caller that suppresses notifications now passes its predicate directly.
reconnect() (triggered externally, e.g. by an attribute update) reset unhealthySince to 0. If the stream endpoint was down while the app called updateUserAttributes() more often than the unhealthy give-up window, every reconnect forgave the clock and the polling fallback was starved indefinitely. Only real liveness (markHealthy) should clear the window, so a genuinely down endpoint still falls back on schedule regardless of how often external reconnects fire.
A protobuf-JSON marshaler that drops zero-valued fields sends no forceUpdate key when it's false. The shape check required forceUpdate to be a boolean, so every such patch failed validation and was silently dropped instead of applied. Treat a missing forceUpdate as false, matching the REST path's existing tolerance for the omission.
An explicit `event: message` line was treated as a named event, so it fell into the "unknown name, no listener" silent-drop path instead of onmessage. Per the SSE/EventSource standard, 'message' is the default type whether it comes from no event: line, an empty one, or an explicit `event: message` — all three must reach onmessage the same way a native EventSource would.
Clearing the user-attributes flag runs on every stream (re)connect and every successful poll, but the flag is almost always already false. Writing anyway re-serializes the entire evaluations map to storage just to set false to false. Return early when there is nothing to clear.
Two guards added earlier duplicated protection that was already present. The client's own destroyed flag restated an identity check against the singleton that the surrounding code already performs, and it missed the destroy-then-reinitialize case, where the flag stays set on an instance that is no longer registered anyway. Stopping the stream connection at the top of openStream was likewise unreachable: clearing the pending recovery timer there is what guarantees every caller arrives with no live connection. Add coverage for the one path where that timer clear actually matters — a stream opened by reconnect that fails terminally, so the open handler never runs and nothing else can cancel the timer armed by the original failure.
The resume offset returned alongside the unconsumed remainder was always just derived from that remainder's length, so the wrapper object carried no information the caller could not compute itself. Returning the remainder alone drops the allocation and puts the back-up-one-char rule at the call site, next to where the next scan actually starts. Also add a test pinning that rule down: a blank-line separator split so one newline ends a chunk and the other begins the next must still yield two events, not one merged one.
Both write methods restated the stale-write rule in their own words, one of them by cross-referencing the other. Keeping three copies in sync is a losing game, so defer to the predicate that actually decides it.
The streaming and scheduler suites each carried their own copy of the same base config and DefaultComponent wiring, so any change to the test setup had to be made twice and could silently drift. A single builder in the test utils keeps both suites on identical defaults while still letting each one pass the overrides it needs.
Initialization is really two phases with different failure semantics: loading the evaluation cache is fatal and must un-register the client, while the first fetch may time out and must leave it registered. Both lived in one instance method that also had to clear the singleton it was registered in, so the object was responsible for un-registering itself. Splitting the phases and moving the register/clear bookkeeping and the post-await identity check into the entry point that creates the instance keeps all singleton handling in one place, and lets each phase's comment state its own contract. Behavior and ordering are unchanged — the cache load still resolves before any task starts.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 32 out of 33 changed files in this pull request and generated 1 comment.
Suppressed comments (1)
src/internal/streaming/FetchEventSource.ts:91
- FetchEventSource.connect() calls the injected fetch directly. If the fetch implementation throws synchronously (some implementations throw on invalid input), this will escape the promise chain and can crash instead of surfacing via onerror like other failures.
const doFetch = this.fetchImpl
doFetch(this.url, {
method: this.init.method ?? 'POST',
headers,
body: this.init.body ?? '',
t-kikuc
approved these changes
Aug 4, 2026
t-kikuc
left a comment
t-kikuc
left a comment
There was a problem hiding this comment.
thank you.
these comments are non-blocking.
you can fix them in other PRs.
| // a redundant forceUpdate snapshot. Streamed data must never clear the | ||
| // flag (race) — only this, the polling/fetch path, does. | ||
| const changed = await this.writeEvaluations(result.value) | ||
| await this.evaluationStorage.clearUserAttributesUpdated( |
There was a problem hiding this comment.
non-blocking: The comment above states "a rejected write must skip the clear — the flag survives and the next poll retries", but the clear below runs unconditionally: changed is false both for a stale-skipped write and for a no-change response, so the boolean can't express that invariant (and gating the clear on changed would wrongly keep the flag set on legit no-change responses).
Practical impact today looks negligible — in streaming mode the debounced reconnect re-evaluates with the latest attributes anyway, and the polling-only path has effectively a single writer — but a future change relying on the documented invariant could turn this into a real bug.
Suggestion: either reword the comment to match the behavior (the clear also runs for stale-skipped writes; the streaming reconnect compensates), or have the storage writes return applied/unchanged/stale and skip the clear only on stale.
| // already retries for the polling API (ClientClosedRequestException) — a | ||
| // backend rollout that polling survives must not kill the stream instead. | ||
| return ( | ||
| status === 400 || status === 408 || status === 429 || status === 499 |
There was a problem hiding this comment.
non-blocking question: Is 400 in the recoverable set intentional? 422 is classified terminal because the same body fails validation every time, and 400 is the generic form of the same condition; the polling path (post.ts) also retries only 499. As is, an endpoint that persistently returns 400 cycles forever: ~120s of backoff retries → polling fallback → a recovery attempt every 5 minutes, each burning another retry window.
If this is meant to cover gateways that can return transient 400s, a short note next to the 499 rationale would help; otherwise moving 400 to TERMINAL_STATUSES seems more consistent with the list's own reasoning.
|
It worked well in my local Minikube. |
duyhungtnn
marked this pull request as draft
August 5, 2026 03:25
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Overview
flowchart TD Init["BKTClientImpl.initializeInternal()"] Init --> InitCache["await evaluationInteractor().initialize()<br/>load the evaluation cache FIRST — StreamingTask reads it<br/>synchronously on its first connect and throws if it isn't loaded"] InitCache --> Sched["scheduleTasks() → new TaskScheduler().start()"] Sched --> Choice{"config.enableStreaming?"} Sched --> EventTask["EventTask<br/>flushes queued analytics events"] Choice -->|false| Poll["EvaluationTask<br/>polls for evaluations on an interval"] Choice -->|true| Stream["StreamingTask<br/>SSE processor"] Stream --> Open["openStream()"] Open --> Transport{"config.eventSource provided?"} Transport -->|yes| Injected["user-injected EventSourceLike"] Transport -->|"no (default)"| Fetch["FetchEventSource(config.fetch)<br/>built-in POST + ReadableStream transport"] Open --> Conn["StreamConnection<br/>transport supervisor — owns one connection at a time"] Injected --> Conn Fetch --> Conn Conn --> Health["Connection health:<br/>70s idle watchdog — any received byte resets it<br/>120s unhealthy → give up, tell the processor<br/>60s open & stable → reset backoff<br/>reconnect backoff 1s→30s with jitter"] Conn -->|"put / patch / unnamed data event"| HandleData["handleData() → EvaluationInteractor.applyEvaluationsResponse()<br/>updates the cache and notifies listeners"] Conn -->|"named 'error' event"| LogErr["console.error — logged distinctly, NOT applied to the cache"] Conn -->|"connection opened"| StopFb["stopFallback()<br/>streaming is healthy again — stop polling"] Conn -->|"gave up: onError(info)"| HandleErr["handleError(info)"] HandleErr --> StartFb["startFallback()"] StartFb -->|"streamingFallbackToPolling = true"| Fallback["EvaluationTask.start(true)<br/>fetch immediately, then keep polling"] HandleErr --> Terminal{"info.terminal?"} Terminal -->|"false (transient / outage)"| Recovery["scheduleRecovery()<br/>reopen the stream in 5 min"] Terminal -->|"true (bad API key / can't stream)"| NoRecovery["no recovery — retrying can never succeed"] Recovery --> Open Poll --> FetchEval["fetchEvaluations()"] Fallback --> FetchEval FetchEval --> Guard{"still running after the await?"} Guard -->|"no — stop() ran mid-fetch"| Drop["return without rescheduling<br/>(prevents an unstoppable background poller)"] Guard -->|yes| Reschedule["reschedule(pollingInterval)"] Reschedule -.next poll.-> FetchEval1. What runs at startup?
flowchart TD A["initializeBKTClient()"] --> B["initialize()<br/>load the evaluation cache first"] B --> C["scheduleTasks()"] C --> D{"enableStreaming?"} D -->|true| E["StreamingTask (SSE)"] D -->|false| F["EvaluationTask (polling)"] C --> G["EventTask (always runs)"]2. How does a streamed update reach the cache?
flowchart TD A["StreamingTask.openStream()"] --> B["StreamConnection"] B --> C{"transport?"} C -->|default| D["FetchEventSource(config.fetch)"] C -->|config.eventSource| E["injected EventSourceLike"] B -->|"put / patch / unnamed data"| F["handleData()"] F --> G["applyEvaluationsResponse()<br/>update cache + notify listeners"] B -->|"named 'error' event"| H["log only — not applied"]3. When is the connection considered healthy or dead?
stateDiagram-v2 [*] --> Connecting Connecting --> Open: onopen Open --> Open: byte received (resets 70s watchdog) Open --> Unhealthy: drop, or 70s with no bytes Unhealthy --> Connecting: backoff retry (1s to 30s) Unhealthy --> GaveUp: unhealthy over 120s GaveUp --> [*]: onError to processor4. What happens when the stream breaks?
flowchart TD A["StreamConnection gives up<br/>onError(info)"] --> B["handleError()"] B --> C["startFallback()<br/>poll while streaming is down"] B --> D{"info.terminal?"} D -->|"false (transient / outage)"| E["scheduleRecovery()<br/>reopen the stream in 5 min"] D -->|"true (bad key / can't stream)"| F["no recovery"] E -. 5 min .-> G["openStream() again"] G -->|"reconnects (onOpen)"| H["stopFallback()<br/>back to streaming"]5. How does the polling task avoid overlapping fetches and stray timers?
flowchart TD A["start(immediate)"] --> B{"immediate?"} B -->|false| C["arm the next-poll timer"] B -->|true| D["fetchEvaluations() now"] C -. timer fires .-> D D --> E{"still running<br/>after the await?"} E -->|"no — stop() ran mid-fetch"| F["return, do NOT reschedule<br/>(no orphaned poller)"] E -->|yes| G["reschedule(pollingInterval)"] G -. next poll .-> D