diff --git a/README.md b/README.md index 5be0351..d925e59 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,76 @@ and attachment links remain valid. Directory-backed Zarr stores are available through result registration or the bulk results ZIP; individual file attachments skip those directories and report the skipped count. +## Storage provenance and repair + +Shallow-result provenance uses the shared schema-2 manifest parser from +`biomero-schema>=0.3.0b1,<0.4` (also required by the matching BIOMERO core). +Plate annotations collect the source identities from the manifest's image +bindings; Image annotations select their own image node. Both v2 and v3 +interchange profiles are read without converting the store. Unsupported +manifest versions, inconsistent bindings or ambiguous references produce a +metadata warning rather than guessed provenance. + +The import records storage evidence before rendering workflow annotations. +If this step or annotation persistence fails, the script's `Message` reports +`WARNING` with the complete/reduced/incomplete counts and points to the activity +log. The imported data remains available; a completed import does not imply +that all its metadata was written. + +For an existing result whose storage evidence is missing, an administrator can +use `backfill_import_storage_provenance()` from the installed import script. +Run this in the worker's Python environment, with an authenticated administrator +`conn` in the result's group and the normal workflow event-store configuration: + +```python +import json +import runpy +from pathlib import Path +from types import SimpleNamespace +from biomero import WorkflowTracker + +scripts_root = Path("/opt/omero/server/OMERO.server/lib/scripts/biomero") +adapter = runpy.run_path(str(scripts_root / "_data/SLURM_Import_Results.py")) +backfill = adapter["backfill_import_storage_provenance"] +maintenance = runpy.run_path(str(scripts_root / "admin/SLURM_Init_environment.py")) +refresh = maintenance["refresh_workflow_metadata"] +# Select the existing result and the workflow that imported it. +result_type, result_id, workflow_id = "Plate", 123, "your-workflow-uuid" + +with WorkflowTracker() as tracker: + client = SimpleNamespace(workflowTracker=tracker) + preview = backfill(conn, client, result_type, result_id, workflow_id) + print(json.dumps(preview, indent=2)) + # After reviewing the preview, append its evidence: + # backfill(conn, client, result_type, result_id, workflow_id, dry_run=False) + + # After recording, preview the corresponding OMERO annotation changes: + # plan = refresh(conn, tracker, result_type, result_id, workflow_id, + # refresh_storage=True) + # print(json.dumps(plan, indent=2)) + # Apply with a new backup filename after reviewing that plan: + # refresh(conn, tracker, result_type, result_id, workflow_id, + # refresh_storage=True, dry_run=False, + # backup_path="/data/biomero-metadata-backups/result-before-repair.json") +``` + +The default is a read-only preview. Apply validates the matching workflow +annotation, managed storage evidence and unique import task, then appends the +observed evidence to the event store. Identical evidence is not recorded twice. +Missing files, mismatched workflows, conflicting recorded evidence and ambiguous +results must be resolved before applying. Source and result pixels are never read, rewritten or +renormalized by this operation. + +The explicit `refresh_storage=True` option requires a coordinated core build +whose `plan_metadata_refresh()` supports that option. It advances only the +import-task snapshot needed to expose newly recorded storage evidence, including +its paired job maps. Other task snapshots remain pinned. Normal metadata +refresh, including the Slurm Init UI, preserves the old snapshots and therefore +does not expose later evidence by itself. See +[metadata administration](https://nl-bioimaging.github.io/NL-BIOMERO/master/sysadmin/metadata-refresh.html) +for backup and refresh controls. Recording evidence and refreshing maps are two +separate steps. If refresh fails, retry it; repeating the backfill is also safe. + ## Development Tests live on the separate diff --git a/_data/SLURM_Import_Results.py b/_data/SLURM_Import_Results.py index d4e2933..d63e735 100644 --- a/_data/SLURM_Import_Results.py +++ b/_data/SLURM_Import_Results.py @@ -3217,9 +3217,10 @@ def add_metadata_to_imported_images( message = f"Added workflow metadata to {metadata_added}/{len(imported_images)} imported images" else: message = ( - f"Workflow metadata for {len(imported_images)} imported images: " + f"WARNING: Workflow metadata for {len(imported_images)} imported images: " f"{metadata_added} complete, {metadata_reduced} reduced, " - f"{len(imported_images) - metadata_added - metadata_reduced} incomplete") + f"{len(imported_images) - metadata_added - metadata_reduced} incomplete. " + "Imported data remains available. Details: activity log (i button).") else: message = f"Matched {len(imported_images)} imported screen images to workflow inputs" messages.append(message) @@ -3228,7 +3229,9 @@ def add_metadata_to_imported_images( return "\n".join(messages) except Exception as e: - error_msg = f"Failed to add metadata to imported images: {str(e)}" + error_msg = ( + f"WARNING: Could not complete metadata for imported images: {e}. " + "Details: activity log (i button).") logger.error(error_msg, exc_info=True) return error_msg @@ -3351,11 +3354,17 @@ def add_metadata_to_imported_plates( f"Workflow metadata for {len(imported_plates)} imported plates: " f"{metadata_added} complete, {metadata_reduced} reduced, " f"{len(imported_plates) - metadata_added - metadata_reduced} incomplete") + if metadata_added != len(imported_plates): + message = ( + f"WARNING: {message}. Imported data remains available. " + "Details: activity log (i button).") logger.info(message) return message except Exception as e: - error_msg = f"Failed to add metadata to imported plates: {str(e)}" + error_msg = ( + f"WARNING: Could not complete metadata for imported plates: {e}. " + "Details: activity log (i button).") logger.error(error_msg, exc_info=True) return error_msg @@ -3566,7 +3575,10 @@ def collect_import_storage_provenance(conn, object_type, object_id, wf_id): import hashlib target = conn.getObject(object_type, object_id) + if target is None: + raise ValueError('Imported result target not found') refs, paths = [], set() + other_workflow_reference = False for ann in target.listAnnotations(): ns = ann.getNs() if ns not in ('biomero.zarr.shallow', @@ -3576,6 +3588,8 @@ def collect_import_storage_provenance(conn, object_type, object_id, wf_id): if ns == 'biomero.zarr.shallow': if values.get('workflowId') == str(wf_id): refs.append(values) + else: + other_workflow_reference = True else: path = values.get('Imported_from') or values.get('Filepath') if path: @@ -3589,6 +3603,8 @@ def collect_import_storage_provenance(conn, object_type, object_id, wf_id): root = (mount / reference['relativePath']).resolve() root.relative_to(mount) else: + if other_workflow_reference: + raise ValueError('Shallow reference belongs to another workflow') roots = set() for path in paths: candidate = Path(path).resolve() @@ -3626,37 +3642,97 @@ def collect_import_storage_provenance(conn, object_type, object_id, wf_id): elif shallow: evidence['location'] = 'unknown' if shallow: + from biomero_schema.zarr import ShallowManifest + raw = manifest.read_bytes() - collection = json.loads(raw) - if collection.get('workflowId') != str(wf_id): + collection = ShallowManifest.from_dict(json.loads(raw)) + if str(collection.workflow_id) != str(wf_id): raise ValueError('Shallow manifest belongs to another workflow') - images = collection['images'] + images = collection.collection.images if reference and 'imageNodePath' in reference: images = [image for image in images - if image['imageNodePath'] == reference['imageNodePath']] + if image.node_path == reference['imageNodePath']] + elif object_type == 'Image': + # Older imports may only have biomero.import/Filepath. Select the + # exact image node it identifies; a Plate root cannot identify one + # of its fields. Never assign all Plate biocodes to one Image. + image_paths = {Path(path).resolve() for path in paths} + images = [image for image in images + if (root / image.node_path).resolve() in image_paths] + if object_type == 'Image' and len(images) != 1: + raise ValueError('Cannot identify a unique image in shallow manifest') if not images: raise ValueError('Result missing from shallow manifest') evidence.update( manifest='import-mount-data/' + manifest.relative_to(mount).as_posix(), manifest_sha256=hashlib.sha256(raw).hexdigest(), - source_biocodes=sorted({image['source']['pixelIdentity']['iscc'] + source_biocodes=sorted({collection.image_binding(image.node_id) + .source.pixel_identity.iscc_code for image in images})) return evidence +def storage_provenance_import_task(tracker, wf_id): + """Resolve one import task from durable workflow history.""" + workflow = tracker.repository.get(uuid.UUID(str(wf_id))) + tasks = [tracker.repository.get(tid) for tid in workflow.tasks] + imports = [task for task in tasks if task.task_name == 'SLURM_Import_Results.py'] + if len(imports) != 1: + raise ValueError('Cannot identify a unique import task for storage provenance') + return imports[0] + + def record_import_storage_provenance(conn, slurm_client, object_type, object_id, wf_id): """Persist observed result facts before the import-task metadata snapshot.""" evidence = collect_import_storage_provenance(conn, object_type, object_id, wf_id) if evidence is None: return tracker = slurm_client.workflowTracker - workflow = tracker.repository.get(uuid.UUID(str(wf_id))) - tasks = [tracker.repository.get(tid) for tid in workflow.tasks] - imports = [task for task in tasks if task.task_name == 'SLURM_Import_Results.py'] - if len(imports) != 1: - raise ValueError('Cannot identify a unique import task for storage provenance') + task = storage_provenance_import_task(tracker, wf_id) tracker.record_storage_provenance( - imports[0].id, f'{object_type}:{object_id}', evidence) + task.id, f'{object_type}:{object_id}', evidence) + + +def backfill_import_storage_provenance( + conn, slurm_client, object_type, object_id, wf_id, *, dry_run=True): + """Preview or append missing storage evidence for an existing result. + + Administrator-only, using the target's group context. Reads managed + metadata files, not pixel arrays. After apply, use the administrative + metadata refresh with refresh_storage=True to expose the new evidence. + Repeating either step is safe after interruption. + """ + if not conn.isAdmin(): + raise ValueError('Storage provenance backfill requires an administrator') + if object_type not in ('Image', 'Plate'): + raise ValueError('Only Image and Plate result targets are supported') + wf_id = str(uuid.UUID(str(wf_id))) + target = conn.getObject(object_type, int(object_id)) + if target is None: + raise ValueError('Imported result target not found') + if not any( + ann.getNs() == 'biomero/workflow' + and ['Workflow_ID', wf_id] in [list(pair) for pair in ann.getValue()] + for ann in target.listAnnotations()): + raise ValueError('Target has no workflow metadata matching this workflow') + evidence = collect_import_storage_provenance( + conn, object_type, int(object_id), wf_id) + if evidence is None: + raise ValueError('No managed storage evidence is available for this result') + tracker = slurm_client.workflowTracker + task = storage_provenance_import_task(tracker, wf_id) + target_key = f'{object_type}:{int(object_id)}' + previous = getattr(task, 'storage_provenance', {}).get(target_key) + if previous is not None and previous != evidence: + raise ValueError('Import task already contains different storage evidence') + unchanged = previous == evidence + status = 'unchanged' if unchanged else 'planned' + if not dry_run and not unchanged: + tracker.record_storage_provenance(task.id, target_key, evidence) + status = 'recorded' + return {'target_key': target_key, 'workflow_id': wf_id, + 'task_id': str(task.id), 'dry_run': dry_run, 'status': status, + 'evidence': evidence, 'metadata_refresh_required': True} def add_object_annotations(conn, slurmClient, object_type, object_id, job_id, wf_id=None): diff --git a/admin/SLURM_Init_environment.py b/admin/SLURM_Init_environment.py index 9ed20be..bda83c7 100644 --- a/admin/SLURM_Init_environment.py +++ b/admin/SLURM_Init_environment.py @@ -120,7 +120,7 @@ def _read_values(pairs): def refresh_workflow_metadata(conn, tracker, object_type, object_id, workflow_id, *, view_version='v0', dry_run=True, backup_path=None, - backup_enabled=True): + backup_enabled=True, refresh_storage=False): """Refresh existing maps in place; unlink obsolete internal-task maps. Default is a dry run. Applying normally saves a new backup file; disabling @@ -129,6 +129,8 @@ def refresh_workflow_metadata(conn, tracker, object_type, object_id, workflow_id annotation, image data or event is modified. Shared annotations are refused. Writes are not one transaction: errors propagate and the backup remains. Quiesce metadata writers for the target during an administrative refresh. + Explicit refresh_storage asks core to advance only the import-task snapshot + when new storage evidence has been backfilled. Normal refresh stays pinned. """ if object_type not in ('Image', 'Plate'): raise ValueError('Only Image and Plate result targets are supported') @@ -153,9 +155,11 @@ def refresh_workflow_metadata(conn, tracker, object_type, object_id, workflow_id rows.append(MetadataAnnotation(ns, values)) records.append({'annotation_id': ann.getId(), 'namespace': ns, 'pairs': [list(p) for p in pairs]}) + repair_options = {'refresh_storage': True} if refresh_storage else {} plan = plan_metadata_refresh(tracker, workflow_id, rows, view_version=view_version, - target_key=f'{object_type}:{object_id}') + target_key=f'{object_type}:{object_id}', + **repair_options) actions = [] for record, change in zip(records, plan): pairs = metadata_pairs(change.after.values) if change.after else None @@ -172,6 +176,8 @@ def refresh_workflow_metadata(conn, tracker, object_type, object_id, workflow_id 'before_pairs': a['pairs'], 'after_pairs': a['new_pairs'] or []} for a in actions]} + if refresh_storage: + summary['refresh_storage'] = True if dry_run: return summary if all(action['action'] == 'unchanged' for action in actions):