From 70ce7b22f34f507c27920a9ccab324dcab56a5a5 Mon Sep 17 00:00:00 2001 From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:32:42 +0000 Subject: [PATCH 1/9] deps: update dependency flagsmith/sdk-spec to v0.9.0 --- .github/workflows/sdk-spec.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/sdk-spec.yml b/.github/workflows/sdk-spec.yml index ca9c03d..ec0b7c2 100644 --- a/.github/workflows/sdk-spec.yml +++ b/.github/workflows/sdk-spec.yml @@ -11,7 +11,7 @@ permissions: jobs: sdk-spec: # Keep the ref in `uses:` and `spec-ref` in sync. - uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.7.0 + uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.9.0 with: - spec-ref: v0.7.0 + spec-ref: v0.9.0 language: python From 53350cb661be361e1280fb086bfabb06c4e0d5b0 Mon Sep 17 00:00:00 2001 From: "flagsmith-engineering[bot]" <216534828+flagsmith-engineering[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:32:49 +0000 Subject: [PATCH 2/9] ci: Run the SDK spec for this SDK's capabilities Its entry in Flagsmith/sdk-spec's compatibility matrix at v0.9.0: `core events local-evaluation`. --- .github/workflows/sdk-spec.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/sdk-spec.yml b/.github/workflows/sdk-spec.yml index ec0b7c2..5b5b483 100644 --- a/.github/workflows/sdk-spec.yml +++ b/.github/workflows/sdk-spec.yml @@ -14,4 +14,5 @@ jobs: uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.9.0 with: spec-ref: v0.9.0 + tags: core events local-evaluation language: python From 739ead361003aaf9af7d9c28c10cb2ae175eb2e6 Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:22:30 +0000 Subject: [PATCH 3/9] feat: Retry event delivery and flush on close Event delivery now survives transient failures of the events API, as SDK spec v0.8.0 requires: - A batch is sent up to 3 times with a 3 s request timeout, backing off from 1 s, doubling, with full jitter, one batch in flight at a time. - A batch whose last attempt fails transiently is kept for the next flush, ahead of newer events, unless that attempt got a 500. - A batch the events API rejects is dropped, and a 401/403 stops event tracking. - A full buffer drops its oldest event. `Flagsmith.flush_events()` sends the tracked events and waits for the outcome, and `Flagsmith.close(timeout_seconds)` stops background work and sends the buffered events within the timeout. The spec steps file wires the new steps, and a client with both local evaluation and events. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit d019e76c09c189ddad76139e07c12d0259d42864) --- flagsmith/analytics.py | 215 ++++++++++++++++---- flagsmith/flagsmith.py | 37 +++- sdk-spec/steps.py | 31 +++ tests/test_event_processor.py | 358 +++++++++++++++++++++++++++++----- tests/test_flagsmith.py | 63 ++++++ 5 files changed, 602 insertions(+), 102 deletions(-) diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index f5af134..44b7368 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -1,11 +1,14 @@ import atexit +import enum import json import logging +import random import threading import typing from dataclasses import dataclass from datetime import datetime +import requests from requests_futures.sessions import FuturesSession # type: ignore from flagsmith.version import DEFAULT_USER_AGENT, __version__ @@ -21,6 +24,12 @@ DEFAULT_EVENT_API_URL: typing.Final[str] = "https://events.api.flagsmith.com/" +EVENTS_REQUEST_TIMEOUT_SECONDS: typing.Final[float] = 3 +EVENTS_MAX_ATTEMPTS: typing.Final[int] = 3 +EVENTS_RETRY_BASE_SECONDS: typing.Final[float] = 1 +DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS: typing.Final[float] = 5 +RETRIED_STATUS_CODES: typing.Final[typing.FrozenSet[int]] = frozenset({408, 429}) + session = FuturesSession(max_workers=4) @@ -96,11 +105,25 @@ class EventProcessorConfig: flush_interval_seconds: float = 10.0 +class _DeliveryOutcome(enum.Enum): + DELIVERED = "delivered" + DROPPED = "dropped" + KEPT = "kept" + REFUSED = "refused" + + class EventProcessor: """ Buffered event processor that batches custom events and POSTs them to the Flagsmith event endpoint. Flushes on a background timer or when the buffer fills. + + A batch is sent up to ``EVENTS_MAX_ATTEMPTS`` times, backing off with full + jitter between attempts, one batch at a time. A batch whose last attempt + fails transiently is kept for the next flush, unless that attempt got a + 500. A batch the events API rejects is dropped, and an events API refusing + the environment key stops event tracking. When the buffer is full, the + oldest event is dropped. """ def __init__( @@ -116,9 +139,17 @@ def __init__( self._max_buffer = config.max_buffer_items self._flush_interval_seconds = config.flush_interval_seconds + self._session = requests.Session() self._buffer: typing.List[typing.Dict[str, typing.Any]] = [] self._buffered_exposure_keys: typing.Set[ExposureKey] = set() self._lock = threading.Lock() + # Held while a batch is in flight, so that one batch is sent at a time. + self._send_lock = threading.Lock() + self._tracked_count = 0 + self._attempted_count = 0 + self._full_buffer_flush_pending = False + self._refused = False + self._closed = threading.Event() self._timer: typing.Optional[threading.Timer] = None def track_event( @@ -164,8 +195,14 @@ def _buffer_event( traits: typing.Optional[typing.Dict[str, typing.Any]], metadata: typing.Optional[typing.Dict[str, typing.Any]], ) -> None: - should_flush = False with self._lock: + if self._refused: + logger.debug( + "Not tracking %s event: the events API refused the" + " environment key.", + event, + ) + return if event == FLAG_EXPOSURE_EVENT: # An exposure is defined by who saw which variant of which # feature; equal exposures within one flush interval add no @@ -184,6 +221,12 @@ def _buffer_event( ) return self._buffered_exposure_keys.add(exposure_key) + if len(self._buffer) >= self._max_buffer: + dropped = self._buffer.pop(0) + logger.warning( + "Event buffer is full, dropping the oldest %s event", + dropped["event"], + ) self._buffer.append( { "event": event, @@ -195,62 +238,150 @@ def _buffer_event( "timestamp": int(datetime.now().timestamp() * 1000), } ) - if len(self._buffer) >= self._max_buffer: - should_flush = True + self._tracked_count += 1 + if len(self._buffer) < self._max_buffer or ( + self._full_buffer_flush_pending + ): + return + self._full_buffer_flush_pending = True + tracked_count = self._tracked_count - if should_flush: - self.flush() + threading.Thread( + target=self._flush_full_buffer, args=(tracked_count,), daemon=True + ).start() def flush(self) -> None: + """ + Sends the events tracked so far, and returns once each of them has + been delivered, dropped, or kept for the next flush. Events already + sent while this was waiting for another batch in flight are not sent + again. + """ with self._lock: - if not self._buffer: - return - events = self._buffer - self._buffer = [] - self._buffered_exposure_keys.clear() + tracked_count = self._tracked_count + self._send(tracked_count) - payload = json.dumps({"events": events}) - try: - future = session.post( - self._batch_endpoint, - data=payload, - timeout=3, - headers={ - "Content-Type": "application/json; charset=utf-8", - "X-Environment-Key": self._environment_key, - "Flagsmith-SDK-User-Agent": DEFAULT_USER_AGENT, - }, - ) - except RuntimeError: - logger.debug("Skipping flush: thread pool already shut down") - return - future.add_done_callback(lambda f: self._handle_flush_result(f, events)) + def _flush_full_buffer(self, tracked_count: int) -> None: + with self._lock: + self._full_buffer_flush_pending = False + self._send(tracked_count) - def _handle_flush_result( - self, - future: typing.Any, - events: typing.List[typing.Dict[str, typing.Any]], - ) -> None: - try: - response = future.result() - response.raise_for_status() - except Exception: + def _send(self, tracked_count: typing.Optional[int] = None) -> None: + """ + Sends the buffered events as one batch. Given `tracked_count`, does + nothing if every event tracked up to it has been sent already. + """ + with self._send_lock: + with self._lock: + if not self._buffer or ( + tracked_count is not None and self._attempted_count >= tracked_count + ): + return + events = self._buffer + self._buffer = [] + self._buffered_exposure_keys.clear() + attempted_count = self._tracked_count + + outcome = self._deliver(events) + + with self._lock: + self._attempted_count = attempted_count + if outcome is _DeliveryOutcome.KEPT: + max_buffer = self._max_buffer + self._buffer = (events + self._buffer)[-max_buffer:] + elif outcome is _DeliveryOutcome.REFUSED: + self._refused = True + self._buffer = [] + self._buffered_exposure_keys.clear() + + def _deliver( + self, events: typing.List[typing.Dict[str, typing.Any]] + ) -> _DeliveryOutcome: + payload = json.dumps({"events": events}) + for attempt in range(1, EVENTS_MAX_ATTEMPTS + 1): + if attempt > 1: + backoff = EVENTS_RETRY_BASE_SECONDS * 2 ** (attempt - 2) + if self._closed.wait(random.uniform(0, backoff)): + logger.warning( + "Not retrying %d events: the client is closed", len(events) + ) + return _DeliveryOutcome.KEPT + try: + response = self._session.post( + self._batch_endpoint, + data=payload, + timeout=EVENTS_REQUEST_TIMEOUT_SECONDS, + headers={ + "Content-Type": "application/json; charset=utf-8", + "X-Environment-Key": self._environment_key, + "Flagsmith-SDK-User-Agent": DEFAULT_USER_AGENT, + }, + ) + except requests.RequestException as e: + logger.warning( + "Failed to send %d events, attempt %d of %d: %s", + len(events), + attempt, + EVENTS_MAX_ATTEMPTS, + e, + ) + status = None + else: + status = response.status_code + if response.ok: + return _DeliveryOutcome.DELIVERED + if status in (401, 403): + logger.error( + "Events API refused the environment key with %d," + " stopping event tracking", + status, + ) + return _DeliveryOutcome.REFUSED + if status not in RETRIED_STATUS_CODES and status < 500: + logger.warning( + "Events API rejected %d events with %d, dropping them", + len(events), + status, + ) + return _DeliveryOutcome.DROPPED + logger.warning( + "Failed to send %d events, attempt %d of %d: status %d", + len(events), + attempt, + EVENTS_MAX_ATTEMPTS, + status, + ) + if status == 500: logger.warning( - "Failed to flush pipeline analytics, re-queuing events", exc_info=True + "Dropping %d events the events API failed with 500", len(events) ) - with self._lock: - self._buffer = events + self._buffer - self._buffer = self._buffer[: self._max_buffer] + return _DeliveryOutcome.DROPPED + logger.warning("Keeping %d events for the next flush", len(events)) + return _DeliveryOutcome.KEPT def start(self) -> None: self._schedule_flush() atexit.register(self.stop) - def stop(self) -> None: + def stop( + self, timeout_seconds: float = DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS + ) -> None: + """ + Stops flushing on a timer, and sends the buffered events, waiting for + at most `timeout_seconds`. Events not sent by then are not retried. + """ atexit.unregister(self.stop) if self._timer is not None: self._timer.cancel() - self.flush() + final_flush = threading.Thread(target=self._send, daemon=True) + final_flush.start() + final_flush.join(timeout_seconds) + self._closed.set() + if final_flush.is_alive(): + logger.warning( + "Gave up sending events after %s seconds while closing", + timeout_seconds, + ) def _schedule_flush(self) -> None: self._timer = threading.Timer(self._flush_interval_seconds, self._timer_flush) @@ -258,5 +389,5 @@ def _schedule_flush(self) -> None: self._timer.start() def _timer_flush(self) -> None: - self.flush() + self._send() self._schedule_flush() diff --git a/flagsmith/flagsmith.py b/flagsmith/flagsmith.py index f50ceec..fd80e57 100644 --- a/flagsmith/flagsmith.py +++ b/flagsmith/flagsmith.py @@ -9,6 +9,7 @@ from urllib3 import Retry from flagsmith.analytics import ( + DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS, FLAG_EXPOSURE_EVENT, AnalyticsProcessor, EventProcessor, @@ -453,6 +454,33 @@ def track_exposure_event( metadata=metadata, ) + def flush_events(self) -> None: + """ + Send the events tracked so far, and return once each of them has been + delivered, dropped, or kept for the next flush, retries included. + """ + if not self._event_processor: + raise ValueError("Events must be enabled to flush events.") + self._event_processor.flush() + + def close( + self, timeout_seconds: float = DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS + ) -> None: + """ + Stop the client's background work, and send the buffered events. + + :param timeout_seconds: how long to wait for the buffered events to be + sent. Events not sent by then are lost. + """ + if hasattr(self, "environment_data_polling_manager_thread"): + self.environment_data_polling_manager_thread.stop() + + if hasattr(self, "event_stream_thread"): + self.event_stream_thread.stop() + + if self._event_processor: + self._event_processor.stop(timeout_seconds) + def update_environment(self) -> None: try: environment_data = self._get_json_response( @@ -592,11 +620,4 @@ def _get_json_response( ) from e def __del__(self) -> None: - if hasattr(self, "environment_data_polling_manager_thread"): - self.environment_data_polling_manager_thread.stop() - - if hasattr(self, "event_stream_thread"): - self.event_stream_thread.stop() - - if self._event_processor: - self._event_processor.stop() + self.close() diff --git a/sdk-spec/steps.py b/sdk-spec/steps.py index 7c7700a..3f2f221 100644 --- a/sdk-spec/steps.py +++ b/sdk-spec/steps.py @@ -50,6 +50,19 @@ def create_client_with_local_evaluation( ) +# Create the SDK client with local evaluation and events enabled +def create_client_with_local_evaluation_and_events( + environment_key: str, api_url: str, events_api_url: str +) -> Flagsmith: + return Flagsmith( + environment_key=environment_key, + api_url=api_url, + enable_local_evaluation=True, + enable_events=True, + event_processor_config=EventProcessorConfig(events_api_url=events_api_url), + ) + + # When the application gets the environment flags def get_environment_flags(flagsmith: Flagsmith) -> list[SpecFlag]: return as_spec_flags(flagsmith.get_environment_flags()) @@ -92,3 +105,21 @@ def track_exposure_event( flagsmith.track_exposure_event( feature, identifier=identifier, value=value, metadata=metadata ) + + +# When the application tracks events "event_{first}" to "event_{last}" for identity "{identifier}" +def track_numbered_events( + flagsmith: Flagsmith, first: int, last: int, identifier: str +) -> None: + for number in range(first, last + 1): + flagsmith.track_event(f"event_{number}", identifier=identifier) + + +# When the application flushes events +def flush_events(flagsmith: Flagsmith) -> None: + flagsmith.flush_events() + + +# When the application closes the client with a timeout of {timeout} seconds +def close_client(flagsmith: Flagsmith, timeout: float) -> None: + flagsmith.close(timeout_seconds=timeout) diff --git a/tests/test_event_processor.py b/tests/test_event_processor.py index 52595a7..6047dff 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -1,7 +1,14 @@ import json -from concurrent.futures import Future +import logging +import threading +import time +import typing from datetime import datetime -from unittest import mock + +import pytest +import requests +import responses +from pytest_mock import MockerFixture from flagsmith.analytics import ( FLAG_EXPOSURE_EVENT, @@ -120,18 +127,19 @@ def test_track_exposure_event__differing_events__not_deduplicated( def test_track_exposure_event__equal_event_after_flush__buffered_again( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: # Given - with mock.patch("flagsmith.analytics.session"): - event_processor.track_exposure_event( - feature_name="checkout_v2", identifier="user1", value="variant_b" - ) - event_processor.flush() + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_exposure_event( + feature_name="checkout_v2", identifier="user1", value="variant_b" + ) + event_processor.flush() - # When - event_processor.track_exposure_event( - feature_name="checkout_v2", identifier="user1", value="variant_b" - ) + # When + event_processor.track_exposure_event( + feature_name="checkout_v2", identifier="user1", value="variant_b" + ) # Then assert len(event_processor._buffer) == 1 @@ -148,69 +156,315 @@ def test_track_event__equal_events__not_deduplicated( assert len(event_processor._buffer) == 2 -def test_auto_flush_on_buffer_full() -> None: - config = EventProcessorConfig(events_api_url="http://test/", max_buffer_items=5) - processor = EventProcessor(config=config, environment_key="key") +def test_flush__full_buffer__sends_events_once( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=5 + ), + environment_key="test_key", + ) + mocked_responses.post("http://test_analytics/v1/events", status=202) + for i in range(5): + processor.track_event(event=f"event_{i}") - with mock.patch("flagsmith.analytics.session"): - for i in range(5): - processor.track_event(event=f"event_{i}") + # When + processor.flush() - assert len(processor._buffer) == 0 + # Then + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == [f"event_{i}" for i in range(5)] + assert processor._buffer == [] + + +def test_flush__full_buffer__drops_oldest_event( + mocker: MockerFixture, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=3 + ), + environment_key="test_key", + ) + for status in (503, 503, 503, 202): + mocked_responses.post("http://test_analytics/v1/events", status=status) + for i in range(1, 4): + processor.track_event(event=f"event_{i}") + processor.flush() + # When + processor.track_event(event="event_4") + processor.flush() -def test_flush_sends_correct_http_request(event_processor: EventProcessor) -> None: - with mock.patch("flagsmith.analytics.session") as mock_session: - event_processor.track_event(event="purchase", identifier="user1") - event_processor.flush() + # Then + assert len(mocked_responses.calls) == 4 + request_body = mocked_responses.calls[3].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == ["event_2", "event_3", "event_4"] + assert "Event buffer is full, dropping the oldest event_1 event" in ( + caplog.messages + ) - mock_session.post.assert_called_once() - call_kwargs = mock_session.post.call_args - assert call_kwargs[0][0] == "http://test_analytics/v1/events" - headers = call_kwargs[1]["headers"] - assert headers["X-Environment-Key"] == "test_key" - assert headers["Content-Type"] == "application/json; charset=utf-8" - assert "flagsmith-python-sdk/" in headers["Flagsmith-SDK-User-Agent"] +def test_flush__sends_events_to_events_api( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + post = mocker.spy(event_processor._session, "post") + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="purchase", identifier="user1") - body = json.loads(call_kwargs[1]["data"]) + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 1 + request = mocked_responses.calls[0].request + assert post.call_args.kwargs["timeout"] == 3 + assert request.headers["X-Environment-Key"] == "test_key" + assert request.headers["Content-Type"] == "application/json; charset=utf-8" + assert "flagsmith-python-sdk/" in request.headers["Flagsmith-SDK-User-Agent"] + request_body = request.body + assert request_body + body = json.loads(request_body) assert "environment_key" not in body assert len(body["events"]) == 1 assert body["events"][0]["event"] == "purchase" + assert event_processor._buffer == [] -def test_flush_noop_when_empty(event_processor: EventProcessor) -> None: - with mock.patch("flagsmith.analytics.session") as mock_session: - event_processor.flush() +def test_flush__no_events__sends_nothing( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 0 + - mock_session.post.assert_not_called() +def test_flush__transient_failures__retried_with_backoff( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + uniform = mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + mocked_responses.post("http://test_analytics/v1/events", status=503) + mocked_responses.post( + "http://test_analytics/v1/events", body=requests.ConnectionError("reset") + ) + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="purchase") + # When + event_processor.flush() -def test_failed_flush_requeues_events(event_processor: EventProcessor) -> None: - future: Future[None] = Future() - future.set_exception(Exception("connection error")) + # Then + assert len(mocked_responses.calls) == 3 + assert uniform.call_args_list == [mocker.call(0, 1), mocker.call(0, 2)] + assert event_processor._buffer == [] + assert caplog.messages == [ + "Failed to send 1 events, attempt 1 of 3: status 503", + "Failed to send 1 events, attempt 2 of 3: reset", + ] - with mock.patch("flagsmith.analytics.session") as mock_session: - mock_session.post.return_value = future - event_processor.track_event(event="purchase") - event_processor.flush() - assert len(event_processor._buffer) == 1 - assert event_processor._buffer[0]["event"] == "purchase" +@pytest.mark.parametrize( + "statuses", + [ + pytest.param([503, 503, 503], id="unavailable"), + pytest.param([500, 500, 503], id="unavailable-after-server-errors"), + pytest.param([429, 429, 429], id="throttled"), + ], +) +def test_flush__last_attempt_fails_transiently__kept_ahead_of_newer_events( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + statuses: typing.List[int], +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + for status in (*statuses, 202): + mocked_responses.post("http://test_analytics/v1/events", status=status) + event_processor.track_event(event="add_to_cart") + event_processor.flush() + + # When + event_processor.track_event(event="checkout") + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 4 + request_body = mocked_responses.calls[3].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == ["add_to_cart", "checkout"] + assert "Keeping 1 events for the next flush" in caplog.messages + + +@pytest.mark.parametrize( + "statuses, expected_log", + [ + pytest.param( + [400], + "Events API rejected 1 events with 400, dropping them", + id="rejected", + ), + pytest.param( + [503, 503, 500], + "Dropping 1 events the events API failed with 500", + id="server-error-last", + ), + ], +) +def test_flush__batch_rejected_or_last_failed_with_server_error__dropped( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + statuses: typing.List[int], + expected_log: str, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + for status in (*statuses, 202): + mocked_responses.post("http://test_analytics/v1/events", status=status) + event_processor.track_event(event="add_to_cart") + event_processor.flush() + + # When + event_processor.track_event(event="checkout") + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == len(statuses) + 1 + request_body = mocked_responses.calls[-1].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == ["checkout"] + assert expected_log in caplog.messages -def test_start_stop_lifecycle() -> None: - config = EventProcessorConfig( - events_api_url="http://test/", flush_interval_seconds=100 +@pytest.mark.parametrize("status", [401, 403]) +def test_flush__environment_key_refused__stops_event_tracking( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + status: int, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=status) + event_processor.track_event(event="add_to_cart") + event_processor.flush() + + # When + with caplog.at_level(logging.DEBUG, logger="flagsmith.analytics"): + event_processor.track_event(event="checkout") + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 1 + assert event_processor._buffer == [] + assert caplog.messages == [ + f"Events API refused the environment key with {status}," + " stopping event tracking", + "Not tracking checkout event: the events API refused the environment key.", + ] + + +def test_start__flushes_on_timer( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", flush_interval_seconds=0.01 + ), + environment_key="test_key", ) - processor = EventProcessor(config=config, environment_key="key") + mocked_responses.post("http://test_analytics/v1/events", status=202) + processor.track_event(event="purchase") + + # When + processor.start() + deadline = time.monotonic() + 5 + while not mocked_responses.calls and time.monotonic() < deadline: + time.sleep(0.01) + processor.stop() + + # Then + assert len(mocked_responses.calls) == 1 + assert processor._buffer == [] + +def test_stop__sends_buffered_events( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", flush_interval_seconds=100 + ), + environment_key="test_key", + ) processor.start() + mocked_responses.post("http://test_analytics/v1/events", status=202) + processor.track_event(event="purchase") + + # When + processor.stop() + + # Then assert processor._timer is not None - assert processor._timer.is_alive() + assert not processor._timer.is_alive() + assert len(mocked_responses.calls) == 1 + assert processor._buffer == [] + + +def test_stop__events_api_not_answering__returns_within_timeout( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + answer = threading.Event() + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=lambda request: (answer.wait(5), (503, {}, ""))[1], + ) + event_processor.track_event(event="purchase") - with mock.patch("flagsmith.analytics.session"): - processor.track_event(event="purchase") - processor.stop() + # When + started = time.monotonic() + event_processor.stop(timeout_seconds=0.1) + elapsed = time.monotonic() - started + answer.set() + with event_processor._send_lock: + pass - assert len(processor._buffer) == 0 + # Then + assert elapsed < 1 + assert len(mocked_responses.calls) == 1 + assert caplog.messages == [ + "Gave up sending events after 0.1 seconds while closing", + "Failed to send 1 events, attempt 1 of 3: status 503", + "Not retrying 1 events: the client is closed", + ] diff --git a/tests/test_flagsmith.py b/tests/test_flagsmith.py index 7998c23..9084211 100644 --- a/tests/test_flagsmith.py +++ b/tests/test_flagsmith.py @@ -1090,6 +1090,69 @@ def test_track_event_delegates_to_event_processor( ) +def test_flush_events__events_tracked__sent_to_events_api( + api_key: str, mocked_responses: responses.RequestsMock +) -> None: + # Given + flagsmith = Flagsmith( + environment_key=api_key, + enable_events=True, + event_processor_config=EventProcessorConfig(events_api_url="http://test/"), + ) + mocked_responses.post("http://test/v1/events", status=202) + flagsmith.track_event("purchase", identifier="user1") + + # When + flagsmith.flush_events() + + # Then + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == ["purchase"] + + +def test_flush_events__events_disabled__raises(api_key: str) -> None: + # Given + flagsmith = Flagsmith(environment_key=api_key) + + # When / Then + with pytest.raises(ValueError, match="Events must be enabled to flush events."): + flagsmith.flush_events() + + +def test_close__events_tracked__sent_within_timeout( + mocker: MockerFixture, api_key: str +) -> None: + # Given + flagsmith = Flagsmith(environment_key=api_key, enable_events=True) + assert flagsmith._event_processor + mock_stop = mocker.patch.object(flagsmith._event_processor, "stop") + + # When + flagsmith.close(timeout_seconds=2) + + # Then + mock_stop.assert_called_once_with(2) + + +def test_close__local_evaluation__stops_polling( + mocker: MockerFixture, + local_eval_flagsmith: Flagsmith, +) -> None: + # Given + mock_stop = mocker.patch.object( + local_eval_flagsmith.environment_data_polling_manager_thread, "stop" + ) + + # When + local_eval_flagsmith.close() + + # Then + mock_stop.assert_called_once_with() + + def test_track_exposure_event_raises_without_config(api_key: str) -> None: flagsmith = Flagsmith(environment_key=api_key) with pytest.raises(ValueError, match="Events must be enabled"): From 96e767f9d623e04f0545d8f55930eb996bb1b325 Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:29:47 +0000 Subject: [PATCH 4/9] fix: Stop flushing events on a timer once the client is closed A timer flush running while the client closed scheduled another one, so the client kept flushing after `close()`. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit ca330ed0791c7574c15d7069ab6d5777d47828da) --- flagsmith/analytics.py | 4 ++++ tests/test_event_processor.py | 24 +++++++++++++++++++++++- 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index 44b7368..77d5e9c 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -149,6 +149,7 @@ def __init__( self._attempted_count = 0 self._full_buffer_flush_pending = False self._refused = False + self._stopped = threading.Event() self._closed = threading.Event() self._timer: typing.Optional[threading.Timer] = None @@ -371,6 +372,7 @@ def stop( at most `timeout_seconds`. Events not sent by then are not retried. """ atexit.unregister(self.stop) + self._stopped.set() if self._timer is not None: self._timer.cancel() final_flush = threading.Thread(target=self._send, daemon=True) @@ -384,6 +386,8 @@ def stop( ) def _schedule_flush(self) -> None: + if self._stopped.is_set(): + return self._timer = threading.Timer(self._flush_interval_seconds, self._timer_flush) self._timer.daemon = True self._timer.start() diff --git a/tests/test_event_processor.py b/tests/test_event_processor.py index 6047dff..19cab36 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -389,6 +389,28 @@ def test_flush__environment_key_refused__stops_event_tracking( ] +def test_stop__timer_flush_in_progress__no_further_timer_flushes( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", flush_interval_seconds=100 + ), + environment_key="test_key", + ) + processor.start() + timer = processor._timer + processor.stop() + + # When + processor._timer_flush() + + # Then + assert processor._timer is timer + assert len(mocked_responses.calls) == 0 + + def test_start__flushes_on_timer( mocked_responses: responses.RequestsMock, ) -> None: @@ -433,7 +455,7 @@ def test_stop__sends_buffered_events( # Then assert processor._timer is not None - assert not processor._timer.is_alive() + assert processor._timer.finished.is_set() assert len(mocked_responses.calls) == 1 assert processor._buffer == [] From cd3aa8ce1113ce7cc7cf424f8e938ffa04d54e11 Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:12:04 +0000 Subject: [PATCH 5/9] refactor: Send events from one long-lived worker thread The event processor created threads three ways (a timer, a thread per full buffer, a thread per close) behind a pile of locks and flags. One worker thread now sends the buffered events on a timer, when the buffer fills, on flush, and once more on close. - `close()` is terminal and idempotent: tracking and flushing stop, and no request starts once it returns. - `__del__` signals the worker without waiting for it. - `flush_events()` never raises, and logs when events are not enabled. - Events are serialised when tracked: one that isn't JSON serialisable is logged and not tracked, instead of failing every later flush. - The retry policy is one set of status codes (408, 429, 500, 502, 503, 504), and only connection errors and timeouts are retried. Other statuses, e.g. 501, drop the batch instead of keeping it forever. Co-Authored-By: Claude Opus 5.5 --- flagsmith/analytics.py | 341 +++++++++++++++++--------------- flagsmith/flagsmith.py | 22 ++- tests/conftest.py | 8 +- tests/test_event_processor.py | 358 ++++++++++++++++++++++------------ tests/test_flagsmith.py | 55 +++++- 5 files changed, 482 insertions(+), 302 deletions(-) diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index 77d5e9c..c5256bf 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -4,7 +4,9 @@ import logging import random import threading +import time import typing +from collections import deque from dataclasses import dataclass from datetime import datetime @@ -28,7 +30,9 @@ EVENTS_MAX_ATTEMPTS: typing.Final[int] = 3 EVENTS_RETRY_BASE_SECONDS: typing.Final[float] = 1 DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS: typing.Final[float] = 5 -RETRIED_STATUS_CODES: typing.Final[typing.FrozenSet[int]] = frozenset({408, 429}) +RETRIED_STATUS_CODES: typing.Final[typing.FrozenSet[int]] = frozenset( + {408, 429, 500, 502, 503, 504} +) session = FuturesSession(max_workers=4) @@ -115,15 +119,15 @@ class _DeliveryOutcome(enum.Enum): class EventProcessor: """ Buffered event processor that batches custom events and POSTs them to the - Flagsmith event endpoint. Flushes on a background timer or when the buffer - fills. + Flagsmith event endpoint, from one worker thread. The worker sends the + buffered events on a timer, when the buffer fills, on `flush`, and once + more on `close`. A batch is sent up to ``EVENTS_MAX_ATTEMPTS`` times, backing off with full - jitter between attempts, one batch at a time. A batch whose last attempt - fails transiently is kept for the next flush, unless that attempt got a - 500. A batch the events API rejects is dropped, and an events API refusing - the environment key stops event tracking. When the buffer is full, the - oldest event is dropped. + jitter between attempts. A batch whose last attempt fails transiently is + kept for the next send, unless that attempt got a 500. A batch the events + API rejects is dropped, and an events API refusing the environment key + stops event tracking. When the buffer is full, the oldest event is dropped. """ def __init__( @@ -140,18 +144,24 @@ def __init__( self._flush_interval_seconds = config.flush_interval_seconds self._session = requests.Session() - self._buffer: typing.List[typing.Dict[str, typing.Any]] = [] + self._condition = threading.Condition() + self._buffer: typing.Deque[str] = deque() self._buffered_exposure_keys: typing.Set[ExposureKey] = set() - self._lock = threading.Lock() - # Held while a batch is in flight, so that one batch is sent at a time. - self._send_lock = threading.Lock() + # Events are numbered as tracked: a flush waits until the worker has + # sent every event up to the number tracked when it was called. self._tracked_count = 0 - self._attempted_count = 0 - self._full_buffer_flush_pending = False + self._flush_requested_count = 0 + self._sent_count = 0 self._refused = False - self._stopped = threading.Event() - self._closed = threading.Event() - self._timer: typing.Optional[threading.Timer] = None + self._closed = False + # Set once `close` returns: the worker starts no more requests. + self._abandoned = threading.Event() + + self._worker = threading.Thread( + target=self._run, name="flagsmith-event-processor", daemon=True + ) + self._worker.start() + atexit.register(self.close) def track_event( self, @@ -196,12 +206,25 @@ def _buffer_event( traits: typing.Optional[typing.Dict[str, typing.Any]], metadata: typing.Optional[typing.Dict[str, typing.Any]], ) -> None: - with self._lock: - if self._refused: + try: + serialised_event = json.dumps( + { + "event": event, + "feature_name": feature_name, + "identifier": identifier, + "value": str(value) if value is not None else None, + "traits": dict(traits) if traits else None, + "metadata": {**(metadata or {}), "sdk_version": __version__}, + "timestamp": int(datetime.now().timestamp() * 1000), + } + ) + except (TypeError, ValueError) as e: + logger.warning("Not tracking %s event: not JSON serialisable: %s", event, e) + return + with self._condition: + if self._closed or self._refused: logger.debug( - "Not tracking %s event: the events API refused the" - " environment key.", - event, + "Not tracking %s event: event tracking has stopped.", event ) return if event == FLAG_EXPOSURE_EVENT: @@ -222,176 +245,176 @@ def _buffer_event( ) return self._buffered_exposure_keys.add(exposure_key) - if len(self._buffer) >= self._max_buffer: - dropped = self._buffer.pop(0) - logger.warning( - "Event buffer is full, dropping the oldest %s event", - dropped["event"], - ) - self._buffer.append( - { - "event": event, - "feature_name": feature_name, - "identifier": identifier, - "value": str(value) if value is not None else None, - "traits": dict(traits) if traits else None, - "metadata": {**(metadata or {}), "sdk_version": __version__}, - "timestamp": int(datetime.now().timestamp() * 1000), - } - ) + self._buffer.append(serialised_event) self._tracked_count += 1 - if len(self._buffer) < self._max_buffer or ( - self._full_buffer_flush_pending - ): - return - self._full_buffer_flush_pending = True - tracked_count = self._tracked_count - - threading.Thread( - target=self._flush_full_buffer, args=(tracked_count,), daemon=True - ).start() + self._drop_oldest_events() + if len(self._buffer) >= self._max_buffer: + self._condition.notify_all() def flush(self) -> None: """ Sends the events tracked so far, and returns once each of them has - been delivered, dropped, or kept for the next flush. Events already - sent while this was waiting for another batch in flight are not sent - again. + been delivered, dropped, or kept for the next send. Does nothing once + closed. """ - with self._lock: - tracked_count = self._tracked_count - self._send(tracked_count) + with self._condition: + if self._closed: + return + flush_count = self._tracked_count + self._flush_requested_count = max(self._flush_requested_count, flush_count) + self._condition.notify_all() + self._condition.wait_for( + lambda: self._sent_count >= flush_count or not self._worker.is_alive() + ) - def _flush_full_buffer(self, tracked_count: int) -> None: - with self._lock: - self._full_buffer_flush_pending = False - self._send(tracked_count) + def close( + self, timeout_seconds: float = DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS + ) -> None: + """ + Stops event tracking, and sends the buffered events, waiting for at + most `timeout_seconds`. No request starts once this returns. Calling + it again does nothing. + """ + if not self.stop(): + return + self._worker.join(timeout_seconds) + self._abandoned.set() + if self._worker.is_alive(): + logger.warning( + "Gave up sending events after %s seconds while closing", + timeout_seconds, + ) - def _send(self, tracked_count: typing.Optional[int] = None) -> None: + def stop(self) -> bool: """ - Sends the buffered events as one batch. Given `tracked_count`, does - nothing if every event tracked up to it has been sent already. + Stops event tracking, and has the worker send the buffered events, + without waiting for it. Returns whether this stopped event tracking. """ - with self._send_lock: - with self._lock: - if not self._buffer or ( - tracked_count is not None and self._attempted_count >= tracked_count + with self._condition: + if self._closed: + return False + self._closed = True + self._condition.notify_all() + atexit.unregister(self.close) + return True + + def _run(self) -> None: + next_send_at = time.monotonic() + self._flush_interval_seconds + while True: + with self._condition: + while not ( + self._closed + or self._flush_requested_count > self._sent_count + or ( + len(self._buffer) >= self._max_buffer + and self._tracked_count > self._sent_count + ) + or time.monotonic() >= next_send_at ): - return - events = self._buffer - self._buffer = [] + self._condition.wait(next_send_at - time.monotonic()) + closing = self._closed + events = list(self._buffer) + self._buffer.clear() self._buffered_exposure_keys.clear() - attempted_count = self._tracked_count + sent_count = self._tracked_count + next_send_at = time.monotonic() + self._flush_interval_seconds - outcome = self._deliver(events) + try: + outcome = ( + self._deliver(events) if events else _DeliveryOutcome.DELIVERED + ) + except Exception: + logger.exception("Dropping %d events: failed to send", len(events)) + outcome = _DeliveryOutcome.DROPPED - with self._lock: - self._attempted_count = attempted_count + with self._condition: + self._sent_count = sent_count if outcome is _DeliveryOutcome.KEPT: - max_buffer = self._max_buffer - self._buffer = (events + self._buffer)[-max_buffer:] + self._buffer.extendleft(reversed(events)) + self._drop_oldest_events() elif outcome is _DeliveryOutcome.REFUSED: self._refused = True - self._buffer = [] + self._buffer.clear() self._buffered_exposure_keys.clear() + self._condition.notify_all() + if closing: + self._session.close() + return - def _deliver( - self, events: typing.List[typing.Dict[str, typing.Any]] - ) -> _DeliveryOutcome: - payload = json.dumps({"events": events}) + def _drop_oldest_events(self) -> None: + while len(self._buffer) > self._max_buffer: + self._buffer.popleft() + logger.warning("Event buffer is full, dropped the oldest event") + + def _deliver(self, events: typing.List[str]) -> _DeliveryOutcome: + payload = f'{{"events": [{", ".join(events)}]}}' + status: typing.Optional[int] = None for attempt in range(1, EVENTS_MAX_ATTEMPTS + 1): if attempt > 1: backoff = EVENTS_RETRY_BASE_SECONDS * 2 ** (attempt - 2) - if self._closed.wait(random.uniform(0, backoff)): - logger.warning( - "Not retrying %d events: the client is closed", len(events) - ) - return _DeliveryOutcome.KEPT - try: - response = self._session.post( - self._batch_endpoint, - data=payload, - timeout=EVENTS_REQUEST_TIMEOUT_SECONDS, - headers={ - "Content-Type": "application/json; charset=utf-8", - "X-Environment-Key": self._environment_key, - "Flagsmith-SDK-User-Agent": DEFAULT_USER_AGENT, - }, - ) - except requests.RequestException as e: + self._abandoned.wait(random.uniform(0, backoff)) + if self._abandoned.is_set(): logger.warning( - "Failed to send %d events, attempt %d of %d: %s", - len(events), - attempt, - EVENTS_MAX_ATTEMPTS, - e, + "Not sending %d events: the client is closed", len(events) ) + return _DeliveryOutcome.DROPPED + try: + response = self._post(payload) + except (requests.ConnectionError, requests.Timeout) as e: status = None + failure = str(e) + except requests.RequestException as e: + logger.error("Dropping %d events: %s", len(events), e) + return _DeliveryOutcome.DROPPED else: status = response.status_code - if response.ok: - return _DeliveryOutcome.DELIVERED - if status in (401, 403): - logger.error( - "Events API refused the environment key with %d," - " stopping event tracking", - status, - ) - return _DeliveryOutcome.REFUSED - if status not in RETRIED_STATUS_CODES and status < 500: - logger.warning( - "Events API rejected %d events with %d, dropping them", - len(events), - status, - ) - return _DeliveryOutcome.DROPPED - logger.warning( - "Failed to send %d events, attempt %d of %d: status %d", - len(events), - attempt, - EVENTS_MAX_ATTEMPTS, - status, - ) + if outcome := self._outcome_of(status, len(events)): + return outcome + failure = f"status {status}" + logger.warning( + "Failed to send %d events, attempt %d of %d: %s", + len(events), + attempt, + EVENTS_MAX_ATTEMPTS, + failure, + ) if status == 500: logger.warning( "Dropping %d events the events API failed with 500", len(events) ) return _DeliveryOutcome.DROPPED - logger.warning("Keeping %d events for the next flush", len(events)) + logger.warning("Keeping %d events for the next send", len(events)) return _DeliveryOutcome.KEPT - def start(self) -> None: - self._schedule_flush() - atexit.register(self.stop) + def _post(self, payload: str) -> requests.Response: + return self._session.post( + self._batch_endpoint, + data=payload, + timeout=EVENTS_REQUEST_TIMEOUT_SECONDS, + headers={ + "Content-Type": "application/json; charset=utf-8", + "X-Environment-Key": self._environment_key, + "Flagsmith-SDK-User-Agent": DEFAULT_USER_AGENT, + }, + ) - def stop( - self, timeout_seconds: float = DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS - ) -> None: - """ - Stops flushing on a timer, and sends the buffered events, waiting for - at most `timeout_seconds`. Events not sent by then are not retried. - """ - atexit.unregister(self.stop) - self._stopped.set() - if self._timer is not None: - self._timer.cancel() - final_flush = threading.Thread(target=self._send, daemon=True) - final_flush.start() - final_flush.join(timeout_seconds) - self._closed.set() - if final_flush.is_alive(): - logger.warning( - "Gave up sending events after %s seconds while closing", - timeout_seconds, + @staticmethod + def _outcome_of(status: int, event_count: int) -> typing.Optional[_DeliveryOutcome]: + """The outcome of a request answered with `status`, or None to retry it.""" + if status < 400: + return _DeliveryOutcome.DELIVERED + if status in (401, 403): + logger.error( + "Events API refused the environment key with %d," + " stopping event tracking", + status, ) - - def _schedule_flush(self) -> None: - if self._stopped.is_set(): - return - self._timer = threading.Timer(self._flush_interval_seconds, self._timer_flush) - self._timer.daemon = True - self._timer.start() - - def _timer_flush(self) -> None: - self._send() - self._schedule_flush() + return _DeliveryOutcome.REFUSED + if status in RETRIED_STATUS_CODES: + return None + logger.warning( + "Events API rejected %d events with %d, dropping them", + event_count, + status, + ) + return _DeliveryOutcome.DROPPED diff --git a/flagsmith/flagsmith.py b/flagsmith/flagsmith.py index fd80e57..7dd0026 100644 --- a/flagsmith/flagsmith.py +++ b/flagsmith/flagsmith.py @@ -233,7 +233,6 @@ def _initialise_events( config=event_processor_config or EventProcessorConfig(), environment_key=environment_key, ) - self._event_processor.start() def _initialise_local_evaluation(self) -> None: # To ensure that the environment is set before allowing subsequent @@ -457,30 +456,35 @@ def track_exposure_event( def flush_events(self) -> None: """ Send the events tracked so far, and return once each of them has been - delivered, dropped, or kept for the next flush, retries included. + delivered, dropped, or kept for the next send, retries included. Does + nothing when events are not enabled, or once the client is closed. """ if not self._event_processor: - raise ValueError("Events must be enabled to flush events.") + logger.warning("Not flushing events: events are not enabled.") + return self._event_processor.flush() def close( self, timeout_seconds: float = DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS ) -> None: """ - Stop the client's background work, and send the buffered events. + Stop the client's background work, and send the buffered events. Events + tracked afterwards are not sent. Calling it again does nothing. :param timeout_seconds: how long to wait for the buffered events to be sent. Events not sent by then are lost. """ + self._stop_background_work() + if self._event_processor: + self._event_processor.close(timeout_seconds) + + def _stop_background_work(self) -> None: if hasattr(self, "environment_data_polling_manager_thread"): self.environment_data_polling_manager_thread.stop() if hasattr(self, "event_stream_thread"): self.event_stream_thread.stop() - if self._event_processor: - self._event_processor.stop(timeout_seconds) - def update_environment(self) -> None: try: environment_data = self._get_json_response( @@ -620,4 +624,6 @@ def _get_json_response( ) from e def __del__(self) -> None: - self.close() + self._stop_background_work() + if self._event_processor: + self._event_processor.stop() diff --git a/tests/conftest.py b/tests/conftest.py index 0663be9..5849a03 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -51,7 +51,7 @@ def tracking_init(self: Flagsmith, *args: typing.Any, **kwargs: typing.Any) -> N stream.stop() stream.join(timeout=5) if flagsmith._event_processor: - flagsmith._event_processor.stop() + flagsmith._event_processor.close(timeout_seconds=0) @pytest.fixture() @@ -69,11 +69,13 @@ def event_processor_config() -> EventProcessorConfig: @pytest.fixture() def event_processor( event_processor_config: EventProcessorConfig, -) -> EventProcessor: - return EventProcessor( +) -> Generator[EventProcessor, None, None]: + processor = EventProcessor( config=event_processor_config, environment_key="test_key", ) + yield processor + processor.close(timeout_seconds=0) @pytest.fixture(scope="session") diff --git a/tests/test_event_processor.py b/tests/test_event_processor.py index 19cab36..458f3b7 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -17,7 +17,15 @@ ) -def test_track_event_buffers_event(event_processor: EventProcessor) -> None: +def test_flush__tracked_event__sent_to_events_api( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + post = mocker.spy(event_processor._session, "post") + mocked_responses.post("http://test_analytics/v1/events", status=202) + before = int(datetime.now().timestamp() * 1000) event_processor.track_event( event="purchase", identifier="user1", @@ -25,10 +33,23 @@ def test_track_event_buffers_event(event_processor: EventProcessor) -> None: traits={"plan": "premium"}, metadata={"sku": "abc"}, ) - event_processor.track_event(event="purchase", identifier="user1") + after = int(datetime.now().timestamp() * 1000) + + # When + event_processor.flush() - assert len(event_processor._buffer) == 2 - event = event_processor._buffer[0] + # Then + assert len(mocked_responses.calls) == 1 + assert post.call_args.kwargs["timeout"] == 3 + request = mocked_responses.calls[0].request + assert request.headers["X-Environment-Key"] == "test_key" + assert request.headers["Content-Type"] == "application/json; charset=utf-8" + assert "flagsmith-python-sdk/" in request.headers["Flagsmith-SDK-User-Agent"] + request_body = request.body + assert request_body + body = json.loads(request_body) + assert "environment_key" not in body + [event] = body["events"] assert event["event"] == "purchase" assert event["feature_name"] is None assert event["identifier"] == "user1" @@ -36,19 +57,29 @@ def test_track_event_buffers_event(event_processor: EventProcessor) -> None: assert event["traits"] == {"plan": "premium"} assert event["metadata"]["sku"] == "abc" assert "sdk_version" in event["metadata"] - assert isinstance(event["timestamp"], int) + assert before <= event["timestamp"] <= after -def test_buffer_stringifies_non_string_values( +def test_flush__event_values__sent_as_strings( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_event(event="purchase", value=99.5) event_processor.track_event(event="opt_in", value=True) event_processor.track_event(event="count", value=3) event_processor.track_event(event="variant", value="control") event_processor.track_event(event="empty") - assert [e["value"] for e in event_processor._buffer] == [ + # When + event_processor.flush() + + # Then + request_body = mocked_responses.calls[0].request.body + assert request_body + body = json.loads(request_body) + assert [e["value"] for e in body["events"]] == [ "99.5", "True", "3", @@ -57,19 +88,12 @@ def test_buffer_stringifies_non_string_values( ] -def test_track_event_defaults_timestamp_to_now( - event_processor: EventProcessor, -) -> None: - before = int(datetime.now().timestamp() * 1000) - event_processor.track_event(event="ping") - after = int(datetime.now().timestamp() * 1000) - - assert before <= event_processor._buffer[0]["timestamp"] <= after - - -def test_track_exposure_event_buffers_with_flag_exposure_event_name( +def test_flush__exposure_event__sent_as_flag_exposure( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_exposure_event( feature_name="checkout_v2", identifier="user1", @@ -78,8 +102,13 @@ def test_track_exposure_event_buffers_with_flag_exposure_event_name( metadata={"source": "homepage"}, ) - assert len(event_processor._buffer) == 1 - event = event_processor._buffer[0] + # When + event_processor.flush() + + # Then + request_body = mocked_responses.calls[0].request.body + assert request_body + [event] = json.loads(request_body)["events"] assert event["event"] == FLAG_EXPOSURE_EVENT == "$flag_exposure" assert event["feature_name"] == "checkout_v2" assert event["identifier"] == "user1" @@ -89,25 +118,32 @@ def test_track_exposure_event_buffers_with_flag_exposure_event_name( assert "sdk_version" in event["metadata"] -def test_track_exposure_event__equal_events_in_flush_interval__deduplicated( +def test_flush__equal_exposures_in_flush_interval__sent_once( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: - # Given / When + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) for _ in range(3): event_processor.track_exposure_event( - feature_name="checkout_v2", - identifier="user1", - value="variant_b", + feature_name="checkout_v2", identifier="user1", value="variant_b" ) + # When + event_processor.flush() + # Then - assert len(event_processor._buffer) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert len(json.loads(request_body)["events"]) == 1 -def test_track_exposure_event__differing_events__not_deduplicated( +def test_flush__differing_exposures__each_sent( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: - # Given / When + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_exposure_event( feature_name="checkout_v2", identifier="user1", value="variant_b" ) @@ -121,11 +157,16 @@ def test_track_exposure_event__differing_events__not_deduplicated( feature_name="banner_test", identifier="user1", value="variant_b" ) + # When + event_processor.flush() + # Then - assert len(event_processor._buffer) == 4 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert len(json.loads(request_body)["events"]) == 4 -def test_track_exposure_event__equal_event_after_flush__buffered_again( +def test_flush__equal_exposure_after_flush__sent_again( event_processor: EventProcessor, mocked_responses: responses.RequestsMock, ) -> None: @@ -135,28 +176,59 @@ def test_track_exposure_event__equal_event_after_flush__buffered_again( feature_name="checkout_v2", identifier="user1", value="variant_b" ) event_processor.flush() - - # When event_processor.track_exposure_event( feature_name="checkout_v2", identifier="user1", value="variant_b" ) + # When + event_processor.flush() + # Then - assert len(event_processor._buffer) == 1 + assert len(mocked_responses.calls) == 2 -def test_track_event__equal_events__not_deduplicated( +def test_flush__equal_custom_events__each_sent( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: - # Given / When + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_event(event="purchase", identifier="user1", value="99.5") event_processor.track_event(event="purchase", identifier="user1", value="99.5") + # When + event_processor.flush() + + # Then + request_body = mocked_responses.calls[0].request.body + assert request_body + assert len(json.loads(request_body)["events"]) == 2 + + +def test_track_event__metadata_not_json_serialisable__event_not_tracked( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) + + # When + event_processor.track_event(event="purchase", metadata={"at": datetime.now()}) + event_processor.track_event(event="checkout") + event_processor.flush() + # Then - assert len(event_processor._buffer) == 2 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == ["checkout"] + assert caplog.messages == [ + "Not tracking purchase event: not JSON serialisable:" + " Object of type datetime is not JSON serializable" + ] -def test_flush__full_buffer__sends_events_once( +def test_track_event__full_buffer__sent_without_flush( mocked_responses: responses.RequestsMock, ) -> None: # Given @@ -167,11 +239,15 @@ def test_flush__full_buffer__sends_events_once( environment_key="test_key", ) mocked_responses.post("http://test_analytics/v1/events", status=202) - for i in range(5): - processor.track_event(event=f"event_{i}") # When + for i in range(5): + processor.track_event(event=f"event_{i}") + deadline = time.monotonic() + 5 + while not mocked_responses.calls and time.monotonic() < deadline: + time.sleep(0.01) processor.flush() + processor.close(timeout_seconds=0) # Then assert len(mocked_responses.calls) == 1 @@ -179,7 +255,6 @@ def test_flush__full_buffer__sends_events_once( assert request_body body = json.loads(request_body) assert [e["event"] for e in body["events"]] == [f"event_{i}" for i in range(5)] - assert processor._buffer == [] def test_flush__full_buffer__drops_oldest_event( @@ -204,6 +279,7 @@ def test_flush__full_buffer__drops_oldest_event( # When processor.track_event(event="event_4") processor.flush() + processor.close(timeout_seconds=0) # Then assert len(mocked_responses.calls) == 4 @@ -211,38 +287,7 @@ def test_flush__full_buffer__drops_oldest_event( assert request_body body = json.loads(request_body) assert [e["event"] for e in body["events"]] == ["event_2", "event_3", "event_4"] - assert "Event buffer is full, dropping the oldest event_1 event" in ( - caplog.messages - ) - - -def test_flush__sends_events_to_events_api( - mocker: MockerFixture, - event_processor: EventProcessor, - mocked_responses: responses.RequestsMock, -) -> None: - # Given - post = mocker.spy(event_processor._session, "post") - mocked_responses.post("http://test_analytics/v1/events", status=202) - event_processor.track_event(event="purchase", identifier="user1") - - # When - event_processor.flush() - - # Then - assert len(mocked_responses.calls) == 1 - request = mocked_responses.calls[0].request - assert post.call_args.kwargs["timeout"] == 3 - assert request.headers["X-Environment-Key"] == "test_key" - assert request.headers["Content-Type"] == "application/json; charset=utf-8" - assert "flagsmith-python-sdk/" in request.headers["Flagsmith-SDK-User-Agent"] - request_body = request.body - assert request_body - body = json.loads(request_body) - assert "environment_key" not in body - assert len(body["events"]) == 1 - assert body["events"][0]["event"] == "purchase" - assert event_processor._buffer == [] + assert "Event buffer is full, dropped the oldest event" in caplog.messages def test_flush__no_events__sends_nothing( @@ -277,7 +322,6 @@ def test_flush__transient_failures__retried_with_backoff( # Then assert len(mocked_responses.calls) == 3 assert uniform.call_args_list == [mocker.call(0, 1), mocker.call(0, 2)] - assert event_processor._buffer == [] assert caplog.messages == [ "Failed to send 1 events, attempt 1 of 3: status 503", "Failed to send 1 events, attempt 2 of 3: reset", @@ -285,11 +329,15 @@ def test_flush__transient_failures__retried_with_backoff( @pytest.mark.parametrize( - "statuses", + "failures", [ pytest.param([503, 503, 503], id="unavailable"), pytest.param([500, 500, 503], id="unavailable-after-server-errors"), - pytest.param([429, 429, 429], id="throttled"), + pytest.param([429, 408, 504], id="throttled-and-timed-out"), + pytest.param( + [502, requests.ConnectionError("reset"), requests.Timeout("timed out")], + id="unreachable", + ), ], ) def test_flush__last_attempt_fails_transiently__kept_ahead_of_newer_events( @@ -297,12 +345,16 @@ def test_flush__last_attempt_fails_transiently__kept_ahead_of_newer_events( event_processor: EventProcessor, mocked_responses: responses.RequestsMock, caplog: pytest.LogCaptureFixture, - statuses: typing.List[int], + failures: typing.List[typing.Union[int, Exception]], ) -> None: # Given mocker.patch("flagsmith.analytics.random.uniform", return_value=0) - for status in (*statuses, 202): - mocked_responses.post("http://test_analytics/v1/events", status=status) + for failure in failures: + if isinstance(failure, Exception): + mocked_responses.post("http://test_analytics/v1/events", body=failure) + else: + mocked_responses.post("http://test_analytics/v1/events", status=failure) + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_event(event="add_to_cart") event_processor.flush() @@ -316,7 +368,7 @@ def test_flush__last_attempt_fails_transiently__kept_ahead_of_newer_events( assert request_body body = json.loads(request_body) assert [e["event"] for e in body["events"]] == ["add_to_cart", "checkout"] - assert "Keeping 1 events for the next flush" in caplog.messages + assert "Keeping 1 events for the next send" in caplog.messages @pytest.mark.parametrize( @@ -327,6 +379,11 @@ def test_flush__last_attempt_fails_transiently__kept_ahead_of_newer_events( "Events API rejected 1 events with 400, dropping them", id="rejected", ), + pytest.param( + [501], + "Events API rejected 1 events with 501, dropping them", + id="not-implemented", + ), pytest.param( [503, 503, 500], "Dropping 1 events the events API failed with 500", @@ -362,6 +419,52 @@ def test_flush__batch_rejected_or_last_failed_with_server_error__dropped( assert expected_log in caplog.messages +def test_flush__request_cannot_be_made__dropped_without_retrying( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocked_responses.post( + "http://test_analytics/v1/events", + body=requests.exceptions.InvalidHeader("invalid header"), + ) + event_processor.track_event(event="purchase") + + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 1 + assert caplog.messages == ["Dropping 1 events: invalid header"] + + +def test_flush__unexpected_error__returns_and_keeps_tracking( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocker.patch.object( + event_processor._session, + "post", + side_effect=[RuntimeError("unexpected"), mocker.DEFAULT], + wraps=event_processor._session.post, + ) + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="purchase") + event_processor.flush() + + # When + event_processor.track_event(event="checkout") + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 1 + assert caplog.messages == ["Dropping 1 events: failed to send"] + + @pytest.mark.parametrize("status", [401, 403]) def test_flush__environment_key_refused__stops_event_tracking( event_processor: EventProcessor, @@ -381,37 +484,14 @@ def test_flush__environment_key_refused__stops_event_tracking( # Then assert len(mocked_responses.calls) == 1 - assert event_processor._buffer == [] assert caplog.messages == [ f"Events API refused the environment key with {status}," " stopping event tracking", - "Not tracking checkout event: the events API refused the environment key.", + "Not tracking checkout event: event tracking has stopped.", ] -def test_stop__timer_flush_in_progress__no_further_timer_flushes( - mocked_responses: responses.RequestsMock, -) -> None: - # Given - processor = EventProcessor( - config=EventProcessorConfig( - events_api_url="http://test_analytics/", flush_interval_seconds=100 - ), - environment_key="test_key", - ) - processor.start() - timer = processor._timer - processor.stop() - - # When - processor._timer_flush() - - # Then - assert processor._timer is timer - assert len(mocked_responses.calls) == 0 - - -def test_start__flushes_on_timer( +def test_events__flush_interval_elapsed__sent_without_flush( mocked_responses: responses.RequestsMock, ) -> None: # Given @@ -422,45 +502,46 @@ def test_start__flushes_on_timer( environment_key="test_key", ) mocked_responses.post("http://test_analytics/v1/events", status=202) - processor.track_event(event="purchase") # When - processor.start() + processor.track_event(event="purchase") deadline = time.monotonic() + 5 while not mocked_responses.calls and time.monotonic() < deadline: time.sleep(0.01) - processor.stop() + processor.close(timeout_seconds=0) # Then assert len(mocked_responses.calls) == 1 - assert processor._buffer == [] -def test_stop__sends_buffered_events( +def test_close__sends_buffered_events_and_stops_event_tracking( + event_processor: EventProcessor, mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, ) -> None: # Given - processor = EventProcessor( - config=EventProcessorConfig( - events_api_url="http://test_analytics/", flush_interval_seconds=100 - ), - environment_key="test_key", - ) - processor.start() mocked_responses.post("http://test_analytics/v1/events", status=202) - processor.track_event(event="purchase") + event_processor.track_event(event="purchase") # When - processor.stop() + event_processor.close() + with caplog.at_level(logging.DEBUG, logger="flagsmith.analytics"): + event_processor.track_event(event="checkout") + event_processor.flush() + event_processor.close() # Then - assert processor._timer is not None - assert processor._timer.finished.is_set() assert len(mocked_responses.calls) == 1 - assert processor._buffer == [] + request_body = mocked_responses.calls[0].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == ["purchase"] + assert not event_processor._worker.is_alive() + assert caplog.messages == [ + "Not tracking checkout event: event tracking has stopped." + ] -def test_stop__events_api_not_answering__returns_within_timeout( +def test_close__events_api_not_answering__returns_within_timeout( event_processor: EventProcessor, mocked_responses: responses.RequestsMock, caplog: pytest.LogCaptureFixture, @@ -476,11 +557,10 @@ def test_stop__events_api_not_answering__returns_within_timeout( # When started = time.monotonic() - event_processor.stop(timeout_seconds=0.1) + event_processor.close(timeout_seconds=0.1) elapsed = time.monotonic() - started answer.set() - with event_processor._send_lock: - pass + event_processor._worker.join(5) # Then assert elapsed < 1 @@ -488,5 +568,31 @@ def test_stop__events_api_not_answering__returns_within_timeout( assert caplog.messages == [ "Gave up sending events after 0.1 seconds while closing", "Failed to send 1 events, attempt 1 of 3: status 503", - "Not retrying 1 events: the client is closed", + "Not sending 1 events: the client is closed", ] + + +def test_stop__events_api_not_answering__returns_without_waiting( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + answer = threading.Event() + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=lambda request: (answer.wait(5), (202, {}, ""))[1], + ) + event_processor.track_event(event="purchase") + + # When + started = time.monotonic() + event_processor.stop() + elapsed = time.monotonic() - started + answer.set() + event_processor._worker.join(5) + + # Then + assert elapsed < 1 + assert len(mocked_responses.calls) == 1 + assert not event_processor._worker.is_alive() diff --git a/tests/test_flagsmith.py b/tests/test_flagsmith.py index 9084211..b6879fc 100644 --- a/tests/test_flagsmith.py +++ b/tests/test_flagsmith.py @@ -1113,13 +1113,17 @@ def test_flush_events__events_tracked__sent_to_events_api( assert [e["event"] for e in body["events"]] == ["purchase"] -def test_flush_events__events_disabled__raises(api_key: str) -> None: +def test_flush_events__events_disabled__does_nothing( + api_key: str, caplog: pytest.LogCaptureFixture +) -> None: # Given flagsmith = Flagsmith(environment_key=api_key) - # When / Then - with pytest.raises(ValueError, match="Events must be enabled to flush events."): - flagsmith.flush_events() + # When + flagsmith.flush_events() + + # Then + assert caplog.messages == ["Not flushing events: events are not enabled."] def test_close__events_tracked__sent_within_timeout( @@ -1128,13 +1132,30 @@ def test_close__events_tracked__sent_within_timeout( # Given flagsmith = Flagsmith(environment_key=api_key, enable_events=True) assert flagsmith._event_processor - mock_stop = mocker.patch.object(flagsmith._event_processor, "stop") + mock_close = mocker.patch.object(flagsmith._event_processor, "close") # When flagsmith.close(timeout_seconds=2) # Then - mock_stop.assert_called_once_with(2) + mock_close.assert_called_once_with(2) + + +def test_del__events_tracked__stops_event_processor_without_waiting( + mocker: MockerFixture, api_key: str +) -> None: + # Given + flagsmith = Flagsmith(environment_key=api_key, enable_events=True) + assert flagsmith._event_processor + mock_stop = mocker.patch.object(flagsmith._event_processor, "stop") + mock_close = mocker.patch.object(flagsmith._event_processor, "close") + + # When + flagsmith.__del__() + + # Then + mock_stop.assert_called_once_with() + mock_close.assert_not_called() def test_close__local_evaluation__stops_polling( @@ -1153,6 +1174,28 @@ def test_close__local_evaluation__stops_polling( mock_stop.assert_called_once_with() +def test_close__realtime_updates__stops_streaming( + requests_session_response_ok: None, server_api_key: str, mocker: MockerFixture +) -> None: + # Given + mocker.patch( + "flagsmith.streaming_manager.requests.get", + side_effect=requests.exceptions.ReadTimeout(), + ) + flagsmith = Flagsmith( + environment_key=server_api_key, + enable_local_evaluation=True, + enable_realtime_updates=True, + ) + mock_stop = mocker.patch.object(flagsmith.event_stream_thread, "stop") + + # When + flagsmith.close() + + # Then + mock_stop.assert_called_once_with() + + def test_track_exposure_event_raises_without_config(api_key: str) -> None: flagsmith = Flagsmith(environment_key=api_key) with pytest.raises(ValueError, match="Events must be enabled"): From de59e681841bed30154ef69b5cefa0d69951daf4 Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Fri, 13 Nov 2026 13:05:48 +0000 Subject: [PATCH 6/9] fix: Send kept events on every flush, and re-track dropped exposures A flush right after a failed send returned without sending the kept batch, because it only waited for events tracked since. It now asks the worker for a send, unless a send already running covers every tracked event. An exposure dropped from a full buffer left its key behind, so the same exposure tracked again was skipped as a duplicate of an event that was no longer buffered. Each buffered event now carries its exposure key, which is forgotten when the event is dropped and restored when a failed batch is kept. Co-Authored-By: Claude Opus 5.5 --- flagsmith/analytics.py | 59 +++++++++++++----- tests/test_event_processor.py | 113 ++++++++++++++++++++++++++++++++++ 2 files changed, 155 insertions(+), 17 deletions(-) diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index c5256bf..6387ac4 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -109,6 +109,11 @@ class EventProcessorConfig: flush_interval_seconds: float = 10.0 +class _BufferedEvent(typing.NamedTuple): + payload: str + exposure_key: typing.Optional[ExposureKey] + + class _DeliveryOutcome(enum.Enum): DELIVERED = "delivered" DROPPED = "dropped" @@ -145,13 +150,15 @@ def __init__( self._session = requests.Session() self._condition = threading.Condition() - self._buffer: typing.Deque[str] = deque() + self._buffer: typing.Deque[_BufferedEvent] = deque() self._buffered_exposure_keys: typing.Set[ExposureKey] = set() - # Events are numbered as tracked: a flush waits until the worker has - # sent every event up to the number tracked when it was called. self._tracked_count = 0 - self._flush_requested_count = 0 self._sent_count = 0 + # Events tracked up to this count are being sent, if a send is running. + self._sending_count: typing.Optional[int] = None + self._flush_requested = False + self._sends_started = 0 + self._sends_completed = 0 self._refused = False self._closed = False # Set once `close` returns: the worker starts no more requests. @@ -227,6 +234,7 @@ def _buffer_event( "Not tracking %s event: event tracking has stopped.", event ) return + exposure_key = None if event == FLAG_EXPOSURE_EVENT: # An exposure is defined by who saw which variant of which # feature; equal exposures within one flush interval add no @@ -245,7 +253,7 @@ def _buffer_event( ) return self._buffered_exposure_keys.add(exposure_key) - self._buffer.append(serialised_event) + self._buffer.append(_BufferedEvent(serialised_event, exposure_key)) self._tracked_count += 1 self._drop_oldest_events() if len(self._buffer) >= self._max_buffer: @@ -253,18 +261,25 @@ def _buffer_event( def flush(self) -> None: """ - Sends the events tracked so far, and returns once each of them has - been delivered, dropped, or kept for the next send. Does nothing once - closed. + Sends the buffered events, including any kept from a failed send, and + returns once each of them has been delivered, dropped, or kept for the + next send. If a send already running covers every event tracked so + far, waits for it instead. Does nothing once closed. """ with self._condition: if self._closed: return - flush_count = self._tracked_count - self._flush_requested_count = max(self._flush_requested_count, flush_count) - self._condition.notify_all() + if ( + self._sending_count is not None + and self._sending_count >= self._tracked_count + ): + send = self._sends_started + else: + self._flush_requested = True + send = self._sends_started + 1 + self._condition.notify_all() self._condition.wait_for( - lambda: self._sent_count >= flush_count or not self._worker.is_alive() + lambda: self._sends_completed >= send or not self._worker.is_alive() ) def close( @@ -304,7 +319,7 @@ def _run(self) -> None: with self._condition: while not ( self._closed - or self._flush_requested_count > self._sent_count + or self._flush_requested or ( len(self._buffer) >= self._max_buffer and self._tracked_count > self._sent_count @@ -316,21 +331,30 @@ def _run(self) -> None: events = list(self._buffer) self._buffer.clear() self._buffered_exposure_keys.clear() - sent_count = self._tracked_count + self._flush_requested = False + self._sends_started += 1 + sending_count = self._sending_count = self._tracked_count next_send_at = time.monotonic() + self._flush_interval_seconds try: outcome = ( - self._deliver(events) if events else _DeliveryOutcome.DELIVERED + self._deliver([event.payload for event in events]) + if events + else _DeliveryOutcome.DELIVERED ) except Exception: logger.exception("Dropping %d events: failed to send", len(events)) outcome = _DeliveryOutcome.DROPPED with self._condition: - self._sent_count = sent_count + self._sent_count = sending_count + self._sending_count = None + self._sends_completed += 1 if outcome is _DeliveryOutcome.KEPT: self._buffer.extendleft(reversed(events)) + self._buffered_exposure_keys.update( + event.exposure_key for event in events if event.exposure_key + ) self._drop_oldest_events() elif outcome is _DeliveryOutcome.REFUSED: self._refused = True @@ -343,7 +367,8 @@ def _run(self) -> None: def _drop_oldest_events(self) -> None: while len(self._buffer) > self._max_buffer: - self._buffer.popleft() + dropped = self._buffer.popleft() + self._buffered_exposure_keys.discard(dropped.exposure_key) logger.warning("Event buffer is full, dropped the oldest event") def _deliver(self, events: typing.List[str]) -> _DeliveryOutcome: diff --git a/tests/test_event_processor.py b/tests/test_event_processor.py index 458f3b7..c64a863 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -371,6 +371,119 @@ def test_flush__last_attempt_fails_transiently__kept_ahead_of_newer_events( assert "Keeping 1 events for the next send" in caplog.messages +def test_flush__kept_batch_and_no_new_events__kept_batch_sent( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + for status in (503, 503, 503, 202): + mocked_responses.post("http://test_analytics/v1/events", status=status) + event_processor.track_event(event="add_to_cart") + event_processor.flush() + + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 4 + request_body = mocked_responses.calls[3].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == ["add_to_cart"] + + +def test_flush__send_in_flight_covers_tracked_events__waits_without_resending( + mocker: MockerFixture, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=2 + ), + environment_key="test_key", + ) + requested = threading.Event() + answer = threading.Event() + + def answer_when_released( + request: requests.PreparedRequest, + ) -> typing.Tuple[int, typing.Dict[str, str], str]: + requested.set() + answer.wait(5) + return 503, {}, "" + + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=answer_when_released, + ) + processor.track_event(event="event_1") + processor.track_event(event="event_2") + requested.wait(5) + threading.Timer(0.1, answer.set).start() + + # When + processor.flush() + + # Then + assert len(mocked_responses.calls) == 3 + processor.close(timeout_seconds=0) + + +def test_track_exposure_event__equal_exposure_dropped_from_full_buffer__sent( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=2 + ), + environment_key="test_key", + ) + requested = threading.Event() + answer = threading.Event() + + def answer_when_released( + request: requests.PreparedRequest, + ) -> typing.Tuple[int, typing.Dict[str, str], str]: + requested.set() + answer.wait(5) + return 202, {}, "" + + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=answer_when_released, + ) + processor.track_event(event="event_1") + processor.track_event(event="event_2") + requested.wait(5) + processor.track_exposure_event( + feature_name="checkout", identifier="user1", value="variant_b" + ) + processor.track_event(event="event_3") + processor.track_event(event="event_4") + + # When + processor.track_exposure_event( + feature_name="checkout", identifier="user1", value="variant_b" + ) + answer.set() + processor.flush() + processor.close(timeout_seconds=0) + + # Then + sent_events = [] + for call in mocked_responses.calls: + request_body = call.request.body + assert request_body + sent_events += [e["event"] for e in json.loads(request_body)["events"]] + assert sent_events == ["event_1", "event_2", "event_4", FLAG_EXPOSURE_EVENT] + + @pytest.mark.parametrize( "statuses, expected_log", [ From 9f27ea9b3b7764162038a26a4f63b4c883a0e1bc Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Fri, 13 Nov 2026 15:02:27 +0000 Subject: [PATCH 7/9] fix: Pace event retries by the flush interval during an outage While the events API failed, the buffer stayed full of kept events, so any event tracked during a retry cycle started another full cycle as soon as it ended, ignoring the flush interval. The worker now wakes for a full buffer only once a buffer's worth of events has been tracked since the last send. `max_buffer_items` below 1 dropped every event as soon as it was buffered. It is now raised to 1, with a warning, so that each event is sent straight away. Tests now wait for a closed processor's worker to finish, so it cannot send its kept events into the next test. Co-Authored-By: Claude Opus 5.5 --- flagsmith/analytics.py | 21 +++++++---- tests/conftest.py | 2 ++ tests/test_event_processor.py | 66 +++++++++++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 6 deletions(-) diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index 6387ac4..f187c5f 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -145,7 +145,12 @@ def __init__( url = f"{url}/" self._batch_endpoint = f"{url}v1/events" self._environment_key = environment_key - self._max_buffer = config.max_buffer_items + if config.max_buffer_items < 1: + logger.warning( + "max_buffer_items must be at least 1, not %d: using 1", + config.max_buffer_items, + ) + self._max_buffer = max(config.max_buffer_items, 1) self._flush_interval_seconds = config.flush_interval_seconds self._session = requests.Session() @@ -256,7 +261,7 @@ def _buffer_event( self._buffer.append(_BufferedEvent(serialised_event, exposure_key)) self._tracked_count += 1 self._drop_oldest_events() - if len(self._buffer) >= self._max_buffer: + if self._has_full_buffer_of_new_events(): self._condition.notify_all() def flush(self) -> None: @@ -320,10 +325,7 @@ def _run(self) -> None: while not ( self._closed or self._flush_requested - or ( - len(self._buffer) >= self._max_buffer - and self._tracked_count > self._sent_count - ) + or self._has_full_buffer_of_new_events() or time.monotonic() >= next_send_at ): self._condition.wait(next_send_at - time.monotonic()) @@ -365,6 +367,13 @@ def _run(self) -> None: self._session.close() return + def _has_full_buffer_of_new_events(self) -> bool: + """ + Whether a buffer's worth of events was tracked since the last send. A + buffer full of events kept from failed sends waits for the timer. + """ + return self._tracked_count - self._sent_count >= self._max_buffer + def _drop_oldest_events(self) -> None: while len(self._buffer) > self._max_buffer: dropped = self._buffer.popleft() diff --git a/tests/conftest.py b/tests/conftest.py index 5849a03..9bd8b58 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -52,6 +52,7 @@ def tracking_init(self: Flagsmith, *args: typing.Any, **kwargs: typing.Any) -> N stream.join(timeout=5) if flagsmith._event_processor: flagsmith._event_processor.close(timeout_seconds=0) + flagsmith._event_processor._worker.join(5) @pytest.fixture() @@ -76,6 +77,7 @@ def event_processor( ) yield processor processor.close(timeout_seconds=0) + processor._worker.join(5) @pytest.fixture(scope="session") diff --git a/tests/test_event_processor.py b/tests/test_event_processor.py index c64a863..fe1ad4f 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -248,6 +248,7 @@ def test_track_event__full_buffer__sent_without_flush( time.sleep(0.01) processor.flush() processor.close(timeout_seconds=0) + processor._worker.join(5) # Then assert len(mocked_responses.calls) == 1 @@ -257,6 +258,67 @@ def test_track_event__full_buffer__sent_without_flush( assert [e["event"] for e in body["events"]] == [f"event_{i}" for i in range(5)] +def test_track_event__buffer_full_of_kept_events__waits_for_flush_interval( + mocker: MockerFixture, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=2 + ), + environment_key="test_key", + ) + mocked_responses.post("http://test_analytics/v1/events", status=503) + processor.track_event(event="event_1") + processor.track_event(event="event_2") + processor.flush() + + # When + processor.track_event(event="event_3") + time.sleep(0.3) + + # Then + assert len(mocked_responses.calls) == 3 + processor.close(timeout_seconds=0) + processor._worker.join(5) + + +@pytest.mark.parametrize("max_buffer_items", [0, -1]) +def test_track_event__max_buffer_items_below_one__event_sent_without_flush( + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + max_buffer_items: int, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", + max_buffer_items=max_buffer_items, + ), + environment_key="test_key", + ) + mocked_responses.post("http://test_analytics/v1/events", status=202) + + # When + processor.track_event(event="purchase") + deadline = time.monotonic() + 5 + while not mocked_responses.calls and time.monotonic() < deadline: + time.sleep(0.01) + processor.close(timeout_seconds=0) + processor._worker.join(5) + + # Then + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == ["purchase"] + assert caplog.messages[0] == ( + f"max_buffer_items must be at least 1, not {max_buffer_items}: using 1" + ) + + def test_flush__full_buffer__drops_oldest_event( mocker: MockerFixture, mocked_responses: responses.RequestsMock, @@ -280,6 +342,7 @@ def test_flush__full_buffer__drops_oldest_event( processor.track_event(event="event_4") processor.flush() processor.close(timeout_seconds=0) + processor._worker.join(5) # Then assert len(mocked_responses.calls) == 4 @@ -431,6 +494,7 @@ def answer_when_released( # Then assert len(mocked_responses.calls) == 3 processor.close(timeout_seconds=0) + processor._worker.join(5) def test_track_exposure_event__equal_exposure_dropped_from_full_buffer__sent( @@ -474,6 +538,7 @@ def answer_when_released( answer.set() processor.flush() processor.close(timeout_seconds=0) + processor._worker.join(5) # Then sent_events = [] @@ -622,6 +687,7 @@ def test_events__flush_interval_elapsed__sent_without_flush( while not mocked_responses.calls and time.monotonic() < deadline: time.sleep(0.01) processor.close(timeout_seconds=0) + processor._worker.join(5) # Then assert len(mocked_responses.calls) == 1 From 0f1eafbde7d87da7eb597499697b8a6546837414 Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Fri, 13 Nov 2026 16:14:06 +0000 Subject: [PATCH 8/9] fix: Send events from forked processes and garbage-collected clients The event worker was started once in the constructor, so a forked child (gunicorn --preload, uWSGI, Celery prefork) tracked and flushed events without ever sending them. A fork hook now starts a new worker in the child, with an empty buffer, as the parent sends the events it buffered. `__del__` stops the processor without waiting, but `stop()` also unregistered the `atexit` hook, so a client collected before exit lost its buffered events when the interpreter killed the worker. The hook now stays registered until `close()` returns, and `close()` waits for the worker after a `stop()` too. Dropping the oldest events from a full buffer logged a warning for each one, under the lock and on the caller's thread. The worker now logs the number dropped once per send. Co-Authored-By: Claude Opus 5.5 --- flagsmith/analytics.py | 53 +++++++++++----- tests/test_event_processor.py | 112 +++++++++++++++++++++++++++++++++- 2 files changed, 150 insertions(+), 15 deletions(-) diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index f187c5f..13faea6 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -2,10 +2,12 @@ import enum import json import logging +import os import random import threading import time import typing +import weakref from collections import deque from dataclasses import dataclass from datetime import datetime @@ -152,11 +154,22 @@ def __init__( ) self._max_buffer = max(config.max_buffer_items, 1) self._flush_interval_seconds = config.flush_interval_seconds + self._refused = False + self._closed = False + self._start_worker() + _processors.add(self) + atexit.register(self.close) + def _start_worker(self) -> None: + """ + Starts the worker, with empty state. Also runs in a forked child, where + the parent's worker doesn't exist and the parent sends its own events. + """ self._session = requests.Session() self._condition = threading.Condition() self._buffer: typing.Deque[_BufferedEvent] = deque() self._buffered_exposure_keys: typing.Set[ExposureKey] = set() + self._dropped_count = 0 self._tracked_count = 0 self._sent_count = 0 # Events tracked up to this count are being sent, if a send is running. @@ -164,16 +177,12 @@ def __init__( self._flush_requested = False self._sends_started = 0 self._sends_completed = 0 - self._refused = False - self._closed = False # Set once `close` returns: the worker starts no more requests. self._abandoned = threading.Event() - self._worker = threading.Thread( target=self._run, name="flagsmith-event-processor", daemon=True ) self._worker.start() - atexit.register(self.close) def track_event( self, @@ -292,31 +301,29 @@ def close( ) -> None: """ Stops event tracking, and sends the buffered events, waiting for at - most `timeout_seconds`. No request starts once this returns. Calling - it again does nothing. + most `timeout_seconds`, also after `stop`. No request starts once this + returns. Calling it again does nothing. """ - if not self.stop(): + self.stop() + if self._abandoned.is_set(): return self._worker.join(timeout_seconds) self._abandoned.set() + atexit.unregister(self.close) if self._worker.is_alive(): logger.warning( "Gave up sending events after %s seconds while closing", timeout_seconds, ) - def stop(self) -> bool: + def stop(self) -> None: """ Stops event tracking, and has the worker send the buffered events, - without waiting for it. Returns whether this stopped event tracking. + without waiting for it. `close` still runs at exit, to wait for it. """ with self._condition: - if self._closed: - return False self._closed = True self._condition.notify_all() - atexit.unregister(self.close) - return True def _run(self) -> None: next_send_at = time.monotonic() + self._flush_interval_seconds @@ -336,7 +343,13 @@ def _run(self) -> None: self._flush_requested = False self._sends_started += 1 sending_count = self._sending_count = self._tracked_count + dropped_count, self._dropped_count = self._dropped_count, 0 next_send_at = time.monotonic() + self._flush_interval_seconds + if dropped_count: + logger.warning( + "Event buffer was full: dropped the %d oldest events", + dropped_count, + ) try: outcome = ( @@ -378,7 +391,7 @@ def _drop_oldest_events(self) -> None: while len(self._buffer) > self._max_buffer: dropped = self._buffer.popleft() self._buffered_exposure_keys.discard(dropped.exposure_key) - logger.warning("Event buffer is full, dropped the oldest event") + self._dropped_count += 1 def _deliver(self, events: typing.List[str]) -> _DeliveryOutcome: payload = f'{{"events": [{", ".join(events)}]}}' @@ -452,3 +465,15 @@ def _outcome_of(status: int, event_count: int) -> typing.Optional[_DeliveryOutco status, ) return _DeliveryOutcome.DROPPED + + +_processors: "weakref.WeakSet[EventProcessor]" = weakref.WeakSet() + + +def _restart_workers_after_fork() -> None: + for processor in list(_processors): + processor._start_worker() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_restart_workers_after_fork) diff --git a/tests/test_event_processor.py b/tests/test_event_processor.py index fe1ad4f..8f65af7 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -1,5 +1,6 @@ import json import logging +import os import threading import time import typing @@ -350,7 +351,7 @@ def test_flush__full_buffer__drops_oldest_event( assert request_body body = json.loads(request_body) assert [e["event"] for e in body["events"]] == ["event_2", "event_3", "event_4"] - assert "Event buffer is full, dropped the oldest event" in caplog.messages + assert "Event buffer was full: dropped the 1 oldest events" in caplog.messages def test_flush__no_events__sends_nothing( @@ -775,3 +776,112 @@ def test_stop__events_api_not_answering__returns_without_waiting( assert elapsed < 1 assert len(mocked_responses.calls) == 1 assert not event_processor._worker.is_alive() + + +def test_close__after_stop__waits_for_buffered_events( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + answer = threading.Event() + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=lambda request: (answer.wait(5), (202, {}, ""))[1], + ) + event_processor.track_event(event="purchase") + event_processor.stop() + threading.Timer(0.2, answer.set).start() + + # When + event_processor.close(timeout_seconds=5) + + # Then + assert len(mocked_responses.calls) == 1 + assert not event_processor._worker.is_alive() + + +def test_track_event__full_buffer_while_sending__drops_logged_once( + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=2 + ), + environment_key="test_key", + ) + requested = threading.Event() + answer = threading.Event() + + def answer_when_released( + request: requests.PreparedRequest, + ) -> typing.Tuple[int, typing.Dict[str, str], str]: + requested.set() + answer.wait(5) + return 202, {}, "" + + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=answer_when_released, + ) + processor.track_event(event="event_1") + processor.track_event(event="event_2") + requested.wait(5) + + # When + for i in range(3, 8): + processor.track_event(event=f"event_{i}") + answer.set() + processor.flush() + processor.close(timeout_seconds=0) + processor._worker.join(5) + + # Then + assert [m for m in caplog.messages if m.startswith("Event buffer")] == [ + "Event buffer was full: dropped the 3 oldest events" + ] + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") +@pytest.mark.filterwarnings( + "ignore:This process .* is multi-threaded:DeprecationWarning" +) +def test_track_event__forked_child__each_process_sends_its_own_events( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="tracked_before_fork") + + # When + pid = os.fork() + if pid == 0: + exit_code = 1 + try: + event_processor.track_event(event="tracked_in_child") + event_processor.flush() + child_sent_events = [] + for call in mocked_responses.calls: + request_body = call.request.body + assert request_body + child_sent_events += [ + e["event"] for e in json.loads(request_body)["events"] + ] + exit_code = 0 if child_sent_events == ["tracked_in_child"] else 2 + finally: + os._exit(exit_code) + _, wait_status = os.waitpid(pid, 0) + event_processor.flush() + + # Then + assert os.waitstatus_to_exitcode(wait_status) == 0 + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == [ + "tracked_before_fork" + ] From 2aa8e6d5808b202014ddf132c86db1dd179cbf5e Mon Sep 17 00:00:00 2001 From: "flagsmithsmith[bot]" <346993984+flagsmithsmith[bot]@users.noreply.github.com> Date: Fri, 13 Nov 2026 19:34:31 +0000 Subject: [PATCH 9/9] fix: Close event processors from one exit hook Each processor registered its bound `close` with `atexit`, which kept the processor and its session alive until `close()` was called. A client garbage-collected without `close()` leaked its processor for the life of the process. One module-level hook now closes every processor still in `_processors`. The worker keeps its processor alive while it runs, so a processor stopped by `__del__` is still waited for at exit while its events are being sent, and is collected once they are. Co-Authored-By: Claude Opus 5.5 --- flagsmith/analytics.py | 15 ++++++++-- tests/test_event_processor.py | 55 +++++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 3 deletions(-) diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index 13faea6..7e47400 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -158,7 +158,6 @@ def __init__( self._closed = False self._start_worker() _processors.add(self) - atexit.register(self.close) def _start_worker(self) -> None: """ @@ -309,7 +308,6 @@ def close( return self._worker.join(timeout_seconds) self._abandoned.set() - atexit.unregister(self.close) if self._worker.is_alive(): logger.warning( "Gave up sending events after %s seconds while closing", @@ -319,7 +317,8 @@ def close( def stop(self) -> None: """ Stops event tracking, and has the worker send the buffered events, - without waiting for it. `close` still runs at exit, to wait for it. + without waiting for it. `close` still runs at exit, to wait for it, as + long as the worker is sending. """ with self._condition: self._closed = True @@ -475,5 +474,15 @@ def _restart_workers_after_fork() -> None: processor._start_worker() +def _close_processors_at_exit() -> None: + # One hook for every processor: a hook per processor would keep each one + # alive until exit. The worker keeps its processor alive while it runs, so + # a stopped processor is still closed here if its events are being sent. + for processor in list(_processors): + processor.close() + + +atexit.register(_close_processors_at_exit) + if hasattr(os, "register_at_fork"): os.register_at_fork(after_in_child=_restart_workers_after_fork) diff --git a/tests/test_event_processor.py b/tests/test_event_processor.py index 8f65af7..2667839 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -1,9 +1,11 @@ +import gc import json import logging import os import threading import time import typing +import weakref from datetime import datetime import pytest @@ -15,6 +17,7 @@ FLAG_EXPOSURE_EVENT, EventProcessor, EventProcessorConfig, + _close_processors_at_exit, ) @@ -885,3 +888,55 @@ def test_track_event__forked_child__each_process_sends_its_own_events( assert [e["event"] for e in json.loads(request_body)["events"]] == [ "tracked_before_fork" ] + + +def test_stop__processor_no_longer_referenced__garbage_collected( + event_processor_config: EventProcessorConfig, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) + processor = EventProcessor( + config=event_processor_config, environment_key="test_key" + ) + processor.track_event(event="purchase") + worker = processor._worker + collected = weakref.ref(processor) + + # When + processor.stop() + del processor + worker.join(5) + gc.collect() + + # Then + assert len(mocked_responses.calls) == 1 + assert collected() is None + + +def test_close_processors_at_exit__stopped_processor_sending__waits_for_it( + event_processor_config: EventProcessorConfig, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + answer = threading.Event() + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=lambda request: (answer.wait(5), (202, {}, ""))[1], + ) + processor = EventProcessor( + config=event_processor_config, environment_key="test_key" + ) + processor.track_event(event="purchase") + worker = processor._worker + processor.stop() + del processor + threading.Timer(0.2, answer.set).start() + + # When + _close_processors_at_exit() + + # Then + assert len(mocked_responses.calls) == 1 + assert not worker.is_alive()