feat: introduce SSE (ref AI-984) by leakonvalinka · Pull Request #28 · Dynatrace/openfeature-provider-python

leakonvalinka · 2026-10-22T09:31:11Z

WIP

What

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

References

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

toddbaert · 2026-10-27T20:16:29Z

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.

Copilot

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

toddbaert · 2026-11-05T18:49:58Z

+            self._emit_failure(type(exc).__name__)
+        return result


Agreed; the drop check should compare against the last applied Last-Modified, not the completed fetch's. That's what .NET does.

I've been trying to wrap my head around the issue here and I think I get it now - can this actually happen in reality? That the CDN returns an error code but also a Last Modified header? According to Claude this "typically shouldn't happen" which is not a definite "no", but if I think about it it doesn't make sense. I can definitely add the fix, this is more so out of curiosity.

Yes it would be nonsense, but I think it's nonsense worth defending against.

toddbaert · 2026-11-05T18:54:08Z

+                self._sse_url,
+                pool=self._pool,
+                urllib3_request_options={
+                    "timeout": Urllib3Timeout(read=SSE_READ_TIMEOUT_S)


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

toddbaert · 2026-11-05T18:49:55Z

+        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


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.

toddbaert · 2026-11-05T18:49:53Z

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


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:

Java

Node

.NET

See providers.md 2.1 and the 4.1 table.

toddbaert · 2026-11-05T18:54:10Z

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


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

leakonvalinka · 2026-11-05T11:01:43Z

+        """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."""


see https://github.com/launchdarkly/python-eventsource/blob/main/ld_eventsource/http.py#L119 (and PR)

Since the reader unblocks at the 150s idle timeout at the latest (sooner if the server sends anything), the Windows leak is bounded: a stale subscriber for up to ~150s per shutdown, on Windows only. IMO that's not worth reaching into urllib3 internals, which might break later. WDYT about dropping shutdown_stream()/_socket_of() (and _StreamPool with them, a plain PoolManager would do) and accepting what every ld_eventsource user already deals with on Windows? The WORST thing that could happen here would bem150s "bounded" leaks if somebody were to restart a provider over and over.

If this causes an issue testing on windows, I would see if we can disable that specific test there somehow, maybe with a new tag... but I think we can do that on a different PR.

toddbaert · 2026-11-05T18:49:43Z

@@ -1 +1 @@
-Subproject commit 5c7697375cfa64ddaec9d3297f435276ab07f5e3
+Subproject commit ec236c732f0fc8bb87cd56b077e04da51a28ed02


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.

toddbaert · 2026-11-05T18:49:44Z

+            self._start_sse_locked(body)
            self._start_poller_locked()


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)

oh, seems like i compared this to an older version of java then, because i think it was the other way around? anyway, will do

Ya it was a recent fix, you probably had the one just before.

toddbaert · 2026-11-05T18:49:44Z

+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


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

toddbaert · 2026-11-05T18:49:44Z

+            self._rescheduled = True
+            self._cond.notify_all()
+
+    def trigger_immediate_fetch(


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(

toddbaert · 2026-11-05T18:49:44Z

+        # 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


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

toddbaert · 2026-11-05T18:49:44Z

+    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


Suggested change

issued_for = self._in_flight_sse_fetch_last_modified

in_flight = self._in_flight_last_modified

I get that issued_for is confusing, but does in_flight fit? If I'm not mistaken this runs after the poll has happened - what do you think about previous/previous_in_flight?

Fair; the field's fine since it's set while the fetch runs, but for the local previous_in_flight works for me... I was just trying to make most of the big funcs/vars named consistently, maybe I went overboard in a couple places.

toddbaert · 2026-11-05T18:49:44Z

        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


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

toddbaert · 2026-11-05T18:49:44Z

+    # 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."""


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

toddbaert · 2026-11-05T18:49:44Z

+            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."""


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

toddbaert · 2026-11-05T18:49:44Z

        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,


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?

Sorry, I don't really understand what you mean here regarding the fetcher 😅 But will remove it, the _poll_interval_s as well? That has been there before already.

Sorry, ignore the fetcher bit. Just dropping sse_connected_poll_interval_s is enough; poll_interval_s predates this PR, so let's leave it (maybe a follow-up).

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

feat: introduce SSE (ref AI-984)

556ca49

leakonvalinka force-pushed the feat/sse branch from 92c7d14 to 556ca49 Compare October 22, 2026 09:34

small fixups

94ce949

leakonvalinka force-pushed the feat/sse branch from 0d65c68 to a41859d Compare October 27, 2026 11:24

remove duplicate method def

a41859d

toddbaert reviewed Oct 28, 2026

View reviewed changes

Comment thread src/dynatrace/openfeature/_sse.py Resolved

small fixes

aab31af

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 started reviewing on behalf of leakonvalinka November 5, 2026 10:50

View session

Copilot AI reviewed Nov 5, 2026

View reviewed changes

leakonvalinka commented Nov 5, 2026

View reviewed changes

toddbaert reviewed Nov 5, 2026

View reviewed changes

		@@ -1 +1 @@
		Subproject commit 5c7697375cfa64ddaec9d3297f435276ab07f5e3
		Subproject commit ec236c732f0fc8bb87cd56b077e04da51a28ed02

	SSE_READ_TIMEOUT_S: float = 150
	SSE_IDLE_TIMEOUT_S: float = 150

	self._pending_sse_fetch_last_modified: float \| None = None
	self._pending_last_modified: float \| None = None

	issued_for = self._in_flight_sse_fetch_last_modified
	in_flight = self._in_flight_last_modified

	# 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

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

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

Conversation

leakonvalinka commented Oct 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

References

Uh oh!

toddbaert commented Oct 27, 2026

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leakonvalinka Nov 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

toddbaert Nov 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

toddbaert Nov 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

toddbaert Nov 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leakonvalinka Nov 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

leakonvalinka commented Oct 22, 2026 •

edited

Loading

leakonvalinka Nov 10, 2026 •

edited

Loading

toddbaert Nov 11, 2026 •

edited

Loading

toddbaert Nov 11, 2026 •

edited

Loading

toddbaert Nov 5, 2026 •

edited

Loading

leakonvalinka Nov 10, 2026 •

edited

Loading