Repository navigation
feat: introduce SSE (ref AI-984) - #28
leakonvalinka wants to merge 4 commits into
Conversation
92c7d14 to
556ca49
Compare
0d65c68 to
a41859d
Compare
|
Heads up: harness v1.4.0 adds a scenario for polling at 10s after a failed fetch, even with SSE connected. Only a successful fetch while SSE is connected relaxes to 10m; connect alone no longer does. Spec: https://github.com/Dynatrace-Internal/feature-management-knowledge-base/pull/177; Java: Dynatrace/openfeature-provider-java#44; Node: Dynatrace/openfeature-provider-js#46. |
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved SSE and polling defects can block reconnection indefinitely, discard needed refetches, miss stream URL changes, and delay recovery after fetch failures.
5 open issues
Do not discard queued refetches after failed result application · New Set a finite connection timeout for SSE requests · New Recreate the SSE manager when the configured URL changes · New Use aggressive polling after fetch failures while SSE is connected · New Register the subscriber before sending response headers · New
What changed in this PR
Adds server-sent events (SSE) support to the Dynatrace OpenFeature provider so flag-change notifications trigger configuration refreshes, with polling as a fallback.
Changes:
- Adds stream discovery, message parsing, reconnection, and shutdown handling.
- Integrates SSE-triggered refreshes with single-flight polling and configurable polling intervals.
- Adds unit and end-to-end coverage, dependencies, and user documentation.
| File | Description |
|---|---|
uv.lock |
Locks the new SSE client dependency. |
tests/test_sse.py |
Tests stream lifecycle, callbacks, and logging. |
tests/test_sse_message.py |
Tests message parsing and invalid payload handling. |
tests/test_provider_sse.py |
Tests provider-level SSE integration and lifecycle. |
tests/test_polling.py |
Tests interval changes and deferred refresh coalescing. |
tests/test_cdn_config.py |
Tests advertised SSE URL extraction. |
tests/sse_server.py |
Adds a local SSE server for socket-based tests. |
tests/e2e/test_e2e.py |
Adds SSE acceptance steps and event assertions. |
tests/e2e/mock_server.py |
Adds SSE controls to the mock-server client. |
tests/e2e/conftest.py |
Enables previously skipped single-flight scenarios. |
tests/conftest.py |
Provides the local SSE server fixture. |
src/dynatrace/openfeature/provider.py |
Connects SSE notifications to provider refresh and polling. |
src/dynatrace/openfeature/_sse.py |
Implements streaming, reconnection, and shutdown. |
src/dynatrace/openfeature/_sse_message.py |
Parses flat and wrapped SSE messages. |
src/dynatrace/openfeature/_polling.py |
Supports dynamic cadence and coalesced immediate refreshes. |
src/dynatrace/openfeature/_cdn.py |
Exposes Last-Modified timestamps for refresh deduplication. |
src/dynatrace/openfeature/_cdn_config.py |
Extracts SSE endpoints from configuration. |
README.md |
Documents SSE behavior and the disable option. |
pyproject.toml |
Declares the SSE dependency and typing override. |
.gitignore |
Excludes IDE and macOS artifacts. |
Review effort: Balanced
| self._emit_failure(type(exc).__name__) | ||
| return result |
There was a problem hiding this comment.
Agreed; the drop check should compare against the last applied Last-Modified, not the completed fetch's. That's what .NET does.
| self._sse_url, | ||
| pool=self._pool, | ||
| urllib3_request_options={ | ||
| "timeout": Urllib3Timeout(read=SSE_READ_TIMEOUT_S) |
There was a problem hiding this comment.
Agreed; something like Urllib3Timeout(connect=REQUEST_TIMEOUT_S, read=SSE_IDLE_TIMEOUT_S) matches the fetch timeout and keeps reconnects from hanging.
| if not self._sse_enabled or self._sse_manager is not None: | ||
| return | ||
| cdn = urlsplit(self._options.cdn_url) | ||
| sse_url = extract_sse_url(body, f"{cdn.scheme}://{cdn.netloc}") | ||
| if sse_url is None: | ||
| # the server always sends eventStreams, so this is not worth logging | ||
| return |
There was a problem hiding this comment.
| poller = self._poller | ||
| if poller is None: | ||
| return | ||
| poller.set_interval(self._sse_connected_poll_interval_s) |
There was a problem hiding this comment.
Agreed; we should fix both:
- connecting shouldn't relax the cadence by itself; it just issues the catch-up fetch, and that fetch's success is what moves us to 10m
- a failed fetch should drop back to 10s, even while SSE is connected
So the cadence moves into _on_fetch_success / _on_fetch_failure, like the other impls:
See providers.md 2.1 and the 4.1 table.
| self.send_response(200) | ||
| self.send_header("Content-Type", "text/event-stream") | ||
| self.send_header("Cache-Control", "no-cache") | ||
| self.send_header("Transfer-Encoding", "chunked") | ||
| self.send_header("Connection", "close") | ||
| self.end_headers() | ||
| self.wfile.flush() | ||
| events: queue.Queue[bytes | None] = queue.Queue() | ||
| owner.register(events) | ||
| try: | ||
| self._stream(events) | ||
| except OSError: | ||
| pass # client went away mid-write | ||
| finally: | ||
| owner.unregister(events) |
There was a problem hiding this comment.
Agreed; registering before end_headers() (and moving the header writes inside the try) prevents this from happening. Cheap fix for a likely CI flake.
| """Shut the stream's socket down so the server sees it close, even on Windows. | ||
| ld_eventsource's own close() only releases the connection there, because shutdown() does not | ||
| wake the reader thread blocked in recv(), and closing under it would deadlock on its buffer lock. | ||
| So the socket stays open until GC and the server keeps a stale subscriber; shutting it down here | ||
| sends the FIN. The reader may stay blocked, but the stop flag keeps it inert: it makes no callbacks.""" |
There was a problem hiding this comment.
| @@ -1 +1 @@ | |||
| Subproject commit 5c7697375cfa64ddaec9d3297f435276ab07f5e3 | |||
| Subproject commit ec236c732f0fc8bb87cd56b077e04da51a28ed02 | |||
There was a problem hiding this comment.
Can we bump this to v1.4.0? It adds "A failed fetch while SSE is connected switches to the aggressive poll cadence until a fetch succeeds", which covers the cadence fix above. It should need no new bindings.
| self._start_sse_locked(body) | ||
| self._start_poller_locked() |
There was a problem hiding this comment.
Spec 3.1 says SSE opens after polling starts. The lock makes this safe here, but we just swapped this order in .NET and Java to match:
| self._start_sse_locked(body) | |
| self._start_poller_locked() | |
| self._start_poller_locked() | |
| self._start_sse_locked(body) |
| log = logging.getLogger(__name__) | ||
|
|
||
| # the stream is long-lived and idle between notifications, so only a dead connection should trip this | ||
| SSE_READ_TIMEOUT_S: float = 150 |
There was a problem hiding this comment.
The spec calls this the "SSE idle timeout", and Java (IDLE_TIMEOUT_SECONDS) and .NET (IdleTimeout) match (plus the usage at L105):
| SSE_READ_TIMEOUT_S: float = 150 | |
| SSE_IDLE_TIMEOUT_S: float = 150 |
| self._rescheduled = True | ||
| self._cond.notify_all() | ||
|
|
||
| def trigger_immediate_fetch( |
There was a problem hiding this comment.
Java, Node and .NET all call this requestOutOfBandFetch; "immediate" is also not quite right when it's deferred (plus usages):
| def trigger_immediate_fetch( | |
| def request_out_of_band_fetch( |
| # set from submit until the fetch has settled (not until its thread exits), so no trigger defers behind a settled fetch | ||
| self._in_flight = False | ||
| # last-modified value of a deferred SSE re-fetch, or None when none is pending | ||
| self._pending_sse_fetch_last_modified: float | None = None |
There was a problem hiding this comment.
Nit: the same value goes by pending_sse_fetch_last_modified, in_flight_sse_fetch_last_modified, sse_refetch_at and issued_for. One name pair would be easier to follow (plus usages):
| self._pending_sse_fetch_last_modified: float | None = None | |
| self._pending_last_modified: float | None = None |
| def _on_fetch_complete_locked(self, completed: FetchResult | None) -> bool: | ||
| """Settle the SSE re-fetch bookkeeping; return True to issue the pending re-fetch now. | ||
| ``completed`` is the finished fetch's result, or None if it raised.""" | ||
| issued_for = self._in_flight_sse_fetch_last_modified |
There was a problem hiding this comment.
| issued_for = self._in_flight_sse_fetch_last_modified | |
| in_flight = self._in_flight_last_modified |
| self._grace_period_s = grace_period_s | ||
| self._poller: PollingManager | None = None | ||
| self._sse_manager: SseManager | None = None | ||
| # true while the SSE stream is open, i.e. the provider is in SSE mode rather than polling mode |
There was a problem hiding this comment.
Nit: the spec doesn't have SSE vs polling "modes"; SSE only picks the poll cadence (and "polling-only" means SSE disabled or absent):
| # true while the SSE stream is open, i.e. the provider is in SSE mode rather than polling mode | |
| # true while the SSE stream is open; only selects the poll cadence |
| # SSE only prompts fetches, so its state picks the poll cadence and never touches READY/STALE/ERROR. | ||
|
|
||
| def _on_sse_connected(self, source: SseManager | None) -> None: | ||
| """Switch to SSE mode: safety-net cadence plus a catch-up fetch for anything missed while down.""" |
There was a problem hiding this comment.
| """Switch to SSE mode: safety-net cadence plus a catch-up fetch for anything missed while down.""" | |
| """Issue a catch-up fetch for anything missed while disconnected; its success relaxes the cadence.""" |
| poller.trigger_immediate_fetch(None) | ||
|
|
||
| def _on_sse_disconnected(self, source: SseManager | None) -> None: | ||
| """Fall back to polling mode; fetch failures from here on are handled like any poll failure.""" |
There was a problem hiding this comment.
| """Fall back to polling mode; fetch failures from here on are handled like any poll failure.""" | |
| """Switch to the 10s cadence; fetch failures from here on are handled like any poll failure.""" |
| fetcher: CdnFetcher | None = None, | ||
| metrics_flusher: EvaluationMetricsFlusher | None = None, | ||
| poll_interval_s: float = _POLL_INTERVAL_S, | ||
| sse_connected_poll_interval_s: float = _SSE_CONNECTED_POLL_INTERVAL_S, |
There was a problem hiding this comment.
Spec 2.1 says these intervals aren't user-configurable. If this is just a test knob, can we make that clear like fetcher, or drop it?

WIP
What
References
https://dt-rnd.atlassian.net/browse/AI-984