diff --git a/AGENTS.md b/AGENTS.md index 2bb33f4..1aa5ca4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,22 +30,29 @@ replication (`mesh.py`, `merge.py`). Know which subsystem you are in. and the caller falls back to the older, dumber path. A missing model, a corrupt index, a locked database, a missing FTS5 build — all degrade search; none may break it. Test the failure branch, not just the happy one. -3. **All note writes go through `OmiStore`** (`notes.upsert_note` for external +3. **Multi-note writes go through `omind.txn`.** One note at a time is already + atomic (same-dir temp + `os.replace`). Two or more is not, and a crash + between them used to leave partial state with no way back. Any new operation + that writes several notes must journal its pre-images through + `txn.Transaction` under `store.write_lock()`, or it reintroduces the gap + `omind recover` exists to close. Recovery refuses to overwrite a note edited + after the crash — that edit is newer than the pre-image. +4. **All note writes go through `OmiStore`** (`notes.upsert_note` for external writers) — flock, atomic rename, Lamport `Rev:` stamping, soft delete. Deletes archive (`Disabled: true`); only `omind mesh purge` truly removes. -4. **`OmiStore.safe_name` guards every read and write.** Path traversal must stay +5. **`OmiStore.safe_name` guards every read and write.** Path traversal must stay impossible; there are tests enforcing this. Don't route around them. -5. **`store.py` stays framework-free.** No FastAPI, no MCP. The CLI and the web +6. **`store.py` stays framework-free.** No FastAPI, no MCP. The CLI and the web app both build on it. -6. **Credential notes are de-prioritised** in search and in the gate's +7. **Credential notes are de-prioritised** in search and in the gate's suggestions unless the query is itself about credentials (`retrieve._CREDENTIAL_PENALTY`). The gate must never steer an agent into the secrets notes. This is load-bearing, not decoration. -7. **No MCP tool may return unbounded output.** Every list-shaped tool pages +8. **No MCP tool may return unbounded output.** Every list-shaped tool pages (`limit`, `offset`, `total`, `has_more`) via `server._page`. `list-notes` once returned ~90,800 tokens in a single result; `tests/test_server.py:: test_every_list_tool_is_bounded` exists so a new tool cannot regress that. -8. **Reserved files are not memories.** `index.md` and `Memory Template.md` are +9. **Reserved files are not memories.** `index.md` and `Memory Template.md` are scaffolding; reading one does *not* clear the consult gate (`paths.NON_CONSULT_FILENAMES` — an anti-dodge measure, issue #109). diff --git a/BACKLOG.md b/BACKLOG.md index 73deec3..2596a35 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -66,24 +66,14 @@ quantized int8 vectors + RRF beats their JSON BM25 index), on multi-machine repl (they have none), on enforcement, and on shipping a real MCP server. What follows are the five places their design is genuinely better and the idea transfers._ -- [ ] **Journaled plan→apply→recover transactions for multi-note operations** ([#194](https://github.com/CryptoJones/omind/issues/194)) — _enhancement (durability)_ — - omind has atomic per-file replace, an advisory write lock, and version - preconditions, but no journal and no rollback, so an interrupted multi-note - operation leaves partial state. `store.create_and_disable_sources` concedes it in - its own docstring ("a process crash can still leave extra recoverable copies"). - Generalize what `omind consolidate` already prototypes into a store-level - primitive plus `omind recover`. Skip their `approved_plan_sha256` handshake. -- [x] **Typed confidence + symmetric `Conflicts with:` provenance** ([#195](https://github.com/CryptoJones/omind/issues/195)) — _enhancement (memory shape)_ — - shipped as two optional note fields that round-trip through Markdown, CLI, MCP, and - mesh merge like `Supersedes:`. A conflict binds both notes even when one side declared - it; retrieval surfaces the disagreement rather than resolving it; lint flags broken and - one-sided claims. No research-grade ledger. Original description follows. - their claim ledger types authority, assessment, confidence, and evidence relation - (`supports`/`contradicts`/`context`), keeping contradictions visible. omind's - `references:` is free text and `Supersedes:` only expresses clean ordered - replacement — there is no way to say "these two memories disagree" or "this was - never verified". Add two optional fields that round-trip like `Supersedes:` does; - do not grow a research-grade ledger. +- [x] **Journaled plan→apply→recover transactions for multi-note operations** ([#194](https://github.com/CryptoJones/omind/issues/194)) — _enhancement (durability)_ — + shipped as `omind.txn` + `omind recover`. Pre-images captured and fsynced before the + first write, atomic per-file replace, a commit record, deterministic rollback. Recovery + refuses to overwrite a note edited after the crash — that edit is newer than the + pre-image — reporting a conflict and keeping the journal instead. In-process failures + roll themselves back. `create_and_disable_sources` migrated; the docstring that conceded + "a process crash can still leave extra recoverable copies" is gone. Skipped their + `approved_plan_sha256` handshake as planned. - [ ] **Machine-readable capability contract verified by `doctor`** ([#196](https://github.com/CryptoJones/omind/issues/196)) — _hardening_ — they declare every capability's tier, read/write scope, network need, and destructiveness in `config/capabilities.json`, verify it, and state explicitly diff --git a/CHANGELOG.md b/CHANGELOG.md index b8ba1d6..ee64151 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,64 @@ All notable changes to this project are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [8.0.0] - 2026-08-02 + +### Added +- **Journaled multi-note transactions and `omind recover`** + ([#194](https://github.com/CryptoJones/omind/issues/194)). Multi-file updates + cannot be truly atomic on the filesystems omind runs on, and this does not + pretend otherwise. It provides the contract that *can* be kept: every target's + pre-image is captured and fsynced before the first write, the writes go + through the store's existing atomic per-file replace, a commit record marks + the point of no return, and an interrupted run is rolled back deterministically. + + omind had the first half of this already — same-dir temp + `os.replace`, an + advisory write lock, version preconditions — and no journal, so an interrupted + multi-note operation left partial state with no recovery path. + `store.create_and_disable_sources` (the `omind consolidate --apply` write path) + conceded it in its own docstring: *"a process crash can still leave extra + recoverable copies."* That fails toward keeping data, but a human still had to + notice and reconcile by hand. + + `omind recover` rolls back anything that did not reach its commit record, and + is a no-op on a clean journal. `--dry-run` reports without touching the vault. + + **Recovery refuses to clobber a later edit.** A pre-image is restored only when + the file still holds either that pre-image (nothing to do) or exactly the bytes + the interrupted run intended to write (ours to undo). Anything else means + someone edited the note after the crash — newer information than the pre-image + — so it is reported as a conflict, left alone, and its journal is kept for + inspection. `omind recover` exits 1 when that happens. Blind rollback would be + data loss wearing recovery's clothes. + + An in-process failure rolls itself back, so `create_and_disable_sources` needs + no manual `recover` at all; the journal is for the case where the process dies. + + Pre-images are restored **byte for byte**, through a binary atomic write + rather than the store's text-mode one. Routing them through a text writer + re-translates line endings: on Windows a `b"old A\r\n"` pre-image came back as + `b"old A\r\r\n"`, so every rollback silently grew a blank line in the note it + was restoring. + + Content identity is hashed over line-ending-normalized text, not raw bytes. + `_atomic_write` writes in text mode, so on Windows every `\n` reaches the disk + as `\r\n` — hashing bytes meant a file never matched what had just been written + to it, every note read as a foreign edit, and recovery rolled back *nothing*. + Both Windows CI legs caught it before merge. + + Journals live in the state dir, never the vault: they describe this machine's + interrupted filesystem work, are meaningless to a mesh peer, and must not + replicate (invariant 1). + + Deliberately not copied from the source design: its `approved_plan_sha256` + handshake, which would put a hash-copying step in front of routine memory + writes. `consolidate` keeps plan/apply; everything else just gets the journal. + +### Changed +- `AGENTS.md` gains invariant 3: any new operation writing several notes must + journal through `txn.Transaction` under `store.write_lock()`, or it + reintroduces the gap `omind recover` exists to close. + ## [7.0.0] - 2026-08-02 _The last of the 2026-08-01 review and 2026-08-02 comparison backlogs, released diff --git a/README.md b/README.md index 7800b24..7729a32 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,12 @@ reads and writes as long-term memory. `omind` does two things with it: French, Arabic, Russian, Chinese), including right-to-left layout for Arabic. Its API is unauthenticated by design — the localhost bind is the security boundary. Read [docs/serve.md](docs/serve.md) before exposing the port. +- **`omind recover`** — roll back a multi-note write that was interrupted + mid-apply (a killed `consolidate --apply`, a power loss). Multi-file updates + can't be truly atomic, so omind journals every target's prior bytes before the + first write and can put them back. A note edited *after* the interruption is + reported as a conflict and left alone — that edit is newer than anything the + journal could restore. - **`omind doctor`** — diagnose the wiring in one shot: Claude CLI + git on `PATH`, the `omi` MCP server registered at user scope with the right command, the OMI folder readable, mesh health (node identity, merge driver, per-peer diff --git a/pyproject.toml b/pyproject.toml index f2533ae..ceaab6e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "omind" -version = "7.0.0" +version = "8.0.0" description = "Reproduce the OMI/Obsidian memory integration for AI agents, plus a local web app to view, edit, and add memory entries." readme = "README.md" requires-python = ">=3.10" diff --git a/src/omind/__init__.py b/src/omind/__init__.py index 0909da6..dde076d 100644 --- a/src/omind/__init__.py +++ b/src/omind/__init__.py @@ -2,4 +2,4 @@ # Copyright 2026 Aaron K. Clark """omind — OMI/Obsidian memory tooling for AI agents.""" -__version__ = "7.0.0" +__version__ = "8.0.0" diff --git a/src/omind/cli.py b/src/omind/cli.py index f516d43..c6fb0ef 100644 --- a/src/omind/cli.py +++ b/src/omind/cli.py @@ -66,7 +66,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--version", action="version", version=f"omind {__version__}") sub = parser.add_subparsers( dest="command", - metavar="{help,setup,quickstart,serve,doctor,self-update,backup,ai,export,import,reindex,note,rollup,hook}", + metavar="{help,setup,quickstart,serve,doctor,self-update,backup,ai,export,import,reindex,note,rollup,recover,hook}", ) help_p = sub.add_parser( @@ -477,6 +477,27 @@ def build_parser() -> argparse.ArgumentParser: for gp in (g_neighbors, g_path, g_orphans, g_dangling, g_stats, g_frontier, g_export): _add_vault_args(gp) + recover = sub.add_parser( + "recover", + help="roll back a multi-note write that was interrupted mid-apply", + description=( + "Roll back any journaled multi-note transaction that did not reach its\n" + "commit record — an `omind consolidate --apply` killed mid-write, a\n" + "power loss, an OOM. A no-op when the journal is clean, which is the\n" + "normal case.\n" + "\n" + "A note whose bytes match neither the pre-image nor what the interrupted\n" + "run intended to write was edited after the crash, and is reported as a\n" + "conflict and left alone: that edit is newer than anything this could\n" + "restore. Its journal is kept so you can inspect it." + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + recover.add_argument( + "--dry-run", action="store_true", help="list what would be rolled back, change nothing" + ) + _add_vault_args(recover) + checkpoint = sub.add_parser( "checkpoint", help="summarize recent activity (journal + compliance log) into a daily " @@ -1218,6 +1239,38 @@ def _run_graph(args: argparse.Namespace) -> int: return 0 +def _run_recover(args: argparse.Namespace) -> int: + """``omind recover``: roll back interrupted multi-note transactions.""" + from omind import txn + from omind.store import OmiStore + + omi_dir = (args.vault / args.folder).expanduser() + store = OmiStore(omi_dir) + if not txn.pending(omi_dir): + print("nothing to recover: no interrupted transactions") + return 0 + # Take the same write lock a normal write takes, so recovery cannot race a + # concurrent MCP/web/cron writer touching the very notes it is restoring. + with store.write_lock(): + reports = txn.recover(omi_dir, dry_run=args.dry_run) + conflicts = 0 + for report in reports: + print(("would roll back " if args.dry_run else "") + report.format()) + for name in report.conflicts: + print(f" CONFLICT: {name} changed after the interruption — left as-is") + conflicts += 1 + if conflicts: + print( + "\nSome notes were edited after the interrupted write. Their journals are kept;\n" + "inspect the notes, then delete the journal directory when you are satisfied.", + file=sys.stderr, + ) + return 1 + if not args.dry_run: + store.update_index() + return 0 + + def _run_checkpoint(args: argparse.Namespace) -> int: from omind import checkpoint @@ -1535,6 +1588,8 @@ def main(argv: list[str] | None = None) -> int: return _run_bench(args) if args.command == "lint": return _run_lint(args) + if args.command == "recover": + return _run_recover(args) if args.command == "graph": return _run_graph(args) if args.command == "checkpoint": diff --git a/src/omind/paths.py b/src/omind/paths.py index 0d9cf0a..7ac651f 100644 --- a/src/omind/paths.py +++ b/src/omind/paths.py @@ -111,3 +111,12 @@ def access_state_path(omi_dir: Path) -> Path: def consolidation_dir(omi_dir: Path) -> Path: """Machine-local proposal/draft storage for one vault's reviewed merges.""" return state_dir() / f"consolidate-{_omi_dir_digest(omi_dir)}" + + +def transaction_dir(omi_dir: Path) -> Path: + """Journal + pre-image storage for one vault's multi-note transactions. + + Machine-local and outside the vault: it records what the *filesystem* was + doing, is meaningless on another peer, and must never be mesh-synced. + """ + return state_dir() / f"txn-{_omi_dir_digest(omi_dir)}" diff --git a/src/omind/server.py b/src/omind/server.py index 360dfe1..23f9a2c 100644 --- a/src/omind/server.py +++ b/src/omind/server.py @@ -471,7 +471,7 @@ def _graph_query( if operation == "stats": return dict(graph.stats(graph_for())) if operation == "frontier": - # Paged like every other list-shaped result (invariant 7): the caller + # Paged like every other list-shaped result (invariant 8): the caller # asks for a page, the ranking is computed over the whole graph. ranked = [ { diff --git a/src/omind/store.py b/src/omind/store.py index 4b5ee2f..baa8030 100644 --- a/src/omind/store.py +++ b/src/omind/store.py @@ -27,7 +27,7 @@ import yaml -from omind import filelock +from omind import filelock, txn from omind.clock import Rev, next_rev from omind.paths import ( INDEX_FILENAME, @@ -1273,8 +1273,13 @@ def create_and_disable_sources( The source tuples are ``(filename, expected_version)``. All versions and the target's nonexistence are checked before the first write, closing the gap where another OmiStore writer could change the second source - between a separate create and two archive calls. A process crash can - still leave extra recoverable copies, never a hard-deleted source. + between a separate create and two archive calls. + + The writes run inside a journaled transaction (:mod:`omind.txn`), so an + interrupted apply is rolled back by ``omind recover`` instead of leaving + the vault half-merged. This used to concede here that "a process crash + can still leave extra recoverable copies" — it fails toward keeping + data, but a human still had to notice and reconcile by hand (#194). """ if not fields.title.strip(): raise NoteError("a note requires a title") @@ -1305,12 +1310,22 @@ def create_and_disable_sources( content = render_fields(fields) if self.node_id is not None: content = self._stamped(target, content) - _atomic_write(target, content) + transaction = txn.Transaction(self.omi_dir) + transaction.write(target, content) for path, _expected in source_paths: archived = _with_disabled(_read_text(path), True) if self.node_id is not None: archived = self._stamped(path, archived) - _atomic_write(path, archived) + transaction.write(path, archived) + transaction.prepare() + try: + transaction.apply(_atomic_write) + except BaseException: + # Put back what we managed to change before re-raising, so the + # in-process failure needs no `omind recover` at all. + transaction.rollback() + raise + transaction.commit() self._write_index() self._signal_write() return target.name diff --git a/src/omind/txn.py b/src/omind/txn.py new file mode 100644 index 0000000..77b5093 --- /dev/null +++ b/src/omind/txn.py @@ -0,0 +1,397 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Aaron K. Clark +"""Journaled multi-note transactions with deterministic recovery (#194). + +Multi-file updates cannot be truly atomic on the filesystems omind runs on. +This module does not pretend otherwise. What it provides instead is a contract +that can actually be kept: + +* every target's **pre-image** is captured and fsynced *before* the first write, +* the writes then happen through the store's existing atomic per-file replace, +* a **commit record** marks the point of no return, and +* an interrupted run is rolled back deterministically by :func:`recover`. + +So an interrupted apply lands in one of two states, never a third: fully +applied, or fully rolled back once ``omind recover`` runs. + +**What this fixes.** ``OmiStore`` already had same-dir temp + ``os.replace``, an +advisory write lock, and version preconditions — the first half. It had no +journal and no rollback, so an interrupted multi-note operation left partial +state with no recovery path. ``store.create_and_disable_sources`` conceded this +in its own docstring: *"a process crash can still leave extra recoverable +copies."* That failed toward keeping data, which is the right direction, but it +still meant a human had to notice and reconcile by hand. + +**The rule that keeps recovery safe.** Rollback restores a pre-image *only* +when the file on disk still holds either that pre-image (nothing to do) or +exactly the bytes this transaction intended to write (ours to undo). If it +holds anything else, someone edited the note after the crash, and their edit is +newer information than our pre-image. We refuse, report, and leave it alone. +Blind rollback would be data loss dressed up as recovery. + +The journal lives in the state dir, never the vault: it describes this +machine's interrupted filesystem work, is meaningless to a mesh peer, and must +not replicate. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import json +import os +import tempfile +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from omind import paths + +#: Journal record states. A record that is not ``COMMITTED`` when found is, by +#: definition, an interrupted run: the process died between preparing and +#: committing, because both transitions are fsynced. +PREPARED = "prepared" +COMMITTED = "committed" + + +def _sha(text: str) -> str: + """Identity of a note's *content*, independent of line-ending translation. + + ``_atomic_write`` writes in text mode, so on Windows every ``\n`` reaches + the disk as ``\r\n``. Hashing the raw bytes therefore never matched what we + had just written, every file looked like somebody else's edit, and recovery + refused to roll back anything at all — the feature was inert on Windows and + silent about it. Normalise, then hash. + """ + return hashlib.sha256(text.replace("\r\n", "\n").encode("utf-8")).hexdigest() + + +def _sha_bytes(data: bytes) -> str: + """Content identity for bytes read back off disk.""" + return _sha(data.decode("utf-8", errors="replace")) + + +def _fsync_path(path: Path) -> None: + """Best-effort fsync of a file, so a crash cannot lose the journal itself.""" + with contextlib.suppress(OSError): + fd = os.open(path, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +def _fsync_dir(directory: Path) -> None: + with contextlib.suppress(OSError, AttributeError): + fd = os.open(directory, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +def _atomic_write_bytes(path: Path, data: bytes) -> None: + """Restore exact bytes: same-dir temp + ``os.replace``, binary mode. + + Deliberately NOT the store's text-mode ``_atomic_write``. A pre-image is + bytes, and routing it through a text writer re-translates line endings — on + Windows, restoring ``b"old A\r\n"`` that way produced ``b"old A\r\r\n"``, + growing a blank line into the note on every rollback. Recovery must put back + exactly what was there, byte for byte. + """ + directory = path.parent + directory.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=directory, prefix=".tmp-recover-", suffix=".md") + try: + with os.fdopen(fd, "wb") as fh: + fh.write(data) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, path) + _fsync_dir(directory) + except BaseException: + with contextlib.suppress(OSError): + os.unlink(tmp) + raise + + +class TransactionError(Exception): + """A transaction could not be prepared, committed, or recovered.""" + + +@dataclass +class _Entry: + """One file this transaction will replace, and how to undo that.""" + + path: Path + #: Bytes to write. ``None`` means "this file should not exist afterwards", + #: which recovery reads as "delete it on rollback if we created it". + content: str + #: sha256 of the file's bytes before we touched it; "" when it did not exist. + prior_sha: str = "" + existed: bool = False + #: sha256 of what we intend to write — the proof, at recovery time, that a + #: file's current bytes are ours to undo rather than someone else's edit. + new_sha: str = "" + + def to_json(self) -> dict[str, Any]: + return { + "path": str(self.path), + "prior_sha": self.prior_sha, + "existed": self.existed, + "new_sha": self.new_sha, + } + + +@dataclass +class RecoveryReport: + """What one :func:`recover` pass did, per journal.""" + + transaction_id: str + restored: list[str] = field(default_factory=list) + removed: list[str] = field(default_factory=list) + skipped: list[str] = field(default_factory=list) + #: Files whose bytes match neither our pre-image nor what we wrote: edited + #: after the crash, so rolling them back would destroy newer work. + conflicts: list[str] = field(default_factory=list) + + @property + def clean(self) -> bool: + return not self.conflicts + + def format(self) -> str: + bits = [ + f"transaction {self.transaction_id}", + f"restored {len(self.restored)}", + f"removed {len(self.removed)}", + f"skipped {len(self.skipped)}", + ] + if self.conflicts: + bits.append(f"CONFLICTS {len(self.conflicts)}") + return ", ".join(bits) + + +class Transaction: + """Collects writes, journals their pre-images, then applies them. + + Not a general database transaction: there is no isolation and no concurrent + reader guarantee. It buys exactly one thing — that an interrupted multi-note + write can be put back the way it was. + + The caller MUST hold ``OmiStore.write_lock()`` around + :meth:`prepare`/:meth:`commit`, which is what makes "no other omind writer + is touching these files" true for the duration. + """ + + def __init__(self, omi_dir: Path | str) -> None: + self.omi_dir = Path(omi_dir) + self.id = f"{time.strftime('%Y%m%dT%H%M%S')}-{uuid.uuid4().hex[:8]}" + self._entries: list[_Entry] = [] + self._prepared = False + + # -- building ----------------------------------------------------------- + + def write(self, path: Path, content: str) -> None: + """Queue ``path`` to be replaced with ``content`` on commit.""" + if self._prepared: + raise TransactionError("cannot add writes after prepare()") + self._entries.append(_Entry(path=Path(path), content=content)) + + # -- storage ------------------------------------------------------------ + + def _dir(self) -> Path: + return paths.transaction_dir(self.omi_dir) / self.id + + def _journal_path(self) -> Path: + return self._dir() / "journal.json" + + def _preimage_path(self, index: int) -> Path: + return self._dir() / f"{index:04d}.pre" + + def _record(self, state: str) -> None: + payload = { + "id": self.id, + "state": state, + "vault": str(self.omi_dir), + "updated": time.time(), + "entries": [entry.to_json() for entry in self._entries], + } + journal = self._journal_path() + tmp = journal.with_suffix(".tmp") + tmp.write_text(json.dumps(payload, indent=2), encoding="utf-8") + _fsync_path(tmp) + os.replace(tmp, journal) + _fsync_dir(journal.parent) + + # -- lifecycle ---------------------------------------------------------- + + def prepare(self) -> None: + """Capture and durably record every target's pre-image. + + Nothing in the vault has changed when this returns — but from here on a + crash is recoverable, because the bytes needed to undo the writes are on + disk and fsynced. + """ + if self._prepared: + return + directory = self._dir() + directory.mkdir(parents=True, exist_ok=True) + for index, entry in enumerate(self._entries): + entry.new_sha = _sha(entry.content) + try: + prior = entry.path.read_bytes() + except FileNotFoundError: + entry.existed = False + continue + except OSError as exc: + raise TransactionError(f"cannot read {entry.path.name}: {exc}") from exc + entry.existed = True + entry.prior_sha = _sha_bytes(prior) + preimage = self._preimage_path(index) + preimage.write_bytes(prior) + _fsync_path(preimage) + _fsync_dir(directory) + self._record(PREPARED) + self._prepared = True + + def apply(self, writer: Any) -> None: + """Perform the writes with ``writer(path, content)`` (the store's + atomic replace). Must follow :meth:`prepare`.""" + if not self._prepared: + raise TransactionError("apply() before prepare()") + for entry in self._entries: + writer(entry.path, entry.content) + + def commit(self) -> None: + """Mark the transaction complete and drop the journal. + + Past this point there is nothing to recover: every write landed. The + commit record is fsynced *before* the journal directory is removed, so a + crash between the two leaves a committed record that :func:`recover` + correctly does nothing with. + """ + if not self._prepared: + raise TransactionError("commit() before prepare()") + self._record(COMMITTED) + self.discard() + + def discard(self) -> None: + """Remove this transaction's journal and pre-images. Never raises.""" + directory = self._dir() + with contextlib.suppress(OSError): + for child in directory.iterdir(): + with contextlib.suppress(OSError): + child.unlink() + with contextlib.suppress(OSError): + directory.rmdir() + + def rollback(self) -> RecoveryReport: + """Undo a prepared-but-uncommitted transaction in this process.""" + report = _rollback_journal(self._dir()) + self.discard() + return report + + +# -- recovery --------------------------------------------------------------- + + +def pending(omi_dir: Path | str) -> list[Path]: + """Journal directories for interrupted transactions, oldest first.""" + root = paths.transaction_dir(Path(omi_dir)) + if not root.is_dir(): + return [] + found: list[Path] = [] + for directory in sorted(root.iterdir()): + journal = directory / "journal.json" + if not journal.is_file(): + continue + try: + payload = json.loads(journal.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + if str(payload.get("state")) != COMMITTED: + found.append(directory) + return found + + +def _rollback_journal(directory: Path) -> RecoveryReport: + """Restore one journal's pre-images. See the module docstring's safety rule.""" + journal = directory / "journal.json" + try: + payload = json.loads(journal.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise TransactionError(f"unreadable journal at {directory}: {exc}") from exc + report = RecoveryReport(transaction_id=str(payload.get("id") or directory.name)) + for index, raw in enumerate(payload.get("entries") or []): + path = Path(str(raw.get("path"))) + prior_sha = str(raw.get("prior_sha") or "") + new_sha = str(raw.get("new_sha") or "") + existed = bool(raw.get("existed")) + try: + current = path.read_bytes() + current_sha = _sha_bytes(current) + except FileNotFoundError: + current_sha = "" + except OSError: + report.conflicts.append(path.name) + continue + + if existed and current_sha == prior_sha: + report.skipped.append(path.name) # never written, or already undone + continue + if not existed and current_sha == "": + report.skipped.append(path.name) # we never created it + continue + # Only bytes we wrote are ours to undo. Anything else is someone's edit + # made after the crash, and it is newer than our pre-image. + if current_sha != new_sha: + report.conflicts.append(path.name) + continue + if existed: + preimage = directory / f"{index:04d}.pre" + try: + _atomic_write_bytes(path, preimage.read_bytes()) + except OSError: + report.conflicts.append(path.name) + continue + report.restored.append(path.name) + else: + with contextlib.suppress(OSError): + path.unlink() + report.removed.append(path.name) + return report + + +def recover(omi_dir: Path | str, *, dry_run: bool = False) -> list[RecoveryReport]: + """Roll back every interrupted transaction for this vault. + + A no-op when the journal is clean, which is the normal case. Journals whose + rollback hit a conflict are **kept**, so a human can look rather than having + the evidence deleted underneath them. + """ + reports: list[RecoveryReport] = [] + for directory in pending(omi_dir): + if dry_run: + try: + payload = json.loads((directory / "journal.json").read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + entries = payload.get("entries") or [] + reports.append( + RecoveryReport( + transaction_id=str(payload.get("id") or directory.name), + skipped=[Path(str(e.get("path"))).name for e in entries], + ) + ) + continue + report = _rollback_journal(directory) + reports.append(report) + if report.clean: + with contextlib.suppress(OSError): + for child in directory.iterdir(): + with contextlib.suppress(OSError): + child.unlink() + directory.rmdir() + return reports diff --git a/tests/test_txn.py b/tests/test_txn.py new file mode 100644 index 0000000..44c20ca --- /dev/null +++ b/tests/test_txn.py @@ -0,0 +1,319 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Aaron K. Clark +"""Tests for omind.txn: journaled multi-note writes and their recovery. + +The interesting cases are all failure cases, so most of these interrupt a +transaction on purpose and then assert what `omind recover` does with the +wreckage. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from omind import txn +from omind.store import NoteFields, OmiStore, _atomic_write + + +@pytest.fixture +def omi(tmp_path: Path) -> Path: + d = tmp_path / "OMI" + d.mkdir() + return d + + +def _write(path: Path, text: str) -> None: + _atomic_write(path, text) + + +def test_commit_applies_every_write_and_leaves_no_journal(omi: Path) -> None: + a, b = omi / "A.md", omi / "B.md" + _write(a, "old A\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.write(b, "new B\n") + t.prepare() + t.apply(_atomic_write) + t.commit() + + assert a.read_text() == "new A\n" + assert b.read_text() == "new B\n" + assert txn.pending(omi) == [] # nothing left to recover + + +def test_prepare_changes_nothing_on_disk(omi: Path) -> None: + """Preparing is the point where a crash becomes *recoverable*, not visible.""" + a = omi / "A.md" + _write(a, "old A\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.prepare() + + assert a.read_text() == "old A\n" + assert len(txn.pending(omi)) == 1 # journaled, uncommitted + + +def test_an_interrupted_apply_is_rolled_back(omi: Path) -> None: + """The whole point: partial state is undone, not left for a human to find.""" + a, b, c = omi / "A.md", omi / "B.md", omi / "C.md" + _write(a, "old A\n") + _write(b, "old B\n") + + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.write(b, "new B\n") + t.write(c, "created C\n") + t.prepare() + + # Die after the first write lands: A is new, B is stale, C never existed. + written = 0 + + def dying_writer(path: Path, content: str) -> None: + nonlocal written + if written >= 1: + raise KeyboardInterrupt("power loss") + written += 1 + _atomic_write(path, content) + + with pytest.raises(KeyboardInterrupt): + t.apply(dying_writer) + assert a.read_text() == "new A\n" # genuinely half-applied + assert b.read_text() == "old B\n" + + reports = txn.recover(omi) + assert len(reports) == 1 and reports[0].clean + assert a.read_text() == "old A\n" # rolled back + assert b.read_text() == "old B\n" # untouched + assert not c.exists() # a file we created is removed again + assert txn.pending(omi) == [] + + +def test_recovery_refuses_to_clobber_an_edit_made_after_the_crash(omi: Path) -> None: + """Someone's later edit is newer information than our pre-image (#194). + + Blind rollback here would be data loss wearing recovery's clothes. + """ + a = omi / "A.md" + _write(a, "old A\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.prepare() + t.apply(_atomic_write) # applied, but the process died before commit() + + a.write_text("a human fixed this by hand\n", encoding="utf-8") + + reports = txn.recover(omi) + assert len(reports) == 1 + assert reports[0].conflicts == ["A.md"] + assert not reports[0].clean + assert a.read_text() == "a human fixed this by hand\n" # preserved + assert txn.pending(omi) # journal kept, so the evidence survives for a human + + +def test_recovery_is_idempotent(omi: Path) -> None: + a = omi / "A.md" + _write(a, "old A\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.prepare() + t.apply(_atomic_write) + + assert txn.recover(omi)[0].clean + assert a.read_text() == "old A\n" + assert txn.recover(omi) == [] # second pass finds nothing + assert a.read_text() == "old A\n" + + +def test_a_committed_transaction_is_never_rolled_back(omi: Path) -> None: + """A crash between the commit record and the cleanup must not undo the work.""" + a = omi / "A.md" + _write(a, "old A\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.prepare() + t.apply(_atomic_write) + t._record(txn.COMMITTED) # commit record fsynced; cleanup did not run + + assert txn.pending(omi) == [] + assert txn.recover(omi) == [] + assert a.read_text() == "new A\n" + + +def test_dry_run_reports_without_changing_anything(omi: Path) -> None: + a = omi / "A.md" + _write(a, "old A\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.prepare() + t.apply(_atomic_write) + + reports = txn.recover(omi, dry_run=True) + assert len(reports) == 1 and reports[0].skipped == ["A.md"] + assert a.read_text() == "new A\n" # untouched + assert txn.pending(omi) # still pending + + +def test_the_journal_lives_outside_the_vault(omi: Path) -> None: + """Invariant 1: derived state never lands in a mesh-synced vault.""" + t = txn.Transaction(omi) + t.write(omi / "A.md", "x\n") + t.prepare() + assert not list(omi.rglob("journal.json")) + assert txn.pending(omi)[0].is_relative_to(txn.paths.state_dir()) + + +def test_consolidate_apply_rolls_back_a_failed_merge(omi: Path, monkeypatch) -> None: # type: ignore[no-untyped-def] + """The write path that used to concede 'extra recoverable copies' (#194).""" + store = OmiStore(omi) + store.create_note(NoteFields(title="Source One", summary="first")) + store.create_note(NoteFields(title="Source Two", summary="second")) + sources = [ + ("Source One.md", store.note_version("Source One.md")), + ("Source Two.md", store.note_version("Source Two.md")), + ] + before = {name: (omi / name).read_text() for name, _ in sources} + + real = txn.Transaction.apply + calls = {"n": 0} + + def flaky(self: txn.Transaction, writer: object) -> None: + calls["n"] += 1 + + def half(path: Path, content: str) -> None: + if calls["n"] and path.name == "Source Two.md": + raise OSError("disk full") + _atomic_write(path, content) + + real(self, half) + + monkeypatch.setattr(txn.Transaction, "apply", flaky) + with pytest.raises(OSError): + store.create_and_disable_sources( + NoteFields(title="Merged", summary="merged"), sources + ) + + # Neither a stray merged note nor a half-archived source survives. + assert not (omi / "Merged.md").exists() + for name, _ in sources: + assert (omi / name).read_text() == before[name] + assert txn.pending(omi) == [] # rolled back in-process; no manual recover needed + + +def test_recover_cli_reports_and_exits_nonzero_on_conflict( + omi: Path, capsys: pytest.CaptureFixture[str] +) -> None: + from omind.cli import main + + a = omi / "A.md" + _write(a, "old A\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.prepare() + t.apply(_atomic_write) + a.write_text("hand-edited\n", encoding="utf-8") + + code = main(["recover", "--vault", str(omi.parent), "--folder", omi.name]) + out = capsys.readouterr() + assert code == 1 + assert "CONFLICT" in out.out + assert a.read_text() == "hand-edited\n" + + +def test_recover_cli_is_a_noop_on_a_clean_journal( + omi: Path, capsys: pytest.CaptureFixture[str] +) -> None: + from omind.cli import main + + assert main(["recover", "--vault", str(omi.parent), "--folder", omi.name]) == 0 + assert "nothing to recover" in capsys.readouterr().out + + +def test_a_corrupt_journal_is_reported_not_silently_skipped(omi: Path) -> None: + t = txn.Transaction(omi) + t.write(omi / "A.md", "x\n") + t.prepare() + journal = txn.pending(omi)[0] / "journal.json" + journal.write_text("{not json", encoding="utf-8") + + assert txn.pending(omi) == [] # unparseable: not claimed as recoverable + with pytest.raises(txn.TransactionError): + txn._rollback_journal(journal.parent) + + +def test_prepare_records_every_target_in_the_journal(omi: Path) -> None: + a = omi / "A.md" + _write(a, "old A\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.write(omi / "B.md", "new B\n") + t.prepare() + + payload = json.loads((txn.pending(omi)[0] / "journal.json").read_text()) + assert payload["state"] == txn.PREPARED + entries = {Path(e["path"]).name: e for e in payload["entries"]} + assert entries["A.md"]["existed"] is True and entries["A.md"]["prior_sha"] + assert entries["B.md"]["existed"] is False and not entries["B.md"]["prior_sha"] + assert all(e["new_sha"] for e in payload["entries"]) # the undo proof + + +def test_rollback_recognizes_its_own_write_after_line_ending_translation( + omi: Path, +) -> None: + """CRLF on disk must not read as "somebody else edited this" (Windows). + + `_atomic_write` writes in text mode, so on Windows every `\\n` lands as + `\\r\\n`. Hashing raw bytes made the file we had just written match neither + the pre-image nor our intended content, so recovery classified *everything* + as a conflict and rolled back nothing — the feature was inert on Windows and + said nothing about it. Both Windows CI legs caught it. + + Reproduced here on any platform by writing the translated bytes directly. + """ + a = omi / "A.md" + a.write_bytes(b"old A\r\n") + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.prepare() + a.write_bytes(b"new A\r\n") # what the writer leaves on a Windows disk + + report = txn.recover(omi)[0] + assert report.clean, f"CRLF read as a foreign edit: {report.conflicts}" + assert report.restored == ["A.md"] + assert a.read_text().replace("\r\n", "\n") == "old A\n" + + +def test_rollback_never_routes_a_preimage_through_the_text_writer( + omi: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A pre-image is bytes and must be restored as bytes (Windows). + + The store's `_atomic_write` is TEXT mode, so on Windows it translates every + `\n` to `\r\n`. Restoring a CRLF pre-image through it translated a second + time — `b"old A\r\n"` was written back as `b"old A\r\r\n"`, so every + rollback silently grew a blank line in the note it was restoring. Both + Windows CI legs caught it. + + On Linux the translation is a no-op, so asserting on bytes alone proves + nothing here. Instead: install a writer that translates the way Windows + does, and assert recovery never touches it. + """ + from omind import store + + def windows_like_text_writer(path: Path, text: str) -> None: + Path(path).write_bytes(text.replace("\n", "\r\n").encode("utf-8")) + + a = omi / "A.md" + original = b"old A\r\nsecond line\r\n" + a.write_bytes(original) + + t = txn.Transaction(omi) + t.write(a, "new A\n") + t.prepare() + a.write_bytes(b"new A\r\n") # what a Windows writer leaves behind + + monkeypatch.setattr(store, "_atomic_write", windows_like_text_writer) + assert txn.recover(omi)[0].clean + assert a.read_bytes() == original # byte-identical, not merely equivalent diff --git a/uv.lock b/uv.lock index 3ff003f..fad624c 100644 --- a/uv.lock +++ b/uv.lock @@ -2354,7 +2354,7 @@ wheels = [ [[package]] name = "omind" -version = "7.0.0" +version = "8.0.0" source = { editable = "." } dependencies = [ { name = "cryptography" },