Stores: lock MappingStore setitem/delitem, snapshot iteration (#68)

lacing/store/mapping.py

@@ -161,6 +161,10 @@ def _put_tiers(self, tiers: Iterable[Tier]) -> None:
     # --- MutableMapping interface ---------------------------------------------
 
     def __setitem__(self, key: TimeInterval, value: list[Annotation]) -> None:
+        with self._lock:
+            self._set_bucket(key, value)
+
+    def _set_bucket(self, key: TimeInterval, value: list[Annotation]) -> None:
         value = list(value)
         for ann in value:
             if ann.interval != key:
@@ -187,6 +191,10 @@ def __setitem__(self, key: TimeInterval, value: list[Annotation]) -> None:
         super().__setitem__(key, value)
 
     def __delitem__(self, key: TimeInterval) -> None:
+        with self._lock:
+            self._del_bucket(key)
+
+    def _del_bucket(self, key: TimeInterval) -> None:
         bucket = self._buckets.get(_bucket_key(key))
         if bucket is None:
             raise KeyError(key)

lacing/store/memory.py

@@ -107,14 +107,20 @@ def __delitem__(self, key: TimeInterval) -> None:
             self._drop_node(k)
 
     def __iter__(self) -> Iterator[TimeInterval]:
-        for ann_list in self._buckets.values():
-            if ann_list:
-                iv = ann_list[0].interval
-                if iv is not None:
-                    yield iv
+        # Iterate a snapshot taken under the lock, so a concurrent ``add`` /
+        # ``update`` / ``remove`` cannot raise "dictionary changed size during
+        # iteration".
+        with self._lock:
+            intervals = [
+                b[0].interval
+                for b in self._buckets.values()
+                if b and b[0].interval is not None
+            ]
+        yield from intervals
 
     def __len__(self) -> int:
-        return sum(1 for ann_list in self._buckets.values() if ann_list)
+        with self._lock:
+            return sum(1 for ann_list in self._buckets.values() if ann_list)
 
     # --- annotation-level convenience ---------------------------------------
 

tests/test_store_postgres.py

@@ -818,3 +818,62 @@ def reader():
             t.join()
             reader_store.close()
         assert missing == []
+
+    def test_moving_update_under_concurrent_readers(self, postgres_store, postgresql):
+        """A moving ``update`` (interval changes each time) never opens a gap (lacing#68).
+
+        The reader looks for the id by whole-store scan *and* by an interval
+        query covering both places the annotation lives, so a gap in either the
+        row or the range index shows up.
+        """
+        import threading
+
+        info = postgresql.info
+        reader_store = PostgresStore(
+            {
+                "host": info.host,
+                "port": info.port,
+                "user": info.user,
+                "password": info.password or "",
+                "dbname": info.dbname,
+            },
+            rate=24000,
+            use_pool=False,
+        )
+        target = _ann(_ti(0, 2), text="0")
+        postgres_store.add(target)
+        everywhere = _ti(0, 100)
+        missing: list[str] = []
+        stop = threading.Event()
+
+        def reader():
+            try:
+                while not stop.is_set():
+                    if target.id not in {a.id for a in reader_store.all()}:
+                        missing.append("all()")
+                    if target.id not in {
+                        a.id for a in reader_store.intersects(everywhere)
+                    }:
+                        missing.append("intersects()")
+            except Exception as exc:  # noqa: BLE001 - a dead reader must fail the test
+                missing.append(f"reader died: {exc!r}")
+
+        t = threading.Thread(target=reader)
+        t.start()
+        try:
+            for i in range(1, 101):
+                iv = _ti(10, 12) if i % 2 else _ti(0, 2)
+                postgres_store.update(
+                    target.model_copy(
+                        update={
+                            "reference": MediaRef(asset_id="blake3:test", interval=iv),
+                            "body": {"text": str(i)},
+                        }
+                    )
+                )
+        finally:
+            stop.set()
+            t.join()
+            reader_store.close()
+        assert missing == []
+        assert [a.body["text"] for a in postgres_store.all()] == ["100"]

tests/test_store_update.py

@@ -151,3 +151,68 @@ def test_stores_with_a_lock_still_copy_and_pickle():
     for clone in (copy.deepcopy(s), pickle.loads(pickle.dumps(s))):
         assert [a.id for a in clone.all()] == [a.id for a in s.all()]
         clone.update(next(iter(clone.all())).model_copy(update={"body": {"text": "y"}}))
+
+
+@pytest.mark.parametrize("kind", ["memory", "mapping"])
+def test_iterating_while_update_moves_an_annotation(kind):
+    """``__iter__`` works from a snapshot: no "dictionary changed size" (lacing#68)."""
+    store = MemoryStore() if kind == "memory" else MappingStore({})
+    store.add_tier(Tier("words"))
+    target = _ann(_ti(0, 2), text="0")
+    store.add(target)
+    errors: list[BaseException] = []
+    stop = threading.Event()
+
+    def iterator():
+        try:
+            while not stop.is_set():
+                list(store)
+                len(store)
+        except Exception as exc:  # noqa: BLE001 - reported by the assert below
+            errors.append(exc)
+
+    old_interval = sys.getswitchinterval()
+    sys.setswitchinterval(1e-6)
+    t = threading.Thread(target=iterator)
+    t.start()
+    try:
+        for i in range(1, N_REPLACEMENTS + 1):
+            # A fresh interval each time grows and shrinks the bucket dict.
+            store.update(_ann(_ti(10 * i, 10 * i + 2), id=target.id, text=str(i)))
+    finally:
+        stop.set()
+        t.join()
+        sys.setswitchinterval(old_interval)
+    assert errors == []
+
+
+def test_mapping_store_delitem_and_setitem_vs_update_keep_mapping_consistent():
+    """A racing ``update`` must not bring a deleted key back in the mapping (lacing#68)."""
+    backing: dict = {}
+    store = MappingStore(backing)
+    iv = _ti(0, 2)
+    old_interval = sys.getswitchinterval()
+    sys.setswitchinterval(1e-6)
+    try:
+        for _ in range(100):
+            target = _ann(iv, text="0")
+            store[iv] = [target]
+            stop = threading.Event()
+
+            def updater():
+                while not stop.is_set():
+                    try:
+                        store.update(_ann(iv, id=target.id, text="u"))
+                    except KeyError:
+                        return  # deleted: update must refuse from here on
+
+            t = threading.Thread(target=updater)
+            t.start()
+            del store[iv]
+            stop.set()
+            t.join()
+            ids_in_store = {str(a.id) for a in store.all()}
+            ids_in_mapping = {k for k in backing if k != store._tiers_key}
+            assert ids_in_mapping == ids_in_store == set()
+    finally:
+        sys.setswitchinterval(old_interval)