Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 21 additions & 5 deletions src/datajoint/gc.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,10 @@ def __init__(self, *schemas: "Schema", store: str | None = None, config=None) ->
spec = self.config.get_store_spec(store)
self._hash_prefix = spec["hash_prefix"].strip("/") # settings applies the "_hash" default
self._schema_prefix = spec["schema_prefix"].strip("/") # ... "_schema" default
# Per-collect() scan errors (walk failures in list_*_paths). Non-empty
# blocks destructive deletion in collect(dry_run=False). Reset at the
# start of each collect() call.
self._scan_errors: list[str] = []

# ------------------------------------------------------------------ #
# References — extracted from database metadata (per-schema connection)
Expand Down Expand Up @@ -205,15 +209,17 @@ def list_hash_paths(self, schema_name: str) -> dict[str, int]:
if not _BASE32_HASH.match(filename):
continue
file_path = f"{root}/{filename}"
relative_path = file_path[len(full_root) :].lstrip("/")
Comment thread
dimitri-yatsenko marked this conversation as resolved.
try:
relative_path = file_path.replace(full_root, "").lstrip("/")
stored[relative_path] = self.backend.fs.size(file_path)
except Exception:
pass
except Exception as e:
logger.warning(f"Could not size {file_path}: {e}")
stored[relative_path] = 0
except FileNotFoundError:
pass # the hash section does not exist yet
except Exception as e:
logger.warning(f"Error listing stored hashes: {e}")
self._scan_errors.append(f"list_hash_paths({schema_name}): {e}")
return stored

def list_schema_paths(self, schema_name: str) -> dict[str, int]:
Expand Down Expand Up @@ -252,15 +258,17 @@ def list_schema_paths(self, schema_name: str) -> dict[str, int]:
# reclaimed with it. _is_covered() keeps a live object's
# manifest out of the orphan set.
file_path = f"{root}/{filename}"
relative_path = file_path.replace(full_root, "").lstrip("/")
relative_path = file_path[len(full_root) :].lstrip("/")
try:
stored[relative_path] = self.backend.fs.size(file_path)
except Exception:
except Exception as e:
logger.warning(f"Could not size {file_path}: {e}")
stored[relative_path] = 0
except FileNotFoundError:
pass # this schema's section does not exist yet
except Exception as e:
logger.warning(f"Error listing stored schemas: {e}")
self._scan_errors.append(f"list_schema_paths({schema_name}): {e}")
return stored

# ------------------------------------------------------------------ #
Expand Down Expand Up @@ -327,6 +335,7 @@ def collect(self, dry_run: bool = True, verbose: bool = False) -> dict[str, Any]
- hash_paths_deleted / schema_paths_deleted / deleted / bytes_freed (0 when
dry_run) / errors / dry_run
"""
self._scan_errors = [] # reset per-call; populated by list_*_paths below
# References can be gathered across all schemas at once: because paths
# embed the schema, a file under schema X is only ever covered by an X
# reference, so combined coverage equals per-schema coverage.
Expand All @@ -350,6 +359,12 @@ def collect(self, dry_run: bool = True, verbose: bool = False) -> dict[str, Any]
errors = 0

if not dry_run:
if self._scan_errors:
raise DataJointError(
f"Refusing to delete: {len(self._scan_errors)} scan error(s) — "
f"partial listing risks classifying live files as orphaned. "
f"Errors: {self._scan_errors}"
)
# The size maps from the scan above are reused for the byte tally —
# no second listing pass.
for path in orphaned_hash_paths:
Expand Down Expand Up @@ -393,5 +408,6 @@ def collect(self, dry_run: bool = True, verbose: bool = False) -> dict[str, Any]
"deleted": hash_paths_deleted + schema_paths_deleted,
"bytes_freed": bytes_freed,
"errors": errors,
"scan_errors": list(self._scan_errors),
"dry_run": dry_run,
}
81 changes: 81 additions & 0 deletions tests/integration/test_gc.py
Original file line number Diff line number Diff line change
Expand Up @@ -762,3 +762,84 @@ def test_collect_one_schema_never_touches_the_other(self, two_schemas):
assert set(collector.list_hash_paths(b.database)) == b_hashes_before
assert set(collector.list_schema_paths(b.database)) == b_paths_before
assert len(b_obj()) == 1 # b's row (and its object) untouched


class TestScanErrorGuard:
Comment thread
dimitri-yatsenko marked this conversation as resolved.
"""A walk failure during list_*_paths (anything other than FileNotFoundError,
which just means the section doesn't exist yet) leaves the stored listing
partial. Comparing partial listings against complete reference sets would
classify live files as orphaned — silent data loss under dry_run=False.
The scan-error guard captures such errors and refuses to delete."""

def test_scan_errors_reset_between_collect_calls(self):
"""_scan_errors is reset at the start of each collect() call, so an
error from a prior collect() never carries over into the next report."""
with ExitStack() as es:
for p in TestCollect._patch_data(TestCollect):
es.enter_context(p)
es.enter_context(patch("datajoint.gc.delete_path"))
collector = _mock_collector()
# Simulate a leftover error from a previous run.
collector._scan_errors = ["stale error from previous call"]

stats = collector.collect() # dry_run=True

assert stats["scan_errors"] == [], "scan_errors must be reset at the start of collect()"
assert collector._scan_errors == [], "the collector's _scan_errors must be cleared for the new run"

def test_dry_run_false_raises_when_scan_errors_present(self):
"""collect(dry_run=False) must refuse to delete when list_*_paths hit a
non-FileNotFoundError walk failure — partial listing means live files
could be misclassified as orphans."""
# Real walk failure: mock fs.walk on list_hash_paths to raise a
# non-FileNotFoundError, exercising the outer except that records the
# scan error.
with ExitStack() as es:
es.enter_context(patch.object(gc.GarbageCollector, "hash_references", return_value=set()))
es.enter_context(patch.object(gc.GarbageCollector, "schema_references", return_value=set()))

# Real list_hash_paths runs and hits fs.walk failure; list_schema_paths
# is stubbed to a clean empty listing so only the hash walk errors.
es.enter_context(patch.object(gc.GarbageCollector, "list_schema_paths", return_value={}))

collector = _mock_collector()
collector.backend.fs.walk.side_effect = PermissionError("s3 access denied")
collector.backend._full_path.return_value = "/root"

mock_delete = es.enter_context(patch("datajoint.gc.delete_path"))
with pytest.raises(DataJointError, match="Refusing to delete"):
collector.collect(dry_run=False)

# And no deletion happened.
mock_delete.assert_not_called()

def test_scan_errors_in_stats_dict(self):
"""The returned stats dict always includes a 'scan_errors' key: empty
list on a clean scan, list of "<method>(<schema>): <exception>" strings
when a walk failed."""
# Clean case — scan_errors is an empty list.
with ExitStack() as es:
for p in TestCollect._patch_data(TestCollect):
es.enter_context(p)
es.enter_context(patch("datajoint.gc.delete_path"))
stats = _mock_collector().collect()

assert "scan_errors" in stats
assert stats["scan_errors"] == []

# Error case — list of strings identifying which listing failed.
with ExitStack() as es:
es.enter_context(patch.object(gc.GarbageCollector, "hash_references", return_value=set()))
es.enter_context(patch.object(gc.GarbageCollector, "schema_references", return_value=set()))
es.enter_context(patch.object(gc.GarbageCollector, "list_schema_paths", return_value={}))

collector = _mock_collector()
collector.schemas[0].database = "s"
collector.backend.fs.walk.side_effect = PermissionError("s3 access denied")
collector.backend._full_path.return_value = "/root"

stats = collector.collect() # dry_run=True → returns rather than raising

assert len(stats["scan_errors"]) == 1
assert stats["scan_errors"][0].startswith("list_hash_paths(s):")
assert "s3 access denied" in stats["scan_errors"][0]
Loading