diff --git a/lacing/store/mapping.py b/lacing/store/mapping.py index bf117fd6..7bca91d2 100644 --- a/lacing/store/mapping.py +++ b/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) diff --git a/lacing/store/memory.py b/lacing/store/memory.py index ec875481..a34c9469 100644 --- a/lacing/store/memory.py +++ b/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 --------------------------------------- diff --git a/tests/test_store_postgres.py b/tests/test_store_postgres.py index 0649d938..0d9ab88f 100644 --- a/tests/test_store_postgres.py +++ b/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"] diff --git a/tests/test_store_update.py b/tests/test_store_update.py index 7b486699..787ce077 100644 --- a/tests/test_store_update.py +++ b/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)