Repository navigation
fix(prompts): share one request between concurrent fetches of the same prompt - #1059
jurajmajerik wants to merge 5 commits into
Conversation
![posthog[bot]](https://avatars.githubusercontent.com/in/1203049?s=60&v=4){kind=small}
| finally: | ||
| with self._in_flight_lock: | ||
| self._in_flight.pop(cache_key, None) | ||
| pending.done.set() |
There was a problem hiding this comment.
P2 · Publish cache state before removing the in-flight entry
When a leader finishes fetching but has not yet updated _cache in _get_internal, _fetch_prompt_once removes its in-flight entry. Another concurrent caller sees both a cache miss and no pending fetch, so it starts a duplicate request. This leaves a gap in the new request sharing on cold caches and expiry; publish the cache or cooldown state before removing the entry.
| if isinstance(error, Exception): | ||
| self._maybe_capture_error( | ||
| error, name=name, version=version, label=label | ||
| ) |
There was a problem hiding this comment.
P2 · Release waiters before invoking error-tracking callbacks
With capture_errors=True, a capture_exception callback such as before_send that calls get() for the same prompt enters pending.done.wait() on the leader's own unfinished fetch. The leader cannot reach finally, so the callback and all waiting callers hang indefinitely, even with a fallback. Complete the shared failure before invoking error-tracking callbacks.
| if not is_leader: | ||
| pending.done.wait() |
There was a problem hiding this comment.
P2 · Reset prompt flight state after a fork
When a process forks while another thread fetches a prompt, the child inherits _in_flight but loses the leader thread. Its next get() for that key blocks forever in pending.done.wait(), even with a fallback; the client's existing fork hook does not reset this state. Replace the lock and clear inherited flights in the child before accepting requests.
posthog-python Compliance ReportDate: 2026-10-14T08:36:17.906833+00:00 ✅ All Tests Passed!121/121 tests passed Capture_V1 Tests✅ 95/95 tests passed View Details
Capture_Ai Tests✅ 5/5 tests passed View Details
Feature_Flags Tests✅ 17/17 tests passed View Details
Feature_Flags_Local_Evaluation Tests✅ 4/4 tests passed View Details
|
| raise | ||
| finally: | ||
| with self._in_flight_lock: | ||
| self._in_flight.pop(cache_key, None) |
There was a problem hiding this comment.
P2 · Preserve the replacement flight when a timed-out leader finishes
When a fetch exceeds the 30-second waiter timeout, another caller replaces its flight and starts fetching. If the original leader then finishes, this unconditional pop() removes the replacement's entry while its request is still running. Subsequent expired-cache or zero-TTL reads start additional requests; remove the entry only when it still refers to pending.
| if nested_results: | ||
| return | ||
| nested_results.append( | ||
| prompts.get("test-prompt", with_metadata=True, fallback="nested") | ||
| ) |
There was a problem hiding this comment.
P2 · Set the callback recursion guard before fetching the nested prompt
With the mocked fetch always failing, the nested prompts.get() invokes capture_exception again before nested_results.append() executes. The guard therefore remains false until recursion exhaustion, and unwinding appends many results, failing the assertion expecting one nested result. Set a separate guard before entering the nested fetch.
| if hasattr(os, "register_at_fork"): | ||
| weak_self = weakref.ref(self) | ||
| os.register_at_fork( | ||
| after_in_child=lambda: Prompts._reset_in_flight_after_fork(weak_self) | ||
| ) |
There was a problem hiding this comment.
P2 · Register one fork callback for a weak registry of prompt instances
On platforms supporting register_at_fork, every Prompts instance permanently registers a new closure. The weak reference lets the instance disappear, but Python retains the callback and its closure for the process lifetime. Applications repeatedly constructing instances accumulate memory and an ever-growing callback list executed on every fork; register one callback over a weak instance registry.
jurajmajerik
changed the title
| with self._in_flight_lock: | ||
| pending = self._in_flight.get(cache_key) | ||
| is_leader = pending is None | ||
| if pending is None: | ||
| pending = _InFlightFetch() | ||
| self._in_flight[cache_key] = pending |
There was a problem hiding this comment.
P2 · Recheck cache freshness before electing a new fetch leader
When a caller passes _get_internal's cache check and pauses before acquiring this lock, another caller can fetch, populate the cache, and remove its flight. The paused caller then creates a new flight and sends a duplicate request despite the fresh cache. Publishing the cache first leaves this race open; recheck freshness under the election lock while preserving zero-TTL behavior.
| if not pending.done.wait(_IN_FLIGHT_WAIT_SECONDS): | ||
| with self._in_flight_lock: | ||
| if self._in_flight.get(cache_key) is pending: | ||
| del self._in_flight[cache_key] | ||
| return self._fetch_prompt_once(name, version, label) |
There was a problem hiding this comment.
P2 · Avoid replacing live fetches after the fixed waiter timeout
When the server returns 429 with Retry-After: 60, _get_session()'s urllib3 retries sleep beyond the 30-second waiter limit. Waiters delete the live flight and launch another request every 30 seconds, amplifying traffic while the server asks the client to back off. The transport has no 30-second total deadline; align replacement with an actual bounded fetch lifetime.
|
Simplified after the review rounds: the timed wait and replacement-fetch logic are gone, waiters just wait for the leader, a single fork hook resets inherited state in a child, and the cache freshness check also runs under the election lock so no caller starts a second request for a prompt that was just fetched. |
marandaneto
left a comment
marandaneto
left a comment
There was a problem hiding this comment.
The code looks good; no qualifying findings. Verdict: correct.
This branch has not been deployed
2 participants
💡 Motivation and Context
An app with many threads, such as a worker with 50 activity slots, sends one request per thread when a prompt is not in the cache or its TTL has expired. That burst repeats every TTL and is enough to hit the per-minute prompt fetch limit.
Now one thread makes the request and the others wait for its result. With
capture_errors=True, a shared failure is reported once instead of once per thread.Same change as PostHog/posthog-js#5266 for the JS client. Part of #106807 in PostHog/posthog.
💚 How did you test it?
Two new tests in
test_prompts.py: ten threads on a cold cache make one request, and ten threads on a failing fetch make one request and onecapture_exceptioncall. Both fail on the old code with 11 and 10 requests.📝 Checklist
If releasing new changes
sampo addto generate a changeset file