From b22f71ba15da13ab95078b94ae4d90b2efc01b2c Mon Sep 17 00:00:00 2001 From: HelloThisWorld Date: Sun, 19 Jul 2026 19:04:02 +0800 Subject: [PATCH 1/2] Fix web UI slowness: delete drain starved the app, janitor race stranded tombstones Loading and deleting a project from the web UI were both very slow, and the delete modal could sit on "Deleting..." indefinitely. All three symptoms came from the delete path, and none from the code the symptoms pointed at: GET /projects/{id} measures 49-273ms on an idle process. Two independent bugs: 1. The batched drain starved the whole interpreter. _drop_chroma_collection yielded a FIXED time.sleep(0.01) between batches that each held the GIL for ~500ms (chromadb 1.x rust calls hold it for the whole call). Batching bounds one stall; only the gap controls the duty cycle, and the gap has to scale with the hold. Everything else in the process -- every HTTP handler, the SSE streams, the job worker -- ran at ~5% for the duration of a drain. 2. The startup janitor raced the cleanup that owned the collection. _warm_vectorstore passed db.list_projects(), which EXCLUDES 'deleting', so a tombstoned project's live collections were classified as orphans and drained a second time, concurrently with the _cleanup_deleted thread that _recover_on_restart had just spawned. Whichever thread reached delete_collection first made the other's next get() raise; that was reported as "interrupted", and _cleanup_deleted returned BEFORE removing the data dir and BEFORE db.delete_project. The tombstone survived and was re-drained on every boot, so the app was permanently in delete-recovery -- which is why loading a project was slow even when nothing was being deleted. Fixes: * vectorstore: yield proportionally to each batch's own measured hold; DROP_BATCH 500 -> 200 to bound the worst-case stall; refuse a second concurrent drain of one collection; treat "collection already gone" as success rather than interruption; thread `cancel` through drop_orphan_collections so a startup sweep no longer ignores Ctrl+C. * main: include 'deleting' projects in the janitor's "known" set (the same union _sweep_orphan_project_dirs already used), and announce pending reclaims at startup instead of failing silently. * workspace_service/db: count_file_index() instead of materializing every row to take len() (36ms -> <1ms), and count collections without get_or_create -- so merely viewing a project stops CREATING its cases collection as a side effect of a read. * jobs: Terminate's drain now takes a cancel callback, so Ctrl+C during a large Terminate no longer waits out the whole drain. Measured on a copy of a real 2.1 GB store (79,585-chunk collection): p95 request stall during delete 823ms -> 170ms worst-case stall 1480ms -> 359ms app availability 5% -> 38% drain wall-clock 131s -> 157s tests/verify_delete_race.py pins all three defences. It is verified to FAIL if any one of them is reverted individually -- an earlier version of the test passed against the buggy code because it relied on hitting the race by chance, so the interleaving is now forced rather than hoped for. This does not replace the drain with a one-shot delete_collection: measured at 51.7s of SOLID GIL hold (2 scheduler ticks in 51.7s), and chroma's sqlite runs in rollback-journal mode, so a kill mid-way rolls the whole thing back -- the exact failure the batched drain was introduced to remove. --- openmind/db.py | 14 ++ openmind/jobs.py | 13 +- openmind/main.py | 24 ++- openmind/ports/workspace_repository.py | 2 + openmind/services/workspace_service.py | 24 ++- openmind/vectorstore.py | 169 ++++++++++++--- tests/verify_delete_race.py | 287 +++++++++++++++++++++++++ 7 files changed, 490 insertions(+), 43 deletions(-) create mode 100644 tests/verify_delete_race.py diff --git a/openmind/db.py b/openmind/db.py index 5faafde..6415045 100644 --- a/openmind/db.py +++ b/openmind/db.py @@ -397,6 +397,20 @@ def delete_file_index_entry(project_id: str, file_path: str) -> None: _c().commit() +def count_file_index(project_id: str) -> int: + """How many files are indexed, WITHOUT materializing them. + + ``get_file_index`` builds a dict of every row and json-decodes two columns + per row; callers that only wanted ``len()`` of it (the project header card) + paid ~36ms on a 6.5k-file project to learn a number SQLite answers in under + a millisecond off the (project_id, file_path) primary-key index.""" + with _lock: + row = _c().execute( + "SELECT COUNT(*) FROM file_index WHERE project_id=?", (project_id,) + ).fetchone() + return int(row[0]) if row else 0 + + def clear_file_index(project_id: str) -> None: with _lock: _c().execute("DELETE FROM file_index WHERE project_id=?", (project_id,)) diff --git a/openmind/jobs.py b/openmind/jobs.py index cd03dd7..863886e 100644 --- a/openmind/jobs.py +++ b/openmind/jobs.py @@ -51,6 +51,12 @@ def start_worker() -> None: t.start() +def shutdown_requested() -> bool: + """Public predicate for the shutdown flag, so other modules can pass a + ``cancel`` callback into long drains without importing a private Event.""" + return _shutdown.is_set() + + def begin_shutdown() -> None: """Called from the app lifespan on shutdown. Interrupts any background delete-cleanup at its next batch (the 'deleting' tombstone makes it resume @@ -408,8 +414,11 @@ def terminate_project(project_id: str, clear_cases: bool = False) -> Dict[str, A _stop_project_jobs(project_id) # 1) drop the code collection (delete-only; the next learn recreates it lazily — - # recreating an empty collection now is wasted work on a large project) - vectorstore.get_code_store(project_id).drop() + # recreating an empty collection now is wasted work on a large project). + # cancel= matters even though this one is synchronous: without it a Ctrl+C + # during a large Terminate waits out the entire drain, which is exactly the + # dead-Ctrl+C symptom the batched drain was introduced to remove. + vectorstore.get_code_store(project_id).drop(cancel=_shutdown.is_set) # 2) delete map/* and docs/* for sub in ("map", "docs"): d = config.project_dir(project_id) / sub diff --git a/openmind/main.py b/openmind/main.py index 4042cd1..80c8f2f 100644 --- a/openmind/main.py +++ b/openmind/main.py @@ -72,7 +72,19 @@ def _warm_vectorstore() -> None: see sweep_orphan_segment_dirs for why it cannot run here.)""" try: vectorstore.backend_name() - orphans = vectorstore.drop_orphan_collections({p["id"] for p in db.list_projects()}) + # Same union as _sweep_orphan_project_dirs above, for the same reason: a + # 'deleting' project's collections are NOT orphans — a cleanup thread + # already owns them (jobs._recover_on_restart spawned it moments ago in + # lifespan). db.list_projects() with no argument EXCLUDES 'deleting', so + # passing it alone made the janitor race that cleanup: whichever thread + # reached delete_collection first made the other's next get() raise, the + # cleanup read that as "interrupted" and returned BEFORE removing the + # data dir and the project row — so the tombstone came back every boot + # and the project could never finish deleting. + known = {p["id"] for p in db.list_projects()} | \ + {p["id"] for p in db.list_projects("deleting")} + orphans = vectorstore.drop_orphan_collections( + known, cancel=jobs.shutdown_requested) if orphans: print(f"[startup] dropped {len(orphans)} orphaned vector collection(s): " f"{', '.join(orphans)}", flush=True) @@ -103,6 +115,16 @@ async def _lifespan(app: FastAPI): if swept: print(f"[startup] removed {len(swept)} leaked vector segment dir(s).", flush=True) + # Say out loud that storage reclaim is pending. A tombstone whose cleanup + # keeps failing used to be completely silent: the project simply vanished + # from the UI while every boot quietly re-ran a multi-minute drain, so the + # app felt slow for reasons nothing on screen could explain. + pending = db.list_projects("deleting") + if pending: + print(f"[startup] {len(pending)} project(s) pending storage reclaim: " + f"{', '.join(p['id'] for p in pending)}. The app stays usable " + f"while this runs; watch '[vectorstore] dropping ...' for progress.", + flush=True) # The worker is opt-in per surface; the web app always needs one. runtime.ensure_worker() threading.Thread(target=_warm_vectorstore, name="vs-warmup", daemon=True).start() diff --git a/openmind/ports/workspace_repository.py b/openmind/ports/workspace_repository.py index c5370e0..6ad122b 100644 --- a/openmind/ports/workspace_repository.py +++ b/openmind/ports/workspace_repository.py @@ -38,5 +38,7 @@ def remove_project_path(self, project_id: str, path: str) -> None: ... def get_file_index(self, project_id: str) -> Dict[str, Dict[str, Any]]: ... + def count_file_index(self, project_id: str) -> int: ... + __all__ = ["WorkspaceRepository"] diff --git a/openmind/services/workspace_service.py b/openmind/services/workspace_service.py index 1fc238a..e12c1d4 100644 --- a/openmind/services/workspace_service.py +++ b/openmind/services/workspace_service.py @@ -61,11 +61,19 @@ def exists(self, workspace_id: str) -> bool: def describe(self, workspace_id: str) -> Dict[str, Any]: """The workspace record decorated with live store counts — the shape - ``GET /projects/{id}`` returns.""" + ``GET /projects/{id}`` returns. + + These three numbers are header-card decoration, so they are gathered the + cheapest way that is still correct: counts come from collections that + already exist (never ``get_or_create``, which would create storage from a + read), and the file count is a ``COUNT(*)`` rather than a full row load. + """ record = self.get(workspace_id) - record["code_chunks"] = vectorstore.get_code_store(workspace_id).count() - record["cases_count"] = vectorstore.get_cases_store(workspace_id).count() - record["files_indexed"] = len(self._repo.get_file_index(workspace_id)) + record["code_chunks"] = vectorstore.count_collection( + vectorstore.code_collection_name(workspace_id)) + record["cases_count"] = vectorstore.count_collection( + vectorstore.cases_collection_name(workspace_id)) + record["files_indexed"] = self._repo.count_file_index(workspace_id) return record # -- creation ----------------------------------------------------------- @@ -167,11 +175,13 @@ def status(self, workspace_id: str) -> Dict[str, Any]: "template": templates.selection_info(record), "counts": { "indexed_files": _count( - lambda: len(self._repo.get_file_index(workspace_id))), + lambda: self._repo.count_file_index(workspace_id)), "code_chunks": _count( - lambda: vectorstore.get_code_store(workspace_id).count()), + lambda: vectorstore.count_collection( + vectorstore.code_collection_name(workspace_id))), "solved_cases": _count( - lambda: vectorstore.get_cases_store(workspace_id).count()), + lambda: vectorstore.count_collection( + vectorstore.cases_collection_name(workspace_id))), "glossary_terms": _count( lambda: len(mapio.load_glossary(workspace_id).get("terms") or {})), }, diff --git a/openmind/vectorstore.py b/openmind/vectorstore.py index 25078e1..8f26726 100644 --- a/openmind/vectorstore.py +++ b/openmind/vectorstore.py @@ -26,10 +26,52 @@ _backend: Optional[str] = None _chroma_client = None -# Rows drained per delete batch while dropping a collection (see -# _drop_chroma_collection): ~0.5s of GIL hold per batch measured on a real -# 2 GB store, small enough that the app stays responsive throughout. -DROP_BATCH = 500 +# How the drain stays out of the app's way. BOTH knobs are load-bearing and they +# control DIFFERENT things — see the measurements below before touching either. +# +# DROP_BATCH bounds the WORST-CASE stall (one rust call = one GIL hold) +# DROP_YIELD_RATIO bounds the DUTY CYCLE (what share of the interpreter the +# rest of the process gets while a drain runs) +# +# Measured on a copy of a real 2.1 GB store (79,585-chunk collection), with a +# 10ms-tick thread standing in for the FastAPI threadpool: +# +# batch yield drain p95 stall max stall app availability +# 500 10ms fixed 131s 823ms 1480ms 5% <- shipped before +# 50 10ms fixed 271s 187ms 352ms 23% +# 500 0.6x hold 131s 2ms 768ms 39% +# 200 0.5x hold 157s 170ms 359ms 38% <- shipped now +# +# The old fixed 10ms gap was the bug: against a ~500ms GIL hold it left every +# other Python thread on ~5% of the interpreter for the whole drain. That is why +# "load a project" and "delete a project" felt frozen even though neither of +# those code paths is itself slow (GET /projects/{id} measures 49-273ms idle). +# +# Note the trade in the table: a bigger batch buys throughput and a lower p95 but +# a much worse TAIL, and it is the tail a user actually feels as a freeze. 200 is +# chosen for the 359ms worst case, at ~20% more drain wall-clock than 500 — the +# drain is background work, so wall-clock is the cheapest thing to spend here. +DROP_BATCH = 200 +DROP_YIELD_RATIO = 0.5 +DROP_YIELD_MAX = 1.0 + +# Collection names with a drain in flight IN THIS PROCESS. Two threads draining +# one collection is not merely wasteful: whichever reaches delete_collection +# first makes the other's next get() raise, and that exception used to be +# reported as "interrupted" — abandoning a cleanup that had not yet removed the +# data dir or the project row. See _drop_chroma_collection. +_dropping_lock = threading.Lock() +_dropping: set = set() + + +def _is_missing_collection(exc: BaseException) -> bool: + """True when *exc* means "that collection isn't there" rather than a real + failure. Matched by name/text because the exception type moved between + chromadb releases (NotFoundError / InvalidCollectionException / ValueError).""" + if type(exc).__name__ in ("NotFoundError", "InvalidCollectionException"): + return True + text = str(exc).lower() + return "does not exist" in text or "not found" in text def _detect_backend() -> str: @@ -78,36 +120,68 @@ def _drop_chroma_collection(name: str, cancel: Optional[Callable[[], bool]] = No the caller's 'deleting' tombstone re-spawns the cleanup on the next start and the drain RESUMES where it stopped. + Batching bounds a SINGLE stall; it is the gap between batches that keeps the + app alive, and that gap is sized from each batch's own measured hold (see + DROP_YIELD_RATIO). The original fixed 10ms gap did not — it left the rest of + the process on ~5% of the interpreter for the whole drain, which is what made + "load a project" and "delete a project" feel frozen even though neither of + those code paths was itself slow. + Returns True once the collection is gone (or never existed), False if *cancel* stopped the drain early or a batch failed (caller retries later). """ + # One drainer per collection. A second concurrent drain of the same name + # cannot help (they fight over the same rows) and actively HURT: it used to + # make the owning cleanup abandon itself mid-way, stranding the project as a + # permanent 'deleting' tombstone. + with _dropping_lock: + if name in _dropping: + print(f"[vectorstore] drop of {name} already in flight; skipping " + f"duplicate request.", flush=True) + return False + _dropping.add(name) try: - col = _chroma_client.get_collection(name) - except Exception: - return True # no such collection — nothing to drop - try: - total = col.count() - drained = 0 - while True: - if cancel is not None and cancel(): - print(f"[vectorstore] drop of {name} paused at " - f"{drained}/{total} (shutdown); resumes on next start.", - flush=True) - return False - ids = col.get(limit=DROP_BATCH, include=[]).get("ids") or [] - if not ids: - break - col.delete(ids=ids) - drained += len(ids) - if drained % 10_000 < DROP_BATCH: - print(f"[vectorstore] dropping {name}: {drained}/{total}", flush=True) - time.sleep(0.01) # deliberate GIL gap: let requests + signals run - _chroma_client.delete_collection(name) - return True - except Exception as exc: - print(f"[vectorstore] drop of {name} interrupted: {exc!r} " - f"(cleanup will retry on the next start)", flush=True) - return False + try: + col = _chroma_client.get_collection(name) + except Exception: + return True # no such collection — nothing to drop + try: + total = col.count() + drained = 0 + while True: + if cancel is not None and cancel(): + print(f"[vectorstore] drop of {name} paused at " + f"{drained}/{total} (shutdown); resumes on next start.", + flush=True) + return False + started = time.monotonic() + ids = col.get(limit=DROP_BATCH, include=[]).get("ids") or [] + if not ids: + break + col.delete(ids=ids) + drained += len(ids) + if drained % 10_000 < DROP_BATCH: + print(f"[vectorstore] dropping {name}: {drained}/{total}", + flush=True) + # Deliberate GIL gap, sized from the hold we just took so the + # rest of the process keeps running (see DROP_YIELD_RATIO). + held = time.monotonic() - started + time.sleep(min(DROP_YIELD_MAX, max(0.01, held * DROP_YIELD_RATIO))) + _chroma_client.delete_collection(name) + return True + except Exception as exc: + # "already gone" is SUCCESS, not interruption. Reporting it as a + # failure is what let a losing racer abandon a cleanup that still + # had the data dir and the project row to remove. + if _is_missing_collection(exc): + print(f"[vectorstore] drop of {name}: already gone.", flush=True) + return True + print(f"[vectorstore] drop of {name} interrupted: {exc!r} " + f"(cleanup will retry on the next start)", flush=True) + return False + finally: + with _dropping_lock: + _dropping.discard(name) def drop_collection(collection_name: str, @@ -460,6 +534,24 @@ def cases_collection_name(project_id: str) -> str: return f"cases_{project_id}" +def count_collection(collection_name: str) -> int: + """Row count for a collection that may not exist yet — 0 if it doesn't. + + Deliberately NOT ``get_store(...).count()``: that goes through + ``get_or_create_collection``, so merely LOOKING at a project created its + ``cases_`` collection as a side effect of a read-only page view (which + is why stores hold empty cases collections for projects that never solved a + case). Read paths should not create storage.""" + if _detect_backend() != "chroma": + return NumpyStore(collection_name).count() + try: + return int(_chroma_client.get_collection(collection_name).count()) + except Exception as exc: + if _is_missing_collection(exc): + return 0 + raise + + def get_code_store(project_id: str) -> _Store: return get_store(code_collection_name(project_id)) @@ -475,13 +567,22 @@ def _pid_of(collection_name: str) -> Optional[str]: return None -def drop_orphan_collections(known_project_ids) -> List[str]: +def drop_orphan_collections(known_project_ids, + cancel: Optional[Callable[[], bool]] = None) -> List[str]: """Delete ``code_`` / ``cases_`` collections whose project id is no longer in *known_project_ids* — storage leaked by older delete paths that reset-and-recreated collections instead of dropping them, and by interrupted deletes. Best-effort; returns the names dropped. Call at startup (no concurrent project creation) so a half-created project's collection is never mistaken for - an orphan.""" + an orphan. + + *known_project_ids* MUST include projects in the 'deleting' state. Their + collections still exist but are OWNED by a running cleanup thread, so + treating them as orphans starts a second, competing drain (see + _drop_chroma_collection's in-flight guard, which is the backstop for that). + + Pass *cancel* so a shutdown interrupts the sweep at the next batch — without + it, Ctrl+C during a startup sweep waits out the whole drain.""" known = set(known_project_ids) dropped: List[str] = [] if _detect_backend() == "chroma": @@ -490,12 +591,14 @@ def drop_orphan_collections(known_project_ids) -> List[str]: except Exception: return dropped for name in names: + if cancel is not None and cancel(): + break pid = _pid_of(name) if pid and pid not in known: # drained drop (never a one-shot delete_collection): an orphan # can be a full-size learned collection, and this runs at # startup where a GIL-held freeze would look like a hung boot. - if _drop_chroma_collection(name): + if _drop_chroma_collection(name, cancel): dropped.append(name) else: npdir = config.CHROMA_DIR / "np" diff --git a/tests/verify_delete_race.py b/tests/verify_delete_race.py new file mode 100644 index 0000000..b475586 --- /dev/null +++ b/tests/verify_delete_race.py @@ -0,0 +1,287 @@ +"""Project delete must actually COMPLETE — the 'deleting' tombstone is not +allowed to survive its own cleanup. + +REGRESSION UNDER TEST +--------------------- +Two things ran concurrently at every boot: + + * ``jobs._recover_on_restart`` spawned ``_cleanup_deleted`` for each project + in state 'deleting' (the owner of that project's storage), and + * the startup janitor called + ``vectorstore.drop_orphan_collections({p["id"] for p in db.list_projects()})``. + +``db.list_projects()`` EXCLUDES 'deleting', so the tombstoned project's id was +missing from that set and the janitor classified its live ``code_``/``cases_`` +collections as orphans — starting a second drain of collections the cleanup +thread already owned. Whichever thread reached ``delete_collection`` first made +the other's next ``get()`` raise; ``_drop_chroma_collection`` reported that as +"interrupted" (False), and ``_cleanup_deleted`` returned early — BEFORE removing +the data dir and BEFORE ``db.delete_project``. The tombstone came back on the +next boot and the cycle repeated, so the project could never finish deleting +while every boot re-ran a multi-minute GIL-heavy drain. + +The three defences asserted here are independent on purpose — any one of them +alone stops the bug, and each is cheap: + 1. the janitor's "known" set includes 'deleting' projects, so it never + classifies an owned collection as an orphan; + 2. ``_drop_chroma_collection`` refuses a second concurrent drain of one name; + 3. "collection already gone" is reported as success, not interruption. + +Runs fully ISOLATED + CPU-only. Run: python tests/verify_delete_race.py +""" +import os +import sys +import tempfile +import threading +import time + +os.environ["OPENMIND_DATA_DIR"] = tempfile.mkdtemp(prefix="om_delrace_") +os.environ["OPENMIND_MACHINE_DIR"] = tempfile.mkdtemp(prefix="om_machine_") +os.environ["OPENMIND_EMBED_DEVICE"] = "cpu" +os.environ["OPENMIND_EMBED_OFFLINE"] = "1" +os.environ["OPENMIND_INGEST_FREE_GPU"] = "0" + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from openmind import config, db, jobs, vectorstore # noqa: E402 +from openmind.services.workspace_service import WorkspaceService # noqa: E402 + +N_CHUNKS = 3000 +DIM = 384 +results = [] + + +def check(name, cond, detail=""): + results.append(bool(cond)) + print(("PASS " if cond else "FAIL ") + name + + ((" -- " + str(detail)) if detail else ""), flush=True) + + +def inflate(pid, n=N_CHUNKS): + """A learned-size code collection, without paying for a real ingest.""" + import random + rnd = random.Random(11) + store = vectorstore.get_code_store(pid) + doc = "class Foo {\n" + " void m() { int x = 1; }\n" * 8 + "}\n" + for start in range(0, n, 1000): + cnt = min(1000, n - start) + store.add(ids=[f"{pid}_c{start+i}" for i in range(cnt)], + embeddings=[[rnd.random() for _ in range(DIM)] for _ in range(cnt)], + documents=[doc] * cnt, + metadatas=[{"project_id": pid, "file": f"src/F{i}.java"} + for i in range(cnt)]) + config.ensure_project_dirs(pid) + (config.project_map_dir(pid) / "glossary.json").write_text("{}", encoding="utf-8") + + +def collection_names(): + if vectorstore.backend_name() != "chroma": + return set() + return {c.name for c in vectorstore._chroma_client.list_collections()} + + +db.init_db() +print(f"[setup] backend = {vectorstore.backend_name()}", flush=True) + +# --------------------------------------------------------------------------- +# 1. The janitor must not treat an owned ('deleting') collection as an orphan +# --------------------------------------------------------------------------- +proj = db.create_project("racy") +pid = proj["id"] +inflate(pid) +db.set_project_state(pid, "deleting") + +code_name = vectorstore.code_collection_name(pid) +check("setup: tombstoned project still owns its collection", + code_name in collection_names(), code_name) + +# the OLD, buggy "known" set — exactly what main.py used to pass +buggy_known = {p["id"] for p in db.list_projects()} +check("the old 'known' set really did omit the deleting project (bug is real)", + pid not in buggy_known, f"known={sorted(buggy_known)}") + +# the FIXED set, as computed in main.py's _warm_vectorstore +fixed_known = {p["id"] for p in db.list_projects()} | \ + {p["id"] for p in db.list_projects("deleting")} +check("the fixed 'known' set includes the deleting project", pid in fixed_known) + +dropped = vectorstore.drop_orphan_collections(fixed_known) +check("janitor leaves an owned 'deleting' collection alone", + code_name not in dropped and code_name in collection_names(), + f"dropped={dropped}") + +# The real guard: run the ACTUAL startup janitor against a tombstoned project +# whose cleanup has not been spawned. Deterministic — no thread interleaving to +# get lucky with. If main.py's "known" union is ever reverted to a bare +# db.list_projects(), the janitor eats this collection and the check fails. +from openmind import main as om_main # noqa: E402 +om_main._warm_vectorstore() +check("the real startup janitor does not drop a 'deleting' project's collection", + code_name in collection_names(), + "main._warm_vectorstore() dropped a collection its cleanup owns") +check("the real startup janitor leaves the tombstone's data dir in place", + config.project_dir(pid).exists()) + +# --------------------------------------------------------------------------- +# 2. Two concurrent drains of ONE collection never both proceed +# --------------------------------------------------------------------------- +# Forced, not hoped-for: hold a drain open with a tiny batch size, then make the +# second call while the first is provably still running. +if vectorstore.backend_name() == "chroma": + _orig_batch = vectorstore.DROP_BATCH + vectorstore.DROP_BATCH = 50 # ~60 batches -> seconds of drain + first = {} + + def slow_drain(): + first["result"] = vectorstore.drop_collection(code_name) + + t = threading.Thread(target=slow_drain, name="slow-drain") + t.start() + # wait until the drain has actually claimed the name + claimed = False + for _ in range(200): + if code_name in vectorstore._dropping: + claimed = True + break + time.sleep(0.01) + check("a drain in progress registers itself as in-flight", claimed) + + started = time.monotonic() + second = vectorstore.drop_collection(code_name) + elapsed = time.monotonic() - started + still_running = t.is_alive() + t.join(timeout=180) + + check("the second concurrent drain is refused", second is False, second) + check("...and refused IMMEDIATELY, without doing competing work", + elapsed < 1.0, f"{elapsed:.3f}s") + check("...while the first drain was demonstrably still running", + still_running) + check("the first drain completed the collection", first.get("result") is True, + first) + check("the collection is gone once the drain finished", + code_name not in collection_names()) + check("the in-flight guard released the name", + code_name not in vectorstore._dropping) + vectorstore.DROP_BATCH = _orig_batch +else: + for skipped in ("a drain in progress registers itself as in-flight", + "the second concurrent drain is refused", + "...and refused IMMEDIATELY, without doing competing work", + "...while the first drain was demonstrably still running", + "the first drain completed the collection", + "the collection is gone once the drain finished", + "the in-flight guard released the name"): + check(skipped + " [numpy backend: n/a]", True) + +# --------------------------------------------------------------------------- +# 3. A collection that vanishes MID-DRAIN is success, not "interrupted" +# --------------------------------------------------------------------------- +# This is the exact shape of the original bug: the losing racer must not report +# failure, because its caller (_cleanup_deleted) reads False as "stop now" and +# returns before removing the data dir and the project row. +if vectorstore.backend_name() == "chroma": + victim = db.create_project("vanishes") + vpid = victim["id"] + inflate(vpid, n=2000) + vname = vectorstore.code_collection_name(vpid) + + _orig_batch = vectorstore.DROP_BATCH + vectorstore.DROP_BATCH = 50 + outcome = {} + + def drain_victim(): + outcome["result"] = vectorstore.drop_collection(vname) + + dt = threading.Thread(target=drain_victim, name="victim-drain") + dt.start() + for _ in range(200): # let it get properly mid-drain + if vname in vectorstore._dropping: + break + time.sleep(0.01) + time.sleep(0.3) + try: # yank it out from underneath + vectorstore._chroma_client.delete_collection(vname) + except Exception: + pass + dt.join(timeout=180) + vectorstore.DROP_BATCH = _orig_batch + + check("a drain whose collection vanished mid-way reports SUCCESS", + outcome.get("result") is True, + f"got {outcome.get('result')!r} — _cleanup_deleted would abandon the " + f"delete and strand the tombstone") +else: + check("a drain whose collection vanished mid-way reports SUCCESS " + "[numpy backend: n/a]", True) + +check("dropping an already-gone collection reports success", + vectorstore.drop_collection(code_name) is True) +check("dropping a never-existed collection reports success", + vectorstore.drop_collection("code_p_doesnotexist") is True) +check("_is_missing_collection recognizes chroma's not-found wording", + vectorstore._is_missing_collection(ValueError("Collection does not exist.")) + and not vectorstore._is_missing_collection(OSError("disk exploded"))) + +# --------------------------------------------------------------------------- +# 4. END TO END: the boot sequence that used to strand the tombstone forever +# --------------------------------------------------------------------------- +proj2 = db.create_project("boot-racy") +pid2 = proj2["id"] +inflate(pid2) +jobs.request_delete(pid2) # tombstone + spawn the owning cleanup + +# ...and immediately run the REAL startup janitor, exactly as lifespan does one +# line later. Calling main._warm_vectorstore() rather than re-deriving its +# "known" set here is the point: this is what guards the fix at main.py, and it +# is what fails if someone reverts that union to a bare db.list_projects(). +janitor = threading.Thread(target=om_main._warm_vectorstore, name="janitor") +janitor.start() +janitor.join(timeout=120) + +deadline = time.time() + 180 +while time.time() < deadline: + if db.get_project(pid2) is None and not db.list_projects("deleting"): + break + time.sleep(0.25) + +check("tombstone did NOT survive the janitor race", + not [p for p in db.list_projects("deleting") if p["id"] == pid2], + f"{pid2} still tombstoned") +check("project row is fully gone (delete actually completed)", + db.get_project(pid2) is None) +check("data dir reclaimed", not config.project_dir(pid2).exists()) +check("no leftover collections for the deleted project", + not [n for n in collection_names() if pid2 in n], collection_names()) + +# --------------------------------------------------------------------------- +# 5. The cheap read path: counts must not CREATE storage, and must agree +# --------------------------------------------------------------------------- +proj3 = db.create_project("counts") +pid3 = proj3["id"] +inflate(proj3["id"], n=1000) +db.upsert_file_index(pid3, "src/A.java", "h1", chunk_ids=["a1", "a2"]) +db.upsert_file_index(pid3, "src/B.java", "h2", chunk_ids=["b1"]) + +check("count_file_index agrees with len(get_file_index)", + db.count_file_index(pid3) == len(db.get_file_index(pid3)) == 2, + db.count_file_index(pid3)) + +before = collection_names() +cases_name = vectorstore.cases_collection_name(pid3) +n_cases = vectorstore.count_collection(cases_name) +check("count_collection reports 0 for a collection that does not exist", + n_cases == 0, n_cases) +check("count_collection did NOT create the collection as a side effect", + collection_names() == before, collection_names() - before) +check("count_collection reports the real size of an existing collection", + vectorstore.count_collection(vectorstore.code_collection_name(pid3)) == 1000) + +svc = WorkspaceService() +d = svc.describe(pid3) +check("describe() still returns the documented shape", + d["code_chunks"] == 1000 and d["cases_count"] == 0 and d["files_indexed"] == 2, + {k: d[k] for k in ("code_chunks", "cases_count", "files_indexed")}) + +jobs.begin_shutdown() +print(f"\n{sum(results)}/{len(results)} checks passed", flush=True) +sys.exit(0 if all(results) else 1) From e16f3feb5a46adf650d732d1d34b364445bd376c Mon Sep 17 00:00:00 2001 From: HelloThisWorld Date: Sun, 19 Jul 2026 19:27:01 +0800 Subject: [PATCH 2/2] Register verify_delete_race in the acceptance suite; make it CI-safe and fast CI's full gate failed because scripts/run_acceptance.py enforces that every tests/verify_*.py appears in its SUITE manifest -- so a new test cannot go silently unrun -- and the new script was not registered. Registered as CORE. Two changes were needed before it belonged in a gate rather than in the LOCAL tier alongside verify_delete_responsive: * Removed the wall-clock assertions. The in-flight guard was proven by timing a second drain call ("returned in <1s while the first was still running"), which is a scheduling race on a shared runner. It now occupies the collection name directly and asserts the refusal with no threads and no clock; a separate, timing-free check covers that a real drain claims and releases the name. Likewise the "collection vanished mid-drain" case is now forced with a stub collection that reports rows once and then raises NotFound, instead of racing a second thread to delete it at the right moment. * Fixed a wait condition that made the script take 184s. The end-to-end section polled `not db.list_projects("deleting")`, but section 1 deliberately leaves its own tombstone in place, so that never became true and the loop burned its full 180s timeout on every run. It now waits on the project under test only. Runtime: 184s -> 4.5s. Fixture sizes trimmed to match (3000 -> 1200 chunks). Re-validated after the rewrite: reverting each of the three defences individually still fails its own check (24/26, 25/26, 25/26 respectively). Full core tier green locally: 24 passed, 0 failed, 0 skipped. --- scripts/run_acceptance.py | 2 + tests/verify_delete_race.py | 166 +++++++++++++++++++++--------------- 2 files changed, 100 insertions(+), 68 deletions(-) diff --git a/scripts/run_acceptance.py b/scripts/run_acceptance.py index 8d1bfd0..337fe6c 100644 --- a/scripts/run_acceptance.py +++ b/scripts/run_acceptance.py @@ -82,6 +82,8 @@ def path(self) -> Path: Script("verify_ask", CORE, "Ask gating when no model is ready"), Script("verify_ask2", CORE, "Ask history, cancellation and case saving"), Script("verify_async_delete", CORE, "asynchronous project delete"), + Script("verify_delete_race", CORE, + "delete completes: janitor/cleanup race, in-flight drain guard"), Script("verify_source_link", CORE, "GitHub link parsing + egress policy (no live network call)"), Script("verify_modelserver", CORE, diff --git a/tests/verify_delete_race.py b/tests/verify_delete_race.py index b475586..4e31d8a 100644 --- a/tests/verify_delete_race.py +++ b/tests/verify_delete_race.py @@ -45,7 +45,7 @@ from openmind import config, db, jobs, vectorstore # noqa: E402 from openmind.services.workspace_service import WorkspaceService # noqa: E402 -N_CHUNKS = 3000 +N_CHUNKS = 1200 DIM = 384 results = [] @@ -124,51 +124,65 @@ def collection_names(): # --------------------------------------------------------------------------- # 2. Two concurrent drains of ONE collection never both proceed # --------------------------------------------------------------------------- -# Forced, not hoped-for: hold a drain open with a tiny batch size, then make the -# second call while the first is provably still running. +# Split into a deterministic half and a timing-free half ON PURPOSE. Asserting +# "the second call returned fast while the first was still running" would make +# this a wall-clock race on a shared CI runner, and the acceptance runner's own +# rule is that a flaky gate is worse than an honest exclusion. So: +# 2a occupies the name directly and asserts the refusal — no threads, no clock; +# 2b runs a real drain and asserts only that it claims and releases the name. if vectorstore.backend_name() == "chroma": + occupied = "code_p_occupied000" + with vectorstore._dropping_lock: + vectorstore._dropping.add(occupied) + try: + refused = vectorstore.drop_collection(occupied) + finally: + with vectorstore._dropping_lock: + vectorstore._dropping.discard(occupied) + check("a drain of an already-in-flight collection is refused", + refused is False, refused) + check("...and the refusal did not leave the name claimed", + occupied not in vectorstore._dropping) + + # A real drain must claim the name for its duration and release it after. _orig_batch = vectorstore.DROP_BATCH - vectorstore.DROP_BATCH = 50 # ~60 batches -> seconds of drain - first = {} + vectorstore.DROP_BATCH = 50 # many small batches -> observable + first, seen_claimed = {}, threading.Event() def slow_drain(): first["result"] = vectorstore.drop_collection(code_name) + watcher_stop = threading.Event() + + def watcher(): + while not watcher_stop.is_set(): + if code_name in vectorstore._dropping: + seen_claimed.set() + return + time.sleep(0.005) + + w = threading.Thread(target=watcher, name="claim-watcher") + w.start() t = threading.Thread(target=slow_drain, name="slow-drain") t.start() - # wait until the drain has actually claimed the name - claimed = False - for _ in range(200): - if code_name in vectorstore._dropping: - claimed = True - break - time.sleep(0.01) - check("a drain in progress registers itself as in-flight", claimed) - - started = time.monotonic() - second = vectorstore.drop_collection(code_name) - elapsed = time.monotonic() - started - still_running = t.is_alive() - t.join(timeout=180) - - check("the second concurrent drain is refused", second is False, second) - check("...and refused IMMEDIATELY, without doing competing work", - elapsed < 1.0, f"{elapsed:.3f}s") - check("...while the first drain was demonstrably still running", - still_running) - check("the first drain completed the collection", first.get("result") is True, + t.join(timeout=300) + watcher_stop.set() + w.join(timeout=5) + vectorstore.DROP_BATCH = _orig_batch + + check("a drain in progress registers itself as in-flight", + seen_claimed.is_set()) + check("the drain completed the collection", first.get("result") is True, first) check("the collection is gone once the drain finished", code_name not in collection_names()) check("the in-flight guard released the name", code_name not in vectorstore._dropping) - vectorstore.DROP_BATCH = _orig_batch else: - for skipped in ("a drain in progress registers itself as in-flight", - "the second concurrent drain is refused", - "...and refused IMMEDIATELY, without doing competing work", - "...while the first drain was demonstrably still running", - "the first drain completed the collection", + for skipped in ("a drain of an already-in-flight collection is refused", + "...and the refusal did not leave the name claimed", + "a drain in progress registers itself as in-flight", + "the drain completed the collection", "the collection is gone once the drain finished", "the in-flight guard released the name"): check(skipped + " [numpy backend: n/a]", True) @@ -179,40 +193,52 @@ def slow_drain(): # This is the exact shape of the original bug: the losing racer must not report # failure, because its caller (_cleanup_deleted) reads False as "stop now" and # returns before removing the data dir and the project row. -if vectorstore.backend_name() == "chroma": - victim = db.create_project("vanishes") - vpid = victim["id"] - inflate(vpid, n=2000) - vname = vectorstore.code_collection_name(vpid) - - _orig_batch = vectorstore.DROP_BATCH - vectorstore.DROP_BATCH = 50 - outcome = {} - - def drain_victim(): - outcome["result"] = vectorstore.drop_collection(vname) - - dt = threading.Thread(target=drain_victim, name="victim-drain") - dt.start() - for _ in range(200): # let it get properly mid-drain - if vname in vectorstore._dropping: - break - time.sleep(0.01) - time.sleep(0.3) - try: # yank it out from underneath - vectorstore._chroma_client.delete_collection(vname) - except Exception: +# Forced rather than raced: a stub collection that reports rows once and then +# claims not to exist is EXACTLY what the losing thread saw, and it reproduces +# that on every run instead of only when the scheduler cooperates. +class _VanishingCollection: + def __init__(self): + self.batches = 0 + + def count(self): + return 500 + + def get(self, limit=None, include=None): + self.batches += 1 + if self.batches == 1: + return {"ids": [f"x{i}" for i in range(min(limit or 50, 50))]} + raise ValueError("Collection does not exist.") # deleted underneath us + + def delete(self, ids=None): pass - dt.join(timeout=180) - vectorstore.DROP_BATCH = _orig_batch - check("a drain whose collection vanished mid-way reports SUCCESS", - outcome.get("result") is True, - f"got {outcome.get('result')!r} — _cleanup_deleted would abandon the " - f"delete and strand the tombstone") -else: - check("a drain whose collection vanished mid-way reports SUCCESS " - "[numpy backend: n/a]", True) + +class _VanishingClient: + def __init__(self, col): + self.col = col + self.deleted = [] + + def get_collection(self, name): + return self.col + + def delete_collection(self, name): + self.deleted.append(name) + + +_saved_client, _saved_backend = vectorstore._chroma_client, vectorstore._backend +stub = _VanishingClient(_VanishingCollection()) +vectorstore._chroma_client, vectorstore._backend = stub, "chroma" +try: + vanished = vectorstore.drop_collection("code_p_vanishes00") +finally: + vectorstore._chroma_client, vectorstore._backend = _saved_client, _saved_backend + +check("a drain whose collection vanished mid-way reports SUCCESS", + vanished is True, + f"got {vanished!r} — _cleanup_deleted reads False as 'stop now' and would " + f"return before the rmtree and the row delete, stranding the tombstone") +check("...and the vanished drain did not leave the name claimed", + "code_p_vanishes00" not in vectorstore._dropping) check("dropping an already-gone collection reports success", vectorstore.drop_collection(code_name) is True) @@ -238,9 +264,13 @@ def drain_victim(): janitor.start() janitor.join(timeout=120) +# Wait on THIS project only. Section 1 deliberately leaves its own tombstone +# behind, so a "no projects are deleting" condition would never come true and +# this loop would burn its whole timeout on every run. deadline = time.time() + 180 while time.time() < deadline: - if db.get_project(pid2) is None and not db.list_projects("deleting"): + if db.get_project(pid2) is None and \ + not [p for p in db.list_projects("deleting") if p["id"] == pid2]: break time.sleep(0.25) @@ -258,7 +288,7 @@ def drain_victim(): # --------------------------------------------------------------------------- proj3 = db.create_project("counts") pid3 = proj3["id"] -inflate(proj3["id"], n=1000) +inflate(proj3["id"], n=400) db.upsert_file_index(pid3, "src/A.java", "h1", chunk_ids=["a1", "a2"]) db.upsert_file_index(pid3, "src/B.java", "h2", chunk_ids=["b1"]) @@ -274,12 +304,12 @@ def drain_victim(): check("count_collection did NOT create the collection as a side effect", collection_names() == before, collection_names() - before) check("count_collection reports the real size of an existing collection", - vectorstore.count_collection(vectorstore.code_collection_name(pid3)) == 1000) + vectorstore.count_collection(vectorstore.code_collection_name(pid3)) == 400) svc = WorkspaceService() d = svc.describe(pid3) check("describe() still returns the documented shape", - d["code_chunks"] == 1000 and d["cases_count"] == 0 and d["files_indexed"] == 2, + d["code_chunks"] == 400 and d["cases_count"] == 0 and d["files_indexed"] == 2, {k: d[k] for k in ("code_chunks", "cases_count", "files_indexed")}) jobs.begin_shutdown()