fix(cache): concurrent writes of one key neither crash the run nor bill twice

CHANGELOG.md

@@ -95,6 +95,12 @@ different event from one that moved because it was wrong.
   timeout, 408, 409, 425, 429 and 5xx), `Retry-After` is honoured up to 60
   seconds, the SDKs' own retries are off, and a failed case records the
   attempts it actually made.
+- Two workers answering cases with the same text wrote the same cache entry
+  through one shared temporary file; on Windows the loser raised
+  `PermissionError` and aborted the run, losing every call already paid for
+  (#27). Each write now has a temporary file of its own, a write that still
+  fails is counted in the cache stats (`write_errors`) instead of raised, and
+  cases sharing a key are answered once per run, the rest from the cache.
 - Every validation message on the site's calculator and planner read
   "[object Object]" (#25). Messages now say what is wrong, mark the field
   invalid, and are tied to it for screen readers; a result is announced as one

src/plumbline/runner/cache.py

@@ -12,13 +12,21 @@
 
 A hit is recorded as a hit. It does not count toward cost or latency, because it
 measures disk rather than the model.
+
+Cases that share a key within one run are answered once: the runner holds a lock
+per key across look up, call, and store, so the second waits and then hits. A
+write that fails is counted and dropped rather than raised, because the answer
+it was storing is already paid for.
 """
 
 from __future__ import annotations
 
+import contextlib
 import dataclasses
 import hashlib
 import json
+import os
+import tempfile
 from collections.abc import Sequence
 from pathlib import Path
 from typing import Any
@@ -94,6 +102,7 @@ def __init__(self, directory: Path | str, *, enabled: bool = True) -> None:
         self.hits = 0
         self.misses = 0
         self.writes = 0
+        self.write_errors = 0
 
     def path_for(self, key: str) -> Path:
         return self.directory / key[:2] / f"{key}.json"
@@ -117,19 +126,48 @@ def get(self, key: str) -> Prediction | None:
         return prediction
 
     def put(self, key: str, prediction: Prediction) -> None:
+        """Store an answer, or count the failure to; never raise.
+
+        Each write goes through a temporary file of its own, so two writers of
+        one key cannot collide on a shared name. A write that still fails (a
+        virus scanner holding the file on Windows, a full disk) is counted in
+        ``stats`` and dropped: the answer is already in hand and paid for, and a
+        missing entry costs one more call on a later run, not this run.
+        """
         if not self.enabled:
             return
         path = self.path_for(key)
-        path.parent.mkdir(parents=True, exist_ok=True)
         payload = {"version": CACHE_FORMAT_VERSION, "prediction": to_jsonable(prediction)}
-        temporary = path.with_suffix(".tmp")
-        temporary.write_text(json.dumps(payload, sort_keys=True, default=str), encoding="utf-8")
-        temporary.replace(path)
+        temporary: Path | None = None
+        try:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            with tempfile.NamedTemporaryFile(
+                "w",
+                encoding="utf-8",
+                dir=path.parent,
+                prefix=f"{key}.",
+                suffix=".tmp",
+                delete=False,
+            ) as handle:
+                temporary = Path(handle.name)
+                handle.write(json.dumps(payload, sort_keys=True, default=str))
+            os.replace(temporary, path)
+        except OSError:
+            self.write_errors += 1
+            if temporary is not None:
+                with contextlib.suppress(OSError):
+                    temporary.unlink()
+            return
         self.writes += 1
 
     @property
     def stats(self) -> dict[str, int]:
-        return {"hits": self.hits, "misses": self.misses, "writes": self.writes}
+        return {
+            "hits": self.hits,
+            "misses": self.misses,
+            "writes": self.writes,
+            "write_errors": self.write_errors,
+        }
 
 
 def to_jsonable(prediction: Prediction) -> dict[str, Any]:

src/plumbline/runner/execute.py

@@ -24,6 +24,7 @@
 
 import hashlib
 import json
+import threading
 import time
 from collections.abc import Callable, Mapping, Sequence
 from concurrent.futures import ThreadPoolExecutor
@@ -524,9 +525,10 @@ def run(
     estimate = check_guard(cases, guard, pricing)
 
     records: list[CaseRecord | None] = [None] * len(cases)
+    locks = _KeyLocks() if cache is not None and cache.enabled else None
 
     def handle(index: int) -> None:
-        records[index] = _one_case(cases[index], adapter, cache, retry, table)
+        records[index] = _one_case(cases[index], adapter, cache, retry, table, locks)
 
     if workers == 1:
         for index in range(len(cases)):
@@ -585,16 +587,51 @@ def handle(index: int) -> None:
     )
 
 
+class _KeyLocks:
+    """One lock per cache key, handed out on demand to the worker threads.
+
+    Two cases with the same text share a key. Without a lock both miss the
+    cache, both call the adapter, and the same answer is billed twice. Holding
+    the key's lock across look up, call, and store makes the second one wait
+    and then find the first one's answer. Different keys never wait on each
+    other.
+    """
+
+    def __init__(self) -> None:
+        self._guard = threading.Lock()
+        self._locks: dict[str, threading.Lock] = {}
+
+    def for_key(self, key: str) -> threading.Lock:
+        with self._guard:
+            return self._locks.setdefault(key, threading.Lock())
+
+
 def _one_case(
     case: Case,
     adapter: Adapter,
     cache: Cache | None,
     retry: RetryPolicy,
     table: PricingTable,
+    locks: _KeyLocks | None = None,
 ) -> CaseRecord:
     labels = list(case.labels)
     key = cache_key(adapter, case.text, labels, case.question_type) if cache is not None else None
+    if key is not None and locks is not None:
+        with locks.for_key(key):
+            return _answer(case, labels, key, adapter, cache, retry, table)
+    return _answer(case, labels, key, adapter, cache, retry, table)
+
 
+def _answer(
+    case: Case,
+    labels: list[str],
+    key: str | None,
+    adapter: Adapter,
+    cache: Cache | None,
+    retry: RetryPolicy,
+    table: PricingTable,
+) -> CaseRecord:
+    """From the cache if it holds the answer, otherwise from the adapter, with retries."""
     if cache is not None and key is not None:
         hit = cache.get(key)
         if hit is not None:

tests/test_runner.py

@@ -172,11 +172,74 @@ def test_reordering_labels_does_not_invalidate_the_cache() -> None:
     assert forward == backward
 
 
+class CountingAdapter(Adapter):
+    """Answers through the mock, counting calls across threads, a little slowly."""
+
+    def __init__(self, inner: Adapter) -> None:
+        import threading
+
+        self.inner = inner
+        self.name = inner.name
+        self.model_requested = inner.model_requested
+        self.revision = inner.revision
+        self.probability_semantics = inner.probability_semantics
+        self.calls = 0
+        self._lock = threading.Lock()
+
+    @property
+    def call_params(self):  # type: ignore[override]
+        return self.inner.call_params
+
+    def classify(self, text: str, labels: list[str], **asked: object) -> Prediction:
+        import time
+
+        with self._lock:
+            self.calls += 1
+        time.sleep(0.02)  # long enough for every worker to arrive at once
+        return self.inner.classify(text, labels, **asked)  # type: ignore[arg-type]
+
+
+def test_identical_cases_under_many_workers_are_answered_once(tmp_path: Path) -> None:
+    """One key is one billed call, and duplicates writing it at once do not crash (#27)."""
+    import dataclasses
+
+    base = make_cases(1, labels=LABELS)[0]
+    cases = [dataclasses.replace(base, id=f"dup-{i}") for i in range(16)]
+    adapter = CountingAdapter(an_adapter(cases))
+    cache = Cache(tmp_path)
+
+    result = execute.run(adapter, cases, cache=cache, workers=16)
+
+    assert adapter.calls == 1
+    assert all(record.ok for record in result.records)
+    assert sum(record.from_cache for record in result.records) == 15
+
+
+def test_a_cache_write_that_fails_does_not_end_the_run(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The answer is already paid for; losing the cache entry must not lose the run."""
+    from plumbline.runner import cache as cache_module
+
+    def refuse(*_: object) -> None:
+        raise PermissionError(32, "The process cannot access the file")
+
+    monkeypatch.setattr(cache_module.os, "replace", refuse)
+    cases = make_cases(5, labels=LABELS)
+    cache = Cache(tmp_path)
+
+    result = execute.run(an_adapter(cases), cases, cache=cache, workers=4)
+
+    assert all(record.ok for record in result.records)
+    assert cache.stats["write_errors"] == 5
+    assert not list(tmp_path.rglob("*.tmp"))
+
+
 def test_a_disabled_cache_never_reads_or_writes(tmp_path: Path) -> None:
     cases = make_cases(20, labels=LABELS)
     cache = Cache(tmp_path / "cache", enabled=False)
     execute.run(an_adapter(cases), cases, cache=cache, workers=2)
-    assert cache.stats == {"hits": 0, "misses": 0, "writes": 0}
+    assert cache.stats == {"hits": 0, "misses": 0, "writes": 0, "write_errors": 0}
     assert not (tmp_path / "cache").exists()