Promote dev to main: geocode P0 fix + ops tooling + privacy logging #52
2 changed files with 178 additions and 50 deletions
|
|
@ -1009,10 +1009,13 @@ _REVGEO_CACHE: dict[str, str | None] = {}
|
||||||
_REVGEO_MIN_INTERVAL = 1.1 # seconds between successive Nominatim reverse calls
|
_REVGEO_MIN_INTERVAL = 1.1 # seconds between successive Nominatim reverse calls
|
||||||
_revgeo_last = 0.0
|
_revgeo_last = 0.0
|
||||||
|
|
||||||
_REVGEO_QUEUE: "queue.Queue[tuple[float, float, str, concurrent.futures.Future]]" = queue.Queue()
|
_REVGEO_QUEUE: "queue.Queue[tuple]" = queue.Queue()
|
||||||
_REVGEO_WORKER_LOCK = threading.Lock()
|
_REVGEO_WORKER_LOCK = threading.Lock()
|
||||||
_revgeo_worker_started = False
|
_revgeo_worker_started = False
|
||||||
_REVGEO_WAIT_TIMEOUT = 10.0 # seconds a caller waits for ITS OWN request before giving up
|
_REVGEO_WAIT_TIMEOUT = 10.0 # seconds a caller waits for ITS OWN request before giving up
|
||||||
|
_GEOCODE_WAIT_TIMEOUT = 35.0 # forward /geocode is a deliberate user search, not a map-pan
|
||||||
|
# enrichment -- worth a longer wait than reverse geocoding's,
|
||||||
|
# covering the 30s per-request timeout plus queue wait.
|
||||||
|
|
||||||
|
|
||||||
def reverse_geocode_cached(lat: float, lon: float) -> tuple[bool, str | None]:
|
def reverse_geocode_cached(lat: float, lon: float) -> tuple[bool, str | None]:
|
||||||
|
|
@ -1065,31 +1068,46 @@ def _fetch_revgeo_label(lat: float, lon: float) -> str | None:
|
||||||
|
|
||||||
|
|
||||||
def _revgeo_worker() -> None:
|
def _revgeo_worker() -> None:
|
||||||
"""Drains _REVGEO_QUEUE one request at a time, forever. The sole caller of
|
"""Drains _REVGEO_QUEUE one job at a time, forever — both reverse ("revgeo")
|
||||||
_fetch_revgeo_label, so the ~1/sec pacing below is enforced just by doing the
|
and forward ("geocode") jobs, so the ~1/sec pacing below is enforced just by
|
||||||
work serially — no lock needed, since nothing else ever touches Nominatim.
|
doing the work serially on this one thread, no lock needed, since nothing
|
||||||
Re-checks the cache before fetching (a request queued behind an identical one
|
else ever touches Nominatim. Reverse jobs re-check the cache before fetching
|
||||||
is answered from what the earlier request just cached, no duplicate call),
|
(a request queued behind an identical one is answered from what the earlier
|
||||||
and always finishes the fetch + persists the result even if the original
|
request just cached, no duplicate call) and always finish + persist even if
|
||||||
caller already gave up waiting (see reverse_geocode's timeout)."""
|
the original caller already gave up waiting (see reverse_geocode's timeout).
|
||||||
|
Forward jobs have no cache (see geocode_nominatim) but share the same pacer."""
|
||||||
global _revgeo_last
|
global _revgeo_last
|
||||||
while True:
|
while True:
|
||||||
lat, lon, key, fut = _REVGEO_QUEUE.get()
|
job = _REVGEO_QUEUE.get()
|
||||||
|
kind = job[0]
|
||||||
try:
|
try:
|
||||||
found, label = reverse_geocode_cached(lat, lon)
|
if kind == "revgeo":
|
||||||
if not found:
|
_, lat, lon, key, fut = job
|
||||||
|
found, label = reverse_geocode_cached(lat, lon)
|
||||||
|
if not found:
|
||||||
|
wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
|
||||||
|
if wait > 0:
|
||||||
|
time.sleep(wait)
|
||||||
|
label = _fetch_revgeo_label(lat, lon)
|
||||||
|
_revgeo_last = time.monotonic()
|
||||||
|
_REVGEO_CACHE[key] = label
|
||||||
|
store.put_revgeo(key, label) # survive restarts (a None label retries after its TTL)
|
||||||
|
if not fut.done():
|
||||||
|
fut.set_result(label)
|
||||||
|
else: # "geocode"
|
||||||
|
_, name, count, fut = job
|
||||||
wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
|
wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
|
||||||
if wait > 0:
|
if wait > 0:
|
||||||
time.sleep(wait)
|
time.sleep(wait)
|
||||||
label = _fetch_revgeo_label(lat, lon)
|
try:
|
||||||
_revgeo_last = time.monotonic()
|
results = _fetch_geocode_forward(name, count)
|
||||||
_REVGEO_CACHE[key] = label
|
finally:
|
||||||
store.put_revgeo(key, label) # survive restarts (a None label retries after its TTL)
|
_revgeo_last = time.monotonic()
|
||||||
if not fut.done():
|
if not fut.done():
|
||||||
fut.set_result(label)
|
fut.set_result(results)
|
||||||
except Exception: # noqa: BLE001 - never let a bad request kill the worker
|
except Exception: # noqa: BLE001 - never let a bad request kill the worker
|
||||||
if not fut.done():
|
if not fut.done():
|
||||||
fut.set_result(None)
|
fut.set_result(None if kind == "revgeo" else [])
|
||||||
finally:
|
finally:
|
||||||
_REVGEO_QUEUE.task_done()
|
_REVGEO_QUEUE.task_done()
|
||||||
|
|
||||||
|
|
@ -1129,44 +1147,28 @@ def reverse_geocode(lat: float, lon: float) -> str | None:
|
||||||
return label
|
return label
|
||||||
_start_revgeo_worker()
|
_start_revgeo_worker()
|
||||||
fut: "concurrent.futures.Future[str | None]" = concurrent.futures.Future()
|
fut: "concurrent.futures.Future[str | None]" = concurrent.futures.Future()
|
||||||
_REVGEO_QUEUE.put((lat, lon, key, fut))
|
_REVGEO_QUEUE.put(("revgeo", lat, lon, key, fut))
|
||||||
try:
|
try:
|
||||||
return fut.result(timeout=_REVGEO_WAIT_TIMEOUT)
|
return fut.result(timeout=_REVGEO_WAIT_TIMEOUT)
|
||||||
except concurrent.futures.TimeoutError:
|
except concurrent.futures.TimeoutError:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def geocode_nominatim(name: str, count: int = 5) -> list[dict]:
|
def _fetch_geocode_forward(name: str, count: int) -> list[dict]:
|
||||||
"""Forward place-name lookup via OpenStreetMap Nominatim (keyless).
|
"""The actual Nominatim forward-search HTTP call + result shaping. Called
|
||||||
|
ONLY from _revgeo_worker (never on a caller's thread), mirroring
|
||||||
Replaces the former Open-Meteo geocoder. Nominatim covers the long tail the
|
_fetch_revgeo_label — but unlike that function, a bad response is allowed to
|
||||||
local GeoNames index can't — neighbourhoods, postcodes, sub-1000-population
|
raise here; _revgeo_worker's own except clause turns it into an empty list,
|
||||||
villages, and alternate/native-language spellings — so /geocode falls back to
|
same net effect without a second layer of exception-swallowing."""
|
||||||
it on a local miss. It carries no population, so results keep Nominatim's own
|
r = _request(
|
||||||
relevance order; name/admin/country map straight across.
|
"https://nominatim.openstreetmap.org/search",
|
||||||
|
{"q": name, "format": "jsonv2", "addressdetails": 1,
|
||||||
Shares the reverse geocoder's lock and ~1/sec pacing: both hit the same host,
|
"limit": count, "accept-language": "en"},
|
||||||
so serializing them together keeps total Nominatim traffic under the usage
|
30,
|
||||||
policy. Only the low-volume /geocode miss path reaches here — autocomplete
|
phase="geocode",
|
||||||
(/suggest) is served purely from the local index and never calls out.
|
headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"},
|
||||||
"""
|
)
|
||||||
global _revgeo_last
|
rows = r.json() or []
|
||||||
with _REVGEO_LOCK:
|
|
||||||
wait = _REVGEO_MIN_INTERVAL - (time.monotonic() - _revgeo_last)
|
|
||||||
if wait > 0:
|
|
||||||
time.sleep(wait)
|
|
||||||
try:
|
|
||||||
r = _request(
|
|
||||||
"https://nominatim.openstreetmap.org/search",
|
|
||||||
{"q": name, "format": "jsonv2", "addressdetails": 1,
|
|
||||||
"limit": count, "accept-language": "en"},
|
|
||||||
30,
|
|
||||||
phase="geocode",
|
|
||||||
headers={"User-Agent": "Thermograph/0.1 (local weather grading app)"},
|
|
||||||
)
|
|
||||||
rows = r.json() or []
|
|
||||||
finally:
|
|
||||||
_revgeo_last = time.monotonic()
|
|
||||||
out = []
|
out = []
|
||||||
for g in rows:
|
for g in rows:
|
||||||
a = g.get("address", {}) or {}
|
a = g.get("address", {}) or {}
|
||||||
|
|
@ -1187,3 +1189,28 @@ def geocode_nominatim(name: str, count: int = 5) -> list[dict]:
|
||||||
"population": None, # Nominatim has no population; order is relevance-based
|
"population": None, # Nominatim has no population; order is relevance-based
|
||||||
})
|
})
|
||||||
return out
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def geocode_nominatim(name: str, count: int = 5) -> list[dict]:
|
||||||
|
"""Forward place-name lookup via OpenStreetMap Nominatim (keyless).
|
||||||
|
|
||||||
|
Replaces the former Open-Meteo geocoder. Nominatim covers the long tail the
|
||||||
|
local GeoNames index can't — neighbourhoods, postcodes, sub-1000-population
|
||||||
|
villages, and alternate/native-language spellings — so /geocode falls back to
|
||||||
|
it on a local miss. It carries no population, so results keep Nominatim's own
|
||||||
|
relevance order; name/admin/country map straight across.
|
||||||
|
|
||||||
|
Shares the reverse geocoder's worker thread and ~1/sec pacing: both hit the
|
||||||
|
same host, and both jobs are drained serially by the one thread in
|
||||||
|
_revgeo_worker, so there is exactly one writer of _revgeo_last — no lock
|
||||||
|
needed, same reasoning as reverse_geocode. Only the low-volume /geocode miss
|
||||||
|
path reaches here — autocomplete (/suggest) is served purely from the local
|
||||||
|
index and never calls out.
|
||||||
|
"""
|
||||||
|
_start_revgeo_worker()
|
||||||
|
fut: "concurrent.futures.Future[list[dict]]" = concurrent.futures.Future()
|
||||||
|
_REVGEO_QUEUE.put(("geocode", name, count, fut))
|
||||||
|
try:
|
||||||
|
return fut.result(timeout=_GEOCODE_WAIT_TIMEOUT) or []
|
||||||
|
except concurrent.futures.TimeoutError:
|
||||||
|
return []
|
||||||
|
|
|
||||||
|
|
@ -615,3 +615,104 @@ def test_reverse_geocode_timeout_returns_none_without_blocking_the_caller(monkey
|
||||||
elapsed = time_mod.monotonic() - t0
|
elapsed = time_mod.monotonic() - t0
|
||||||
assert label is None
|
assert label is None
|
||||||
assert elapsed < 0.2 # returned near the wait timeout, not after the 0.3s fetch
|
assert elapsed < 0.2 # returned near the wait timeout, not after the 0.3s fetch
|
||||||
|
|
||||||
|
|
||||||
|
# --- forward geocode: shares the reverse-geocode worker, not a second lock -----
|
||||||
|
# Regression coverage for the NameError _REVGEO_LOCK bug (geocode_nominatim
|
||||||
|
# referenced a lock that was removed when reverse geocoding moved to the
|
||||||
|
# worker/queue design, so every forward lookup 502'd in production). These
|
||||||
|
# exercise the real queue/worker plumbing rather than mocking geocode_nominatim
|
||||||
|
# itself away, so a reintroduced bare `with _REVGEO_LOCK:` or any other
|
||||||
|
# not-actually-defined-name bug fails loudly here instead of shipping unseen.
|
||||||
|
|
||||||
|
def test_geocode_nominatim_resolves_via_the_worker_thread(monkeypatch):
|
||||||
|
monkeypatch.setattr(
|
||||||
|
climate, "_fetch_geocode_forward",
|
||||||
|
lambda name, count: [{"name": name, "admin1": None, "country": "Testland",
|
||||||
|
"country_code": "TL", "lat": 1.0, "lon": 2.0,
|
||||||
|
"population": None}],
|
||||||
|
)
|
||||||
|
results = climate.geocode_nominatim("Nowheresville")
|
||||||
|
assert results[0]["name"] == "Nowheresville"
|
||||||
|
assert results[0]["country"] == "Testland"
|
||||||
|
|
||||||
|
|
||||||
|
def test_geocode_nominatim_timeout_returns_empty_list(monkeypatch):
|
||||||
|
"""Same shape as reverse_geocode's timeout test: a caller waits only up to
|
||||||
|
_GEOCODE_WAIT_TIMEOUT, not as long as the fetch itself takes."""
|
||||||
|
import time as time_mod
|
||||||
|
monkeypatch.setattr(climate, "_GEOCODE_WAIT_TIMEOUT", 0.05)
|
||||||
|
|
||||||
|
def slow_fetch(name, count):
|
||||||
|
time_mod.sleep(0.3)
|
||||||
|
return [{"name": "Too Slow"}]
|
||||||
|
|
||||||
|
monkeypatch.setattr(climate, "_fetch_geocode_forward", slow_fetch)
|
||||||
|
t0 = time_mod.monotonic()
|
||||||
|
results = climate.geocode_nominatim("anywhere")
|
||||||
|
elapsed = time_mod.monotonic() - t0
|
||||||
|
assert results == []
|
||||||
|
assert elapsed < 0.2
|
||||||
|
|
||||||
|
|
||||||
|
def test_geocode_nominatim_a_bad_fetch_degrades_to_empty_list_not_a_crash(monkeypatch):
|
||||||
|
"""_fetch_geocode_forward is allowed to raise (matches _fetch_revgeo_label's
|
||||||
|
contract loosely -- the worker's except clause is the actual safety net);
|
||||||
|
confirm a raising fetch never reaches the caller as an exception."""
|
||||||
|
def boom(name, count):
|
||||||
|
raise RuntimeError("Nominatim is down")
|
||||||
|
monkeypatch.setattr(climate, "_fetch_geocode_forward", boom)
|
||||||
|
assert climate.geocode_nominatim("anywhere") == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_geocode_nominatim_shares_the_reverse_geocode_pacer(monkeypatch):
|
||||||
|
"""Forward and reverse jobs are drained by the SAME worker thread off the
|
||||||
|
SAME queue, so a forward call advances _revgeo_last exactly like a reverse
|
||||||
|
one does -- this is what makes a second lock unnecessary."""
|
||||||
|
monkeypatch.setattr(climate, "_revgeo_last", 0.0)
|
||||||
|
monkeypatch.setattr(climate, "_fetch_geocode_forward", lambda name, count: [])
|
||||||
|
before = climate._revgeo_last
|
||||||
|
climate.geocode_nominatim("anywhere")
|
||||||
|
assert climate._revgeo_last > before
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_geocode_forward_parses_nominatim_response(monkeypatch):
|
||||||
|
"""Executes _fetch_geocode_forward's real body (the function the NameError
|
||||||
|
bug prevented from ever running) against a stubbed HTTP transport -- not a
|
||||||
|
monkeypatch of geocode_nominatim itself, unlike the API-layer tests."""
|
||||||
|
class _FakeResp:
|
||||||
|
def json(self):
|
||||||
|
return [
|
||||||
|
{"name": "West Seattle", "lat": "47.57", "lon": "-122.38",
|
||||||
|
"display_name": "West Seattle, Seattle, King County, Washington, United States",
|
||||||
|
"address": {"suburb": "West Seattle", "city": "Seattle",
|
||||||
|
"state": "Washington", "country": "United States",
|
||||||
|
"country_code": "us"}},
|
||||||
|
# No `name`, no recognized address component -- falls back to the
|
||||||
|
# head of display_name, exercising that branch too.
|
||||||
|
{"lat": "51.5", "lon": "-0.1",
|
||||||
|
"display_name": "Some Unnamed Place, Greater London, England",
|
||||||
|
"address": {"country": "United Kingdom", "country_code": "gb"}},
|
||||||
|
]
|
||||||
|
|
||||||
|
captured = {}
|
||||||
|
|
||||||
|
def fake_request(url, params, timeout, *, phase, headers=None, attempts=climate.MAX_ATTEMPTS):
|
||||||
|
captured["url"] = url
|
||||||
|
captured["params"] = params
|
||||||
|
captured["phase"] = phase
|
||||||
|
return _FakeResp()
|
||||||
|
|
||||||
|
monkeypatch.setattr(climate, "_request", fake_request)
|
||||||
|
results = climate._fetch_geocode_forward("west seattle", 5)
|
||||||
|
|
||||||
|
assert captured["url"] == "https://nominatim.openstreetmap.org/search"
|
||||||
|
assert captured["params"]["q"] == "west seattle"
|
||||||
|
assert captured["phase"] == "geocode"
|
||||||
|
|
||||||
|
assert results[0] == {
|
||||||
|
"name": "West Seattle", "admin1": "Washington", "country": "United States",
|
||||||
|
"country_code": "US", "lat": 47.57, "lon": -122.38, "population": None,
|
||||||
|
}
|
||||||
|
assert results[1]["name"] == "Some Unnamed Place"
|
||||||
|
assert results[1]["country_code"] == "GB"
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue