diff --git a/_data/_SLURM_Image_Transfer.py b/_data/_SLURM_Image_Transfer.py index d650ca0..38919ba 100644 --- a/_data/_SLURM_Image_Transfer.py +++ b/_data/_SLURM_Image_Transfer.py @@ -210,6 +210,45 @@ def _annotation_values(annotation): return values +def _canonical_registration_annotations(obj, namespace): + return [ + annotation for annotation in obj.listAnnotations() + if _annotation_namespace(annotation) == namespace + ] + + +def _canonical_content_signature(source): + """Compare canonical content while ignoring mutable trust metadata.""" + ignored = { + "sourceGeneration", + "pixelIdentityOrigin", + "canonicalPixelVerified", + } + + def normalize(value): + if isinstance(value, dict): + return { + key: normalize(item) + for key, item in value.items() + if key not in ignored + } + if isinstance(value, (list, tuple)): + return [normalize(item) for item in value] + return value + + return normalize(source.to_dict()) + + +def _update_canonical_annotation(annotation, values): + """Update one private BIOMERO MapAnnotation in place.""" + setter = getattr(annotation, "setValue", None) + saver = getattr(annotation, "save", None) + if not callable(setter) or not callable(saver): + raise ValueError("Canonical MapAnnotation cannot be updated in place") + setter([[key, value] for key, value in values.items()]) + return saver() + + def get_shallow_reference(obj): """Resolve one deterministic managed shallow reference from an OMERO Image.""" candidates = [] @@ -365,23 +404,12 @@ def get_canonical_source(obj, object_type): object_id, ) return None - current_generation = max( - candidate.source_generation for candidate in candidates - ) - current = [ - candidate for candidate in candidates - if candidate.source_generation == current_generation - ] - distinct = { - json.dumps(candidate.to_dict(), sort_keys=True) - for candidate in current - } - if len(distinct) != 1: + if len(candidates) != 1: raise ValueError( - f"Canonical source metadata is ambiguous for {object_type} " - f"{object_id} generation {current_generation}" + f"{object_type} {object_id} has multiple canonical registrations; " + "run BIOMERO Migrate Shallow Storage" ) - source = current[0] + source = candidates[0] logger.info( "Located canonical Zarr for %s %s: generation=%s, " "storage=%s:%s, node=%s, pixel ISCC=%s", @@ -396,6 +424,38 @@ def get_canonical_source(obj, object_type): return source +def bind_canonical_plate_source(source, index): + """Bind one physical Plate inventory to an OMERO Plate registration.""" + images = [] + for image in source.images: + image_source = image.source.model_copy(update={ + "storage_root": index.storage_root, + "relative_path": index.relative_path, + "source_object_id": index.source_object_id, + "source_generation": index.source_generation, + }) + labels = tuple( + label.model_copy(update={ + "source": label.source.model_copy(update={ + "storage_root": index.storage_root, + "relative_path": index.relative_path, + }) if label.source is not None else None, + }) + for label in image.labels + ) + images.append(image.model_copy(update={ + "source": image_source, + "labels": labels, + })) + return source.model_copy(update={ + "storage_root": index.storage_root, + "relative_path": index.relative_path, + "source_object_id": index.source_object_id, + "source_generation": index.source_generation, + "images": tuple(images), + }) + + def get_canonical_plate_source(plate): """Resolve compact storage-backed or legacy canonical Plate metadata.""" plate_id = int(plate.getId()) @@ -452,16 +512,18 @@ def get_canonical_plate_source(plate): managed_source = load_canonical_marker(managed_path) if ( isinstance(managed_source, CanonicalPlateSource) - and managed_source.source_object_id == plate_id - and managed_source.source_generation - == index.source_generation and managed_source.storage_root == index.storage_root and managed_source.relative_path == index.relative_path and len(managed_source.images) == index.image_count and sum(len(image.labels) for image in managed_source.images) == index.label_count ): - candidates.append(managed_source) + # Multiple OMERO Plates may intentionally reuse the same + # managed source Zarr. The marker inventories its pixels; + # this Plate's compact index supplies the OMERO binding. + candidates.append(bind_canonical_plate_source( + managed_source, index, + )) continue except Exception as exc: logger.warning( @@ -564,19 +626,12 @@ def get_canonical_plate_source(plate): if not candidates: logger.info("No BIOMERO canonical Zarr record found for Plate %s", plate_id) return None - generation = max(item.source_generation for item in candidates) - current = [ - item for item in candidates if item.source_generation == generation - ] - distinct = { - json.dumps(item.to_dict(), sort_keys=True) for item in current - } - if len(distinct) != 1: + if len(candidates) != 1: raise ValueError( - f"Canonical Plate metadata is ambiguous for Plate {plate_id} " - f"generation {generation}" + f"Plate {plate_id} has multiple canonical registrations; run " + "BIOMERO Migrate Shallow Storage" ) - source = current[0] + source = candidates[0] logger.info( "Located canonical Zarr for Plate %s: generation=%s, storage=%s:%s, " "images=%s (cached ISCC-BIO identities; not recalculated)", @@ -1100,19 +1155,34 @@ def attach_canonical_source( object_type, source, annotation_writer=None, + annotation_updater=None, ): - """Attach one canonical record without creating same-generation ambiguity.""" + """Create or update the object's single canonical registration.""" existing = get_canonical_source(obj, object_type) if existing == source: return None - if ( - existing is not None - and existing.source_generation >= source.source_generation - ): + registrations = _canonical_registration_annotations( + obj, CANONICAL_SOURCE_NAMESPACE, + ) + if existing is not None: + if len(registrations) != 1: + raise ValueError( + f"{object_type} {obj.getId()} has multiple canonical " + "registrations; run BIOMERO Migrate Shallow Storage" + ) + if _canonical_content_signature(existing) != ( + _canonical_content_signature(source) + ): + raise ValueError( + f"Cannot replace canonical pixels for {object_type} " + f"{obj.getId()}" + ) + updater = annotation_updater or _update_canonical_annotation + return updater(registrations[0], source.to_annotation_values()) + if registrations: raise ValueError( - f"Cannot replace canonical {object_type} {obj.getId()} generation " - f"{existing.source_generation} with generation " - f"{source.source_generation}" + f"{object_type} {obj.getId()} has an invalid canonical " + "registration; run BIOMERO Migrate Shallow Storage" ) if annotation_writer is None: from ezomero import post_map_annotation @@ -1133,25 +1203,39 @@ def attach_canonical_plate_source( plate, source, annotation_writer=None, + annotation_updater=None, ): - """Attach one compact Plate index; detailed identities live in storage.""" + """Create or update one compact Plate canonical registration.""" existing = get_canonical_plate_source(plate) if existing == source: return None - if ( - existing is not None - and existing.source_generation >= source.source_generation - ): + registrations = _canonical_registration_annotations( + plate, CANONICAL_PLATE_SOURCE_NAMESPACE, + ) + index = CanonicalPlateIndex.from_source(source) + if existing is not None: + if len(registrations) != 1: + raise ValueError( + f"Plate {plate.getId()} has multiple canonical registrations; " + "run BIOMERO Migrate Shallow Storage" + ) + if _canonical_content_signature(existing) != ( + _canonical_content_signature(source) + ): + raise ValueError( + f"Cannot replace canonical pixels for Plate {plate.getId()}" + ) + updater = annotation_updater or _update_canonical_annotation + return updater(registrations[0], index.to_annotation_values()) + if registrations: raise ValueError( - f"Cannot replace canonical Plate {plate.getId()} generation " - f"{existing.source_generation} with generation " - f"{source.source_generation}" + f"Plate {plate.getId()} has an invalid canonical registration; " + "run BIOMERO Migrate Shallow Storage" ) if annotation_writer is None: from ezomero import post_map_annotation annotation_writer = post_map_annotation - index = CanonicalPlateIndex.from_source(source) return annotation_writer( conn=conn, object_type="Plate", @@ -1421,7 +1505,7 @@ def upgrade_reused_canonical(conn, obj, source, source_path): A cached export is not authoritative merely because it is on managed disk. Require the existing import provenance to identify the same backing store. - Publish a new metadata generation without copying or rehashing pixel data. + Update trust metadata in place without copying or rehashing pixel data. """ is_plate = isinstance(source, CanonicalPlateSource) verified = (all(i.source.canonical_pixel_verified for i in source.images) @@ -1434,14 +1518,6 @@ def upgrade_reused_canonical(conn, obj, source, source_path): upgraded = (verified_plate_source(source) if is_plate else source.model_copy(update={"canonical_pixel_verified": True, "pixel_identity_origin": "canonical-bootstrap"})) - generation = source.source_generation + 1 - upgraded = upgraded.model_copy(update={"source_generation": generation}) - if is_plate: - upgraded = upgraded.model_copy(update={"images": tuple( - image.model_copy(update={"source": image.source.model_copy( - update={"source_generation": generation})}) - for image in upgraded.images - )}) write_indexed_canonical_marker(source_path, upgraded) if is_plate: attach_canonical_plate_source(conn, obj, upgraded) diff --git a/admin/BIOMERO_Migrate_Shallow_Storage.py b/admin/BIOMERO_Migrate_Shallow_Storage.py index 94fef58..9dbfd4e 100644 --- a/admin/BIOMERO_Migrate_Shallow_Storage.py +++ b/admin/BIOMERO_Migrate_Shallow_Storage.py @@ -5,6 +5,8 @@ import json import logging from pathlib import Path +import re +import shutil from uuid import uuid4 import omero @@ -14,12 +16,28 @@ from omero.sys import ParametersI from biomero_schema.zarr import ( + CANONICAL_PLATE_IMAGE_NAMESPACE, + CANONICAL_PLATE_LABEL_NAMESPACE, + CANONICAL_PLATE_SOURCE_NAMESPACE, + CANONICAL_SOURCE_NAMESPACE, SHALLOW_COLLECTION_MANIFEST, SHALLOW_COLLECTION_NAMESPACE, + CanonicalPlateIndex, + CanonicalPlateSource, + CanonicalZarrSource, ShallowManifest, + ShallowPlateReference, + ShallowZarrReference, +) +from biomero_importer.utils.canonical_store import ( + CANONICAL_MARKER_NAME, + external_canonical_marker_path, + load_canonical_marker, + write_indexed_canonical_marker, ) from biomero_shallower.migration import ( migrate_shallow_store_v1, + rebind_shallow_store_sources, restore_shallow_store_v1, upgrade_annotation_reference_v1, upgrade_manifest_v1, @@ -35,6 +53,7 @@ DRY_RUN = "Dry Run" BACKUP_DIRECTORY = "Backup Directory (optional)" REQUIRED_MIGRATION_CAPABILITIES = ( + "canonical-single-store", "schema-1-to-2", "schema-1-path-only-labels", ) @@ -77,6 +96,635 @@ def _resolve_store_path(values, storage_roots): return store +def _stable_canonical_relative_path(relative_path, object_type, object_id): + """Remove the prerelease generation suffix from generated stores.""" + relative = Path(relative_path) + legacy = re.fullmatch( + rf"{re.escape(object_type)}-{int(object_id)}\.g\d+\.ome\.zarr", + relative.name, + ) + if legacy: + return relative.with_name( + f"{object_type}-{int(object_id)}.ome.zarr" + ) + return relative + + +def _canonical_content_signature(source): + """Identify canonical content independently of path/trust revisions.""" + ignored = { + "relativePath", + "sourceGeneration", + "pixelIdentityOrigin", + "canonicalPixelVerified", + } + + def normalize(value): + if isinstance(value, dict): + return { + key: normalize(item) + for key, item in value.items() + if key not in ignored + } + if isinstance(value, (list, tuple)): + return [normalize(item) for item in value] + return value + + return json.dumps(normalize(source.to_dict()), sort_keys=True) + + +def _normalize_canonical_source(source, relative_path): + """Point one canonical contract at its stable physical store.""" + relative_path = Path(relative_path).as_posix() + if not isinstance(source, CanonicalPlateSource): + return source.model_copy(update={ + "relative_path": relative_path, + "source_generation": 1, + }) + + images = [] + for image in source.images: + image_source = image.source.model_copy(update={ + "relative_path": relative_path, + "source_generation": 1, + }) + labels = tuple( + label.model_copy(update={ + "source": label.source.model_copy(update={ + "relative_path": relative_path, + }) if label.source is not None else None, + }) + for label in image.labels + ) + images.append(image.model_copy(update={ + "source": image_source, + "labels": labels, + })) + return source.model_copy(update={ + "relative_path": relative_path, + "source_generation": 1, + "images": tuple(images), + }) + + +def _plan_canonical_consolidation(records, storage_roots): + """Select one equivalent canonical store and its stable destination.""" + if not records: + raise ValueError("Canonical consolidation requires registrations") + signatures = { + _canonical_content_signature(record["source"]) + for record in records + } + if len(signatures) != 1: + raise ValueError("Canonical registrations contain different canonical content") + + first = records[0]["source"] + object_type = ( + "Plate" if isinstance(first, CanonicalPlateSource) + else first.source_object_type + ) + object_id = first.source_object_id + + native_locations = { + (record["source"].storage_root, record["source"].relative_path) + for record in records + if _stable_canonical_relative_path( + record["source"].relative_path, object_type, object_id, + ) == Path(record["source"].relative_path) + and Path(record["source"].relative_path).name + != f"{object_type}-{object_id}.ome.zarr" + } + if len(native_locations) > 1: + raise ValueError( + "Canonical registrations point at multiple managed input Zarrs" + ) + + def preference(record): + source = record["source"] + relative = Path(source.relative_path) + stable = _stable_canonical_relative_path( + relative, object_type, object_id, + ) + stable_name = f"{object_type}-{object_id}.ome.zarr" + if stable == relative and relative.name != stable_name: + kind = 0 # Existing managed input Zarr: never copy it. + elif stable == relative: + kind = 1 # Already uses the stable generated name. + else: + kind = 2 # Legacy gN store; prefer the latest only as input. + return kind, -source.source_generation + + selected = min(records, key=preference) + source = selected["source"] + relative = _stable_canonical_relative_path( + source.relative_path, object_type, object_id, + ) + try: + root = Path(storage_roots[source.storage_root]).resolve() + except KeyError as error: + raise ValueError( + f"Unknown managed storage root: {source.storage_root}" + ) from error + destination = (root / relative).resolve() + if not destination.is_relative_to(root): + raise ValueError("Canonical destination escapes its managed storage root") + return { + "selected": selected, + "source": _normalize_canonical_source(source, relative), + "relative_path": relative, + "destination": destination, + } + + +def _annotation_links(conn, object_type, namespace): + offset = 0 + while True: + page = list(conn.getAnnotationLinks( + object_type, + ns=namespace, + params=ParametersI().page(offset, 500), + )) + yield from page + if len(page) < 500: + return + offset += len(page) + + +def _canonical_marker_status(marker, source, annotation_id): + """Classify repairable marker drift without accepting new pixels.""" + if marker is None: + return "missing" + if _canonical_content_signature(marker) != _canonical_content_signature( + source + ): + raise ValueError( + "Canonical marker pixel identity disagrees with annotation " + f"{annotation_id}" + ) + return "matching" if marker == source else "locator-drift" + + +def _bind_canonical_plate_source(source, index): + """Bind a shared physical Plate inventory to one OMERO Plate.""" + images = [] + for image in source.images: + image_source = image.source.model_copy(update={ + "storage_root": index.storage_root, + "relative_path": index.relative_path, + "source_object_id": index.source_object_id, + "source_generation": index.source_generation, + }) + labels = tuple( + label.model_copy(update={ + "source": label.source.model_copy(update={ + "storage_root": index.storage_root, + "relative_path": index.relative_path, + }) if label.source is not None else None, + }) + for label in image.labels + ) + images.append(image.model_copy(update={ + "source": image_source, + "labels": labels, + })) + return source.model_copy(update={ + "storage_root": index.storage_root, + "relative_path": index.relative_path, + "source_object_id": index.source_object_id, + "source_generation": index.source_generation, + "images": tuple(images), + }) + + +def discover_canonical_registrations( + conn, storage_roots, *, object_type="All", object_ids=None, +): + """Load every canonical registration and its authoritative marker.""" + if not conn.isAdmin(): + raise ValueError("Canonical consolidation requires an administrator") + kinds = ("Image", "Plate") if object_type == "All" else (object_type,) + selected = set(object_ids or ()) + records = [] + original_group = conn.SERVICE_OPTS.getOmeroGroup() + conn.SERVICE_OPTS.setOmeroGroup("-1") + try: + for kind in kinds: + namespace = ( + CANONICAL_SOURCE_NAMESPACE if kind == "Image" + else CANONICAL_PLATE_SOURCE_NAMESPACE + ) + for link in _annotation_links(conn, kind, namespace): + parent = link.getParent() + ident = parent.getId() + if selected and ident not in selected: + continue + annotation = link.getAnnotation() + pairs = [list(pair) for pair in annotation.getValue()] + values = _pairs_to_values(pairs) + shared_plate_marker = False + if kind == "Image": + source = CanonicalZarrSource.from_annotation_values(values) + elif "images" in values: + source = CanonicalPlateSource.from_annotation_values(values) + else: + index = CanonicalPlateIndex.from_annotation_values(values) + path = _resolve_store_path(index.to_annotation_values(), + storage_roots) + marker_source = load_canonical_marker(path) + if not isinstance(marker_source, CanonicalPlateSource): + raise ValueError( + f"Plate {ident} canonical marker is missing or invalid" + ) + source = _bind_canonical_plate_source( + marker_source, index, + ) + shared_plate_marker = True + if source.source_object_id != ident: + raise ValueError( + f"Canonical annotation {annotation.getId()} is linked " + f"to the wrong {kind}" + ) + path = _resolve_store_path(source.to_annotation_values(), + storage_roots) + if not path.is_dir(): + raise ValueError( + f"Canonical store does not exist for annotation " + f"{annotation.getId()}: {path}" + ) + marker = load_canonical_marker(path) + # Early prerelease registrations did not consistently write + # a sidecar for reused managed Zarrs. Apply mode repairs that + # omission, but a different pixel identity remains fatal. + if shared_plate_marker and marker is not None: + marker_status = ( + "matching" + if marker.storage_root == source.storage_root + and marker.relative_path == source.relative_path + and len(marker.images) == len(source.images) + and sum(len(image.labels) for image in marker.images) + == sum(len(image.labels) for image in source.images) + else "locator-drift" + ) + else: + marker_status = _canonical_marker_status( + marker, source, annotation.getId(), + ) + records.append({ + "object_type": kind, + "object_id": ident, + "group_id": parent.getDetails().group.id.val, + "annotation_id": annotation.getId(), + "pairs": pairs, + "values": values, + "source": source, + "path": path, + "marker_status": marker_status, + }) + finally: + conn.SERVICE_OPTS.setOmeroGroup(original_group) + return records + + +def discover_shallow_references( + conn, *, object_type="All", object_ids=None, +): + """Discover schema-2 result references that may use canonical stores.""" + kinds = ("Image", "Plate") if object_type == "All" else (object_type,) + selected = set(object_ids or ()) + records = [] + original_group = conn.SERVICE_OPTS.getOmeroGroup() + conn.SERVICE_OPTS.setOmeroGroup("-1") + try: + for kind in kinds: + for link in _annotation_links( + conn, kind, SHALLOW_COLLECTION_NAMESPACE, + ): + parent = link.getParent() + ident = parent.getId() + if selected and ident not in selected: + continue + annotation = link.getAnnotation() + pairs = [list(pair) for pair in annotation.getValue()] + values = _pairs_to_values(pairs) + if values.get("schema") != "2": + continue + reference = ( + ShallowZarrReference.from_annotation_values(values) + if kind == "Image" + else ShallowPlateReference.from_annotation_values(values) + ) + records.append({ + "object_type": kind, + "object_id": ident, + "group_id": parent.getDetails().group.id.val, + "annotation_id": annotation.getId(), + "pairs": pairs, + "values": values, + "reference": reference, + }) + finally: + conn.SERVICE_OPTS.setOmeroGroup(original_group) + return records + + +def _managed_locators(value): + """Collect managed storage locators from a contract tree.""" + locators = set() + if isinstance(value, dict): + if "storageRoot" in value and "relativePath" in value: + locators.add((str(value["storageRoot"]), str(value["relativePath"]))) + for item in value.values(): + locators.update(_managed_locators(item)) + elif isinstance(value, (list, tuple)): + for item in value: + locators.update(_managed_locators(item)) + return locators + + +def _requires_locator_update(value, replacements): + if isinstance(value, dict): + locator = (value.get("storageRoot"), value.get("relativePath")) + if locator in replacements and ( + value.get("relativePath") != replacements[locator] + or value.get("sourceGeneration", 1) != 1 + ): + return True + return any( + _requires_locator_update(item, replacements) + for item in value.values() + ) + if isinstance(value, (list, tuple)): + return any(_requires_locator_update(item, replacements) + for item in value) + return False + + +def _shallow_reference_values(record, manifest): + values = record["values"] + common = { + "storage_root": values["storageRoot"], + "relative_path": values["relativePath"], + } + if record["object_type"] == "Image": + reference = ShallowZarrReference.from_manifest( + manifest, + image_node_path=values["imageNodePath"], + label_node_paths=tuple(json.loads(values["labelNodePaths"])), + **common, + ) + else: + reference = ShallowPlateReference.from_manifest(manifest, **common) + return reference.to_annotation_values() + + +def _tree_metrics(path): + files = 0 + size = 0 + for item in path.rglob("*"): + if item.is_file() and not item.is_symlink(): + files += 1 + size += item.stat().st_size + return {"files": files, "bytes": size} + + +def _canonical_marker_path(path): + internal = path / CANONICAL_MARKER_NAME + external = external_canonical_marker_path(path) + markers = [candidate for candidate in (internal, external) + if candidate.is_file()] + if len(markers) != 1: + raise ValueError( + f"Canonical store must have exactly one marker: {path}" + ) + return markers[0] + + +def _write_canonical_marker(path, source): + try: + marker_path = _canonical_marker_path(path) + except ValueError: + internal = path / CANONICAL_MARKER_NAME + external = external_canonical_marker_path(path) + if internal.exists() or external.exists(): + raise + return write_indexed_canonical_marker(path, source) + marker = json.loads(marker_path.read_text(encoding="utf-8")) + if marker.get("state") != "committed" or "markerSchema" not in marker: + raise ValueError(f"Canonical marker is not committed: {marker_path}") + marker["source"] = source.to_dict() + _write_json(marker_path, marker) + if load_canonical_marker(path) != source: + raise ValueError(f"Canonical marker update did not validate: {path}") + return marker_path + + +def _move_canonical_store(source, destination): + """Rename one generated store and any external marker without copying.""" + if source == destination: + return + if destination.exists(): + raise FileExistsError(f"Stable canonical destination exists: {destination}") + old_external = external_canonical_marker_path(source) + new_external = external_canonical_marker_path(destination) + source.rename(destination) + if old_external.is_file(): + new_external.parent.mkdir(parents=True, exist_ok=True) + old_external.rename(new_external) + + +def _canonical_registration_values(source): + if isinstance(source, CanonicalPlateSource): + return CanonicalPlateIndex.from_source(source).to_annotation_values() + return source.to_annotation_values() + + +def _delete_annotation(conn, record): + conn.SERVICE_OPTS.setOmeroGroup(str(record["group_id"])) + conn.deleteObjects("Annotation", [record["annotation_id"]], wait=True) + + +def _legacy_plate_records(conn, plate_id): + records = [] + for namespace in ( + CANONICAL_PLATE_IMAGE_NAMESPACE, + CANONICAL_PLATE_LABEL_NAMESPACE, + ): + for link in _annotation_links(conn, "Plate", namespace): + parent = link.getParent() + if parent.getId() != plate_id: + continue + annotation = link.getAnnotation() + records.append({ + "group_id": parent.getDetails().group.id.val, + "annotation_id": annotation.getId(), + }) + return records + + +def plan_canonical_consolidation( + conn, storage_roots, *, object_type="All", object_ids=None, +): + records = discover_canonical_registrations( + conn, storage_roots, object_type=object_type, object_ids=object_ids, + ) + grouped = {} + for record in records: + grouped.setdefault( + (record["object_type"], record["object_id"]), [] + ).append(record) + + plans = [] + replacements = {} + for (kind, ident), group_records in grouped.items(): + plan = _plan_canonical_consolidation(group_records, storage_roots) + legacy_metadata = ( + _legacy_plate_records(conn, ident) if kind == "Plate" else [] + ) + selected_path = plan["selected"]["path"] + legacy_paths = [] + for record in group_records: + source = record["source"] + old_path = record["path"] + if old_path != plan["destination"]: + expected = _stable_canonical_relative_path( + source.relative_path, kind, ident, + ) + if expected == Path(source.relative_path): + if old_path != selected_path: + raise ValueError( + f"Refusing to remove non-generated canonical " + f"store: {old_path}" + ) + else: + legacy_paths.append(old_path) + changed = ( + len(group_records) != 1 + or selected_path != plan["destination"] + or plan["selected"]["source"] != plan["source"] + or any(record["marker_status"] != "matching" + for record in group_records) + or bool(legacy_metadata) + ) + if not changed: + continue + for record in group_records: + source = record["source"] + replacements[(source.storage_root, source.relative_path)] = ( + plan["source"].relative_path + ) + reclaim = {"files": 0, "bytes": 0} + for path in set(legacy_paths): + if path == selected_path and selected_path != plan["destination"]: + continue + metrics = _tree_metrics(path) + reclaim["files"] += metrics["files"] + reclaim["bytes"] += metrics["bytes"] + plan.update({ + "object_type": kind, + "object_id": ident, + "records": group_records, + "legacy_paths": tuple(sorted(set(legacy_paths))), + "legacy_metadata": tuple(legacy_metadata), + "reclaim": reclaim, + }) + plans.append(plan) + return plans, replacements + + +def _apply_canonical_consolidation( + conn, plans, replacements, storage_roots, directory, +): + """Apply prevalidated consolidation, deleting duplicates only at the end.""" + if not plans: + return [] + relevant = set(replacements) + shallow_records = discover_shallow_references(conn) + shallow_groups = {} + for record in shallow_records: + store = _resolve_store_path(record["values"], storage_roots) + manifest = ShallowManifest.from_dict(json.loads( + (store / SHALLOW_COLLECTION_MANIFEST).read_text(encoding="utf-8") + )) + manifest_value = manifest.to_dict() + used = _managed_locators(manifest_value).intersection(relevant) + if used and _requires_locator_update(manifest_value, replacements): + group = shallow_groups.setdefault(store, { + "manifest": manifest, + "records": [], + "replacements": {}, + }) + group["records"].append(record) + group["replacements"].update({ + locator: replacements[locator] for locator in used + }) + + results = [] + for index, plan in enumerate(plans, start=1): + selected_path = plan["selected"]["path"] + destination = plan["destination"] + snapshot = { + "objectType": plan["object_type"], + "objectId": plan["object_id"], + "selected": str(selected_path), + "destination": str(destination), + "annotations": [record["annotation_id"] + for record in plan["records"]], + "markerStatus": { + str(record["annotation_id"]): record["marker_status"] + for record in plan["records"] + }, + "reclaim": plan["reclaim"], + } + _write_json(directory / f"canonical-{index:06d}.json", snapshot) + _move_canonical_store(selected_path, destination) + marker_backup = directory / f"canonical-{index:06d}-marker.json" + try: + marker_path = _canonical_marker_path(destination) + except ValueError: + marker_path = None + if marker_path is not None: + shutil.copy2(marker_path, marker_backup) + _write_canonical_marker(destination, plan["source"]) + results.append(snapshot) + + for index, (store, group) in enumerate(shallow_groups.items(), start=1): + migration = rebind_shallow_store_sources( + store, + group["replacements"], + backup_path=directory / f"shallow-{index:06d}-files", + ) + for record in group["records"]: + _write_annotation( + conn, record, + _shallow_reference_values(record, migration.manifest), + ) + + for plan in plans: + selected = plan["selected"] + _write_annotation( + conn, selected, _canonical_registration_values(plan["source"]), + ) + for record in plan["records"]: + if record["annotation_id"] != selected["annotation_id"]: + _delete_annotation(conn, record) + if plan["object_type"] == "Plate": + for record in plan["legacy_metadata"]: + _delete_annotation(conn, record) + + # This is the only destructive phase. Every live reference and marker now + # resolves to the stable store, and only generated gN directories qualify. + for plan in plans: + destination = plan["destination"] + selected_old = plan["selected"]["path"] + for path in plan["legacy_paths"]: + if path == selected_old and selected_old != destination: + continue + if path.exists(): + shutil.rmtree(path) + external_canonical_marker_path(path).unlink(missing_ok=True) + return results + + def _build_group_plan(records, storage_roots, manifest): """Build the coupled filesystem/OMERO plan for one shallow store.""" stores = { @@ -263,7 +911,7 @@ def _migration_directory_path(value): raise ValueError("Backup Directory must be an absolute worker path") else: root = Path("/data/biomero-shallow-migrations") - directory = root / f"schema-1-to-2-{uuid4()}" + directory = root / f"shallow-storage-{uuid4()}" return directory @@ -337,6 +985,9 @@ def migrate_schema_1_references( planned_directory = _migration_directory_path(backup_directory) _preflight_migration_directory(planned_directory) roots = load_managed_storage_roots() + canonical_plans, canonical_replacements = plan_canonical_consolidation( + conn, roots, object_type=object_type, object_ids=object_ids, + ) records = discover_schema_1_references( conn, object_type=object_type, object_ids=object_ids, ) @@ -352,7 +1003,8 @@ def migrate_schema_1_references( None if dry_run else _create_migration_directory(planned_directory) ) report = { - "migration": "biomero-shallow-schema-1-to-2", + "migration": "biomero-shallow-storage", + "operations": ["schema-1-to-2", "canonical-single-store"], "dryRun": dry_run, "backupDirectory": str(planned_directory), "stores": [], @@ -362,7 +1014,27 @@ def migrate_schema_1_references( "migrated": 0, "recovered": 0, "failed": 0, + "canonicalObjects": len(canonical_plans), + "canonicalStoresRemoved": sum( + len(plan["legacy_paths"]) - ( + 1 if plan["selected"]["path"] != plan["destination"] + and plan["selected"]["path"] in plan["legacy_paths"] + else 0 + ) + for plan in canonical_plans + ), + "canonicalBytesReclaimed": sum( + plan["reclaim"]["bytes"] for plan in canonical_plans + ), }, + "canonical": [{ + "objectType": plan["object_type"], + "objectId": plan["object_id"], + "selected": str(plan["selected"]["path"]), + "destination": str(plan["destination"]), + "registrations": len(plan["records"]), + "reclaim": plan["reclaim"], + } for plan in canonical_plans], } original_group = conn.SERVICE_OPTS.getOmeroGroup() try: @@ -426,6 +1098,15 @@ def migrate_schema_1_references( str(migration.backup_path) if migration else None ) _write_json(directory / "report.json", report) + if not dry_run and canonical_plans: + report["canonical"] = _apply_canonical_consolidation( + conn, + canonical_plans, + canonical_replacements, + roots, + directory, + ) + _write_json(directory / "report.json", report) finally: conn.SERVICE_OPTS.setOmeroGroup(original_group) return report @@ -437,7 +1118,10 @@ def _summary(report): return ( "Shallow-storage migration dry run: " f"{counts['stores']} stores and {counts['annotations']} OMERO " - "references would be upgraded from schema 1 to schema 2. " + "references would be upgraded from schema 1 to schema 2; " + f"{counts['canonicalObjects']} canonical objects would be " + "consolidated, reclaiming approximately " + f"{counts['canonicalBytesReclaimed']} bytes. " "No data was changed. Backup location validated inside " f"biomeroworker: {report['backupDirectory']}" ) @@ -445,6 +1129,7 @@ def _summary(report): "Shallow-storage migration complete: " f"{counts['migrated']} stores migrated, " f"{counts['recovered']} reference-only recoveries, " + f"{counts['canonicalObjects']} canonical objects consolidated, " f"{counts['failed']} failures. Backups and report: " f"{report['backupDirectory']}" ) @@ -453,9 +1138,9 @@ def _summary(report): def runScript(): client = scripts.client( "BIOMERO Migrate Shallow Storage (Admin Only)", - """Upgrade prerelease BIOMERO shallow-Zarr schema-1 storage metadata - and its linked OMERO references to schema 2. Run a dry run first. - Applying always creates recovery files; image pixels are not copied.""", + """Migrate older BIOMERO shallow-storage metadata and consolidate + canonical source registrations. Run a dry run first. Applying always + creates recovery files; image pixels are not copied.""", scripts.String( OBJECT_TYPE, default="All",