fix: flush pending assignment and exposure events on LocalEvaluationClient.stop()

[cmyui]

Problem

LocalEvaluationClient.stop() stops the flag config poller and closes the connection pool, but never touches the Amplitude analytics instances created for the assignment and exposure services. Events still sitting in their buffers — up to flush_queue_size per instance, accumulating for up to flush_interval_millis (defaults: 200 events / 10s) — are silently dropped.

The analytics SDK does register an atexit shutdown hook, but that only fires on a clean interpreter exit of the main thread. In the environments where server-side local evaluation typically runs (gunicorn/uwsgi workers being recycled, forked job runners, containers receiving SIGKILL after a grace period), that hook frequently never runs — so the tail of assignment/$exposure events for every worker lifecycle is lost. stop() / the context-manager __exit__ is the documented lifecycle point, and it should deliver what was tracked.

Fix

stop() now, after stopping the poller and closing the connection pool:

1. Calls flush() on the assignment and exposure Amplitude instances (public API) and collects the returned futures.
2. Waits for them with a bounded timeout — new timeout parameter, default 10 seconds, consistent with the SDK's other network timeout defaults. None waits indefinitely. On timeout, a warning is logged with the number of unsent batches.
3. Calls shutdown() on both instances — after the flush, so events tracked post-stop() are dropped deliberately rather than accumulating in a stopped client.

Clients with no assignment/exposure config are unaffected.

Behavior change

stop() previously returned immediately; it can now block up to timeout seconds performing network sends. That is the point of the fix, but it is a change — flagging it for review. stop(timeout=0) effectively restores fire-and-forget (flush is still triggered; the wait is skipped).

Tests

tests/local/stop_flush_test.py: flush-then-shutdown ordering on both services, timeout bounding with a never-completing future, no-op with no event services, and context-manager exit. The 4 pre-existing errors in tests/util/user_test.py occur identically on unmodified main in my environment.

🤖 Generated with Claude Code

---

Note

Medium Risk
Changes documented client lifecycle semantics and can block up to 10s on shutdown; behavior is intentional but may affect worker recycle timing in production.

Overview
LocalEvaluationClient.stop() now drains buffered assignment and exposure analytics before tearing down, instead of only stopping the flag poller and closing the HTTP pool.

After the existing shutdown steps, it **flush()**es each configured assignment/exposure Amplitude instance, **wait()**s on returned futures up to a new timeout argument (10s default; None = wait forever), logs a warning if batches remain, then calls shutdown() on those instances. __exit__ still calls stop(), so context-manager use gets the same behavior. Clients without assignment/exposure config are unchanged.

stop() may now block on network I/O (up to timeout); stop(timeout=0) still triggers flush but skips waiting.

Adds tests/local/stop_flush_test.py for flush/shutdown ordering, timeout bounding, no-op without event services, and context-manager exit.

^{Reviewed by Cursor Bugbot for commit 41c2df7. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Bugbot Autofix is ON, but it could not run because the branch was deleted or merged before autofix could start.}

^{Reviewed by Cursor Bugbot for commit 41c2df7. Configure here.}

[cursor]

Flush futures nested, wait breaks

High Severity

Amplitude.flush() returns a list whose elements are themselves lists of batch futures (from each destination plugin), or None when a destination had nothing to send. This code treats that top-level list as a flat sequence of futures and passes it to wait(), which expects Future instances. When assignment or exposure events are actually pending, stop() raises instead of waiting, so the new flush path fails in the case it is meant to fix and shutdown() is never reached.

 

^{Reviewed by Cursor Bugbot for commit 41c2df7. Configure here.}

[cmyui]

Verified against every released amplitude-analytics version in the supported range (floor >=1.1.1 through current 1.2.3): Workers.flush() never returns a list — it returns None, a single Future (1.1.1–1.2.0, and the single-batch case since 1.2.1), or a combined Future created via threads_pool.submit(wait_for_all) (multi-batch case since 1.2.1). Timeline.flush() therefore returns a flat list of Future | None, and the Nones are filtered before wait(). Also confirmed empirically: with real pending events queued in both the assignment and exposure instances, stop() flushes and returns without raising.