sync: matter cache + breaker, runtime threading lock

matter/impact.py

@@ -31,9 +31,41 @@ from typing import Any
 from matter.passport import Environmental, Passport, Value
 
 
+import threading
+
 CLIMATIQ_BASE_URL = "https://api.climatiq.io"
 CLIMATIQ_API_KEY_ENV = "CLIMATIQ_API_KEY"
-CLIMATIQ_TIMEOUT_S = 4.0  # tight — caller wants fast inference, not perfect data
+CLIMATIQ_TIMEOUT_S = 3.0  # tight — caller wants fast inference, not perfect data
+CLIMATIQ_BREAKER_THRESHOLD = 3  # consecutive failures before we stop trying
+
+# Cache: free-text query → (activity_id, data_version). Populated on the first
+# successful /search; subsequent calls for the same query skip /search entirely.
+_climatiq_cache: dict[str, tuple[str, str]] = {}
+_climatiq_cache_lock = threading.Lock()
+
+# Per-process circuit breaker. If three consecutive Climatiq calls fail,
+# stop trying for the rest of this process's lifetime — fall back to static.
+# Reset to closed on any successful call.
+_climatiq_breaker = {"failures": 0, "tripped": False}
+_climatiq_breaker_lock = threading.Lock()
+
+
+def _breaker_record_failure() -> None:
+    with _climatiq_breaker_lock:
+        _climatiq_breaker["failures"] += 1
+        if _climatiq_breaker["failures"] >= CLIMATIQ_BREAKER_THRESHOLD:
+            _climatiq_breaker["tripped"] = True
+
+
+def _breaker_record_success() -> None:
+    with _climatiq_breaker_lock:
+        _climatiq_breaker["failures"] = 0
+        _climatiq_breaker["tripped"] = False
+
+
+def _breaker_open() -> bool:
+    with _climatiq_breaker_lock:
+        return _climatiq_breaker["tripped"]
 
 
 @dataclass(frozen=True)
@@ -141,26 +173,40 @@ def estimate_climatiq(
     query = _CLIMATIQ_QUERIES.get((head_name, identity_class, next_best_action))
     if query is None:
         return None
+    if _breaker_open():
+        # Circuit breaker tripped — don't waste time/budget on more API calls.
+        return None
     try:
         # Lazy import — httpx is in the package deps but optional at module load
         import httpx
         with httpx.Client(timeout=timeout) as client:
-            # 1. Search for the matching factor
-            search = client.get(
-                f"{CLIMATIQ_BASE_URL}/data/v1/search",
-                params={"query": query, "results_per_page": 1},
-                headers={"Authorization": f"Bearer {api_key}"},
-            )
-            if search.status_code != 200:
-                return None
-            results = (search.json() or {}).get("results") or []
-            if not results:
-                return None
-            top = results[0]
-            activity_id = top.get("activity_id")
-            data_version = top.get("data_version", "^21")
-            if not activity_id:
-                return None
+            headers = {"Authorization": f"Bearer {api_key}"}
+
+            # 1. Resolve activity_id — check in-process cache first, hit /search
+            # only on miss.
+            with _climatiq_cache_lock:
+                cached = _climatiq_cache.get(query)
+            if cached is not None:
+                activity_id, data_version = cached
+            else:
+                search = client.get(
+                    f"{CLIMATIQ_BASE_URL}/data/v1/search",
+                    params={"query": query, "results_per_page": 1},
+                    headers=headers,
+                )
+                if search.status_code != 200:
+                    _breaker_record_failure()
+                    return None
+                results = (search.json() or {}).get("results") or []
+                if not results:
+                    return None
+                top = results[0]
+                activity_id = top.get("activity_id")
+                data_version = top.get("data_version", "^21")
+                if not activity_id:
+                    return None
+                with _climatiq_cache_lock:
+                    _climatiq_cache[query] = (activity_id, data_version)
 
             # 2. Estimate emissions for our weight
             estimate = client.post(
@@ -169,29 +215,24 @@ def estimate_climatiq(
                     "emission_factor": {"activity_id": activity_id, "data_version": data_version},
                     "parameters": {"weight": weight_kg, "weight_unit": "kg"},
                 },
-                headers={"Authorization": f"Bearer {api_key}"},
+                headers=headers,
             )
             if estimate.status_code != 200:
+                _breaker_record_failure()
                 return None
             payload: dict[str, Any] = estimate.json() or {}
             co2e_kg = payload.get("co2e")
             if not isinstance(co2e_kg, (int, float)):
                 return None
 
-            # Climatiq returns *emissions caused* by the disposal pathway. Our
-            # convention is *kg avoided vs. BAU landfill*. We sign-flip: a
-            # recycling factor of, say, +0.8 kg CO2e/kg becomes -0.8 in our
-            # terms (recycling itself emits, but it AVOIDS landfill emissions
-            # which are separately measured). The static table already encodes
-            # the avoided-vs-BAU framing; the Climatiq path here surfaces the
-            # raw emissions of the routing — caller should treat as a positive
-            # footprint number, not a savings.
+            _breaker_record_success()
             basis = (
                 f"climatiq:{activity_id} (data_version {data_version}) "
                 f"× {weight_kg:.3f} kg via search '{query}'"
             )
             return float(co2e_kg), basis
     except Exception:
+        _breaker_record_failure()
         return None
 
 

transformers_runtime.py

@@ -11,6 +11,7 @@ the MATTER_MODEL_ID Space secret.
 from __future__ import annotations
 
 import os
+import threading
 from pathlib import Path
 from typing import Literal
 
@@ -53,28 +54,42 @@ class TransformersRuntime:
         self.max_new_tokens = max_new_tokens
         self._model = None
         self._processor = None
+        # Guards _ensure_loaded against concurrent first-call races. Two users
+        # hitting a cold Space simultaneously could both enter `from_pretrained`
+        # without this lock and double-allocate, OOM'ing CUDA.
+        self._load_lock = threading.Lock()
 
     def _ensure_loaded(self) -> None:
+        # Fast path: already loaded, no lock needed.
         if self._model is not None:
             return
-        from transformers import AutoModelForImageTextToText, AutoProcessor
-
-        dtype = torch.bfloat16 if torch.cuda.is_available() else torch.float32
-        device = "cuda" if torch.cuda.is_available() else "cpu"
-
-        self._processor = AutoProcessor.from_pretrained(self.model_id)
-        self._model = AutoModelForImageTextToText.from_pretrained(
-            self.model_id,
-            torch_dtype=dtype,
-            device_map=device,
-        )
-        if self.lora_id:
-            try:
-                from peft import PeftModel
-                self._model = PeftModel.from_pretrained(self._model, self.lora_id)
-            except Exception as e:
-                print(f"[TransformersRuntime] LoRA load failed ({self.lora_id}): {e}")
-        self._model.eval()
+        with self._load_lock:
+            # Double-checked locking: another thread may have completed the
+            # load while we were waiting for the lock.
+            if self._model is not None:
+                return
+            from transformers import AutoModelForImageTextToText, AutoProcessor
+
+            dtype = torch.bfloat16 if torch.cuda.is_available() else torch.float32
+            device = "cuda" if torch.cuda.is_available() else "cpu"
+
+            processor = AutoProcessor.from_pretrained(self.model_id)
+            model = AutoModelForImageTextToText.from_pretrained(
+                self.model_id,
+                torch_dtype=dtype,
+                device_map=device,
+            )
+            if self.lora_id:
+                try:
+                    from peft import PeftModel
+                    model = PeftModel.from_pretrained(model, self.lora_id)
+                except Exception as e:
+                    print(f"[TransformersRuntime] LoRA load failed ({self.lora_id}): {e}")
+            model.eval()
+            # Publish atomically — readers without the lock should never see a
+            # half-initialized state.
+            self._processor = processor
+            self._model = model
 
     def infer(self, prompt: str, image: Path | None) -> str:
         return self._infer_gpu(prompt, str(image) if image is not None else None)