83 lines
3.6 KiB
Python
83 lines
3.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""echo_concurrency.py — H3: make the multi-writer story actually safe. [v1.0 SCAFFOLD — not yet wired]
|
||
|
|
|
||
|
|
The plugin advertises a shared vault (Claude Code + CoWork), but two real races exist:
|
||
|
|
1. The entity index is a full-file load -> mutate -> PUT (echo_index.load/save) with
|
||
|
|
NO lock. Two concurrent `capture` calls both read the old index; last PUT wins;
|
||
|
|
the other entity is silently dropped from the index (the note file survives, but
|
||
|
|
resolve/recall miss it until the next sweep).
|
||
|
|
2. The advisory lock is MANUAL (echo.py cmd_lock) — capture / scope set never take it;
|
||
|
|
it relies on the model remembering to.
|
||
|
|
|
||
|
|
This module provides (a) a context manager that auto-acquires/releases the existing
|
||
|
|
advisory lock around a write burst, with crash-safe release, and (b) an atomic
|
||
|
|
index-update that re-reads UNDER the lock before saving, so concurrent captures merge
|
||
|
|
instead of clobber.
|
||
|
|
|
||
|
|
ACCEPTANCE (see ROADMAP-1.0.md > H3):
|
||
|
|
- two concurrent captures both land in the index (no lost entries)
|
||
|
|
- lock auto-released on normal AND error exit
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import contextlib
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
|
import echo # noqa: E402
|
||
|
|
import echo_index as idx_mod # noqa: E402
|
||
|
|
|
||
|
|
|
||
|
|
def auto_owner() -> str:
|
||
|
|
"""A stable-per-process owner id: client tag + pid. Avoids Math.random/time-based
|
||
|
|
ids; pid is unique enough for cooperative locking and is reproducible in tests."""
|
||
|
|
tag = os.environ.get("ECHO_CLIENT", "cc")
|
||
|
|
return f"{tag}-{os.getpid()}"
|
||
|
|
|
||
|
|
|
||
|
|
@contextlib.contextmanager
|
||
|
|
def vault_lock(owner: str | None = None, required: bool = False):
|
||
|
|
"""Hold the advisory lock for a write burst; release on exit (incl. exceptions).
|
||
|
|
|
||
|
|
Yields `held: bool`. The lock is cooperative (read-back-confirmed, TTL-reclaimable),
|
||
|
|
so if another session holds it we do NOT block: with `required=False` we yield
|
||
|
|
`held=False` and let the caller proceed (advisory) or warn; with `required=True` we
|
||
|
|
raise. Release is best-effort and never masks the body's own exception.
|
||
|
|
"""
|
||
|
|
owner = owner or auto_owner()
|
||
|
|
rc = echo.cmd_lock(owner, quiet=True)
|
||
|
|
held = rc == 0
|
||
|
|
if not held and required:
|
||
|
|
raise echo.EchoError(f"vault_lock: could not acquire lock for '{owner}' (held elsewhere)", 75)
|
||
|
|
try:
|
||
|
|
yield held
|
||
|
|
finally:
|
||
|
|
if held:
|
||
|
|
try:
|
||
|
|
echo.cmd_unlock(owner, quiet=True)
|
||
|
|
except Exception: # noqa: BLE001 — release is best-effort; TTL reclaims a leak
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def atomic_index_update(mutate, owner: str | None = None):
|
||
|
|
"""Apply `mutate(index)` to the entity index without losing a concurrent writer's
|
||
|
|
entry: take the lock, **re-read the index fresh** (not a stale in-memory copy),
|
||
|
|
run the mutator, save, release. Returns the freshly-read-and-saved index.
|
||
|
|
|
||
|
|
The fresh re-read inside the critical section is the core of the fix — even if the
|
||
|
|
advisory lock can't be taken (another session active), reading immediately before
|
||
|
|
the save collapses the read-modify-write window to two calls instead of spanning the
|
||
|
|
whole capture, so an unlocked update is still far safer than the prior code.
|
||
|
|
"""
|
||
|
|
with vault_lock(owner) as held:
|
||
|
|
index = idx_mod.load()
|
||
|
|
mutate(index)
|
||
|
|
idx_mod.save(index)
|
||
|
|
if not held:
|
||
|
|
print("echo_concurrency: entity index updated WITHOUT the advisory lock "
|
||
|
|
"(another session may be active); the fresh re-read minimizes the race.",
|
||
|
|
file=sys.stderr)
|
||
|
|
return index
|