From 5b91208779d04784e625297e9f64d7d6136b3981 Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 16 Apr 2026 16:18:49 +0200 Subject: [PATCH 001/120] Add backend models and APIs for bulk storage operations --- lib/galaxy/celery/tasks.py | 131 ++++++++ lib/galaxy/model/__init__.py | 59 ++++ lib/galaxy/schema/schema.py | 103 ++++++ .../webapps/galaxy/api/history_contents.py | 42 +++ .../galaxy/services/history_contents.py | 295 +++++++++++++++++- 5 files changed, 627 insertions(+), 3 deletions(-) diff --git a/lib/galaxy/celery/tasks.py b/lib/galaxy/celery/tasks.py index b7145b99676..d8f68aa770a 100644 --- a/lib/galaxy/celery/tasks.py +++ b/lib/galaxy/celery/tasks.py @@ -5,6 +5,7 @@ from collections.abc import Callable from concurrent.futures import TimeoutError from functools import lru_cache from pathlib import Path +from types import SimpleNamespace from typing import ( Any, Optional, @@ -50,6 +51,10 @@ from galaxy.managers.tool_data import ToolDataImportManager from galaxy.managers.workflow_completion import WorkflowCompletionManager from galaxy.metadata.set_metadata import set_metadata_portable from galaxy.model import ( + Dataset, + DatasetStorageOperationRun, + DatasetStorageOperationRunItem, + DatasetStorageOperationSnapshot, Job, User, ) @@ -308,6 +313,132 @@ def set_metadata( sa_session.commit() +@galaxy_task(action="execute bulk storage relocate run") +def bulk_relocate_storage( + sa_session: galaxy_scoped_session, + dataset_manager: DatasetManager, + app: MinimalManagerApp, + run_db_id: int, + task_user_id: Optional[int] = None, +): + run = sa_session.get(DatasetStorageOperationRun, run_db_id) + if run is None: + log.error(f"Bulk storage relocate run not found: {run_db_id}") + return + + snapshot = sa_session.get(DatasetStorageOperationSnapshot, run.snapshot_id) + if snapshot is None: + run.state = "failed" + sa_session.add(run) + sa_session.commit() + log.error(f"Snapshot not found for run {run_db_id}") + return + + if snapshot.expires_at <= now(): + run.state = "failed" + sa_session.add(run) + sa_session.commit() + log.error(f"Snapshot expired for run {run_db_id}") + return + + run.state = "running" + sa_session.add(run) + sa_session.commit() + + user = sa_session.get(User, run.user_id) if run.user_id else None + trans = SimpleNamespace(user=user) + + succeeded_count = 0 + failed_count = 0 + skipped_count = 0 + resolved_dataset_ids = snapshot.resolved_dataset_ids + + for dataset_id in resolved_dataset_ids: + dataset = sa_session.get(Dataset, dataset_id) + if dataset is None: + failed_count += 1 + sa_session.add( + DatasetStorageOperationRunItem( + run_id=run.id, + dataset_id=dataset_id, + state="failed", + reason_code="dataset_not_found", + message="Dataset was not found at execution time.", + ) + ) + continue + + reason_code: Optional[str] = None + message: Optional[str] = None + device_source_map = app.object_store.get_device_source_map() + target_device_id = device_source_map.get_device_id(run.target_object_store_id) + source_object_store_id = dataset.object_store_id + source_device_id = device_source_map.get_device_id(source_object_store_id) if source_object_store_id else None + + if target_device_id is None: + reason_code = "invalid_target_object_store" + message = f"Target object store '{run.target_object_store_id}' is invalid." + elif source_object_store_id is None: + reason_code = "missing_source_object_store" + message = "Dataset does not define a source object store." + elif source_object_store_id == run.target_object_store_id: + reason_code = "already_in_target" + message = "Dataset is already in the requested object store." + elif source_device_id != target_device_id: + reason_code = "cross_device_relocate_not_allowed" + message = "Cannot relocate to an object store that does not share the same device." + elif not app.security_agent.can_change_object_store_id(user, dataset): + reason_code = "insufficient_permissions" + message = "Dataset is not eligible for object store changes." + + if reason_code is not None: + state = "skipped" if run.skip_ineligible else "failed" + if state == "skipped": + skipped_count += 1 + else: + failed_count += 1 + sa_session.add( + DatasetStorageOperationRunItem( + run_id=run.id, + dataset_id=dataset_id, + state=state, + reason_code=reason_code, + message=message, + ) + ) + continue + + try: + dataset_manager.update_object_store_id(trans, dataset, run.target_object_store_id) + succeeded_count += 1 + sa_session.add( + DatasetStorageOperationRunItem( + run_id=run.id, + dataset_id=dataset_id, + state="succeeded", + ) + ) + except Exception as exc: + failed_count += 1 + sa_session.add( + DatasetStorageOperationRunItem( + run_id=run.id, + dataset_id=dataset_id, + state="failed", + reason_code="execution_error", + message=str(exc), + ) + ) + + run.state = "completed" + run.total_count = len(resolved_dataset_ids) + run.succeeded_count = succeeded_count + run.failed_count = failed_count + run.skipped_count = skipped_count + sa_session.add(run) + sa_session.commit() + + def _get_dataset_manager( hda_manager: HDAManager, ldda_manager: LDDAManager, model_class: str = "HistoryDatasetAssociation" ) -> DatasetAssociationManager: diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 6dfff4660ee..83bae9db04c 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -12726,6 +12726,65 @@ class CeleryUserActiveTask(Base): started_at: Mapped[datetime] +class DatasetStorageOperationSnapshot(Base): + """Immutable snapshot of a resolved storage operation selection.""" + + __tablename__ = "dataset_storage_operation_snapshot" + + id: Mapped[int] = mapped_column(primary_key=True) + history_id: Mapped[int] = mapped_column(ForeignKey("history.id", ondelete="CASCADE"), index=True) + user_id: Mapped[int] = mapped_column(ForeignKey("galaxy_user.id", ondelete="CASCADE"), index=True) + mode: Mapped[str] = mapped_column(String(32), index=True) + target_object_store_id: Mapped[str] = mapped_column(String(255)) + resolved_dataset_ids: Mapped[list[int]] = mapped_column(JSONType) + eligible_dataset_ids: Mapped[list[int]] = mapped_column(JSONType) + create_time: Mapped[datetime] = mapped_column(default=now, nullable=True) + update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, nullable=True) + expires_at: Mapped[datetime] = mapped_column(index=True) + + +class DatasetStorageOperationRun(Base): + """Tracks one execution attempt for a storage operation snapshot.""" + + __tablename__ = "dataset_storage_operation_run" + + id: Mapped[int] = mapped_column(primary_key=True) + snapshot_id: Mapped[int] = mapped_column( + ForeignKey("dataset_storage_operation_snapshot.id", ondelete="CASCADE"), index=True + ) + history_id: Mapped[int] = mapped_column(ForeignKey("history.id", ondelete="CASCADE"), index=True) + user_id: Mapped[int] = mapped_column(ForeignKey("galaxy_user.id", ondelete="CASCADE"), index=True) + mode: Mapped[str] = mapped_column(String(32), index=True) + target_object_store_id: Mapped[str] = mapped_column(String(255)) + state: Mapped[str] = mapped_column(String(32), index=True) + skip_ineligible: Mapped[bool] = mapped_column(Boolean, default=True) + task_id: Mapped[Optional[Union[UUID, str]]] = mapped_column(UUIDType(), index=True) + total_count: Mapped[int] = mapped_column(default=0) + succeeded_count: Mapped[int] = mapped_column(default=0) + failed_count: Mapped[int] = mapped_column(default=0) + skipped_count: Mapped[int] = mapped_column(default=0) + create_time: Mapped[datetime] = mapped_column(default=now, nullable=True) + update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, nullable=True) + + +class DatasetStorageOperationRunItem(Base): + """Per-dataset status for a storage operation run.""" + + __tablename__ = "dataset_storage_operation_run_item" + __table_args__ = (UniqueConstraint("run_id", "dataset_id"),) + + id: Mapped[int] = mapped_column(primary_key=True) + run_id: Mapped[int] = mapped_column(ForeignKey("dataset_storage_operation_run.id", ondelete="CASCADE"), index=True) + dataset_id: Mapped[int] = mapped_column(ForeignKey("dataset.id", ondelete="CASCADE"), index=True) + state: Mapped[str] = mapped_column(String(32), index=True) + reason_code: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) + message: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + attempt_count: Mapped[int] = mapped_column(default=1) + bytes_processed: Mapped[int] = mapped_column(default=0) + create_time: Mapped[datetime] = mapped_column(default=now, nullable=True) + update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, nullable=True) + + class UserCredentials(Base): """ Represents a credential associated with a user for a specific service. diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py index faadc6cdbc0..eedf4331a5b 100644 --- a/lib/galaxy/schema/schema.py +++ b/lib/galaxy/schema/schema.py @@ -1406,6 +1406,109 @@ class HistoryContentBulkOperationResult(Model): errors: list[BulkOperationItemError] +class StorageOperationMode(str, Enum): + relocate = "relocate" + copy = "copy" + move = "move" + + +class StorageOperationEligibilityState(str, Enum): + eligible = "eligible" + ineligible = "ineligible" + + +class StorageOperationRunState(str, Enum): + pending = "pending" + running = "running" + completed = "completed" + failed = "failed" + + +class StorageOperationRunItemState(str, Enum): + pending = "pending" + running = "running" + succeeded = "succeeded" + failed = "failed" + skipped = "skipped" + + +class StorageOperationPreviewRequest(Model): + mode: StorageOperationMode + target_object_store_id: str + items: Optional[list[HistoryContentItem]] = None + + +class StorageOperationPreviewItemResult(Model): + item: EncodedHistoryContentItem + state: StorageOperationEligibilityState + reason_code: Optional[str] = None + message: Optional[str] = None + + +class StorageOperationSelectionCounts(Model): + selected_items_count: int + expanded_leaf_count: int + unique_dataset_count: int + + +class StorageOperationEligibilitySummary(Model): + eligible_count: int + ineligible_count: int + items: list[StorageOperationPreviewItemResult] + + +class StorageOperationEstimateSummary(Model): + bytes_to_transfer: int = 0 + quota_delta_by_source: dict[str, int] = Field(default_factory=dict) + + +class StorageOperationPreviewResponse(Model): + snapshot_id: EncodedDatabaseIdField + selection_counts: StorageOperationSelectionCounts + eligibility: StorageOperationEligibilitySummary + estimates: StorageOperationEstimateSummary + warnings: list[str] = Field(default_factory=list) + expires_at: datetime + + +class StorageOperationExecutePolicy(Model): + skip_ineligible: bool = True + max_retries: Optional[int] = None + + +class StorageOperationExecuteRequest(Model): + snapshot_id: DecodedDatabaseIdField + execution_policy: StorageOperationExecutePolicy = Field(default_factory=StorageOperationExecutePolicy) + + +class StorageOperationRunSummary(Model): + run_id: EncodedDatabaseIdField + state: StorageOperationRunState + mode: StorageOperationMode + target_object_store_id: str + total_count: int + succeeded_count: int + failed_count: int + skipped_count: int + task_id: Optional[UUID4] = None + + +class StorageOperationRunItemStatus(Model): + dataset_id: EncodedDatabaseIdField + state: StorageOperationRunItemState + reason_code: Optional[str] = None + message: Optional[str] = None + + +class StorageOperationExecuteResponse(Model): + run: StorageOperationRunSummary + + +class StorageOperationRunResponse(Model): + run: StorageOperationRunSummary + items: list[StorageOperationRunItemStatus] + + class UpdateHistoryContentsPayload(Model): """Can contain arbitrary/dynamic fields that will be updated for a particular history item.""" diff --git a/lib/galaxy/webapps/galaxy/api/history_contents.py b/lib/galaxy/webapps/galaxy/api/history_contents.py index 54df48caef3..89781afa50f 100644 --- a/lib/galaxy/webapps/galaxy/api/history_contents.py +++ b/lib/galaxy/webapps/galaxy/api/history_contents.py @@ -52,6 +52,11 @@ from galaxy.schema.schema import ( HistoryContentType, MaterializeDatasetInstanceAPIRequest, MaterializeDatasetInstanceRequest, + StorageOperationExecuteRequest, + StorageOperationExecuteResponse, + StorageOperationPreviewRequest, + StorageOperationPreviewResponse, + StorageOperationRunResponse, StoreExportPayload, UpdateHistoryContentsBatchPayload, UpdateHistoryContentsPayload, @@ -801,6 +806,43 @@ class FastAPIHistoryContents: """ return self.service.bulk_operation(trans, history_id, filter_query_params, payload) + @router.post( + "/api/histories/{history_id}/contents/bulk/storage/preview", + summary="Previews a storage bulk operation for selected history contents.", + ) + def storage_operation_preview( + self, + history_id: HistoryIDPathParam, + trans: ProvidesHistoryContext = DependsOnTrans, + filter_query_params: ValueFilterQueryParams = Depends(get_value_filter_query_params), + payload: StorageOperationPreviewRequest = Body(...), + ) -> StorageOperationPreviewResponse: + return self.service.storage_operation_preview(trans, history_id, filter_query_params, payload) + + @router.post( + "/api/histories/{history_id}/contents/bulk/storage/execute", + summary="Executes a previously previewed storage bulk operation snapshot.", + ) + def storage_operation_execute( + self, + history_id: HistoryIDPathParam, + trans: ProvidesHistoryContext = DependsOnTrans, + payload: StorageOperationExecuteRequest = Body(...), + ) -> StorageOperationExecuteResponse: + return self.service.storage_operation_execute(trans, history_id, payload) + + @router.get( + "/api/histories/{history_id}/contents/bulk/storage/runs/{run_id}", + summary="Returns run status and per-item details for a storage bulk operation.", + ) + def storage_operation_run( + self, + history_id: HistoryIDPathParam, + run_id: DecodedDatabaseIdField, + trans: ProvidesHistoryContext = DependsOnTrans, + ) -> StorageOperationRunResponse: + return self.service.storage_operation_run(trans, history_id, run_id) + @router.put( "/api/histories/{history_id}/contents/{id}/validate", summary="Validates the metadata associated with a dataset within a History.", diff --git a/lib/galaxy/webapps/galaxy/services/history_contents.py b/lib/galaxy/webapps/galaxy/services/history_contents.py index 53926007bd3..2c19e8900c3 100644 --- a/lib/galaxy/webapps/galaxy/services/history_contents.py +++ b/lib/galaxy/webapps/galaxy/services/history_contents.py @@ -2,6 +2,7 @@ import logging import os import re from collections.abc import Iterable +from datetime import timedelta from typing import ( Any, cast, @@ -10,19 +11,19 @@ from typing import ( TYPE_CHECKING, Union, ) +from uuid import UUID from celery import chain from pydantic import ( ConfigDict, Field, ) -from typing_extensions import ( - Protocol, -) +from typing_extensions import Protocol from galaxy import exceptions from galaxy.celery.helpers import async_task_summary from galaxy.celery.tasks import ( + bulk_relocate_storage, change_datatype, materialize as materialize_task, prepare_dataset_collection_download, @@ -31,6 +32,7 @@ from galaxy.celery.tasks import ( write_history_content_to, ) from galaxy.managers import ( + datasets, folders, hdas, hdcas, @@ -57,12 +59,17 @@ from galaxy.managers.jobs import ( ) from galaxy.managers.library_datasets import LibraryDatasetsManager from galaxy.model import ( + Dataset, + DatasetStorageOperationRun, + DatasetStorageOperationRunItem, + DatasetStorageOperationSnapshot, History, HistoryDatasetAssociation, HistoryDatasetCollectionAssociation, LibraryDataset, User, ) +from galaxy.model.orm.now import now from galaxy.model.security import GalaxyRBACAgent from galaxy.objectstore import BaseObjectStore from galaxy.schema import ( @@ -100,6 +107,21 @@ from galaxy.schema.schema import ( JobSourceType, MaterializeDatasetInstanceRequest, Model, + StorageOperationEligibilityState, + StorageOperationEligibilitySummary, + StorageOperationEstimateSummary, + StorageOperationExecuteRequest, + StorageOperationExecuteResponse, + StorageOperationMode, + StorageOperationPreviewItemResult, + StorageOperationPreviewRequest, + StorageOperationPreviewResponse, + StorageOperationRunItemState, + StorageOperationRunItemStatus, + StorageOperationRunResponse, + StorageOperationRunState, + StorageOperationRunSummary, + StorageOperationSelectionCounts, StoreContentSource, StoreExportPayload, TagOperationParams, @@ -259,12 +281,15 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto and pydantic models to declare its parameters and return types. """ + _storage_snapshot_ttl = timedelta(hours=24) + def __init__( self, security: IdEncodingHelper, object_store: BaseObjectStore, history_manager: histories.HistoryManager, history_contents_manager: HistoryContentsManager, + dataset_manager: datasets.DatasetManager, hda_manager: hdas.HDAManager, hdca_manager: hdcas.HDCAManager, dataset_collection_manager: DatasetCollectionManager, @@ -280,6 +305,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto super().__init__(security) self.history_manager = history_manager self.history_contents_manager = history_contents_manager + self.dataset_manager = dataset_manager self.hda_manager = hda_manager self.hdca_manager = hdca_manager self.dataset_collection_manager = dataset_collection_manager @@ -734,6 +760,269 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto success_count = len(contents) - len(errors) return HistoryContentBulkOperationResult(success_count=success_count, errors=errors) + def storage_operation_preview( + self, + trans: ProvidesHistoryContext, + history_id: DecodedDatabaseIdField, + filter_query_params: ValueFilterQueryParams, + payload: StorageOperationPreviewRequest, + ) -> StorageOperationPreviewResponse: + if payload.mode != StorageOperationMode.relocate: + raise exceptions.RequestParameterInvalidException("Only 'relocate' mode is supported in phase 1.") + + user = self.get_authenticated_user(trans) + history = self.history_manager.get_mutable(history_id, user, current_history=trans.history) + filters = self.history_contents_filters.parse_query_filters(filter_query_params) + if payload.items: + contents = self._get_contents_by_item_list(trans, history, payload.items) + else: + contents = self.history_contents_manager.contents(history, filters) + + selected_items_count = len(contents) + unique_datasets, expanded_leaf_count = self._resolve_unique_datasets_from_contents(contents) + + eligible_dataset_ids: list[int] = [] + eligibility_items: list[StorageOperationPreviewItemResult] = [] + eligible_count = 0 + ineligible_count = 0 + quota_delta_by_source: dict[str, int] = {} + quota_source_map = self.object_store.get_quota_source_map() + + for dataset_id, dataset in unique_datasets.items(): + reason = self._validate_relocate_dataset(trans, dataset, payload.target_object_store_id) + item = EncodedHistoryContentItem( + id=self.encode_id(dataset_id), + history_content_type=HistoryContentType.dataset, + ) + if reason is None: + eligible_count += 1 + eligible_dataset_ids.append(dataset_id) + eligibility_items.append( + StorageOperationPreviewItemResult(item=item, state=StorageOperationEligibilityState.eligible) + ) + if quota_source_map: + old_label = quota_source_map.get_quota_source_label(dataset.object_store_id) or "default" + new_label = quota_source_map.get_quota_source_label(payload.target_object_store_id) or "default" + if old_label != new_label: + key = f"{old_label}->{new_label}" + dataset_size = int(dataset.get_total_size() or 0) + quota_delta_by_source[key] = quota_delta_by_source.get(key, 0) + dataset_size + else: + ineligible_count += 1 + reason_code, message = reason + eligibility_items.append( + StorageOperationPreviewItemResult( + item=item, + state=StorageOperationEligibilityState.ineligible, + reason_code=reason_code, + message=message, + ) + ) + + expires_at = now() + self._storage_snapshot_ttl + snapshot = DatasetStorageOperationSnapshot( + history_id=history.id, + user_id=user.id, + mode=payload.mode.value, + target_object_store_id=payload.target_object_store_id, + resolved_dataset_ids=list(unique_datasets.keys()), + eligible_dataset_ids=eligible_dataset_ids, + expires_at=expires_at, + ) + trans.sa_session.add(snapshot) + trans.sa_session.commit() + + warnings: list[str] = [] + if not payload.items: + warnings.append("Selection was query-based. Execute uses a fixed snapshot to avoid selection drift.") + + return StorageOperationPreviewResponse( + snapshot_id=self.encode_id(snapshot.id), + selection_counts=StorageOperationSelectionCounts( + selected_items_count=selected_items_count, + expanded_leaf_count=expanded_leaf_count, + unique_dataset_count=len(unique_datasets), + ), + eligibility=StorageOperationEligibilitySummary( + eligible_count=eligible_count, + ineligible_count=ineligible_count, + items=eligibility_items, + ), + estimates=StorageOperationEstimateSummary( + bytes_to_transfer=0, + quota_delta_by_source=quota_delta_by_source, + ), + warnings=warnings, + expires_at=expires_at, + ) + + def storage_operation_execute( + self, + trans: ProvidesHistoryContext, + history_id: DecodedDatabaseIdField, + payload: StorageOperationExecuteRequest, + ) -> StorageOperationExecuteResponse: + user = self.get_authenticated_user(trans) + history = self.history_manager.get_mutable(history_id, user, current_history=trans.history) + snapshot = self._get_storage_snapshot(trans, payload.snapshot_id) + if snapshot.history_id != history.id: + raise exceptions.RequestParameterInvalidException("Snapshot does not belong to the requested history.") + if snapshot.user_id != user.id: + raise exceptions.RequestParameterInvalidException("Snapshot does not belong to the current user.") + if snapshot.mode != StorageOperationMode.relocate.value: + raise exceptions.RequestParameterInvalidException("Only 'relocate' mode is supported in phase 1.") + + run = DatasetStorageOperationRun( + snapshot_id=snapshot.id, + history_id=history.id, + user_id=user.id, + mode=snapshot.mode, + target_object_store_id=snapshot.target_object_store_id, + state=StorageOperationRunState.pending.value, + skip_ineligible=payload.execution_policy.skip_ineligible, + ) + trans.sa_session.add(run) + resolved_dataset_ids = snapshot.resolved_dataset_ids + run.total_count = len(resolved_dataset_ids) + trans.sa_session.add(run) + trans.sa_session.commit() + + task_result = bulk_relocate_storage.delay( + run_db_id=run.id, + task_user_id=user.id, + ) + run.task_id = UUID(task_result.id) + trans.sa_session.add(run) + trans.sa_session.commit() + + run_summary = StorageOperationRunSummary( + run_id=self.encode_id(run.id), + state=StorageOperationRunState.pending, + mode=StorageOperationMode.relocate, + target_object_store_id=run.target_object_store_id, + total_count=len(resolved_dataset_ids), + succeeded_count=0, + failed_count=0, + skipped_count=0, + task_id=UUID(str(run.task_id)) if run.task_id is not None else None, + ) + return StorageOperationExecuteResponse(run=run_summary) + + def storage_operation_run( + self, + trans: ProvidesHistoryContext, + history_id: DecodedDatabaseIdField, + run_id: DecodedDatabaseIdField, + ) -> StorageOperationRunResponse: + user = self.get_authenticated_user(trans) + history = self.history_manager.get_mutable(history_id, user, current_history=trans.history) + run = trans.sa_session.get(DatasetStorageOperationRun, run_id) + if run is None: + raise exceptions.ObjectNotFound("Storage operation run not found.") + if run.history_id != history.id: + raise exceptions.RequestParameterInvalidException("Run does not belong to the requested history.") + if run.user_id != user.id: + raise exceptions.RequestParameterInvalidException("Run does not belong to the current user.") + + run_items = ( + trans.sa_session.query(DatasetStorageOperationRunItem) + .filter(DatasetStorageOperationRunItem.run_id == run.id) + .order_by(DatasetStorageOperationRunItem.id.asc()) + .all() + ) + + summary = StorageOperationRunSummary( + run_id=self.encode_id(run.id), + state=StorageOperationRunState(run.state), + mode=StorageOperationMode(run.mode), + target_object_store_id=run.target_object_store_id, + total_count=run.total_count, + succeeded_count=run.succeeded_count, + failed_count=run.failed_count, + skipped_count=run.skipped_count, + task_id=UUID(str(run.task_id)) if run.task_id is not None else None, + ) + items = [ + StorageOperationRunItemStatus( + dataset_id=self.encode_id(run_item.dataset_id), + state=StorageOperationRunItemState(run_item.state), + reason_code=run_item.reason_code, + message=run_item.message, + ) + for run_item in run_items + ] + return StorageOperationRunResponse(run=summary, items=items) + + def _resolve_unique_datasets_from_contents( + self, + contents: list["HistoryItem"], + ) -> tuple[dict[int, Dataset], int]: + unique_datasets: dict[int, Dataset] = {} + expanded_leaf_count = 0 + + for content in contents: + if isinstance(content, HistoryDatasetAssociation): + hda_dataset = content.dataset + if hda_dataset is None: + continue + expanded_leaf_count += 1 + unique_datasets[hda_dataset.id] = hda_dataset + elif isinstance(content, HistoryDatasetCollectionAssociation): + hdas = self.hdca_manager.map_datasets(content, fn=lambda item, *parents: item) + for hda in hdas: + if isinstance(hda, HistoryDatasetAssociation): + hda_dataset = hda.dataset + if hda_dataset is None: + continue + expanded_leaf_count += 1 + unique_datasets[hda_dataset.id] = hda_dataset + + return unique_datasets, expanded_leaf_count + + def _validate_relocate_dataset( + self, + trans: ProvidesHistoryContext, + dataset: Dataset, + target_object_store_id: str, + ) -> Optional[tuple[str, str]]: + device_source_map = self.object_store.get_device_source_map() + target_device_id = device_source_map.get_device_id(target_object_store_id) + if target_device_id is None: + return "invalid_target_object_store", f"Target object store '{target_object_store_id}' is invalid." + + source_object_store_id = dataset.object_store_id + if source_object_store_id is None: + return "missing_source_object_store", "Dataset does not define a source object store." + source_device_id = device_source_map.get_device_id(source_object_store_id) + if source_object_store_id == target_object_store_id: + return "already_in_target", "Dataset is already in the requested object store." + if source_device_id != target_device_id: + return ( + "cross_device_relocate_not_allowed", + "Cannot relocate to an object store that does not share the same device.", + ) + + if not trans.app.security_agent.can_change_object_store_id(trans.user, dataset): + return "insufficient_permissions", "Dataset is not eligible for object store changes." + + ok_to_edit_metadata = getattr(dataset, "ok_to_edit_metadata", None) + if callable(ok_to_edit_metadata) and not ok_to_edit_metadata(): + return "dataset_in_use", "Dataset is currently used by an active job." + + return None + + def _get_storage_snapshot( + self, trans: ProvidesHistoryContext, snapshot_id: DecodedDatabaseIdField + ) -> DatasetStorageOperationSnapshot: + snapshot = trans.sa_session.get(DatasetStorageOperationSnapshot, snapshot_id) + if snapshot is None: + raise exceptions.ObjectNotFound("Storage operation snapshot not found.") + if snapshot.expires_at <= now(): + trans.sa_session.delete(snapshot) + trans.sa_session.commit() + raise exceptions.RequestParameterInvalidException("Storage operation snapshot has expired.") + return snapshot + def validate(self, trans, history_id: DecodedDatabaseIdField, history_content_id: DecodedDatabaseIdField): """ Validates the metadata associated with a dataset within a History. From 78c513b30ea9e5972383b5863afbd7ba46bbded2 Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 16 Apr 2026 16:18:50 +0200 Subject: [PATCH 002/120] Add OpenAPI types and endpoints for storage operations --- client/src/api/schema/schema.ts | 346 ++++++++++++++++++++++++++++++++ 1 file changed, 346 insertions(+) diff --git a/client/src/api/schema/schema.ts b/client/src/api/schema/schema.ts index 79b0abc1924..6d38758e208 100644 --- a/client/src/api/schema/schema.ts +++ b/client/src/api/schema/schema.ts @@ -2339,6 +2339,57 @@ export interface paths { patch?: never; trace?: never; }; + "/api/histories/{history_id}/contents/bulk/storage/execute": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Executes a previously previewed storage bulk operation snapshot. */ + post: operations["storage_operation_execute_api_histories__history_id__contents_bulk_storage_execute_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/api/histories/{history_id}/contents/bulk/storage/preview": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Previews a storage bulk operation for selected history contents. */ + post: operations["storage_operation_preview_api_histories__history_id__contents_bulk_storage_preview_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/api/histories/{history_id}/contents/bulk/storage/runs/{run_id}": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** Returns run status and per-item details for a storage bulk operation. */ + get: operations["storage_operation_run_api_histories__history_id__contents_bulk_storage_runs__run_id__get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/histories/{history_id}/contents/dataset_collections/{hdca_id}/download": { parameters: { query?: never; @@ -22558,6 +22609,155 @@ export interface components { /** Total Item Count */ total_item_count: number; }; + /** + * StorageOperationEligibilityState + * @enum {string} + */ + StorageOperationEligibilityState: "eligible" | "ineligible"; + /** StorageOperationEligibilitySummary */ + StorageOperationEligibilitySummary: { + /** Eligible Count */ + eligible_count: number; + /** Ineligible Count */ + ineligible_count: number; + /** Items */ + items: components["schemas"]["StorageOperationPreviewItemResult"][]; + }; + /** StorageOperationEstimateSummary */ + StorageOperationEstimateSummary: { + /** + * Bytes To Transfer + * @default 0 + */ + bytes_to_transfer: number; + /** Quota Delta By Source */ + quota_delta_by_source?: { + [key: string]: number; + }; + }; + /** StorageOperationExecutePolicy */ + StorageOperationExecutePolicy: { + /** Max Retries */ + max_retries?: number | null; + /** + * Skip Ineligible + * @default true + */ + skip_ineligible: boolean; + }; + /** StorageOperationExecuteRequest */ + StorageOperationExecuteRequest: { + execution_policy?: components["schemas"]["StorageOperationExecutePolicy"]; + /** + * Snapshot Id + * @example 0123456789ABCDEF + */ + snapshot_id: string; + }; + /** StorageOperationExecuteResponse */ + StorageOperationExecuteResponse: { + run: components["schemas"]["StorageOperationRunSummary"]; + }; + /** + * StorageOperationMode + * @enum {string} + */ + StorageOperationMode: "relocate" | "copy" | "move"; + /** StorageOperationPreviewItemResult */ + StorageOperationPreviewItemResult: { + item: components["schemas"]["EncodedHistoryContentItem"]; + /** Message */ + message?: string | null; + /** Reason Code */ + reason_code?: string | null; + state: components["schemas"]["StorageOperationEligibilityState"]; + }; + /** StorageOperationPreviewRequest */ + StorageOperationPreviewRequest: { + /** Items */ + items?: components["schemas"]["HistoryContentItem"][] | null; + mode: components["schemas"]["StorageOperationMode"]; + /** Target Object Store Id */ + target_object_store_id: string; + }; + /** StorageOperationPreviewResponse */ + StorageOperationPreviewResponse: { + eligibility: components["schemas"]["StorageOperationEligibilitySummary"]; + estimates: components["schemas"]["StorageOperationEstimateSummary"]; + /** + * Expires At + * Format: date-time + */ + expires_at: string; + selection_counts: components["schemas"]["StorageOperationSelectionCounts"]; + /** + * Snapshot Id + * @example 0123456789ABCDEF + */ + snapshot_id: string; + /** Warnings */ + warnings?: string[]; + }; + /** + * StorageOperationRunItemState + * @enum {string} + */ + StorageOperationRunItemState: "pending" | "running" | "succeeded" | "failed" | "skipped"; + /** StorageOperationRunItemStatus */ + StorageOperationRunItemStatus: { + /** + * Dataset Id + * @example 0123456789ABCDEF + */ + dataset_id: string; + /** Message */ + message?: string | null; + /** Reason Code */ + reason_code?: string | null; + state: components["schemas"]["StorageOperationRunItemState"]; + }; + /** StorageOperationRunResponse */ + StorageOperationRunResponse: { + /** Items */ + items: components["schemas"]["StorageOperationRunItemStatus"][]; + run: components["schemas"]["StorageOperationRunSummary"]; + }; + /** + * StorageOperationRunState + * @enum {string} + */ + StorageOperationRunState: "pending" | "running" | "completed" | "failed"; + /** StorageOperationRunSummary */ + StorageOperationRunSummary: { + /** Failed Count */ + failed_count: number; + mode: components["schemas"]["StorageOperationMode"]; + /** + * Run Id + * @example 0123456789ABCDEF + */ + run_id: string; + /** Skipped Count */ + skipped_count: number; + state: components["schemas"]["StorageOperationRunState"]; + /** Succeeded Count */ + succeeded_count: number; + /** Target Object Store Id */ + target_object_store_id: string; + /** Task Id */ + task_id?: string | null; + /** Total Count */ + total_count: number; + }; + /** StorageOperationSelectionCounts */ + StorageOperationSelectionCounts: { + /** Expanded Leaf Count */ + expanded_leaf_count: number; + /** Selected Items Count */ + selected_items_count: number; + /** Unique Dataset Count */ + unique_dataset_count: number; + }; /** StoreExportPayload */ StoreExportPayload: { /** @@ -37765,6 +37965,152 @@ export interface operations { }; }; }; + storage_operation_execute_api_histories__history_id__contents_bulk_storage_execute_post: { + parameters: { + query?: never; + header?: { + /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */ + "run-as"?: string | null; + }; + path: { + /** @description The encoded database identifier of the History. */ + history_id: string; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["StorageOperationExecuteRequest"]; + }; + }; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["StorageOperationExecuteResponse"]; + }; + }; + /** @description Request Error */ + "4XX": { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MessageExceptionModel"]; + }; + }; + /** @description Server Error */ + "5XX": { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MessageExceptionModel"]; + }; + }; + }; + }; + storage_operation_preview_api_histories__history_id__contents_bulk_storage_preview_post: { + parameters: { + query?: { + /** @description Generally a property name to filter by followed by an (often optional) hyphen and operator string. */ + q?: string[] | null; + /** @description The value to filter by. */ + qv?: string[] | null; + }; + header?: { + /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */ + "run-as"?: string | null; + }; + path: { + /** @description The encoded database identifier of the History. */ + history_id: string; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["StorageOperationPreviewRequest"]; + }; + }; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["StorageOperationPreviewResponse"]; + }; + }; + /** @description Request Error */ + "4XX": { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MessageExceptionModel"]; + }; + }; + /** @description Server Error */ + "5XX": { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MessageExceptionModel"]; + }; + }; + }; + }; + storage_operation_run_api_histories__history_id__contents_bulk_storage_runs__run_id__get: { + parameters: { + query?: never; + header?: { + /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */ + "run-as"?: string | null; + }; + path: { + /** @description The encoded database identifier of the History. */ + history_id: string; + run_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["StorageOperationRunResponse"]; + }; + }; + /** @description Request Error */ + "4XX": { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MessageExceptionModel"]; + }; + }; + /** @description Server Error */ + "5XX": { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MessageExceptionModel"]; + }; + }; + }; + }; history_contents__download_collection: { parameters: { query?: never; From 18cab50731ec80a2e7e2d895ff904eee54d2383b Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 16 Apr 2026 16:18:51 +0200 Subject: [PATCH 003/120] Add client API and models for storage operation flows --- client/src/api/histories.ts | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/client/src/api/histories.ts b/client/src/api/histories.ts index 03d3542feff..f411191b5ec 100644 --- a/client/src/api/histories.ts +++ b/client/src/api/histories.ts @@ -51,6 +51,11 @@ export type PublishedHistory = SharedHistory & { */ export type AnyHistoryEntry = MyHistory | SharedHistory | PublishedHistory | ArchivedHistorySummary; +/** + * Represents a reference to a history, containing only the fields necessary for identifying a history in API calls. + */ +export type HistoryReference = Pick; + /** * Represents the options for fetching histories. */ @@ -301,6 +306,29 @@ export async function getHistoryCounts(historyId: string): Promise { + const { data, error } = await GalaxyApi().GET("/api/histories/{history_id}/contents/bulk/storage/runs/{run_id}", { + params: { + path: { history_id: history.id, run_id: runId }, + }, + }); + + if (error) { + rethrowSimple(error); + } + + return data; +} + /** * Fetches the workflow extraction summary for a specific history entry. * @param {string} historyId The ID of the history entry to fetch the workflow extraction summary for From 6d86d2be7a8c2ac60a8dded88420f2fe23cf567a Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 16 Apr 2026 16:18:51 +0200 Subject: [PATCH 004/120] Add composition utilities for polling storage run status --- .../src/composables/useStorageRunWatcher.ts | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 client/src/composables/useStorageRunWatcher.ts diff --git a/client/src/composables/useStorageRunWatcher.ts b/client/src/composables/useStorageRunWatcher.ts new file mode 100644 index 00000000000..66e649c092d --- /dev/null +++ b/client/src/composables/useStorageRunWatcher.ts @@ -0,0 +1,33 @@ +import { computed, ref } from "vue"; + +import { getStorageOperationRunStatus, type HistoryReference, type StorageOperationRunResponse } from "@/api/histories"; +import { useResourceWatcher } from "@/composables/resourceWatcher"; + +const STORAGE_RUN_TERMINAL_STATES = new Set(["completed", "failed"]); +const STORAGE_RUN_POLLING_INTERVAL = 10000; // 10 seconds + +/** + * Poll storage-operation run status until it reaches a terminal state. + */ +export function useStorageRunWatcher(history: HistoryReference, runId: string) { + const runStatus = ref(null); + const isTerminal = computed(() => { + const state = runStatus.value?.run.state; + return state !== undefined && STORAGE_RUN_TERMINAL_STATES.has(state); + }); + + async function fetchStatus() { + const data = await getStorageOperationRunStatus(history, runId); + runStatus.value = data ?? null; + if (isTerminal.value) { + stopPolling(); + } + } + + const { startWatchingResource: startPolling, stopWatchingResource: stopPolling } = useResourceWatcher(fetchStatus, { + shortPollingInterval: STORAGE_RUN_POLLING_INTERVAL, + enableBackgroundPolling: false, + }); + + return { runStatus, isTerminal, startPolling, stopPolling }; +} From 19e2f2d7bf1ed4c74e0ba34bb8d4410cc4a524b1 Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 16 Apr 2026 16:18:52 +0200 Subject: [PATCH 005/120] Integrate storage operation APIs into history model queries --- .../src/components/History/model/queries.ts | 69 ++++++++++++++++++- 1 file changed, 66 insertions(+), 3 deletions(-) diff --git a/client/src/components/History/model/queries.ts b/client/src/components/History/model/queries.ts index 8c49e23d0dc..f071fdfcfe9 100644 --- a/client/src/components/History/model/queries.ts +++ b/client/src/components/History/model/queries.ts @@ -1,5 +1,15 @@ -import { type components, GalaxyApi, type HistoryItemSummary, type HistorySummary } from "@/api"; +import type { components, HistoryItemSummary } from "@/api"; +import { GalaxyApi } from "@/api"; import { createHistoryDatasetCollectionInstanceSimple } from "@/api/datasetCollections"; +import type { + HistoryReference, + StorageOperationExecutePolicy, + StorageOperationExecuteResponse, + StorageOperationMode, + StorageOperationPreviewResponse, + StorageOperationRunResponse, +} from "@/api/histories"; +import { getStorageOperationRunStatus } from "@/api/histories"; import { rethrowSimple } from "@/utils/simple-error"; type BulkOperation = components["schemas"]["HistoryContentItemOperation"]; @@ -57,7 +67,7 @@ export async function updateContentFields(content: HistoryItemSummary, newFields * the filters will determine which items are processed. */ export async function bulkUpdate( - history: HistorySummary, + history: HistoryReference, operation: BulkOperation, filters: QueryFilters, items = [], @@ -81,7 +91,7 @@ export async function bulkUpdate( return data; } -export async function createDatasetCollection(history: HistorySummary, inputs = {}) { +export async function createDatasetCollection(history: HistoryReference, inputs = {}) { const defaults = { collection_type: "list", copy_elements: true, @@ -94,3 +104,56 @@ export async function createDatasetCollection(history: HistorySummary, inputs = const payload = Object.assign({}, defaults, inputs); return createHistoryDatasetCollectionInstanceSimple(payload); } + +export async function bulkStoragePreview( + history: HistoryReference, + mode: StorageOperationMode, + targetObjectStoreId: string, + filters: QueryFilters, + items = [], +): Promise { + const { data, error } = await GalaxyApi().POST("/api/histories/{history_id}/contents/bulk/storage/preview", { + params: { + path: { history_id: history.id }, + query: filtersToQueryValues(filters), + }, + body: { + mode, + target_object_store_id: targetObjectStoreId, + items, + }, + }); + + if (error) { + rethrowSimple(error); + } + return data; +} + +export async function bulkStorageExecute( + history: HistoryReference, + snapshotId: string, + executionPolicy: StorageOperationExecutePolicy = { skip_ineligible: true }, +): Promise { + const { data, error } = await GalaxyApi().POST("/api/histories/{history_id}/contents/bulk/storage/execute", { + params: { + path: { history_id: history.id }, + }, + body: { + snapshot_id: snapshotId, + execution_policy: executionPolicy, + }, + }); + + if (error) { + rethrowSimple(error); + } + return data; +} + +export async function bulkStorageRunStatus( + history: HistoryReference, + runId: string, +): Promise { + return getStorageOperationRunStatus(history, runId); +} From e91471ab6804b6dd772af95de880ade9db5849e9 Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:35:27 +0200 Subject: [PATCH 006/120] Add backend notifications for storage operations --- lib/galaxy/celery/tasks.py | 88 ++++++++++++++++++- lib/galaxy/schema/notifications.py | 29 ++++++ lib/galaxy/webapps/galaxy/buildapp.py | 2 + .../galaxy/services/history_contents.py | 55 +++++++++++- 4 files changed, 170 insertions(+), 4 deletions(-) diff --git a/lib/galaxy/celery/tasks.py b/lib/galaxy/celery/tasks.py index d8f68aa770a..5690ea67794 100644 --- a/lib/galaxy/celery/tasks.py +++ b/lib/galaxy/celery/tasks.py @@ -62,7 +62,15 @@ from galaxy.model.scoped_session import galaxy_scoped_session from galaxy.objectstore import BaseObjectStore from galaxy.objectstore.caching import check_caches from galaxy.queue_worker import GalaxyQueueWorker -from galaxy.schema.notifications import NotificationCreateRequest +from galaxy.schema.notifications import ( + NotificationCreateData, + NotificationCreateRequest, + NotificationRecipients, + NotificationVariant, + PersonalNotificationCategory, + StorageOperationNotificationContent, + StorageOperationNotificationState, +) from galaxy.schema.tasks import ( ComputeDatasetHashTaskRequest, GenerateHistoryContentDownload, @@ -317,6 +325,7 @@ def set_metadata( def bulk_relocate_storage( sa_session: galaxy_scoped_session, dataset_manager: DatasetManager, + notification_manager: NotificationManager, app: MinimalManagerApp, run_db_id: int, task_user_id: Optional[int] = None, @@ -326,18 +335,78 @@ def bulk_relocate_storage( log.error(f"Bulk storage relocate run not found: {run_db_id}") return + user = sa_session.get(User, run.user_id) if run.user_id else None + + def send_storage_notification( + *, + state: StorageOperationNotificationState, + variant: NotificationVariant, + message: str, + ): + if user is None: + return + encoded_history_id = app.security.encode_id(run.history_id) + encoded_run_id = app.security.encode_id(run.id) + relative_run_url = f"/histories/{encoded_history_id}/storage/runs/{encoded_run_id}" + galaxy_url = app.config.galaxy_infrastructure_url + run_url = f"{galaxy_url}{relative_run_url}" if galaxy_url else relative_run_url + try: + notification_request = NotificationCreateRequest( + recipients=NotificationRecipients.model_construct(user_ids=[user.id]), + notification=NotificationCreateData( + source="galaxy", + category=PersonalNotificationCategory.storage_operation, + variant=variant, + publication_time=None, + expiration_time=None, + content=StorageOperationNotificationContent( + subject=( + "Storage operation completed" + if state == StorageOperationNotificationState.completed + else "Storage operation failed" + ), + message=message, + history_id=encoded_history_id, + run_id=encoded_run_id, + run_url=run_url, + mode=run.mode, + state=state, + total_count=run.total_count, + succeeded_count=run.succeeded_count, + failed_count=run.failed_count, + skipped_count=run.skipped_count, + ), + ), + galaxy_url=galaxy_url, + ) + notification_manager.send_notification_to_recipients(notification_request) + except Exception: + log.exception("Failed to send storage operation notification for run %s", run.id) + snapshot = sa_session.get(DatasetStorageOperationSnapshot, run.snapshot_id) if snapshot is None: run.state = "failed" + run.failed_count = run.total_count if run.total_count else 0 sa_session.add(run) sa_session.commit() + send_storage_notification( + state=StorageOperationNotificationState.failed, + variant=NotificationVariant.urgent, + message="Bulk relocate run failed because its preview snapshot could not be found.", + ) log.error(f"Snapshot not found for run {run_db_id}") return if snapshot.expires_at <= now(): run.state = "failed" + run.failed_count = run.total_count if run.total_count else 0 sa_session.add(run) sa_session.commit() + send_storage_notification( + state=StorageOperationNotificationState.failed, + variant=NotificationVariant.urgent, + message="Bulk relocate run failed because its preview snapshot expired before execution.", + ) log.error(f"Snapshot expired for run {run_db_id}") return @@ -345,7 +414,6 @@ def bulk_relocate_storage( sa_session.add(run) sa_session.commit() - user = sa_session.get(User, run.user_id) if run.user_id else None trans = SimpleNamespace(user=user) succeeded_count = 0 @@ -438,6 +506,22 @@ def bulk_relocate_storage( sa_session.add(run) sa_session.commit() + if failed_count > 0 or skipped_count > 0: + variant = NotificationVariant.warning + message = ( + f"Bulk relocate run finished with partial success. " + f"Succeeded: {succeeded_count}, Failed: {failed_count}, Skipped: {skipped_count}." + ) + else: + variant = NotificationVariant.info + message = f"Bulk relocate run completed successfully for {succeeded_count} dataset(s)." + + send_storage_notification( + state=StorageOperationNotificationState.completed, + variant=variant, + message=message, + ) + def _get_dataset_manager( hda_manager: HDAManager, ldda_manager: LDDAManager, model_class: str = "HistoryDatasetAssociation" diff --git a/lib/galaxy/schema/notifications.py b/lib/galaxy/schema/notifications.py index a59b9d9b624..7aaa1a6c2a5 100644 --- a/lib/galaxy/schema/notifications.py +++ b/lib/galaxy/schema/notifications.py @@ -61,6 +61,7 @@ class PersonalNotificationCategory(str, Enum): message = "message" new_shared_item = "new_shared_item" + storage_operation = "storage_operation" # TODO: enable this and create content model when we have a hook for completed workflows # workflow_execution_completed = "workflow_execution_completed" @@ -115,6 +116,33 @@ class NewSharedItemNotificationContent(Model): slug: str = Field(..., title="Slug", description="The slug of the shared item. Used for the link to the item.") +class StorageOperationNotificationState(str, Enum): + started = "started" + completed = "completed" + failed = "failed" + + +class StorageOperationNotificationContent(MessageNotificationContentBase): + category: Literal[PersonalNotificationCategory.storage_operation] = PersonalNotificationCategory.storage_operation + history_id: EncodedDatabaseIdField = Field(..., title="History ID", description="The encoded history ID.") + run_id: EncodedDatabaseIdField = Field(..., title="Run ID", description="The encoded storage operation run ID.") + run_url: AbsoluteOrRelativeUrl = Field( + ..., + title="Run URL", + description="Absolute or relative URL to the storage operation run status view.", + ) + mode: str = Field(..., title="Mode", description="Storage operation mode.") + state: StorageOperationNotificationState = Field( + ..., + title="State", + description="The current state of the storage operation run when this notification was generated.", + ) + total_count: int = Field(..., title="Total Count", description="Total datasets in the run.") + succeeded_count: int = Field(default=0, title="Succeeded Count", description="Succeeded datasets count.") + failed_count: int = Field(default=0, title="Failed Count", description="Failed datasets count.") + skipped_count: int = Field(default=0, title="Skipped Count", description="Skipped datasets count.") + + NotificationContentField = Field( default=..., discriminator="category", @@ -126,6 +154,7 @@ AnyUserNotificationContent = Annotated[ Union[ MessageNotificationContent, NewSharedItemNotificationContent, + StorageOperationNotificationContent, ], NotificationContentField, ] diff --git a/lib/galaxy/webapps/galaxy/buildapp.py b/lib/galaxy/webapps/galaxy/buildapp.py index e4d35f253bf..f71731d6311 100644 --- a/lib/galaxy/webapps/galaxy/buildapp.py +++ b/lib/galaxy/webapps/galaxy/buildapp.py @@ -279,6 +279,8 @@ def app_pair(global_conf, load_app_kwds=None, wsgi_preflight=True, **kwargs): webapp.add_client_route("/histories/{history_id}/graph") webapp.add_client_route("/histories/{history_id}/pages") webapp.add_client_route("/histories/{history_id}/pages/{page_id}") + webapp.add_client_route("/histories/{history_id}/storage/operations") + webapp.add_client_route("/histories/{history_id}/storage/runs/{path:.*?}") # change to /storage/operations/runs? webapp.add_client_route("/histories/archived") webapp.add_client_route("/histories/list_published") webapp.add_client_route("/histories/list_shared") diff --git a/lib/galaxy/webapps/galaxy/services/history_contents.py b/lib/galaxy/webapps/galaxy/services/history_contents.py index 2c19e8900c3..28ed716825a 100644 --- a/lib/galaxy/webapps/galaxy/services/history_contents.py +++ b/lib/galaxy/webapps/galaxy/services/history_contents.py @@ -81,6 +81,15 @@ from galaxy.schema.fields import ( DecodedDatabaseIdField, LibraryFolderDatabaseIdField, ) +from galaxy.schema.notifications import ( + NotificationCreateData, + NotificationCreateRequest, + NotificationRecipients, + NotificationVariant, + PersonalNotificationCategory, + StorageOperationNotificationContent, + StorageOperationNotificationState, +) from galaxy.schema.schema import ( AnyBulkOperationParams, AnyHistoryContentItem, @@ -896,16 +905,56 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto trans.sa_session.commit() run_summary = StorageOperationRunSummary( - run_id=self.encode_id(run.id), + run_id=run.id, state=StorageOperationRunState.pending, mode=StorageOperationMode.relocate, target_object_store_id=run.target_object_store_id, + create_time=run.create_time, + update_time=run.update_time, total_count=len(resolved_dataset_ids), succeeded_count=0, failed_count=0, skipped_count=0, task_id=UUID(str(run.task_id)) if run.task_id is not None else None, ) + + encoded_history_id = self.encode_id(history.id) + encoded_run_id = self.encode_id(run.id) + relative_run_url = f"/histories/{encoded_history_id}/storage/runs/{encoded_run_id}" + galaxy_url = trans.app.config.galaxy_infrastructure_url + run_url = f"{galaxy_url}{relative_run_url}" if galaxy_url else relative_run_url + try: + notification_request = NotificationCreateRequest( + recipients=NotificationRecipients.model_construct(user_ids=[user.id]), + notification=NotificationCreateData( + source="galaxy", + category=PersonalNotificationCategory.storage_operation, + variant=NotificationVariant.info, + publication_time=None, + expiration_time=None, + content=StorageOperationNotificationContent( + subject="Storage operation started", + message=( + f"Bulk relocate run started for {len(resolved_dataset_ids)} dataset(s). " + "You can monitor progress in real time." + ), + history_id=encoded_history_id, + run_id=encoded_run_id, + run_url=run_url, + mode=run.mode, + state=StorageOperationNotificationState.started, + total_count=len(resolved_dataset_ids), + succeeded_count=0, + failed_count=0, + skipped_count=0, + ), + ), + galaxy_url=galaxy_url, + ) + trans.app.notification_manager.send_notification_to_recipients(notification_request) + except Exception: + log.exception("Failed to send storage operation start notification for run %s", run.id) + return StorageOperationExecuteResponse(run=run_summary) def storage_operation_run( @@ -932,10 +981,12 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto ) summary = StorageOperationRunSummary( - run_id=self.encode_id(run.id), + run_id=run.id, state=StorageOperationRunState(run.state), mode=StorageOperationMode(run.mode), target_object_store_id=run.target_object_store_id, + create_time=run.create_time, + update_time=run.update_time, total_count=run.total_count, succeeded_count=run.succeeded_count, failed_count=run.failed_count, From 85a0dff5949f247a67cf6a406fa2f4854b791cfa Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:35:28 +0200 Subject: [PATCH 007/120] Update OpenAPI schema for storage operation notification support --- client/src/api/schema/schema.ts | 128 +++++++++++++++++++++++++++++++- 1 file changed, 125 insertions(+), 3 deletions(-) diff --git a/client/src/api/schema/schema.ts b/client/src/api/schema/schema.ts index 6d38758e208..7344d8e88e1 100644 --- a/client/src/api/schema/schema.ts +++ b/client/src/api/schema/schema.ts @@ -19358,6 +19358,7 @@ export interface components { content: | components["schemas"]["MessageNotificationContent"] | components["schemas"]["NewSharedItemNotificationContent"] + | components["schemas"]["StorageOperationNotificationContent"] | components["schemas"]["BroadcastNotificationContent"]; /** * Expiration time @@ -19446,6 +19447,7 @@ export interface components { content: | components["schemas"]["MessageNotificationContent"] | components["schemas"]["NewSharedItemNotificationContent"] + | components["schemas"]["StorageOperationNotificationContent"] | components["schemas"]["BroadcastNotificationContent"]; /** * Create time @@ -20533,7 +20535,7 @@ export interface components { * displayed in the notification preferences. * @enum {string} */ - PersonalNotificationCategory: "message" | "new_shared_item"; + PersonalNotificationCategory: "message" | "new_shared_item" | "storage_operation"; /** PluginAspectStatus */ PluginAspectStatus: { /** Message */ @@ -22663,9 +22665,86 @@ export interface components { * @enum {string} */ StorageOperationMode: "relocate" | "copy" | "move"; + /** StorageOperationNotificationContent */ + StorageOperationNotificationContent: { + /** + * @description discriminator enum property added by openapi-typescript + * @enum {string} + */ + category: "storage_operation"; + /** + * Failed Count + * @description Failed datasets count. + * @default 0 + */ + failed_count: number; + /** + * History ID + * @description The encoded history ID. + * @example 0123456789ABCDEF + */ + history_id: string; + /** + * Message + * @description The message of the notification (supports Markdown). + */ + message: string; + /** + * Mode + * @description Storage operation mode. + */ + mode: string; + /** + * Run ID + * @description The encoded storage operation run ID. + * @example 0123456789ABCDEF + */ + run_id: string; + /** + * Run URL + * @description Absolute or relative URL to the storage operation run status view. + */ + run_url: string; + /** + * Skipped Count + * @description Skipped datasets count. + * @default 0 + */ + skipped_count: number; + /** + * State + * @description The current state of the storage operation run when this notification was generated. + */ + state: components["schemas"]["StorageOperationNotificationState"]; + /** + * Subject + * @description The subject of the notification. + */ + subject: string; + /** + * Succeeded Count + * @description Succeeded datasets count. + * @default 0 + */ + succeeded_count: number; + /** + * Total Count + * @description Total datasets in the run. + */ + total_count: number; + }; + /** + * StorageOperationNotificationState + * @enum {string} + */ + StorageOperationNotificationState: "started" | "completed" | "failed"; /** StorageOperationPreviewItemResult */ StorageOperationPreviewItemResult: { - item: components["schemas"]["EncodedHistoryContentItem"]; + /** + * Dataset Id + * @example 0123456789ABCDEF + */ + dataset_id: string; /** Message */ message?: string | null; /** Reason Code */ @@ -22705,6 +22784,16 @@ export interface components { StorageOperationRunItemState: "pending" | "running" | "succeeded" | "failed" | "skipped"; /** StorageOperationRunItemStatus */ StorageOperationRunItemStatus: { + /** Attempt Count */ + attempt_count: number; + /** Bytes Processed */ + bytes_processed: number; + /** + * Create Time + * Format: date-time + * @description The time and date this item was created. + */ + create_time: string; /** * Dataset Id * @example 0123456789ABCDEF @@ -22715,6 +22804,12 @@ export interface components { /** Reason Code */ reason_code?: string | null; state: components["schemas"]["StorageOperationRunItemState"]; + /** + * Update Time + * Format: date-time + * @description The last time and date this item was updated. + */ + update_time: string; }; /** StorageOperationRunResponse */ StorageOperationRunResponse: { @@ -22729,6 +22824,12 @@ export interface components { StorageOperationRunState: "pending" | "running" | "completed" | "failed"; /** StorageOperationRunSummary */ StorageOperationRunSummary: { + /** + * Create Time + * Format: date-time + * @description The time and date this item was created. + */ + create_time: string; /** Failed Count */ failed_count: number; mode: components["schemas"]["StorageOperationMode"]; @@ -22748,6 +22849,12 @@ export interface components { task_id?: string | null; /** Total Count */ total_count: number; + /** + * Update Time + * Format: date-time + * @description The last time and date this item was updated. + */ + update_time: string; }; /** StorageOperationSelectionCounts */ StorageOperationSelectionCounts: { @@ -24783,6 +24890,13 @@ export interface components { * "push": true * }, * "enabled": true + * }, + * "storage_operation": { + * "channels": { + * "email": true, + * "push": true + * }, + * "enabled": true * } * } * } @@ -25186,6 +25300,13 @@ export interface components { * "push": true * }, * "enabled": true + * }, + * "storage_operation": { + * "channels": { + * "email": true, + * "push": true + * }, + * "enabled": true * } * } * } @@ -25215,7 +25336,8 @@ export interface components { */ content: | components["schemas"]["MessageNotificationContent"] - | components["schemas"]["NewSharedItemNotificationContent"]; + | components["schemas"]["NewSharedItemNotificationContent"] + | components["schemas"]["StorageOperationNotificationContent"]; /** * Create time * Format: date-time From 22421521bcdfa8575a523064e0d63381a419419f Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:35:29 +0200 Subject: [PATCH 008/120] Fix getStorageOperationRunStatus to always return a value --- client/src/api/histories.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client/src/api/histories.ts b/client/src/api/histories.ts index f411191b5ec..10a4f1a8cca 100644 --- a/client/src/api/histories.ts +++ b/client/src/api/histories.ts @@ -315,7 +315,7 @@ export type StorageOperationRunResponse = components["schemas"]["StorageOperatio export async function getStorageOperationRunStatus( history: HistoryReference, runId: string, -): Promise { +): Promise { const { data, error } = await GalaxyApi().GET("/api/histories/{history_id}/contents/bulk/storage/runs/{run_id}", { params: { path: { history_id: history.id, run_id: runId }, From 80992fa48377cfd2eb024ffc8add4e3f45e8a70e Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:35:29 +0200 Subject: [PATCH 009/120] Support storage operation run tracking in API and schema --- lib/galaxy/schema/schema.py | 8 ++++++- .../galaxy/services/history_contents.py | 21 +++++++++++-------- 2 files changed, 19 insertions(+), 10 deletions(-) diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py index eedf4331a5b..7b2dc5d8215 100644 --- a/lib/galaxy/schema/schema.py +++ b/lib/galaxy/schema/schema.py @@ -1439,7 +1439,7 @@ class StorageOperationPreviewRequest(Model): class StorageOperationPreviewItemResult(Model): - item: EncodedHistoryContentItem + dataset_id: EncodedDatabaseIdField state: StorageOperationEligibilityState reason_code: Optional[str] = None message: Optional[str] = None @@ -1486,6 +1486,8 @@ class StorageOperationRunSummary(Model): state: StorageOperationRunState mode: StorageOperationMode target_object_store_id: str + create_time: datetime = CreateTimeField + update_time: datetime = UpdateTimeField total_count: int succeeded_count: int failed_count: int @@ -1498,6 +1500,10 @@ class StorageOperationRunItemStatus(Model): state: StorageOperationRunItemState reason_code: Optional[str] = None message: Optional[str] = None + attempt_count: int + bytes_processed: int + create_time: datetime = CreateTimeField + update_time: datetime = UpdateTimeField class StorageOperationExecuteResponse(Model): diff --git a/lib/galaxy/webapps/galaxy/services/history_contents.py b/lib/galaxy/webapps/galaxy/services/history_contents.py index 28ed716825a..2c6104a7994 100644 --- a/lib/galaxy/webapps/galaxy/services/history_contents.py +++ b/lib/galaxy/webapps/galaxy/services/history_contents.py @@ -799,15 +799,14 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto for dataset_id, dataset in unique_datasets.items(): reason = self._validate_relocate_dataset(trans, dataset, payload.target_object_store_id) - item = EncodedHistoryContentItem( - id=self.encode_id(dataset_id), - history_content_type=HistoryContentType.dataset, - ) if reason is None: eligible_count += 1 eligible_dataset_ids.append(dataset_id) eligibility_items.append( - StorageOperationPreviewItemResult(item=item, state=StorageOperationEligibilityState.eligible) + StorageOperationPreviewItemResult( + dataset_id=dataset_id, + state=StorageOperationEligibilityState.eligible, + ) ) if quota_source_map: old_label = quota_source_map.get_quota_source_label(dataset.object_store_id) or "default" @@ -821,7 +820,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto reason_code, message = reason eligibility_items.append( StorageOperationPreviewItemResult( - item=item, + dataset_id=dataset_id, state=StorageOperationEligibilityState.ineligible, reason_code=reason_code, message=message, @@ -832,7 +831,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto snapshot = DatasetStorageOperationSnapshot( history_id=history.id, user_id=user.id, - mode=payload.mode.value, + mode=payload.mode, target_object_store_id=payload.target_object_store_id, resolved_dataset_ids=list(unique_datasets.keys()), eligible_dataset_ids=eligible_dataset_ids, @@ -846,7 +845,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto warnings.append("Selection was query-based. Execute uses a fixed snapshot to avoid selection drift.") return StorageOperationPreviewResponse( - snapshot_id=self.encode_id(snapshot.id), + snapshot_id=snapshot.id, selection_counts=StorageOperationSelectionCounts( selected_items_count=selected_items_count, expanded_leaf_count=expanded_leaf_count, @@ -995,10 +994,14 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto ) items = [ StorageOperationRunItemStatus( - dataset_id=self.encode_id(run_item.dataset_id), + dataset_id=run_item.dataset_id, state=StorageOperationRunItemState(run_item.state), reason_code=run_item.reason_code, message=run_item.message, + attempt_count=run_item.attempt_count, + bytes_processed=run_item.bytes_processed, + create_time=run_item.create_time, + update_time=run_item.update_time, ) for run_item in run_items ] From 84f42caa39349ae30536ff5d9091527670f82fdf Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:35:30 +0200 Subject: [PATCH 010/120] Add Pinia store and utility for storage operation runs --- client/src/stores/storageOperationsStore.ts | 100 ++++++++++++++++++++ client/src/utils/storageOperations.ts | 97 +++++++++++++++++++ 2 files changed, 197 insertions(+) create mode 100644 client/src/stores/storageOperationsStore.ts create mode 100644 client/src/utils/storageOperations.ts diff --git a/client/src/stores/storageOperationsStore.ts b/client/src/stores/storageOperationsStore.ts new file mode 100644 index 00000000000..9e3e6ae673c --- /dev/null +++ b/client/src/stores/storageOperationsStore.ts @@ -0,0 +1,100 @@ +import { defineStore } from "pinia"; + +import { useUserLocalStorage } from "@/composables/userLocalStorage"; +import { isTerminalRunState, type StorageOperationRunSummary, type TrackedStorageRun } from "@/utils/storageOperations"; + +const ACTIVE_RUN_TTL_MS = 24 * 60 * 60 * 1000; + +export type StorageRun = TrackedStorageRun; + +export type RunStatusUpdate = { + state: StorageOperationRunSummary["state"]; + succeeded_count?: StorageOperationRunSummary["succeeded_count"]; + failed_count?: StorageOperationRunSummary["failed_count"]; + skipped_count?: StorageOperationRunSummary["skipped_count"]; + total_count?: StorageOperationRunSummary["total_count"]; + update_time?: StorageOperationRunSummary["update_time"]; +}; + +export const useStorageOperationsStore = defineStore("storageOperationsStore", () => { + const runs = useUserLocalStorage("storage-operations-active-runs", []); + + function pruneExpiredRuns() { + const now = Date.now(); + runs.value = runs.value.filter((run) => { + const startedAtRaw = run.create_time || run.update_time; + if (!startedAtRaw) { + return true; + } + + const startedAt = Date.parse(startedAtRaw); + if (!Number.isFinite(startedAt)) { + return true; + } + + return now - startedAt <= ACTIVE_RUN_TTL_MS; + }); + } + + function startRun(run: StorageRun): void { + pruneExpiredRuns(); + // Remove any existing run with the same ID + runs.value = runs.value.filter((r) => r.run_id !== run.run_id); + runs.value.push(run); + } + + function updateRunStatus(runId: string, update: RunStatusUpdate): void { + pruneExpiredRuns(); + runs.value = runs.value.map((run) => { + if (run.run_id !== runId) { + return run; + } + + return { + ...run, + ...update, + }; + }); + } + + function clearRun(runId: string): void { + runs.value = runs.value.filter((run) => run.run_id !== runId); + } + + function getRuns(historyId: string): StorageRun[] { + pruneExpiredRuns(); + return runs.value.filter((run) => run.historyId === historyId); + } + + function getActiveRuns(historyId: string): StorageRun[] { + pruneExpiredRuns(); + return runs.value.filter((run) => run.historyId === historyId && !isTerminalRunState(run.state)); + } + + function getCompletedRuns(historyId: string): StorageRun[] { + pruneExpiredRuns(); + return runs.value.filter((run) => run.historyId === historyId && isTerminalRunState(run.state)); + } + + function getActiveRunCount(historyId: string): number { + return getActiveRuns(historyId).length; + } + + function getCompletedRunCount(historyId: string): number { + return getCompletedRuns(historyId).length; + } + + pruneExpiredRuns(); + + return { + runs, + startRun, + updateRunStatus, + getRuns, + getActiveRuns, + getCompletedRuns, + getActiveRunCount, + getCompletedRunCount, + clearRun, + }; +}); diff --git a/client/src/utils/storageOperations.ts b/client/src/utils/storageOperations.ts new file mode 100644 index 00000000000..b2819909a85 --- /dev/null +++ b/client/src/utils/storageOperations.ts @@ -0,0 +1,97 @@ +import type { components } from "@/api/schema"; + +export type StorageOperationRunState = components["schemas"]["StorageOperationRunState"]; +export type StorageOperationRunSummary = components["schemas"]["StorageOperationRunSummary"]; +export type TrackedStorageRun = StorageOperationRunSummary & { + historyId: string; + runUrl: string; +}; + +export type IneligibleReasonCode = + | "dataset_not_found" + | "invalid_target_object_store" + | "missing_source_object_store" + | "already_in_target" + | "cross_device_relocate_not_allowed" + | "insufficient_permissions" + | "execution_error"; + +export interface IneligibleReasonDescription { + label: string; + description: string; +} + +/** + * Mapping of ineligibility reason codes to user-friendly labels and descriptions. + * These correspond to the reason codes defined in the backend (galaxy/celery/tasks.py). + */ +const INELIGIBLE_REASON_DESCRIPTIONS: Record = { + dataset_not_found: { + label: "Dataset not found", + description: "The dataset could not be found at the time of the operation.", + }, + invalid_target_object_store: { + label: "Invalid storage location", + description: "The selected storage location is not available or is invalid.", + }, + missing_source_object_store: { + label: "Missing source location", + description: "The dataset does not have a source storage location defined.", + }, + already_in_target: { + label: "Already in target location", + description: "The dataset is already stored in the selected location.", + }, + cross_device_relocate_not_allowed: { + label: "Cannot relocate to different device", + description: "The dataset cannot be moved because the source and target locations are on different devices.", + }, + insufficient_permissions: { + label: "Insufficient permissions", + description: "You do not have permission to move this dataset to a different storage location.", + }, + execution_error: { + label: "Execution error", + description: "An error occurred while attempting to move the dataset.", + }, +}; + +/** + * Get a user-friendly description for an ineligibility reason code. + * @param reasonCode - The technical reason code from the backend + * @returns An object with a human-readable label and description + */ +export function getIneligibleReasonDescription(reasonCode: string): IneligibleReasonDescription { + return ( + INELIGIBLE_REASON_DESCRIPTIONS[reasonCode as IneligibleReasonCode] || { + label: "Unknown reason", + description: `Reason code: ${reasonCode}`, + } + ); +} + +/** + * Check if a storage operation run state is a terminal state (no longer running). + * @param state - The run state (e.g., "pending", "running", "completed", "failed") + * @returns true if the state is terminal (completed or failed) + */ +export function isTerminalRunState(state?: StorageOperationRunState): boolean { + return state === "completed" || state === "failed"; +} + +export function toTrackedStorageRun(historyId: string, run: StorageOperationRunSummary): TrackedStorageRun { + const createTime = run.create_time || new Date().toISOString(); + const updateTime = run.update_time || createTime; + + return { + ...run, + create_time: createTime, + update_time: updateTime, + historyId, + runUrl: `/histories/${historyId}/storage/runs/${run.run_id}`, + }; +} + +export function getCompletedAtForRun(run: StorageOperationRunSummary): string | undefined { + return isTerminalRunState(run.state) ? run.update_time : undefined; +} From 771b5de05d17b632f748730c12dd7bf2b2e321f3 Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:35:30 +0200 Subject: [PATCH 011/120] Add composables for polling storage operation run state --- .../src/composables/useStorageRunWatcher.ts | 122 +++++++++++++++++- 1 file changed, 116 insertions(+), 6 deletions(-) diff --git a/client/src/composables/useStorageRunWatcher.ts b/client/src/composables/useStorageRunWatcher.ts index 66e649c092d..47f732b9a1a 100644 --- a/client/src/composables/useStorageRunWatcher.ts +++ b/client/src/composables/useStorageRunWatcher.ts @@ -1,19 +1,38 @@ -import { computed, ref } from "vue"; +import { computed, readonly, ref } from "vue"; import { getStorageOperationRunStatus, type HistoryReference, type StorageOperationRunResponse } from "@/api/histories"; import { useResourceWatcher } from "@/composables/resourceWatcher"; +import { useStorageOperationsStore } from "@/stores/storageOperationsStore"; +import { isTerminalRunState } from "@/utils/storageOperations"; -const STORAGE_RUN_TERMINAL_STATES = new Set(["completed", "failed"]); const STORAGE_RUN_POLLING_INTERVAL = 10000; // 10 seconds +export interface StorageRunWatcherOptions { + pollInterval?: number; + enableBackgroundPolling?: boolean; +} + +const DEFAULT_RUN_WATCHER_OPTIONS: Required = { + pollInterval: STORAGE_RUN_POLLING_INTERVAL, + enableBackgroundPolling: false, +}; + +function resolveWatcherOptions(options?: StorageRunWatcherOptions): Required { + return { + ...DEFAULT_RUN_WATCHER_OPTIONS, + ...options, + }; +} + /** * Poll storage-operation run status until it reaches a terminal state. */ -export function useStorageRunWatcher(history: HistoryReference, runId: string) { +export function useStorageRunWatcher(history: HistoryReference, runId: string, options?: StorageRunWatcherOptions) { + const { pollInterval, enableBackgroundPolling } = resolveWatcherOptions(options); const runStatus = ref(null); const isTerminal = computed(() => { const state = runStatus.value?.run.state; - return state !== undefined && STORAGE_RUN_TERMINAL_STATES.has(state); + return state !== undefined && isTerminalRunState(state); }); async function fetchStatus() { @@ -25,9 +44,100 @@ export function useStorageRunWatcher(history: HistoryReference, runId: string) { } const { startWatchingResource: startPolling, stopWatchingResource: stopPolling } = useResourceWatcher(fetchStatus, { - shortPollingInterval: STORAGE_RUN_POLLING_INTERVAL, - enableBackgroundPolling: false, + shortPollingInterval: pollInterval, + enableBackgroundPolling, }); return { runStatus, isTerminal, startPolling, stopPolling }; } + +/** + * Poll status for all tracked storage-operation runs in a history. + */ +export function useStorageHistoryRunsWatcher(historyId: string, options?: StorageRunWatcherOptions) { + const { pollInterval, enableBackgroundPolling } = resolveWatcherOptions(options); + const storageOperationsStore = useStorageOperationsStore(); + const runSummariesByRunId = ref>({}); + + const historyReference: HistoryReference = { + id: historyId, + model_class: "History", + }; + + async function watchStorageOperationRuns() { + const runs = storageOperationsStore.getRuns(historyId); + if (!runs.length) { + runSummariesByRunId.value = {}; + return; + } + + const previous = runSummariesByRunId.value; + + const runsToFetch = runs.filter((run) => !isTerminalRunState(run.state)); + + if (!runsToFetch.length) { + const nextFromPrevious: Record = {}; + runs.forEach((run) => { + const summary = previous[run.run_id]; + if (summary) { + nextFromPrevious[run.run_id] = summary; + } + }); + runSummariesByRunId.value = nextFromPrevious; + return; + } + + const settled = await Promise.allSettled( + runsToFetch.map(async (run) => { + const status = await getStorageOperationRunStatus(historyReference, run.run_id); + return { run_id: run.run_id, summary: status?.run }; + }), + ); + + const next: Record = {}; + + runs.forEach((run) => { + const existingSummary = previous[run.run_id]; + if (existingSummary) { + next[run.run_id] = existingSummary; + } + }); + + settled.forEach((result, index) => { + const runId = runsToFetch[index]?.run_id; + if (!runId) { + return; + } + + if (result.status === "fulfilled" && result.value.summary) { + const summary = result.value.summary; + next[runId] = summary; + storageOperationsStore.updateRunStatus(runId, { + state: summary.state, + total_count: summary.total_count, + succeeded_count: summary.succeeded_count, + failed_count: summary.failed_count, + skipped_count: summary.skipped_count, + update_time: summary.update_time, + }); + } + }); + + runSummariesByRunId.value = next; + } + + const { startWatchingResource, stopWatchingResource, isWatchingResource } = useResourceWatcher( + watchStorageOperationRuns, + { + shortPollingInterval: pollInterval, + enableBackgroundPolling, + }, + ); + + return { + isPolling: readonly(isWatchingResource), + startPolling: startWatchingResource, + stopPolling: stopWatchingResource, + runSummariesByRunId: readonly(runSummariesByRunId), + }; +} From 49993f6daeb4ecc0a794b2f4e970875e58c51c71 Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:35:30 +0200 Subject: [PATCH 012/120] Implement UI for viewing and managing storage operation runs --- .../components/Common/DatasetPopoverLink.vue | 71 ++++ .../StorageOperationHistoryView.vue | 396 ++++++++++++++++++ .../StorageOperationRunView.vue | 312 ++++++++++++++ 3 files changed, 779 insertions(+) create mode 100644 client/src/components/Common/DatasetPopoverLink.vue create mode 100644 client/src/components/History/StorageOperations/StorageOperationHistoryView.vue create mode 100644 client/src/components/History/StorageOperations/StorageOperationRunView.vue diff --git a/client/src/components/Common/DatasetPopoverLink.vue b/client/src/components/Common/DatasetPopoverLink.vue new file mode 100644 index 00000000000..7c0b28dbfbb --- /dev/null +++ b/client/src/components/Common/DatasetPopoverLink.vue @@ -0,0 +1,71 @@ + + + + + diff --git a/client/src/components/History/StorageOperations/StorageOperationHistoryView.vue b/client/src/components/History/StorageOperations/StorageOperationHistoryView.vue new file mode 100644 index 00000000000..ba3956dd159 --- /dev/null +++ b/client/src/components/History/StorageOperations/StorageOperationHistoryView.vue @@ -0,0 +1,396 @@ + + + + + diff --git a/client/src/components/History/StorageOperations/StorageOperationRunView.vue b/client/src/components/History/StorageOperations/StorageOperationRunView.vue new file mode 100644 index 00000000000..47bf2d8c9ee --- /dev/null +++ b/client/src/components/History/StorageOperations/StorageOperationRunView.vue @@ -0,0 +1,312 @@ + + + + + From 691ec32c28b31ed7f7eb55c3347357e57bc252a7 Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Thu, 23 Apr 2026 15:35:31 +0200 Subject: [PATCH 013/120] Integrate storage operation run indicators into history panel --- .../HistoryOperations/HistoryOperations.vue | 26 +++++--- .../HistoryStorageOperationsIndicator.vue | 62 +++++++++++++++++++ .../History/CurrentHistory/HistoryPanel.vue | 7 +++ 3 files changed, 87 insertions(+), 8 deletions(-) create mode 100644 client/src/components/History/CurrentHistory/HistoryOperations/HistoryStorageOperationsIndicator.vue diff --git a/client/src/components/History/CurrentHistory/HistoryOperations/HistoryOperations.vue b/client/src/components/History/CurrentHistory/HistoryOperations/HistoryOperations.vue index 8622468e4a8..dc5a9593730 100644 --- a/client/src/components/History/CurrentHistory/HistoryOperations/HistoryOperations.vue +++ b/client/src/components/History/CurrentHistory/HistoryOperations/HistoryOperations.vue @@ -3,11 +3,13 @@ import { faCheckSquare, faCompress } from "@fortawesome/free-solid-svg-icons"; import { FontAwesomeIcon } from "@fortawesome/vue-fontawesome"; import type { HistorySummaryExtended } from "@/api"; +import type { ActiveStorageRun } from "@/stores/storageOperationsStore"; import localize from "@/utils/localization"; import GButton from "@/components/BaseComponents/GButton.vue"; import GButtonGroup from "@/components/BaseComponents/GButtonGroup.vue"; import DefaultOperations from "@/components/History/CurrentHistory/HistoryOperations/DefaultOperations.vue"; +import HistoryStorageOperationsIndicator from "@/components/History/CurrentHistory/HistoryOperations/HistoryStorageOperationsIndicator.vue"; interface Props { history: HistorySummaryExtended; @@ -16,6 +18,7 @@ interface Props { expandedCount: number; showSelection: boolean; isMultiViewItem: boolean; + activeStorageRuns: ActiveStorageRun[]; } const props = defineProps(); @@ -61,15 +64,22 @@ function onUpdateOperationStatus(updateTime: number) { - - - +
+ + + - + + + +