perf(merge): ⚡ derive global ids from tile counts, dropping two full passes

The merge rewrote the whole staged store twice for bookkeeping. Pass 0
(_make_globally_unique) streamed every chunk and renumbered it in place, so
tile-local 1..N ids stopped colliding. relabel_sequential_zarr then made a
second full read+write to compact the ids, building a Python set of every
label id on the way (~60 bytes each).

Neither pass needs to touch the volume. stage_tile already knows how many
labels it wrote, so it now renumbers its trimmed output densely and returns
that count; the segment job records it in the .done marker, which until now
was an empty file. Global ids are then offset + local with offset an
exclusive cumulative sum over those counts -- O(n_tiles) arithmetic. The
offsets are applied where they are needed instead: on the two chunks either
side of each boundary during the scan, and per chunk in the relabel worker.

That makes the id domain dense by construction, so the sequential renumber
becomes np.unique over the LUT -- length = object count, not voxels -- and
folds into the lookup the relabel pass already performs. Passing no counts
keeps the old streaming path, so the dask-array entry point and any external
caller are unaffected.

The relabel worker also stops widening whole blocks to int64, gathering only
the non-zero ids instead; labels are sparse, so its peak drops accordingly.

Verified on a synthetic workflow run: byte-identical labels and object count
to the previous implementation, and a test asserts the two paths agree.

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

Author: lguerard

src/patchworks/_distributed.py

@@ -18,6 +18,8 @@
 import numpy as np
 import zarr
 
+from ._relabel import relabel_sequential_array
+
 Overlap = Union[int, Sequence[int]]
 
 
@@ -158,7 +160,11 @@ def stage_tile(
     Returns
     -------
     int
-        The processed tile *index*.
+        Number of labels this tile wrote, i.e. its ids are exactly ``1..n``.
+        Record it (the workflow puts it in the tile's ``.done`` marker): with
+        one count per tile the merge can derive every tile's global id range
+        by a cumulative sum, instead of rewriting the whole store to make the
+        ids unique.
     """
     shape = image.shape
     sl = spatial_tiles(shape, tile_shape)[index]
@@ -175,8 +181,13 @@ def stage_tile(
         slice(left, out.shape[i] - right)
         for i, (left, right) in enumerate(trims)
     )
-    # Local labels (1..N) are fine here — they collide across tiles, but the
-    # merge's first pass makes them globally unique before stitching.
+    trimmed = out[sel]
+    # Local labels collide across tiles; the merge resolves that. Renumber to a
+    # dense 1..n first: trimming the halo can drop objects entirely, and the
+    # merge's offset arithmetic needs each tile's ids to be exactly 1..n with
+    # no gaps so that `offset[tile] + local` is globally unique AND compact.
+    trimmed = relabel_sequential_array(trimmed)
+    n_labels = int(trimmed.max())
     dst = zarr.open_group(str(stage_path), mode="r+")[component]
-    dst[sl] = out[sel].astype(dst.dtype)
-    return index
+    dst[sl] = trimmed.astype(dst.dtype)
+    return n_labels