diff --git a/AGENTS.md b/AGENTS.md index c0be4b3dc..6a759531f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -190,6 +190,7 @@ Each domain has exactly **one** write-owning module (or one tightly-scoped famil | Awareness (`awareness/current.json`) | `solstone/think/awareness.py` | | Todos (`facets/*/todos/*.jsonl`) | `solstone/apps/todos/todo.py` + `solstone/apps/todos/call.py` | | Config (`config/journal.json`) | `solstone/think/journal_config.py` | +| Convey config (`config/convey.json`) | `solstone/convey/config.py` + `solstone/think/facets.py` | | Chat config (`config/chat.json`) | `solstone/apps/chat/config.py` | | Vertex credentials (`.config/vertex-credentials.json`) | `solstone/apps/settings/vertex_credentials.py` | | Speaker labels (`chronicle/**/talents/speaker_labels.json`) | `solstone/apps/speakers/attribution.py` | diff --git a/scripts/check_journal_io_access.py b/scripts/check_journal_io_access.py index 2c9309868..bbde78b4d 100644 --- a/scripts/check_journal_io_access.py +++ b/scripts/check_journal_io_access.py @@ -81,6 +81,7 @@ OWNER_FILES: frozenset[str] = frozenset( { "solstone/apps/entities/call.py", "solstone/apps/chat/config.py", + "solstone/convey/config.py", "solstone/apps/speakers/attribution.py", "solstone/apps/speakers/owner.py", "solstone/apps/speakers/routes.py", diff --git a/solstone/apps/curation/routes.py b/solstone/apps/curation/routes.py index 00990859f..12df7a3ff 100644 --- a/solstone/apps/curation/routes.py +++ b/solstone/apps/curation/routes.py @@ -74,7 +74,10 @@ def accept_facet() -> Response | tuple[Response, int]: name_key = str(_required(data, "name_key")) except KeyError: return _missing_field("name_key") - return _result_response(accept_facet_candidate(name_key)) + try: + return _result_response(accept_facet_candidate(name_key)) + except LockTimeout: + return error_response(ENTITY_BUSY, detail="suggestions are busy; try again") @curation_bp.post("/api/facet/dismiss") @@ -84,7 +87,10 @@ def dismiss_facet() -> Response | tuple[Response, int]: name_key = str(_required(data, "name_key")) except KeyError: return _missing_field("name_key") - return _result_response(dismiss_facet_candidate(name_key)) + try: + return _result_response(dismiss_facet_candidate(name_key)) + except LockTimeout: + return error_response(ENTITY_BUSY, detail="suggestions are busy; try again") @curation_bp.post("/api/entity/preview") diff --git a/solstone/apps/settings/maint/003_seed_default_app_navigation.py b/solstone/apps/settings/maint/003_seed_default_app_navigation.py index 1ecf207e8..e6cd9d99e 100644 --- a/solstone/apps/settings/maint/003_seed_default_app_navigation.py +++ b/solstone/apps/settings/maint/003_seed_default_app_navigation.py @@ -7,11 +7,11 @@ from __future__ import annotations import logging import sys +from typing import Any import solstone.convey.state as convey_state from solstone.convey.config import ( - load_convey_config, - save_convey_config, + locked_modify_convey_config, seed_default_app_navigation, ) from solstone.think.utils import get_journal @@ -36,18 +36,17 @@ def main(): convey_state.journal_root = str(journal) - config = load_convey_config() - if not seed_default_app_navigation(config): - print("Default app navigation already present.") - return + def _seed(config: dict[str, Any]) -> dict[str, Any] | None: + return config if seed_default_app_navigation(config) else None try: - saved = save_convey_config(config) + result = locked_modify_convey_config(_seed) except Exception as exc: _fail("default app navigation seed convey-config PERSIST failed", exc) - if not saved: - _fail("default app navigation seed convey-config PERSIST failed") + if result is None: + print("Default app navigation already present.") + return print("Seeded default app navigation.") diff --git a/solstone/convey/config.py b/solstone/convey/config.py index bb18196b7..9f3c73c73 100644 --- a/solstone/convey/config.py +++ b/solstone/convey/config.py @@ -5,20 +5,25 @@ from __future__ import annotations +import json import logging +from collections.abc import Callable from pathlib import Path from typing import Any from flask import Blueprint, request +from solstone.think.journal_io import LockTimeout, atomic_replace, hold_lock + from . import state from .reasons import ( + CONVEY_BUSY, CONVEY_OPERATION_FAILED, INVALID_CONFIG_VALUE, INVALID_JSON_REQUEST, MISSING_REQUIRED_FIELD, ) -from .utils import error_response, load_json, save_json, success_response +from .utils import error_response, load_json, success_response logger = logging.getLogger(__name__) @@ -69,21 +74,32 @@ def reporting_enabled() -> bool: return config.get("reporting", {}).get("enabled", True) -def save_convey_config(config: dict[str, Any]) -> bool: - """Save config/convey.json atomically. - - Args: - config: Configuration dict to save +def _write_convey_config(config: dict[str, Any]) -> None: + """Atomically persist convey config. Caller MUST hold the convey.json lock.""" + atomic_replace( + _get_config_path(), + json.dumps(config, indent=2, ensure_ascii=False) + "\n", + ) - Returns: - True if successful, False otherwise - """ - config_path = _get_config_path() - # Ensure config directory exists - config_path.parent.mkdir(parents=True, exist_ok=True) +def locked_modify_convey_config( + transform: Callable[[dict[str, Any]], dict[str, Any] | None], +) -> dict[str, Any] | None: + """Apply a locked read-modify-write to config/convey.json. - return save_json(config_path, config, indent=2) + Reads the current config under an exclusive lock, applies ``transform``, + and atomically persists the result. If ``transform`` returns ``None`` the + write is skipped (no-op). Returns the persisted config, or ``None`` when + skipped. + """ + path = _get_config_path() + with hold_lock(path): + config = load_convey_config() + new_config = transform(config) + if new_config is None: + return None + _write_convey_config(new_config) + return new_config def seed_default_app_navigation(config: dict[str, Any]) -> bool: @@ -117,24 +133,16 @@ def get_selected_facet() -> str | None: def set_selected_facet(facet: str | None) -> None: - """Update selected facet in config. + """Update selected facet in config (best-effort; never raises).""" - Args: - facet: Facet name to select, or None to clear selection - """ - config = load_convey_config() - - # Ensure facets section exists - if "facets" not in config: - config["facets"] = {} + def _transform(config: dict[str, Any]) -> dict[str, Any]: + config.setdefault("facets", {})["selected"] = facet + return config - # Update selected field - config["facets"]["selected"] = facet - - # Save config (async safe - doesn't block if write fails) - success = save_convey_config(config) - if not success: - logger.warning(f"Failed to save selected facet: {facet}") + try: + locked_modify_convey_config(_transform) + except (LockTimeout, OSError) as exc: + logger.warning("Failed to save selected facet %s: %s", facet, exc) def apply_facet_order(facets: list[dict], config: dict) -> list[dict]: @@ -350,37 +358,23 @@ def update_config() -> tuple[Any, int]: detail=f"Invalid config: {error_msg}", ) - # Merge with existing config (partial updates supported) - current_config = load_convey_config() - - # Deep merge facets section - if "facets" in new_config: - if "facets" not in current_config: - current_config["facets"] = {} - current_config["facets"].update(new_config["facets"]) - - # Deep merge apps section - if "apps" in new_config: - if "apps" not in current_config: - current_config["apps"] = {} - current_config["apps"].update(new_config["apps"]) - - # Deep merge reporting section - if "reporting" in new_config: - if "reporting" not in current_config: - current_config["reporting"] = {} - current_config["reporting"].update(new_config["reporting"]) - - # Save updated config - success = save_convey_config(current_config) - if not success: - return error_response( - CONVEY_OPERATION_FAILED, - detail="Failed to save configuration", - ) + def _transform(config: dict[str, Any]) -> dict[str, Any]: + if "facets" in new_config: + config.setdefault("facets", {}).update(new_config["facets"]) + if "apps" in new_config: + config.setdefault("apps", {}).update(new_config["apps"]) + if "reporting" in new_config: + config.setdefault("reporting", {}).update(new_config["reporting"]) + return config - return success_response({"config": current_config}) + persisted = locked_modify_convey_config(_transform) + return success_response({"config": persisted}) + except LockTimeout: + return error_response( + CONVEY_BUSY, + detail="Interface settings are busy; try again", + ) except Exception as e: logger.error(f"Failed to update config: {e}", exc_info=True) return error_response( @@ -418,22 +412,19 @@ def update_facet_order() -> tuple[Any, int]: detail="'order' must contain only strings", ) - # Load config and update facets.order - config = load_convey_config() - if "facets" not in config: - config["facets"] = {} - config["facets"]["order"] = order + def _transform(config: dict[str, Any]) -> dict[str, Any]: + config.setdefault("facets", {})["order"] = order + return config - # Save - success = save_convey_config(config) - if not success: - return error_response( - CONVEY_OPERATION_FAILED, - detail="Failed to save facet order", - ) + locked_modify_convey_config(_transform) return success_response({"order": order}) + except LockTimeout: + return error_response( + CONVEY_BUSY, + detail="Interface settings are busy; try again", + ) except Exception as e: logger.error(f"Failed to update facet order: {e}", exc_info=True) return error_response( @@ -471,22 +462,19 @@ def update_app_order() -> tuple[Any, int]: detail="'order' must contain only strings", ) - # Load config and update apps.order - config = load_convey_config() - if "apps" not in config: - config["apps"] = {} - config["apps"]["order"] = order + def _transform(config: dict[str, Any]) -> dict[str, Any]: + config.setdefault("apps", {})["order"] = order + return config - # Save - success = save_convey_config(config) - if not success: - return error_response( - CONVEY_OPERATION_FAILED, - detail="Failed to save app order", - ) + locked_modify_convey_config(_transform) return success_response({"order": order}) + except LockTimeout: + return error_response( + CONVEY_BUSY, + detail="Interface settings are busy; try again", + ) except Exception as e: logger.error(f"Failed to update app order: {e}", exc_info=True) return error_response( @@ -524,30 +512,25 @@ def toggle_app_star() -> tuple[Any, int]: detail="'starred' must be a boolean", ) - # Load config and update apps.starred - config = load_convey_config() - if "apps" not in config: - config["apps"] = {} - - starred_apps = set(config["apps"].get("starred", [])) + def _transform(config: dict[str, Any]) -> dict[str, Any]: + apps_config = config.setdefault("apps", {}) + starred_apps = set(apps_config.get("starred", [])) + if starred: + starred_apps.add(app_name) + else: + starred_apps.discard(app_name) + apps_config["starred"] = sorted(starred_apps) + return config - if starred: - starred_apps.add(app_name) - else: - starred_apps.discard(app_name) - - config["apps"]["starred"] = sorted(starred_apps) - - # Save - success = save_convey_config(config) - if not success: - return error_response( - CONVEY_OPERATION_FAILED, - detail="Failed to save app starred status", - ) + locked_modify_convey_config(_transform) return success_response({"app": app_name, "starred": starred}) + except LockTimeout: + return error_response( + CONVEY_BUSY, + detail="Interface settings are busy; try again", + ) except Exception as e: logger.error(f"Failed to toggle app star: {e}", exc_info=True) return error_response( diff --git a/solstone/convey/reasons.py b/solstone/convey/reasons.py index a3bf8ea19..1febee335 100644 --- a/solstone/convey/reasons.py +++ b/solstone/convey/reasons.py @@ -110,6 +110,11 @@ CONVEY_OPERATION_FAILED = Reason( "I couldn't update the interface settings.", 500, ) +CONVEY_BUSY = Reason( + "convey_busy", + "I couldn't update the interface settings right now because they were busy. Try again in a moment.", + 503, +) NETWORK_SECURITY_REQUIRES_PASSWORD = Reason( "network_security_requires_password", "I couldn't change network access until a password is set.", diff --git a/solstone/convey/root.py b/solstone/convey/root.py index 0aaf57654..ac42b7473 100644 --- a/solstone/convey/root.py +++ b/solstone/convey/root.py @@ -41,8 +41,7 @@ from solstone.think.utils import ( from . import bridge as convey_bridge from .config import ( - load_convey_config, - save_convey_config, + locked_modify_convey_config, seed_default_app_navigation, ) from .copy import LOGIN_NO_PASSWORD_CONFIGURED @@ -369,8 +368,12 @@ def init_finalize() -> Any: write_journal_config(config) - config = load_convey_config() - if seed_default_app_navigation(config) and not save_convey_config(config): + def _seed(config: dict[str, Any]) -> dict[str, Any] | None: + return config if seed_default_app_navigation(config) else None + + try: + locked_modify_convey_config(_seed) + except Exception: logger.error("default app navigation seed convey-config PERSIST failed") session["logged_in"] = True diff --git a/solstone/think/facet_review_candidates.py b/solstone/think/facet_review_candidates.py index 74fb8c5a2..24f0a58de 100644 --- a/solstone/think/facet_review_candidates.py +++ b/solstone/think/facet_review_candidates.py @@ -8,14 +8,13 @@ Sole write-owner of: from __future__ import annotations -import fcntl import json import logging from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable -from solstone.think.entities.core import atomic_write +from solstone.think.journal_io import atomic_replace, hold_lock from solstone.think.utils import get_journal logger = logging.getLogger(__name__) @@ -33,11 +32,6 @@ def facet_review_candidates_path() -> Path: return facet_review_candidates_dir() / "review-candidates.jsonl" -def facet_review_candidates_lock_path() -> Path: - """Return the sibling lock path for review-candidates.jsonl.""" - return facet_review_candidates_dir() / ".review-candidates.lock" - - def _load_jsonl_rows(path: Path) -> list[dict[str, Any]]: """Load JSONL rows from *path*, skipping blanks and malformed lines.""" if not path.exists(): @@ -77,10 +71,12 @@ def load_candidates() -> list[dict[str, Any]]: def _save_jsonl_rows(path: Path, rows: list[dict[str, Any]]) -> None: """Write *rows* to *path* as JSONL using an atomic replace.""" - content = "" - if rows: - content = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n" - atomic_write(path, content) + content = ( + "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n" + if rows + else "" + ) + atomic_replace(path, content) def save_candidates(rows: list[dict[str, Any]]) -> None: @@ -107,18 +103,11 @@ def locked_modify_candidates( fn: Callable[[list[dict[str, Any]]], list[dict[str, Any]]], ) -> list[dict[str, Any]]: """Apply a locked read-modify-write cycle to review-candidates.jsonl.""" - facet_review_candidates_dir() - lock_path = facet_review_candidates_lock_path() - # Lock file contents are irrelevant; opening with "w" matches the existing pattern. - with open(lock_path, "w", encoding="utf-8") as lock_file: - fcntl.flock(lock_file, fcntl.LOCK_EX) - try: - rows = load_candidates() - new_rows = fn(rows) - save_candidates(new_rows) - return new_rows - finally: - fcntl.flock(lock_file, fcntl.LOCK_UN) + with hold_lock(facet_review_candidates_path()): + rows = load_candidates() + new_rows = fn(rows) + save_candidates(new_rows) + return new_rows def utc_now_iso() -> str: diff --git a/solstone/think/facets.py b/solstone/think/facets.py index df8ded463..ff29a0210 100644 --- a/solstone/think/facets.py +++ b/solstone/think/facets.py @@ -13,6 +13,7 @@ from pathlib import Path from typing import Any, Optional from solstone.think.entities import get_identity_names +from solstone.think.journal_io import append_text, atomic_replace, hold_lock from solstone.think.utils import day_dirs, day_path, get_journal, iter_segments @@ -185,9 +186,6 @@ def _write_action_log( else: log_path = Path(journal) / "config" / "actions" / f"{day}.jsonl" - # Ensure parent directory exists - log_path.parent.mkdir(parents=True, exist_ok=True) - # Create log entry entry = { "timestamp": datetime.now(timezone.utc).isoformat(), @@ -206,8 +204,7 @@ def _write_action_log( entry["use_id"] = use_id # Append to log file - with open(log_path, "a", encoding="utf-8") as f: - f.write(json.dumps(entry, ensure_ascii=False) + "\n") + append_text(log_path, json.dumps(entry, ensure_ascii=False)) def log_call_action( @@ -244,20 +241,7 @@ def log_call_action( def _write_facet_json(path: Path, data: dict[str, Any]) -> None: """Write facet metadata atomically.""" - import tempfile - - temp_fd, temp_path = tempfile.mkstemp(dir=path.parent, suffix=".json", text=True) - try: - with os.fdopen(temp_fd, "w", encoding="utf-8") as f: - json.dump(data, f, indent=2, ensure_ascii=False) - f.write("\n") - os.replace(temp_path, path) - except Exception: - try: - os.unlink(temp_path) - except Exception: - pass - raise + atomic_replace(path, json.dumps(data, indent=2, ensure_ascii=False) + "\n") def ensure_facet(slug: str) -> bool: @@ -757,23 +741,7 @@ def set_facet_muted(facet: str, muted: bool) -> None: facet_data.pop("muted", None) # Write back atomically - import tempfile - - temp_fd, temp_path = tempfile.mkstemp( - dir=facet_json_path.parent, suffix=".json", text=True - ) - try: - with os.fdopen(temp_fd, "w", encoding="utf-8") as f: - json.dump(facet_data, f, indent=2, ensure_ascii=False) - f.write("\n") - os.replace(temp_path, facet_json_path) - except Exception: - # Clean up temp file on error - try: - os.unlink(temp_path) - except Exception: - pass - raise + _write_facet_json(facet_json_path, facet_data) # Log the change action = "facet_mute" if muted else "facet_unmute" @@ -886,22 +854,7 @@ def update_facet(name: str, **kwargs: Any) -> dict[str, Any]: changed_fields[field] = {"old": old_value, "new": new_value} facet_data[field] = new_value - import tempfile - - temp_fd, temp_path = tempfile.mkstemp( - dir=facet_json_path.parent, suffix=".json", text=True - ) - try: - with os.fdopen(temp_fd, "w", encoding="utf-8") as f: - json.dump(facet_data, f, indent=2, ensure_ascii=False) - f.write("\n") - os.replace(temp_path, facet_json_path) - except Exception: - try: - os.unlink(temp_path) - except Exception: - pass - raise + _write_facet_json(facet_json_path, facet_data) if changed_fields: log_call_action( @@ -930,29 +883,31 @@ def delete_facet(name: str, *, consent: bool = False) -> None: convey_config_path = Path(get_journal()) / "config" / "convey.json" if convey_config_path.exists(): - try: - with open(convey_config_path, "r", encoding="utf-8") as f: - config = json.load(f) - - changed = False - facets_config = config.get("facets", {}) - - if facets_config.get("selected") == name: - facets_config["selected"] = "" - changed = True - - order = facets_config.get("order", []) - if name in order: - facets_config["order"] = [item for item in order if item != name] - changed = True - - if changed: - config["facets"] = facets_config - with open(convey_config_path, "w", encoding="utf-8") as f: - json.dump(config, f, indent=2, ensure_ascii=False) - f.write("\n") - except (json.JSONDecodeError, OSError): - pass + with hold_lock(convey_config_path): + try: + with open(convey_config_path, "r", encoding="utf-8") as f: + config = json.load(f) + + changed = False + facets_config = config.get("facets", {}) + + if facets_config.get("selected") == name: + facets_config["selected"] = "" + changed = True + + order = facets_config.get("order", []) + if name in order: + facets_config["order"] = [item for item in order if item != name] + changed = True + + if changed: + config["facets"] = facets_config + atomic_replace( + convey_config_path, + json.dumps(config, indent=2, ensure_ascii=False) + "\n", + ) + except (json.JSONDecodeError, OSError): + pass log_params: dict = {"name": name} if consent: @@ -1294,34 +1249,36 @@ def rename_facet(old_name: str, new_name: str) -> None: # Step 2: Update config/convey.json convey_config_path = Path(journal) / "config" / "convey.json" if convey_config_path.exists(): - try: - with open(convey_config_path, "r", encoding="utf-8") as f: - config = json.load(f) - - changed = False - facets_config = config.get("facets", {}) - - if facets_config.get("selected") == old_name: - facets_config["selected"] = new_name - changed = True - - order = facets_config.get("order", []) - if old_name in order: - facets_config["order"] = [ - new_name if name == old_name else name for name in order - ] - changed = True - - if changed: - config["facets"] = facets_config - with open(convey_config_path, "w", encoding="utf-8") as f: - json.dump(config, f, indent=2, ensure_ascii=False) - f.write("\n") - print("Updated config/convey.json") - else: - print("No changes needed in config/convey.json") - except (json.JSONDecodeError, OSError) as exc: - logging.warning("Failed to update convey config: %s", exc) + with hold_lock(convey_config_path): + try: + with open(convey_config_path, "r", encoding="utf-8") as f: + config = json.load(f) + + changed = False + facets_config = config.get("facets", {}) + + if facets_config.get("selected") == old_name: + facets_config["selected"] = new_name + changed = True + + order = facets_config.get("order", []) + if old_name in order: + facets_config["order"] = [ + new_name if name == old_name else name for name in order + ] + changed = True + + if changed: + config["facets"] = facets_config + atomic_replace( + convey_config_path, + json.dumps(config, indent=2, ensure_ascii=False) + "\n", + ) + print("Updated config/convey.json") + else: + print("No changes needed in config/convey.json") + except (json.JSONDecodeError, OSError) as exc: + logging.warning("Failed to update convey config: %s", exc) # Step 3: Advise index rebuild print( diff --git a/tests/test_convey_config.py b/tests/test_convey_config.py index 4fdeab638..aff2cd44c 100644 --- a/tests/test_convey_config.py +++ b/tests/test_convey_config.py @@ -145,7 +145,11 @@ def test_init_finalize_logs_convey_seed_persist_failure( from solstone.convey import root as root_module (journal_copy / "config" / "convey.json").unlink() - monkeypatch.setattr(root_module, "save_convey_config", lambda _config: False) + + def _fail_seed(_transform): + raise OSError("simulated persist failure") + + monkeypatch.setattr(root_module, "locked_modify_convey_config", _fail_seed) caplog.set_level(logging.ERROR, logger="solstone.convey.root") app = create_app(str(journal_copy)) app.config["TESTING"] = True diff --git a/tests/test_convey_config_locking.py b/tests/test_convey_config_locking.py new file mode 100644 index 000000000..decaeabfe --- /dev/null +++ b/tests/test_convey_config_locking.py @@ -0,0 +1,186 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import json +import multiprocessing +import os +import time +import traceback +from pathlib import Path +from queue import Empty +from typing import Any + +import pytest + + +def _write_convey(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(payload, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + + +def _seed_cross_writer_journal(journal: Path) -> Path: + config_path = journal / "config" / "convey.json" + _write_convey( + config_path, + { + "facets": {"order": ["a", "b"], "selected": "a"}, + "apps": {"order": ["x"]}, + }, + ) + facet_dir = journal / "facets" / "b" + facet_dir.mkdir(parents=True, exist_ok=True) + (facet_dir / "facet.json").write_text( + json.dumps( + { + "title": "B", + "description": "", + "color": "#667eea", + "emoji": "📦", + }, + indent=2, + ensure_ascii=False, + ) + + "\n", + encoding="utf-8", + ) + return config_path + + +def _delete_facet_worker( + journal_path: str, + barrier: Any, + errors: Any, + delay: float, +) -> None: + os.environ["SOLSTONE_JOURNAL"] = journal_path + try: + barrier.wait(timeout=5) + if delay: + time.sleep(delay) + + from solstone.think.facets import delete_facet + + delete_facet("b") + except BaseException: + errors.put(traceback.format_exc()) + raise + + +def _convey_config_worker( + journal_path: str, + barrier: Any, + errors: Any, + delay: float, +) -> None: + os.environ["SOLSTONE_JOURNAL"] = journal_path + try: + barrier.wait(timeout=5) + if delay: + time.sleep(delay) + + import solstone.convey.state as convey_state + from solstone.convey.config import locked_modify_convey_config + + convey_state.journal_root = journal_path + + def _transform(config: dict[str, Any]) -> dict[str, Any]: + config.setdefault("apps", {})["order"] = ["x", "y"] + return config + + locked_modify_convey_config(_transform) + except BaseException: + errors.put(traceback.format_exc()) + raise + + +def _drain_errors(errors: Any) -> list[str]: + found = [] + while True: + try: + found.append(errors.get_nowait()) + except Empty: + return found + + +def _join_processes(processes: list[Any], errors: Any) -> None: + for process in processes: + process.start() + for process in processes: + process.join(timeout=10) + for process in processes: + if process.is_alive(): + process.terminate() + process.join(timeout=2) + + error_text = "\n".join(_drain_errors(errors)) + assert all(not process.is_alive() for process in processes), error_text + assert all(process.exitcode == 0 for process in processes), error_text + + +def test_convey_config_atomic_failure_preserves_existing_bytes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + journal = tmp_path / "journal" + config_path = journal / "config" / "convey.json" + original = { + "facets": {"order": ["a"], "selected": "a"}, + "apps": {"order": ["x"]}, + } + _write_convey(config_path, original) + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) + + import solstone.convey.state as convey_state + from solstone.convey.config import locked_modify_convey_config + + convey_state.journal_root = str(journal) + original_bytes = config_path.read_bytes() + + def fail_replace(_src: str, _dst: str) -> None: + raise OSError("simulated replace failure") + + monkeypatch.setattr("solstone.think.journal_io.atomic.os.replace", fail_replace) + + with pytest.raises(OSError): + locked_modify_convey_config(lambda config: {**config, "apps": {"order": ["z"]}}) + + assert config_path.read_bytes() == original_bytes + assert list(config_path.parent.glob(".tmp_*")) == [] + + +@pytest.mark.parametrize( + ("delete_delay", "convey_delay"), + [(0.0, 0.2), (0.2, 0.0)], +) +def test_convey_config_lock_shared_with_think_facet_delete( + tmp_path: Path, + delete_delay: float, + convey_delay: float, +) -> None: + ctx = multiprocessing.get_context("spawn") + journal = tmp_path / f"journal-{delete_delay}-{convey_delay}" + config_path = _seed_cross_writer_journal(journal) + barrier = ctx.Barrier(2) + errors = ctx.Queue() + processes = [ + ctx.Process( + target=_delete_facet_worker, + args=(str(journal), barrier, errors, delete_delay), + ), + ctx.Process( + target=_convey_config_worker, + args=(str(journal), barrier, errors, convey_delay), + ), + ] + + _join_processes(processes, errors) + + data = json.loads(config_path.read_text(encoding="utf-8")) + assert data["facets"]["order"] == ["a"] + assert data["apps"]["order"] == ["x", "y"] + assert not (journal / "facets" / "b").exists() diff --git a/tests/test_facet_review_candidates.py b/tests/test_facet_review_candidates.py index 58c5a3b2f..5ca218a12 100644 --- a/tests/test_facet_review_candidates.py +++ b/tests/test_facet_review_candidates.py @@ -15,7 +15,6 @@ from solstone.think.facet_review_candidates import ( candidate_key, dismiss_candidate, facet_review_candidates_dir, - facet_review_candidates_lock_path, facet_review_candidates_path, find_candidate, load_candidates, @@ -46,10 +45,6 @@ def test_path_helpers_return_expected_names(candidate_journal): facet_review_candidates_path() == candidate_journal / "facets" / "review-candidates.jsonl" ) - assert ( - facet_review_candidates_lock_path() - == candidate_journal / "facets" / ".review-candidates.lock" - ) def test_load_candidates_missing_file_returns_empty(candidate_journal): diff --git a/tests/test_facet_review_candidates_locking.py b/tests/test_facet_review_candidates_locking.py new file mode 100644 index 000000000..b351b04b4 --- /dev/null +++ b/tests/test_facet_review_candidates_locking.py @@ -0,0 +1,125 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import json +import multiprocessing +import os +import traceback +from pathlib import Path +from queue import Empty +from typing import Any + +import pytest + +from solstone.think.facet_review_candidates import ( + facet_review_candidates_path, + load_candidates, + save_candidates, +) + + +def _record_candidate_worker( + journal_path: str, + barrier: Any, + errors: Any, + index: int, +) -> None: + os.environ["SOLSTONE_JOURNAL"] = journal_path + try: + barrier.wait(timeout=5) + + from solstone.think.facet_review_candidates import record_facet_candidate + + record_facet_candidate( + name=f"Candidate {index}", + name_key=f"candidate {index}", + count=index + 1, + window_days=14, + samples=[ + { + "day": "20260602", + "stream": "archon", + "segment": f"09000{index}_300", + } + ], + day="20260602", + ) + except BaseException: + errors.put(traceback.format_exc()) + raise + + +def _drain_errors(errors: Any) -> list[str]: + found = [] + while True: + try: + found.append(errors.get_nowait()) + except Empty: + return found + + +def _join_processes(processes: list[Any], errors: Any) -> None: + for process in processes: + process.start() + for process in processes: + process.join(timeout=10) + for process in processes: + if process.is_alive(): + process.terminate() + process.join(timeout=2) + + error_text = "\n".join(_drain_errors(errors)) + assert all(not process.is_alive() for process in processes), error_text + assert all(process.exitcode == 0 for process in processes), error_text + + +def test_facet_review_candidate_atomic_failure_preserves_existing_bytes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + path = facet_review_candidates_path() + seed = {"name": "Café", "name_key": "café", "status": "open"} + original = json.dumps(seed, ensure_ascii=False) + "\n" + path.write_text(original, encoding="utf-8") + + def fail_replace(_src: str, _dst: str) -> None: + raise OSError("simulated replace failure") + + monkeypatch.setattr("solstone.think.journal_io.atomic.os.replace", fail_replace) + + with pytest.raises(OSError): + save_candidates([{"name": "Changed", "name_key": "changed"}]) + + assert path.read_text(encoding="utf-8") == original + assert list(path.parent.glob(".tmp_*")) == [] + + +def test_facet_review_candidate_locked_modify_survives_multiprocess_writers( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + ctx = multiprocessing.get_context("spawn") + journal = tmp_path / "journal" + barrier = ctx.Barrier(4) + errors = ctx.Queue() + processes = [ + ctx.Process( + target=_record_candidate_worker, + args=(str(journal), barrier, errors, index), + ) + for index in range(4) + ] + + _join_processes(processes, errors) + + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) + rows = load_candidates() + assert sorted(row["name_key"] for row in rows) == [ + "candidate 0", + "candidate 1", + "candidate 2", + "candidate 3", + ] diff --git a/tests/test_facets_durability.py b/tests/test_facets_durability.py new file mode 100644 index 000000000..1770bd65d --- /dev/null +++ b/tests/test_facets_durability.py @@ -0,0 +1,31 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from solstone.think.facets import create_facet, update_facet + + +def test_facet_json_atomic_failure_preserves_existing_bytes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + slug = create_facet("Home Reno") + facet_json = tmp_path / "facets" / slug / "facet.json" + original = facet_json.read_bytes() + + def fail_replace(_src: str, _dst: str) -> None: + raise OSError("simulated replace failure") + + monkeypatch.setattr("solstone.think.journal_io.atomic.os.replace", fail_replace) + + with pytest.raises(OSError): + update_facet(slug, title="New Home Reno") + + assert facet_json.read_bytes() == original + assert list(facet_json.parent.glob(".tmp_*")) == []