Repository navigation
feat: Retry event delivery and flush on close - #283
flagsmithsmith[bot] wants to merge 8 commits into
Conversation
Its entry in Flagsmith/sdk-spec's compatibility matrix at v0.9.0: `core events local-evaluation`.
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 <[email protected]> (cherry picked from commit d019e76)
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 <[email protected]> (cherry picked from commit ca330ed)
flagsmithsmith
Bot
added
the
sdk-spec
|
Important Review skippedBot user detected. To trigger a single review, invoke the
 |
Zaimwa9
left a comment
Zaimwa9
left a comment
There was a problem hiding this comment.
I started a review but will stop here, grab the first comments to understand the issues in the implementation.
Overall spec passes, but the event processor's structure is not good and needs a rework, not patches: threads are created three ways, with a pile of locks and flags. Use one long-lived worker thread, like EnvironmentDataPollingManager. __del__ must not block, flush_events() must never raise, and close() must be terminal.
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 <[email protected]>
|
Replying to the review:
Reworked rather than patched, in cd3aa8c.
It now calls
It logs
Once it is called, tracking and flushing are no-ops. It waits at most Checks: 398 unit tests pass, with full coverage of the changed lines. |
|
@themis-blindfold review |
|
Review failed after 2 attempt(s) (OutputError). Check the worker logs. |
This asked the themis-blindfold bot for a review, and its run failed before it produced one ("Review failed after 2 attempt(s) (OutputError)"). That failure is in the bot's worker, not in this repository, so there is no review feedback to address and I haven't changed any code. I haven't re-triggered it, so that you decide whether to run it again. The branch is unchanged at cd3aa8c, and every check is green, including |
|
@themis-blindfold review |
|
Review failed after 2 attempt(s) (OutputError). Check the worker logs. |
The re-run failed the same way ("Review failed after 2 attempt(s) (OutputError)"), so there is still no review feedback to address, and the code is unchanged at cd3aa8c. Both failures happened in the bot's worker, and its logs aren't visible from this pull request, so I can't tell why it fails here. All checks are green, including |
|
@themis-blindfold review |
|
Review failed after 2 attempt(s) (OutputError). Check the worker logs. |
The third run also failed before producing a review ("Review failed after 2 attempt(s) (OutputError)"), so there is nothing new to address. The code is unchanged at cd3aa8c, and all checks are green. The bot fails the same way every time on this pull request, so the cause is likely on its side; its worker logs would show why. |
|
@themis-blindfold review |
![themis-blindfold[bot]](https://avatars.githubusercontent.com/in/4278417?s=60&v=4){kind=small}
| "X-Environment-Key": self._environment_key, | ||
| "Flagsmith-SDK-User-Agent": DEFAULT_USER_AGENT, | ||
| }, | ||
| flush_count = self._tracked_count |
There was a problem hiding this comment.
🟠 Major · 🏗️ Heavy lift
Flush retained events even when no new events were tracked
Observed: after three 503 responses, the batch is retained but _sent_count already equals _tracked_count; a second flush() returns without another request. This violates the requirement to retry the retained batch on the next flush, so applications cannot explicitly drain it until a new event arrives or the timer fires. Track flush requests independently of the tracked-event count and wait for the worker to process the retained batch.
There was a problem hiding this comment.
Right, fixed in de59e68. flush() used to wait only for events tracked since the last send, so a flush right after a failed send returned without sending the kept batch. It now asks the worker for a send of the whole buffer, kept events included. The exception is a send already running that covers every tracked event: then it waits for that send instead. That exception keeps the spec's "A full buffer drops its oldest event" passing, where a flush called while the full-buffer send is in flight must not send the batch again.
Tests: test_flush__kept_batch_and_no_new_events__kept_batch_sent (fails before the fix) and test_flush__send_in_flight_covers_tracked_events__waits_without_resending.
There was a problem hiding this comment.
Fixed: a subsequent flush requests another send of the retained batch (flagsmith/analytics.py:278), with a regression test covering three 503 responses (tests/test_event_processor.py:374). Resolving this thread.
|
|
||
| def _drop_oldest_events(self) -> None: | ||
| while len(self._buffer) > self._max_buffer: | ||
| self._buffer.popleft() |
There was a problem hiding this comment.
🟠 Major · 🏗️ Heavy lift
Remove deduplication keys when exposures are evicted
Observed: with a one-item buffer, tracking an exposure, then a custom event, then the same exposure leaves only the custom event buffered: eviction removes the exposure but leaves its key in _buffered_exposure_keys. The later exposure is discarded as a duplicate even though no copy remains to send. Store enough information with each buffered event to remove its exposure key on eviction, and keep the keys consistent when retained batches are requeued.
There was a problem hiding this comment.
Right, fixed in de59e68. Each buffered event now carries its exposure key. Dropping the oldest event forgets its key, and a kept batch puts its keys back when it is re-queued, so the keys always match the exposures in the buffer.
Test: test_track_exposure_event__equal_exposure_dropped_from_full_buffer__sent, which follows your scenario and fails before the fix.
There was a problem hiding this comment.
The eviction case is fixed, but requeuing can still leave keys inconsistent. If an exposure is tracked again while its earlier batch is in flight, a kept batch can contain the same key as the buffer. When the older copy is evicted, flagsmith/analytics.py:371 removes the shared key even though the newer copy remains, so another duplicate is accepted. Account for duplicate keys when requeuing and evicting (flagsmith/analytics.py:353).
| 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( |
There was a problem hiding this comment.
🟠 Major · ⚡ Quick win
Drop a batch immediately on HTTP 500
Observed: 500 is in the retry set, so the worker retries it and can retain the batch if a later attempt fails differently. The linked delivery requirements specify dropping the batch on any 500 without retrying. Predicted: repeated requests could duplicate events if the server processed the first request before returning 500. Remove 500 from the retry set and update the retry tests to require an immediate drop.
There was a problem hiding this comment.
Not changed: the spec this PR implements requires retrying a 500. The issue's first summary did say "Drop the batch on any other status, including 500", but the spec was later changed on purpose (sdk-spec commit "Retry a 500, then drop the batch instead of keeping it"). features/event_delivery.feature at v0.9.0 now has:
- "A batch whose last attempt gets a server error is dropped": 503, 503, 500, so the batch is dropped after the 500.
- "A batch whose last attempt fails otherwise is kept, after server errors": the SDK must send 500, 500, 503, then 202 for the same batch.
Dropping on the first 500 would fail the second scenario, and the human review here also asked for 500 in the retry set. The duplicate-delivery risk you raise is real, though, so if the events API cannot deduplicate, it would be worth raising in sdk-spec.
There was a problem hiding this comment.
You're right; I relied on the outdated issue summary. I withdraw the finding. The checkout tests both sequences in tests/test_event_processor.py:335 and tests/test_event_processor.py:501.
⚖️ Themis review: 🟠 Fix before mergeThe single-worker delivery path adds bounded shutdown and retries, but retained batches cannot be retried by a subsequent explicit flush, and buffer eviction can suppress an exposure that is no longer buffered. The 500 response policy also differs from the linked delivery requirements. The recorded CI checks passed.
🟠 Majors
⚖️ Acknowledged
📝 Walkthrough
🧪 How to verify
Product take: Reliable event delivery is a solid improvement for experimentation analytics; missed or repeated delivery can affect the data users rely on. 🧭 Assumptions & unverified claimsNo unverified assumptions or claims. May the next flush actually flush · reviewed at cd3aa8c |
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 <[email protected]>
This run produced a review (🟠 Fix before merge, at cd3aa8c). Replies to its three majors, also answered inline:
Fixed in de59e68. A flush now always sends the buffer, kept events included, unless a send already running covers every tracked event.
Fixed in de59e68. An exposure's key is forgotten when the event is dropped from a full buffer, and restored when a failed batch is kept.
Not changed. The spec at v0.9.0 requires retrying a 500: "A batch whose last attempt fails otherwise is kept, after server errors" expects 500, 500, 503, then 202 for the same batch. Only a 500 on the last attempt drops the batch. The text the bot quotes is from the issue's earlier summary, which the spec later changed on purpose. Details are in the inline reply. Checks: 401 unit tests pass, stable over 40 repeated runs, and the new regression tests fail without the fix. |
Zaimwa9
left a comment
Zaimwa9
left a comment
There was a problem hiding this comment.
Requesting changes. During an events API outage the full-buffer wake restarts the retry cycle back to back instead of waiting for the flush interval, and max_buffer_items=0 now drops every event (inline).
| len(self._buffer) >= self._max_buffer | ||
| and self._tracked_count > self._sent_count |
There was a problem hiding this comment.
During an outage the buffer stays full of kept events, so any event tracked during a retry cycle wakes the worker for another full cycle as soon as the current one ends, ignoring the flush interval (in a scaled-down 503 outage: 62 requests where timer pacing gives about 22). Wake on a full buffer only once a full buffer of new events has arrived, e.g. self._tracked_count - self._sent_count >= self._max_buffer.
There was a problem hiding this comment.
Agreed, fixed in 9f27ea9 with the rule you suggested. Both the wake condition and the notify in _buffer_event now use _has_full_buffer_of_new_events(), which is self._tracked_count - self._sent_count >= self._max_buffer. A buffer full of kept events plus a few new ones now waits for the flush interval (or an explicit flush) instead of starting another retry cycle.
The spec's "A full buffer drops its oldest event" still passes. The first 1000 events are a full buffer of new events and trigger a send. The 1001st doesn't trigger one, and the scenario's explicit flush sends it. I re-ran spec v0.9.0 locally: 24 passed.
Test: test_track_event__buffer_full_of_kept_events__waits_for_flush_interval uses a buffer of 2 and an events API that always answers 503. After one failed cycle, tracking a third event makes no request within 0.3 s. It fails before the fix.
| return | ||
|
|
||
| def _drop_oldest_events(self) -> None: | ||
| while len(self._buffer) > self._max_buffer: |
There was a problem hiding this comment.
With max_buffer_items=0 every tracked event is dropped here as soon as it is buffered, whereas on main each event was sent straight away. Clamp values below 1 in __init__.
There was a problem hiding this comment.
Agreed, fixed in 9f27ea9. __init__ now raises max_buffer_items below 1 to 1 and logs max_buffer_items must be at least 1, not <n>: using 1. With a buffer of 1, each tracked event is a full buffer of new events, so the worker sends it straight away, as main did.
Test: test_track_event__max_buffer_items_below_one__event_sent_without_flush, for 0 and -1. It fails before the fix.
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 <[email protected]>
Both fixed in 9f27ea9, with details inline:
Separately, the new tests exposed a flaky teardown: about 1 run in 100 failed. Tests closed their processors with Checks: 404 unit tests pass in 60 of 60 repeated runs, and the new tests fail without the fix. |
Zaimwa9
left a comment
Zaimwa9
left a comment
There was a problem hiding this comment.
Requesting changes. The single-worker design is good and the earlier points are fixed, but it loses events silently in two cases that work on main: in a forked child, and when a client is garbage-collected before the process exits.
| self._worker = threading.Thread( | ||
| target=self._run, name="flagsmith-event-processor", daemon=True | ||
| ) | ||
| self._worker.start() |
There was a problem hiding this comment.
The worker is started once in the constructor, so it does not exist in a forked child (gunicorn --preload, uWSGI, Celery prefork): tracking and flush succeed there but nothing is ever sent. Start the worker lazily with a PID check, or rebuild it with os.register_at_fork(after_in_child=...).
There was a problem hiding this comment.
Right, fixed in 0f1eafb with os.register_at_fork(after_in_child=...) (guarded with hasattr, since Windows has no fork). Live processors are tracked in a module-level weakref.WeakSet, and in the child each one gets a fresh worker. It also gets a fresh condition, which a parent thread may have held at the moment of the fork, and a fresh requests.Session, so it doesn't share the parent's pooled sockets. The child starts with an empty buffer: the parent's worker still sends the events the parent buffered, so a child resending its copy would duplicate them. Closed and refused state carry over.
I chose the fork hook over a lazy start with a PID check, because the hook keeps the PID check off the tracking path and also replaces the lock.
Test: test_track_event__forked_child__each_process_sends_its_own_events forks, then tracks and flushes in the child, which must send only its own event. The parent must then send only the event tracked before the fork. It fails before the fix. Coverage doesn't count the two lines of the hook, because they run only in the child, which exits with os._exit.
The same limitation exists on main for EnvironmentDataPollingManager: it is started in the constructor, so local evaluation in a forked child never refreshes its environment. I left that alone, as it's outside this PR. I can do it in a follow-up if you'd like.
| return False | ||
| self._closed = True | ||
| self._condition.notify_all() | ||
| atexit.unregister(self.close) |
There was a problem hiding this comment.
__del__ calls stop(), which unregisters the atexit hook and does not wait, so a client that goes out of scope before exit has its daemon worker killed mid-send and loses its buffered events. Keep the atexit hook registered in stop() and let close() still join the worker after a stop().
There was a problem hiding this comment.
Right, fixed in 0f1eafb as you suggested. stop() no longer unregisters the atexit hook: it only stops tracking and wakes the worker. close() always joins the worker, including after a stop(), and unregisters the hook once it returns. It is still idempotent: a second call returns straight away. So a client collected before exit has its worker send the buffered events in the background, and the close at exit waits up to its timeout for that send.
One consequence: atexit holds a reference to close, so a processor that has only been stop()ped stays in memory until exit. It's small once its worker has finished. Clients closed explicitly are released as before.
Test: test_close__after_stop__waits_for_buffered_events fails before the fix, because close() used to return immediately after a stop().
| 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") |
There was a problem hiding this comment.
This logs one WARNING per dropped event while holding the condition on the caller's thread, which floods the logs during an outage on a busy service. Count the drops and log once per send.
There was a problem hiding this comment.
Right, fixed in 0f1eafb. _drop_oldest_events now only increments a counter under the lock. The worker reads and resets it when it takes a batch, and logs Event buffer was full: dropped the <n> oldest events once per send, after releasing the lock. This also counts events dropped when a kept batch is put back into a full buffer.
Test: test_track_event__full_buffer_while_sending__drops_logged_once tracks five events into a buffer of 2 while a send is in flight, and expects a single message for the 3 dropped. It fails before the fix.
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 <[email protected]>
Both fixed in 0f1eafb, along with the logging point. Details are inline:
The fork limitation also exists on Checks: 407 unit tests pass in 40 of 40 repeated runs, and in 30 of 30 runs under artificial CPU load. That load run caught a timing flaw in one of the new tests, which I fixed before pushing. The new tests fail without the fix. |
Zaimwa9
left a comment
Zaimwa9
left a comment
There was a problem hiding this comment.
Requesting changes. The fork, exit and log fixes all hold; one side effect of keeping the atexit hook remains.
| self._closed = False | ||
| self._start_worker() | ||
| _processors.add(self) | ||
| atexit.register(self.close) |
There was a problem hiding this comment.
atexit holds this bound method until close(), so every client garbage-collected without close() now leaves its processor and session alive for the life of the process; main retains none. Use one module-level atexit hook that closes every processor in _processors instead.
There was a problem hiding this comment.
To be fair, I'm not sure the SDK should own an exit hook at all. I'd rather document this approach than enforce it on all users.
Contributes to Flagsmith/sdk-spec#23