fix: flush() delivers pending events without waiting out flush_interval - #797
Draft
posthog[bot] wants to merge 2 commits into
Draft
fix: flush() delivers pending events without waiting out flush_interval#797posthog[bot] wants to merge 2 commits into
posthog[bot] wants to merge 2 commits into
Conversation
The sdk-specs `flush` contract requires an explicit flush to bypass the batching thresholds: "Do not wait for `flushAt` or `flushInterval`; attempt delivery now" (openspec/specs/flush/spec.md, Requirement "Canonical flush behavior", Behavior #2). `Consumer.next()` accumulates until it has `flush_at` items or `flush_interval` elapses, and nothing connected `flush()` to a consumer parked mid-batch — `_Lane.flush()` only blocked on `queue.join()`. So a consumer holding a partial batch kept waiting for more events and the caller waited with it: on defaults, `capture()` + `flush()` took ~5s. Worse, `flush()` defaults to `timeout_seconds=10`, so any `flush_interval` above that made the default `flush()` give up and deliver nothing. Add `DrainSignal`, a generation counter shared by a lane and its consumers. `flush()` and `join()` bump it; while a request is outstanding a consumer takes only what is already queued and uploads as soon as the queue comes up empty, which is when it marks the request served. A consumer already holding a partial batch waits in slices so it notices a request that lands while it is parked; an idle consumer still parks for the whole interval, since a new put wakes its get() anyway. Timer-based batching without an explicit flush is unchanged: a slice expiring only re-arms the wait, and the batch window still closes on `flush_at`, the batch size limit, or a full `flush_interval`. Generated-By: PostHog Code Task-Id: 2bc2b5c5-03be-4da2-b7df-97c3364b33c9
Contributor
posthog-python Compliance ReportDate: 2026-07-29 08:37:38 UTC ✅ All Tests Passed!111/111 tests passed Capture_V1 Tests✅ 94/94 tests passed View Details
Feature_Flags Tests✅ 17/17 tests passed View Details
|
`capture_message_only_logs()` attaches a DEBUG handler to the process-wide "posthog" logger, and `Consumer.upload()` spans a whole `flush_interval`, so any background thread left running by an earlier test writes into the same captured stream. Asserting on the entire capture made the test depend on suite-wide timing: with flush() no longer waiting out flush_interval the suite runs ~40s faster, and feature-flag warnings from lingering pollers now reliably land inside that 5s window. Assert on the "error uploading" line instead, which is what the test is about. Matches how every other user of this helper asserts. Generated-By: PostHog Code Task-Id: 2bc2b5c5-03be-4da2-b7df-97c3364b33c9
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
💡 Motivation and Context
Why: this came out of a scheduled audit of posthog-python against the cross-SDK contracts in PostHog/sdk-specs, looking for places where the SDK does not yet satisfy the spec.
The
flushcontract says an explicit flush must not wait for the batching timer:openspec/specs/flush/spec.md— Requirement Canonical flush behavior, Behavior Update README.md #2: "Bypass normal wait conditions. Do not wait forflushAtorflushInterval; attempt delivery now."acceptance/public/flush.feature)openspec/specs/event-batcher/spec.md— Behavior Fix tests and clean up integration #8: "Publicflush()or shutdown paths bypass the normal timer/threshold waiting and drain what is pending."Consumer.next()accumulates until it hasflush_atitems orflush_intervalelapses, and there was no channel fromflush()to a consumer telling it to stop waiting —_Lane.flush()only blocks onqueue.join(). So a consumer holding fewer thanflush_atevents kept waiting for more, and the caller waited with it. Measured on defaults (flush_at=100,flush_interval=5.0) against a local mock server:capture()× 1, thenflush()capture()× 1000, thenflush()flush()on an empty queueflush_interval=20, 1 event, defaultflush()The last row is the sharp edge:
flush()defaults totimeout_seconds=10, so anyflush_intervalabove that meant the defaultflush()gave up before the consumer ever tried to send, logged "flush ran out of budget", and delivered nothing.Change
DrainSignalis a generation counter shared by a lane and its consumers._Lane.flush()(and_Lane.join()) bump it before waiting. While a request is outstanding a consumer takes only what is already queued — non-blocking gets, no timer wait — and uploads as soon as the queue comes up empty, which is also when it marks the request served.A consumer that is already holding a partial batch is parked on
queue.get()and cannot see a flag flip, so it now waits in slices (DRAIN_POLL_INTERVAL, 50ms) instead of one long block. An idle consumer still parks for the whole interval: anything a caller enqueued beforeflush()is already in the queue and wakes itsget()by itself, so there is nothing for a flush to wait on.Batching without an explicit flush is unchanged. A slice expiring only re-arms the wait; the batch window still closes on
flush_at,BATCH_SIZE_LIMIT, or a fullflush_interval. Verified directly: one event every 100ms for 1s withflush_at=100, flush_interval=1.0produces a single batch of 10 both before and after.flush()andshutdown()return much sooner. Any code that (perhaps accidentally) relied onflush()blocking for roughlyflush_intervalwill see it return promptly instead.flush_at, since the point is to stop waiting for a full batch. The spec allows this; a flush of a large backlog still sends full-size batches (verified: 1000 queued events → ten batches of 100).Consumergains an optionaldrain_signalargument; constructed without one it behaves exactly as before.Not fixed here (follow-up):
shutdown()still spends up to oneflush_intervalinsidejoin(), because a consumer parked with an empty batch only noticespause()when itsget()times out. That is a thread-teardown delay, not a delivery delay — events are already flushed by then — and waking those consumers needs a separate mechanism (a queue sentinel per consumer), so it is left out to keep this change focused. It is already an improvement:shutdown()with one queued event previously cost two intervals, now one.💚 How did you test it?
posthog/test/test_consumer.py(drain returns a partial batch without waiting; still respectsflush_at; request is served once the queue empties, so no spin; consecutive requests each drain; a request that lands while the consumer is parked mid-batch wakes it; a consumer with no drain signal batches exactly as before) andposthog/test/test_client.py(flush()does not wait outflush_interval; delivers whenflush_intervalexceeds the flush timeout; batches stay whole).pytest --timeout=120 --ignore=integration_tests --ignore=posthog/test/ai→ 1306 passed, 0 failed. (posthog/test/aineeds live provider credentials unavailable in this environment; those tests were not run and are untouched here.)ruff format --check .,ruff check .,mypy --no-site-packages --config-file mypy.ini . | mypy-baseline filter, andpython .github/scripts/check_public_api.pyall clean; the public API snapshot is regenerated.test_message_only_error_logs_include_posthog_prefixasserted on the entirecapture_message_only_logs()stream, which taps the process-wideposthoglogger for the ~5s thatConsumer.upload()takes. Because this change makes the suite ~40s faster, feature-flag warnings from background pollers left running by earlier tests started landing inside that window and failed the equality assertion (it went red in CI on the first push). Second commit narrows the assertion to the "error uploading" line — what the test is actually about, and how every other user of that helper asserts.Clientagainst a local mock HTTP server to produce the timing table above, plus repeated flushes, four concurrent flushes, and athread=4lane (all drain in ~50ms with no lost or duplicated events).📝 Checklist
flush()'s documented contract is what this makes true.If releasing new changes
sampo addto generate a changeset file — hand-written at.sampo/changesets/flush-bypasses-flush-interval.md(thesampoCLI isn't available in this environment); please sanity-check the format.🤖 Agent context
Autonomy: Fully autonomous
Opened by the SDK Spec Compliance Enforcer loop, which audits posthog-python against the contracts in PostHog/sdk-specs. This run re-checked the 22 spec capabilities whose
Applicabilityisboth(so in scope for a server SDK) using parallel read-only Claude Code sub-agents, then confirmed each candidate finding by running the SDK locally against a mock server.This violation was chosen because the spec language is unambiguous, it is confirmed by measurement rather than by reading, and it has real consequences for the recommended serverless pattern (
capture()thenflush()/shutdown()before the process exits) — including outright non-delivery whenflush_intervalexceeds the flush timeout. The concurrency design was picked over two alternatives: a plainthreading.Event(a shared flag cleared by whichever flush finishes first can strand a concurrent flush for another interval) and a queue sentinel per consumer (wakes parked consumers with no polling, but unconsumed sentinels leaveunfinished_tasksnon-zero, which would hang a laterqueue.join()).Runners-up deliberately left alone:
feature_enabled()has nodefault_value, which the recently updatedis-feature-enabledspec makes a SHALL — but the method is deprecated in favour ofevaluate_flags(), so it is a spec-vs-roadmap question for humans.alias()andgroup_identify()don't validate their required identities, which two@bothscenarios require them to drop with a warning — correct per spec, but it turns currently-emitted (malformed) events into dropped ones, so it wants a human call. The$feature_flag_calleddedupe map clears itself wholesale at capacity, which thefeature-flag-called-trackerspec explicitly says SHOULD NOT happen; worth its own PR.Agent-authored — requires human review; not self-merged.