cellmap_flow.finetune.session.sync

Bringing painted chunks from MinIO back to the volumes on disk.

The browser paints into MinIO; the trainer reads the volume on disk. A sync lists a volume’s chunks in the bucket with their ETags, copies the ones that changed since the last sync into the volume’s own zarr, and records what it saw in the volume’s registry record (chunk_sync_state). Local disk is the source of truth: nothing on disk is ever deleted because MinIO lacks it.

state is the dashboard’s minio_state and volumes its volume registry, passed to every call; only the lock and the failure counter of the periodic sync are kept here.

MinIO is read with s3fs (and zarr-python over it, for the metadata), which take the session’s own credentials; tensorstore, under worldarray, takes S3 credentials only from the environment.

Attributes

`logger`
`SYNC_PERIOD_SECONDS`
`SYNC_FAILURE_WARNING_INTERVAL`

Functions

`chunk_version`(entry) → str	What identifies one version of a remote chunk: its ETag, from a listing.
`copy_chunks_parallel`(s3, copy_pairs) → set	Copy `(src, dst)` chunk files from MinIO in parallel.
`sync_zarr_group_metadata`(s3, src_path, dst_path) → set	Sync a zarr group's structure and metadata from MinIO to local disk.
`diff_and_sync_chunks`(s3, s0_path, dst_s0_path, known_state) → tuple	Copy the chunks of MinIO's `s0_path` that changed since `known_state` to disk.
`volume_record`(volume_id[, zarr_path])	The registry record of `volume_id`, rebuilt from `zarr_path` if need be.
`sync_volume`(volume_id[, force, zarr_path]) → bool	Pull an annotation volume's changed chunks from MinIO to local disk.
`sync_all`([force]) → int	Sync every registered annotation volume from MinIO to local disk.
`retire_volume`(volume_id, store, *, state) → Optional[str]	Pull a volume's last strokes from MinIO, then take it out of the
`periodic_sync_once`(*, state, volumes) → None	One round of the periodic sync; failures are warned about, not hidden.
`periodic_sync`(*, state, volumes) → None	The periodic sync thread's loop: a round every SYNC_PERIOD_SECONDS, forever.
`start_periodic_sync`(state, volumes) → None	Start the periodic annotation sync thread if not already running.

Module Contents

cellmap_flow.finetune.session.sync.logger

cellmap_flow.finetune.session.sync.SYNC_PERIOD_SECONDS = 30

cellmap_flow.finetune.session.sync.SYNC_FAILURE_WARNING_INTERVAL = 300

cellmap_flow.finetune.session.sync.chunk_version(entry) → str

What identifies one version of a remote chunk: its ETag, from a listing.

The ETag changes with the content. LastModified has one-second resolution: two brush strokes to the same chunk within a second, with a sync between them, would leave the second stroke unsynced for good. LastModified and size are only the fallback for a store that lists no ETag.

cellmap_flow.finetune.session.sync.copy_chunks_parallel(s3, copy_pairs) → set

Copy (src, dst) chunk files from MinIO in parallel.

Each chunk is downloaded to a temporary file beside its destination and moved into place with os.replace, so a reader – the trainer on its LSF node reads this same volume – sees the old chunk or the new one, never a half-written file (which blosc rejects).

As many at once as the job has threads (worldarray’s thread_budget). Returns the source paths that could not be copied. Their chunks keep what was on disk, and the caller must not record them as synced.

cellmap_flow.finetune.session.sync.sync_zarr_group_metadata(s3, src_path, dst_path) → set

Sync a zarr group’s structure and metadata from MinIO to local disk.

Creates the arrays of src_path that dst_path does not have yet, and copies the attrs. An array that exists locally with a different shape, chunking or dtype is left alone and reported: re-creating it would delete every local chunk of that array, which the “never delete on-disk chunks” rule of diff_and_sync_chunks exists to prevent.

Returns the keys of the arrays whose local layout does not match MinIO’s. The caller must not copy chunks into those, since they would not fit.

cellmap_flow.finetune.session.sync.diff_and_sync_chunks(s3, s0_path, dst_s0_path, known_state, force=False) → tuple

Copy the chunks of MinIO’s s0_path that changed since known_state to disk.

Local disk is the source of truth: YAML imports are written locally first and only later mirrored to MinIO, and painted scribbles flow MinIO -> local through this function. Nothing on disk is ever deleted because MinIO lacks it: a chunk absent from a listing is almost always a transient (a truncated page, an mc mirror in flight, a server restart, a network blip), not a real erase, since painting background over a chunk in neuroglancer rewrites the chunk file rather than removing it. Trusting one bad listing once wiped 3456 chunks of a session from disk.

Returns (changed_keys, remote_state): the chunks copied, and the state to record as synced. A chunk that failed to copy is left out of changed_keys and keeps its previous state (or none) in remote_state, so the next sync tries it again.

cellmap_flow.finetune.session.sync.volume_record(volume_id, zarr_path=None, *, volumes)

The registry record of volume_id, rebuilt from zarr_path if need be.

When the registry has none yet (after a dashboard restart, say), the record is rebuilt from zarr_path’s attrs and registered. None if neither gives one.

cellmap_flow.finetune.session.sync.sync_volume(volume_id, force=False, zarr_path=None, *, state, volumes) → bool

Pull an annotation volume’s changed chunks from MinIO to local disk.

Syncs the annotation group’s metadata, then diffs MinIO’s chunk listing against what was last synced and copies the chunks that changed. The trainer reads the volume itself, through the session’s manifest. Returns True if any chunk was pulled.

The chunks go to the volume’s own zarr: zarr_path (a resumed copy), else the registered one, else – only for a volume this dashboard has no record of – <output_base>/<volume_id>.zarr. output_base is fixed by the first ensure_serving of the dashboard’s life, so a volume created under another output path would have its strokes land in the first session’s directory, where its trainer never looks.

cellmap_flow.finetune.session.sync.sync_all(force: bool = True, *, state, volumes) → int

Sync every registered annotation volume from MinIO to local disk.

The registry’s volumes, not the bucket’s. The bucket keeps every volume MinIO ever served from its data directory: one retired by Use this model (store.SessionStore.retire_volume), and every volume of a resumed session’s source whose .minio came along. Synced from the bucket’s listing, each of those would be registered as it went (volume_record), back into the session it had left.

Returns the number of volumes that had changed chunks, or -1 if MinIO is not initialized.

cellmap_flow.finetune.session.sync.retire_volume(volume_id, store, *, state) → str | None

Pull a volume’s last strokes from MinIO, then take it out of the session, kept on disk (store’s SessionStore.retire_volume); returns where it went, as that does.

Both under the sync’s lock: a periodic sync between the two, or during the move, would write chunks into the zarr as it moved and make its old path again. What MinIO holds of it stays there, unsynced from then on.

cellmap_flow.finetune.session.sync.periodic_sync_once(*, state, volumes) → None: One round of the periodic sync; failures are warned about, not hidden.

cellmap_flow.finetune.session.sync.periodic_sync(*, state, volumes) → None: The periodic sync thread’s loop: a round every SYNC_PERIOD_SECONDS, forever.

cellmap_flow.finetune.session.sync.start_periodic_sync(state, volumes) → None: Start the periodic annotation sync thread if not already running.