Collapse concurrent cache refreshes into a single fetch

[dchaudhari7177]

Addresses item 2 of #6 — the thundering herd. Independent of the item-4 change in #18 (merged) and the item-1 change in #19.

The problem

_collect_shared() released _cache_lock before calling collect():

with _cache_lock:
    cached = _cache
    if cached is not None and ...:
        return cached[1]

response = collect(adapters)      # <- outside the lock

The TTL check and the fetch are therefore not atomic. Every caller that arrives while the cache is cold or expired sees a miss and starts its own full fetch. N concurrent requests produce N fetches at WMO and USGS — precisely the burst the cache was added to prevent, and at ~774KB per unfiltered SWIC fetch it is not a cheap one.

As the issue says, this is not a correctness bug: they all compute the same answer and the last writer wins. It just undoes the rate-limiting.

The change

A second lock serialises the fetch, with a double-check after acquiring it:

hit = _fresh_cached()
if hit is not None:
    return hit

with _refresh_lock:
    hit = _fresh_cached()          # someone may have refreshed while we waited
    if hit is not None:
        return hit
    response = collect(adapters)
    with _cache_lock:
        _cache = (time.monotonic(), response)
    return response

The freshness check is extracted into _fresh_cached() so both call sites use identical logic rather than a duplicated inequality.

Why two locks and not one. Reusing _cache_lock for the whole operation would mean holding it across the fetch, so every reader — including the ones whose answer is already cached and fresh — would block behind a slow fetch. That trades a thundering herd for a convoy. As written, _cache_lock is never held across a fetch, and a cache hit never touches _refresh_lock at all.

Tests

Two in tests/test_api.py:

• test_concurrent_cold_requests_trigger_one_fetch — 8 threads released simultaneously through a threading.Barrier onto a cold cache, against an adapter that sleeps 0.2s so every thread is inside the window the old code left open. Asserts all 8 get 200 and the adapter was called exactly once.
• test_a_cache_hit_does_not_wait_on_an_in_flight_refresh — the control for the convoy risk above: a caller served from a warm cache must not be gated by the refresh lock.

import time is added at the top of the file; it was not previously imported there.

Verifying the test catches the bug

Reverting api.py to main while keeping the new tests:

AssertionError: expected one fetch, got 8

With the fix applied, tests/test_api.py is 23 passed. So the test measures the actual behaviour change, not just the presence of a lock.

Full suite

238 passed, 1 skipped, 7 deselected. The single failure — test_gdacs.py::test_geometry_falls_back_to_bbox_polygon_when_no_point — is pre-existing on main and unrelated.

Item 3 of #6 (the ignored adapters argument) is untouched and still open; it interacts with this code, so it seemed better as its own change than folded in here.

[coderabbitai]

Important

• 🔍 Trigger review

This repository does not receive automatic reviews because it has fewer than 10 stars.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 3605cedd-e207-4e66-8bc9-14d1860c3575

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[jamiusaliu]

Queued and will review — thanks for continuing to work through #6.

Flagging the situation honestly rather than letting you rebase three times: you now have three open PRs (#19, #22, #23) that all rewrite the same cache path in api.py, and all three conflict with current main and with each other on tests/test_api.py.

Rebasing them one at a time means three rounds of the same conflict. Options, your call:

1. Stack them — rebase Re-check a degraded fetch sooner than a complete one #19 first, then Collapse concurrent cache refreshes into a single fetch #22 on top, then Key the response cache by the adapter set it was fetched for #23. Clean history, but you wait on me between each.
2. Combine into one PR. They are all cache work on the same function, and a single "cache: shorter TTL when degraded, single-flight refresh, key by adapter set" PR would be easier for both of us. You lose per-change granularity in the log.

I have a slight preference for (2) given the overlap, but you are the one doing the work — pick whichever you prefer and I will review it that way.

One more thing: main is mid-change on api.py right now (issue #8 adds a second field list that surfaces in /alerts). Give me a few hours before rebasing so you are not rebasing onto a moving target.

[jamiusaliu]

This needs one more rebase, and there is a specific regression to watch for beyond the textual conflict.

#19 merged, so main now has _ttl_for(), which returns the shorter PARTIAL_CACHE_TTL_SECONDS for a degraded response and the full CACHE_TTL_SECONDS otherwise. Your _fresh_cached() hardcodes the constant:

def _fresh_cached() -> AlertsResponse | None:
    if cached is not None and (time.monotonic() - cached[0]) < CACHE_TTL_SECONDS:

where main now reads:

    if cached is not None and (time.monotonic() - cached[0]) < _ttl_for(cached[1]):

Rebased naively, _ttl_for becomes dead code and a degraded source is cached for the full 60s again, undoing #19. No test catches this because every cache test sets an explicit TTL, which is why I am flagging it rather than leaving you to find it.

The fix is small: _fresh_cached() should call _ttl_for(cached[1]) rather than the constant, so the single-flight path and the degraded-TTL behaviour compose instead of one overwriting the other.

Git did raise a conflict in src/alertmux/api.py and tests/test_api.py here, so you would probably have caught it. Worth saying explicitly because had the two edits been a few lines further apart, this would have auto-merged clean into a silent regression.

Verification I will run after the rebase, so you can check it yourself first. Frozen clock, a source that fails once then recovers:

t=0     partial=True   fetches=1
t=+5s   partial=True   fetches=1   <- cached
t=+13s  partial=False  fetches=2   <- degraded TTL expired, refetched
healthy source at +13s: fetches=1  <- full 60s TTL, not refetched

Plus your own single-flight assertion: N concurrent callers on a cold cache produce exactly one upstream fetch. Both need to hold at once.

#23 will need the same treatment afterwards, since it widens the cache tuple and cached[1] becomes the key rather than the response. Happy to take #23 first if that is easier for you, whichever order you prefer.

[dchaudhari7177]

Rebased onto current main (ac8e9c9). Went with option (1), stacked — #22 first, #23 rebased on top of it, both pushed. Keeping them separate seemed worth one extra round given they are three genuinely different failure modes, but say the word if you'd rather have the single combined PR and I'll squash them.

The _ttl_for regression you flagged. Caught and composed: _fresh_cached() now calls _ttl_for(cached[1]) rather than the constant, so the single-flight path and the degraded TTL stack instead of one overwriting the other. I put the reason in the docstring, because the next person to touch that line will see a bare constant as the obvious simplification.

One correction, in your favour. You said no test catches it. One does — test_a_degraded_fetch_is_rechecked_before_the_full_ttl. I checked rather than assumed, by reintroducing the naive resolution on the rebased branch:

_fresh_cached -> CACHE_TTL_SECONDS   1 failed, 37 passed
  FAILED test_a_degraded_fetch_is_rechecked_before_the_full_ttl
_fresh_cached -> _ttl_for(cached[1])  38 passed

It survives because it drives the TTL through a frozen clock and api.PARTIAL_CACHE_TTL_SECONDS rather than setting one explicitly, so it reads whatever the code actually uses. Your #19 suite is stronger than you gave it credit for — the silent-regression scenario would have been caught even if the two edits had been far enough apart to auto-merge.

Verification. Your table is exactly what the three #19 tests assert, and they pass on the rebased branch alongside my single-flight assertion — the two hold at once rather than trading off:

t=+5s   partial, cached          test_a_degraded_fetch_is_still_cached_within_its_shorter_ttl
t=+13s  partial, refetched       test_a_degraded_fetch_is_rechecked_before_the_full_ttl
+13s    healthy, not refetched   test_a_healthy_fetch_is_not_rechecked_at_the_degraded_interval
8 threads, cold cache -> 1 fetch test_concurrent_cold_requests_trigger_one_fetch

tests/test_api.py: 38 passed. Full suite: 421 passed, 10 failed — the same 10 as pristine main (test_gdacs ×1, test_mcp_server ×8, test_tsunami ×1), none of them touched here. ruff check src tests has the same 3 pre-existing findings before and after.

Thanks for flagging it rather than letting me find it after the fact — and for the heads-up about main moving on issue #8. #23 is rebased on top of this one and ready whenever this lands.

[jamiusaliu]

Verified on a fresh clone rather than taking the report at face value, and you are right and I was wrong. Reproduced exactly as you described, by reverting the resolution on ac8e9c9:

_fresh_cached -> CACHE_TTL_SECONDS     FAILED test_a_degraded_fetch_is_rechecked_before_the_full_ttl (assert 503 == 200)
_fresh_cached -> _ttl_for(cached[1])   38 passed

"No test catches this" was a claim I made without running it, which is exactly the thing I asked you to guard against. The test survives because it drives the TTL through a frozen clock and api.PARTIAL_CACHE_TTL_SECONDS rather than an explicit value, so it reads what the code actually does rather than what the test assumes. Thank you for checking instead of accepting it.

Stacking was the right call. Three failure modes, three PRs.

One thing about your environment, because you should not be treating a red suite as normal. Fresh clone of main at 2a0ee16:

pytest -q            421 passed, 1 skipped, 8 deselected, 0 failed
pytest -q -m ""      429 passed, 1 skipped, 0 failed          (live included)
pytest -q tests/test_mcp_server.py   8 passed                 (after pip install -e ".[mcp]")

The 8 MCP failures are what you get when those tests are named explicitly without the mcp extra installed: pytest.importorskip only skips them when they are collected, not when they are force-selected. test_gdacs and test_tsunami carry @pytest.mark.live and hit real endpoints, which is why addopts = "-m 'not live'" excludes them by default. Both pass here right now.

So main is green, and so is your branch. On 75db996: tests/test_api.py 40 passed, full suite 433 passed, 1 skipped, 0 failed with live included.

On _refresh_lock staying global: leave it. You reasoned it correctly. Per-key means a dict of locks and an eviction question, and #6 is not asking for that. Deliberate and documented beats optimal and speculative, and the docstring note means the next person meets the reasoning rather than the bare constant.

Rebase-merging this rather than squashing, so ac8e9c9 stays intact underneath #23. Squashing here would rewrite the commit #23 is stacked on and manufacture a conflict for no reason.

Merging now, #23 straight after.