From 375d8c2d2a15add8fc6917a671bf9496d8539728 Mon Sep 17 00:00:00 2001 From: wang-qisen Date: Wed, 9 Sep 2026 15:55:08 +0800 Subject: [PATCH] refactor(auto_resource): streamline batch lookup refresh --- reme/steps/evolve/auto_resource.py | 15 +- reme/steps/evolve/base_auto_resource.py | 78 ++++------ .../test_auto_resource_concurrent_lookup.py | 143 +++++++----------- .../test_auto_resource_external_lookup.py | 31 ++++ .../test_auto_resource_snapshot_lookup.py | 73 +++++++-- 5 files changed, 181 insertions(+), 159 deletions(-) diff --git a/reme/steps/evolve/auto_resource.py b/reme/steps/evolve/auto_resource.py index 286e87ba..5c13022e 100644 --- a/reme/steps/evolve/auto_resource.py +++ b/reme/steps/evolve/auto_resource.py @@ -168,16 +168,15 @@ class AutoResourceStep(BaseStep): route_counts = ", ".join(f"{spec['backend']}={len(batch)}" for spec, _, batch in routes) self.logger.info(f"[{self.name}] route changes={len(changes)} processors=({route_counts})") - # One common lookup scope covers every configured modality. Base builds - # history only when a loose-root resource needs it, regardless of order. - with _resource_lookup_scope(self.context): - try: + try: + # Base lazily builds history in a scope shared by every modality. + with _resource_lookup_scope(self.context): for spec, _, indexed_changes in routes: await self._dispatch_processor(spec, indexed_changes, result_slots) - finally: - # dispatch_steps merges the sub-batch into the shared context. - # Downstream steps and the result hook must see the original batch. - self.context["changes"] = changes + finally: + # dispatch_steps merges the sub-batch into the shared context. + # Downstream steps and the result hook must see the original batch. + self.context["changes"] = changes results = [item for item in result_slots if item is not None] success_count = sum(1 for item in results if item.get("success")) diff --git a/reme/steps/evolve/base_auto_resource.py b/reme/steps/evolve/base_auto_resource.py index d287caaa..afef583d 100644 --- a/reme/steps/evolve/base_auto_resource.py +++ b/reme/steps/evolve/base_auto_resource.py @@ -7,7 +7,7 @@ import os import re import stat from abc import abstractmethod -from collections.abc import Awaitable, Callable, Mapping +from collections.abc import Mapping from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path, PurePosixPath @@ -29,26 +29,17 @@ _LOOKUP_BATCH_KEY = "_auto_resource_lookup_batch" _LookupKey = tuple[Path, str, int, int] -def _stat_signature(value: os.stat_result) -> tuple[int, ...]: - """Detect replacement, in-place edits, and edits which restore mtime.""" - return (value.st_dev, value.st_ino, value.st_mode, value.st_size, value.st_mtime_ns, value.st_ctime_ns) - - @dataclass(frozen=True) class _ResourceDaySnapshot: """Metadata only: all direct Markdown files, not just currently owned notes.""" directory: Path - signature: tuple[int, ...] - files: dict[str, tuple] + files: dict[str, tuple[Path | None, int, int, int]] def _snapshot_note_day(workspace: Path, directory: Path) -> _ResourceDaySnapshot | None: """Inspect file metadata without reading bodies or parsing frontmatter.""" try: - directory_stat = directory.stat() - if not stat.S_ISDIR(directory_stat.st_mode): - return None files = {} with os.scandir(directory) as entries: for entry in entries: @@ -70,11 +61,8 @@ def _snapshot_note_day(workspace: Path, directory: Path) -> _ResourceDaySnapshot try: value = entry.stat() if stat.S_ISREG(value.st_mode): - files[entry.name] = ( - target, - _stat_signature(value), - _stat_signature(entry.stat(follow_symlinks=False)) if target is not None else None, - ) + # ctime also catches edits which restore mtime on POSIX. + files[entry.name] = (target, value.st_mtime_ns, value.st_size, value.st_ctime_ns) except FileNotFoundError: # A removed file or a dangling internal link is not a note. continue @@ -83,7 +71,7 @@ def _snapshot_note_day(workspace: Path, directory: Path) -> _ResourceDaySnapshot if exc.errno == errno.ELOOP: continue raise - return _ResourceDaySnapshot(directory, _stat_signature(directory_stat), files) + return _ResourceDaySnapshot(directory, files) except (FileNotFoundError, NotADirectoryError): return None except OSError as exc: @@ -102,7 +90,7 @@ class _ResourceDayLookup: class _ResourceLookupBatch: - """Own modality-independent index construction, refresh and invocation lifetime.""" + """Keep lookup data and its refresh lock within one resource invocation.""" def __init__(self): self.indexes: dict[_LookupKey, dict[str, _ResourceDayLookup]] = {} @@ -112,33 +100,6 @@ class _ResourceLookupBatch: """Drop ownership data; never carry it to another batch.""" self.indexes.clear() - async def find_matches( - self, - key: _LookupKey, - source: str, - snapshot: Callable[[], dict[str, _ResourceDaySnapshot]], - read_day: Callable[[str], Awaitable[list[dict]]], - ) -> list[tuple[str, str]]: - """Refresh changed days before lookup, without making each scan a transaction.""" - async with self.lock: - before = snapshot() - days = self.indexes.get(key, {}).copy() - for day in sorted(before.keys() | days.keys()): - if day not in before: - days.pop(day) - elif day not in days or days[day].snapshot != before[day]: - owners = {} - for note in await read_day(day): - owner = str(note.get(_SOURCE_RESOURCE_KEY, "")).strip() - if owner and owner not in owners: - owners[owner] = str(note["path"]) - # Keep the pre-read version: a write during this read must - # invalidate the cache at the next resource's lookup. - days[day] = _ResourceDayLookup(before[day], owners) - # Failed reads never publish a partial refresh or a new version. - self.indexes[key] = days - return sorted((day, cached.owners[source]) for day, cached in days.items() if source in cached.owners) - @contextmanager def _resource_lookup_scope(context: RuntimeContext, *, reuse: bool = False): @@ -418,12 +379,27 @@ class BaseAutoResourceStep(BaseStep): id(self.file_store), id(self.app_context), ) - matches = await batch.find_matches( - key, - self._source_resource_link(file_path), - self._resource_note_snapshot, - self._list_daily_notes, - ) + async with batch.lock: + snapshots = self._resource_note_snapshot() + days = batch.indexes.get(key, {}).copy() + for day in sorted(snapshots.keys() | days.keys()): + if day not in snapshots: + days.pop(day) + elif day not in days or days[day].snapshot != snapshots[day]: + owners = {} + for note in await self._list_daily_notes(day): + owner = str(note.get(_SOURCE_RESOURCE_KEY, "")).strip() + if owner and owner not in owners: + owners[owner] = str(note["path"]) + # Save the pre-read version so later resources detect writes + # during this read, without retrying the current lookup. + days[day] = _ResourceDayLookup(snapshots[day], owners) + # Failed reads leave the previous cache available for a later retry. + batch.indexes[key] = days + source = self._source_resource_link(file_path) + matches = sorted( + (day, cached.owners[source]) for day, cached in days.items() if source in cached.owners + ) if len(matches) > 1: paths = ", ".join(path for _, path in matches) raise RuntimeError(f"Multiple daily resource notes claim {file_path}: {paths}") diff --git a/tests/unit/test_auto_resource_concurrent_lookup.py b/tests/unit/test_auto_resource_concurrent_lookup.py index 32e5fb09..2c02455b 100644 --- a/tests/unit/test_auto_resource_concurrent_lookup.py +++ b/tests/unit/test_auto_resource_concurrent_lookup.py @@ -1,6 +1,7 @@ """Concurrent resource invocations must not reuse another batch's stale ownership.""" import asyncio +from contextlib import asynccontextmanager import pytest @@ -32,6 +33,41 @@ def _assert_no_active_batches(env): assert env.app_context.metadata == {} +@asynccontextmanager +async def _paused_between_resources(env, monkeypatch, consumer, *, change="deleted"): + """Pause a two-item batch after its gate lookup; yield an explicit resume operation.""" + ready, resume = asyncio.Event(), asyncio.Event() + original_delete = consumer._handle_delete # pylint: disable=protected-access + + async def pause_gate(file_path, date_str, note_stem): + if file_path == "resource/gate.png": + ready.set() + await resume.wait() + await original_delete(file_path, date_str, note_stem) + + monkeypatch.setattr(consumer, "_handle_delete", pause_gate) + task = asyncio.create_task( + env.run( + consumer, + [ + {"change": "deleted", "path": "resource/gate.png"}, + {"change": change, "path": "resource/photo.png"}, + ], + ), + ) + + async def resume_consumer(): + resume.set() + return await asyncio.wait_for(task, timeout=5) + + try: + await asyncio.wait_for(ready.wait(), timeout=5) + yield resume_consumer + finally: + resume.set() + await _stop_task(task) + + @pytest.mark.parametrize("consumer_change", ["modified", "deleted"]) @pytest.mark.parametrize("existing_day", [False, True], ids=["new-day", "existing-day"]) async def test_another_batch_new_owner_is_seen_by_later_resource( @@ -49,40 +85,19 @@ async def test_another_batch_new_owner_is_seen_by_later_resource( env.write_binary("resource/photo.png", image_bytes()) consumer = _processor(env, "2026-01-02") producer = _processor(env, "2026-01-01") - ready, resume = asyncio.Event(), asyncio.Event() - original_delete = consumer._handle_delete # pylint: disable=protected-access - async def pause_after_lookup(file_path, date_str, note_stem): - if file_path == "resource/gate.png": - ready.set() - await resume.wait() - await original_delete(file_path, date_str, note_stem) - - monkeypatch.setattr(consumer, "_handle_delete", pause_after_lookup) - context = RuntimeContext( - changes=[ - {"change": "deleted", "path": "resource/gate.png"}, - {"change": consumer_change, "path": "resource/photo.png"}, - ], - ) - task = asyncio.create_task(consumer(context)) - try: - await asyncio.wait_for(ready.wait(), timeout=5) + async with _paused_between_resources(env, monkeypatch, consumer, change=consumer_change) as resume: created = await env.run(producer, [{"change": "added", "path": "resource/photo.png"}]) assert created.success is True card_path = created.metadata["results"][0]["metadata"]["path"] assert card_path.startswith("daily/2026-01-01/") - resume.set() - response = await asyncio.wait_for(task, timeout=5) + response = await resume() result = response.metadata["results"][-1] assert result["success"] is True assert result["metadata"]["action"] == consumer_change assert result["metadata"]["path"] == card_path assert (env.workspace / card_path).exists() is (consumer_change != "deleted") assert not list((env.workspace / "daily/2026-01-02").glob("*.md")) - finally: - resume.set() - await _stop_task(task) _assert_no_active_batches(env) @@ -205,51 +220,30 @@ async def test_cancelled_writer_invalidates_other_batch_after_partial_write(auto env.write_binary("resource/photo.png", image_bytes()) consumer = _processor(env, "2026-01-02") producer = _processor(env, "2026-01-01") - ready, resume, written = asyncio.Event(), asyncio.Event(), asyncio.Event() - original_delete = consumer._handle_delete # pylint: disable=protected-access - - async def pause_consumer(file_path, date_str, note_stem): - if file_path == "resource/gate.png": - ready.set() - await resume.wait() - await original_delete(file_path, date_str, note_stem) + written = asyncio.Event() async def pause_after_write(_day): written.set() await asyncio.Event().wait() - monkeypatch.setattr(consumer, "_handle_delete", pause_consumer) monkeypatch.setattr(producer, "_refresh_day_index", pause_after_write) - consumer_task = asyncio.create_task( - env.run( - consumer, - [ - {"change": "deleted", "path": "resource/gate.png"}, - {"change": "deleted", "path": "resource/photo.png"}, - ], - ), - ) - producer_task = None - try: - await asyncio.wait_for(ready.wait(), timeout=5) + + async with _paused_between_resources(env, monkeypatch, consumer) as resume: producer_task = asyncio.create_task(env.run(producer, [{"change": "added", "path": "resource/photo.png"}])) - await asyncio.wait_for(written.wait(), timeout=5) - card = env.workspace / "daily/2026-01-01/caption.md" - assert card.exists() - producer_task.cancel() - with pytest.raises(asyncio.CancelledError): - await producer_task - resume.set() - response = await asyncio.wait_for(consumer_task, timeout=5) - result = response.metadata["results"][-1] - assert result["success"] is True - assert result["metadata"]["action"] == "deleted" - assert result["metadata"]["path"] == "daily/2026-01-01/caption.md" - assert not card.exists() - finally: - resume.set() - await _stop_task(consumer_task) - if producer_task is not None: + try: + await asyncio.wait_for(written.wait(), timeout=5) + card = env.workspace / "daily/2026-01-01/caption.md" + assert card.exists() + producer_task.cancel() + with pytest.raises(asyncio.CancelledError): + await producer_task + response = await resume() + result = response.metadata["results"][-1] + assert result["success"] is True + assert result["metadata"]["action"] == "deleted" + assert result["metadata"]["path"] == "daily/2026-01-01/caption.md" + assert not card.exists() + finally: await _stop_task(producer_task) _assert_no_active_batches(env) @@ -262,40 +256,17 @@ async def test_another_batch_deletion_removes_cached_day(auto_resource_env, monk env.write_binary("resource/photo.png", image_bytes()) consumer = _processor(env, "2026-01-02") producer = _processor(env, "2026-01-01") - ready, resume = asyncio.Event(), asyncio.Event() - original_delete = consumer._handle_delete # pylint: disable=protected-access - async def pause_consumer(file_path, date_str, note_stem): - if file_path == "resource/gate.png": - ready.set() - await resume.wait() - await original_delete(file_path, date_str, note_stem) - - monkeypatch.setattr(consumer, "_handle_delete", pause_consumer) - task = asyncio.create_task( - env.run( - consumer, - [ - {"change": "deleted", "path": "resource/gate.png"}, - {"change": "added", "path": "resource/photo.png"}, - ], - ), - ) - try: - await asyncio.wait_for(ready.wait(), timeout=5) + async with _paused_between_resources(env, monkeypatch, consumer, change="added") as resume: removed = await env.run(producer, [{"change": "deleted", "path": "resource/photo.png"}]) assert removed.success is True assert not old_card.exists() - resume.set() - response = await asyncio.wait_for(task, timeout=5) + response = await resume() result = response.metadata["results"][-1] assert result["success"] is True assert result["metadata"]["created"] is True assert result["metadata"]["path"] == "daily/2026-01-02/caption.md" assert not list((env.workspace / "daily/2026-01-01").glob("*.md")) - finally: - resume.set() - await _stop_task(task) _assert_no_active_batches(env) diff --git a/tests/unit/test_auto_resource_external_lookup.py b/tests/unit/test_auto_resource_external_lookup.py index 5d842daf..f10f6434 100644 --- a/tests/unit/test_auto_resource_external_lookup.py +++ b/tests/unit/test_auto_resource_external_lookup.py @@ -158,6 +158,37 @@ async def test_external_owner_removal_stops_reusing_previous_date(new_source, au assert card.read_bytes() == replacement +@pytest.mark.parametrize("restore_mtime", [False, True], ids=["normal-save", "preserved-mtime"]) +async def test_atomic_replacement_refreshes_owner(restore_mtime, auto_resource_env, monkeypatch): + """Replacing a same-size file must not preserve its previous ownership in the cache.""" + env = auto_resource_env + card = env.write_binary(_ORIGINAL_CARD, _note_text("resource/other.png").encode()) + replacement = card.with_suffix(".tmp") + + async def replace_card(): + before = card.stat() + content = _note_text(_SOURCE).encode() + assert len(content) == before.st_size + replacement.write_bytes(content) + if restore_mtime: + os.utime(replacement, ns=(before.st_atime_ns, before.st_mtime_ns)) + replacement.replace(card) + after = card.stat() + assert after.st_size == before.st_size + if restore_mtime: + assert after.st_mtime_ns == before.st_mtime_ns + if (after.st_mtime_ns, after.st_ctime_ns) == (before.st_mtime_ns, before.st_ctime_ns): + pytest.skip("Filesystem does not expose this replacement through a distinct mtime or ctime") + + response, result = await _after_history_edit(env, monkeypatch, replace_card) + + assert response.success is True + assert result["metadata"]["action"] == "deleted" + assert result["metadata"]["path"] == _ORIGINAL_CARD + assert not card.exists() + assert not replacement.exists() + + @pytest.mark.parametrize("operation", ["move", "delete"]) async def test_external_move_or_delete_updates_cached_owner(operation, auto_resource_env, monkeypatch): """Manual moves and deletion cannot leave a cached historical owner behind.""" diff --git a/tests/unit/test_auto_resource_snapshot_lookup.py b/tests/unit/test_auto_resource_snapshot_lookup.py index 05517bf2..9818918a 100644 --- a/tests/unit/test_auto_resource_snapshot_lookup.py +++ b/tests/unit/test_auto_resource_snapshot_lookup.py @@ -4,7 +4,6 @@ import asyncio import errno from collections import Counter from contextlib import nullcontext -from pathlib import Path from types import SimpleNamespace from unittest.mock import Mock @@ -251,14 +250,32 @@ async def test_failed_changed_day_read_does_not_accept_new_fingerprint(auto_reso assert not card.exists() -async def test_unchanged_snapshots_do_not_reparse_historical_notes(auto_resource_env, monkeypatch): - """Every resource checks metadata, but unchanged historical days are parsed once.""" +@pytest.mark.parametrize("irrelevant_edit", ["none", "non-markdown", "same-target-link"]) +async def test_unchanged_snapshots_do_not_reparse_historical_notes(irrelevant_edit, auto_resource_env, monkeypatch): + """Only changes to note contents or targets require another historical read.""" env = auto_resource_env env.app_context.metadata = {} days = ("2025-12-01", "2025-12-02") for day in days: for index in range(8): env.write_note(f"daily/{day}/archive-{index}.md", f"[[resource/archive-{day}-{index}.txt]]") + step = _processor(env) + original_delete = step._handle_delete # pylint: disable=protected-access + alias = env.workspace / f"daily/{days[0]}/alias.md" + target = env.workspace / "storage/archive.md" + if irrelevant_edit == "same-target-link": + env.write_note("storage/archive.md", "[[resource/archive.txt]]") + _symlink(alias, target) + + async def edit_unrelated_metadata(file_path, date_str, note_stem): + await original_delete(file_path, date_str, note_stem) + if irrelevant_edit == "non-markdown": + (alias.parent / "unrelated.txt").write_text(file_path, encoding="utf-8") + elif irrelevant_edit == "same-target-link": + alias.unlink() + _symlink(alias, target) + + monkeypatch.setattr(step, "_handle_delete", edit_unrelated_metadata) daily_list = env.app_context.jobs["daily_list"] calls = Counter() @@ -268,7 +285,7 @@ async def test_unchanged_snapshots_do_not_reparse_historical_notes(auto_resource monkeypatch.setitem(env.app_context.jobs, "daily_list", counted_daily_list) response = await env.run( - _processor(env), + step, [{"change": "deleted", "path": f"resource/missing-{index}.png"} for index in range(8)], ) assert response.success is True @@ -277,6 +294,42 @@ async def test_unchanged_snapshots_do_not_reparse_historical_notes(auto_resource assert {day: calls[day] for day in days} == dict.fromkeys(days, 1) +async def test_note_link_retarget_refreshes_cached_ownership(auto_resource_env, monkeypatch): + """A different safe target is discovered by the next resource lookup.""" + env = auto_resource_env + old_target = env.write_note("storage/old.md", "[[resource/other.png]]") + new_target = env.write_note("storage/new.md", "[[resource/photo.png]]") + day = env.workspace / "daily/2026-01-01" + day.mkdir(parents=True) + alias = day / "photo.md" + _symlink(alias, old_target) + step = _processor(env) + original_delete = step._handle_delete # pylint: disable=protected-access + + async def retarget_after_first_resource(file_path, date_str, note_stem): + await original_delete(file_path, date_str, note_stem) + if file_path == "resource/missing.png": + alias.unlink() + _symlink(alias, new_target) + + monkeypatch.setattr(step, "_handle_delete", retarget_after_first_resource) + response = await env.run( + step, + [ + {"change": "deleted", "path": "resource/missing.png"}, + {"change": "deleted", "path": "resource/photo.png"}, + ], + ) + assert response.success is True + assert response.metadata["results"][0]["metadata"]["action"] == "skipped" + result = response.metadata["results"][1]["metadata"] + assert result["action"] == "deleted" + assert result["path"] == "daily/2026-01-01/photo.md" + assert old_target.exists() + assert new_target.exists() + assert not alias.is_symlink() + + @pytest.mark.parametrize("cycle", ["self", "mutual"]) async def test_cyclic_note_links_do_not_block_unrelated_resource(cycle, auto_resource_env): """Like daily_list, metadata scans skip cyclic links that are not readable files.""" @@ -309,7 +362,7 @@ async def test_cyclic_daily_links_do_not_block_unrelated_resource(cycle, auto_re assert link.is_symlink() -@pytest.mark.parametrize("failure_site", ["directory-stat", "scandir", "file-stat"]) +@pytest.mark.parametrize("failure_site", ["scandir", "file-stat"]) @pytest.mark.parametrize("error_number", [errno.ELOOP, errno.EACCES, errno.EIO], ids=["loop", "permission", "io"]) async def test_snapshot_handles_only_loop_errors(failure_site, error_number, tmp_path, monkeypatch): """Exercise late OS errors on every Python version, without bypassing safety failures.""" @@ -317,19 +370,11 @@ async def test_snapshot_handles_only_loop_errors(failure_site, error_number, tmp day.mkdir(parents=True) (day / "valid.md").write_text("A normal note.", encoding="utf-8") error = OSError(error_number, "snapshot probe failed") - original_stat = Path.stat with base_auto_resource.os.scandir(day) as entries: valid_entries = list(entries) - def stat_with_failure(path, *args, **kwargs): - if path == day: - raise error - return original_stat(path, *args, **kwargs) - with monkeypatch.context() as patch: - if failure_site == "directory-stat": - patch.setattr(Path, "stat", stat_with_failure) - elif failure_site == "scandir": + if failure_site == "scandir": patch.setattr(base_auto_resource.os, "scandir", Mock(side_effect=error)) else: unreadable = SimpleNamespace(