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.
This commit is contained in:
wang-qisen 2026-09-09 15:03:29 +08:00
parent 16460b8cbb
commit 43b9d25bec
3 changed files with 155 additions and 155 deletions

View file

@ -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}")

View file

@ -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)

View file

@ -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