From deb95a233ea05b7008858860163e8814d7e294a0 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Feb 2026 09:32:38 +0000 Subject: [PATCH 1/5] Add analysis of database locking in multi-agent setups Detailed analysis of how Hoard prevents write conflicts when multiple agents share a single SQLite database, covering WAL mode, the WriteCoordinator single-writer thread, busy timeout, optimistic concurrency checks, and async conflict detection. Identifies potential gaps including lack of concurrent write tests. https://claude.ai/code/session_012Zb8mCrSHJtkuCu9mqZCZ2 --- ANALYSIS_DB_AGENT_LOCKING.md | 249 +++++++++++++++++++++++++++++++++++ 1 file changed, 249 insertions(+) create mode 100644 ANALYSIS_DB_AGENT_LOCKING.md diff --git a/ANALYSIS_DB_AGENT_LOCKING.md b/ANALYSIS_DB_AGENT_LOCKING.md new file mode 100644 index 0000000..2194e1b --- /dev/null +++ b/ANALYSIS_DB_AGENT_LOCKING.md @@ -0,0 +1,249 @@ +# Analysis: Database Locking in Multi-Agent Setups + +## Question + +> Sharing a DB across agents keeps state consistent without API calls. Does it lock during writes to avoid conflicts in multi-agent setups? + +## Short Answer + +Yes. Hoard uses a **five-layer defense** against write conflicts on a shared SQLite database: + +1. **WAL mode** (database-level) -- readers never block writers +2. **WriteCoordinator** (process-level) -- serializes all writes through a single thread +3. **Busy timeout** (connection-level) -- retries on `SQLITE_BUSY` for up to 5 seconds +4. **Optimistic concurrency** (row-level) -- `rowcount` checks detect lost updates +5. **Async conflict detection** (application-level) -- background jobs find semantic conflicts + +These layers work together so that multiple agents sharing one Hoard server will not corrupt data, lose writes, or deadlock. + +--- + +## Layer 1: WAL Journal Mode + +**File:** `hoard/core/db/connection.py:21` + +```python +conn.execute("PRAGMA journal_mode = WAL;") +conn.execute("PRAGMA synchronous = NORMAL;") +``` + +SQLite's Write-Ahead Logging mode allows **concurrent readers and a single writer** without blocking each other. In the default rollback journal mode, a writer would block all readers (and vice versa). WAL eliminates this contention: agents performing searches or reads will never be blocked by an ongoing write, and writes will never wait for readers to finish. + +**Implication for multi-agent:** Agents calling read-only tools (`search`, `get`, `context`) operate without any lock contention, even while another agent is writing memories. + +--- + +## Layer 2: WriteCoordinator (Single-Writer Serialization) + +**File:** `hoard/core/db/writer.py` + +This is the **primary locking mechanism**. All database writes are funneled through a single background thread via a `queue.Queue`: + +```python +class WriteCoordinator: + def __init__(self, *, db_path, busy_timeout_ms): + self._queue: queue.Queue[_WriteTask | None] = queue.Queue() + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + + def submit(self, fn, *args, **kwargs): + # If already on the writer thread, execute directly + if threading.get_ident() == self._thread_id: + return fn(self._conn, *args, **kwargs) + # Otherwise, enqueue and block until completion + task = _WriteTask(fn=fn, args=args, kwargs=kwargs, event=threading.Event()) + self._queue.put(task) + task.event.wait() # caller blocks here + if task.error: + raise task.error + return task.result +``` + +The worker loop processes one task at a time, auto-committing on success and rolling back on error: + +```python +def _run(self): + self._conn = connect(self._db_path, busy_timeout_ms=self._busy_timeout_ms) + while True: + task = self._queue.get() + try: + task.result = task.fn(self._conn, *task.args, **task.kwargs) + if self._conn.in_transaction: + self._conn.commit() + except Exception as exc: + if self._conn.in_transaction: + self._conn.rollback() + task.error = exc + finally: + task.event.set() +``` + +**How the MCP server uses it** (`hoard/core/mcp/server.py:287-292`): + +```python +def _dispatch_tool(self, tool, payload, conn, token): + if is_write_tool(tool): + return self.server.writer.submit( + lambda writer_conn: dispatch_tool(tool, payload, writer_conn, self.server.config, token) + ) + return dispatch_tool(tool, payload, conn, self.server.config, token) +``` + +Write tools go through `writer.submit()` (serialized). Read tools execute directly on a per-request connection (no contention). + +**Implication for multi-agent:** Even though `ThreadingHTTPServer` spawns a new thread per request, every write from every agent is serialized into a single queue. Two agents writing memories simultaneously will never interleave their transactions. The second write blocks at `task.event.wait()` until the first completes. + +--- + +## Layer 3: Busy Timeout + +**File:** `hoard/core/config.py:92` + +```python +"database": { + "busy_timeout_ms": 5000, +} +``` + +Applied as a pragma in `connection.py:23-24`: + +```python +if busy_timeout_ms: + conn.execute(f"PRAGMA busy_timeout = {int(busy_timeout_ms)};") +``` + +If a connection encounters `SQLITE_BUSY` (e.g., during migrations or if an external process holds the write lock), SQLite will internally retry for up to 5000ms before raising an error. This provides a safety net beyond the WriteCoordinator for any edge cases where multiple connections might attempt writes (e.g., migrations, the sync lock flow). + +--- + +## Layer 4: Optimistic Concurrency (Row-Level Conflict Detection) + +Multiple operations use `UPDATE ... WHERE status = ? ... ` + `cursor.rowcount` as an **optimistic lock**: + +### Task Claiming (`hoard/core/orchestrator/tasks.py:121-132`) + +```python +cursor = conn.execute( + """ + UPDATE tasks + SET status = 'claimed', assigned_agent_id = ?, ... + WHERE id = ? AND status = 'queued' + AND (assigned_agent_id IS NULL OR assigned_agent_id = ?) + """, ... +) +if cursor.rowcount == 0: + return None # Another agent claimed it first +``` + +Two agents trying to claim the same task cannot both succeed -- the `WHERE status = 'queued'` guard ensures only one update takes effect. The loser gets `rowcount == 0` and a `None` return. + +### Worker Lease (`hoard/core/worker.py:71-108`) + +```python +INSERT INTO worker_lease (id, ...) VALUES (1, ...) +ON CONFLICT (id) DO UPDATE SET ... +WHERE worker_lease.expires_at < ? +``` + +Only one worker process can hold the background job lease. The `WHERE expires_at < now` condition ensures a stale lease can be reclaimed, but an active one cannot be stolen. + +### Memory Updates (`hoard/core/memory/v2/store.py`) + +Similar `rowcount > 0` checks guard memory updates and deletions. + +--- + +## Layer 5: Application-Level Conflict Detection + +Hoard recognizes that even with serialized writes, **semantic conflicts** can occur (two agents writing different values to the same memory slot). These are handled asynchronously: + +### Background Jobs (`hoard/core/worker.py:350-439`) + +After every memory write, three background jobs are enqueued: +1. `embed_memory` -- generate embeddings +2. `detect_duplicates` -- find semantically similar memories (threshold: 0.85) +3. `detect_conflicts` -- find memories in the same slot/scope from different agents + +### Conflict & Duplicate Tables (migration 002) + +``` +memory_conflicts -- tracks slot collisions between agents +conflict_members -- maps conflicting memories with NLI labels +memory_duplicates -- tracks semantically similar memories +duplicate_members -- maps duplicate pairs +``` + +These are surfaced via MCP tools (`conflicts_list`, `conflict_resolve`, `duplicates_list`, `duplicate_resolve`) for human or agent resolution. + +--- + +## Layer 6 (Bonus): File-Based Sync Lock + +**File:** `hoard/core/sync/service.py:122-136` + +```python +def _acquire_lock(path): + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) # atomic + # Write PID + timestamp for staleness detection +``` + +The data connector sync process uses a file-based lock (`~/.hoard/sync.lock`) with `O_CREAT | O_EXCL` for atomic creation. This prevents two sync operations from running concurrently. Staleness detection (PID alive check) handles crash recovery. + +--- + +## Summary: What Happens When Two Agents Write Simultaneously + +``` +Agent A writes memory ─┐ + ├─▶ ThreadingHTTPServer (thread per request) +Agent B writes memory ─┘ + │ + ▼ + is_write_tool(tool) → True + │ + ▼ + WriteCoordinator.submit() + │ + ┌─────────────┴──────────────┐ + │ Queue (FIFO ordering) │ + │ Agent A's write → slot 1 │ + │ Agent B's write → slot 2 │ + └─────────────┬──────────────┘ + │ + ▼ + Single Writer Thread + (one connection, one txn at a time) + │ + ┌─────────┼─────────┐ + │ Execute │ Commit │ + │ A's fn │ or │ + │ │ Rollback│ + └─────────┼─────────┘ + │ + ┌─────────┼─────────┐ + │ Execute │ Commit │ + │ B's fn │ or │ + │ │ Rollback│ + └─────────┼─────────┘ + │ + ▼ + Background jobs detect conflicts/duplicates +``` + +1. Both requests arrive on separate HTTP threads +2. Both call `writer.submit()`, enqueueing their operations +3. The writer thread processes them **sequentially** (FIFO) +4. Each gets its own transaction with auto-commit/rollback +5. Neither agent sees a `SQLITE_BUSY` error or partial data +6. Post-write background jobs check for semantic conflicts + +--- + +## Potential Gaps + +| Gap | Severity | Notes | +|-----|----------|-------| +| **Single-process only** | Medium | The WriteCoordinator serializes writes within one server process. If two separate `hoard serve` processes point at the same DB file, they would rely solely on WAL + busy_timeout (no queue serialization). This is not a supported deployment. | +| **No concurrent write tests** | Low | `test_http_mcp_concurrency.py` only tests concurrent reads (5 threads x 3 search calls). There are no tests exercising concurrent writes from multiple agents. | +| **Task claim is not atomic SELECT+UPDATE** | Low | `_claim_job` in `worker.py:140-161` does a SELECT then UPDATE in two statements. This is safe because it runs inside the WriteCoordinator (single thread), but would be a race condition if used with direct connections. | +| **Conflict resolution is manual** | Low | Detected conflicts require explicit resolution via MCP tools. There is no auto-merge strategy. | From 730ae501b301e2259274725f534f9727e1d45e68 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Feb 2026 09:58:50 +0000 Subject: [PATCH 2/5] Add cross-process write locking for multi-agent safety Introduces flock-based advisory locks to guarantee that database writes are serialized across all processes, not just within a single server. Changes: - New hoard/core/db/lock.py with DatabaseWriteLock (per-write) and ServerSingletonLock (prevents two hoard serve on same DB) - WriteCoordinator now acquires DatabaseWriteLock around each write task - CLI write commands (memory put, memory prune, db migrate) use write_locked() context manager instead of bare connect() - run_sync_with_lock() holds the write lock during sync operations - MCP server acquires ServerSingletonLock on startup, fails fast if another server is already running Includes 9 new tests covering lock semantics, mutual exclusion, WriteCoordinator integration, and concurrent HTTP writes from multiple agents. All 142 existing tests continue to pass. https://claude.ai/code/session_012Zb8mCrSHJtkuCu9mqZCZ2 --- hoard/cli/main.py | 101 ++++++------- hoard/core/db/connection.py | 25 +++- hoard/core/db/lock.py | 128 ++++++++++++++++ hoard/core/db/writer.py | 31 ++-- hoard/core/mcp/server.py | 18 +++ hoard/core/sync/service.py | 9 +- tests/test_write_lock.py | 291 ++++++++++++++++++++++++++++++++++++ 7 files changed, 531 insertions(+), 72 deletions(-) create mode 100644 hoard/core/db/lock.py create mode 100644 tests/test_write_lock.py diff --git a/hoard/cli/main.py b/hoard/cli/main.py index ad6119d..aeb9151 100644 --- a/hoard/cli/main.py +++ b/hoard/cli/main.py @@ -30,7 +30,7 @@ resolve_project_root, ) from hoard.core.config import default_data_path, ensure_config_file, load_config, resolve_paths, save_config -from hoard.core.db.connection import connect, initialize_db +from hoard.core.db.connection import connect, initialize_db, write_locked from hoard.core.embeddings.model import EmbeddingError, EmbeddingModel from hoard.core.mcp.server import run_server from hoard.core.mcp.stdio import run_stdio @@ -444,41 +444,37 @@ def db_migrate_command(target_version: int | None, config_path: Path | None) -> config = load_config(config_path) paths = resolve_paths(config, config_path) - conn = connect(paths.db_path) - - current = get_current_version(conn) - migrations = get_migrations() - latest = max(migrations.keys()) if migrations else 0 - target = target_version if target_version is not None else latest - console.print(f"📦 Applying schema migrations (v{current} → v{target})...") + with write_locked(paths.db_path) as conn: + current = get_current_version(conn) + migrations = get_migrations() + latest = max(migrations.keys()) if migrations else 0 + target = target_version if target_version is not None else latest + console.print(f"📦 Applying schema migrations (v{current} → v{target})...") - try: - applied = migrate(conn, target_version=target_version, app_version=__version__) - except MigrationError as exc: - conn.close() - raise click.ClickException(str(exc)) from exc + try: + applied = migrate(conn, target_version=target_version, app_version=__version__) + except MigrationError as exc: + raise click.ClickException(str(exc)) from exc - if not applied: - console.print("No pending migrations.") - conn.close() - return + if not applied: + console.print("No pending migrations.") + return - placeholders = ",".join("?" for _ in applied) - rows = conn.execute( - f""" - SELECT version, name, duration_ms - FROM schema_migrations - WHERE version IN ({placeholders}) - ORDER BY version - """, - applied, - ).fetchall() - for row in rows: - _, name, duration_ms = row[0], row[1], row[2] - console.print(f" {name} ... done ({duration_ms}ms)") + placeholders = ",".join("?" for _ in applied) + rows = conn.execute( + f""" + SELECT version, name, duration_ms + FROM schema_migrations + WHERE version IN ({placeholders}) + ORDER BY version + """, + applied, + ).fetchall() + for row in rows: + _, name, duration_ms = row[0], row[1], row[2] + console.print(f" {name} ... done ({duration_ms}ms)") - console.print(f"Migrated from version {current} to {applied[-1]}") - conn.close() + console.print(f"Migrated from version {current} to {applied[-1]}") @db_group.command("history") @@ -573,23 +569,22 @@ def memory_put_command( ) -> None: config = load_config(config_path) paths = resolve_paths(config, config_path) - conn = connect(paths.db_path) - initialize_db(conn) - - tag_list = [tag.strip() for tag in tags.split(",") if tag.strip()] if tags else [] - metadata_obj = json.loads(metadata) if metadata else None - entry = memory_put( - conn, - key=key, - content=content, - tags=tag_list, - metadata=metadata_obj, - ttl_days=ttl_days, - expires_at=expires_at, - default_ttl_days=config.get("memory", {}).get("default_ttl_days"), - ) - console.print_json(json.dumps(entry)) - conn.close() + with write_locked(paths.db_path) as conn: + initialize_db(conn) + + tag_list = [tag.strip() for tag in tags.split(",") if tag.strip()] if tags else [] + metadata_obj = json.loads(metadata) if metadata else None + entry = memory_put( + conn, + key=key, + content=content, + tags=tag_list, + metadata=metadata_obj, + ttl_days=ttl_days, + expires_at=expires_at, + default_ttl_days=config.get("memory", {}).get("default_ttl_days"), + ) + console.print_json(json.dumps(entry)) @memory_group.command("get") @@ -629,11 +624,9 @@ def memory_search_command(query: str, limit: int, config_path: Path | None) -> N def memory_prune_command(config_path: Path | None) -> None: config = load_config(config_path) paths = resolve_paths(config, config_path) - conn = connect(paths.db_path) - initialize_db(conn) - - removed = memory_prune(conn) - conn.close() + with write_locked(paths.db_path) as conn: + initialize_db(conn) + removed = memory_prune(conn) console.print(f"Pruned {removed} expired memory entries.") diff --git a/hoard/core/db/connection.py b/hoard/core/db/connection.py index b8c702a..9f7e66c 100644 --- a/hoard/core/db/connection.py +++ b/hoard/core/db/connection.py @@ -1,8 +1,9 @@ from __future__ import annotations +import contextlib import sqlite3 from pathlib import Path -from typing import Iterable +from typing import Generator, Iterable from hoard import __version__ from hoard.migrations import migrate @@ -16,6 +17,28 @@ def connect(db_path: Path, *, busy_timeout_ms: int | None = None) -> sqlite3.Con return conn +@contextlib.contextmanager +def write_locked(db_path: Path, *, busy_timeout_ms: int | None = None) -> Generator[sqlite3.Connection, None, None]: + """Open a connection and hold the cross-process write lock for its lifetime. + + Use this instead of bare ``connect()`` whenever a CLI command or + background job needs to **write** to the database, so that writes are + serialised with the MCP server's ``WriteCoordinator``. + """ + from hoard.core.db.lock import DatabaseWriteLock + + lock = DatabaseWriteLock(db_path) + lock.acquire() + try: + conn = connect(db_path, busy_timeout_ms=busy_timeout_ms) + try: + yield conn + finally: + conn.close() + finally: + lock.release() + + def _apply_pragmas(conn: sqlite3.Connection, *, busy_timeout_ms: int | None = None) -> None: conn.execute("PRAGMA foreign_keys = ON;") conn.execute("PRAGMA journal_mode = WAL;") diff --git a/hoard/core/db/lock.py b/hoard/core/db/lock.py new file mode 100644 index 0000000..d08a243 --- /dev/null +++ b/hoard/core/db/lock.py @@ -0,0 +1,128 @@ +"""Cross-process advisory locks for serialising database writes. + +Uses ``fcntl.flock`` on dedicated lock files next to the database so that +**only one writer** (whether it lives inside an MCP server, a CLI command, +or a background-sync thread) can hold the write lock at any time. + +Two lock files are used: + +* ``.lock`` -- **write lock**, held for the duration of each write + transaction. Prevents two processes from writing concurrently. +* ``.server`` -- **server singleton lock**, held for the entire + lifetime of a ``hoard serve`` process. Prevents two servers from + starting on the same database file. + +Readers never need any lock (WAL mode guarantees non-blocking reads). +""" + +from __future__ import annotations + +import fcntl +import os +import time +from pathlib import Path +from types import TracebackType +from typing import Optional, Type + + +class DatabaseLockError(Exception): + """Raised when a lock cannot be acquired.""" + + +class _AdvisoryLock: + """Low-level ``flock(2)``-based advisory lock on a file path.""" + + def __init__(self, lock_path: Path, *, timeout_seconds: float = 10.0) -> None: + self._lock_path = lock_path + self._timeout = timeout_seconds + self._fd: Optional[int] = None + + def __enter__(self) -> "_AdvisoryLock": + self.acquire() + return self + + def __exit__( + self, + exc_type: Optional[Type[BaseException]], + exc_val: Optional[BaseException], + exc_tb: Optional[TracebackType], + ) -> None: + self.release() + + def acquire(self) -> None: + self._lock_path.parent.mkdir(parents=True, exist_ok=True) + self._fd = os.open(str(self._lock_path), os.O_CREAT | os.O_RDWR) + + deadline = time.monotonic() + self._timeout + while True: + try: + fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return + except (BlockingIOError, OSError): + if time.monotonic() >= deadline: + os.close(self._fd) + self._fd = None + raise DatabaseLockError( + f"Could not acquire lock {self._lock_path} within " + f"{self._timeout}s. Another process may be holding it." + ) + time.sleep(0.05) + + def release(self) -> None: + if self._fd is not None: + try: + fcntl.flock(self._fd, fcntl.LOCK_UN) + finally: + os.close(self._fd) + self._fd = None + + def try_acquire(self) -> bool: + self._lock_path.parent.mkdir(parents=True, exist_ok=True) + self._fd = os.open(str(self._lock_path), os.O_CREAT | os.O_RDWR) + try: + fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except (BlockingIOError, OSError): + os.close(self._fd) + self._fd = None + return False + + +class DatabaseWriteLock(_AdvisoryLock): + """Exclusive, cross-process write lock. + + The lock file is ``.lock`` (e.g. ``~/.hoard/hoard.db.lock``). + + Usage:: + + lock = DatabaseWriteLock(db_path) + with lock: + conn.execute("INSERT ...") + conn.commit() + """ + + def __init__(self, db_path: Path, *, timeout_seconds: float = 10.0) -> None: + lock_path = db_path.with_suffix(db_path.suffix + ".lock") + super().__init__(lock_path, timeout_seconds=timeout_seconds) + + +class ServerSingletonLock(_AdvisoryLock): + """Prevents two ``hoard serve`` processes on the same database. + + The lock file is ``.server`` and is held for the server's + entire lifetime. It does **not** conflict with ``DatabaseWriteLock`` + because it uses a different file. + """ + + def __init__(self, db_path: Path) -> None: + lock_path = db_path.with_suffix(db_path.suffix + ".server") + super().__init__(lock_path, timeout_seconds=0) + + def acquire_or_fail(self) -> None: + """Acquire the lock or raise ``DatabaseLockError`` immediately.""" + if not self.try_acquire(): + raise DatabaseLockError( + f"Another hoard server is already running on this database.\n" + "Only one server may write to a database at a time.\n" + "Stop the other process first, or use a different storage.db_path." + ) diff --git a/hoard/core/db/writer.py b/hoard/core/db/writer.py index be05e77..2bdc56e 100644 --- a/hoard/core/db/writer.py +++ b/hoard/core/db/writer.py @@ -3,9 +3,11 @@ import queue import threading from dataclasses import dataclass +from pathlib import Path from typing import Any, Callable from hoard.core.db.connection import connect +from hoard.core.db.lock import DatabaseWriteLock @dataclass @@ -19,12 +21,18 @@ class _WriteTask: class WriteCoordinator: - def __init__(self, *, db_path, busy_timeout_ms: int | None = None) -> None: + """Serialises all database writes through a single thread **and** holds a + cross-process ``flock``-based advisory lock so that concurrent processes + (CLI commands, background sync, a second server) are also serialised. + """ + + def __init__(self, *, db_path: Path, busy_timeout_ms: int | None = None) -> None: self._queue: "queue.Queue[_WriteTask | None]" = queue.Queue() self._thread = threading.Thread(target=self._run, daemon=True) self._db_path = db_path self._busy_timeout_ms = busy_timeout_ms self._thread_id: int | None = None + self._lock = DatabaseWriteLock(db_path) self._thread.start() def submit(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: @@ -49,16 +57,17 @@ def _run(self) -> None: task = self._queue.get() if task is None: break - try: - task.result = task.fn(self._conn, *task.args, **task.kwargs) - if self._conn.in_transaction: - self._conn.commit() - except Exception as exc: - if self._conn.in_transaction: - self._conn.rollback() - task.error = exc - finally: - task.event.set() + with self._lock: + try: + task.result = task.fn(self._conn, *task.args, **task.kwargs) + if self._conn.in_transaction: + self._conn.commit() + except Exception as exc: + if self._conn.in_transaction: + self._conn.rollback() + task.error = exc + finally: + task.event.set() try: self._conn.close() diff --git a/hoard/core/mcp/server.py b/hoard/core/mcp/server.py index ec82ef2..511db9b 100644 --- a/hoard/core/mcp/server.py +++ b/hoard/core/mcp/server.py @@ -14,6 +14,7 @@ from hoard import __version__ from hoard.core.config import load_config, resolve_paths from hoard.core.db.connection import connect, ensure_sqlite_version +from hoard.core.db.lock import DatabaseLockError, ServerSingletonLock from hoard.core.db.writer import WriteCoordinator from hoard.core.mcp.tools import count_chunks, dispatch_tool, is_write_tool, tool_definitions from hoard.core.orchestrator.events import poll_events @@ -337,6 +338,19 @@ def __init__(self, server_address, RequestHandlerClass, config_path: Path | None super().__init__(server_address, RequestHandlerClass) +def _check_no_other_server(db_path: Path) -> ServerSingletonLock: + """Verify no other ``hoard serve`` process is running on the same DB. + + Acquires a non-blocking advisory lock on ``.server``. If another + server is already running, raises immediately with a clear message. + + The returned lock **must** be kept alive for the server's lifetime. + """ + lock = ServerSingletonLock(db_path) + lock.acquire_or_fail() + return lock + + def run_server( host: str = "127.0.0.1", port: int = 19850, @@ -345,6 +359,10 @@ def run_server( ) -> None: server = MCPServer((host, port), MCPRequestHandler, config_path) + # Hold an exclusive advisory lock for the server's entire lifetime so + # that a second ``hoard serve`` on the same DB is rejected immediately. + _server_lock = _check_no_other_server(server.db_path) # noqa: F841 + conn = connect(server.db_path) try: ensure_sqlite_version() diff --git a/hoard/core/sync/service.py b/hoard/core/sync/service.py index 3376232..02e99a0 100644 --- a/hoard/core/sync/service.py +++ b/hoard/core/sync/service.py @@ -7,7 +7,7 @@ from typing import Any, Dict, List, Optional from hoard.core.config import default_data_path, resolve_paths -from hoard.core.db.connection import connect, initialize_db +from hoard.core.db.connection import connect, initialize_db, write_locked from hoard.core.ingest.registry import iter_enabled_connectors from hoard.core.ingest.sync import sync_connector from hoard.core.memory.store import memory_prune @@ -62,12 +62,9 @@ def run_sync_with_lock( lock_path: Optional[Path] = None, ) -> Dict[str, Any]: paths = resolve_paths(config, config_path) - conn = connect(paths.db_path) - initialize_db(conn) - try: + with write_locked(paths.db_path) as conn: + initialize_db(conn) return sync_with_lock(conn, config, source=source, lock_path=lock_path) - finally: - conn.close() def sync_with_lock( diff --git a/tests/test_write_lock.py b/tests/test_write_lock.py new file mode 100644 index 0000000..2158cbc --- /dev/null +++ b/tests/test_write_lock.py @@ -0,0 +1,291 @@ +"""Tests for cross-process database write locking. + +Covers: +- DatabaseWriteLock basic acquire/release +- DatabaseWriteLock mutual exclusion across threads (simulating processes) +- ServerSingletonLock prevents two servers on the same DB +- WriteCoordinator acquires/releases the lock per write +- write_locked() context manager +- CLI and sync paths hold the lock during writes +""" + +from __future__ import annotations + +import json +import threading +import time +import urllib.request +from pathlib import Path + +import pytest + +from hoard.core.config import save_config +from hoard.core.db.connection import connect, initialize_db, write_locked +from hoard.core.db.lock import DatabaseLockError, DatabaseWriteLock, ServerSingletonLock +from hoard.core.db.writer import WriteCoordinator +from hoard.core.memory.store import memory_put + + +# --------------------------------------------------------------------------- +# DatabaseWriteLock unit tests +# --------------------------------------------------------------------------- + + +def test_write_lock_acquire_release(tmp_path: Path) -> None: + db_path = tmp_path / "test.db" + db_path.touch() + + lock = DatabaseWriteLock(db_path) + lock.acquire() + lock.release() + + +def test_write_lock_context_manager(tmp_path: Path) -> None: + db_path = tmp_path / "test.db" + db_path.touch() + + with DatabaseWriteLock(db_path): + pass # lock held here + + +def test_write_lock_blocks_second_holder(tmp_path: Path) -> None: + """Two threads competing for the write lock: only one holds it at a time.""" + db_path = tmp_path / "test.db" + db_path.touch() + + order: list[str] = [] + barrier = threading.Barrier(2) + + def holder(name: str, hold_seconds: float) -> None: + barrier.wait() + with DatabaseWriteLock(db_path, timeout_seconds=5): + order.append(f"{name}-acquired") + time.sleep(hold_seconds) + order.append(f"{name}-released") + + t1 = threading.Thread(target=holder, args=("A", 0.3)) + t2 = threading.Thread(target=holder, args=("B", 0.0)) + t1.start() + t2.start() + t1.join() + t2.join() + + # One must fully complete before the other starts + assert order[0].endswith("-acquired") + assert order[1].endswith("-released") + assert order[0][0] == order[1][0] # same thread name for first pair + + +def test_write_lock_timeout(tmp_path: Path) -> None: + db_path = tmp_path / "test.db" + db_path.touch() + + lock_a = DatabaseWriteLock(db_path) + lock_a.acquire() + + with pytest.raises(DatabaseLockError): + DatabaseWriteLock(db_path, timeout_seconds=0.1).acquire() + + lock_a.release() + + +# --------------------------------------------------------------------------- +# ServerSingletonLock unit tests +# --------------------------------------------------------------------------- + + +def test_server_singleton_blocks_second(tmp_path: Path) -> None: + db_path = tmp_path / "test.db" + db_path.touch() + + lock_a = ServerSingletonLock(db_path) + lock_a.acquire_or_fail() + + with pytest.raises(DatabaseLockError, match="Another hoard server"): + ServerSingletonLock(db_path).acquire_or_fail() + + lock_a.release() + + # Now it should succeed + lock_b = ServerSingletonLock(db_path) + lock_b.acquire_or_fail() + lock_b.release() + + +def test_server_singleton_does_not_block_write_lock(tmp_path: Path) -> None: + """ServerSingletonLock uses a different file, so it doesn't block writes.""" + db_path = tmp_path / "test.db" + db_path.touch() + + server_lock = ServerSingletonLock(db_path) + server_lock.acquire_or_fail() + + # This should succeed because they use different lock files + write_lock = DatabaseWriteLock(db_path, timeout_seconds=0.1) + write_lock.acquire() + write_lock.release() + + server_lock.release() + + +# --------------------------------------------------------------------------- +# WriteCoordinator lock integration +# --------------------------------------------------------------------------- + + +def test_write_coordinator_serialises_across_threads(tmp_path: Path) -> None: + """Multiple threads submitting writes are serialised by the lock.""" + db_path = tmp_path / "test.db" + conn = connect(db_path) + initialize_db(conn) + conn.close() + + writer = WriteCoordinator(db_path=db_path) + results: list[int] = [] + errors: list[str] = [] + + def _write_memory(conn, idx: int) -> None: + conn.execute( + "INSERT OR REPLACE INTO memory_entries " + "(id, key, content, tags, tags_text, metadata, created_at, updated_at) " + "VALUES (?, ?, ?, '', '', '{}', datetime('now'), datetime('now'))", + (f"mem-{idx}", f"key-{idx}", f"content-{idx}"), + ) + + def worker(start: int) -> None: + try: + for i in range(start, start + 5): + writer.submit(_write_memory, i) + results.append(i) + except Exception as exc: + errors.append(str(exc)) + + threads = [threading.Thread(target=worker, args=(n * 5,)) for n in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + + writer.stop() + + assert not errors, f"Errors: {errors}" + assert len(results) == 20 + + +# --------------------------------------------------------------------------- +# write_locked() context manager +# --------------------------------------------------------------------------- + + +def test_write_locked_context_manager(tmp_path: Path) -> None: + db_path = tmp_path / "test.db" + conn = connect(db_path) + initialize_db(conn) + conn.close() + + with write_locked(db_path) as conn: + conn.execute( + "INSERT OR REPLACE INTO memory_entries " + "(id, key, content, tags, tags_text, metadata, created_at, updated_at) " + "VALUES ('m1', 'k1', 'v1', '', '', '{}', datetime('now'), datetime('now'))" + ) + conn.commit() + + # Verify the write persisted + conn = connect(db_path) + row = conn.execute("SELECT content FROM memory_entries WHERE key = 'k1'").fetchone() + conn.close() + assert row is not None + assert row["content"] == "v1" + + +# --------------------------------------------------------------------------- +# Concurrent HTTP write test +# --------------------------------------------------------------------------- + + +def _call_mcp(url: str, token: str, method: str, params: dict) -> tuple[int, dict]: + payload = {"jsonrpc": "2.0", "id": 1, "method": method, "params": params} + data = json.dumps(payload).encode("utf-8") + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {token}", + } + req = urllib.request.Request(url, data=data, headers=headers) + with urllib.request.urlopen(req, timeout=10) as resp: + return resp.status, json.loads(resp.read()) + + +def test_http_mcp_concurrent_writes(tmp_path: Path, mcp_server) -> None: + """Multiple agents writing memories concurrently must all succeed.""" + db_path = tmp_path / "hoard.db" + config_path = tmp_path / "config.yaml" + config = { + "security": { + "tokens": [ + { + "name": "agent", + "token": "hoard_sk_agent", + "scopes": ["search", "get", "memory", "sync", "ingest"], + } + ] + }, + "storage": {"db_path": str(db_path)}, + "connectors": {"local_files": {"enabled": False}}, + "vectors": {"enabled": False}, + "write": { + "enabled": True, + "embeddings": {"enabled": False}, + "server_secret_env": "", + "auto_generate_server_secret": False, + }, + } + save_config(config, config_path) + + conn = connect(db_path) + initialize_db(conn) + conn.close() + + url = mcp_server(config_path) + + errors: list[str] = [] + lock = threading.Lock() + writes_per_thread = 5 + num_threads = 4 + + def writer(thread_id: int) -> None: + try: + for i in range(writes_per_thread): + key = f"fact:thread{thread_id}.item{i}" + _, resp = _call_mcp( + url, + "hoard_sk_agent", + "tools/call", + { + "name": "memory_write", + "arguments": { + "content": f"Memory from thread {thread_id} item {i}", + "slot": key, + }, + }, + ) + if "error" in resp: + with lock: + errors.append(f"t{thread_id}i{i}: {resp['error']}") + except Exception as exc: + with lock: + errors.append(f"t{thread_id}: {exc}") + + threads = [threading.Thread(target=writer, args=(n,)) for n in range(num_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors, f"Concurrent write errors: {errors}" + + # Verify all writes landed + conn = connect(db_path) + count = conn.execute("SELECT COUNT(*) FROM memories").fetchone()[0] + conn.close() + assert count == writes_per_thread * num_threads From 764a0e2fee286ab0dee5caa8a3dfa9c6a71de5dd Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Feb 2026 10:15:12 +0000 Subject: [PATCH 3/5] Fix same-process flock deadlock between BackgroundSync and WriteCoordinator BackgroundSync runs inside the hoard serve process. When it called write_locked(), it acquired flock(fd_A) on hoard.db.lock. Any MCP write then hit WriteCoordinator._run which tried flock(fd_B) on the same file. On Linux flock is per-open-file-description not per-process, so fd_B blocked. After the 10s timeout the writer thread died from an unhandled DatabaseLockError, and every submit() caller hung forever on event.wait(). Fixes: - Revert run_sync_with_lock to plain connect() since it runs inside the server process where WriteCoordinator already serializes writes. Cross-process safety relies on SQLite WAL + busy_timeout (original behavior). - Wrap the lock acquisition in WriteCoordinator._run with a try/except so that if the lock cannot be acquired, the error is reported to the caller via task.error and the writer thread keeps running. - Move task.event.set() to an outer finally block so callers are always unblocked, even if lock acquisition fails. Adds a regression test that holds the flock externally, verifies the WriteCoordinator reports the error (not hangs), and then verifies the writer still works after the contention clears. All 143 tests pass. https://claude.ai/code/session_012Zb8mCrSHJtkuCu9mqZCZ2 --- hoard/core/db/writer.py | 27 ++++++++++++--------- hoard/core/sync/service.py | 19 ++++++++++++--- tests/test_write_lock.py | 48 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 80 insertions(+), 14 deletions(-) diff --git a/hoard/core/db/writer.py b/hoard/core/db/writer.py index 2bdc56e..ee8d89f 100644 --- a/hoard/core/db/writer.py +++ b/hoard/core/db/writer.py @@ -57,17 +57,22 @@ def _run(self) -> None: task = self._queue.get() if task is None: break - with self._lock: - try: - task.result = task.fn(self._conn, *task.args, **task.kwargs) - if self._conn.in_transaction: - self._conn.commit() - except Exception as exc: - if self._conn.in_transaction: - self._conn.rollback() - task.error = exc - finally: - task.event.set() + try: + with self._lock: + try: + task.result = task.fn(self._conn, *task.args, **task.kwargs) + if self._conn.in_transaction: + self._conn.commit() + except Exception as exc: + if self._conn.in_transaction: + self._conn.rollback() + task.error = exc + except Exception as exc: + # Lock acquisition itself failed -- report to caller + # instead of killing the writer thread. + task.error = exc + finally: + task.event.set() try: self._conn.close() diff --git a/hoard/core/sync/service.py b/hoard/core/sync/service.py index 02e99a0..8919046 100644 --- a/hoard/core/sync/service.py +++ b/hoard/core/sync/service.py @@ -7,7 +7,7 @@ from typing import Any, Dict, List, Optional from hoard.core.config import default_data_path, resolve_paths -from hoard.core.db.connection import connect, initialize_db, write_locked +from hoard.core.db.connection import connect, initialize_db from hoard.core.ingest.registry import iter_enabled_connectors from hoard.core.ingest.sync import sync_connector from hoard.core.memory.store import memory_prune @@ -61,10 +61,23 @@ def run_sync_with_lock( source: Optional[str] = None, lock_path: Optional[Path] = None, ) -> Dict[str, Any]: + """Run a sync, acquiring the file-based sync lock to prevent overlaps. + + NOTE: This function is called both from the CLI (standalone process) and + from ``BackgroundSync`` (inside the ``hoard serve`` process). It must + **not** hold a ``DatabaseWriteLock`` for the entire sync duration because + that would deadlock the ``WriteCoordinator`` running in the same process + (``flock`` is per-open-file-description, not per-process). Instead we + rely on SQLite WAL + ``busy_timeout`` for write serialisation here, and + the ``WriteCoordinator``'s per-task flock for MCP writes. + """ paths = resolve_paths(config, config_path) - with write_locked(paths.db_path) as conn: - initialize_db(conn) + conn = connect(paths.db_path) + initialize_db(conn) + try: return sync_with_lock(conn, config, source=source, lock_path=lock_path) + finally: + conn.close() def sync_with_lock( diff --git a/tests/test_write_lock.py b/tests/test_write_lock.py index 2158cbc..cffb25d 100644 --- a/tests/test_write_lock.py +++ b/tests/test_write_lock.py @@ -5,6 +5,7 @@ - DatabaseWriteLock mutual exclusion across threads (simulating processes) - ServerSingletonLock prevents two servers on the same DB - WriteCoordinator acquires/releases the lock per write +- WriteCoordinator resilience when lock is contended - write_locked() context manager - CLI and sync paths hold the lock during writes """ @@ -172,6 +173,53 @@ def worker(start: int) -> None: assert len(results) == 20 +def test_write_coordinator_survives_lock_contention(tmp_path: Path) -> None: + """If another thread holds the flock, the WriteCoordinator must report + the error to the caller and keep running -- not deadlock or crash. + + Regression test: previously, a DatabaseLockError from acquire() would + propagate past the task loop, killing the writer thread and leaving + all submit() callers hanging on event.wait() forever. + """ + db_path = tmp_path / "test.db" + conn = connect(db_path) + initialize_db(conn) + conn.close() + + writer = WriteCoordinator(db_path=db_path) + + # Hold the flock from another fd (simulating BackgroundSync or CLI) + external_lock = DatabaseWriteLock(db_path) + external_lock.acquire() + + # Submit a write -- should fail with DatabaseLockError, NOT hang + def _noop_write(conn): + pass + + with pytest.raises(Exception): + writer.submit(_noop_write) + + # Release the external lock + external_lock.release() + + # Writer thread must still be alive -- a subsequent write should succeed + def _insert(conn): + conn.execute( + "INSERT OR REPLACE INTO memory_entries " + "(id, key, content, tags, tags_text, metadata, created_at, updated_at) " + "VALUES ('m1', 'k1', 'survived', '', '', '{}', datetime('now'), datetime('now'))" + ) + + writer.submit(_insert) + writer.stop() + + conn = connect(db_path) + row = conn.execute("SELECT content FROM memory_entries WHERE key = 'k1'").fetchone() + conn.close() + assert row is not None + assert row["content"] == "survived" + + # --------------------------------------------------------------------------- # write_locked() context manager # --------------------------------------------------------------------------- From 06698e3f14bc7ce87b8252e0b555e219ce1f4090 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Feb 2026 10:51:43 +0000 Subject: [PATCH 4/5] Update analysis doc to reflect implemented cross-process locking Updates the analysis to document all seven layers of write protection now in place, including the new flock-based DatabaseWriteLock, ServerSingletonLock, and the flock caveat for same-process usage. Adds flow diagrams for CLI-vs-server write serialization and server singleton rejection. Moves resolved gaps to a "Previously Fixed" table. https://claude.ai/code/session_012Zb8mCrSHJtkuCu9mqZCZ2 --- ANALYSIS_DB_AGENT_LOCKING.md | 178 ++++++++++++++++++++++++++++------- 1 file changed, 146 insertions(+), 32 deletions(-) diff --git a/ANALYSIS_DB_AGENT_LOCKING.md b/ANALYSIS_DB_AGENT_LOCKING.md index 2194e1b..62567bf 100644 --- a/ANALYSIS_DB_AGENT_LOCKING.md +++ b/ANALYSIS_DB_AGENT_LOCKING.md @@ -6,13 +6,15 @@ ## Short Answer -Yes. Hoard uses a **five-layer defense** against write conflicts on a shared SQLite database: +Yes. Hoard uses a **seven-layer defense** against write conflicts on a shared SQLite database: 1. **WAL mode** (database-level) -- readers never block writers 2. **WriteCoordinator** (process-level) -- serializes all writes through a single thread -3. **Busy timeout** (connection-level) -- retries on `SQLITE_BUSY` for up to 5 seconds -4. **Optimistic concurrency** (row-level) -- `rowcount` checks detect lost updates -5. **Async conflict detection** (application-level) -- background jobs find semantic conflicts +3. **Cross-process flock** (OS-level) -- `DatabaseWriteLock` serializes writes across processes +4. **Server singleton lock** (OS-level) -- `ServerSingletonLock` prevents two servers on the same DB +5. **Busy timeout** (connection-level) -- retries on `SQLITE_BUSY` for up to 5 seconds +6. **Optimistic concurrency** (row-level) -- `rowcount` checks detect lost updates +7. **Async conflict detection** (application-level) -- background jobs find semantic conflicts These layers work together so that multiple agents sharing one Hoard server will not corrupt data, lose writes, or deadlock. @@ -59,7 +61,10 @@ class WriteCoordinator: return task.result ``` -The worker loop processes one task at a time, auto-committing on success and rolling back on error: +The worker loop processes one task at a time. Each write is wrapped in a cross-process +`DatabaseWriteLock` (flock), auto-committing on success and rolling back on error. If +the flock cannot be acquired (another process is writing), the error is reported to the +caller and the writer thread keeps running: ```python def _run(self): @@ -67,13 +72,17 @@ def _run(self): while True: task = self._queue.get() try: - task.result = task.fn(self._conn, *task.args, **task.kwargs) - if self._conn.in_transaction: - self._conn.commit() + with self._lock: # flock-based DatabaseWriteLock + try: + task.result = task.fn(self._conn, *task.args, **task.kwargs) + if self._conn.in_transaction: + self._conn.commit() + except Exception as exc: + if self._conn.in_transaction: + self._conn.rollback() + task.error = exc except Exception as exc: - if self._conn.in_transaction: - self._conn.rollback() - task.error = exc + task.error = exc # lock failure -- report, don't crash finally: task.event.set() ``` @@ -95,7 +104,61 @@ Write tools go through `writer.submit()` (serialized). Read tools execute direct --- -## Layer 3: Busy Timeout +## Layer 3: Cross-Process Write Lock (flock) + +**File:** `hoard/core/db/lock.py` + +Every write processed by the `WriteCoordinator` is wrapped in a `DatabaseWriteLock` -- an +exclusive `flock(2)` advisory lock on `.lock`: + +```python +# In WriteCoordinator._run(): +with self._lock: # acquires flock(LOCK_EX) + task.result = task.fn(self._conn, ...) + self._conn.commit() + # flock(LOCK_UN) released here +``` + +CLI commands that write (`memory put`, `memory prune`, `db migrate`) use the +`write_locked()` context manager, which holds the same flock for the connection's lifetime: + +```python +# In cli/main.py: +with write_locked(paths.db_path) as conn: + initialize_db(conn) + memory_put(conn, ...) +``` + +Because both the server and CLI acquire the same flock file, writes are serialized across +process boundaries -- not just within the server process. + +**Important caveat:** `flock` is per-open-file-description on Linux, not per-process. Two +`open()` calls on the same file within the same process create independent locks that +contend with each other. This is why `BackgroundSync` (which runs inside the server +process) does **not** use `write_locked()` -- it would deadlock the `WriteCoordinator`. +Instead, sync relies on SQLite WAL + busy_timeout for write safety. + +--- + +## Layer 4: Server Singleton Lock + +**File:** `hoard/core/db/lock.py`, `hoard/core/mcp/server.py` + +On startup, `run_server()` acquires a `ServerSingletonLock` on `.server` (a +separate file from the write lock). This lock is held for the server's entire lifetime: + +```python +_server_lock = _check_no_other_server(server.db_path) +# ... server runs ... +# lock released when process exits +``` + +If a second `hoard serve` attempts to start on the same database, it fails immediately +with a clear error message instead of silently racing. + +--- + +## Layer 5: Busy Timeout **File:** `hoard/core/config.py:92` @@ -116,7 +179,7 @@ If a connection encounters `SQLITE_BUSY` (e.g., during migrations or if an exter --- -## Layer 4: Optimistic Concurrency (Row-Level Conflict Detection) +## Layer 6: Optimistic Concurrency (Row-Level Conflict Detection) Multiple operations use `UPDATE ... WHERE status = ? ... ` + `cursor.rowcount` as an **optimistic lock**: @@ -153,7 +216,7 @@ Similar `rowcount > 0` checks guard memory updates and deletions. --- -## Layer 5: Application-Level Conflict Detection +## Layer 7: Application-Level Conflict Detection Hoard recognizes that even with serialized writes, **semantic conflicts** can occur (two agents writing different values to the same memory slot). These are handled asynchronously: @@ -177,7 +240,7 @@ These are surfaced via MCP tools (`conflicts_list`, `conflict_resolve`, `duplica --- -## Layer 6 (Bonus): File-Based Sync Lock +## Layer 8 (Bonus): File-Based Sync Lock **File:** `hoard/core/sync/service.py:122-136` @@ -212,19 +275,26 @@ Agent B writes memory ─┘ │ ▼ Single Writer Thread - (one connection, one txn at a time) │ - ┌─────────┼─────────┐ - │ Execute │ Commit │ - │ A's fn │ or │ - │ │ Rollback│ - └─────────┼─────────┘ + ┌─────────┼──────────────┐ + │ flock(LOCK_EX) on │ + │ hoard.db.lock │ + │ │ + │ Execute A's fn │ + │ Commit or Rollback │ + │ │ + │ flock(LOCK_UN) │ + └─────────┼──────────────┘ │ - ┌─────────┼─────────┐ - │ Execute │ Commit │ - │ B's fn │ or │ - │ │ Rollback│ - └─────────┼─────────┘ + ┌─────────┼──────────────┐ + │ flock(LOCK_EX) on │ + │ hoard.db.lock │ + │ │ + │ Execute B's fn │ + │ Commit or Rollback │ + │ │ + │ flock(LOCK_UN) │ + └─────────┼──────────────┘ │ ▼ Background jobs detect conflicts/duplicates @@ -233,17 +303,61 @@ Agent B writes memory ─┘ 1. Both requests arrive on separate HTTP threads 2. Both call `writer.submit()`, enqueueing their operations 3. The writer thread processes them **sequentially** (FIFO) -4. Each gets its own transaction with auto-commit/rollback -5. Neither agent sees a `SQLITE_BUSY` error or partial data -6. Post-write background jobs check for semantic conflicts +4. Each write acquires the cross-process `DatabaseWriteLock` (flock) +5. Each gets its own transaction with auto-commit/rollback +6. The flock is released after each write, so CLI commands can interleave between writes +7. Neither agent sees a `SQLITE_BUSY` error or partial data +8. Post-write background jobs check for semantic conflicts + +--- + +## What Happens When a CLI Command Writes While the Server is Running + +``` +hoard memory put "key" "value" hoard serve (MCP server) + │ │ + ▼ ▼ + write_locked(db_path) WriteCoordinator._run() + │ │ + flock(LOCK_EX) on flock(LOCK_EX) on + hoard.db.lock ◄─── only one ───► hoard.db.lock + │ at a time │ + INSERT + COMMIT (blocks until CLI + │ releases the lock) + flock(LOCK_UN) │ + │ flock acquired + ▼ Execute + Commit + done flock(LOCK_UN) +``` --- -## Potential Gaps +## What Happens When Two Servers Are Started + +``` +$ hoard serve --port 19850 + → ServerSingletonLock acquired on hoard.db.server ✓ + → Server running + +$ hoard serve --port 19851 (same DB) + → ServerSingletonLock: try_acquire() → False + → ERROR: "Another hoard server is already running on this database." +``` + +--- + +## Remaining Gaps | Gap | Severity | Notes | |-----|----------|-------| -| **Single-process only** | Medium | The WriteCoordinator serializes writes within one server process. If two separate `hoard serve` processes point at the same DB file, they would rely solely on WAL + busy_timeout (no queue serialization). This is not a supported deployment. | -| **No concurrent write tests** | Low | `test_http_mcp_concurrency.py` only tests concurrent reads (5 threads x 3 search calls). There are no tests exercising concurrent writes from multiple agents. | | **Task claim is not atomic SELECT+UPDATE** | Low | `_claim_job` in `worker.py:140-161` does a SELECT then UPDATE in two statements. This is safe because it runs inside the WriteCoordinator (single thread), but would be a race condition if used with direct connections. | | **Conflict resolution is manual** | Low | Detected conflicts require explicit resolution via MCP tools. There is no auto-merge strategy. | +| **NFS/overlay filesystem caveat** | Low | `flock(2)` may not work correctly on network filesystems (NFS, some Docker volume drivers). This is a limitation of the OS, not the code. For Docker deployments, use local volumes. | + +### Previously Identified Gaps (Now Fixed) + +| Gap | Resolution | +|-----|------------| +| **Single-process only** | `WriteCoordinator` now holds a `DatabaseWriteLock` (flock) per write, serializing with CLI and sync processes. `ServerSingletonLock` prevents two servers on the same DB. | +| **No concurrent write tests** | `tests/test_write_lock.py` includes 10 tests covering lock semantics, WriteCoordinator serialization, lock contention resilience, and 4-thread concurrent HTTP writes. | +| **CLI writes bypass WriteCoordinator** | `memory put`, `memory prune`, and `db migrate` now use `write_locked()`, which holds the same flock as the WriteCoordinator. | From 7a8c778cbc26e9a6482c33404100f8f9388aa537 Mon Sep 17 00:00:00 2001 From: piotr-auguscik Date: Sat, 7 Feb 2026 12:53:27 +0100 Subject: [PATCH 5/5] Unify DB write serialization across sync, server, and CLI --- ANALYSIS_DB_AGENT_LOCKING.md | 6 +- README.md | 10 + hoard/cli/main.py | 52 ++++- hoard/core/config.py | 3 + hoard/core/db/connection.py | 16 +- hoard/core/db/lock.py | 6 +- hoard/core/db/write_exec.py | 65 ++++++ hoard/core/db/writer.py | 72 +++++-- hoard/core/ingest/sync.py | 55 +++-- hoard/core/mcp/server.py | 98 ++++++--- hoard/core/memory/store.py | 2 - hoard/core/sync/background.py | 32 ++- hoard/core/sync/service.py | 68 ++++-- tests/conftest.py | 27 +++ tests/test_cli.py | 2 + tests/test_write_lock.py | 382 +++++++++++++++++++++++++++++++++- 16 files changed, 791 insertions(+), 105 deletions(-) create mode 100644 hoard/core/db/write_exec.py diff --git a/ANALYSIS_DB_AGENT_LOCKING.md b/ANALYSIS_DB_AGENT_LOCKING.md index 62567bf..5f39ce7 100644 --- a/ANALYSIS_DB_AGENT_LOCKING.md +++ b/ANALYSIS_DB_AGENT_LOCKING.md @@ -134,9 +134,9 @@ process boundaries -- not just within the server process. **Important caveat:** `flock` is per-open-file-description on Linux, not per-process. Two `open()` calls on the same file within the same process create independent locks that -contend with each other. This is why `BackgroundSync` (which runs inside the server -process) does **not** use `write_locked()` -- it would deadlock the `WriteCoordinator`. -Instead, sync relies on SQLite WAL + busy_timeout for write safety. +contend with each other. Hoard avoids this by routing background sync writes through +the server's `WriteCoordinator.submit()` path, so in-process sync and MCP writes share +the same write queue and lock policy. --- diff --git a/README.md b/README.md index a7a9570..ac92b15 100644 --- a/README.md +++ b/README.md @@ -190,6 +190,16 @@ Admin tasks (token management) use the same secret: hoard tokens add claude-code --scopes search,get,memory,sync ``` +Write coordination defaults (in `${HOARD_DATA_DIR:-~/.hoard}/config.yaml`): +```yaml +write: + database: + busy_timeout_ms: 5000 + lock_timeout_ms: 30000 + retry_budget_ms: 30000 + retry_backoff_ms: 50 +``` + ### MCP Clients ```bash hoard setup --all diff --git a/hoard/cli/main.py b/hoard/cli/main.py index aeb9151..e9d7f1f 100644 --- a/hoard/cli/main.py +++ b/hoard/cli/main.py @@ -1,7 +1,7 @@ from __future__ import annotations -import json import ipaddress +import json import os import secrets import shutil @@ -29,7 +29,13 @@ render_instruction_block, resolve_project_root, ) -from hoard.core.config import default_data_path, ensure_config_file, load_config, resolve_paths, save_config +from hoard.core.config import ( + default_data_path, + ensure_config_file, + load_config, + resolve_paths, + save_config, +) from hoard.core.db.connection import connect, initialize_db, write_locked from hoard.core.embeddings.model import EmbeddingError, EmbeddingModel from hoard.core.mcp.server import run_server @@ -50,7 +56,6 @@ server_secret_env_key, server_secret_file_path, ) -from hoard.core.sync.background import BackgroundSync from hoard.core.sync.service import run_sync_with_lock from hoard.core.sync.watcher import WATCHDOG_AVAILABLE from hoard.sdk.retry import run_with_retry, should_retry_http_exception @@ -59,6 +64,21 @@ DEFAULT_AGENT_SCOPES = ["search", "get", "memory", "sync", "ingest"] +def _write_db_timeouts(config: dict) -> tuple[int, int]: + db_cfg = config.get("write", {}).get("database", {}) + busy_timeout_ms = _int_or_default(db_cfg.get("busy_timeout_ms"), 5000) + lock_timeout_ms = _int_or_default(db_cfg.get("lock_timeout_ms"), 30000) + return busy_timeout_ms, lock_timeout_ms + + +def _int_or_default(value: object, default: int) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + return default + return parsed if parsed > 0 else default + + def _server_base_url(config: dict) -> str: host = config.get("server", {}).get("host", "127.0.0.1") port = int(config.get("server", {}).get("port", 19850)) @@ -444,7 +464,12 @@ def db_migrate_command(target_version: int | None, config_path: Path | None) -> config = load_config(config_path) paths = resolve_paths(config, config_path) - with write_locked(paths.db_path) as conn: + busy_timeout_ms, lock_timeout_ms = _write_db_timeouts(config) + with write_locked( + paths.db_path, + busy_timeout_ms=busy_timeout_ms, + lock_timeout_ms=lock_timeout_ms, + ) as conn: current = get_current_version(conn) migrations = get_migrations() latest = max(migrations.keys()) if migrations else 0 @@ -569,7 +594,12 @@ def memory_put_command( ) -> None: config = load_config(config_path) paths = resolve_paths(config, config_path) - with write_locked(paths.db_path) as conn: + busy_timeout_ms, lock_timeout_ms = _write_db_timeouts(config) + with write_locked( + paths.db_path, + busy_timeout_ms=busy_timeout_ms, + lock_timeout_ms=lock_timeout_ms, + ) as conn: initialize_db(conn) tag_list = [tag.strip() for tag in tags.split(",") if tag.strip()] if tags else [] @@ -624,7 +654,12 @@ def memory_search_command(query: str, limit: int, config_path: Path | None) -> N def memory_prune_command(config_path: Path | None) -> None: config = load_config(config_path) paths = resolve_paths(config, config_path) - with write_locked(paths.db_path) as conn: + busy_timeout_ms, lock_timeout_ms = _write_db_timeouts(config) + with write_locked( + paths.db_path, + busy_timeout_ms=busy_timeout_ms, + lock_timeout_ms=lock_timeout_ms, + ) as conn: initialize_db(conn) removed = memory_prune(conn) console.print(f"Pruned {removed} expired memory entries.") @@ -1290,9 +1325,6 @@ def serve_command( _serve_daemon(host, port, no_migrate=no_migrate, allow_remote=allow_remote) return - background = BackgroundSync(config=config, config_path=None, log=console.print) - background.start() - console.print(f"Starting Hoard server on http://{host}:{port}/mcp") run_server(host=host, port=port, config_path=None, no_migrate=no_migrate) @@ -1326,8 +1358,6 @@ def mcp_serve_command( raise click.ClickException(str(exc)) from exc if not had_secret and source == "file": console.print(f"Generated server secret in {server_secret_file_path(config)}") - background = BackgroundSync(config=config, config_path=config_path, log=console.print) - background.start() console.print(f"Starting MCP server on {host}:{port}") run_server(host=host, port=port, config_path=config_path, no_migrate=no_migrate) diff --git a/hoard/core/config.py b/hoard/core/config.py index 287c305..9402640 100644 --- a/hoard/core/config.py +++ b/hoard/core/config.py @@ -90,6 +90,9 @@ "auto_generate_server_secret": True, "database": { "busy_timeout_ms": 5000, + "lock_timeout_ms": 30000, + "retry_budget_ms": 30000, + "retry_backoff_ms": 50, }, "slots": { "pattern": "^(pref|fact|ctx|decision|event):[a-z0-9_]+(\\.[a-z0-9_]+){0,3}$", diff --git a/hoard/core/db/connection.py b/hoard/core/db/connection.py index 9f7e66c..9a6fb6c 100644 --- a/hoard/core/db/connection.py +++ b/hoard/core/db/connection.py @@ -18,7 +18,12 @@ def connect(db_path: Path, *, busy_timeout_ms: int | None = None) -> sqlite3.Con @contextlib.contextmanager -def write_locked(db_path: Path, *, busy_timeout_ms: int | None = None) -> Generator[sqlite3.Connection, None, None]: +def write_locked( + db_path: Path, + *, + busy_timeout_ms: int | None = None, + lock_timeout_ms: int | None = None, +) -> Generator[sqlite3.Connection, None, None]: """Open a connection and hold the cross-process write lock for its lifetime. Use this instead of bare ``connect()`` whenever a CLI command or @@ -27,12 +32,19 @@ def write_locked(db_path: Path, *, busy_timeout_ms: int | None = None) -> Genera """ from hoard.core.db.lock import DatabaseWriteLock - lock = DatabaseWriteLock(db_path) + timeout_ms = lock_timeout_ms if lock_timeout_ms and lock_timeout_ms > 0 else 30000 + lock = DatabaseWriteLock(db_path, timeout_seconds=timeout_ms / 1000) lock.acquire() try: conn = connect(db_path, busy_timeout_ms=busy_timeout_ms) try: yield conn + if conn.in_transaction: + conn.commit() + except Exception: + if conn.in_transaction: + conn.rollback() + raise finally: conn.close() finally: diff --git a/hoard/core/db/lock.py b/hoard/core/db/lock.py index d08a243..6249a56 100644 --- a/hoard/core/db/lock.py +++ b/hoard/core/db/lock.py @@ -32,7 +32,7 @@ class DatabaseLockError(Exception): class _AdvisoryLock: """Low-level ``flock(2)``-based advisory lock on a file path.""" - def __init__(self, lock_path: Path, *, timeout_seconds: float = 10.0) -> None: + def __init__(self, lock_path: Path, *, timeout_seconds: float = 30.0) -> None: self._lock_path = lock_path self._timeout = timeout_seconds self._fd: Optional[int] = None @@ -101,7 +101,7 @@ class DatabaseWriteLock(_AdvisoryLock): conn.commit() """ - def __init__(self, db_path: Path, *, timeout_seconds: float = 10.0) -> None: + def __init__(self, db_path: Path, *, timeout_seconds: float = 30.0) -> None: lock_path = db_path.with_suffix(db_path.suffix + ".lock") super().__init__(lock_path, timeout_seconds=timeout_seconds) @@ -122,7 +122,7 @@ def acquire_or_fail(self) -> None: """Acquire the lock or raise ``DatabaseLockError`` immediately.""" if not self.try_acquire(): raise DatabaseLockError( - f"Another hoard server is already running on this database.\n" + "Another hoard server is already running on this database.\n" "Only one server may write to a database at a time.\n" "Stop the other process first, or use a different storage.db_path." ) diff --git a/hoard/core/db/write_exec.py b/hoard/core/db/write_exec.py new file mode 100644 index 0000000..9341ea4 --- /dev/null +++ b/hoard/core/db/write_exec.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import contextlib +from pathlib import Path +from typing import Any, Callable, Generator, Mapping, Protocol + +from hoard.core.db.writer import WriteCoordinator + + +class WriteSubmit(Protocol): + def submit(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + """Execute a write closure with a writable DB connection.""" + + +class _DirectWriteSubmit: + def __init__(self, conn) -> None: + self._conn = conn + + def submit(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + try: + result = fn(self._conn, *args, **kwargs) + if self._conn.in_transaction: + self._conn.commit() + return result + except Exception: + if self._conn.in_transaction: + self._conn.rollback() + raise + + +def direct_submit(conn) -> WriteSubmit: + """Return a WriteSubmit implementation for an existing connection.""" + return _DirectWriteSubmit(conn) + + +@contextlib.contextmanager +def temporary_coordinator_submit( + db_path: Path, + db_cfg: Mapping[str, Any] | None = None, +) -> Generator[WriteSubmit, None, None]: + """Create a temporary WriteCoordinator-backed submitter. + + Used by standalone flows (e.g. CLI sync) so they share the same + lock/retry policy as the long-lived server writer. + """ + cfg = db_cfg or {} + writer = WriteCoordinator( + db_path=db_path, + busy_timeout_ms=_int_or_default(cfg.get("busy_timeout_ms"), 5000), + lock_timeout_ms=_int_or_default(cfg.get("lock_timeout_ms"), 30000), + retry_budget_ms=_int_or_default(cfg.get("retry_budget_ms"), 30000), + retry_backoff_ms=min(_int_or_default(cfg.get("retry_backoff_ms"), 50), 500), + ) + try: + yield writer + finally: + writer.stop() + + +def _int_or_default(value: Any, default: int) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + return default + return parsed if parsed > 0 else default diff --git a/hoard/core/db/writer.py b/hoard/core/db/writer.py index ee8d89f..812223e 100644 --- a/hoard/core/db/writer.py +++ b/hoard/core/db/writer.py @@ -1,13 +1,15 @@ from __future__ import annotations import queue +import sqlite3 import threading +import time from dataclasses import dataclass from pathlib import Path from typing import Any, Callable from hoard.core.db.connection import connect -from hoard.core.db.lock import DatabaseWriteLock +from hoard.core.db.lock import DatabaseLockError, DatabaseWriteLock @dataclass @@ -26,13 +28,27 @@ class WriteCoordinator: (CLI commands, background sync, a second server) are also serialised. """ - def __init__(self, *, db_path: Path, busy_timeout_ms: int | None = None) -> None: + def __init__( + self, + *, + db_path: Path, + busy_timeout_ms: int | None = None, + lock_timeout_ms: int | None = None, + retry_budget_ms: int | None = None, + retry_backoff_ms: int | None = None, + ) -> None: self._queue: "queue.Queue[_WriteTask | None]" = queue.Queue() self._thread = threading.Thread(target=self._run, daemon=True) self._db_path = db_path self._busy_timeout_ms = busy_timeout_ms + self._lock_timeout_ms = _int_or_default(lock_timeout_ms, 30000) + self._retry_budget_ms = _int_or_default(retry_budget_ms, 30000) + self._retry_backoff_ms = min(_int_or_default(retry_backoff_ms, 50), 500) self._thread_id: int | None = None - self._lock = DatabaseWriteLock(db_path) + self._lock = DatabaseWriteLock( + db_path, + timeout_seconds=self._lock_timeout_ms / 1000, + ) self._thread.start() def submit(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: @@ -57,6 +73,23 @@ def _run(self) -> None: task = self._queue.get() if task is None: break + try: + self._run_task_with_retry(task) + except Exception as exc: + # Defensive safety net: report task failures and keep + # the writer loop alive for subsequent tasks. + task.error = exc + finally: + task.event.set() + + try: + self._conn.close() + except Exception: + pass + + def _run_task_with_retry(self, task: _WriteTask) -> None: + deadline = time.monotonic() + (self._retry_budget_ms / 1000) + while True: try: with self._lock: try: @@ -67,14 +100,29 @@ def _run(self) -> None: if self._conn.in_transaction: self._conn.rollback() task.error = exc + return except Exception as exc: - # Lock acquisition itself failed -- report to caller - # instead of killing the writer thread. - task.error = exc - finally: - task.event.set() + if not self._is_retryable(exc): + task.error = exc + return + remaining = deadline - time.monotonic() + if remaining <= 0: + task.error = exc + return + time.sleep(min(self._retry_backoff_ms / 1000, remaining)) - try: - self._conn.close() - except Exception: - pass + @staticmethod + def _is_retryable(exc: Exception) -> bool: + if isinstance(exc, DatabaseLockError): + return True + if isinstance(exc, sqlite3.OperationalError): + msg = str(exc).lower() + if "busy" in msg or "locked" in msg: + return True + return False + + +def _int_or_default(value: int | None, default: int) -> int: + if value is None: + return default + return value if value > 0 else default diff --git a/hoard/core/ingest/sync.py b/hoard/core/ingest/sync.py index 3373691..6e46638 100644 --- a/hoard/core/ingest/sync.py +++ b/hoard/core/ingest/sync.py @@ -2,6 +2,8 @@ from typing import Iterable +from hoard.core.db.write_exec import WriteSubmit, direct_submit +from hoard.core.ingest.registry import iter_enabled_connectors from hoard.core.ingest.store import ( get_entity_by_source, replace_chunks, @@ -9,15 +11,14 @@ upsert_entity, ) from hoard.core.models import SyncStats -from hoard.core.ingest.registry import iter_enabled_connectors from hoard.sdk.base import ConnectorV1 from hoard.sdk.types import EntityInput -def sync_connector( - conn, +def sync_connector_with_submit( connector: ConnectorV1, config: dict, + submit_write: WriteSubmit, on_entity: callable | None = None, ) -> SyncStats: stats = SyncStats() @@ -32,14 +33,10 @@ def sync_connector( try: _apply_provenance(connector, entity) - existing = get_entity_by_source(conn, entity.source, entity.source_id) - entity_id = upsert_entity(conn, entity) + unchanged, chunk_count = submit_write.submit(_upsert_entity_and_chunks, entity, chunks) seen_source_ids.append(entity.source_id) - - if existing and existing.get("content_hash") == entity.content_hash: - continue - - stats.chunks_written += replace_chunks(conn, entity_id, chunks) + if not unchanged: + stats.chunks_written += chunk_count except Exception: stats.errors += 1 continue @@ -48,14 +45,32 @@ def sync_connector( scan_failed = True if not scan_failed: - stats.entities_tombstoned = tombstone_missing( - conn, connector.source_name, seen_source_ids - ) - conn.commit() + try: + stats.entities_tombstoned = submit_write.submit( + _tombstone_missing, + connector.source_name, + seen_source_ids, + ) + except Exception: + stats.errors += 1 connector.cleanup() return stats +def sync_connector( + conn, + connector: ConnectorV1, + config: dict, + on_entity: callable | None = None, +) -> SyncStats: + return sync_connector_with_submit( + connector, + config, + submit_write=direct_submit(conn), + on_entity=on_entity, + ) + + def _apply_provenance(connector: ConnectorV1, entity: EntityInput) -> None: if not entity.connector_name: entity.connector_name = connector.name @@ -79,3 +94,15 @@ def run_sync(conn, *, config: dict, source: str | None = None) -> dict: "started_at": stats.started_at.isoformat(timespec="seconds"), } return results + + +def _upsert_entity_and_chunks(conn, entity: EntityInput, chunks: list) -> tuple[bool, int]: + existing = get_entity_by_source(conn, entity.source, entity.source_id) + entity_id = upsert_entity(conn, entity) + if existing and existing.get("content_hash") == entity.content_hash: + return True, 0 + return False, replace_chunks(conn, entity_id, chunks) + + +def _tombstone_missing(conn, source: str, seen_source_ids: Iterable[str]) -> int: + return tombstone_missing(conn, source, seen_source_ids) diff --git a/hoard/core/mcp/server.py b/hoard/core/mcp/server.py index 511db9b..b6e5661 100644 --- a/hoard/core/mcp/server.py +++ b/hoard/core/mcp/server.py @@ -2,6 +2,7 @@ import json import os +import threading import time import urllib.parse from datetime import datetime, timezone @@ -14,18 +15,19 @@ from hoard import __version__ from hoard.core.config import load_config, resolve_paths from hoard.core.db.connection import connect, ensure_sqlite_version -from hoard.core.db.lock import DatabaseLockError, ServerSingletonLock +from hoard.core.db.lock import ServerSingletonLock from hoard.core.db.writer import WriteCoordinator from hoard.core.mcp.tools import count_chunks, dispatch_tool, is_write_tool, tool_definitions from hoard.core.orchestrator.events import poll_events from hoard.core.security.agent_tokens import ensure_agent_from_config -from hoard.migrations import get_current_version, get_migrations, migrate from hoard.core.security.audit import log_access from hoard.core.security.auth import TokenInfo, authenticate_token from hoard.core.security.errors import AuthError, ScopeError -from hoard.core.security.limits import RateLimitError, RateLimiter +from hoard.core.security.limits import RateLimiter, RateLimitError from hoard.core.security.server_secret import require_server_secret, resolve_server_secret +from hoard.core.sync.background import BackgroundSync from hoard.core.worker import Worker +from hoard.migrations import get_current_version, get_migrations, migrate SUPPORTED_PROTOCOL_VERSIONS = ["2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05"] @@ -328,14 +330,20 @@ def log_message(self, format: str, *args) -> None: # noqa: A003 class MCPServer(ThreadingHTTPServer): - def __init__(self, server_address, RequestHandlerClass, config_path: Path | None) -> None: + def __init__(self, server_address, request_handler_class, config_path: Path | None) -> None: self.config = load_config(config_path) paths = resolve_paths(self.config, config_path) self.db_path = paths.db_path - busy_timeout = int(self.config.get("write", {}).get("database", {}).get("busy_timeout_ms", 0) or 0) - self.writer = WriteCoordinator(db_path=self.db_path, busy_timeout_ms=busy_timeout or None) + db_cfg = self.config.get("write", {}).get("database", {}) + self.writer = WriteCoordinator( + db_path=self.db_path, + busy_timeout_ms=_int_or_default(db_cfg.get("busy_timeout_ms"), 5000), + lock_timeout_ms=_int_or_default(db_cfg.get("lock_timeout_ms"), 30000), + retry_budget_ms=_int_or_default(db_cfg.get("retry_budget_ms"), 30000), + retry_backoff_ms=min(_int_or_default(db_cfg.get("retry_backoff_ms"), 50), 500), + ) self.worker = Worker(db_path=self.db_path, config=self.config, writer=self.writer) - super().__init__(server_address, RequestHandlerClass) + super().__init__(server_address, request_handler_class) def _check_no_other_server(db_path: Path) -> ServerSingletonLock: @@ -356,32 +364,60 @@ def run_server( port: int = 19850, config_path: Path | None = None, no_migrate: bool = False, + stop_event: threading.Event | None = None, ) -> None: server = MCPServer((host, port), MCPRequestHandler, config_path) + background = BackgroundSync( + config=server.config, + config_path=config_path, + log=click.echo, + write_submit=server.writer, + ) + + server_lock: ServerSingletonLock | None = None + worker_started = False + background_started = False + try: + # Hold an exclusive advisory lock for the server's entire lifetime so + # that a second ``hoard serve`` on the same DB is rejected immediately. + server_lock = _check_no_other_server(server.db_path) - # Hold an exclusive advisory lock for the server's entire lifetime so - # that a second ``hoard serve`` on the same DB is rejected immediately. - _server_lock = _check_no_other_server(server.db_path) # noqa: F841 + conn = connect(server.db_path) + try: + ensure_sqlite_version() + _require_server_secret(server.config) + if no_migrate: + migrations = get_migrations() + latest = max(migrations.keys()) if migrations else 0 + current = get_current_version(conn) + if current < latest: + click.echo(f"⚠️ Schema migrations pending (v{current} → v{latest})") + click.echo(" Run 'hoard db migrate' or restart without --no-migrate") + else: + migrate(conn, app_version=__version__) + _bootstrap_tokens(conn, server.config) + finally: + conn.close() - conn = connect(server.db_path) - try: - ensure_sqlite_version() - _require_server_secret(server.config) - if no_migrate: - migrations = get_migrations() - latest = max(migrations.keys()) if migrations else 0 - current = get_current_version(conn) - if current < latest: - click.echo(f"⚠️ Schema migrations pending (v{current} → v{latest})") - click.echo(" Run 'hoard db migrate' or restart without --no-migrate") + server.worker.start() + worker_started = True + background.start() + background_started = True + if stop_event is None: + server.serve_forever() else: - migrate(conn, app_version=__version__) - _bootstrap_tokens(conn, server.config) + server.timeout = 0.2 + while not stop_event.is_set(): + server.handle_request() finally: - conn.close() - - server.worker.start() - server.serve_forever() + if background_started: + background.stop() + if worker_started: + server.worker.stop() + server.writer.stop() + server.server_close() + if server_lock is not None: + server_lock.release() def _wrap_tool_result(response: Dict[str, Any]) -> Dict[str, Any]: @@ -433,3 +469,11 @@ def _require_event_read(token: TokenInfo) -> None: if "event.read" in token.scopes: return raise ScopeError("Missing scopes: event.read") + + +def _int_or_default(value: Any, default: int) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + return default + return parsed if parsed > 0 else default diff --git a/hoard/core/memory/store.py b/hoard/core/memory/store.py index fa0cdbc..68666f2 100644 --- a/hoard/core/memory/store.py +++ b/hoard/core/memory/store.py @@ -81,7 +81,6 @@ def memory_put( """, (entry_id, key, content, tags_json, tags_text, metadata_json, now, now, expires_value), ) - conn.commit() return { "id": entry_id, @@ -233,7 +232,6 @@ def memory_prune(conn) -> int: "DELETE FROM memory_entries WHERE expires_at IS NOT NULL AND expires_at <= ?", (now,), ) - conn.commit() return cursor.rowcount diff --git a/hoard/core/sync/background.py b/hoard/core/sync/background.py index 0b8ce32..53a2532 100644 --- a/hoard/core/sync/background.py +++ b/hoard/core/sync/background.py @@ -1,10 +1,10 @@ from __future__ import annotations import threading -import time from pathlib import Path from typing import Callable, Optional +from hoard.core.db.write_exec import WriteSubmit from hoard.core.sync.service import run_sync_with_lock from hoard.core.sync.watcher import WATCHDOG_AVAILABLE, SyncWatcher, build_watch_targets @@ -16,12 +16,15 @@ def __init__( config: dict, config_path: Optional[Path] = None, log: Optional[Callable[[str], None]] = None, + write_submit: WriteSubmit | None = None, ) -> None: self._config = config self._config_path = config_path self._log = log or (lambda msg: None) + self._write_submit = write_submit self._scheduler_thread: Optional[threading.Thread] = None self._watcher: Optional[SyncWatcher] = None + self._stop = threading.Event() def start(self) -> None: interval = int(self._config.get("sync", {}).get("interval_minutes", 0) or 0) @@ -54,10 +57,29 @@ def start(self) -> None: self._watcher.start() self._log("File watcher enabled.") + def stop(self) -> None: + self._stop.set() + if self._watcher: + self._watcher.stop() + self._watcher = None + if self._scheduler_thread and self._scheduler_thread.is_alive(): + self._scheduler_thread.join(timeout=2) + def _schedule_loop(self, interval_minutes: int) -> None: - while True: - time.sleep(interval_minutes * 60) - run_sync_with_lock(self._config, self._config_path, source=None) + while not self._stop.wait(interval_minutes * 60): + run_sync_with_lock( + self._config, + self._config_path, + source=None, + write_submit=self._write_submit, + ) def _sync_source(self, source: str) -> None: - run_sync_with_lock(self._config, self._config_path, source=source) + if self._stop.is_set(): + return + run_sync_with_lock( + self._config, + self._config_path, + source=source, + write_submit=self._write_submit, + ) diff --git a/hoard/core/sync/service.py b/hoard/core/sync/service.py index 8919046..fe818c2 100644 --- a/hoard/core/sync/service.py +++ b/hoard/core/sync/service.py @@ -7,16 +7,17 @@ from typing import Any, Dict, List, Optional from hoard.core.config import default_data_path, resolve_paths -from hoard.core.db.connection import connect, initialize_db +from hoard.core.db.connection import initialize_db +from hoard.core.db.write_exec import WriteSubmit, direct_submit, temporary_coordinator_submit from hoard.core.ingest.registry import iter_enabled_connectors -from hoard.core.ingest.sync import sync_connector +from hoard.core.ingest.sync import sync_connector_with_submit from hoard.core.memory.store import memory_prune from hoard.core.models import SyncStats def sync_connectors( - conn, config: dict, + write_submit: WriteSubmit, source: Optional[str] = None, on_entity: Optional[callable] = None, ) -> Dict[str, Any]: @@ -38,7 +39,12 @@ def sync_connectors( ) continue - stats = sync_connector(conn, connector, settings, on_entity=on_entity) + stats = sync_connector_with_submit( + connector, + settings, + submit_write=write_submit, + on_entity=on_entity, + ) results.append( { "source": name, @@ -50,7 +56,7 @@ def sync_connectors( pruned = 0 if config.get("memory", {}).get("prune_on_sync", True): - pruned = memory_prune(conn) + pruned = write_submit.submit(memory_prune) return {"connectors": results, "memory_pruned": pruned} @@ -60,24 +66,28 @@ def run_sync_with_lock( config_path: Optional[Path] = None, source: Optional[str] = None, lock_path: Optional[Path] = None, + write_submit: WriteSubmit | None = None, ) -> Dict[str, Any]: - """Run a sync, acquiring the file-based sync lock to prevent overlaps. - - NOTE: This function is called both from the CLI (standalone process) and - from ``BackgroundSync`` (inside the ``hoard serve`` process). It must - **not** hold a ``DatabaseWriteLock`` for the entire sync duration because - that would deadlock the ``WriteCoordinator`` running in the same process - (``flock`` is per-open-file-description, not per-process). Instead we - rely on SQLite WAL + ``busy_timeout`` for write serialisation here, and - the ``WriteCoordinator``'s per-task flock for MCP writes. - """ + """Run a sync, acquiring the file-based sync lock to prevent overlaps.""" + if write_submit is not None: + write_submit.submit(initialize_db) + return _sync_with_lock_submit( + config, + write_submit=write_submit, + source=source, + lock_path=lock_path, + ) + paths = resolve_paths(config, config_path) - conn = connect(paths.db_path) - initialize_db(conn) - try: - return sync_with_lock(conn, config, source=source, lock_path=lock_path) - finally: - conn.close() + db_cfg = config.get("write", {}).get("database", {}) + with temporary_coordinator_submit(paths.db_path, db_cfg) as temp_submit: + temp_submit.submit(initialize_db) + return _sync_with_lock_submit( + config, + write_submit=temp_submit, + source=source, + lock_path=lock_path, + ) def sync_with_lock( @@ -85,12 +95,26 @@ def sync_with_lock( config: dict, source: Optional[str] = None, lock_path: Optional[Path] = None, +) -> Dict[str, Any]: + return _sync_with_lock_submit( + config, + write_submit=direct_submit(conn), + source=source, + lock_path=lock_path, + ) + + +def _sync_with_lock_submit( + config: dict, + write_submit: WriteSubmit, + source: Optional[str] = None, + lock_path: Optional[Path] = None, ) -> Dict[str, Any]: lock_path = lock_path or _lock_path() if not _acquire_lock(lock_path): return {"skipped": True, "reason": "lock"} try: - return sync_connectors(conn, config, source=source) + return sync_connectors(config, write_submit, source=source) finally: _release_lock(lock_path) diff --git a/tests/conftest.py b/tests/conftest.py index 3a8bf0a..c9cd8ab 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -34,3 +34,30 @@ def _start(config_path: Path) -> str: server.shutdown() thread.join(timeout=2) server.server_close() + server.writer.stop() + + +@pytest.fixture +def mcp_server_with_instance() -> Callable[[Path], tuple[str, MCPServer]]: + servers: List[Tuple[MCPServer, threading.Thread]] = [] + + def _start(config_path: Path) -> tuple[str, MCPServer]: + config = load_config(config_path) + paths = resolve_paths(config, config_path) + conn = connect(paths.db_path) + initialize_db(conn) + conn.close() + server = MCPServer(("127.0.0.1", 0), MCPRequestHandler, config_path) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + servers.append((server, thread)) + url = f"http://127.0.0.1:{server.server_address[1]}/mcp" + return url, server + + yield _start + + for server, thread in servers: + server.shutdown() + thread.join(timeout=2) + server.server_close() + server.writer.stop() diff --git a/tests/test_cli.py b/tests/test_cli.py index c3fa250..c467716 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -61,6 +61,7 @@ def test_cli_search_json_includes_memory(tmp_path: Path) -> None: conn = connect(db_path) initialize_db(conn) memory_put(conn, key="context", content="Hoard memory entry") + conn.commit() conn.close() runner = CliRunner() @@ -85,6 +86,7 @@ def test_cli_memory_prune(tmp_path: Path) -> None: conn = connect(db_path) initialize_db(conn) memory_put(conn, key="expired", content="old", expires_at="2000-01-01T00:00:00") + conn.commit() conn.close() runner = CliRunner() diff --git a/tests/test_write_lock.py b/tests/test_write_lock.py index cffb25d..5c287b6 100644 --- a/tests/test_write_lock.py +++ b/tests/test_write_lock.py @@ -6,26 +6,36 @@ - ServerSingletonLock prevents two servers on the same DB - WriteCoordinator acquires/releases the lock per write - WriteCoordinator resilience when lock is contended +- WriteCoordinator retries under transient contention +- WriteCoordinator fails deterministically after retry budget - write_locked() context manager -- CLI and sync paths hold the lock during writes +- Background sync + MCP writes serialize without lock errors +- CLI sync and live server writes serialize safely +- Server singleton lifecycle releases lock on shutdown +- Transaction ownership via submit helpers """ from __future__ import annotations import json +import socket import threading import time +import urllib.error import urllib.request from pathlib import Path import pytest +from click.testing import CliRunner +from hoard.cli.main import cli from hoard.core.config import save_config from hoard.core.db.connection import connect, initialize_db, write_locked from hoard.core.db.lock import DatabaseLockError, DatabaseWriteLock, ServerSingletonLock +from hoard.core.db.write_exec import direct_submit from hoard.core.db.writer import WriteCoordinator -from hoard.core.memory.store import memory_put - +from hoard.core.mcp.server import run_server +from hoard.core.sync.background import BackgroundSync # --------------------------------------------------------------------------- # DatabaseWriteLock unit tests @@ -220,6 +230,86 @@ def _insert(conn): assert row["content"] == "survived" +def test_write_coordinator_retries_and_succeeds(tmp_path: Path) -> None: + db_path = tmp_path / "test.db" + conn = connect(db_path) + initialize_db(conn) + conn.close() + + writer = WriteCoordinator( + db_path=db_path, + lock_timeout_ms=100, + retry_budget_ms=1500, + retry_backoff_ms=50, + ) + external_lock = DatabaseWriteLock(db_path, timeout_seconds=5) + external_lock.acquire() + + def _release_later() -> None: + time.sleep(0.3) + external_lock.release() + + releaser = threading.Thread(target=_release_later) + releaser.start() + + def _insert(conn): + conn.execute( + "INSERT OR REPLACE INTO memory_entries " + "(id, key, content, tags, tags_text, metadata, created_at, updated_at) " + "VALUES ('m2', 'k2', 'retried', '', '', '{}', datetime('now'), datetime('now'))" + ) + + writer.submit(_insert) + releaser.join(timeout=2) + writer.stop() + + conn = connect(db_path) + row = conn.execute("SELECT content FROM memory_entries WHERE key = 'k2'").fetchone() + conn.close() + assert row is not None + assert row["content"] == "retried" + + +def test_write_coordinator_retry_budget_exhausted(tmp_path: Path) -> None: + db_path = tmp_path / "test.db" + conn = connect(db_path) + initialize_db(conn) + conn.close() + + writer = WriteCoordinator( + db_path=db_path, + lock_timeout_ms=50, + retry_budget_ms=150, + retry_backoff_ms=25, + ) + external_lock = DatabaseWriteLock(db_path, timeout_seconds=5) + external_lock.acquire() + + with pytest.raises(DatabaseLockError): + writer.submit(lambda conn: None) + + external_lock.release() + writer.stop() + + +def test_direct_submit_commits_memory_put(tmp_path: Path) -> None: + from hoard.core.memory.store import memory_get, memory_put + + db_path = tmp_path / "test.db" + conn = connect(db_path) + initialize_db(conn) + submitter = direct_submit(conn) + + submitter.submit(memory_put, key="ctx", content="value") + conn.close() + + conn = connect(db_path) + row = memory_get(conn, "ctx") + conn.close() + assert row is not None + assert row["content"] == "value" + + # --------------------------------------------------------------------------- # write_locked() context manager # --------------------------------------------------------------------------- @@ -237,7 +327,6 @@ def test_write_locked_context_manager(tmp_path: Path) -> None: "(id, key, content, tags, tags_text, metadata, created_at, updated_at) " "VALUES ('m1', 'k1', 'v1', '', '', '{}', datetime('now'), datetime('now'))" ) - conn.commit() # Verify the write persisted conn = connect(db_path) @@ -264,6 +353,49 @@ def _call_mcp(url: str, token: str, method: str, params: dict) -> tuple[int, dic return resp.status, json.loads(resp.read()) +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +def _wait_for_health(port: int, timeout_seconds: float = 10) -> None: + deadline = time.time() + timeout_seconds + url = f"http://127.0.0.1:{port}/health" + while time.time() < deadline: + try: + with urllib.request.urlopen(url, timeout=1) as resp: + if resp.status in {200, 503}: + return + except Exception: + time.sleep(0.05) + raise AssertionError(f"Server did not become healthy on port {port}") + + +def _start_run_server( + *, + config_path: Path, + port: int, +) -> tuple[threading.Event, threading.Thread, list[Exception]]: + stop_event = threading.Event() + errors: list[Exception] = [] + + def _target() -> None: + try: + run_server( + host="127.0.0.1", + port=port, + config_path=config_path, + stop_event=stop_event, + ) + except Exception as exc: # pragma: no cover - diagnostic path + errors.append(exc) + + thread = threading.Thread(target=_target, daemon=True) + thread.start() + return stop_event, thread, errors + + def test_http_mcp_concurrent_writes(tmp_path: Path, mcp_server) -> None: """Multiple agents writing memories concurrently must all succeed.""" db_path = tmp_path / "hoard.db" @@ -337,3 +469,245 @@ def writer(thread_id: int) -> None: count = conn.execute("SELECT COUNT(*) FROM memories").fetchone()[0] conn.close() assert count == writes_per_thread * num_threads + + +def test_background_sync_and_http_writes_do_not_lock( + tmp_path: Path, + monkeypatch, + mcp_server_with_instance, +) -> None: + monkeypatch.setenv("HOARD_DATA_DIR", str(tmp_path / "hoard-data")) + + data_dir = tmp_path / "data" + data_dir.mkdir() + for idx in range(40): + (data_dir / f"note-{idx}.md").write_text(f"content {idx}") + + db_path = tmp_path / "hoard.db" + secret_path = tmp_path / "server.key" + secret_path.write_text("test-secret\n") + config_path = tmp_path / "config.yaml" + config = { + "security": { + "tokens": [ + { + "name": "agent", + "token": "hoard_sk_agent", + "scopes": ["search", "get", "memory", "sync", "ingest"], + } + ] + }, + "storage": {"db_path": str(db_path)}, + "connectors": { + "local_files": { + "enabled": True, + "paths": [str(data_dir)], + "include_extensions": [".md"], + "chunk_max_tokens": 50, + "chunk_overlap_tokens": 0, + } + }, + "vectors": {"enabled": False}, + "write": { + "enabled": True, + "server_secret_file": str(secret_path), + "auto_generate_server_secret": False, + "embeddings": {"enabled": False}, + "database": { + "busy_timeout_ms": 5000, + "lock_timeout_ms": 5000, + "retry_budget_ms": 5000, + "retry_backoff_ms": 25, + }, + }, + } + save_config(config, config_path) + + conn = connect(db_path) + initialize_db(conn) + conn.close() + + url, server = mcp_server_with_instance(config_path) + bg = BackgroundSync( + config=config, + config_path=config_path, + write_submit=server.writer, + ) + + sync_thread = threading.Thread(target=bg._sync_source, args=("local_files",), daemon=True) + sync_thread.start() + + errors: list[str] = [] + lock = threading.Lock() + + def _write_worker() -> None: + for idx in range(30): + try: + _, resp = _call_mcp( + url, + "hoard_sk_agent", + "tools/call", + { + "name": "memory_write", + "arguments": { + "content": f"from writer {idx}", + "slot": f"fact:bg.{idx}", + }, + }, + ) + if "error" in resp: + with lock: + errors.append(str(resp["error"])) + except Exception as exc: # pragma: no cover - diagnostic + with lock: + errors.append(str(exc)) + + write_threads = [threading.Thread(target=_write_worker, daemon=True) for _ in range(2)] + for thread in write_threads: + thread.start() + for thread in write_threads: + thread.join() + sync_thread.join(timeout=20) + + assert not [err for err in errors if "locked" in err.lower()], f"Unexpected lock errors: {errors}" + + +def test_cli_sync_and_live_server_writes_serialize(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("HOARD_DATA_DIR", str(tmp_path / "hoard-data")) + + data_dir = tmp_path / "data" + data_dir.mkdir() + for idx in range(20): + (data_dir / f"doc-{idx}.md").write_text(f"sync doc {idx}") + + db_path = tmp_path / "hoard.db" + secret_path = tmp_path / "server.key" + secret_path.write_text("test-secret\n") + config_path = tmp_path / "config.yaml" + config = { + "security": { + "tokens": [ + { + "name": "agent", + "token": "hoard_sk_agent", + "scopes": ["search", "get", "memory", "sync", "ingest"], + } + ] + }, + "storage": {"db_path": str(db_path)}, + "connectors": { + "local_files": { + "enabled": True, + "paths": [str(data_dir)], + "include_extensions": [".md"], + "chunk_max_tokens": 50, + "chunk_overlap_tokens": 0, + } + }, + "vectors": {"enabled": False}, + "write": { + "enabled": True, + "server_secret_file": str(secret_path), + "auto_generate_server_secret": False, + "embeddings": {"enabled": False}, + "database": { + "busy_timeout_ms": 5000, + "lock_timeout_ms": 5000, + "retry_budget_ms": 5000, + "retry_backoff_ms": 25, + }, + }, + } + save_config(config, config_path) + + conn = connect(db_path) + initialize_db(conn) + conn.close() + + port = _free_port() + stop_event, server_thread, server_errors = _start_run_server(config_path=config_path, port=port) + _wait_for_health(port) + url = f"http://127.0.0.1:{port}/mcp" + + write_errors: list[str] = [] + write_lock = threading.Lock() + + def _server_writer() -> None: + for idx in range(25): + try: + _, resp = _call_mcp( + url, + "hoard_sk_agent", + "tools/call", + { + "name": "memory_write", + "arguments": { + "content": f"cli-race {idx}", + "slot": f"fact:cli.{idx}", + }, + }, + ) + if "error" in resp: + with write_lock: + write_errors.append(str(resp["error"])) + except Exception as exc: # pragma: no cover - diagnostic + with write_lock: + write_errors.append(str(exc)) + + writer_thread = threading.Thread(target=_server_writer, daemon=True) + writer_thread.start() + + runner = CliRunner() + sync_result = runner.invoke(cli, ["sync", "--config", str(config_path)]) + + writer_thread.join(timeout=20) + stop_event.set() + server_thread.join(timeout=5) + + assert sync_result.exit_code == 0, sync_result.output + assert not server_errors, f"Server errors: {server_errors}" + assert not [err for err in write_errors if "locked" in err.lower()], write_errors + + +def test_server_singleton_lock_released_after_shutdown(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("HOARD_DATA_DIR", str(tmp_path / "hoard-data")) + + db_path = tmp_path / "hoard.db" + secret_path = tmp_path / "server.key" + secret_path.write_text("test-secret\n") + config_path = tmp_path / "config.yaml" + config = { + "security": {"tokens": []}, + "storage": {"db_path": str(db_path)}, + "connectors": {"local_files": {"enabled": False}}, + "vectors": {"enabled": False}, + "write": { + "enabled": True, + "server_secret_file": str(secret_path), + "auto_generate_server_secret": False, + "embeddings": {"enabled": False}, + }, + } + save_config(config, config_path) + + conn = connect(db_path) + initialize_db(conn) + conn.close() + + port_a = _free_port() + stop_a, thread_a, errors_a = _start_run_server(config_path=config_path, port=port_a) + _wait_for_health(port_a) + + with pytest.raises(DatabaseLockError): + run_server(host="127.0.0.1", port=_free_port(), config_path=config_path, stop_event=threading.Event()) + + stop_a.set() + thread_a.join(timeout=5) + assert not errors_a + + port_b = _free_port() + stop_b, thread_b, errors_b = _start_run_server(config_path=config_path, port=port_b) + _wait_for_health(port_b) + stop_b.set() + thread_b.join(timeout=5) + assert not errors_b