From 0c72225311179e1ee02fa0b6534dbeea48d3500e Mon Sep 17 00:00:00 2001 From: KunjShah95 Date: Sun, 26 Jul 2026 22:58:12 +0530 Subject: [PATCH] fix: corrupt JSON store files must not crash server startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The small file-backed stores (Inbox, inbox routing, mention threads, channel subscriptions, unrouted/dead-letter, unattended flags, per-persona/per-session connection maps, and user risk overrides) each loaded their JSON file with an unguarded json.loads inside __init__ — and every one is constructed eagerly in SessionManager.__init__. So a single truncated or malformed file turned into an exception that aborted server startup entirely, recoverable only by finding and deleting the file by hand. Several of the same stores also wrote their files non-atomically (write_text in place), which is exactly what produces the corrupt file after a crash/kill/disk-full mid-write. This generalizes the resilient pattern already used inline by ChannelBuffer and WorkspaceTrustStore into two shared helpers (coworker/jsonstore.py): - read_json(): missing/corrupt file -> caller default, moving the bad file aside to .corrupt so it is neither lost nor re-hit on the next boot. - write_json_atomic(): temp sibling + os.replace, atomic on POSIX and Windows. Every affected store now uses them. Adds tests covering the helpers and each store's construct-from-corrupt-file path plus a save round-trip. Related to the WakeStore-specific report in #158 (that fix, #159, covers only selfwake.py); this addresses the same failure mode across the rest of the persistence layer. Co-Authored-By: Claude Opus 4.8 --- coworker/connections.py | 41 ++++----- coworker/inbox.py | 19 ++--- coworker/inbox_routing.py | 36 ++++---- coworker/jsonstore.py | 79 ++++++++++++++++++ coworker/mentions.py | 16 ++-- coworker/overrides.py | 25 ++---- coworker/subscriptions.py | 15 ++-- coworker/unattended.py | 12 +-- coworker/unrouted.py | 16 ++-- tests/test_json_store_resilience.py | 124 ++++++++++++++++++++++++++++ 10 files changed, 266 insertions(+), 117 deletions(-) create mode 100644 coworker/jsonstore.py create mode 100644 tests/test_json_store_resilience.py diff --git a/coworker/connections.py b/coworker/connections.py index 9088b53f..1a81bc21 100644 --- a/coworker/connections.py +++ b/coworker/connections.py @@ -22,11 +22,12 @@ from __future__ import annotations -import json import threading from pathlib import Path from typing import Optional +from .jsonstore import read_json, write_json_atomic + class PersonaConnectionStore: """``{persona_id: {connector: bool}}`` — the per-persona default on/off for each connector.""" @@ -38,21 +39,14 @@ def __init__(self, path: Optional[str | Path] = None) -> None: self._load() def _load(self) -> None: - if self.path and self.path.is_file(): - data = json.loads(self.path.read_text(encoding="utf-8")) - self._rows = { - pid: {str(c): bool(v) for c, v in (row or {}).items()} - for pid, row in data.get("personas", {}).items() - } + data = read_json(self.path, {}) or {} + self._rows = { + pid: {str(c): bool(v) for c, v in (row or {}).items()} + for pid, row in data.get("personas", {}).items() + } def _save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text( - json.dumps({"personas": self._rows}, indent=2), - encoding="utf-8", - ) + write_json_atomic(self.path, {"personas": self._rows}) # -- queries ---------------------------------------------------------------- def get(self, persona_id: str) -> dict[str, bool]: @@ -111,21 +105,14 @@ def __init__(self, path: Optional[str | Path] = None) -> None: self._load() def _load(self) -> None: - if self.path and self.path.is_file(): - data = json.loads(self.path.read_text(encoding="utf-8")) - self._rows = { - sid: {str(c): bool(v) for c, v in (row or {}).items()} - for sid, row in data.get("sessions", {}).items() - } + data = read_json(self.path, {}) or {} + self._rows = { + sid: {str(c): bool(v) for c, v in (row or {}).items()} + for sid, row in data.get("sessions", {}).items() + } def _save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text( - json.dumps({"sessions": self._rows}, indent=2), - encoding="utf-8", - ) + write_json_atomic(self.path, {"sessions": self._rows}) # -- queries ---------------------------------------------------------------- def get(self, session_id: str) -> dict[str, bool]: diff --git a/coworker/inbox.py b/coworker/inbox.py index 924e707a..fc96e0f4 100644 --- a/coworker/inbox.py +++ b/coworker/inbox.py @@ -22,6 +22,8 @@ from pathlib import Path from typing import Any, Optional +from .jsonstore import read_json, write_json_atomic + KIND_APPROVAL = "approval" KIND_QUESTION = "question" KIND_NOTIFICATION = "notification" @@ -97,19 +99,14 @@ def __init__(self, path: Optional[str | Path] = None) -> None: # -- persistence ------------------------------------------------------------ def _load(self) -> None: - if self.path and self.path.is_file(): - data = json.loads(self.path.read_text(encoding="utf-8")) - for raw in data.get("items", []): - item = InboxItem(**raw) - self._items[item.id] = item + data = read_json(self.path, {}) or {} + for raw in data.get("items", []): + item = InboxItem(**raw) + self._items[item.id] = item def _save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text( - json.dumps({"items": [asdict(i) for i in self._items.values()]}, indent=2), - encoding="utf-8", + write_json_atomic( + self.path, {"items": [asdict(i) for i in self._items.values()]} ) # -- adding ----------------------------------------------------------------- diff --git a/coworker/inbox_routing.py b/coworker/inbox_routing.py index ef115e95..104c178f 100644 --- a/coworker/inbox_routing.py +++ b/coworker/inbox_routing.py @@ -11,13 +11,14 @@ from __future__ import annotations -import json import re import threading from dataclasses import asdict, dataclass from pathlib import Path from typing import Callable, Optional +from .jsonstore import read_json, write_json_atomic + DEFAULT_INBOX = "default" # Embeds the item id in a delivered message. Emitted as [ow:…] since the bot's rebrand # to OpenWorker (2026-07-22); the legacy [ocw:…] spelling stays parseable so replies to @@ -44,28 +45,21 @@ def __init__(self, path: Optional[str | Path] = None) -> None: self._load() def _load(self) -> None: - if self.path and self.path.is_file(): - data = json.loads(self.path.read_text(encoding="utf-8")) - for raw in data.get("bindings", []): - b = InboxBinding(**raw) - self._bindings[b.name] = b - self._persona_default = dict(data.get("persona_default", {})) - self._session_override = dict(data.get("session_override", {})) + data = read_json(self.path, {}) or {} + for raw in data.get("bindings", []): + b = InboxBinding(**raw) + self._bindings[b.name] = b + self._persona_default = dict(data.get("persona_default", {})) + self._session_override = dict(data.get("session_override", {})) def _save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text( - json.dumps( - { - "bindings": [asdict(b) for b in self._bindings.values()], - "persona_default": self._persona_default, - "session_override": self._session_override, - }, - indent=2, - ), - encoding="utf-8", + write_json_atomic( + self.path, + { + "bindings": [asdict(b) for b in self._bindings.values()], + "persona_default": self._persona_default, + "session_override": self._session_override, + }, ) # -- config ----------------------------------------------------------------- diff --git a/coworker/jsonstore.py b/coworker/jsonstore.py new file mode 100644 index 00000000..a65dcc44 --- /dev/null +++ b/coworker/jsonstore.py @@ -0,0 +1,79 @@ +"""Durable JSON persistence helpers shared by the file-backed stores. + +The tiny JSON-backed stores (Inbox, subscriptions, mention threads, connection +overrides, …) are all constructed in ``SessionManager.__init__``, and each one loads its +file eagerly there. That makes their read/write robustness a *startup* concern, and two +failure modes bite: + +- **A corrupt/truncated file must never block startup.** An unguarded ``json.loads`` in a + store's ``_load`` turns one unreadable file into a server that won't boot at all — + recoverable only by finding and deleting the file by hand. ``read_json`` returns the + caller's default instead, moving the bad file aside (``.corrupt``) so it is neither + lost nor silently re-hit on the next load. +- **A save must be atomic.** Writing in place leaves a half-written file if the process dies + mid-write — which is exactly what produces the corrupt file above. ``write_json_atomic`` + writes a sibling temp file and ``os.replace``s it into place (atomic on POSIX and Windows). + +This just generalizes the pattern already applied inline to ``ChannelBuffer`` and +``WorkspaceTrustStore``, so every store gets the same treatment instead of a few. +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path +from typing import Any + +logger = logging.getLogger("coworker.jsonstore") + + +def read_json(path: str | Path | None, default: Any = None) -> Any: + """Load JSON from ``path``, tolerating a missing or unreadable file. + + Returns ``default`` when the path is ``None``, absent, or unreadable/corrupt. A corrupt + file is moved aside to ``.corrupt`` (best effort) so the bad data is preserved for + inspection and the next ``write_json_atomic`` starts from a clean slate rather than the + store re-hitting the same parse error on every boot. + """ + if path is None: + return default + p = Path(path) + if not p.is_file(): + return default + try: + return json.loads(p.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + logger.warning("ignoring unreadable JSON at %s (%s); moving it aside", p, exc) + _quarantine(p) + return default + + +def _quarantine(path: Path) -> None: + try: + path.replace(path.with_name(path.name + ".corrupt")) + except OSError: + pass + + +def write_json_atomic( + path: str | Path | None, data: Any, *, indent: int | None = 2 +) -> None: + """Serialize ``data`` and write it to ``path`` atomically, creating parent dirs. + + A no-op when ``path`` is ``None`` (the in-memory-only store case). Writes to a sibling + temp file first, then ``os.replace``s it over the target, so a crash mid-write can never + leave a truncated file that would brick the next startup. + """ + if path is None: + return + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + tmp = p.with_name(f".{p.name}.{os.getpid()}.tmp") + try: + tmp.write_text(json.dumps(data, indent=indent), encoding="utf-8") + os.replace(tmp, p) + except BaseException: + tmp.unlink(missing_ok=True) + raise diff --git a/coworker/mentions.py b/coworker/mentions.py index cd4a8717..bbce22ef 100644 --- a/coworker/mentions.py +++ b/coworker/mentions.py @@ -15,12 +15,13 @@ from __future__ import annotations -import json import threading from dataclasses import asdict, dataclass from pathlib import Path from typing import Optional +from .jsonstore import read_json, write_json_atomic + @dataclass class MentionThread: @@ -37,18 +38,11 @@ def __init__(self, path: Optional[str | Path] = None) -> None: self._load() def _load(self) -> None: - if self.path and self.path.is_file(): - data = json.loads(self.path.read_text(encoding="utf-8")) - self._threads = [MentionThread(**raw) for raw in data.get("threads", [])] + data = read_json(self.path, {}) or {} + self._threads = [MentionThread(**raw) for raw in data.get("threads", [])] def _save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text( - json.dumps({"threads": [asdict(t) for t in self._threads]}, indent=2), - encoding="utf-8", - ) + write_json_atomic(self.path, {"threads": [asdict(t) for t in self._threads]}) # -- mutations -------------------------------------------------------------- def set(self, thread_target: str, session_id: str, channel: str) -> MentionThread: diff --git a/coworker/overrides.py b/coworker/overrides.py index 0abb2cf8..8925c99b 100644 --- a/coworker/overrides.py +++ b/coworker/overrides.py @@ -11,12 +11,12 @@ from __future__ import annotations -import json from dataclasses import dataclass from fnmatch import fnmatchcase from pathlib import Path from typing import Callable, Optional +from .jsonstore import read_json, write_json_atomic from .risk import RiskClass @@ -39,32 +39,19 @@ def __init__(self, path: Optional[str | Path] = None) -> None: self._rules: list[_Rule] = self._load() def _load(self) -> list[_Rule]: - if not (self.path and self.path.is_file()): - return [] - data = json.loads(self.path.read_text(encoding="utf-8")) + data = read_json(self.path, {}) or {} rules = [] for r in data.get("rules", []): try: rules.append(_Rule(str(r["pattern"]), RiskClass(str(r["risk"])))) - except (KeyError, ValueError): + except (KeyError, ValueError, TypeError): continue # skip malformed rules rather than failing the whole store return rules def save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text( - json.dumps( - { - "rules": [ - {"pattern": r.pattern, "risk": r.risk.value} - for r in self._rules - ] - }, - indent=2, - ), - encoding="utf-8", + write_json_atomic( + self.path, + {"rules": [{"pattern": r.pattern, "risk": r.risk.value} for r in self._rules]}, ) def set_rule(self, pattern: str, risk: RiskClass | str) -> None: diff --git a/coworker/subscriptions.py b/coworker/subscriptions.py index f8b5d33a..3ec5961d 100644 --- a/coworker/subscriptions.py +++ b/coworker/subscriptions.py @@ -25,6 +25,8 @@ from pathlib import Path from typing import Optional +from .jsonstore import read_json, write_json_atomic + @dataclass class Subscription: @@ -42,17 +44,12 @@ def __init__(self, path: Optional[str | Path] = None) -> None: self._load() def _load(self) -> None: - if self.path and self.path.is_file(): - data = json.loads(self.path.read_text(encoding="utf-8")) - self._subs = [Subscription(**raw) for raw in data.get("subscriptions", [])] + data = read_json(self.path, {}) or {} + self._subs = [Subscription(**raw) for raw in data.get("subscriptions", [])] def _save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text( - json.dumps({"subscriptions": [asdict(s) for s in self._subs]}, indent=2), - encoding="utf-8", + write_json_atomic( + self.path, {"subscriptions": [asdict(s) for s in self._subs]} ) # -- mutations -------------------------------------------------------------- diff --git a/coworker/unattended.py b/coworker/unattended.py index 6b60873f..8224e9d0 100644 --- a/coworker/unattended.py +++ b/coworker/unattended.py @@ -8,25 +8,21 @@ from __future__ import annotations -import json import threading from pathlib import Path from typing import Optional +from .jsonstore import read_json, write_json_atomic + class UnattendedRegistry: def __init__(self, path: Optional[str | Path] = None) -> None: self.path = Path(path) if path else None self._lock = threading.Lock() - self._flags: dict[str, bool] = {} - if self.path and self.path.is_file(): - self._flags = dict(json.loads(self.path.read_text(encoding="utf-8"))) + self._flags: dict[str, bool] = dict(read_json(self.path, {}) or {}) def _save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text(json.dumps(self._flags, indent=2), encoding="utf-8") + write_json_atomic(self.path, self._flags) def is_unattended(self, session_id: str) -> bool: return bool(self._flags.get(session_id, False)) diff --git a/coworker/unrouted.py b/coworker/unrouted.py index e17328ac..38103916 100644 --- a/coworker/unrouted.py +++ b/coworker/unrouted.py @@ -11,13 +11,14 @@ from __future__ import annotations -import json import threading import time from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Optional +from .jsonstore import read_json, write_json_atomic + @dataclass class UnroutedItem: @@ -37,18 +38,11 @@ def __init__(self, path: Optional[str | Path] = None, *, cap: int = 200) -> None self._load() def _load(self) -> None: - if self.path and self.path.is_file(): - data = json.loads(self.path.read_text(encoding="utf-8")) - self._items = [UnroutedItem(**raw) for raw in data.get("items", [])] + data = read_json(self.path, {}) or {} + self._items = [UnroutedItem(**raw) for raw in data.get("items", [])] def _save(self) -> None: - if not self.path: - return - self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text( - json.dumps({"items": [asdict(i) for i in self._items]}, indent=2), - encoding="utf-8", - ) + write_json_atomic(self.path, {"items": [asdict(i) for i in self._items]}) def record(self, source: str, sender: str, text: str, reason: str) -> UnroutedItem: item = UnroutedItem( diff --git a/tests/test_json_store_resilience.py b/tests/test_json_store_resilience.py new file mode 100644 index 00000000..1e36e3e6 --- /dev/null +++ b/tests/test_json_store_resilience.py @@ -0,0 +1,124 @@ +"""A corrupt on-disk JSON file must never crash a store's construction (startup). + +Every file-backed store is built eagerly in ``SessionManager.__init__`` and loads its file +there, so an unreadable file used to raise straight out of the constructor and take the whole +server down. These tests pin the resilient contract for the shared helpers and for each store: +a garbage file loads as empty (and is quarantined), and a normal save round-trips atomically. +""" + +from __future__ import annotations + +from coworker.connections import PersonaConnectionStore, SessionConnectionStore +from coworker.inbox import InboxStore +from coworker.inbox_routing import DEFAULT_INBOX, InboxRouting +from coworker.jsonstore import read_json, write_json_atomic +from coworker.mentions import MentionSessionStore +from coworker.overrides import RiskOverrideStore +from coworker.risk import RiskClass +from coworker.subscriptions import SubscriptionStore +from coworker.unattended import UnattendedRegistry +from coworker.unrouted import UnroutedStore + +CORRUPT = '{"items": [ {"broken": ' # truncated mid-object + + +# -- jsonstore helpers ---------------------------------------------------------- +def test_read_json_missing_returns_default(tmp_path): + assert read_json(tmp_path / "nope.json", {"a": 1}) == {"a": 1} + assert read_json(None, []) == [] + + +def test_read_json_valid_round_trips(tmp_path): + p = tmp_path / "ok.json" + write_json_atomic(p, {"x": [1, 2, 3]}) + assert read_json(p) == {"x": [1, 2, 3]} + + +def test_read_json_corrupt_returns_default_and_quarantines(tmp_path): + p = tmp_path / "bad.json" + p.write_text(CORRUPT, encoding="utf-8") + assert read_json(p, {"fallback": True}) == {"fallback": True} + # The bad file is moved aside, not silently clobbered or left to re-crash next boot. + assert not p.exists() + assert (tmp_path / "bad.json.corrupt").read_text(encoding="utf-8") == CORRUPT + + +def test_write_json_atomic_leaves_no_temp_file(tmp_path): + p = tmp_path / "out.json" + write_json_atomic(p, {"k": "v"}) + assert read_json(p) == {"k": "v"} + # No stray sibling temp files left behind. + assert [f.name for f in tmp_path.iterdir()] == ["out.json"] + + +def test_write_json_atomic_none_path_is_noop(): + write_json_atomic(None, {"anything": 1}) # must not raise + + +# -- every store survives a corrupt file at construction ------------------------ +def _write_corrupt(path): + path.write_text(CORRUPT, encoding="utf-8") + return path + + +def test_inbox_store_survives_corrupt_file(tmp_path): + p = _write_corrupt(tmp_path / "inbox.json") + store = InboxStore(p) # must not raise + assert store.list() == [] + assert (tmp_path / "inbox.json.corrupt").exists() + + +def test_inbox_routing_survives_corrupt_file(tmp_path): + p = _write_corrupt(tmp_path / "inbox_routing.json") + routing = InboxRouting(p) + assert routing.route_for("s1") == DEFAULT_INBOX + + +def test_mentions_store_survives_corrupt_file(tmp_path): + p = _write_corrupt(tmp_path / "mentions.json") + assert MentionSessionStore(p).all() == [] + + +def test_subscription_store_survives_corrupt_file(tmp_path): + p = _write_corrupt(tmp_path / "subs.json") + assert SubscriptionStore(p).all() == [] + + +def test_unrouted_store_survives_corrupt_file(tmp_path): + p = _write_corrupt(tmp_path / "unrouted.json") + assert UnroutedStore(p).list() == [] + + +def test_unattended_registry_survives_corrupt_file(tmp_path): + p = _write_corrupt(tmp_path / "unattended.json") + assert UnattendedRegistry(p).sessions() == [] + + +def test_connection_stores_survive_corrupt_file(tmp_path): + persona = PersonaConnectionStore(_write_corrupt(tmp_path / "persona.json")) + session = SessionConnectionStore(_write_corrupt(tmp_path / "session.json")) + assert persona.get("p1") == {} + assert session.get("s1") == {} + + +def test_override_store_survives_corrupt_file(tmp_path): + p = _write_corrupt(tmp_path / "overrides.json") + assert RiskOverrideStore(p).resolve("some_tool") is None + + +# -- round-trip after recovery: a corrupt file heals on the next save ---------- +def test_store_save_after_corrupt_load_round_trips(tmp_path): + p = _write_corrupt(tmp_path / "subs.json") + store = SubscriptionStore(p) # corrupt file quarantined, in-memory empty + store.subscribe("session-1", "slack:C123") + # A fresh store reads back exactly what was saved — the file is valid again. + assert [(s.session_id, s.channel) for s in SubscriptionStore(p).all()] == [ + ("session-1", "slack:C123") + ] + + +def test_override_store_save_round_trips(tmp_path): + p = tmp_path / "overrides.json" + store = RiskOverrideStore(p) + store.set_rule("mcp__notion__*", RiskClass.READ) + assert RiskOverrideStore(p).resolve("mcp__notion__search") == RiskClass.READ