From 43b9d25bec74855bccc528c45e26e69a0764cbcb Mon Sep 17 00:00:00 2001 From: wang-qisen Date: Wed, 9 Sep 2026 15:03:29 +0800 Subject: [PATCH] refactor(auto_resource): simplify batch ownership lookup Keep per-day ownership caches paired with pre-read metadata and refresh changed days before subsequent resources. Remove reverse-index bookkeeping, post-read stabilization retries, and recursive scope fallback. Handle late ELOOP errors on Python 3.13 while preserving other OS errors. Update regression tests for cross-resource freshness and retain safety and lifecycle coverage. --- reme/steps/evolve/base_auto_resource.py | 128 +++++++--------- .../test_auto_resource_concurrent_lookup.py | 45 +++--- .../test_auto_resource_snapshot_lookup.py | 137 ++++++++++-------- 3 files changed, 155 insertions(+), 155 deletions(-) diff --git a/reme/steps/evolve/base_auto_resource.py b/reme/steps/evolve/base_auto_resource.py index b7b817a8..d287caaa 100644 --- a/reme/steps/evolve/base_auto_resource.py +++ b/reme/steps/evolve/base_auto_resource.py @@ -9,7 +9,7 @@ import stat from abc import abstractmethod from collections.abc import Awaitable, Callable, Mapping from contextlib import contextmanager -from dataclasses import dataclass, field +from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Any @@ -78,85 +78,66 @@ def _snapshot_note_day(workspace: Path, directory: Path) -> _ResourceDaySnapshot except FileNotFoundError: # A removed file or a dangling internal link is not a note. continue + except OSError as exc: + # Python 3.13 can report loops at stat rather than resolve. + if exc.errno == errno.ELOOP: + continue + raise return _ResourceDaySnapshot(directory, _stat_signature(directory_stat), files) except (FileNotFoundError, NotADirectoryError): return None + except OSError as exc: + # Cyclic date directories are not readable history either. + if exc.errno == errno.ELOOP: + return None + raise -@dataclass +@dataclass(frozen=True) class _ResourceDayLookup: - """Small, rebuildable ownership index; note bodies never enter the cache.""" + """One day's first owner per source, paired with its pre-read metadata.""" - owners: dict[str, dict[str, str]] = field(default_factory=dict) - day_sources: dict[str, set[str]] = field(default_factory=dict) - snapshots: dict[str, _ResourceDaySnapshot] = field(default_factory=dict) - - def replace_day(self, day: str, notes: list[dict]) -> None: - """Replace one day's owners, preserving daily_list's first-match order.""" - day_owners: dict[str, str] = {} - for note in notes: - source = str(note.get(_SOURCE_RESOURCE_KEY, "")).strip() - if source and source not in day_owners: - day_owners[source] = str(note["path"]) - for source in self.day_sources.pop(day, set()): - owners = self.owners[source] - owners.pop(day, None) - if not owners: - del self.owners[source] - for source, path in day_owners.items(): - self.owners.setdefault(source, {})[day] = path - self.day_sources[day] = set(day_owners) + snapshot: _ResourceDaySnapshot + owners: dict[str, str] class _ResourceLookupBatch: """Own modality-independent index construction, refresh and invocation lifetime.""" def __init__(self): - self.indexes: dict[_LookupKey, _ResourceDayLookup] = {} + self.indexes: dict[_LookupKey, dict[str, _ResourceDayLookup]] = {} self.lock = asyncio.Lock() def close(self) -> None: """Drop ownership data; never carry it to another batch.""" self.indexes.clear() - async def get_lookup( + async def find_matches( self, key: _LookupKey, + source: str, snapshot: Callable[[], dict[str, _ResourceDaySnapshot]], read_day: Callable[[str], Awaitable[list[dict]]], - ) -> _ResourceDayLookup: - """Reuse parsed ownership, checking disk changes from any writer before each lookup. - - Metadata checks still visit the files; unchanged days avoid daily_list - and frontmatter parsing. Before/after snapshots detect writes during - awaited reads, without requiring writers to cooperate with this batch. - """ + ) -> list[tuple[str, str]]: + """Refresh changed days before lookup, without making each scan a transaction.""" async with self.lock: - lookup = self.indexes.get(key) or _ResourceDayLookup() before = snapshot() - for _ in range(4): - changed_days = { - day - for day in before.keys() | lookup.snapshots.keys() - if before.get(day) != lookup.snapshots.get(day) - } - if not changed_days: - self.indexes[key] = lookup - return lookup - for day in sorted(changed_days): - lookup.replace_day(day, await read_day(day) if day in before else []) - if day in before: - lookup.snapshots[day] = before[day] - else: - lookup.snapshots.pop(day, None) - after = snapshot() - if before == after: - # Publish only a complete, stable scan. On failure the - # recorded pre-read snapshots ensure the next call retries. - self.indexes[key] = lookup - return lookup - before = after - raise RuntimeError("Resource note ownership kept changing during lookup; retry the resource") + 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 @@ -398,21 +379,8 @@ class BaseAutoResourceStep(BaseStep): raise RuntimeError(f"daily_list failed: {list_response.answer}") return list_response.metadata.get("notes") or [] - def _resource_lookup_cache(self) -> tuple[_ResourceLookupBatch | None, tuple]: - """Keep overridden stores/configurations from sharing another lookup domain.""" - batch = self.context.get(_LOOKUP_BATCH_KEY) if self.context is not None else None - if not isinstance(batch, _ResourceLookupBatch): - return None, () - key = ( - self.workspace_path.resolve(), - str(self.config_value("daily_dir")), - id(self.file_store), - id(self.app_context), - ) - return batch, key - def _resource_note_snapshot(self) -> dict[str, _ResourceDaySnapshot]: - """Fingerprint daily notes, including unowned files, using metadata only.""" + """Inspect daily-note metadata, including files without resource ownership.""" workspace = self.workspace_path.resolve() daily_dir = str(self.config_value("daily_dir")) daily_root, error = resolve_path(workspace, daily_dir) @@ -441,13 +409,21 @@ class BaseAutoResourceStep(BaseStep): async def _find_loose_resource_day(self, file_path: str) -> str | None: """Resolve ownership through the common batch index, for any resource modality.""" - batch, key = self._resource_lookup_cache() - if batch is None: - assert self.context is not None - with _resource_lookup_scope(self.context): - return await self._find_loose_resource_day(file_path) - lookup = await batch.get_lookup(key, self._resource_note_snapshot, self._list_daily_notes) - matches = sorted(lookup.owners.get(self._source_resource_link(file_path), {}).items()) + assert self.context is not None + with _resource_lookup_scope(self.context, reuse=True) as batch: + # Overridden stores/configurations must not share lookup contents. + key = ( + self.workspace_path.resolve(), + str(self.config_value("daily_dir")), + 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, + ) 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 539d6d89..32e5fb09 100644 --- a/tests/unit/test_auto_resource_concurrent_lookup.py +++ b/tests/unit/test_auto_resource_concurrent_lookup.py @@ -87,12 +87,12 @@ async def test_another_batch_new_owner_is_seen_by_later_resource( @pytest.mark.parametrize("read_phase", ["history", "dirty-day"]) -async def test_concurrent_write_during_daily_list_does_not_lose_invalidation( +async def test_concurrent_write_during_lookup_is_seen_by_next_resource( read_phase, auto_resource_env, monkeypatch, ): - """An awaited stale scan must not erase an invalidation arriving while it runs.""" + """A stale read keeps its pre-read fingerprint, so the next resource refreshes it.""" env = auto_resource_env env.app_context.metadata = {} env.write_note("daily/2026-01-01/gate.md", "[[resource/gate.png]]") @@ -102,6 +102,7 @@ async def test_concurrent_write_during_daily_list_does_not_lose_invalidation( ready, resume = asyncio.Event(), asyncio.Event() daily_list = env.app_context.jobs["daily_list"] consumer_reads = 0 + # Deleting gate takes a history read and a targeted read; its dirty refresh is third. blocked_read = 1 if read_phase == "history" else 3 task = None @@ -116,7 +117,10 @@ async def test_concurrent_write_during_daily_list_does_not_lose_invalidation( return response monkeypatch.setitem(env.app_context.jobs, "daily_list", pause_stale_response) - changes = [{"change": "deleted", "path": "resource/photo.png"}] + changes = [ + {"change": "deleted", "path": "resource/missing.png"}, + {"change": "deleted", "path": "resource/photo.png"}, + ] if read_phase == "dirty-day": changes.insert(0, {"change": "deleted", "path": "resource/gate.png"}) task = asyncio.create_task(env.run(consumer, changes)) @@ -127,6 +131,7 @@ async def test_concurrent_write_during_daily_list_does_not_lose_invalidation( card_path = created.metadata["results"][0]["metadata"]["path"] resume.set() response = await asyncio.wait_for(task, timeout=5) + assert response.metadata["results"][-2]["metadata"]["action"] == "skipped" result = response.metadata["results"][-1] assert result["success"] is True assert result["metadata"]["action"] == "deleted" @@ -295,8 +300,8 @@ async def test_another_batch_deletion_removes_cached_day(auto_resource_env, monk @pytest.mark.parametrize("mutation", [False, True], ids=["no-op", "continuous-writes"]) -async def test_concurrent_activity_does_not_cause_unbounded_refresh(mutation, auto_resource_env, monkeypatch): - """No-ops preserve a scan; uninterrupted real writes fail closed in bounded reads.""" +async def test_concurrent_activity_refreshes_only_on_next_resource(mutation, auto_resource_env, monkeypatch): + """Do not chase concurrent writes within a lookup; recheck them for the next resource.""" env = auto_resource_env env.app_context.metadata = {} card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]") @@ -314,7 +319,7 @@ async def test_concurrent_activity_does_not_cause_unbounded_refresh(mutation, au response = await daily_list(**kwargs) if asyncio.current_task() is task and kwargs["date"] == "2026-01-01": consumer_reads += 1 - assert consumer_reads <= 5, "lookup failed to bound its refresh attempts" + assert consumer_reads <= 3, "lookup retried instead of leaving freshness to the next resource" if mutation: model.text = caption_json("churn", "Changed image", f"version {consumer_reads}") changes = [{"change": "modified", "path": "resource/churn.png"}] @@ -326,21 +331,23 @@ async def test_concurrent_activity_does_not_cause_unbounded_refresh(mutation, au return response monkeypatch.setitem(env.app_context.jobs, "daily_list", inject_concurrent_activity) - task = asyncio.create_task(env.run(consumer, [{"change": "deleted", "path": "resource/photo.png"}])) + task = asyncio.create_task( + env.run( + consumer, + [ + {"change": "deleted", "path": "resource/missing.png"}, + {"change": "deleted", "path": "resource/photo.png"}, + ], + ), + ) try: response = await asyncio.wait_for(task, timeout=5) - result = response.metadata["results"][0] - if mutation: - assert result["success"] is False - assert result["metadata"]["action"] == "failed" - assert "ownership kept changing" in result["metadata"]["error"] - assert 2 <= consumer_reads <= 4 - assert card.exists() - else: - assert result["success"] is True - assert result["metadata"]["action"] == "deleted" - assert consumer_reads == 2 # One history scan and one fresh, targeted ownership check. - assert not card.exists() + assert response.success is True + assert [item["metadata"]["action"] for item in response.metadata["results"]] == ["skipped", "deleted"] + # One history scan and a targeted check; only real writes require a + # changed-day refresh when the second resource starts its lookup. + assert consumer_reads == (3 if mutation else 2) + assert not card.exists() finally: await _stop_task(task) _assert_no_active_batches(env) diff --git a/tests/unit/test_auto_resource_snapshot_lookup.py b/tests/unit/test_auto_resource_snapshot_lookup.py index aae226a1..05517bf2 100644 --- a/tests/unit/test_auto_resource_snapshot_lookup.py +++ b/tests/unit/test_auto_resource_snapshot_lookup.py @@ -3,10 +3,15 @@ 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 import pytest from reme.schema import Response +from reme.steps.evolve import base_auto_resource from .auto_resource_test_support import FakeVisionModel, caption_json, write_note @@ -42,8 +47,8 @@ def _symlink(link, target, *, directory=False): @pytest.mark.parametrize("read_phase", ["history", "dirty-day", "new-day"]) -async def test_external_change_during_daily_list_is_reconciled(read_phase, auto_resource_env, monkeypatch): - """A write outside every ReMe mutation hook invalidates an in-flight scan.""" +async def test_external_change_during_lookup_is_seen_by_next_resource(read_phase, auto_resource_env, monkeypatch): + """Keep the pre-read version so an external write is detected by the next resource.""" env = auto_resource_env env.app_context.metadata = {} env.write_note("daily/2026-01-01/gate.md", "[[resource/gate.png]]") @@ -51,6 +56,7 @@ async def test_external_change_during_daily_list_is_reconciled(read_phase, auto_ daily_list = env.app_context.jobs["daily_list"] ready, resume = asyncio.Event(), asyncio.Event() reads = 0 + # Deleting gate takes a history read and a targeted read; its dirty refresh is third. blocked_read = 3 if read_phase == "dirty-day" else 1 async def pause_stale_response(**kwargs): @@ -64,7 +70,10 @@ async def test_external_change_during_daily_list_is_reconciled(read_phase, auto_ return response monkeypatch.setitem(env.app_context.jobs, "daily_list", pause_stale_response) - changes = [{"change": "deleted", "path": "resource/photo.png"}] + changes = [ + {"change": "deleted", "path": "resource/missing.png"}, + {"change": "deleted", "path": "resource/photo.png"}, + ] if read_phase == "dirty-day": changes.insert(0, {"change": "deleted", "path": "resource/gate.png"}) task = asyncio.create_task(env.run(consumer, changes)) @@ -74,6 +83,7 @@ async def test_external_change_during_daily_list_is_reconciled(read_phase, auto_ card = env.write_note(f"daily/{day}/photo.md", "[[resource/photo.png]]") resume.set() response = await asyncio.wait_for(task, timeout=5) + assert response.metadata["results"][-2]["metadata"]["action"] == "skipped" result = response.metadata["results"][-1] assert result["success"] is True assert result["metadata"]["action"] == "deleted" @@ -196,63 +206,6 @@ async def test_snapshot_failure_remains_retryable(failure_phase, auto_resource_e assert not card.exists() -@pytest.mark.parametrize("failure_phase", ["initial", "refresh"]) -async def test_post_read_snapshot_failure_rechecks_ownership(failure_phase, auto_resource_env, monkeypatch): - """A failed post-read check must not hide an ownership change made during the read.""" - env = auto_resource_env - env.write_note("daily/2026-01-01/gate.md", "[[resource/gate.png]]") - card = env.write_note("daily/2026-01-01/photo.md", "[[resource/other.png]]") - consumer = _processor(env) - snapshot = consumer._resource_note_snapshot # pylint: disable=protected-access - original_delete = consumer._handle_delete # pylint: disable=protected-access - daily_list = env.app_context.jobs["daily_list"] - warmed = False - read_completed = False - failed = False - - def fail_after_read(): - nonlocal failed - if read_completed and not failed: - failed = True - raise OSError("temporary post-read snapshot failure") - return snapshot() - - async def mark_warmed(file_path, date_str, note_stem): - nonlocal warmed - await original_delete(file_path, date_str, note_stem) - if file_path == "resource/gate.png": - warmed = True - - async def change_owner_after_read(**kwargs): - nonlocal read_completed - response = await daily_list(**kwargs) - if kwargs["date"] == "2026-01-01" and not read_completed and (failure_phase == "initial" or warmed): - # The returned real daily_list result still contains the previous owner. - env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]") - read_completed = True - return response - - monkeypatch.setattr(consumer, "_resource_note_snapshot", fail_after_read) - monkeypatch.setattr(consumer, "_handle_delete", mark_warmed) - monkeypatch.setitem(env.app_context.jobs, "daily_list", change_owner_after_read) - changes = [ - {"change": "deleted", "path": "resource/photo.png"}, - {"change": "deleted", "path": "resource/photo.png"}, - ] - if failure_phase == "refresh": - changes.insert(0, {"change": "deleted", "path": "resource/gate.png"}) - response = await env.run(consumer, changes) - results = response.metadata["results"] - assert read_completed and failed - assert results[-2]["success"] is False - assert results[-2]["metadata"]["modified"] is False - assert "temporary post-read snapshot failure" in results[-2]["metadata"]["error"] - assert results[-1]["success"] is True - assert results[-1]["metadata"]["action"] == "deleted" - assert results[-1]["metadata"]["path"] == "daily/2026-01-01/photo.md" - assert not card.exists() - - async def test_failed_changed_day_read_does_not_accept_new_fingerprint(auto_resource_env, monkeypatch): """A failed refresh must not mark an externally added owner as already indexed.""" env = auto_resource_env @@ -340,6 +293,70 @@ async def test_cyclic_note_links_do_not_block_unrelated_resource(cycle, auto_res assert link.is_symlink() +@pytest.mark.parametrize("cycle", ["self", "mutual"]) +async def test_cyclic_daily_links_do_not_block_unrelated_resource(cycle, auto_resource_env): + """Unresolvable date directories do not hide a normal historical owner.""" + env = auto_resource_env + card = env.write_note("daily/2026-01-01/photo.md", "[[resource/photo.png]]") + link = card.parent.parent / "2025-12-01" + _symlink(link, "2025-12-01" if cycle == "self" else "2025-12-02", directory=True) + if cycle == "mutual": + _symlink(link.with_name("2025-12-02"), "2025-12-01", directory=True) + response = await env.run(_processor(env), [{"change": "deleted", "path": "resource/photo.png"}]) + assert response.success is True + assert response.metadata["results"][0]["metadata"]["action"] == "deleted" + assert not card.exists() + assert link.is_symlink() + + +@pytest.mark.parametrize("failure_site", ["directory-stat", "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.""" + day = tmp_path / "daily" / "2026-01-01" + 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": + patch.setattr(base_auto_resource.os, "scandir", Mock(side_effect=error)) + else: + unreadable = SimpleNamespace( + name="unreadable.md", + is_symlink=lambda: False, + stat=Mock(side_effect=error), + ) + patch.setattr( + base_auto_resource.os, + "scandir", + lambda _: nullcontext(iter([unreadable, *valid_entries])), + ) + if error_number != errno.ELOOP: + with pytest.raises(OSError) as caught: + base_auto_resource._snapshot_note_day(tmp_path, day) # pylint: disable=protected-access + assert caught.value is error + return + snapshot = base_auto_resource._snapshot_note_day(tmp_path, day) # pylint: disable=protected-access + + if failure_site == "file-stat": + assert snapshot is not None + assert set(snapshot.files) == {"valid.md"} + else: + assert snapshot is None + assert (day / "valid.md").read_text(encoding="utf-8") == "A normal note." + + async def test_escaping_note_link_fails_before_daily_list(auto_resource_env, monkeypatch): """Metadata inspection refuses an escaping Markdown target before parsing any notes.""" env = auto_resource_env