fix(gc): don't act on partial storage listings

Three compounding completeness bugs in GarbageCollector.list_hash_paths
and list_schema_paths that could cause collect(dry_run=False) to silently
delete live external-store files:

1. str.replace(full_root, '') strips every occurrence — corrupts the
   relative path when the store-root string recurs (e.g., S3 bucket name
   == schema name). Uses prefix slice instead.

2. fs.size() error handling diverged: list_hash_paths dropped the entry
   silently (invisible to orphan detection); list_schema_paths kept it
   at 0 silently. Now both keep at 0 and log a warning.

3. fs.walk errors returned a partial-or-empty stored dict with only a
   log-warning signal — indistinguishable from a clean store. collect()
   then computed orphaned = stored - referenced against incomplete data
   and permanently deleted the 'orphans'. Adds a _scan_errors tracker
   on the collector; collect(dry_run=False) refuses to delete when set,
   with a clear DataJointError. New scan_errors list added to the stats
   dict (empty on clean runs).

Tests: 3 new TestScanErrorGuard cases; full local suite green
(42/42 test_gc.py, 319/319 unit, 626 integration — no new failures).

Follow-up to the completeness discussion on #1478.

Author: MilagrosMarin

src/datajoint/gc.py

@@ -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)
@@ -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("/")
                     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]:
@@ -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
 
     # ------------------------------------------------------------------ #
@@ -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.
@@ -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:
@@ -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,
         }

tests/integration/test_gc.py

@@ -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:
+    """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]