Skip to content

feat: introduce SSE (ref AI-984) - #28

Open
leakonvalinka wants to merge 4 commits into
mainfrom
feat/sse
Open

leakonvalinka wants to merge 4 commits into
mainfrom
feat/sse

Conversation

@leakonvalinka

@leakonvalinka leakonvalinka commented Oct 22, 2026 •

Copy link
Copy Markdown

WIP

What

  • introduces SSE functionality to enable real-time notifications for flag config changes

References

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

@leakonvalinka leakonvalinka changed the title feat: introduce SSE feat: introduce SSE (ref AI-984) Oct 22, 2026
@toddbaert

Copy link
Copy Markdown
Collaborator

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.

Comment thread src/dynatrace/openfeature/_sse.py Resolved
@leakonvalinka
leakonvalinka marked this pull request as ready for review November 5, 2026 10:49
Copilot AI balanced review requested due to automatic review settings November 5, 2026 10:49

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 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
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

Comment on lines +220 to +221
self._emit_failure(type(exc).__name__)
return result

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed; something like Urllib3Timeout(connect=REQUEST_TIMEOUT_S, read=SSE_IDLE_TIMEOUT_S) matches the fetch timeout and keeps reconnects from hanging.

Comment on lines +208 to +214
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

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. Node (here) and .NET (here) re-read eventStreams on every applied config: reconnect if the URL changed, close if it's gone. Java has the same gap as this PR, and the spec doesn't cover rotation yet either (we never opened that PR). IMO fine as a follow-up, but we should line up all 4.

poller = self._poller
if poller is None:
return
poller.set_interval(self._sse_connected_poll_interval_s)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread tests/sse_server.py
Comment on lines +39 to +53
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)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed; registering before end_headers() (and moving the header writes inside the try) prevents this from happening. Cheap fix for a likely CI flake.

Comment on lines +49 to +53
"""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."""

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Comment thread fm-provider-test-harness
@@ -1 +1 @@
Subproject commit 5c7697375cfa64ddaec9d3297f435276ab07f5e3
Subproject commit ec236c732f0fc8bb87cd56b077e04da51a28ed02

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment on lines +154 to 155
self._start_sse_locked(body)
self._start_poller_locked()

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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:

Suggested change
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

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The spec calls this the "SSE idle timeout", and Java (IDLE_TIMEOUT_SECONDS) and .NET (IdleTimeout) match (plus the usage at L105):

Suggested change
SSE_READ_TIMEOUT_S: float = 150
SSE_IDLE_TIMEOUT_S: float = 150

self._rescheduled = True
self._cond.notify_all()

def trigger_immediate_fetch(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Java, Node and .NET all call this requestOutOfBandFetch; "immediate" is also not quite right when it's deferred (plus usages):

Suggested change
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

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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):

Suggested change
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

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
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

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: the spec doesn't have SSE vs polling "modes"; SSE only picks the poll cadence (and "polling-only" means SSE disabled or absent):

Suggested change
# 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."""

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
"""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."""

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
"""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,

@toddbaert toddbaert Nov 5, 2026 •

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants