diff --git a/ANALYSIS_DB_AGENT_LOCKING.md b/ANALYSIS_DB_AGENT_LOCKING.md new file mode 100644 index 0000000..5f39ce7 --- /dev/null +++ b/ANALYSIS_DB_AGENT_LOCKING.md @@ -0,0 +1,363 @@ +# 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 **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. **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. + +--- + +## 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. 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): + self._conn = connect(self._db_path, busy_timeout_ms=self._busy_timeout_ms) + while True: + task = self._queue.get() + try: + 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: + task.error = exc # lock failure -- report, don't crash + 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: 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. 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. + +--- + +## 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` + +```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 6: 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 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: + +### 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 8 (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 + │ + ┌─────────┼──────────────┐ + │ flock(LOCK_EX) on │ + │ hoard.db.lock │ + │ │ + │ Execute A's fn │ + │ Commit or Rollback │ + │ │ + │ flock(LOCK_UN) │ + └─────────┼──────────────┘ + │ + ┌─────────┼──────────────┐ + │ flock(LOCK_EX) on │ + │ hoard.db.lock │ + │ │ + │ Execute B's fn │ + │ Commit or Rollback │ + │ │ + │ flock(LOCK_UN) │ + └─────────┼──────────────┘ + │ + ▼ + 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 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) +``` + +--- + +## 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 | +|-----|----------|-------| +| **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. | 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 ad6119d..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,8 +29,14 @@ 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.db.connection import connect, initialize_db +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 from hoard.core.mcp.stdio import run_stdio @@ -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,41 +464,42 @@ 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) + 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 + target = target_version if target_version is not None else latest + console.print(f"📦 Applying schema migrations (v{current} → v{target})...") - 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 +594,27 @@ 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() + 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 [] + 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 +654,14 @@ 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() + 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.") @@ -1297,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) @@ -1333,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 b8c702a..9a6fb6c 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,40 @@ 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, + 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 + 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 + + 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: + 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..6249a56 --- /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 = 30.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 = 30.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( + "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 be05e77..812223e 100644 --- a/hoard/core/db/writer.py +++ b/hoard/core/db/writer.py @@ -1,11 +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 DatabaseLockError, DatabaseWriteLock @dataclass @@ -19,12 +23,32 @@ 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, + 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, + timeout_seconds=self._lock_timeout_ms / 1000, + ) self._thread.start() def submit(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: @@ -50,12 +74,10 @@ def _run(self) -> None: if task is None: break try: - task.result = task.fn(self._conn, *task.args, **task.kwargs) - if self._conn.in_transaction: - self._conn.commit() + self._run_task_with_retry(task) except Exception as exc: - if self._conn.in_transaction: - self._conn.rollback() + # Defensive safety net: report task failures and keep + # the writer loop alive for subsequent tasks. task.error = exc finally: task.event.set() @@ -64,3 +86,43 @@ def _run(self) -> None: 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: + 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 + return + except Exception as exc: + 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)) + + @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 ec82ef2..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,17 +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 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"] @@ -327,14 +330,33 @@ 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: + """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( @@ -342,28 +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) - - conn = connect(server.db_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: - 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") + # 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) + + 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() + + 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]: @@ -415,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 3376232..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,14 +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.""" + 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( @@ -75,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 new file mode 100644 index 0000000..5c287b6 --- /dev/null +++ b/tests/test_write_lock.py @@ -0,0 +1,713 @@ +"""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 +- WriteCoordinator resilience when lock is contended +- WriteCoordinator retries under transient contention +- WriteCoordinator fails deterministically after retry budget +- write_locked() context manager +- 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.mcp.server import run_server +from hoard.core.sync.background import BackgroundSync + +# --------------------------------------------------------------------------- +# 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 + + +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" + + +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 +# --------------------------------------------------------------------------- + + +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'))" + ) + + # 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 _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" + 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 + + +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