From 4bca1b30df4b05356e840f271fd22fbc78a28b47 Mon Sep 17 00:00:00 2001 From: kristofr Date: Tue, 26 May 2026 08:30:01 +0200 Subject: [PATCH 1/7] Add daemon lock handling, tests, and startup scan status --- AGENTS.md | 113 +++++++++ indexserver/backend.py | 20 +- indexserver/csv_log.py | 211 ++++++++++++++++ indexserver/daemon.py | 141 ++++++++++- indexserver/index_queue.py | 47 ++-- indexserver/indexer.py | 4 +- indexserver/verifier.py | 15 +- indexserver/watcher.py | 11 + lib/daemon_lock.mjs | 96 +++++++ query/config.py | 12 + scripts/diag_mtime.py | 80 ++++++ scripts/diag_relpath.py | 58 +++++ scripts/scan_csv.py | 445 +++++++++++++++++++++++++++++++++ tests/test_daemon_lock.mjs | 301 ++++++++++++++++++++++ tests/unit/test_daemon_lock.py | 351 ++++++++++++++++++++++++++ ts.mjs | 80 ++++-- 16 files changed, 1941 insertions(+), 44 deletions(-) create mode 100644 AGENTS.md create mode 100644 indexserver/csv_log.py create mode 100644 lib/daemon_lock.mjs create mode 100644 scripts/diag_mtime.py create mode 100644 scripts/diag_relpath.py create mode 100644 scripts/scan_csv.py create mode 100644 tests/test_daemon_lock.mjs create mode 100644 tests/unit/test_daemon_lock.py diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..ba18d17 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,113 @@ +# Agent notes + +Operational notes for Claude / future agents working in this repo. For +architecture, module map, and conventions see `CLAUDE.md`. + +## Debugging the indexer: real-time CSV logs + +When something goes wrong on the indexer side (the classic symptom: files +get re-indexed on every restart even though the index already contains +them), turn on the CSV debug log. + +### Enable + +Set `csv_debug` in `config.json`: + +```json +{ + "port": 8108, + "csv_debug": true, + "roots": { "default": { "path": "q:/spocore/src" } } +} +``` + +Then `node ts.mjs restart` (or `ts restart` if the cmd is on PATH). + +Accepted values: + +| value | effect | +|------------------------------------|-----------------------------------------------------------------| +| `false` / `""` / unset | off | +| `true` / `"1"` / `"on"` / `"true"` | logs to `%LOCALAPPDATA%/tscodesearch/csv/` | +| any other string | treated as an explicit directory path -- logs land there | + +Logging is opt-in and append-only; rows from successive restarts share a +file and are distinguished by the per-row `pid` column. + +### CSV files + +Written by `indexserver/csv_log.py`, one file per event type. Each row +starts with `ts` (ms-precision local time) and `pid`. + +| file | rows are written when | useful columns | +|----------------------|--------------------------------------------------------------------|------------------------------------------------------| +| `session.csv` | daemon start/stop | `action` = `start`/`stop` | +| `backend_export.csv` | verifier exports the index map at scan start (once per session) | `doc_id`, `mtime`, `relative_path` | +| `fs_walk.csv` | every file the verifier walks against the exported map | `decision` = `matched` / `missing` / `stale` | +| `orphan.csv` | index entries with no matching file on disk (about to be deleted) | `doc_id` | +| `enqueue.csv` | every `IndexQueue.enqueue` (including dedup hits) | `action`, `reason`, `is_new` | +| `parse.csv` | every tree-sitter parse the queue worker runs | `parse_ms`, `ok`, `error` | +| `commit.csv` | every Tantivy commit, success or failure | `duration_ms`, `success`, `error` | +| `watcher.csv` | every watchdog event | `src_path`, `action` | + +Decision values in `fs_walk.csv`: + +* `matched` -- file's mtime equals the indexed mtime; nothing to do +* `missing` -- file's `doc_id` is not in the index; enqueued as `new` +* `stale` -- file is in the index but its mtime has changed; enqueued as `modified` + +### Scanning the logs + +`scripts/scan_csv.py` ingests the CSV directory and prints per-session +breakdowns plus anomaly flags. + +```powershell +# default: summary table (one row per restart, anomaly notes on the right) +.client-venv/Scripts/python.exe -m scripts.scan_csv + +# list each session with timestamps and counts +.client-venv/Scripts/python.exe -m scripts.scan_csv --mode sessions + +# drill into one session +.client-venv/Scripts/python.exe -m scripts.scan_csv --session 1 --mode summary + +# top stale files with mtime-delta buckets (was the change recent? batched?) +.client-venv/Scripts/python.exe -m scripts.scan_csv --mode stales + +# list missing files (filename + on-disk mtime) +.client-venv/Scripts/python.exe -m scripts.scan_csv --mode missing + +# orphan doc ids with relative_path (joined from the same session's export) +.client-venv/Scripts/python.exe -m scripts.scan_csv --mode orphans + +# every commit / parse failure with the error string +.client-venv/Scripts/python.exe -m scripts.scan_csv --mode errors + +# diff the last two index exports: what disappeared, what appeared, what changed mtime +.client-venv/Scripts/python.exe -m scripts.scan_csv --mode index-diff + +# files that flipped between matched and stale/missing across sessions +.client-venv/Scripts/python.exe -m scripts.scan_csv --mode flapping +``` + +Override the log directory with `--csv-dir DIR` or +`TSCODESEARCH_CSV_DIR=DIR`. + +### Anomaly flags emitted by `summary` + +* `INDEX_EMPTY_ON_OPEN` -- the index reported 0 docs at scan start; if + the previous session ended with a populated index this means the + Tantivy state was lost between restarts (e.g. `meta.json` blown away, + segments unreachable). Look for prior `commit.csv` failures. +* `ALL_MISSING` -- every walked file is missing from the index. Same + signal as the empty-index case but observed via the fs walk. +* `COMMIT_FAIL=N` -- Tantivy commit failed N times. The error column in + `commit.csv` shows the cause (typically Windows `Access is denied`). +* `ORPHAN_HEAVY (X/Y)` -- more than half the index entries didn't match + any file on disk; check whether the configured root path changed. +* `PARSE_FAIL=N` -- file read / tree-sitter parse failures. + +### Disabling + +Set `"csv_debug": false` in `config.json` and restart. The files stick +around for offline analysis; delete the `csv/` directory when done. diff --git a/indexserver/backend.py b/indexserver/backend.py index 49b4af4..95674c5 100644 --- a/indexserver/backend.py +++ b/indexserver/backend.py @@ -433,20 +433,34 @@ def searcher(self) -> tantivy.Searcher: idx.reload() return idx.searcher() - def export_id_mtime(self) -> dict[str, int]: - """Return {doc_id: mtime} for every document. Used by the verifier.""" + def export_id_mtime(self, collection: str = "") -> dict[str, int]: + """Return {doc_id: mtime} for every document. Used by the verifier. + + When CSV debug logging is enabled, each row is also written to + ``backend_export.csv`` so a human can see exactly what the verifier + saw as the "already indexed" set at scan start. ``collection`` is + passed through to the CSV; pass the empty string when the caller + doesn't know it (the CSV tag will just be blank). + """ + from indexserver import csv_log + searcher = self.searcher() n = searcher.num_docs if n == 0: return {} result = searcher.search(tantivy.Query.all_query(), limit=n) out: dict[str, int] = {} + log_export = csv_log.enabled() for _, addr in result.hits: doc = searcher.doc(addr).to_dict() doc_id = (doc.get("id") or [""])[0] mtime = (doc.get("mtime") or [0])[0] if doc_id: - out[doc_id] = int(mtime) + mtime_int = int(mtime) + out[doc_id] = mtime_int + if log_export: + rel = (doc.get("relative_path") or [""])[0] + csv_log.backend_export(collection, doc_id, mtime_int, rel) return out diff --git a/indexserver/csv_log.py b/indexserver/csv_log.py new file mode 100644 index 0000000..0646783 --- /dev/null +++ b/indexserver/csv_log.py @@ -0,0 +1,211 @@ +""" +Real-time CSV debug logging for the indexer pipeline. + +Off by default. Enable via the ``csv_debug`` key in ``config.json`` -- +either a truthy value (``true``, ``"1"``, ``"on"``) or an explicit +directory path. Truthy values write CSVs under ``/csv/``; +a path string overrides that location. The daemon calls ``configure()`` +once at startup with the value from config; an empty / falsy value leaves +every log call as a no-op. + +One CSV file per event type lives in the log directory; rows are appended +across daemon restarts so post-mortem analysis can span multiple sessions. +Every row starts with a millisecond timestamp and the daemon PID so a +sequence can be reconstructed even when several restarts share the file. + +Events written today: + + session daemon start/stop boundaries + backend_export one row per (doc_id, mtime, relative_path) the verifier + sees when it exports the index map at the start of a scan + fs_walk one row per file the verifier walks; ``decision`` is + one of ``matched``, ``missing``, ``stale`` + orphan one row per index doc that didn't match any file on disk + (about to be deleted from the index) + enqueue one row per index queue enqueue (or dedup hit) + parse one row per parse the index queue worker performed + commit one row per Tantivy commit (success or failure) + watcher one row per filesystem event picked up by the watcher + +The module does its own file handling: line-buffered append-mode files so +each row hits disk immediately (real-time). All writes guard against +exceptions so a logging bug can never crash the daemon. +""" + +from __future__ import annotations + +import os +import sys +import threading +import time +from pathlib import Path +from typing import Any, IO + + +# -- module state --------------------------------------------------------------- + +_enabled: bool = False +_dir: Path | None = None +_files: dict[str, IO[str]] = {} +_lock = threading.Lock() +_pid: int = os.getpid() + + +# -- public API ----------------------------------------------------------------- + +def configure(default_dir: Path, setting: str = "") -> None: + """Turn logging on if ``setting`` is non-empty / truthy. + + ``default_dir`` is used when ``setting`` is a plain truthy value + (``1``, ``true``, ...). Any other non-empty string is treated as an + explicit directory path that overrides ``default_dir``. + """ + global _enabled, _dir + raw = (setting or "").strip() + if not raw: + return + if raw.lower() in ("0", "false", "no", "off"): + return + if raw.lower() in ("1", "true", "yes", "on"): + _dir = default_dir + else: + _dir = Path(raw) + try: + _dir.mkdir(parents=True, exist_ok=True) + except OSError as e: + print(f"[csv_log] cannot create {_dir}: {e}", file=sys.stderr, flush=True) + return + _enabled = True + print(f"[csv_log] CSV debug logging enabled -> {_dir}", flush=True) + # Mark a session boundary so multi-restart logs can be split apart. + session("start") + + +def enabled() -> bool: + return _enabled + + +def shutdown(reason: str = "stop") -> None: + """Flush+close all open csv files. Safe to call multiple times.""" + if not _enabled: + return + session(reason) + with _lock: + for f in _files.values(): + try: + f.flush() + f.close() + except Exception: + pass + _files.clear() + + +# -- per-event helpers ---------------------------------------------------------- + +def session(action: str, detail: str = "") -> None: + _write("session", + ("ts", "pid", "action", "detail"), + (action, detail)) + + +def backend_export(collection: str, doc_id: str, mtime: int, relative_path: str) -> None: + _write("backend_export", + ("ts", "pid", "collection", "doc_id", "mtime", "relative_path"), + (collection, doc_id, mtime, relative_path)) + + +def fs_walk(collection: str, rel: str, mtime: int, doc_id: str, + idx_mtime: int | None, decision: str) -> None: + _write("fs_walk", + ("ts", "pid", "collection", "rel", "mtime", "doc_id", + "idx_mtime", "decision"), + (collection, rel, mtime, doc_id, + "" if idx_mtime is None else idx_mtime, decision)) + + +def orphan(collection: str, doc_id: str) -> None: + _write("orphan", + ("ts", "pid", "collection", "doc_id"), + (collection, doc_id)) + + +def enqueue(collection: str, rel: str, action: str, mtime: int | None, + reason: str, is_new: bool) -> None: + _write("enqueue", + ("ts", "pid", "collection", "rel", "action", "mtime", + "reason", "is_new"), + (collection, rel, action, + "" if mtime is None else mtime, reason, "1" if is_new else "0")) + + +def parse(collection: str, rel: str, size: int, parse_ms: float, ok: bool, + error: str = "") -> None: + _write("parse", + ("ts", "pid", "collection", "rel", "size", "parse_ms", "ok", "error"), + (collection, rel, size, f"{parse_ms:.1f}", "1" if ok else "0", error)) + + +def commit(collection: str, n_buffered: int, duration_ms: float, + success: bool, error: str = "") -> None: + _write("commit", + ("ts", "pid", "collection", "n_buffered", "duration_ms", + "success", "error"), + (collection, n_buffered, f"{duration_ms:.1f}", + "1" if success else "0", error)) + + +def watcher(collection: str, src_path: str, action: str) -> None: + _write("watcher", + ("ts", "pid", "collection", "src_path", "action"), + (collection, src_path, action)) + + +# -- internals ------------------------------------------------------------------ + +def _ts() -> str: + now = time.time() + return time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(now)) \ + + f".{int((now - int(now)) * 1000):03d}" + + +def _escape(value: Any) -> str: + s = "" if value is None else str(value) + # ASCII only -- replace any stray non-ASCII (paths can contain anything). + s = s.encode("ascii", "replace").decode("ascii") + if any(c in s for c in (",", '"', "\n", "\r")): + s = '"' + s.replace('"', '""') + '"' + return s + + +def _get_file(event: str, header: tuple[str, ...]) -> IO[str] | None: + if _dir is None: + return None + f = _files.get(event) + if f is not None: + return f + with _lock: + f = _files.get(event) + if f is not None: + return f + path = _dir / f"{event}.csv" + new_file = not path.exists() or path.stat().st_size == 0 + f = open(path, "a", encoding="ascii", errors="replace", + buffering=1, newline="") + if new_file: + f.write(",".join(header) + "\n") + _files[event] = f + return f + + +def _write(event: str, header: tuple[str, ...], row: tuple) -> None: + if not _enabled: + return + try: + f = _get_file(event, header) + if f is None: + return + cells = [_ts(), str(_pid)] + [_escape(v) for v in row] + f.write(",".join(cells) + "\n") + except Exception as e: + # Logging must never crash the daemon. Print and move on. + print(f"[csv_log] write error ({event}): {e}", file=sys.stderr, flush=True) diff --git a/indexserver/daemon.py b/indexserver/daemon.py index 4ccb2e9..88c7813 100644 --- a/indexserver/daemon.py +++ b/indexserver/daemon.py @@ -49,6 +49,7 @@ from indexserver.search_modes import build_filter_by, resolve_query_params from indexserver.verifier import run_verify from indexserver.watcher import run_watcher +from indexserver import csv_log from indexserver import search as _search_mod _cfg = _load_config() @@ -61,8 +62,14 @@ if sys.platform == "win32" else _HOME / ".local" / "tscodesearch" ) -_RUN_DIR = Path(os.environ.get("TSCODESEARCH_DATA", _DEFAULT_RUN_DIR)) -_DAEMON_PID = _RUN_DIR / "daemon.pid" +_RUN_DIR = Path(os.environ.get("TSCODESEARCH_DATA", _DEFAULT_RUN_DIR)) +_DAEMON_PID = _RUN_DIR / "daemon.pid" +_DAEMON_LOCK = _RUN_DIR / "daemon.lock" + +# File handle for the OS-level exclusive lock. Intentionally never closed -- +# the OS releases the lock when the process truly exits (clean or crash), which +# prevents a second daemon from starting before the first is fully gone. +_lock_fh: object = None _QUERY_CODEBASE_MAX_LIMIT = 250 @@ -78,6 +85,15 @@ _server: HTTPServer | None = None _shutdown_event = threading.Event() +_verify_status_lock = threading.Lock() +_verify_status: dict = { + "state": "idle", + "active_root": "", + "started_at": "", + "last_update": "", + "roots": {}, +} + # -- Tree-sitter query helper --------------------------------------------------- @@ -201,6 +217,71 @@ def _collections_status() -> dict: return collections +def _verify_status_begin() -> None: + now = time.strftime("%Y-%m-%dT%H:%M:%S") + with _verify_status_lock: + _verify_status.update({ + "state": "running", + "active_root": "", + "started_at": now, + "last_update": now, + "roots": {}, + }) + + +def _verify_status_progress(root_name: str, progress: dict) -> None: + now = time.strftime("%Y-%m-%dT%H:%M:%S") + row = { + "status": progress.get("status", "running"), + "phase": progress.get("phase", ""), + "fs_files": progress.get("fs_files", 0), + "missing": progress.get("missing", 0), + "stale": progress.get("stale", 0), + "orphaned": progress.get("orphaned", 0), + "total_to_update": progress.get("total_to_update", 0), + "updated": progress.get("updated", 0), + "errors": progress.get("errors", 0), + "last_update": progress.get("last_update", now), + } + with _verify_status_lock: + _verify_status["state"] = "running" + _verify_status["active_root"] = root_name + _verify_status["last_update"] = now + _verify_status["roots"][root_name] = row + + +def _verify_status_root_error(root_name: str, error: str) -> None: + now = time.strftime("%Y-%m-%dT%H:%M:%S") + with _verify_status_lock: + _verify_status["roots"][root_name] = { + "status": "error", + "phase": "initial scan failed", + "error": str(error), + "last_update": now, + } + _verify_status["last_update"] = now + + +def _verify_status_finish(cancelled: bool = False) -> None: + now = time.strftime("%Y-%m-%dT%H:%M:%S") + with _verify_status_lock: + _verify_status["state"] = "cancelled" if cancelled else "complete" + _verify_status["active_root"] = "" + _verify_status["last_update"] = now + + +def _verify_status_snapshot() -> dict: + with _verify_status_lock: + roots = {name: dict(info) for name, info in _verify_status.get("roots", {}).items()} + return { + "state": _verify_status.get("state", "idle"), + "active_root": _verify_status.get("active_root", ""), + "started_at": _verify_status.get("started_at", ""), + "last_update": _verify_status.get("last_update", ""), + "roots": roots, + } + + # -- Watcher -------------------------------------------------------------------- def _start_watcher() -> None: @@ -285,11 +366,15 @@ def _initialize_async(stop_event: threading.Event) -> None: _init_backends() _index_queue.start(lambda c: _backends.get(c)) _start_watcher() + _verify_status_begin() for root in ALL_ROOTS.values(): if _shutdown_event.is_set(): break try: + def _on_progress(p: dict, root_name: str = root.name) -> None: + _verify_status_progress(root_name, p) + run_verify( _cfg, src_root=root.path, @@ -299,12 +384,15 @@ def _initialize_async(stop_event: threading.Event) -> None: stop_event=_shutdown_event, extensions=root.extensions, on_complete=lambda: None, - on_progress=lambda _: None, + on_progress=_on_progress, backend=_backends.get(root.collection), ) except Exception as e: + _verify_status_root_error(root.name, str(e)) print(f"[tsquery_server] Initial scan error ({root.collection}): {e}", flush=True) + _verify_status_finish(cancelled=_shutdown_event.is_set()) + print("[tsquery_server] Initialization complete.", flush=True) @@ -356,6 +444,7 @@ def _handle(self) -> None: "watcher": _watcher_status(), "queue": _index_queue.stats(), "collections": _collections_status(), + "scan": _verify_status_snapshot(), # Compat: clients still inspect these keys; backend is in-process # so it's always "ok" once the daemon is up. "typesense_ok": True, @@ -538,12 +627,55 @@ class _ThreadedHTTPServer(ThreadingMixIn, HTTPServer): # -- Public API ----------------------------------------------------------------- +def _try_acquire_lock() -> bool: + """Acquire an exclusive OS-level file lock on _DAEMON_LOCK. + + The file handle is stored in _lock_fh and intentionally never closed. + The OS releases the lock automatically when the process exits (clean, + killed, or crashed), so a second daemon cannot start until the first + is truly dead -- not merely mid-shutdown. + + Returns True if the lock was acquired, False if another process holds it. + """ + global _lock_fh + _RUN_DIR.mkdir(parents=True, exist_ok=True) + try: + fh = open(_DAEMON_LOCK, "w+") + if sys.platform == "win32": + import msvcrt + # LK_NBLCK: non-blocking exclusive lock on 1 byte at position 0. + msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + fh.write(str(os.getpid())) + fh.flush() + _lock_fh = fh # keep handle alive for the process lifetime + return True + except (OSError, IOError): + try: + fh.close() + except Exception: + pass + return False + + def start_daemon() -> bool: - """Try to bind PORT and start all daemon threads.""" + """Acquire the process lock and start all daemon threads.""" global _server _RUN_DIR.mkdir(parents=True, exist_ok=True) + # Acquire the exclusive process lock before doing anything else. + # This is checked before the port bind, so a second daemon that loses + # the lock race exits cleanly without leaving a ghost entry in session.csv. + if not _try_acquire_lock(): + return False + + # Only configure CSV logging (which writes the session-start row) after we + # own the lock -- so every session.csv entry corresponds to a real startup. + csv_log.configure(_RUN_DIR / "csv", _cfg.csv_debug) + try: _server = _ThreadedHTTPServer(("127.0.0.1", _cfg.port), _Handler) except OSError: @@ -660,6 +792,7 @@ def _elapsed() -> str: if _DAEMON_PID.exists(): _DAEMON_PID.unlink() + csv_log.shutdown("stop") print(f"[tsquery_server] Stopped. total={_elapsed()}", flush=True) diff --git a/indexserver/index_queue.py b/indexserver/index_queue.py index 7008cb6..bb53a18 100644 --- a/indexserver/index_queue.py +++ b/indexserver/index_queue.py @@ -25,6 +25,7 @@ from indexserver.indexer import build_document, file_id as _file_id from indexserver.backend import Backend +from indexserver import csv_log # Sentinel mtime value stored in queue items for delete actions. MTIME_DELETE = None @@ -154,6 +155,8 @@ def enqueue( self._cond.notify() else: self._n_deduped += 1 + if csv_log.enabled(): + csv_log.enqueue(collection, rel, action, mtime, reason, is_new) return is_new def enqueue_bulk(self, file_pairs, collection: str, stop_event: threading.Event | None = None) -> tuple[int, int]: @@ -286,21 +289,25 @@ def _parse_one(item): t0 = time.perf_counter() with self._busy_lock: self._busy[tid] = (rel, size, t0) + err_msg = "" + doc = None try: doc = build_document(full_path, rel) - t = time.perf_counter() - t0 - if t > 1.0: - print( - f"[index-queue] SLOW parse: {rel} took {t:.2f}s ({size:,} bytes)", - flush=True, - ) - if doc: - return ("upsert", collection, doc) - except OSError: - pass - finally: - with self._busy_lock: - self._busy.pop(tid, None) + except OSError as e: + err_msg = f"OSError: {e}" + t = time.perf_counter() - t0 + if csv_log.enabled(): + csv_log.parse(collection, rel, size, t * 1000.0, + doc is not None, err_msg) + if t > 1.0: + print( + f"[index-queue] SLOW parse: {rel} took {t:.2f}s ({size:,} bytes)", + flush=True, + ) + with self._busy_lock: + self._busy.pop(tid, None) + if doc: + return ("upsert", collection, doc) return None t_parse_start = time.perf_counter() @@ -371,13 +378,21 @@ def _commit_all(self, dirty: set[str]) -> None: if backend is None or not backend.has_pending: dirty.discard(coll) continue + buffered = getattr(backend, "buffered_count", 0) t0 = time.perf_counter() + ok = True + err_msg = "" try: backend.commit() - t_commit = time.perf_counter() - t0 - print(f"[index-queue] committed {coll} in {t_commit:.2f}s", flush=True) except Exception as e: + ok = False + err_msg = f"{type(e).__name__}: {e}" with self._cond: self._n_errors += 1 - print(f"[index-queue] commit error ({coll}): {type(e).__name__}: {e}", flush=True) + print(f"[index-queue] commit error ({coll}): {err_msg}", flush=True) + t_commit = time.perf_counter() - t0 + if ok: + print(f"[index-queue] committed {coll} in {t_commit:.2f}s", flush=True) + if csv_log.enabled(): + csv_log.commit(coll, buffered, t_commit * 1000.0, ok, err_msg) dirty.discard(coll) diff --git a/indexserver/indexer.py b/indexserver/indexer.py index 30f29b6..59519e4 100644 --- a/indexserver/indexer.py +++ b/indexserver/indexer.py @@ -518,9 +518,9 @@ def index_file_list( return total, errors -def export_index_map(backend: Backend) -> dict[str, int]: +def export_index_map(backend: Backend, collection: str = "") -> dict[str, int]: """Return {doc_id: mtime} for every document in *backend*.""" - return backend.export_id_mtime() + return backend.export_id_mtime(collection=collection) def walk_and_enqueue( diff --git a/indexserver/verifier.py b/indexserver/verifier.py index 18b6159..395147c 100644 --- a/indexserver/verifier.py +++ b/indexserver/verifier.py @@ -26,6 +26,7 @@ sys.path.insert(0, _base) from query.config import normalize_path +from indexserver import csv_log from indexserver.indexer import ( walk_source_files, file_id, ensure_backend, index_file_list, @@ -58,7 +59,7 @@ def check_ready(cfg, src_root: str | None = None, try: backend = ensure_backend(cfg, coll, write=False) - index_map = export_index_map(backend) + index_map = export_index_map(backend, collection=coll) except Exception as e: return { "ready": False, "poll_ok": False, "index_ok": False, @@ -162,7 +163,7 @@ def _verify_with_backend(cfg, src_root, coll_name, queue, on_progress, progress["phase"] = "collecting: exporting index" if on_progress: on_progress(progress) - index_map = export_index_map(backend) + index_map = export_index_map(backend, collection=coll_name) progress["index_docs"] = len(index_map) print(f"[verifier] {len(index_map):,} documents in index", flush=True) @@ -179,6 +180,7 @@ def _verify_with_backend(cfg, src_root, coll_name, queue, on_progress, n_fs = 0 last_scan_print = time.time() + log_walk = csv_log.enabled() for sf in walk_source_files(src_root, cfg, extensions=extensions): if stop_event and stop_event.is_set(): break @@ -190,13 +192,19 @@ def _verify_with_backend(cfg, src_root, coll_name, queue, on_progress, progress["missing"] += 1 needs_update = True reason = "new" + decision = "missing" elif sf.mtime != idx_mtime: progress["stale"] += 1 needs_update = True reason = "modified" + decision = "stale" else: needs_update = False reason = "" + decision = "matched" + + if log_walk: + csv_log.fs_walk(coll_name, sf.rel, sf.mtime, doc_id, idx_mtime, decision) if needs_update: if queue is not None: @@ -224,6 +232,9 @@ def _verify_with_backend(cfg, src_root, coll_name, queue, on_progress, orphaned_ids = list(remaining) progress["fs_files"] = n_fs progress["orphaned"] = len(orphaned_ids) + if csv_log.enabled(): + for _oid in orphaned_ids: + csv_log.orphan(coll_name, _oid) if queue is None: progress["total_to_update"] = len(to_update) diff --git a/indexserver/watcher.py b/indexserver/watcher.py index 4deb465..d060f0e 100644 --- a/indexserver/watcher.py +++ b/indexserver/watcher.py @@ -26,6 +26,7 @@ from watchdog.events import FileSystemEventHandler from query.config import normalize_path +from indexserver import csv_log DEBOUNCE_SECONDS = 2.0 @@ -62,6 +63,8 @@ def on_created(self, event): with self._lock: self._pending[event.src_path] = "created" self._schedule_flush() + if csv_log.enabled(): + csv_log.watcher(self._collection, event.src_path, "created") def on_modified(self, event): if not event.is_directory and self._is_indexed(event.src_path): @@ -72,21 +75,29 @@ def on_modified(self, event): if self._pending.get(event.src_path) != "created": self._pending[event.src_path] = "modified" self._schedule_flush() + if csv_log.enabled(): + csv_log.watcher(self._collection, event.src_path, "modified") def on_deleted(self, event): if not event.is_directory and self._is_indexed(event.src_path): with self._lock: self._pending[event.src_path] = "deleted" self._schedule_flush() + if csv_log.enabled(): + csv_log.watcher(self._collection, event.src_path, "deleted") def on_moved(self, event): if not event.is_directory: if self._is_indexed(event.src_path): with self._lock: self._pending[event.src_path] = "deleted" + if csv_log.enabled(): + csv_log.watcher(self._collection, event.src_path, "moved_from") if self._is_indexed(event.dest_path) and not self._is_excluded(event.dest_path): with self._lock: self._pending[event.dest_path] = "created" + if csv_log.enabled(): + csv_log.watcher(self._collection, event.dest_path, "moved_to") self._schedule_flush() def _flush(self): diff --git a/lib/daemon_lock.mjs b/lib/daemon_lock.mjs new file mode 100644 index 0000000..fe44e22 --- /dev/null +++ b/lib/daemon_lock.mjs @@ -0,0 +1,96 @@ +/** + * daemon_lock.mjs -- helpers for probing the daemon OS-level file lock. + * + * The Python daemon holds an exclusive OS lock on daemon.lock for its entire + * lifetime (acquired in _try_acquire_lock(), never explicitly released). + * The OS releases the lock the moment the process truly dies -- clean, killed, + * or crashed. + * + * isLockHeld() lets Node callers ask "is the daemon process still alive?" + * without trusting the port (which closes mid-shutdown) or the PID file + * (which is written at startup, not on exit). + * + * Locking mechanism mirrors daemon.py exactly: + * Windows -- msvcrt.locking() byte-range lock on byte 0 + * Unix -- fcntl.flock() exclusive advisory lock + * + * Both are checked by spawning a tiny Python snippet so that the same OS + * primitive is used on both sides of the check. + */ + +import fs from 'node:fs'; +import { spawnSync } from 'node:child_process'; + +// -- Lock probe scripts (ASCII only, no Unicode) -------------------------------- + +// Each script: +// * opens the lock file for writing (r+b / r+) +// * attempts a non-blocking exclusive lock +// * exits 0 if the lock was free (acquired + released) +// * exits 1 if the lock is held by another process (or any error) +// +// Using try/except with bare `except:` ensures any unexpected error (file +// missing, permission denied, bad path) also returns 1 = "treat as held" +// which is the safe direction: we will wait or re-check rather than blindly +// proceeding. + +const _PROBE_WIN = [ + 'import msvcrt,sys', + 'fh=open(sys.argv[1],"r+b");fh.seek(0);msvcrt.locking(fh.fileno(),msvcrt.LK_NBLCK,1);fh.close()', +].join('\n'); + +const _PROBE_UNIX = [ + 'import fcntl,sys', + 'fh=open(sys.argv[1],"r+");fcntl.flock(fh,fcntl.LOCK_EX|fcntl.LOCK_NB);fh.close()', +].join('\n'); + +const _PROBE_SCRIPT = process.platform === 'win32' ? _PROBE_WIN : _PROBE_UNIX; + +// -- Public API ---------------------------------------------------------------- + +/** + * Synchronously check whether daemon.lock is held by a live process. + * + * @param {string} lockPath Absolute path to daemon.lock. + * @param {string} pythonExe Path to the Python interpreter to use for the probe. + * @returns {boolean} true = lock is held (process alive) + * false = lock is free (process dead or file absent) + */ +export function isLockHeld(lockPath, pythonExe) { + if (!fs.existsSync(lockPath)) return false; + + const result = spawnSync( + pythonExe, + ['-c', _PROBE_SCRIPT, lockPath], + { timeout: 3000, stdio: 'pipe' }, + ); + + // status 0 = probe acquired the lock => it was free => no daemon alive + // status !== 0 = probe failed to acquire => lock is held => daemon alive + // (also covers timeout / spawn error -- treat as held = safe) + if (result.error) return true; // spawn failed -- conservative: treat as held + return result.status !== 0; +} + +/** + * Poll until daemon.lock is released or the timeout elapses. + * + * @param {string} lockPath Absolute path to daemon.lock. + * @param {string} pythonExe Python interpreter for the probe. + * @param {object} [opts] + * @param {number} [opts.timeoutMs=10000] Max wait in milliseconds. + * @param {number} [opts.intervalMs=300] Poll interval in milliseconds. + * @returns {Promise} true = released within timeout, false = timed out. + */ +export async function waitForLockReleased(lockPath, pythonExe, { + timeoutMs = 10_000, + intervalMs = 300, +} = {}) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (!isLockHeld(lockPath, pythonExe)) return true; + await new Promise(r => setTimeout(r, intervalMs)); + } + // One final check: the last poll might have been just before release. + return !isLockHeld(lockPath, pythonExe); +} diff --git a/query/config.py b/query/config.py index f465090..744f58b 100644 --- a/query/config.py +++ b/query/config.py @@ -94,6 +94,11 @@ class Config: exclude_dirs: frozenset = field(default_factory=lambda: EXCLUDE_DIRS) max_file_bytes: int = 3 * 1024 * 1024 max_content_chars: int = 30000 + # Real-time CSV debug logging. Empty / false-y disables it. A truthy + # value (``true``, ``1``, ``on``) writes CSVs under the daemon run dir; + # any other string is treated as the directory path. See + # ``indexserver.csv_log`` for the file layout. + csv_debug: str = "" @property def src_root(self) -> str: @@ -166,8 +171,15 @@ def load_config(config_file: str | None = None) -> Config: if "port" not in raw: raise RuntimeError(f"'port' is required in {path}") + csv_debug_raw = raw.get("csv_debug", "") + if isinstance(csv_debug_raw, bool): + csv_debug = "1" if csv_debug_raw else "" + else: + csv_debug = str(csv_debug_raw or "") + return Config( port=int(raw["port"]), api_key=raw.get("api_key", "codesearch-local"), roots=_parse_roots(raw.get("roots", {})), + csv_debug=csv_debug, ) diff --git a/scripts/diag_mtime.py b/scripts/diag_mtime.py new file mode 100644 index 0000000..420bdc8 --- /dev/null +++ b/scripts/diag_mtime.py @@ -0,0 +1,80 @@ +""" +Diagnostic: compare stored mtime in the index against current disk mtime +for every file under the configured root. Categorise mismatches so we can +see whether the "stale on restart" symptom is genuine fs changes or a bug +in the indexer / verifier path. +""" + +import os +import sys +from collections import Counter + +_base = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _base not in sys.path: + sys.path.insert(0, _base) + +from query.config import load_config, normalize_path +from indexserver.backend import Backend +from indexserver.indexer import file_id, walk_source_files, ensure_backend + + +def main() -> None: + cfg = load_config() + root = next(iter(cfg.roots.values())) + print(f"root: {root.name} path: {root.path}") + print(f"collection: {root.collection}") + print(f"index_dir: {root.index_dir}") + print() + + backend = ensure_backend(cfg, root.collection, write=False) + index_map = backend.export_id_mtime() + print(f"index entries: {len(index_map):,}") + + matched = 0 + missing = 0 + stale = 0 + stale_diffs: Counter[int] = Counter() + stale_examples: list = [] + missing_examples: list = [] + + n_fs = 0 + for sf in walk_source_files(root.path, cfg, extensions=root.extensions): + n_fs += 1 + doc_id = file_id(sf.rel) + idx_mtime = index_map.get(doc_id) + if idx_mtime is None: + missing += 1 + if len(missing_examples) < 5: + missing_examples.append((sf.rel, sf.mtime)) + elif sf.mtime != idx_mtime: + stale += 1 + diff = sf.mtime - idx_mtime + stale_diffs[diff] += 1 + if len(stale_examples) < 10: + stale_examples.append((sf.rel, sf.mtime, idx_mtime, diff)) + else: + matched += 1 + if n_fs % 10000 == 0: + print(f" scanned {n_fs:,} ...", flush=True) + + print() + print(f"fs files: {n_fs:,}") + print(f"matched: {matched:,}") + print(f"missing: {missing:,}") + print(f"stale: {stale:,}") + print() + print("Top 20 mtime diff buckets (disk_mtime - idx_mtime):") + for diff, count in stale_diffs.most_common(20): + print(f" diff={diff:+d}s count={count:,}") + print() + print("Sample stale files (rel, disk_mtime, idx_mtime, diff):") + for rel, dm, im, d in stale_examples: + print(f" {rel} disk={dm} idx={im} diff={d}") + print() + print("Sample missing files (rel, disk_mtime):") + for rel, dm in missing_examples: + print(f" {rel} disk={dm}") + + +if __name__ == "__main__": + main() diff --git a/scripts/diag_relpath.py b/scripts/diag_relpath.py new file mode 100644 index 0000000..d775f7e --- /dev/null +++ b/scripts/diag_relpath.py @@ -0,0 +1,58 @@ +""" +Diagnostic: dump a sample of relative_path values stored in the index and +compare to what walk_source_files currently produces. We want to know if +the rel-path representation changed between runs. +""" +import os +import sys + +_base = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _base not in sys.path: + sys.path.insert(0, _base) + +import tantivy +from query.config import load_config, normalize_path +from indexserver.indexer import file_id, walk_source_files, ensure_backend + + +def main() -> None: + cfg = load_config() + root = next(iter(cfg.roots.values())) + print(f"root: {root.path}") + print(f"collection: {root.collection}") + backend = ensure_backend(cfg, root.collection, write=False) + + idx = backend._require_index() + idx.reload() + searcher = idx.searcher() + n = searcher.num_docs + print(f"index docs: {n:,}") + + print() + print("--- 20 sample stored docs ---") + result = searcher.search(tantivy.Query.all_query(), limit=20) + for _, addr in result.hits: + doc = searcher.doc(addr).to_dict() + rel = (doc.get("relative_path") or [""])[0] + did = (doc.get("id") or [""])[0] + mt = (doc.get("mtime") or [0])[0] + # recompute file_id from stored rel to see if matches stored id + recomputed = file_id(rel) + match = "OK" if recomputed == did else "MISMATCH" + # check whether file currently exists on disk + local_path = root.path.rstrip("/") + "/" + rel + exists = os.path.exists(local_path) + print(f" id={did[:8]} rel_id={recomputed[:8]} {match} exists={exists} mtime={mt} rel={rel}") + + print() + print("--- 10 files from fs walk ---") + walker = walk_source_files(root.path, cfg, extensions=root.extensions) + for i, sf in enumerate(walker): + if i >= 10: + break + did = file_id(sf.rel) + print(f" rel={sf.rel} id={did[:8]} mtime={sf.mtime}") + + +if __name__ == "__main__": + main() diff --git a/scripts/scan_csv.py b/scripts/scan_csv.py new file mode 100644 index 0000000..a23bf54 --- /dev/null +++ b/scripts/scan_csv.py @@ -0,0 +1,445 @@ +""" +Scan the CSV debug logs produced by ``indexserver.csv_log`` and surface +what the daemon decided. Built for the "stop / restart doesn't dedup" +class of bug: per-session breakdowns of how many files matched, how many +were re-indexed, what commits failed, and which files keep bouncing +between matched and stale across restarts. + +Usage: + python -m scripts.scan_csv # default summary + python -m scripts.scan_csv --mode sessions # list sessions only + python -m scripts.scan_csv --mode stales # per-stale-file detail + python -m scripts.scan_csv --mode missing # per-missing-file detail + python -m scripts.scan_csv --mode orphans # orphan id list + python -m scripts.scan_csv --mode errors # commit/parse errors + python -m scripts.scan_csv --mode index-diff # what changed between + # the last two index exports + python -m scripts.scan_csv --mode flapping # files that flipped state + # across sessions + python -m scripts.scan_csv --session N # filter to one session + # (1 = oldest) + +Defaults to ``%LOCALAPPDATA%/tscodesearch/csv`` on Windows. Override with +``--csv-dir`` or env ``TSCODESEARCH_CSV_DIR``. +""" +from __future__ import annotations + +import argparse +import csv +import os +import sys +from collections import Counter, defaultdict +from dataclasses import dataclass, field +from pathlib import Path +from typing import Iterable + + +# -- paths --------------------------------------------------------------------- + +def _default_csv_dir() -> Path: + explicit = os.environ.get("TSCODESEARCH_CSV_DIR", "").strip() + if explicit: + return Path(explicit) + if sys.platform == "win32": + base = os.environ.get("LOCALAPPDATA") or str(Path.home() / "AppData" / "Local") + else: + base = str(Path.home() / ".local") + return Path(base) / "tscodesearch" / "csv" + + +# -- session model ------------------------------------------------------------- + +@dataclass +class Session: + pid: str + start_ts: str = "" + stop_ts: str = "" + + # backend_export: count + mtime histogram + (id -> mtime, id -> rel) + export_count: int = 0 + export_id_to_mtime: dict[str, int] = field(default_factory=dict) + export_id_to_rel: dict[str, str] = field(default_factory=dict) + + # fs_walk: rows per decision; per-stale and per-missing details for drill-down + walk_decision: Counter = field(default_factory=Counter) + stale_rows: list[tuple[str, int, int]] = field(default_factory=list) # (rel, disk_mtime, idx_mtime) + missing_rows: list[tuple[str, int]] = field(default_factory=list) # (rel, disk_mtime) + matched_rels: set[str] = field(default_factory=set) + walk_total: int = 0 + + # orphan id list (deleted from index) + orphans: list[str] = field(default_factory=list) + + # commit / parse / enqueue counters + commits_ok: int = 0 + commits_fail: int = 0 + commit_errors: list[tuple[str, str, str]] = field(default_factory=list) # (ts, coll, err) + parse_total: int = 0 + parse_fail: int = 0 + parse_errors: list[tuple[str, str, str]] = field(default_factory=list) + enqueued: int = 0 + deduped: int = 0 + enqueue_by_reason: Counter = field(default_factory=Counter) + + # watcher + watcher_events: Counter = field(default_factory=Counter) + + +# -- loader -------------------------------------------------------------------- + +def _iter_csv(path: Path) -> Iterable[dict]: + if not path.exists(): + return + with open(path, newline="", encoding="ascii", errors="replace") as f: + reader = csv.DictReader(f) + for row in reader: + yield row + + +def load(csv_dir: Path) -> list[Session]: + """Build per-session aggregates from every CSV in *csv_dir*.""" + # session.csv defines the order. Without it, we still gather by PID. + sessions: dict[str, Session] = {} + order: list[str] = [] # PIDs in session-start order + + def _get(pid: str) -> Session: + s = sessions.get(pid) + if s is None: + s = Session(pid=pid) + sessions[pid] = s + return s + + for row in _iter_csv(csv_dir / "session.csv"): + pid = row["pid"] + s = _get(pid) + action = row["action"] + if action == "start": + s.start_ts = row["ts"] + if pid not in order: + order.append(pid) + elif action in ("stop", "test-done"): + s.stop_ts = row["ts"] + + for row in _iter_csv(csv_dir / "backend_export.csv"): + s = _get(row["pid"]) + if s.pid not in order: + order.append(s.pid) + if not s.start_ts: + s.start_ts = row["ts"] + s.export_count += 1 + did = row["doc_id"] + try: + s.export_id_to_mtime[did] = int(row["mtime"]) + except (TypeError, ValueError): + s.export_id_to_mtime[did] = 0 + s.export_id_to_rel[did] = row.get("relative_path", "") + + for row in _iter_csv(csv_dir / "fs_walk.csv"): + s = _get(row["pid"]) + if s.pid not in order: + order.append(s.pid) + decision = row["decision"] + s.walk_decision[decision] += 1 + s.walk_total += 1 + rel = row["rel"] + try: + disk_mtime = int(row["mtime"]) + except (TypeError, ValueError): + disk_mtime = 0 + if decision == "matched": + s.matched_rels.add(rel) + elif decision == "stale": + try: + idx_mtime = int(row["idx_mtime"]) + except (TypeError, ValueError): + idx_mtime = 0 + s.stale_rows.append((rel, disk_mtime, idx_mtime)) + elif decision == "missing": + s.missing_rows.append((rel, disk_mtime)) + + for row in _iter_csv(csv_dir / "orphan.csv"): + s = _get(row["pid"]) + s.orphans.append(row["doc_id"]) + + for row in _iter_csv(csv_dir / "commit.csv"): + s = _get(row["pid"]) + ok = row.get("success", "0") == "1" + if ok: + s.commits_ok += 1 + else: + s.commits_fail += 1 + s.commit_errors.append((row["ts"], row["collection"], row.get("error", ""))) + + for row in _iter_csv(csv_dir / "parse.csv"): + s = _get(row["pid"]) + s.parse_total += 1 + if row.get("ok", "0") != "1": + s.parse_fail += 1 + s.parse_errors.append((row["ts"], row["rel"], row.get("error", ""))) + + for row in _iter_csv(csv_dir / "enqueue.csv"): + s = _get(row["pid"]) + if row.get("is_new", "0") == "1": + s.enqueued += 1 + reason = row.get("reason", "") + if reason: + s.enqueue_by_reason[reason] += 1 + else: + s.deduped += 1 + + for row in _iter_csv(csv_dir / "watcher.csv"): + s = _get(row["pid"]) + s.watcher_events[row["action"]] += 1 + + return [sessions[pid] for pid in order] + + +# -- presenters --------------------------------------------------------------- + +def _bucket_diff(seconds: int) -> str: + """Human-readable mtime delta bucket.""" + a = abs(seconds) + if a < 60: return f"{seconds:+d}s" + if a < 3600: return f"{seconds // 60:+d}m" + if a < 86400: return f"{seconds // 3600:+d}h" + return f"{seconds // 86400:+d}d" + + +def cmd_sessions(sessions: list[Session]) -> None: + print(f"{'#':>3} {'pid':>6} {'started':<23} {'stopped':<23} {'exported':>9} {'walked':>8}") + for i, s in enumerate(sessions, 1): + print(f"{i:>3} {s.pid:>6} {s.start_ts or '-':<23} {s.stop_ts or '-':<23} " + f"{s.export_count:>9,} {s.walk_total:>8,}") + + +def cmd_summary(sessions: list[Session]) -> None: + cmd_sessions(sessions) + print() + print(f"{'#':>3} {'matched':>9} {'stale':>7} {'missing':>8} {'orphans':>8} " + f"{'enq_new':>8} {'enq_dup':>8} {'commit_ok':>10} {'commit_fail':>12} notes") + for i, s in enumerate(sessions, 1): + notes = [] + if s.export_count == 0 and s.walk_total > 0: + notes.append("INDEX_EMPTY_ON_OPEN") + if s.walk_total and s.walk_decision.get("missing", 0) == s.walk_total: + notes.append("ALL_MISSING") + if s.commits_fail: + notes.append(f"COMMIT_FAIL={s.commits_fail}") + if s.orphans and s.export_count and len(s.orphans) > s.export_count * 0.5: + notes.append(f"ORPHAN_HEAVY ({len(s.orphans)}/{s.export_count})") + if s.parse_fail: + notes.append(f"PARSE_FAIL={s.parse_fail}") + print( + f"{i:>3} " + f"{s.walk_decision.get('matched',0):>9,} " + f"{s.walk_decision.get('stale',0):>7,} " + f"{s.walk_decision.get('missing',0):>8,} " + f"{len(s.orphans):>8,} " + f"{s.enqueued:>8,} " + f"{s.deduped:>8,} " + f"{s.commits_ok:>10,} " + f"{s.commits_fail:>12,} " + f"{' '.join(notes)}" + ) + + # Cross-session signal: did the index size collapse between two restarts? + print() + print("Cross-session checks:") + if len(sessions) < 2: + print(" (need at least two sessions)") + return + prev = sessions[0] + for cur in sessions[1:]: + delta = cur.export_count - prev.export_count + flag = "" + if cur.export_count == 0 and prev.export_count > 0: + flag = " <-- INDEX WIPED" + elif prev.export_count and abs(delta) > prev.export_count * 0.25: + flag = f" <-- LARGE DELTA ({delta:+,})" + print(f" pid {prev.pid} -> pid {cur.pid}: exported {prev.export_count:,} -> {cur.export_count:,} ({delta:+,}){flag}") + prev = cur + + +def cmd_stales(sessions: list[Session], top: int = 30, samples: int = 20) -> None: + for i, s in enumerate(sessions, 1): + if not s.stale_rows: + continue + print(f"=== session #{i} pid={s.pid} stales={len(s.stale_rows):,} ===") + buckets: Counter = Counter() + for _, dm, im in s.stale_rows: + buckets[_bucket_diff(dm - im)] += 1 + print(" top mtime-delta buckets:") + for b, n in buckets.most_common(top): + print(f" {b:>10} {n:,}") + print(" sample stales:") + for rel, dm, im in s.stale_rows[:samples]: + print(f" {_bucket_diff(dm-im):>8} disk={dm} idx={im} {rel}") + print() + + +def cmd_missing(sessions: list[Session], samples: int = 30) -> None: + for i, s in enumerate(sessions, 1): + if not s.missing_rows: + continue + print(f"=== session #{i} pid={s.pid} missing={len(s.missing_rows):,} ===") + for rel, dm in s.missing_rows[:samples]: + print(f" disk={dm} {rel}") + if len(s.missing_rows) > samples: + print(f" ... and {len(s.missing_rows) - samples:,} more") + print() + + +def cmd_orphans(sessions: list[Session], samples: int = 30) -> None: + for i, s in enumerate(sessions, 1): + if not s.orphans: + continue + print(f"=== session #{i} pid={s.pid} orphans={len(s.orphans):,} ===") + # cross-reference against the export to print a relative_path if we have one + rel_lookup = s.export_id_to_rel + for did in s.orphans[:samples]: + rel = rel_lookup.get(did, "(unknown)") + print(f" {did} {rel}") + if len(s.orphans) > samples: + print(f" ... and {len(s.orphans) - samples:,} more") + print() + + +def cmd_errors(sessions: list[Session]) -> None: + any_errors = False + for i, s in enumerate(sessions, 1): + if not (s.commit_errors or s.parse_errors): + continue + any_errors = True + print(f"=== session #{i} pid={s.pid} ===") + if s.commit_errors: + print(f" commit failures: {len(s.commit_errors)}") + for ts, coll, err in s.commit_errors[:20]: + print(f" {ts} {coll} {err}") + if len(s.commit_errors) > 20: + print(f" ... and {len(s.commit_errors) - 20} more") + if s.parse_errors: + print(f" parse failures: {len(s.parse_errors)}") + for ts, rel, err in s.parse_errors[:20]: + print(f" {ts} {rel} {err}") + if len(s.parse_errors) > 20: + print(f" ... and {len(s.parse_errors) - 20} more") + print() + if not any_errors: + print("No commit or parse errors recorded.") + + +def cmd_index_diff(sessions: list[Session]) -> None: + """Compare the index-export set between the most recent two sessions.""" + if len(sessions) < 2: + print("Need at least two sessions to diff.") + return + prev, cur = sessions[-2], sessions[-1] + prev_ids = set(prev.export_id_to_mtime) + cur_ids = set(cur.export_id_to_mtime) + + disappeared = prev_ids - cur_ids + appeared = cur_ids - prev_ids + common = prev_ids & cur_ids + mtime_changed = [ + did for did in common + if prev.export_id_to_mtime[did] != cur.export_id_to_mtime[did] + ] + + print(f"=== diff: session pid={prev.pid} -> pid={cur.pid} ===") + print(f" prev exported: {len(prev_ids):,}") + print(f" cur exported: {len(cur_ids):,}") + print(f" disappeared: {len(disappeared):,}") + print(f" appeared (new): {len(appeared):,}") + print(f" mtime changed: {len(mtime_changed):,}") + print() + if disappeared: + print(" sample disappeared (in prev index, missing from cur):") + for did in list(disappeared)[:15]: + rel = prev.export_id_to_rel.get(did, "") + mt = prev.export_id_to_mtime.get(did, 0) + print(f" {did} prev_mtime={mt} {rel}") + if appeared: + print(" sample appeared (in cur index, not in prev):") + for did in list(appeared)[:15]: + rel = cur.export_id_to_rel.get(did, "") + mt = cur.export_id_to_mtime.get(did, 0) + print(f" {did} cur_mtime={mt} {rel}") + + +def cmd_flapping(sessions: list[Session], top: int = 30) -> None: + """Files that switched decision (matched <-> stale/missing) across sessions.""" + if len(sessions) < 2: + print("Need at least two sessions to detect flapping.") + return + per_rel: dict[str, list[str]] = defaultdict(list) + session_rels: list[set[str]] = [] + for s in sessions: + rels: set[str] = set() + rels.update(s.matched_rels) + for rel, _, _ in s.stale_rows: + per_rel[rel].append("stale") + rels.add(rel) + for rel, _ in s.missing_rows: + per_rel[rel].append("missing") + rels.add(rel) + # matched files contribute "matched" once + for rel in s.matched_rels: + per_rel[rel].append("matched") + session_rels.append(rels) + flap: Counter = Counter() + for rel, states in per_rel.items(): + unique = set(states) + if len(unique) > 1: + flap[rel] += sum(1 for s in states if s != "matched") + print(f"Files seen in >1 distinct decision state (top {top}):") + for rel, score in flap.most_common(top): + states = per_rel[rel] + print(f" score={score:>3} {rel} states={','.join(states)}") + + +# -- main --------------------------------------------------------------------- + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--csv-dir", default=None, help="CSV log dir (default: %LOCALAPPDATA%/tscodesearch/csv)") + ap.add_argument("--mode", default="summary", + choices=["summary", "sessions", "stales", "missing", + "orphans", "errors", "index-diff", "flapping"]) + ap.add_argument("--session", type=int, default=0, + help="Filter output to session N (1 = oldest). 0 = all.") + ap.add_argument("--samples", type=int, default=20) + args = ap.parse_args(argv) + + csv_dir = Path(args.csv_dir) if args.csv_dir else _default_csv_dir() + if not csv_dir.exists(): + print(f"CSV directory not found: {csv_dir}", file=sys.stderr) + return 1 + print(f"CSV dir: {csv_dir}") + print() + + sessions = load(csv_dir) + if not sessions: + print("No sessions found.") + return 0 + + if args.session: + if not (1 <= args.session <= len(sessions)): + print(f"--session must be 1..{len(sessions)}", file=sys.stderr) + return 1 + sessions = [sessions[args.session - 1]] + + mode = args.mode + if mode == "sessions": cmd_sessions(sessions) + elif mode == "summary": cmd_summary(sessions) + elif mode == "stales": cmd_stales(sessions, samples=args.samples) + elif mode == "missing": cmd_missing(sessions, samples=args.samples) + elif mode == "orphans": cmd_orphans(sessions, samples=args.samples) + elif mode == "errors": cmd_errors(sessions) + elif mode == "index-diff": cmd_index_diff(sessions) + elif mode == "flapping": cmd_flapping(sessions) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_daemon_lock.mjs b/tests/test_daemon_lock.mjs new file mode 100644 index 0000000..cad3c30 --- /dev/null +++ b/tests/test_daemon_lock.mjs @@ -0,0 +1,301 @@ +/** + * test_daemon_lock.mjs -- unit tests for lib/daemon_lock.mjs + * + * Run with: + * node --test tests/test_daemon_lock.mjs + * + * Each test spawns a real Python subprocess using the client venv so that + * the OS locking primitive is the same one used by the daemon. No daemon + * needs to be running. + */ + +import { test, describe, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import path from 'node:path'; +import os from 'node:os'; +import { fileURLToPath } from 'node:url'; +import { spawnSync, spawn } from 'node:child_process'; + +import { isLockHeld, waitForLockReleased } from '../lib/daemon_lock.mjs'; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +const REPO = path.dirname(path.dirname(fileURLToPath(import.meta.url))); + +function clientVenvPython() { + if (process.platform === 'win32') { + return path.join(REPO, '.client-venv', 'Scripts', 'python.exe'); + } + return path.join(REPO, '.client-venv', 'bin', 'python'); +} + +const PY = clientVenvPython(); + +/** Create a fresh temp file path (file does not exist yet). */ +function tmpLockPath() { + return path.join(os.tmpdir(), `tstest-lock-${process.pid}-${Date.now()}.lock`); +} + +/** + * Spawn a Python subprocess that acquires an exclusive OS lock on `lockPath`, + * writes "READY\n" to stdout, then sleeps for `holdSecs` seconds. + * + * Returns the child process. Call child.kill() to release the lock early. + * + * The caller must await `whenReady` (a Promise that resolves when "READY" is + * received) before assuming the lock is held. + */ +function spawnLockHolder(lockPath, holdSecs = 10) { + const script = process.platform === 'win32' + ? [ + 'import msvcrt,os,sys,time', + 'fh=open(sys.argv[1],"w")', + 'fh.seek(0)', + 'msvcrt.locking(fh.fileno(),msvcrt.LK_NBLCK,1)', + 'fh.write(str(os.getpid()))', + 'fh.flush()', + 'sys.stdout.write("READY\\n")', + 'sys.stdout.flush()', + 'time.sleep(float(sys.argv[2]))', + ].join('\n') + : [ + 'import fcntl,os,sys,time', + 'fh=open(sys.argv[1],"w")', + 'fcntl.flock(fh,fcntl.LOCK_EX|fcntl.LOCK_NB)', + 'fh.write(str(os.getpid()))', + 'fh.flush()', + 'sys.stdout.write("READY\\n")', + 'sys.stdout.flush()', + 'time.sleep(float(sys.argv[2]))', + ].join('\n'); + + const child = spawn(PY, ['-c', script, lockPath, String(holdSecs)], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + + let buf = ''; + const whenReady = new Promise((resolve, reject) => { + child.stdout.on('data', chunk => { + buf += chunk.toString(); + if (buf.includes('READY')) resolve(); + }); + child.on('error', reject); + child.on('exit', code => { + if (code !== 0 && !buf.includes('READY')) { + reject(new Error(`lock-holder exited ${code} before READY`)); + } + }); + // Safety: reject after 5s if READY never arrives. + setTimeout(() => reject(new Error('lock-holder READY timeout')), 5000); + }); + + return { child, whenReady }; +} + +// --------------------------------------------------------------------------- +// isLockHeld tests +// --------------------------------------------------------------------------- + +describe('isLockHeld', () => { + + test('returns false when lock file does not exist', () => { + const lockPath = tmpLockPath(); + assert.equal(isLockHeld(lockPath, PY), false); + }); + + test('returns false when lock file exists but is not locked', () => { + const lockPath = tmpLockPath(); + fs.writeFileSync(lockPath, '99999'); + try { + assert.equal(isLockHeld(lockPath, PY), false); + } finally { + fs.unlinkSync(lockPath); + } + }); + + test('returns true when a live process holds the lock', async () => { + const lockPath = tmpLockPath(); + const { child, whenReady } = spawnLockHolder(lockPath, 10); + try { + await whenReady; + assert.equal(isLockHeld(lockPath, PY), true); + } finally { + child.kill(); + } + }); + + test('returns false after the holding process exits', async () => { + const lockPath = tmpLockPath(); + const { child, whenReady } = spawnLockHolder(lockPath, 0.1); + await whenReady; + // Wait for the short sleep + process exit. + await new Promise(r => child.on('exit', r)); + assert.equal(isLockHeld(lockPath, PY), false); + }); + + test('probe itself does not leave the lock held (re-entrant safety)', async () => { + // Call isLockHeld twice in quick succession when the file is unlocked -- + // the second call should also return false, not block on its own probe. + const lockPath = tmpLockPath(); + fs.writeFileSync(lockPath, '0'); + try { + assert.equal(isLockHeld(lockPath, PY), false); + assert.equal(isLockHeld(lockPath, PY), false); + } finally { + fs.unlinkSync(lockPath); + } + }); + + test('returns true when called while holder is mid-sleep (stress x5)', async () => { + const lockPath = tmpLockPath(); + const { child, whenReady } = spawnLockHolder(lockPath, 10); + try { + await whenReady; + for (let i = 0; i < 5; i++) { + assert.equal(isLockHeld(lockPath, PY), true, `iteration ${i}`); + } + } finally { + child.kill(); + } + }); + +}); + +// --------------------------------------------------------------------------- +// waitForLockReleased tests +// --------------------------------------------------------------------------- + +describe('waitForLockReleased', () => { + + test('resolves immediately when file does not exist', async () => { + const lockPath = tmpLockPath(); + const released = await waitForLockReleased(lockPath, PY, { timeoutMs: 2000 }); + assert.equal(released, true); + }); + + test('resolves immediately when file exists but is not locked', async () => { + const lockPath = tmpLockPath(); + fs.writeFileSync(lockPath, '0'); + try { + const released = await waitForLockReleased(lockPath, PY, { timeoutMs: 2000 }); + assert.equal(released, true); + } finally { + fs.unlinkSync(lockPath); + } + }); + + test('returns false on timeout when lock is never released', async () => { + const lockPath = tmpLockPath(); + const { child, whenReady } = spawnLockHolder(lockPath, 30); + try { + await whenReady; + const released = await waitForLockReleased(lockPath, PY, { + timeoutMs: 600, + intervalMs: 100, + }); + assert.equal(released, false); + } finally { + child.kill(); + } + }); + + test('resolves true when lock releases during the wait window', async () => { + const lockPath = tmpLockPath(); + // Holder holds for 0.4s; waitForLockReleased waits up to 5s. + const { child, whenReady } = spawnLockHolder(lockPath, 0.4); + await whenReady; + const released = await waitForLockReleased(lockPath, PY, { + timeoutMs: 5000, + intervalMs: 100, + }); + assert.equal(released, true); + child.kill(); // no-op if already exited + }); + + test('resolves true when holder is killed during the wait', async () => { + const lockPath = tmpLockPath(); + const { child, whenReady } = spawnLockHolder(lockPath, 30); + await whenReady; + + // Kill the holder 300ms into the wait. + setTimeout(() => child.kill(), 300); + + const released = await waitForLockReleased(lockPath, PY, { + timeoutMs: 5000, + intervalMs: 100, + }); + assert.equal(released, true); + }); + + test('custom intervalMs is respected (short polling interval)', async () => { + const lockPath = tmpLockPath(); + const { child, whenReady } = spawnLockHolder(lockPath, 0.2); + await whenReady; + + const t0 = Date.now(); + const released = await waitForLockReleased(lockPath, PY, { + timeoutMs: 5000, + intervalMs: 50, + }); + const elapsed = Date.now() - t0; + + assert.equal(released, true); + // Should have finished in well under 2s (holder sleeps only 200ms). + assert.ok(elapsed < 2000, `elapsed ${elapsed}ms should be < 2000ms`); + child.kill(); + }); + +}); + +// --------------------------------------------------------------------------- +// Edge / adversarial cases +// --------------------------------------------------------------------------- + +describe('edge cases', () => { + + test('isLockHeld handles a bad python path without throwing', () => { + const lockPath = tmpLockPath(); + fs.writeFileSync(lockPath, '0'); + try { + // Bad python path -> spawnSync gets an error -> treated as held (conservative). + const result = isLockHeld(lockPath, '/no/such/python'); + // We accept either true (spawn error = held) or false (file-not-locked fallback). + // The important thing is it does not throw. + assert.ok(typeof result === 'boolean'); + } finally { + fs.unlinkSync(lockPath); + } + }); + + test('isLockHeld handles an empty file (zero bytes)', () => { + const lockPath = tmpLockPath(); + fs.writeFileSync(lockPath, ''); + try { + // An empty unlocked file should read as not held. + assert.equal(isLockHeld(lockPath, PY), false); + } finally { + fs.unlinkSync(lockPath); + } + }); + + test('concurrent isLockHeld calls do not interfere', async () => { + const lockPath = tmpLockPath(); + const { child, whenReady } = spawnLockHolder(lockPath, 10); + try { + await whenReady; + // Fire 4 concurrent isLockHeld calls -- all should return true. + const results = await Promise.all( + Array.from({ length: 4 }, () => + Promise.resolve(isLockHeld(lockPath, PY)) + ) + ); + assert.ok(results.every(r => r === true), `all should be true: ${results}`); + } finally { + child.kill(); + } + }); + +}); diff --git a/tests/unit/test_daemon_lock.py b/tests/unit/test_daemon_lock.py new file mode 100644 index 0000000..a55a4e2 --- /dev/null +++ b/tests/unit/test_daemon_lock.py @@ -0,0 +1,351 @@ +""" +Unit tests for _try_acquire_lock() and the daemon process-lock mechanism. + +Tests cover: + TryAcquireLock -- basic acquire / lock-file content / idempotency guard + LockExclusivity -- a subprocess cannot steal the lock while the parent holds it + LockRelease -- lock is released when the holding process exits + GhostSessions -- csv_log.configure() is only called AFTER the lock is held + (i.e. no session.csv row is written for a failed start) +""" + +from __future__ import annotations + +import importlib +import os +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest.mock import patch, MagicMock + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +REPO = Path(__file__).parent.parent.parent +_PY = sys.executable # the venv Python running the tests + +# Platform-appropriate snippet: acquire the lock non-blocking, exit 0 if +# free, exit 1 if held. +if sys.platform == 'win32': + _PROBE = '\n'.join([ + 'import msvcrt,sys', + 'fh=open(sys.argv[1],"r+b");fh.seek(0);msvcrt.locking(fh.fileno(),msvcrt.LK_NBLCK,1);fh.close()', + ]) +else: + _PROBE = '\n'.join([ + 'import fcntl,sys', + 'fh=open(sys.argv[1],"r+");fcntl.flock(fh,fcntl.LOCK_EX|fcntl.LOCK_NB);fh.close()', + ]) + +# Lock-holder script: acquires lock, prints READY, sleeps for argv[2] seconds. +if sys.platform == 'win32': + _HOLD = '\n'.join([ + 'import msvcrt,os,sys,time', + 'fh=open(sys.argv[1],"w")', + 'fh.seek(0)', + 'msvcrt.locking(fh.fileno(),msvcrt.LK_NBLCK,1)', + 'fh.write(str(os.getpid()))', + 'fh.flush()', + 'sys.stdout.write("READY\\n")', + 'sys.stdout.flush()', + 'time.sleep(float(sys.argv[2]))', + ]) +else: + _HOLD = '\n'.join([ + 'import fcntl,os,sys,time', + 'fh=open(sys.argv[1],"w")', + 'fcntl.flock(fh,fcntl.LOCK_EX|fcntl.LOCK_NB)', + 'fh.write(str(os.getpid()))', + 'fh.flush()', + 'sys.stdout.write("READY\\n")', + 'sys.stdout.flush()', + 'time.sleep(float(sys.argv[2]))', + ]) + + +def _probe_lock(lock_path: str) -> bool: + """Return True if lock_path is currently held by another process.""" + r = subprocess.run( + [_PY, '-c', _PROBE, lock_path], + capture_output=True, timeout=5, + ) + return r.returncode != 0 + + +def _spawn_holder(lock_path: str, hold_secs: float = 10) -> subprocess.Popen: + """Spawn a subprocess that holds the lock for hold_secs seconds. + + The caller must read one line from proc.stdout to know the lock is held. + """ + return subprocess.Popen( + [_PY, '-c', _HOLD, lock_path, str(hold_secs)], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + ) + + +def _wait_ready(proc: subprocess.Popen, timeout: float = 5.0) -> None: + """Block until the holder subprocess prints READY.""" + deadline = time.monotonic() + timeout + buf = b'' + while time.monotonic() < deadline: + chunk = proc.stdout.read1(64) # type: ignore[attr-defined] + if chunk: + buf += chunk + if b'READY' in buf: + return + time.sleep(0.05) + raise TimeoutError('lock-holder did not print READY in time') + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +def _load_daemon_module(): + """(Re)import indexserver.daemon so we get a fresh module state.""" + import indexserver.daemon as mod + return mod + + +class _LockTestBase(unittest.TestCase): + """Base class: patches _DAEMON_LOCK and _RUN_DIR to a temp dir, resets + _lock_fh after each test so the OS lock is released.""" + + def setUp(self): + self._tmpdir = tempfile.mkdtemp() + self._lock_path = os.path.join(self._tmpdir, 'daemon.lock') + import indexserver.daemon as _daemon + self._mod = _daemon + # Patch module-level paths so _try_acquire_lock uses our temp dir. + self._patches = [ + patch.object(_daemon, '_DAEMON_LOCK', Path(self._lock_path)), + patch.object(_daemon, '_RUN_DIR', Path(self._tmpdir)), + ] + for p in self._patches: + p.start() + + def tearDown(self): + # Release the OS lock by closing the file handle, then reset the global. + fh = self._mod._lock_fh + if fh is not None: + try: + if sys.platform == 'win32': + import msvcrt + try: + fh.seek(0) + msvcrt.locking(fh.fileno(), msvcrt.LK_UNLCK, 1) + except Exception: + pass + fh.close() + except Exception: + pass + self._mod._lock_fh = None + + for p in self._patches: + p.stop() + + import shutil + shutil.rmtree(self._tmpdir, ignore_errors=True) + + +# --------------------------------------------------------------------------- +# TryAcquireLock -- basic behaviour +# --------------------------------------------------------------------------- + +class TestTryAcquireLock(_LockTestBase): + + def test_returns_true_when_no_file_exists(self): + self.assertFalse(os.path.exists(self._lock_path)) + result = self._mod._try_acquire_lock() + self.assertTrue(result) + + def test_creates_lock_file_on_success(self): + self._mod._try_acquire_lock() + self.assertTrue(os.path.exists(self._lock_path)) + + def test_writes_pid_to_lock_file(self): + self._mod._try_acquire_lock() + fh = self._mod._lock_fh + # Read via the same file handle that holds the lock -- avoids the + # Windows byte-range lock that blocks other handles from reading byte 0. + fh.seek(0) + content = fh.read().strip() + self.assertEqual(content, str(os.getpid())) + + def test_sets_module_level_lock_fh(self): + self.assertIsNone(self._mod._lock_fh) + self._mod._try_acquire_lock() + self.assertIsNotNone(self._mod._lock_fh) + + def test_lock_file_is_actually_locked_after_acquire(self): + """After _try_acquire_lock succeeds, a probe subprocess must fail.""" + self._mod._try_acquire_lock() + self.assertTrue(_probe_lock(self._lock_path), + 'external probe should see the lock as held') + + def test_returns_true_when_existing_unlocked_file(self): + """Pre-existing lock file with no live holder (stale) is acquirable.""" + Path(self._lock_path).write_text('99999', encoding='ascii') + result = self._mod._try_acquire_lock() + self.assertTrue(result) + + def test_returns_false_when_lock_held_by_other_process(self): + """If another process holds the lock, _try_acquire_lock returns False.""" + holder = _spawn_holder(self._lock_path, hold_secs=10) + try: + _wait_ready(holder) + result = self._mod._try_acquire_lock() + self.assertFalse(result) + finally: + holder.kill() + holder.wait() + + def test_lock_fh_stays_none_on_failed_acquire(self): + holder = _spawn_holder(self._lock_path, hold_secs=10) + try: + _wait_ready(holder) + self._mod._try_acquire_lock() + self.assertIsNone(self._mod._lock_fh) + finally: + holder.kill() + holder.wait() + + +# --------------------------------------------------------------------------- +# LockExclusivity -- only one process holds the lock at a time +# --------------------------------------------------------------------------- + +class TestLockExclusivity(_LockTestBase): + + def test_two_concurrent_processes_cannot_both_acquire(self): + """Race two subprocesses: exactly one must win the lock.""" + results = [] + lock_path = self._lock_path + + def _try_in_thread(): + r = subprocess.run( + [_PY, '-c', _PROBE, lock_path], + capture_output=True, timeout=5, + ) + results.append(r.returncode) + + t1 = threading.Thread(target=_try_in_thread) + t2 = threading.Thread(target=_try_in_thread) + t1.start(); t2.start() + t1.join(); t2.join() + # When the file doesn't exist yet and both probes run concurrently, + # at most one should be able to take the lock; the other sees it held. + # Since neither holds long, both may see free -- that is acceptable. + # What must never happen: both return 1 (held) when the file was never + # locked by a third party. Just assert we got two boolean results. + self.assertEqual(len(results), 2) + self.assertTrue(all(r in (0, 1) for r in results)) + + def test_parent_holds_lock_subprocess_cannot_steal(self): + """After _try_acquire_lock, subprocesses must not be able to acquire.""" + self._mod._try_acquire_lock() + # Attempt acquisition from 3 separate subprocesses. + for i in range(3): + held = _probe_lock(self._lock_path) + self.assertTrue(held, f'subprocess {i} should not steal the lock') + + +# --------------------------------------------------------------------------- +# LockRelease -- lock freed when holding process exits +# --------------------------------------------------------------------------- + +class TestLockRelease(unittest.TestCase): + + def test_lock_released_when_holder_process_exits(self): + with tempfile.TemporaryDirectory() as tmpdir: + lock_path = os.path.join(tmpdir, 'daemon.lock') + holder = _spawn_holder(lock_path, hold_secs=0.2) + _wait_ready(holder) + + # Confirm held while alive. + self.assertTrue(_probe_lock(lock_path)) + + # Wait for the short sleep and process exit. + holder.wait(timeout=5) + + # Now the lock should be free. + self.assertFalse(_probe_lock(lock_path)) + + def test_lock_released_when_holder_is_killed(self): + with tempfile.TemporaryDirectory() as tmpdir: + lock_path = os.path.join(tmpdir, 'daemon.lock') + holder = _spawn_holder(lock_path, hold_secs=30) + _wait_ready(holder) + + self.assertTrue(_probe_lock(lock_path)) + + holder.kill() + holder.wait(timeout=5) + + # OS must release the byte-range / flock lock on process death. + self.assertFalse(_probe_lock(lock_path)) + + def test_probe_does_not_leave_lock_held(self): + """Verify that _probe_lock (and therefore isLockHeld) is not leaving a + lingering lock behind after it returns False.""" + with tempfile.TemporaryDirectory() as tmpdir: + lock_path = os.path.join(tmpdir, 'daemon.lock') + Path(lock_path).write_text('0', encoding='ascii') + + # Call probe twice -- it should succeed both times (no leak). + self.assertFalse(_probe_lock(lock_path)) + self.assertFalse(_probe_lock(lock_path)) + + +# --------------------------------------------------------------------------- +# GhostSessions -- csv_log.configure only runs after the lock is acquired +# --------------------------------------------------------------------------- + +class TestGhostSessions(_LockTestBase): + + def test_csv_log_not_configured_when_lock_fails(self): + """When _try_acquire_lock fails, csv_log.configure must not be called. + + This is the ghost-session fix: session.csv must never receive a start + row for a process that failed to get the lock. + """ + # Have a real holder so the lock is taken. + holder = _spawn_holder(self._lock_path, hold_secs=10) + try: + _wait_ready(holder) + + with patch.object(self._mod.csv_log, 'configure') as mock_cfg, \ + patch.object(self._mod, 'csv_log', self._mod.csv_log): + # Simulate the start_daemon() preamble: acquire lock first. + result = self._mod._try_acquire_lock() + self.assertFalse(result) + # csv_log.configure must NOT have been called -- it is only + # called inside start_daemon() after _try_acquire_lock succeeds. + mock_cfg.assert_not_called() + finally: + holder.kill() + holder.wait() + + def test_csv_log_configured_when_lock_succeeds(self): + """When the lock is free, start_daemon() must configure csv_log.""" + import indexserver.daemon as _daemon + # We cannot call start_daemon() in tests (it binds a port), so we + # validate the ordering by inspecting the source: csv_log.configure + # must appear AFTER the _try_acquire_lock() call in start_daemon(). + import inspect + src = inspect.getsource(_daemon.start_daemon) + lock_pos = src.find('_try_acquire_lock()') + csv_pos = src.find('csv_log.configure') + self.assertGreater(lock_pos, -1, 'start_daemon must call _try_acquire_lock()') + self.assertGreater(csv_pos, -1, 'start_daemon must call csv_log.configure') + self.assertGreater(csv_pos, lock_pos, + 'csv_log.configure must appear AFTER _try_acquire_lock() in start_daemon()') + + +if __name__ == '__main__': + unittest.main() diff --git a/ts.mjs b/ts.mjs index dad9a17..e45d069 100644 --- a/ts.mjs +++ b/ts.mjs @@ -25,6 +25,7 @@ import path from 'path'; import { fileURLToPath } from 'url'; import { spawnSync, spawn } from 'child_process'; import http from 'http'; +import { isLockHeld, waitForLockReleased } from './lib/daemon_lock.mjs'; const __filename = fileURLToPath(import.meta.url); const __dirname = path.dirname(__filename); @@ -216,10 +217,11 @@ async function waitForPortClosed(port, timeoutMs = 10_000) { } async function shutdownDaemon() { - const pidFile = path.join( - process.env.LOCALAPPDATA ?? path.join(process.env.USERPROFILE ?? '', 'AppData', 'Local'), - 'tscodesearch', 'daemon.pid' - ); + const runDir = daemonRunDir(); + const pidFile = path.join(runDir, 'daemon.pid'); + const lockFile = path.join(runDir, 'daemon.lock'); + const py = clientVenvPython(); + let daemonPid = null; try { daemonPid = fs.readFileSync(pidFile, 'utf-8').trim() || null; } catch { /* no pid file */ } @@ -231,26 +233,41 @@ async function shutdownDaemon() { return; } - // The daemon's stop_daemon() does up to 5s queue stop + - // backend close (waits for merge threads) + 5s server shutdown -- so allow - // up to 30s before assuming it's stuck. - const closed = await waitForPortClosed(PORT, 30_000); + // Phase 1: wait for the HTTP port to close (daemon shuts down its server + // mid-teardown, so this fires before backends are fully closed). + await waitForPortClosed(PORT, 30_000); - if (!closed && daemonPid) { - log(`Daemon did not exit after 30s -- force-killing pid ${daemonPid}...`); - if (process.platform === 'win32') { - spawnSync('taskkill', ['/F', '/PID', daemonPid], { stdio: 'pipe' }); - } else { - spawnSync('kill', ['-9', daemonPid], { stdio: 'pipe' }); + // Phase 2: wait for the OS file lock to release -- this only happens when + // the process is truly dead (port close is not sufficient: the process is + // still alive closing Tantivy backends after the server socket is gone). + const released = await waitForLockReleased(lockFile, py, { timeoutMs: 15_000 }); + + if (!released) { + // Verify the lock is still held before resorting to force-kill -- the + // final poll in waitForLockReleased may have raced with process exit. + if (!isLockHeld(lockFile, py)) return; + + if (daemonPid) { + log(`Daemon did not exit after 45s -- force-killing pid ${daemonPid}...`); + if (process.platform === 'win32') { + spawnSync('taskkill', ['/F', '/PID', daemonPid], { stdio: 'pipe' }); + } else { + spawnSync('kill', ['-9', daemonPid], { stdio: 'pipe' }); + } + // Confirm the kill took effect. + await waitForLockReleased(lockFile, py, { timeoutMs: 5_000 }); } - await waitForPortClosed(PORT, 5_000); } } -function daemonLogFile() { +function daemonRunDir() { const base = process.env.LOCALAPPDATA ?? path.join(process.env.USERPROFILE ?? '', 'AppData', 'Local'); - return path.join(base, 'tscodesearch', 'daemon.log'); + return path.join(base, 'tscodesearch'); +} + +function daemonLogFile() { + return path.join(daemonRunDir(), 'daemon.log'); } function startDaemon() { @@ -279,6 +296,7 @@ function printStatus(apiBody) { const collections = apiBody?.collections ?? {}; const watcher = apiBody?.watcher ?? {}; const queue = apiBody?.queue ?? {}; + const scan = apiBody?.scan ?? {}; const qDepth = queue.depth ?? 0; const qEnqueued = queue.enqueued ?? 0; @@ -310,6 +328,34 @@ function printStatus(apiBody) { console.log(` Queue : depth=${fmtNum(qDepth)} enqueued=${fmtNum(qEnqueued)} upserted=${fmtNum(qUpserted)} deduped=${fmtNum(qDeduped)} deleted=${fmtNum(qDeleted)}${errStr}${throttle}`); } + const scanState = scan.state ?? 'idle'; + const activeRoot = scan.active_root ?? ''; + const roots = scan.roots ?? {}; + const rootParts = Object.entries(roots).map(([name, info]) => { + const status = info?.status ?? '?'; + const phase = info?.phase ?? ''; + const fsFiles = info?.fs_files ?? 0; + const missing = info?.missing ?? 0; + const stale = info?.stale ?? 0; + const err = info?.errors ?? 0; + const errStr = err > 0 ? ` err=${fmtNum(err)}` : ''; + const phaseStr = phase ? ` ${phase}` : ''; + return `${name}:${status}${phaseStr} fs=${fmtNum(fsFiles)} miss=${fmtNum(missing)} stale=${fmtNum(stale)}${errStr}`; + }); + if (scanState === 'running') { + const active = activeRoot ? ` active=${activeRoot}` : ''; + console.log(` Scan : [>>] ${scanState}${active}`); + } else if (scanState === 'complete') { + console.log(` Scan : [OK] ${scanState}`); + } else if (scanState === 'cancelled') { + console.log(` Scan : [--] ${scanState}`); + } else { + console.log(` Scan : [--] ${scanState}`); + } + if (rootParts.length > 0) { + console.log(` ${rootParts.join(' | ')}`); + } + const state = watcher.state ?? (watcher.running ? 'watching' : 'stopped'); const watchQD = watcher.queue_depth ?? 0; if (state === 'watching') { From d130fa389d0a66846134b99247f82a624728679e Mon Sep 17 00:00:00 2001 From: Kristof Roomp Date: Tue, 26 May 2026 08:32:56 +0200 Subject: [PATCH 2/7] Potential fix for pull request finding 'Unused variable, import, function or class' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> --- tests/test_daemon_lock.mjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_daemon_lock.mjs b/tests/test_daemon_lock.mjs index cad3c30..5236362 100644 --- a/tests/test_daemon_lock.mjs +++ b/tests/test_daemon_lock.mjs @@ -9,7 +9,7 @@ * needs to be running. */ -import { test, describe, beforeEach, afterEach } from 'node:test'; +import { test, describe } from 'node:test'; import assert from 'node:assert/strict'; import fs from 'node:fs'; import path from 'node:path'; From 66b848135514f2d5b57328d11ea5d952840c7a9f Mon Sep 17 00:00:00 2001 From: Kristof Roomp Date: Tue, 26 May 2026 08:33:04 +0200 Subject: [PATCH 3/7] Potential fix for pull request finding 'Unused variable, import, function or class' Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com> --- tests/test_daemon_lock.mjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_daemon_lock.mjs b/tests/test_daemon_lock.mjs index 5236362..6186169 100644 --- a/tests/test_daemon_lock.mjs +++ b/tests/test_daemon_lock.mjs @@ -15,7 +15,7 @@ import fs from 'node:fs'; import path from 'node:path'; import os from 'node:os'; import { fileURLToPath } from 'node:url'; -import { spawnSync, spawn } from 'node:child_process'; +import { spawn } from 'node:child_process'; import { isLockHeld, waitForLockReleased } from '../lib/daemon_lock.mjs'; From f498489539c2a1ac9b08d4b61adf0265583e643f Mon Sep 17 00:00:00 2001 From: kristofr Date: Tue, 26 May 2026 10:15:06 +0200 Subject: [PATCH 4/7] Centralize daemon logging and expose runtime logs via ts log --- AGENTS.md | 305 ++++++++++++++++++++++++++++++++- CLAUDE.md | 302 +------------------------------- indexserver/backend.py | 58 +++++-- indexserver/csv_log.py | 211 ----------------------- indexserver/daemon.py | 143 +++++----------- indexserver/debug_logger.py | 254 +++++++++++++++++++++++++++ indexserver/index_queue.py | 76 +++++--- indexserver/indexer.py | 42 +++-- indexserver/runtime_logger.py | 52 ++++++ indexserver/verifier.py | 263 +++++++++++++++------------- indexserver/watcher.py | 40 +++-- query/config.py | 4 +- scripts/scan_csv.py | 13 +- tests/unit/test_daemon_lock.py | 28 +-- ts.mjs | 31 +++- 15 files changed, 996 insertions(+), 826 deletions(-) delete mode 100644 indexserver/csv_log.py create mode 100644 indexserver/debug_logger.py create mode 100644 indexserver/runtime_logger.py diff --git a/AGENTS.md b/AGENTS.md index ba18d17..dd3144a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,7 +1,10 @@ # Agent notes -Operational notes for Claude / future agents working in this repo. For -architecture, module map, and conventions see `CLAUDE.md`. +This is the canonical instructions file for repository guidance. +Put all long-form agent instructions in AGENTS.md. +CLAUDE.md is a placeholder that points here. + +Operational notes for agents working in this repo are maintained below. ## Debugging the indexer: real-time CSV logs @@ -28,7 +31,7 @@ Accepted values: | value | effect | |------------------------------------|-----------------------------------------------------------------| | `false` / `""` / unset | off | -| `true` / `"1"` / `"on"` / `"true"` | logs to `%LOCALAPPDATA%/tscodesearch/csv/` | +| `true` / `"1"` / `"on"` / `"true"` | logs to `/.tantivy_csv/` | | any other string | treated as an explicit directory path -- logs land there | Logging is opt-in and append-only; rows from successive restarts share a @@ -36,7 +39,7 @@ file and are distinguished by the per-row `pid` column. ### CSV files -Written by `indexserver/csv_log.py`, one file per event type. Each row +Written by `indexserver/debug_logger.py`, one file per event type. Each row starts with `ts` (ms-precision local time) and `pid`. | file | rows are written when | useful columns | @@ -111,3 +114,297 @@ Override the log directory with `--csv-dir DIR` or Set `"csv_debug": false` in `config.json` and restart. The files stick around for offline analysis; delete the `csv/` directory when done. + + +--- + +## Migrated Reference (from CLAUDE.md) + +# codesearch -- developer notes for Claude + +## CRITICAL: ASCII only -- no Unicode in source files + +Never introduce non-ASCII characters in code, comments, docstrings, output strings, or docs. +Reason: Windows `cp1252` and downstream tooling can misrender or fail on Unicode. + +Use plain ASCII substitutions, for example: +- arrows -> `->`, `<-`, `<->`, `=>` +- long dashes -> `-` or `--` +- ellipsis -> `...` +- bullets/checks/math symbols -> `-`/`*`, `OK`/`NO`, `>=`/`<=` + +Validation: +- `python -m scripts.find_nonascii` +- `python -m scripts.replace_nonascii --apply` + +## CRITICAL: running Python scripts from the Bash tool + +Everything runs in the **client venv** on Windows -- there is no separate WSL venv anymore. + +```bash +.client-venv/Scripts/python.exe