Simplify multi-actor stateful tests and fix delete+recreate rebase

- Refactor ClaimTracker: remove _protected, _try_clear, _stash_session, _on_clear
- Add delete_dir() and rewrite_manifests() to MultiActorModel
- Simplify IcechunkModel with cleaner lifecycle methods
- Extract actor logic to multi-actor case only
- Fix delete+recreate in rebase (Rust conflict detector)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: ianhi

icechunk-python/python/icechunk/testing/multi_actor.py

@@ -0,0 +1,102 @@
+"""Multi-actor coordinator for stateful tests.
+
+Manages multiple actors sharing a repository. Each actor has its own
+IcechunkModel. The coordinator swaps the active model and computes
+the blocked set (paths other actors have modified).
+
+The test class's self.model is always a plain IcechunkModel — the
+coordinator just tracks which one is active and what's blocked.
+The committed baseline lives on the test (self.committed), not here.
+"""
+
+from __future__ import annotations
+
+from pathlib import PurePosixPath
+
+from icechunk.testing.stateful_models import IcechunkModel
+
+
+class MultiActorCoordinator:
+    """Coordinates multiple actors sharing a repository.
+
+    Holds the actor dict. Computes the blocked set from other actors'
+    changes. The test class owns self.model and self.committed directly.
+    """
+
+    @staticmethod
+    def _has_blocked_ancestor(path: str, blocked: set[str]) -> bool:
+        """True if path or any ancestor of path is in the blocked set."""
+        pp = PurePosixPath(path)
+        return any(pp.is_relative_to(p) for p in blocked)
+
+    @staticmethod
+    def _has_blocked_relative(path: str, blocked: set[str]) -> bool:
+        """True if any blocked path is an ancestor, descendant, or equal."""
+        pp = PurePosixPath(path)
+        return any(
+            pp.is_relative_to(p) or PurePosixPath(p).is_relative_to(path) for p in blocked
+        )
+
+    def __init__(self) -> None:
+        self.current_actor: str = ""
+        self._actors: dict[str, IcechunkModel] = {}
+        self.blocked: set[str] = set()
+
+    @property
+    def current(self) -> IcechunkModel:
+        """The current actor's model."""
+        return self._actors[self.current_actor]
+
+    @current.setter
+    def current(self, model: IcechunkModel) -> None:
+        self._actors[self.current_actor] = model
+
+    def recompute_blocked(self, committed: IcechunkModel | None) -> None:
+        """Recompute the blocked set for the current actor.
+
+        The blocked set is:
+        1. Other actors' uncommitted changes (structural + data)
+        2. Main drift: paths that changed on committed since this actor's baseline
+        """
+        actor = self.current_actor
+        result: set[str] = set()
+
+        for name, state in self._actors.items():
+            if name != actor:
+                result |= state.changes()
+
+        current = self._actors.get(actor)
+        if current is not None and committed is not None:
+            committed_nodes = frozenset(committed.all_arrays) | frozenset(
+                committed.all_groups
+            )
+            result |= committed_nodes ^ current.baseline
+
+        self.blocked = result
+
+    # ── Actor management ─────────────────────────────────────────────
+
+    def switch(self, actor: str, committed: IcechunkModel | None) -> None:
+        """Switch to an existing actor and recompute the blocked set."""
+        self.current_actor = actor
+        self.recompute_blocked(committed)
+
+    def init_actor(
+        self, committed: IcechunkModel | None, bootstrap: str = "default"
+    ) -> None:
+        """Promote the bootstrap model to the current actor's model."""
+        self._actors[self.current_actor] = self._actors.pop(bootstrap)
+        if committed is not None:
+            self.current.sync_baseline(committed)
+        self.recompute_blocked(committed)
+
+    def other_actors_clean(self) -> bool:
+        """True if no other actor has uncommitted changes."""
+        return all(
+            not state.changes()
+            for name, state in self._actors.items()
+            if name != self.current_actor
+        )
+
+    def can_add(self, path: str) -> bool:
+        return not self._has_blocked_ancestor(path, self.blocked)

icechunk-python/python/icechunk/testing/stateful_models.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+from zarr.core.buffer import default_buffer_prototype
+from zarr.storage import MemoryStore
+
+PROTOTYPE = default_buffer_prototype()
+
+
+class IcechunkModel(MemoryStore):
+    """MemoryStore with move, copy, and lifecycle methods for testing.
+
+    Tracks all_arrays/all_groups alongside store data so that copy()
+    preserves them.
+    """
+
+    spec_version: int
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.all_arrays: set[str] = set()
+        self.all_groups: set[str] = set()
+        self.baseline: frozenset[str] = frozenset()
+        self.claims: set[str] = set()
+
+    @classmethod
+    def from_store(cls, store: MemoryStore) -> IcechunkModel:
+        """Promote a plain MemoryStore to an IcechunkModel."""
+        model = cls()
+        model._store_dict = store._store_dict
+        return model
+
+    def _claim_node(self, key: str) -> None:
+        """Record which array or group a key belongs to."""
+        for array in self.all_arrays:
+            if key.startswith(array + "/"):
+                self.claims.add(array)
+                return
+        for group in self.all_groups:
+            if key.startswith(group + "/"):
+                self.claims.add(group)
+                return
+
+    async def set(
+        self, key: str, value: object, byte_range: tuple[int, int] | None = None
+    ) -> None:
+        self._claim_node(key)
+        await super().set(key, value, byte_range)
+
+    async def delete(self, key: str) -> None:
+        self._claim_node(key)
+        await super().delete(key)
+
+    async def move(self, source: str, dest: str) -> None:
+        """Move all keys from source to dest.
+
+        Store keys always have form "node/zarr.json" or "node/c/...", never bare "node".
+        """
+        all_keys = [k async for k in self.list_prefix("")]
+        keys_to_move = [k for k in all_keys if k.startswith(source + "/")]
+        for old_key in keys_to_move:
+            new_key = dest + old_key[len(source) :]
+            data = await self.get(old_key, prototype=PROTOTYPE)
+            if data is not None:
+                await self.set(new_key, data)
+                await self.delete(old_key)
+
+    async def copy(self) -> IcechunkModel:
+        """Create a copy of this store (data + arrays + groups)."""
+        new_store = IcechunkModel()
+        new_store.spec_version = self.spec_version
+        new_store.all_arrays = self.all_arrays.copy()
+        new_store.all_groups = self.all_groups.copy()
+        async for key in self.list_prefix(""):
+            data = await self.get(key, prototype=PROTOTYPE)
+            if data is not None:
+                await new_store.set(key, data)
+        return new_store
+
+    # things for tracking changes relative to committed baseline
+    @property
+    def nodes(self) -> frozenset[str]:
+        """Current set of all node paths in this model."""
+        return frozenset(self.all_arrays) | frozenset(self.all_groups)
+
+    def changes(self) -> set[str]:
+        """Paths modified since last sync: structural diff + data claims."""
+        return set(self.claims) | (self.nodes ^ self.baseline)
+
+    def sync_baseline(self, committed) -> None:
+        """Record committed state as this actor's baseline."""
+        self.baseline = frozenset(committed.all_arrays) | frozenset(committed.all_groups)
+
+    # ── Lifecycle methods ─────────────────────────────────────────────
+
+    async def commit(self) -> IcechunkModel:
+        """Snapshot this model as the new committed baseline.
+
+        Returns the committed copy; caller stores it as shared state.
+        """
+        committed = await self.copy()
+        self.claims.clear()
+        self.sync_baseline(committed)
+        return committed
+
+    async def rebase(self, committed: IcechunkModel) -> IcechunkModel:
+        """Merge committed baseline with local changes.
+
+        Returns a new model with the merge result; caller swaps it in.
+        """
+        merged = await committed.copy()
+        my_changes = self.changes()
+
+        merged.all_arrays = (committed.all_arrays - my_changes) | (
+            my_changes & self.all_arrays
+        )
+        merged.all_groups = (committed.all_groups - my_changes) | (
+            my_changes & self.all_groups
+        )
+
+        for path in my_changes:
+            keys_to_delete = [k async for k in merged.list_prefix(path + "/")]
+            for key in keys_to_delete:
+                await merged.delete(key)
+            async for key in self.list_prefix(path + "/"):
+                data = await self.get(key, prototype=PROTOTYPE)
+                if data is not None:
+                    await merged.set(key, data)
+
+        merged.sync_baseline(committed)
+        return merged
+
+    async def new_session(self, committed: IcechunkModel) -> IcechunkModel:
+        """Reset to the committed baseline, discarding local changes."""
+        fresh = await committed.copy()
+        fresh.claims.clear()
+        fresh.sync_baseline(committed)
+        return fresh
+
+    async def rewrite_manifests(self, committed: IcechunkModel) -> IcechunkModel:
+        """Reset to committed baseline (rewrite_manifests doesn't change data)."""
+        fresh = await committed.copy()
+        fresh.claims.clear()
+        fresh.sync_baseline(committed)
+        return fresh