From 28dbc9c4e0489249af788dc2606e2586a524c714 Mon Sep 17 00:00:00 2001 From: "Aaron K. Clark" Date: Sun, 2 Aug 2026 14:53:14 -0500 Subject: [PATCH 1/3] feat(store): journaled multi-note transactions and `omind recover` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Multi-file updates cannot be truly atomic on the filesystems omind runs on. This does not pretend otherwise; it provides the contract that can actually 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 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 way back. `store.create_and_disable_sources` conceded exactly this in its own docstring: "a process crash can still leave extra recoverable copies". That fails toward keeping data, which is the right direction, but a human still had to notice and reconcile by hand. The rule that makes recovery safe: 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 meant to write (ours to undo). Anything else means someone edited the note after the crash, and their edit is newer information than our pre-image — so it is reported as a conflict, left alone, and its journal kept for inspection. `omind recover` exits 1 in that case. Blind rollback would be data loss wearing recovery's clothes. In-process failures roll themselves back, so consolidate needs no manual recover; the journal is for when the process dies outright. 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). AGENTS.md gains invariant 3 so the next multi-note operation does not quietly reintroduce the gap. Skipped their approved_plan_sha256 handshake as the issue planned: right for an LLM-driven research vault, wrong in front of routine memory writes. Closes #194. Co-Authored-By: Claude Opus 5 --- AGENTS.md | 19 ++- BACKLOG.md | 26 +-- CHANGELOG.md | 46 ++++++ README.md | 6 + pyproject.toml | 2 +- src/omind/__init__.py | 2 +- src/omind/cli.py | 57 ++++++- src/omind/paths.py | 9 ++ src/omind/server.py | 2 +- src/omind/store.py | 25 ++- src/omind/txn.py | 358 ++++++++++++++++++++++++++++++++++++++++++ tests/test_txn.py | 259 ++++++++++++++++++++++++++++++ uv.lock | 2 +- 13 files changed, 779 insertions(+), 34 deletions(-) create mode 100644 src/omind/txn.py create mode 100644 tests/test_txn.py 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..5ddc291 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,52 @@ 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. + + 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..799c2f4 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, _atomic_write + + 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, _atomic_write, 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..8fb1592 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(_atomic_write) + 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..44aa8e9 --- /dev/null +++ b/src/omind/txn.py @@ -0,0 +1,358 @@ +# 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 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(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +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) + + +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.encode("utf-8")) + 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(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, writer: Any) -> RecoveryReport: + """Undo a prepared-but-uncommitted transaction in this process.""" + report = _rollback_journal(self._dir(), writer) + 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, writer: Any) -> 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(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: + writer(path, preimage.read_bytes().decode("utf-8")) + except (OSError, UnicodeDecodeError): + 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, writer: Any, *, 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, writer) + 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..c821c66 --- /dev/null +++ b/tests/test_txn.py @@ -0,0 +1,259 @@ +# 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, _atomic_write) + 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, _atomic_write) + 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, _atomic_write)[0].clean + assert a.read_text() == "old A\n" + assert txn.recover(omi, _atomic_write) == [] # 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, _atomic_write) == [] + 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, _atomic_write, 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, _atomic_write) + + +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 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" }, From 5cd386a7235e5c12b88dd39c988aef2613f1255a Mon Sep 17 00:00:00 2001 From: "Aaron K. Clark" Date: Sun, 2 Aug 2026 15:01:32 -0500 Subject: [PATCH 2/3] fix(txn): hash normalized text, so CRLF is not read as a foreign edit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both windows-latest legs failed. `_atomic_write` writes in text mode, so on Windows every `\n` reaches the disk as `\r\n`. `new_sha` was computed over the pre-translation string, so the bytes on disk matched neither the pre-image nor what we had just written — recovery classified every file as "edited after the crash", refused to touch any of them, and rolled back nothing. The whole feature was inert on Windows, and silent about it: `omind recover` would report conflicts and exit 1 forever. Same class as #202 — a platform-specific path that turns into permanent no-op rather than an error. Content identity is now hashed over line-ending-normalized text. Pre-images are still stored and restored as exact bytes; only the *comparison* normalizes. `test_rollback_recognizes_its_own_write_after_line_ending_translation` reproduces it on any platform by writing the translated bytes directly — verified to fail without this fix and pass with it, so a Linux-only run still catches a regression. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 6 ++++++ src/omind/txn.py | 23 ++++++++++++++++++----- tests/test_txn.py | 26 ++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ddc291..66137bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. + 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). diff --git a/src/omind/txn.py b/src/omind/txn.py index 44aa8e9..d519c46 100644 --- a/src/omind/txn.py +++ b/src/omind/txn.py @@ -55,8 +55,21 @@ COMMITTED = "committed" -def _sha(data: bytes) -> str: - return hashlib.sha256(data).hexdigest() +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: @@ -200,7 +213,7 @@ def prepare(self) -> None: directory = self._dir() directory.mkdir(parents=True, exist_ok=True) for index, entry in enumerate(self._entries): - entry.new_sha = _sha(entry.content.encode("utf-8")) + entry.new_sha = _sha(entry.content) try: prior = entry.path.read_bytes() except FileNotFoundError: @@ -209,7 +222,7 @@ def prepare(self) -> None: except OSError as exc: raise TransactionError(f"cannot read {entry.path.name}: {exc}") from exc entry.existed = True - entry.prior_sha = _sha(prior) + entry.prior_sha = _sha_bytes(prior) preimage = self._preimage_path(index) preimage.write_bytes(prior) _fsync_path(preimage) @@ -292,7 +305,7 @@ def _rollback_journal(directory: Path, writer: Any) -> RecoveryReport: existed = bool(raw.get("existed")) try: current = path.read_bytes() - current_sha = _sha(current) + current_sha = _sha_bytes(current) except FileNotFoundError: current_sha = "" except OSError: diff --git a/tests/test_txn.py b/tests/test_txn.py index c821c66..d6d329a 100644 --- a/tests/test_txn.py +++ b/tests/test_txn.py @@ -257,3 +257,29 @@ def test_prepare_records_every_target_in_the_journal(omi: Path) -> None: 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, _atomic_write)[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" From 8203d71d4a43c2dace65904ebf952ef1118edfb6 Mon Sep 17 00:00:00 2001 From: "Aaron K. Clark" Date: Sun, 2 Aug 2026 15:08:48 -0500 Subject: [PATCH 3/3] fix(txn): restore pre-images byte for byte, not through the text writer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second Windows failure, different cause. Rollback restored a pre-image by decoding it and handing the string to the store's `_atomic_write`, which is TEXT mode — so on Windows the `\n` was translated a second time. A pre-image of b"old A\r\n" was written back as b"old A\r\r\n", meaning every rollback silently grew a blank line in the note it was supposed to be restoring. `read_text()` saw 'old A\n\n'. A pre-image is bytes. Recovery now writes it back with a binary same-dir temp + os.replace, so what was there is exactly what comes back. The `writer` argument to recover()/rollback() is gone with it — restoration was never the caller's business. The first version of this test asserted on bytes and passed against the bug, because on Linux the text writer translates nothing. Replaced with one that installs a Windows-like translating writer and asserts recovery never routes through it — verified to fail without this fix and pass with it. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 6 ++++++ src/omind/cli.py | 4 ++-- src/omind/store.py | 2 +- src/omind/txn.py | 40 ++++++++++++++++++++++++++++++------- tests/test_txn.py | 50 ++++++++++++++++++++++++++++++++++++++-------- 5 files changed, 84 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 66137bd..ee64151 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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 diff --git a/src/omind/cli.py b/src/omind/cli.py index 799c2f4..c6fb0ef 100644 --- a/src/omind/cli.py +++ b/src/omind/cli.py @@ -1242,7 +1242,7 @@ def _run_graph(args: argparse.Namespace) -> int: def _run_recover(args: argparse.Namespace) -> int: """``omind recover``: roll back interrupted multi-note transactions.""" from omind import txn - from omind.store import OmiStore, _atomic_write + from omind.store import OmiStore omi_dir = (args.vault / args.folder).expanduser() store = OmiStore(omi_dir) @@ -1252,7 +1252,7 @@ def _run_recover(args: argparse.Namespace) -> int: # 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, _atomic_write, dry_run=args.dry_run) + 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()) diff --git a/src/omind/store.py b/src/omind/store.py index 8fb1592..baa8030 100644 --- a/src/omind/store.py +++ b/src/omind/store.py @@ -1323,7 +1323,7 @@ def create_and_disable_sources( 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(_atomic_write) + transaction.rollback() raise transaction.commit() self._write_index() diff --git a/src/omind/txn.py b/src/omind/txn.py index d519c46..77b5093 100644 --- a/src/omind/txn.py +++ b/src/omind/txn.py @@ -40,6 +40,7 @@ import hashlib import json import os +import tempfile import time import uuid from dataclasses import dataclass, field @@ -91,6 +92,31 @@ def _fsync_dir(directory: Path) -> None: 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.""" @@ -261,9 +287,9 @@ def discard(self) -> None: with contextlib.suppress(OSError): directory.rmdir() - def rollback(self, writer: Any) -> RecoveryReport: + def rollback(self) -> RecoveryReport: """Undo a prepared-but-uncommitted transaction in this process.""" - report = _rollback_journal(self._dir(), writer) + report = _rollback_journal(self._dir()) self.discard() return report @@ -290,7 +316,7 @@ def pending(omi_dir: Path | str) -> list[Path]: return found -def _rollback_journal(directory: Path, writer: Any) -> RecoveryReport: +def _rollback_journal(directory: Path) -> RecoveryReport: """Restore one journal's pre-images. See the module docstring's safety rule.""" journal = directory / "journal.json" try: @@ -326,8 +352,8 @@ def _rollback_journal(directory: Path, writer: Any) -> RecoveryReport: if existed: preimage = directory / f"{index:04d}.pre" try: - writer(path, preimage.read_bytes().decode("utf-8")) - except (OSError, UnicodeDecodeError): + _atomic_write_bytes(path, preimage.read_bytes()) + except OSError: report.conflicts.append(path.name) continue report.restored.append(path.name) @@ -338,7 +364,7 @@ def _rollback_journal(directory: Path, writer: Any) -> RecoveryReport: return report -def recover(omi_dir: Path | str, writer: Any, *, dry_run: bool = False) -> list[RecoveryReport]: +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 @@ -360,7 +386,7 @@ def recover(omi_dir: Path | str, writer: Any, *, dry_run: bool = False) -> list[ ) ) continue - report = _rollback_journal(directory, writer) + report = _rollback_journal(directory) reports.append(report) if report.clean: with contextlib.suppress(OSError): diff --git a/tests/test_txn.py b/tests/test_txn.py index d6d329a..44c20ca 100644 --- a/tests/test_txn.py +++ b/tests/test_txn.py @@ -83,7 +83,7 @@ def dying_writer(path: Path, content: str) -> None: assert a.read_text() == "new A\n" # genuinely half-applied assert b.read_text() == "old B\n" - reports = txn.recover(omi, _atomic_write) + 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 @@ -105,7 +105,7 @@ def test_recovery_refuses_to_clobber_an_edit_made_after_the_crash(omi: Path) -> a.write_text("a human fixed this by hand\n", encoding="utf-8") - reports = txn.recover(omi, _atomic_write) + reports = txn.recover(omi) assert len(reports) == 1 assert reports[0].conflicts == ["A.md"] assert not reports[0].clean @@ -121,9 +121,9 @@ def test_recovery_is_idempotent(omi: Path) -> None: t.prepare() t.apply(_atomic_write) - assert txn.recover(omi, _atomic_write)[0].clean + assert txn.recover(omi)[0].clean assert a.read_text() == "old A\n" - assert txn.recover(omi, _atomic_write) == [] # second pass finds nothing + assert txn.recover(omi) == [] # second pass finds nothing assert a.read_text() == "old A\n" @@ -138,7 +138,7 @@ def test_a_committed_transaction_is_never_rolled_back(omi: Path) -> None: t._record(txn.COMMITTED) # commit record fsynced; cleanup did not run assert txn.pending(omi) == [] - assert txn.recover(omi, _atomic_write) == [] + assert txn.recover(omi) == [] assert a.read_text() == "new A\n" @@ -150,7 +150,7 @@ def test_dry_run_reports_without_changing_anything(omi: Path) -> None: t.prepare() t.apply(_atomic_write) - reports = txn.recover(omi, _atomic_write, dry_run=True) + 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 @@ -240,7 +240,7 @@ def test_a_corrupt_journal_is_reported_not_silently_skipped(omi: Path) -> None: assert txn.pending(omi) == [] # unparseable: not claimed as recoverable with pytest.raises(txn.TransactionError): - txn._rollback_journal(journal.parent, _atomic_write) + txn._rollback_journal(journal.parent) def test_prepare_records_every_target_in_the_journal(omi: Path) -> None: @@ -279,7 +279,41 @@ def test_rollback_recognizes_its_own_write_after_line_ending_translation( t.prepare() a.write_bytes(b"new A\r\n") # what the writer leaves on a Windows disk - report = txn.recover(omi, _atomic_write)[0] + 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