diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/quarantine.py b/backend/packages/harness/deerflow/community/aio_sandbox/quarantine.py index 1e9c5d8e9a..aa85f1d608 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/quarantine.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/quarantine.py @@ -41,6 +41,28 @@ class SandboxQuarantine: return False return True + @staticmethod + def _records(directory: Path) -> list[Path]: + """List the fences a teardown pass is about to drop. + + The teardown exclusion is a TTL lease, not a lock, so a peer's retire() + can finish between the caller's existence check and this listing. Its + records are then gone, which is the state both callers were asked to + produce; ``contains()`` reports the same interleaving as "no fence". + """ + try: + return list(directory.iterdir()) + except FileNotFoundError: + return [] + + @staticmethod + def _drop(record: Path) -> None: + """Unlink one fence, tolerating a peer that unlinked it first.""" + try: + record.unlink() + except FileNotFoundError: + pass + def mark(self, info: SandboxInfo) -> None: directory = self._directory(info) directory.mkdir(parents=True, exist_ok=True, mode=0o700) @@ -84,10 +106,10 @@ class SandboxQuarantine: if not self._exists(directory): return keep = self._key(current.container_id) if current.container_id else None - for record in directory.iterdir(): + for record in self._records(directory): if record.name == "unknown" or (keep is not None and record.name == keep): continue - record.unlink() + self._drop(record) try: directory.rmdir() except OSError: @@ -117,18 +139,20 @@ class SandboxQuarantine: This is best-effort, not full fence preservation: it closes the listing → ``rmdir`` window only. A peer record landing before the listing is - still unlinked along with the rest, and the unlink loop has unguarded - sibling windows of its own — a concurrent ``retire()`` can remove the - directory between the existence check and the listing - (``FileNotFoundError``), and on Windows a peer's ``mark()`` still - holding the fresh record open for its ``fsync`` raises - ``PermissionError``. + still unlinked along with the rest. Both the listing and each unlink + tolerate a peer that removed the directory or the record first, matching + the ``FileNotFoundError`` handling in :meth:`contains`, so a teardown + that finds its work already done returns instead of raising out of + ``_retire_absent_quarantine``. What stays unclosed on purpose: on Windows + a peer's ``mark()`` still holding the fresh record open for its + ``fsync`` raises ``PermissionError``, which is a real storage failure + rather than an absent fence. """ directory = self._directory(info) if not self._exists(directory): return - for record in directory.iterdir(): - record.unlink() + for record in self._records(directory): + self._drop(record) try: directory.rmdir() except OSError: diff --git a/backend/tests/test_quarantine_retire_race_regression.py b/backend/tests/test_quarantine_retire_race_regression.py index 7805764332..28a5b4c585 100644 --- a/backend/tests/test_quarantine_retire_race_regression.py +++ b/backend/tests/test_quarantine_retire_race_regression.py @@ -9,12 +9,23 @@ teardown path. Its sibling `retire_replaced_generations()` guards the identical Deterministic: the race is replayed by wrapping `Path.rmdir` so a peer record lands between the `iterdir()` and the `rmdir()` call. +The `rmdir()` window above is closed. Two windows ahead of it are not, in both +`retire()` and `retire_replaced_generations()`: the `iterdir()` listing and each +`record.unlink()`. A peer running the same teardown one step ahead can remove the +directory, or one record, in exactly those windows, and the `FileNotFoundError` +then escapes the unwrapped `_retire_absent_quarantine()` call at the top of +sandbox creation. The class already reports that interleaving as a normal answer +two methods away: `_exists()` maps it to "no quarantine" and `contains()` catches +it on its own listing. The cases below replay each remaining window by +monkeypatching `Path.iterdir` or `Path.unlink`, so they stay deterministic. """ from __future__ import annotations from pathlib import Path +import pytest + from deerflow.community.aio_sandbox.quarantine import SandboxQuarantine from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo @@ -45,3 +56,101 @@ def test_retire_survives_a_peer_fencing_between_listing_and_rmdir(tmp_path, monk assert directory.exists() assert (directory / store._key("gen-2")).exists() + + +@pytest.mark.parametrize("method", ["retire", "retire_replaced_generations"]) +def test_teardown_survives_the_directory_disappearing_before_the_listing(tmp_path, monkeypatch, method): + """A peer finishing its teardown first leaves nothing to remove, and must not raise. + + `_exists()` reported the directory, then the peer unlinked every record and + removed it, so the listing itself fails with FileNotFoundError. The + `contains()` listing two methods above already treats this exact + interleaving as "the fence is gone"; neither teardown loop does. + """ + store = SandboxQuarantine(tmp_path, "provisioner") + info = SandboxInfo("thread-sandbox", "http://sandbox", container_id="gen-1") + store.mark(info) + directory = tmp_path / store._key("provisioner") / store._key("thread-sandbox") + + real_iterdir = Path.iterdir + raced = [] + + def racing_iterdir(self): + if self == directory and not raced: + raced.append(self) + # The peer's retire() completes inside our own teardown window. + for record in list(real_iterdir(self)): + record.unlink() + Path.rmdir(self) + return real_iterdir(self) + + monkeypatch.setattr(Path, "iterdir", racing_iterdir) + + getattr(store, method)(info) # must not raise + + assert not directory.exists() + + +def test_retire_survives_a_record_disappearing_mid_loop(tmp_path, monkeypatch): + """A peer's unlink landing between the listing and ours must not abort the loop. + + Two generations are fenced. The peer drops the first record after `retire()` + listed the directory, so `record.unlink()` fails on a name that is already + gone while the second record and the `rmdir()` still need the loop to finish. + """ + store = SandboxQuarantine(tmp_path, "provisioner") + info = SandboxInfo("thread-sandbox", "http://sandbox", container_id="gen-1") + for container_id in ("gen-1", "gen-2"): + store.mark(SandboxInfo("thread-sandbox", "http://sandbox", container_id=container_id)) + directory = tmp_path / store._key("provisioner") / store._key("thread-sandbox") + raced_record = directory / store._key("gen-1") + + real_unlink = Path.unlink + raced = [] + + def racing_unlink(self): + if self == raced_record and not raced: + raced.append(self) + real_unlink(self) + return real_unlink(self) + + monkeypatch.setattr(Path, "unlink", racing_unlink) + + store.retire(info) # must not raise + + assert not raced_record.exists() + # The loop reached the second record, so the emptied directory was removed. + assert not directory.exists() + + +def test_retire_replaced_generations_keeps_the_live_fence_when_a_peer_drops_a_stale_one_first(tmp_path, monkeypatch): + """The prune path must survive losing a stale record mid-loop and still keep the live fence. + + `retire_replaced_generations()` skips the live generation's record by design. + A peer unlinking the stale record first raises inside the loop, before the + live record is even reached, so the prune aborts with FileNotFoundError. + """ + store = SandboxQuarantine(tmp_path, "provisioner") + stale = SandboxInfo("thread-sandbox", "http://sandbox", container_id="gen-1") + live = SandboxInfo("thread-sandbox", "http://sandbox", container_id="gen-2") + store.mark(stale) + store.mark(live) + directory = tmp_path / store._key("provisioner") / store._key("thread-sandbox") + raced_record = directory / store._key("gen-1") + + real_unlink = Path.unlink + raced = [] + + def racing_unlink(self): + if self == raced_record and not raced: + raced.append(self) + real_unlink(self) + return real_unlink(self) + + monkeypatch.setattr(Path, "unlink", racing_unlink) + + store.retire_replaced_generations(live) # must not raise + + assert not raced_record.exists() + assert (directory / store._key("gen-2")).exists() + assert directory.exists()