fix(gc): coverage-based orphan matching for directory-valued objects

The file-level orphan matching introduced earlier in this PR broke
DIRECTORY-valued objects (e.g. Zarr stores via <object@>): a row's metadata
references the directory PREFIX, while list_schema_paths enumerates the chunk
FILES under it — exact set difference therefore flagged every chunk of a LIVE
store as an orphan (empirically confirmed), and collect() would delete live
data. Folder objects also write a {path}.manifest.json sidecar, which the
listing skipped — orphaned folders would leak their manifests forever.

Orphan detection is now coverage-based via _is_covered(): a stored file is
referenced when it IS a referenced path (single-file object), lies UNDER a
referenced path (directory object), or is the .manifest.json sidecar of a
referenced object. Manifests are now listed (so orphaned ones are reclaimed);
lookup is O(path-depth) ancestor-prefix set membership per file.

Note the pre-existing code also mishandled directory objects (the PK-level
directory holding the manifest was orphan-flagged, so collect() would rm -r
the live object) — neither version was safe; this makes the new file-level
model correct for both single-file and directory objects.

Tests: a live folder object's chunks+manifest are never flagged (fails before
this commit); after deleting one of two rows, collect() reclaims exactly the
orphaned folder's chunks+manifest and preserves the survivor's. Full gc suite
34/34.

Author: dimitri-yatsenko

src/datajoint/gc.py

@@ -387,12 +387,14 @@ def list_schema_paths(store_name: str | None = None, config=None) -> dict[str, i
     """
     List all schema-addressed object files in storage.
 
-    Enumerates the individual object **files** written by schema-addressed
-    codecs, whose paths follow ``{schema}/{table}/{pk}/{field}_{token}[.ext]``.
-    Returning file paths (rather than the enclosing directory) is what lets
-    them match the file paths recorded in each row's stored metadata — the two
-    must be comparable for orphan detection, and per-token granularity lets
-    superseded versions be reclaimed while the current token is retained.
+    Enumerates the individual **files** written by schema-addressed codecs.
+    A single-file object is one file at ``{schema}/{table}/{pk}/{field}_{token}[.ext]``;
+    a directory-valued object (e.g. a Zarr store) is many files under that
+    path prefix, plus a ``{path}.manifest.json`` sidecar beside it — all are
+    listed. Orphan detection matches these files against each row's referenced
+    object path via :func:`_is_covered` (exact, ancestor-prefix, or
+    manifest-sidecar), so superseded per-token versions are reclaimable while
+    every file belonging to a live object — including its manifest — is kept.
 
     Parameters
     ----------
@@ -419,10 +421,11 @@ def list_schema_paths(store_name: str | None = None, config=None) -> dict[str, i
                 continue
 
             for filename in files:
-                # Skip manifest sidecars (mirrors list_stored_hashes)
-                if filename.endswith(".manifest.json"):
-                    continue
-
+                # Manifest sidecars ({object}.manifest.json, written for
+                # folder-valued objects) are deliberately INCLUDED: they are
+                # owned by their object and must be reclaimed with it when the
+                # object is orphaned. _is_covered() keeps a live object's
+                # manifest out of the orphan set.
                 file_path = f"{root}/{filename}"
                 relative_path = file_path.replace(full_prefix, "").lstrip("/")
 
@@ -494,6 +497,32 @@ def delete_schema_path(path: str, store_name: str | None = None, config=None) ->
     return False
 
 
+def _is_covered(path: str, referenced: set[str]) -> bool:
+    """
+    Return True if a stored file is accounted for by a referenced object path.
+
+    A stored file is covered when:
+
+    - it IS a referenced path (single-file object, e.g. ``field_token.npy``);
+    - it lies UNDER a referenced path (directory-valued object, e.g. a Zarr
+      store ``field_token/`` whose stored form is many chunk files); or
+    - it is the ``.manifest.json`` sidecar of a referenced object (written
+      alongside folder-valued objects).
+
+    Uses O(path-depth) ancestor-prefix lookups against the referenced set.
+    """
+    if path.endswith(".manifest.json"):
+        path = path[: -len(".manifest.json")]
+    if path in referenced:
+        return True
+    idx = path.rfind("/")
+    while idx > 0:
+        if path[:idx] in referenced:
+            return True
+        idx = path.rfind("/", 0, idx)
+    return False
+
+
 def scan(
     *schemas: "Schema",
     store_name: str | None = None,
@@ -545,7 +574,10 @@ def scan(
     # --- Schema-addressed storage ---
     schema_paths_referenced = scan_schema_references(*schemas, store_name=store_name, verbose=verbose)
     schema_paths_stored = list_schema_paths(store_name, config=_config)
-    orphaned_paths = set(schema_paths_stored.keys()) - schema_paths_referenced
+    # Coverage, not exact set difference: a referenced path may be a
+    # DIRECTORY-valued object (e.g. a Zarr store), whose stored form is many
+    # files under that prefix, plus a `.manifest.json` sidecar beside it.
+    orphaned_paths = {p for p in schema_paths_stored if not _is_covered(p, schema_paths_referenced)}
     schema_paths_orphaned_bytes = sum(schema_paths_stored.get(p, 0) for p in orphaned_paths)
 
     return {

tests/integration/test_gc.py

@@ -545,3 +545,76 @@ def test_custom_codec_survives_collect(self, schema_custom):
         stored_after = set(gc.list_schema_paths("local", config=schema_custom.connection._config))
         assert live_path in stored_after, f"collect() deleted the live custom-codec file {live_path} (#1469)"
         assert not (deleted_paths & stored_after), f"deleted row's orphan not reclaimed: {deleted_paths & stored_after}"
+
+
+class TestDirectoryObjects:
+    """Directory-valued objects (e.g. Zarr stores) — the referenced metadata
+    path is a directory PREFIX whose stored form is many chunk files plus a
+    `{path}.manifest.json` sidecar. Orphan matching must be coverage-based
+    (exact / ancestor-prefix / manifest), not exact set difference: with exact
+    matching, every chunk of a LIVE store is misclassified as an orphan and
+    collect() deletes live data."""
+
+    @pytest.fixture
+    def schema_dirobj(self, connection_test, prefix, mock_stores):
+        schema = dj.Schema(
+            f"{prefix}_test_gc_dirobj",
+            context={"GcObjectTest": GcObjectTest},
+            connection=connection_test,
+        )
+        schema(GcObjectTest)
+        yield schema
+        schema.drop()
+
+    @staticmethod
+    def _make_store_dir(tmp_path, name, n_chunks=3):
+        d = tmp_path / name
+        d.mkdir()
+        (d / ".zarray").write_bytes(b"{}")
+        for i in range(n_chunks):
+            (d / f"0.{i}").write_bytes(f"chunk{i}".encode())
+        return d
+
+    def test_live_directory_object_not_orphaned(self, schema_dirobj, tmp_path):
+        """A live folder object's chunk files and manifest must be covered by
+        its referenced prefix — none may appear in orphaned_paths."""
+        GcObjectTest.insert1({"rid": 1, "results": str(self._make_store_dir(tmp_path, "live.zarr"))})
+
+        refs = gc.scan_schema_references(schema_dirobj, store_name="local")
+        assert len(refs) == 1
+        ref = next(iter(refs))
+
+        stored = gc.list_schema_paths("local", config=schema_dirobj.connection._config)
+        chunk_files = [p for p in stored if p.startswith(ref + "/")]
+        assert len(chunk_files) >= 4, f"expected the folder's files under {ref}; got {sorted(stored)}"
+        assert f"{ref}.manifest.json" in stored, "manifest sidecar must be listed (it is reclaimable state)"
+
+        stats = gc.scan(schema_dirobj, store_name="local")
+        flagged = [p for p in stats["orphaned_paths"] if p == ref or p.startswith(ref + "/") or p == f"{ref}.manifest.json"]
+        assert not flagged, f"live directory object misclassified as orphaned: {flagged}"
+
+    def test_orphaned_directory_object_fully_reclaimed(self, schema_dirobj, tmp_path):
+        """After deleting one row, collect() must reclaim that folder object's
+        chunks AND manifest while leaving the surviving row's folder intact."""
+        GcObjectTest.insert1({"rid": 1, "results": str(self._make_store_dir(tmp_path, "gone.zarr"))})
+        GcObjectTest.insert1({"rid": 2, "results": str(self._make_store_dir(tmp_path, "kept.zarr"))})
+
+        refs_all = gc.scan_schema_references(schema_dirobj, store_name="local")
+        assert len(refs_all) == 2
+
+        (GcObjectTest & {"rid": 1}).delete(prompt=False)
+        refs_live = gc.scan_schema_references(schema_dirobj, store_name="local")
+        assert len(refs_live) == 1
+        live_ref = next(iter(refs_live))
+        dead_ref = next(iter(refs_all - refs_live))
+
+        gc.collect(schema_dirobj, store_name="local", dry_run=False)
+
+        stored_after = set(gc.list_schema_paths("local", config=schema_dirobj.connection._config))
+        live_files = {p for p in stored_after if p.startswith(live_ref + "/")}
+        assert len(live_files) >= 4, f"live folder object lost files: {sorted(stored_after)}"
+        assert f"{live_ref}.manifest.json" in stored_after, "live object's manifest must survive collect()"
+
+        dead_files = {p for p in stored_after if p == dead_ref or p.startswith(dead_ref + "/")}
+        assert not dead_files, f"orphaned folder object not reclaimed: {sorted(dead_files)}"
+        assert f"{dead_ref}.manifest.json" not in stored_after, "orphaned object's manifest must be reclaimed"