diff --git a/CHANGELOG.md b/CHANGELOG.md index adaa887..f2db89f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,35 @@ All notable changes to Code Context Control (C3) are documented here. The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.158.2] - 2026-10-21 + +### Fixed — approval requests that never reached the desk or the phone + +Clients learn of an override request from its line in +`.c3/notifications.jsonl`, and two things kept that line from existing. +`NotificationStore.add` deduplicated on agent and title, and a request's +title names only the tool, the operation and the file name. Once one request +on a file had been read, the next request on that file inside the cooldown +(5 minutes at the default `critical` severity) wrote nothing, so the desk and +the phone never showed it. A notification with a `ref_id` now matches only +entries with the same `ref_id`, in the cooldown, the duplicate collapse and +the retro-cleanup; decision lines carry the request id as `ref_id` for the +same reason. Second, every process that rewrote the file (the auto-ack on +each tool response, acks from the gateway, the collapse) did so without a +cross-process lock and by truncating it first, so a line appended in between, +or read as a truncated file, was lost. On one machine, 36 of the 63 requests +whose project still existed had no notification line. Rewrites now hold a lock on +`notifications.jsonl.lock` and publish by atomic replace; the Hub's "clear" +route goes through the store. + +A third gap was on the gateway: a session in a worktree nested inside a +registered project (`.claude/worktrees/`) files requests under the +worktree path, which is not registered, so `/api/mobile/overrides` dropped +the row and `GET`, `decide` and `mute` answered 404. A row inside a served +project is now served by that project. Its notification still lands in the +worktree's `.c3/`, outside the feed, so clients see it on their next +`/overrides` read (C3 Desk 0.15.1 re-reads every 30 s). + ## [2.158.1] - 2026-10-21 ### Fixed — FileMemory and BranchWatch stop re-reading every file map diff --git a/cli/c3.py b/cli/c3.py index 20f7be8..0c44411 100644 --- a/cli/c3.py +++ b/cli/c3.py @@ -91,7 +91,7 @@ # Config CONFIG_DIR = ".c3" CONFIG_FILE = ".c3/config.json" -__version__ = "2.158.1" +__version__ = "2.158.2" # The PreToolUse matcher for native subagent calls (installer and hub migration). AGENT_MATCHER = "Agent|Task" diff --git a/cli/hub_server.py b/cli/hub_server.py index 8ce9f56..343188a 100644 --- a/cli/hub_server.py +++ b/cli/hub_server.py @@ -1072,25 +1072,10 @@ def api_project_notifications_clear(): if not nf.exists(): return jsonify({"cleared": 0}) try: - entries = [] - for line in nf.read_text(encoding="utf-8").strip().splitlines(): - if not line.strip(): - continue - try: - entries.append(json.loads(line)) - except json.JSONDecodeError: - continue - count = 0 - for e in entries: - if not e.get("acknowledged"): - e["acknowledged"] = True - count += 1 - if count: - nf.write_text( - "\n".join(json.dumps(e, ensure_ascii=False) for e in entries) + "\n", - encoding="utf-8", - ) - return jsonify({"cleared": count}) + # Through the store: its rewrite holds the cross-process lock, so a + # notification another process appends meanwhile is kept. + from services.notifications import NotificationStore + return jsonify({"cleared": NotificationStore(path).acknowledge_all()}) except Exception as e: return jsonify({"error": str(e)}), 500 diff --git a/docs/override-requests.md b/docs/override-requests.md index ae7bc3a..6e13e1a 100644 --- a/docs/override-requests.md +++ b/docs/override-requests.md @@ -679,7 +679,12 @@ withheld for that reason: it is a canonical local-filesystem identity used for grant matching, and the client already has the human-readable `path`. The cross-project listing additionally drops rows whose project the gateway's scanner does not serve; the store is one file for the whole machine, and a -token's reach must not exceed the projects it can enumerate. +token's reach must not exceed the projects it can enumerate. A row filed from +a directory inside a served project, such as a session in a worktree under +`.claude/worktrees/`, is served by that project (since 2.158.2): the listing, +`GET /overrides/`, `decide` and `mute` all accept it. Its notification +lands in the worktree's own `.c3/`, which the feed does not scan, so a client +learns of it from its next `/overrides` read rather than from the feed. See the deviation list at the top of this document for the four places the implementation had to decide something §8 left open. diff --git a/oracle/services/mobile_api.py b/oracle/services/mobile_api.py index b5545f4..12df2a6 100644 --- a/oracle/services/mobile_api.py +++ b/oracle/services/mobile_api.py @@ -500,6 +500,38 @@ def _resolve_registered(raw: str) -> Path | None: return None +def _serving_project(raw: str) -> Path | None: + """The registered project that serves an override row's ``project_path``: + that project itself, or the deepest registered project containing it. + + A session in a worktree nested inside a registered project (Claude Code's + ``.claude/worktrees/``) files its requests under the worktree path, + which is never registered itself. Refusing those rows left the desk and + the phone unable to show or answer a hold the agent was blocked on. + """ + raw = (raw or "").strip() + if not raw or _scanner is None: + return None + try: + target = os.path.normcase(str(Path(raw).resolve())) + except (OSError, ValueError): + return None + best: Path | None = None + best_len = -1 + for proj in _scanner.discover(): + if not proj.get("has_c3"): + continue + try: + candidate = Path(proj["path"]).resolve() + except (OSError, ValueError): + continue + key = os.path.normcase(str(candidate)) + inside = target == key or target.startswith(key.rstrip("\\/") + os.sep) + if inside and len(key) > best_len: + best, best_len = candidate, len(key) + return best + + def _project_or_404(raw: str): """Returns (Path, None) or (None, error_response).""" if not (raw or "").strip(): @@ -2453,7 +2485,7 @@ def mobile_overrides_list(): # does not serve — the store is one file across every project on the # machine, including ones the scanner has since dropped. rows = [r for r in rows - if _resolve_registered(str(r.get("project_path") or "")) is not None] + if _serving_project(str(r.get("project_path") or "")) is not None] channels = _override_channels(rows) cards = [] for r in rows: @@ -2739,7 +2771,7 @@ def mobile_overrides_costs(): # whole machine, and a cross-project view must not name a project # this token cannot otherwise see. rules = [r for r in rules - if _resolve_registered(str(r.get("project_path") or "")) is not None] + if _serving_project(str(r.get("project_path") or "")) is not None] return jsonify({"days": days, "rules": rules, "count": len(rules), "project": project_path}) @@ -2754,7 +2786,7 @@ def mobile_override_get(request_id): row = orq.get(str(request_id)) if row is None: return jsonify({"error": "unknown request"}), 404 - if _resolve_registered(str(row.get("project_path") or "")) is None: + if _serving_project(str(row.get("project_path") or "")) is None: # Same refusal as an unknown id: whether a request exists for a project # this gateway does not serve is not this token's business. return jsonify({"error": "unknown request"}), 404 @@ -2823,7 +2855,7 @@ def mobile_override_decide(request_id): f"{', '.join(orq.MODES)}"}), 400 row = orq.get(str(request_id)) - if row is None or _resolve_registered(str(row.get("project_path") or "")) is None: + if row is None or _serving_project(str(row.get("project_path") or "")) is None: return jsonify({"error": "unknown request"}), 404 if row.get("status") != orq.STATUS_PENDING: # §12.7: it lapsed while the phone was showing it. 409, and the card @@ -2961,7 +2993,7 @@ def mobile_override_mute(request_id): from services import override_requests as orq data = request.get_json(silent=True) or {} row = orq.get(str(request_id)) - if row is None or _resolve_registered(str(row.get("project_path") or "")) is None: + if row is None or _serving_project(str(row.get("project_path") or "")) is None: return jsonify({"error": "unknown request"}), 404 if row.get("status") != orq.STATUS_PENDING: return jsonify({"error": f"request is {row.get('status')}, not pending", diff --git a/pyproject.toml b/pyproject.toml index 51fcf49..6b61f85 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "code-context-control" -version = "2.158.1" +version = "2.158.2" description = "Local MCP code-intelligence for AI coding tools: surgical search/read/edit, agent-config version history, path-level access + masking guards, and a multi-project hub." readme = "README.md" requires-python = ">=3.10" diff --git a/services/notifications.py b/services/notifications.py index caf6c36..f71c030 100644 --- a/services/notifications.py +++ b/services/notifications.py @@ -2,20 +2,38 @@ Persists to .c3/notifications.jsonl. Supports dedup, severity filtering, and auto-acknowledgement when surfaced to Claude via tool responses. + +Several processes write one project's file at once: every MCP server (the +auto-ack on each tool response), hook subprocesses (session open and end, the +override request filed at a confirm denial), the Oracle (the phone's and the +desk's acks) and the job supervisor. So every method that can rewrite the +file holds a cross-process lock, and a rewrite is published by atomic +replace; otherwise a line appended between another writer's read and its +rewrite is lost. """ import hashlib import json +import os import threading +import time import uuid +from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pathlib import Path +from services.atomic_json import write_text_atomic + # How long to suppress a repeated agent+title after it has been acknowledged. _COOLDOWN_MINUTES = {"critical": 5, "warning": 30, "info": 60} # Duplicate collapse: keep the most severe label when merging records. _SEVERITY_RANK = {"info": 0, "warning": 1, "critical": 2} +# How long a writer waits for another process's rewrite. Past it the write +# goes ahead unlocked rather than being dropped. +_LOCK_TIMEOUT_S = 5.0 +_LOCK_POLL_S = 0.01 + # Event kinds written by C3's own producers (v2.126.0). A client ROUTES on # ``kind`` + ``ref_id`` rather than parsing the title; every kind here also # lands in .c3/notifications.jsonl, which is one of the files the mobile @@ -63,18 +81,84 @@ def _entry_count(entry: dict) -> int: return 1 +class _FileLock: + """Advisory cross-process lock on a sidecar file, released by the OS if + the holder dies. The shape of ``services.task_store._FileLock``, except + that ``acquire`` gives up quietly: it returns False on timeout instead of + raising, and the caller carries on without the lock.""" + + def __init__(self, path: Path): + self.path = path + self._fh = None + + def acquire(self, timeout_s: float = _LOCK_TIMEOUT_S) -> bool: + try: + fh = open(self.path, "ab") # never read/written; fd exists to hold the lock + except OSError: + return False + deadline = time.monotonic() + timeout_s + while True: + try: + if os.name == "nt": + import msvcrt + fh.seek(0) + msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + self._fh = fh + return True + except OSError: + if time.monotonic() >= deadline: + fh.close() + return False + time.sleep(_LOCK_POLL_S) + + def release(self) -> None: + fh, self._fh = self._fh, None + if fh is None: + return + try: + if os.name == "nt": + import msvcrt + fh.seek(0) + msvcrt.locking(fh.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + fcntl.flock(fh.fileno(), fcntl.LOCK_UN) + except OSError: + pass + finally: + fh.close() + + class NotificationStore: """Thread-safe JSONL notification store for background agents.""" def __init__(self, project_path: str): self._file = Path(project_path) / ".c3" / "notifications.jsonl" self._file.parent.mkdir(parents=True, exist_ok=True) + self._lock_file = self._file.with_name(self._file.name + ".lock") self._lock = threading.Lock() # Retro-cleanup runs at most once per store instance, lazily on first # access, so pre-existing duplicate backlogs self-heal without a # dedicated maintenance pass. self._collapsed = False + @contextmanager + def _locked(self): + """This instance's thread lock plus the cross-process file lock. Every + method that can rewrite the file runs under both; instances are made + per call all over C3, so the thread lock alone never covered two + threads, let alone two processes.""" + with self._lock: + file_lock = _FileLock(self._lock_file) + file_lock.acquire() + try: + yield + finally: + file_lock.release() + def add(self, agent: str, severity: str, title: str, message: str, ai_enhanced: bool = False, replace_if_unacked: bool = False, kind: str = "", ref_id: str = "") -> dict | None: @@ -95,9 +179,12 @@ def add(self, agent: str, severity: str, title: str, message: str, ref_id: the id of the thing the notification is ABOUT (e.g. the override request), which is not the notification's own ``id``. A client needs both: ``id`` to acknowledge, ``ref_id`` to navigate. + It is also part of the dedup identity: an entry with a ref_id + only collapses into, or is suppressed by, an entry with the same + ref_id (unless ``replace_if_unacked``). Returns the entry if written/updated, None if deduped. """ - with self._lock: + with self._locked(): self._maybe_collapse_locked() message_hash = hashlib.md5((message or "").encode("utf-8")).hexdigest()[:12] cooldown = timedelta(minutes=_COOLDOWN_MINUTES.get(severity, 30)) @@ -106,6 +193,12 @@ def add(self, agent: str, severity: str, title: str, message: str, for existing in entries: if existing.get("agent") != agent or existing.get("title") != title: continue + # A second request on the same file is a new event, not a + # repeat: clients learn of requests only from this line. + # `replace_if_unacked` keeps title scope by request. + if (ref_id and not replace_if_unacked + and (existing.get("ref_id") or "") != ref_id): + continue if not existing.get("acknowledged"): if replace_if_unacked: # Update in-place — prevents repeated pile-up for chatty agents @@ -168,7 +261,7 @@ def add(self, agent: str, severity: str, title: str, message: str, def get_pending_count(self) -> int: """Return count of unacknowledged warning/critical notifications without consuming them.""" - with self._lock: + with self._locked(): self._maybe_collapse_locked() return sum( 1 for e in self._read_all() @@ -190,7 +283,7 @@ def get_unacknowledged(self, limit: int = 5, severities=None) -> list: """ if severities is None: severities = self._ACTIONABLE - with self._lock: + with self._locked(): self._maybe_collapse_locked() entries = [ e for e in self._read_all() @@ -202,7 +295,7 @@ def get_unacknowledged(self, limit: int = 5, severities=None) -> list: def get_suppressed_info_count(self) -> int: """Count of unacknowledged 'info' notifications not shown in actionable view.""" - with self._lock: + with self._locked(): self._maybe_collapse_locked() return sum( 1 for e in self._read_all() @@ -218,12 +311,12 @@ def get_history(self, limit: int = 50) -> list: def acknowledge(self, notification_id: str) -> bool: """Mark a single notification as acknowledged.""" - with self._lock: + with self._locked(): return self._set_ack(lambda e: e.get("id") == notification_id) def acknowledge_all(self) -> int: """Mark all unacknowledged notifications as acknowledged. Returns count.""" - with self._lock: + with self._locked(): entries = self._read_all() count = 0 for e in entries: @@ -239,7 +332,7 @@ def get_pending_summary(self) -> str: Auto-acknowledges those included. Returns empty string if none. """ - with self._lock: + with self._locked(): self._maybe_collapse_locked() entries = self._read_all() pending = [ @@ -282,7 +375,7 @@ def rotate_acknowledged(self, archive_fn, min_age_minutes: int = 120) -> int: than ``min_age_minutes`` also stay, so add()'s post-ack cooldown suppression window keeps working. Returns the count moved. """ - with self._lock: + with self._locked(): entries = self._read_all() horizon = datetime.now(timezone.utc) - timedelta( minutes=max(0, min_age_minutes)) @@ -312,14 +405,15 @@ def _old_enough(e: dict) -> bool: def collapse_duplicates(self) -> int: """Retro-cleanup: merge unacknowledged duplicates into one record each. - Unacknowledged entries sharing (agent, title) collapse into the most - recent record, which absorbs the group's total ``count``, the max - ``last_seen``, and the highest severity. The newest message wins — + Unacknowledged entries sharing (agent, title, ref_id) collapse into + the most recent record, which absorbs the group's total ``count``, the + max ``last_seen``, and the highest severity. The newest message wins — older same-title messages are stale snapshots of the same condition. + Entries about different things (different ``ref_id``) never merge. Acknowledged history is never touched. Returns the number of duplicate records removed. """ - with self._lock: + with self._locked(): self._collapsed = True return self._collapse_locked() @@ -334,13 +428,14 @@ def _maybe_collapse_locked(self): pass # maintenance is best-effort — never break reads/writes def _collapse_locked(self) -> int: - """Merge unacked duplicates (same agent+title). Caller must hold _lock.""" + """Merge unacked duplicates (same agent+title+ref_id). Caller must hold _locked().""" entries = self._read_all() groups: dict[tuple, list] = {} for e in entries: if e.get("acknowledged"): continue - groups.setdefault((e.get("agent"), e.get("title")), []).append(e) + groups.setdefault((e.get("agent"), e.get("title"), + e.get("ref_id") or ""), []).append(e) drop_ids: set[int] = set() removed = 0 @@ -377,10 +472,16 @@ def _read_all(self) -> list: return entries def _write_all(self, entries: list): - """Rewrite entire file. Caller must hold _lock.""" - with open(self._file, "w", encoding="utf-8") as f: - for e in entries: - f.write(json.dumps(e) + "\n") + """Rewrite entire file. Caller must hold _locked(). + + Published by atomic replace, so a reader that takes no lock (the + gateway feed, the hub) sees the old file or the new one, never the + truncated middle of a rewrite. No fsync: this is a feed, and what + readers need is the replace, not durability across a power cut. + """ + write_text_atomic(self._file, + "".join(json.dumps(e) + "\n" for e in entries), + fsync=False) def _set_ack(self, predicate) -> bool: """Acknowledge entries matching predicate. Caller must hold _lock.""" diff --git a/services/override_requests.py b/services/override_requests.py index b94a951..8764594 100644 --- a/services/override_requests.py +++ b/services/override_requests.py @@ -559,6 +559,11 @@ def _notify_decision(project_path: str, row: dict, grant=None) -> None: severity="info", title=f"Override {status}: {name}", message=message, + # Scopes the store's dedup to this request: two decisions on one + # file inside the info cooldown are two lines, and a client drops + # a card answered elsewhere when it sees one. No `kind`, so the + # line still reads as a decision rather than a request to route to. + ref_id=str(row.get("id") or ""), ) except Exception: pass # the decision already happened; a missing feed line never undoes it diff --git a/tests/test_mobile_override_routes.py b/tests/test_mobile_override_routes.py index 1ad5a86..286d04c 100644 --- a/tests/test_mobile_override_routes.py +++ b/tests/test_mobile_override_routes.py @@ -427,6 +427,33 @@ def test_unregistered_project_rows_are_not_served(self): self.assertEqual( self.get(f"/api/mobile/overrides/{row['id']}").status_code, 404) + def test_rows_from_a_worktree_inside_a_registered_project_are_served(self): + worktree = self.proj / ".claude" / "worktrees" / "wt-1" + (worktree / ".c3").mkdir(parents=True) + self.write_policy(worktree) + row = self.make_request(project=worktree) + + ids = [r["id"] for r in + self.get("/api/mobile/overrides?status=pending").get_json()["requests"]] + self.assertIn(row["id"], ids) + self.assertEqual( + self.get(f"/api/mobile/overrides/{row['id']}").status_code, 200) + res = self.post(f"/api/mobile/overrides/{row['id']}/decide", + {"decision": "deny"}) + self.assertEqual(res.status_code, 200) + + def test_a_sibling_sharing_a_name_prefix_is_not_inside_the_project(self): + sibling = Path(f"{self.proj}-copy") + (sibling / ".c3").mkdir(parents=True) + self.write_policy(sibling) + row = self.make_request(project=sibling) + + ids = [r["id"] for r in + self.get("/api/mobile/overrides").get_json()["requests"]] + self.assertNotIn(row["id"], ids) + self.assertEqual( + self.get(f"/api/mobile/overrides/{row['id']}").status_code, 404) + def test_filter_by_status(self): pending = self.make_request(name="p.env", rule="**/*.env") denied = self.make_request(name="d.env", rule="**/*.env") diff --git a/tests/test_notification_ref_scope.py b/tests/test_notification_ref_scope.py new file mode 100644 index 0000000..50862bb --- /dev/null +++ b/tests/test_notification_ref_scope.py @@ -0,0 +1,179 @@ +"""Approval requests reach the feed: ref_id-scoped dedup and a locked store. + +The desk and the phone learn of an override request only from its line in +``.c3/notifications.jsonl``. Two things kept that line from existing: + +- dedup keyed on agent+title: a second request on the same file inside the + cooldown after the first was acknowledged wrote nothing at all; +- unlocked read-modify-write from several processes: a rewrite from a stale + or truncated read dropped lines another process had just appended. +""" +from __future__ import annotations + +import json +import subprocess +import sys +import tempfile +import textwrap +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT)) + +from services import override_requests as orq # noqa: E402 +from services.notifications import NotificationStore # noqa: E402 + +_TITLE = "Approve Edit write on SKILL.md?" + + +def _lines(project: Path) -> list[dict]: + path = project / ".c3" / "notifications.jsonl" + if not path.is_file(): + return [] + return [json.loads(line) for line in + path.read_text(encoding="utf-8").splitlines() if line.strip()] + + +def _request_row(request_id: str, status: str = "pending") -> dict: + return {"id": request_id, "tool": "Edit", "op": "write", + "path": "skills/demo/SKILL.md", "rule": ".claude/**", + "rule_class": "access_confirm", "status": status} + + +class _Base(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.project = Path(self._tmp.name) / "proj" + (self.project / ".c3").mkdir(parents=True) + + def tearDown(self): + for _ in range(20): + try: + self._tmp.cleanup() + break + except OSError: + time.sleep(0.05) + + +class TestRefScopedDedup(_Base): + def test_acknowledged_request_does_not_suppress_the_next_on_the_same_file(self): + store = NotificationStore(str(self.project)) + first = store.add("override", "critical", _TITLE, "approve ovr_a", + kind="override", ref_id="ovr_a") + store.acknowledge(first["id"]) + + second = store.add("override", "critical", _TITLE, "approve ovr_b", + kind="override", ref_id="ovr_b") + + self.assertIsNotNone(second) + self.assertEqual({e.get("ref_id") for e in _lines(self.project)}, + {"ovr_a", "ovr_b"}) + + def test_pending_requests_sharing_a_title_survive_the_collapse(self): + store = NotificationStore(str(self.project)) + store.add("override", "critical", _TITLE, "approve ovr_a", + kind="override", ref_id="ovr_a") + store.add("override", "critical", _TITLE, "approve ovr_b", + kind="override", ref_id="ovr_b") + + removed = NotificationStore(str(self.project)).collapse_duplicates() + + self.assertEqual(removed, 0) + self.assertEqual(len(_lines(self.project)), 2) + + def test_the_same_ref_id_still_collapses_into_one_record(self): + store = NotificationStore(str(self.project)) + store.add("ci", "warning", "CI failed", "run 7", kind="ci", ref_id="run_7") + again = store.add("ci", "warning", "CI failed", "run 7", + kind="ci", ref_id="run_7") + + self.assertIsNone(again) + (only,) = _lines(self.project) + self.assertEqual(only["count"], 2) + + def test_replace_if_unacked_still_retargets_by_title(self): + store = NotificationStore(str(self.project)) + store.add("agent", "warning", "Index stale", "one", ref_id="r1") + store.add("agent", "warning", "Index stale", "two", ref_id="r2", + replace_if_unacked=True) + + (only,) = _lines(self.project) + self.assertEqual((only["ref_id"], only["message"]), ("r2", "two")) + + +class TestOverrideLines(_Base): + def test_second_request_on_a_file_is_notified_after_the_first_was_read(self): + policy = SimpleNamespace(notify_severity="critical") + orq._notify(str(self.project), _request_row("ovr_a"), policy) + store = NotificationStore(str(self.project)) + store.acknowledge(_lines(self.project)[0]["id"]) + + orq._notify(str(self.project), _request_row("ovr_b"), policy) + + refs = [e.get("ref_id") for e in _lines(self.project) + if e.get("kind") == "override"] + self.assertEqual(refs, ["ovr_a", "ovr_b"]) + + def test_two_decisions_on_one_file_are_two_lines(self): + orq._notify_decision(str(self.project), _request_row("ovr_a", "approved")) + store = NotificationStore(str(self.project)) + store.acknowledge(_lines(self.project)[0]["id"]) + + orq._notify_decision(str(self.project), _request_row("ovr_b", "approved")) + + decisions = _lines(self.project) + self.assertEqual([e.get("ref_id") for e in decisions], ["ovr_a", "ovr_b"]) + self.assertTrue(all(not e.get("kind") for e in decisions)) + + +_WRITER = textwrap.dedent(""" + import sys + sys.path.insert(0, sys.argv[1]) + from services.notifications import NotificationStore + project, tag, count = sys.argv[2], sys.argv[3], int(sys.argv[4]) + for i in range(count): + NotificationStore(project).add( + "override", "critical", f"req {tag}-{i}", "m", + kind="override", ref_id=f"{tag}-{i}") +""") + +_REWRITER = textwrap.dedent(""" + import os, sys + sys.path.insert(0, sys.argv[1]) + from services.notifications import NotificationStore + project, stop = sys.argv[2], sys.argv[3] + while not os.path.exists(stop): + store = NotificationStore(project) + store.get_pending_summary() + store.acknowledge_all() +""") + + +class TestConcurrentWriters(_Base): + def test_lines_appended_by_one_process_survive_rewrites_by_others(self): + writers, per_writer = 3, 30 + stop = Path(self._tmp.name) / "stop" + args = [str(REPO_ROOT), str(self.project)] + rewriters = [subprocess.Popen([sys.executable, "-c", _REWRITER, *args, str(stop)]) + for _ in range(2)] + try: + procs = [subprocess.Popen([sys.executable, "-c", _WRITER, *args, + f"w{n}", str(per_writer)]) + for n in range(writers)] + for proc in procs: + self.assertEqual(proc.wait(timeout=120), 0) + finally: + stop.touch() + for proc in rewriters: + proc.wait(timeout=60) + + refs = [e.get("ref_id") for e in _lines(self.project)] + self.assertEqual(len(refs), writers * per_writer) + self.assertEqual(len(set(refs)), writers * per_writer) + + +if __name__ == "__main__": + unittest.main()