Fix lake-wide capacity-scan topology race before v0.7.4

src/quant_data_kit/capture_v2/epoch.py

@@ -30,6 +30,7 @@
 from quant_data_kit.data_lake import (
     RawObjectReference,
     StoragePolicy,
+    _capacity_tree_lock,
     validate_raw_reference,
     write_normalized_batches,
 )
@@ -1548,16 +1549,17 @@ def _seal_part(self) -> None:
         final_name = f"part-{index:08d}-sha256-{digest}.ndjson"
         final_path = self.root / final_name
         checked_final = _validate_safe_path(self.hot_root, final_path, allow_missing=True)
-        try:
-            os.link(self._open_path, checked_final)
-        except FileExistsError as exc:
-            raise ValidationError(
-                f"Normalized journal part already exists: {checked_final}"
-            ) from exc
+        with _capacity_tree_lock(self.hot_root):
+            try:
+                os.link(self._open_path, checked_final)
+            except FileExistsError as exc:
+                raise ValidationError(
+                    f"Normalized journal part already exists: {checked_final}"
+                ) from exc
+            self._open_path.unlink()
         _validate_safe_path(self.hot_root, checked_final, allow_missing=False)
         if _sha256_file(checked_final, trusted_root=self.hot_root) != digest:
             raise ValidationError(f"Normalized sealed journal part hash changed: {checked_final}")
-        self._open_path.unlink()
         _fsync_directory(self.root)
         self._parts.append(
             EpochPart(

src/quant_data_kit/capture_v2/storage.py

@@ -30,6 +30,8 @@
 from quant_data_kit.data_lake import (
     RawObjectManifest,
     StoragePolicy,
+    _capacity_tree_lock,
+    _unlink_tree_entry,
     evaluate_capacity,
     load_raw_object,
     write_raw_bytes,
@@ -758,6 +760,21 @@ def _atomic_immutable_write(path: Path, body: bytes, *, root: Path | None = None
     temporary = parent / f".{path.name}.{uuid.uuid4().hex}.tmp"
     _validate_safe_path(trusted_root, temporary, allow_missing=True)
     expected_hash = hashlib.sha256(body).hexdigest()
+    relative = checked_path.relative_to(trusted_root)
+    coordinated = bool(
+        relative.parts
+        and relative.parts[0] in {"capture", "curated", "normalized", "quarantine", "raw"}
+    )
+
+    def publish_link() -> None:
+        try:
+            os.link(temporary, checked_path)
+            _fsync_directory(parent)
+        except FileExistsError:
+            pass
+        if coordinated and temporary.exists():
+            temporary.unlink()
+
     try:
         with temporary.open("xb") as stream:
             stream.write(body)
@@ -766,19 +783,22 @@ def _atomic_immutable_write(path: Path, body: bytes, *, root: Path | None = None
         if _sha256_file(temporary) != expected_hash:
             raise ValidationError(f"immutable staging hash mismatch: {temporary}")
         _validate_safe_path(trusted_root, parent, allow_missing=False)
-        try:
-            os.link(temporary, checked_path)
-            _fsync_directory(parent)
-        except FileExistsError:
-            pass
+        if coordinated:
+            with _capacity_tree_lock(trusted_root):
+                publish_link()
+        else:
+            publish_link()
         _validate_safe_path(trusted_root, checked_path, allow_missing=False)
         if not checked_path.is_file() or _sha256_file(checked_path) != expected_hash:
             raise ValidationError(
                 f"immutable archive path already contains different bytes: {checked_path}"
             )
     finally:
         if temporary.exists():
-            temporary.unlink()
+            if coordinated:
+                _unlink_tree_entry(trusted_root, temporary)
+            else:
+                temporary.unlink()
 
 
 class DurableAuditStore:

src/quant_data_kit/curated.py

@@ -4,7 +4,6 @@
 
 import hashlib
 import json
-import os
 import re
 from collections import defaultdict
 from collections.abc import Iterable, Mapping
@@ -22,6 +21,7 @@
     StoragePolicy,
     _atomic_write_bytes,
     _mkdir_in_lake,
+    _publish_tree_entry,
     _resolved_lake_root,
     _stable_staging_directory,
     _validate_lake_path,
@@ -388,8 +388,7 @@ def _publish_curated_snapshot(
         if existing != snapshot:
             raise ValidationError(f"Curated snapshot collision: {snapshot_dir}")
     else:
-        require_collection_capacity(lake_root, projected_write_bytes=0, policy=policy)
-        os.replace(stage, snapshot_dir)
+        _publish_tree_entry(lake_root, stage, snapshot_dir, policy=policy)
     if not revision_path.exists():
         _atomic_write_bytes(
             lake_root,
@@ -429,11 +428,6 @@ def _write_curated_bars(
         groups[(str(record["trading_day"]), str(record["instrument_id"]))].append(record)
 
     estimated_bytes = sum(len(_canonical(_json_value(item))) for item in records)
-    require_collection_capacity(
-        lake_root,
-        projected_write_bytes=estimated_bytes,
-        policy=policy,
-    )
     curated_root = _mkdir_in_lake(lake_root, lake_root / "curated" / dataset)
     staging_root = curated_root / "staging"
     partition_items: list[CuratedPartition] = []
@@ -444,6 +438,11 @@ def _write_curated_bars(
         namespace="curated-revision",
         identity=revision_identity,
     ) as stage:
+        require_collection_capacity(
+            lake_root,
+            projected_write_bytes=estimated_bytes,
+            policy=policy,
+        )
         for (trading_date, instrument_id), group in sorted(groups.items()):
             ordered = sorted(group, key=lambda row: (row["event_time"], row["event_id"]))
             table = pa.Table.from_pylist(