be nicer to system memory

Author: jackharrhy

src/where_the_plow/cache.py

@@ -1,8 +1,8 @@
 # src/where_the_plow/cache.py
-"""Disk cache for coverage trail responses.
+"""Disk cache for JSON responses (coverage trails, search results, etc.).
 
-Stores JSON in /tmp/where-the-plow-cache/ keyed by a hash of
-(since, until, source). Each entry carries an absolute expiry time.
+Stores JSON in /tmp/where-the-plow-cache/ keyed by a hash of the
+request parameters. Each entry carries an absolute expiry time.
 Uses LRU-style eviction by file access time when total cache size
 exceeds a budget.
 """
@@ -79,7 +79,9 @@ def _delete_if_expired(path: Path, expires_at: float) -> bool:
     return True
 
 
-def get(since: datetime, until: datetime, source: str | None = None) -> list[dict] | None:
+def get(
+    since: datetime, until: datetime, source: str | None = None
+) -> list[dict] | None:
     """Return cached trails or None."""
     path = CACHE_DIR / f"{_cache_key(since, until, source)}.json"
     if not path.exists():
@@ -99,7 +101,9 @@ def get(since: datetime, until: datetime, source: str | None = None) -> list[dic
         return None
 
 
-def put(since: datetime, until: datetime, trails: list[dict], source: str | None = None):
+def put(
+    since: datetime, until: datetime, trails: list[dict], source: str | None = None
+):
     """Store trails in disk cache with endpoint-specific TTL policy."""
     _ensure_dir()
     _evict_if_needed()
@@ -114,3 +118,48 @@ def put(since: datetime, until: datetime, trails: list[dict], source: str | None
         logger.debug("cache put: %s (%d trails, ttl=%ds)", path.name, len(trails), ttl)
     except OSError:
         pass
+
+
+# ── Search cache (Nominatim geocoding results) ───────
+
+SEARCH_CACHE_TTL = 86400  # 24 hours — addresses don't change often
+
+
+def _search_key(query: str) -> str:
+    return "search_" + hashlib.sha256(query.encode()).hexdigest()[:24]
+
+
+def search_get(query: str) -> list[dict] | None:
+    """Return cached search results or None."""
+    path = CACHE_DIR / f"{_search_key(query)}.json"
+    if not path.exists():
+        return None
+    try:
+        payload = json.loads(path.read_text())
+        expires_at = float(payload.get("expires_at", 0))
+        results = payload.get("results")
+        if not isinstance(results, list):
+            return None
+        if _delete_if_expired(path, expires_at):
+            return None
+        os.utime(path)
+        logger.debug("search cache hit: %s", query)
+        return results
+    except (OSError, ValueError, json.JSONDecodeError):
+        return None
+
+
+def search_put(query: str, results: list[dict]) -> None:
+    """Store search results on disk."""
+    _ensure_dir()
+    _evict_if_needed()
+    path = CACHE_DIR / f"{_search_key(query)}.json"
+    payload = {
+        "expires_at": time.time() + SEARCH_CACHE_TTL,
+        "results": results,
+    }
+    try:
+        path.write_text(json.dumps(payload))
+        logger.debug("search cache put: %s (%d results)", query, len(results))
+    except OSError:
+        pass

src/where_the_plow/db.py

@@ -12,6 +12,9 @@ def __init__(self, path: str):
         self.path = path
         os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
         self.conn = duckdb.connect(path)
+        # Cap memory so DuckDB doesn't claim 80% of system RAM on a small VPS.
+        self.conn.execute("SET memory_limit = '512MB'")
+        self.conn.execute("SET threads = 2")
 
     def _cursor(self) -> duckdb.DuckDBPyConnection:
         """Create a thread-local cursor for safe concurrent access."""

src/where_the_plow/routes.py

@@ -2,7 +2,6 @@
 import asyncio
 import logging
 import time
-from collections import defaultdict
 from datetime import datetime, timezone, timedelta
 
 import httpx
@@ -23,15 +22,22 @@ class RateLimiter:
     def __init__(self, max_hits: int, window_seconds: int):
         self.max_hits = max_hits
         self.window = window_seconds
-        self._hits: dict[str, list[float]] = defaultdict(list)
+        self._hits: dict[str, list[float]] = {}
 
     def is_limited(self, key: str) -> bool:
         now = time.monotonic()
-        bucket = self._hits[key]
-        self._hits[key] = [t for t in bucket if now - t < self.window]
-        if len(self._hits[key]) >= self.max_hits:
+        bucket = self._hits.get(key)
+        if bucket is not None:
+            bucket = [t for t in bucket if now - t < self.window]
+            if not bucket:
+                del self._hits[key]
+                bucket = []
+        else:
+            bucket = []
+        if len(bucket) >= self.max_hits:
             return True
-        self._hits[key].append(now)
+        bucket.append(now)
+        self._hits[key] = bucket
         return False
 
 
@@ -42,39 +48,17 @@ def is_limited(self, key: str) -> bool:
 )  # 6 searches per min per IP
 
 
-# ── Nominatim search proxy with in-memory cache ──────
+# ── Nominatim search proxy ────────────────────────────
 
 NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
 NOMINATIM_USER_AGENT = (
     "WhereThePlow/1.0 (St. John's snowplow tracker; https://plow.jackharrhy.dev)"
 )
 ST_JOHNS_VIEWBOX = "-52.85,47.45,-52.55,47.65"
-SEARCH_CACHE_TTL = 86400  # 24 hours — addresses don't change often
-SEARCH_CACHE_MAX = 500  # max entries before evicting oldest
 
-_search_cache: dict[str, tuple[float, list[dict]]] = {}  # key -> (expires_at, results)
 _nominatim_last_request: float = 0.0  # monotonic timestamp of last outbound request
 
 
-def _search_cache_get(key: str) -> list[dict] | None:
-    entry = _search_cache.get(key)
-    if entry is None:
-        return None
-    expires_at, results = entry
-    if time.monotonic() > expires_at:
-        del _search_cache[key]
-        return None
-    return results
-
-
-def _search_cache_put(key: str, results: list[dict]) -> None:
-    # Evict oldest entries if cache is full
-    if len(_search_cache) >= SEARCH_CACHE_MAX:
-        oldest_key = min(_search_cache, key=lambda k: _search_cache[k][0])
-        del _search_cache[oldest_key]
-    _search_cache[key] = (time.monotonic() + SEARCH_CACHE_TTL, results)
-
-
 def _client_ip(request: Request) -> str:
     return request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (
         request.client.host if request.client else "unknown"
@@ -460,7 +444,7 @@ async def search_address(
 
     cache_key = q.strip().lower()
 
-    cached = _search_cache_get(cache_key)
+    cached = cache.search_get(cache_key)
     if cached is not None:
         return JSONResponse(content=cached)
 
@@ -495,7 +479,7 @@ async def search_address(
 
         raw = resp.json()
         results = [_format_search_result(r) for r in raw]
-        _search_cache_put(cache_key, results)
+        cache.search_put(cache_key, results)
         return JSONResponse(content=results)
 
     except httpx.TimeoutException: