fix(inference): concurrent suggests encode once and build one provider

src/visionset/inference/cache.py

@@ -24,6 +24,7 @@
 
 from __future__ import annotations
 
+import threading
 from collections import OrderedDict
 from typing import Final
 
@@ -50,13 +51,20 @@
 
 
 class BoundedCache[K, V]:
-    """Bounded, least-recently-used, and deliberately not thread safe.
+    """Bounded, least-recently-used, and safe to share between threads.
 
-    Not thread safe for the same reason ``LocalTransformersProvider`` is not: a
-    worker process runs one task at a time, and a server handler holding a
-    model is already serialised by the device it is talking to. A lock here
-    would buy nothing and would suggest a concurrency this design does not
-    have.
+    **It did not used to be, and the assumption behind that was wrong.** This
+    was written as "deliberately not thread safe" on the reasoning that a worker
+    process runs one task at a time and a server handler is serialised by the
+    device it talks to. Neither holds: the suggest route is a plain ``def``, so
+    FastAPI runs concurrent requests in parallel threadpool threads, and two of
+    them reaching one cache is the ordinary case rather than an exotic one.
+
+    The lock here protects the container's own integrity — two threads calling
+    ``move_to_end`` for *different* keys are still mutating one linked list — and
+    nothing more. It is never held across a computation, so it cannot serialise
+    two encodes; stopping the same value from being computed twice is a separate
+    problem with a separate answer, :class:`KeyedLocks` at the caller.
 
     ``get`` counts as a use, which is what makes this LRU rather than
     first-in-first-out: the asset somebody is clicking on repeatedly is the one
@@ -69,6 +77,7 @@ def __init__(self, capacity: int) -> None:
             raise ValueError(f"a cache holding {capacity} things is not a cache")
         self._capacity = capacity
         self._held: OrderedDict[K, V] = OrderedDict()
+        self._lock = threading.Lock()
 
     @property
     def capacity(self) -> int:
@@ -77,35 +86,78 @@ def capacity(self) -> int:
 
     def get(self, key: K) -> V | None:
         """What is held under that key, or ``None`` — and a hit is a use."""
-        if key not in self._held:
-            return None
-        self._held.move_to_end(key)
-        return self._held[key]
+        with self._lock:
+            if key not in self._held:
+                return None
+            self._held.move_to_end(key)
+            return self._held[key]
 
     def put(self, key: K, value: V) -> V:
         """Hold that, evicting the least recently used if the bound is reached.
 
         Returns the value, so a caller can write ``return cache.put(k, compute())``
         rather than putting and then reading back.
         """
-        if key in self._held:
-            self._held.move_to_end(key)
-        self._held[key] = value
-        while len(self._held) > self._capacity:
-            self._held.popitem(last=False)
+        with self._lock:
+            if key in self._held:
+                self._held.move_to_end(key)
+            self._held[key] = value
+            while len(self._held) > self._capacity:
+                self._held.popitem(last=False)
         return value
 
     def discard(self, key: K) -> None:
         """Forget that key if it is held. A no-op if it is not."""
-        self._held.pop(key, None)
+        with self._lock:
+            self._held.pop(key, None)
 
     def clear(self) -> None:
         """Forget everything."""
-        self._held.clear()
+        with self._lock:
+            self._held.clear()
 
     def __len__(self) -> int:
-        return len(self._held)
+        with self._lock:
+            return len(self._held)
 
     def __contains__(self, key: object) -> bool:
         """Membership **without** counting as a use, for tests that assert eviction."""
-        return key in self._held
+        with self._lock:
+            return key in self._held
+
+
+class KeyedLocks[K]:
+    """One lock per key, so that a value is computed once and only once.
+
+    **What a bounded cache alone cannot do.** A cache tells a caller whether a
+    value is *there*; it has nothing to say about a value that is on its way. Two
+    threads asking for the same un-cached thing both see nothing and both compute
+    it, which is how one image came to be encoded twice and one model came to be
+    loaded four times in a single process. The remedy is to make the second
+    thread wait for the first rather than duplicate it, and a lock held across
+    the computation is what waiting means.
+
+    **Per key, because a global lock would be a different bug.** Holding one lock
+    across every encode would make a click on one asset wait for an unrelated
+    click on another — turning a duplicated-work problem into a queueing problem
+    and costing exactly the latency the cache exists to save. Two keys never
+    contend here; only two callers wanting the same key do.
+
+    The map grows with distinct keys seen in a process and is never pruned. That
+    is one ``threading.Lock`` — tens of bytes — per asset anybody has ever
+    suggested on, against the 16 MB the same asset's embedding occupies while it
+    is cached, so the bound that matters is already elsewhere.
+    """
+
+    def __init__(self) -> None:
+        self._guard = threading.Lock()
+        self._locks: dict[K, threading.Lock] = {}
+
+    def for_key(self, key: K) -> threading.Lock:
+        """The lock for that key, made on first ask.
+
+        The guard is held only for the lookup, never for the work the returned
+        lock goes on to protect.
+        """
+        with self._guard:
+            return self._locks.setdefault(key, threading.Lock())

src/visionset/inference/providers.py

@@ -41,7 +41,7 @@
 from typing import Any, Final
 
 from visionset.inference._extra import require
-from visionset.inference.cache import DEFAULT_PROVIDER_CAPACITY, BoundedCache
+from visionset.inference.cache import DEFAULT_PROVIDER_CAPACITY, BoundedCache, KeyedLocks
 from visionset.inference.families import (
     DETECTOR_FAMILIES,
     SEGMENTER_FAMILIES,
@@ -82,6 +82,7 @@ class ProviderPool:
 
     def __init__(self, capacity: int = DEFAULT_PROVIDER_CAPACITY) -> None:
         self._held: BoundedCache[_Key, Runner] = BoundedCache(capacity)
+        self._building: KeyedLocks[_Key] = KeyedLocks()
         self._builds = 0
 
     @property
@@ -99,14 +100,25 @@ def get(self, connection: InferenceConnection, *, workspace_root: Path) -> Runne
         Every refusal ``provider_for`` can raise is raised here too, and raised
         *before* anything is cached: a connection that is not ready must not
         leave a half-answer behind for the request that follows its download.
+
+        **Built once even when the first several clicks arrive together.** The
+        window between finding nothing here and storing what was built is wide —
+        a build reads a config off disk and the adapter then loads gigabytes of
+        weights — and four concurrent first clicks went through it four times in
+        one process. The lock is per connection, so two connections still build
+        in parallel, and a refusal still caches nothing.
         """
         key = (str(connection.id), connection.updated_at.isoformat())
         held = self._held.get(key)
         if held is not None:
             return held
-        built = provider_for(connection, workspace_root=workspace_root)
-        self._builds += 1
-        return self._held.put(key, built)
+        with self._building.for_key(key):
+            held = self._held.get(key)
+            if held is not None:
+                return held
+            built = provider_for(connection, workspace_root=workspace_root)
+            self._builds += 1
+            return self._held.put(key, built)
 
     def clear(self) -> None:
         """Drop everything held. What a test does between cases."""

src/visionset/inference/sam_provider.py

@@ -48,7 +48,7 @@
 
 from visionset.inference import _fp16
 from visionset.inference._extra import imported
-from visionset.inference.cache import DEFAULT_EMBEDDING_CAPACITY, BoundedCache
+from visionset.inference.cache import DEFAULT_EMBEDDING_CAPACITY, BoundedCache, KeyedLocks
 from visionset.kernel.domain import (
     AssetSegmentation,
     PointPrompt,
@@ -146,6 +146,7 @@ def __init__(
         self._embeddings: BoundedCache[UUID, tuple[Any, tuple[int, int]]] = BoundedCache(
             embedding_capacity
         )
+        self._encoding: KeyedLocks[UUID] = KeyedLocks()
         self._encodes = 0
 
     @property
@@ -266,24 +267,63 @@ def _embedding(
         content-addressed: the bytes behind an id cannot change, so a hit can
         never be stale. Editing the *connection* is what would invalidate these,
         and that replaces the whole provider rather than reaching in here.
+
+        **Once means once even when two clicks arrive together.** The route is a
+        plain ``def``, so FastAPI answers concurrent suggests in parallel
+        threadpool threads; a bare check-then-compute let two clicks on the same
+        un-encoded asset both encode it, which is the most expensive thing this
+        adapter does. The lock is taken per asset, so a click on another asset
+        neither waits for this one nor is waited for.
+
+        The cache is read twice on purpose. The first read is the common case and
+        takes no lock at all; the second is what the loser of a race sees, and
+        without it the winner's work would be redone by everybody who queued
+        behind it.
         """
         held = self._embeddings.get(target.asset_id)
         if held is not None:
             return held
-        image = Image.open(BytesIO(target.content)).convert("RGB")
-        size = (image.height, image.width)
-        inputs = processor(images=image, return_tensors="pt").to(device)
-        self._encodes += 1
-        return self._embeddings.put(
-            target.asset_id, (model.get_image_embeddings(inputs["pixel_values"]), size)
-        )
+        with self._encoding.for_key(target.asset_id):
+            held = self._embeddings.get(target.asset_id)
+            if held is not None:
+                return held
+            image = Image.open(BytesIO(target.content)).convert("RGB")
+            size = (image.height, image.width)
+            inputs = processor(images=image, return_tensors="pt").to(device)
+            self._encodes += 1
+            return self._embeddings.put(
+                target.asset_id, (model.get_image_embeddings(inputs["pixel_values"]), size)
+            )
 
     def _ready(self) -> tuple[Any, Any, str, bool]:
         if self._loaded is None:
             self._loaded = self._load()
         return self._loaded
 
     def _load(self) -> tuple[Any, Any, str, bool]:
+        """Processor and model, once per provider.
+
+        **``transformers`` warns here on every load, and the warning is expected.**
+        The published SAM 2 checkpoints declare ``model_type: sam2_video``, so
+        loading one into ``Sam2Model`` prints *"You are using a model of type
+        ``sam2_video`` to instantiate a model of type ``sam2``"*. Why that is the
+        right class rather than a mistake is argued where the families are
+        declared, in ``families.py``; what is worth recording at the load site is
+        that it was **measured** and not assumed. Asking
+        ``from_pretrained(..., output_loading_info=True)`` for
+        ``facebook/sam2.1-hiera-base-plus`` reports ``missing_keys: 0``,
+        ``unexpected_keys: 0``, ``mismatched_keys: 0`` and no errors — every
+        parameter this class needs came out of the checkpoint and nothing in the
+        checkpoint went unused, so no weight is left randomly initialised.
+
+        The alternative class does not fit: ``Sam2VideoModel.forward`` takes an
+        ``inference_session`` and a frame index, which is the video-tracking path
+        and has no way to answer a point on a single image.
+
+        The warning is therefore left where a reader can see it. Silencing it
+        would hide the same sentence on the day a checkpoint genuinely does not
+        match, and that day it is the only warning there is.
+        """
         torch = imported("torch")
         transformers = imported("transformers")
         device, half = self._resolved_device(torch)

tests/inference/test_cache.py

@@ -12,6 +12,7 @@
     DEFAULT_EMBEDDING_CAPACITY,
     DEFAULT_PROVIDER_CAPACITY,
     BoundedCache,
+    KeyedLocks,
 )
 
 
@@ -88,3 +89,19 @@ def test_the_shipped_capacities_leave_room_for_the_co_residency_the_design_assum
     """Two providers is a detector and a segmenter, which is what D1 describes."""
     assert DEFAULT_PROVIDER_CAPACITY >= 2
     assert DEFAULT_EMBEDDING_CAPACITY >= 2
+
+
+# --- the single-flight primitive ----------------------------------------------
+
+
+def test_one_key_answers_with_one_lock() -> None:
+    """The whole of single-flight: two callers asking about the same thing wait on
+    the same object. A fresh lock per call would let both compute."""
+    locks: KeyedLocks[str] = KeyedLocks()
+    assert locks.for_key("a") is locks.for_key("a")
+
+
+def test_two_keys_answer_with_different_locks() -> None:
+    """Per key, so an encode of one asset never queues behind another's."""
+    locks: KeyedLocks[str] = KeyedLocks()
+    assert locks.for_key("a") is not locks.for_key("b")