Extend on disk checks to running, queued and error states

Fixes
https://sentry.galaxyproject.org/share/issue/c1c4c9b0ab324b4ea4f98c3b8c402d21/:
```
Message
Uncaught Exception
Stack Trace

Newest

FileNotFoundError: [Errno 2] No such file or directory: ''
  File "galaxy/web/framework/middleware/error.py", line 167, in __call__
    app_iter = self.application(environ, sr_checker)
  File "galaxy/web/framework/middleware/statsd.py", line 29, in __call__
    req = self.application(environ, start_response)
  File "/cvmfs/main.galaxyproject.org/venv/lib/python3.11/site-packages/paste/httpexceptions.py", line 635, in __call__
    return self.application(environ, start_response)
  File "galaxy/web/framework/base.py", line 176, in __call__
    return self.handle_request(request_id, path_info, environ, start_response)
  File "galaxy/web/framework/base.py", line 271, in handle_request
    body = method(trans, **kwargs)
  File "galaxy/webapps/galaxy/controllers/dataset.py", line 155, in display
    display_data, headers = data.datatype.display_data(
  File "galaxy/datatypes/tabular.py", line 226, in display_data
    chunk=self.get_chunk(trans, dataset, 0),
  File "galaxy/datatypes/tabular.py", line 149, in get_chunk
    ck_data, last_read = self._read_chunk(trans, dataset, offset, ck_size)
  File "galaxy/datatypes/tabular.py", line 159, in _read_chunk
    with compression_utils.get_fileobj(dataset.get_file_name()) as f:
  File "galaxy/util/compression_utils.py", line 79, in get_fileobj
    return get_fileobj_raw(filename, mode, compressed_formats)[1]
  File "galaxy/util/compression_utils.py", line 139, in get_fileobj_raw
    return compressed_format, open(filename, mode, encoding="utf-8")
```

lib/galaxy/managers/datasets.py

@@ -497,8 +497,8 @@ def ensure_dataset_on_disk(self, trans, dataset):
             raise exceptions.ItemDeletionException("The dataset you are attempting to view has been deleted.")
         elif dataset.state == Dataset.states.UPLOAD:
             raise exceptions.Conflict("Please wait until this dataset finishes uploading before attempting to view it.")
-        elif dataset.state == Dataset.states.NEW:
-            raise exceptions.Conflict("The dataset you are attempting to view is new and has no data.")
+        elif dataset.state in (Dataset.states.NEW, Dataset.states.QUEUED):
+            raise exceptions.Conflict(f"The dataset you are attempting to view is {dataset.state} and has no data.")
         elif dataset.state == Dataset.states.DISCARDED:
             raise exceptions.ItemDeletionException("The dataset you are attempting to view has been discarded.")
         elif dataset.state == Dataset.states.DEFERRED:
@@ -509,6 +509,14 @@ def ensure_dataset_on_disk(self, trans, dataset):
             raise exceptions.Conflict(
                 "The dataset you are attempting to view is in paused state. One of the inputs for the job that creates this dataset has failed."
             )
+        elif dataset.state == Dataset.states.RUNNING:
+            if not self.app.object_store.exists(dataset.dataset):
+                raise exceptions.Conflict(
+                    "The dataset you are attempting to view is still being created and has no data yet."
+                )
+        elif dataset.state == Dataset.states.ERROR:
+            if not self.app.object_store.exists(dataset.dataset):
+                raise exceptions.RequestParameterInvalidException("The dataset is in error and has no data.")
 
     def ensure_can_change_datatype(self, dataset: model.DatasetInstance, raiseException: bool = True) -> bool:
         if not dataset.datatype.is_datatype_change_allowed():

lib/galaxy/webapps/galaxy/controllers/dataset.py

@@ -25,7 +25,6 @@
     HDAManager,
 )
 from galaxy.managers.histories import HistoryManager
-from galaxy.model import Dataset
 from galaxy.model.base import transaction
 from galaxy.model.item_attrs import (
     UsesAnnotations,
@@ -108,33 +107,14 @@ def _check_dataset(self, trans, hda_id):
             raise web.httpexceptions.HTTPNotFound(f"Invalid reference dataset id: {str(hda_id)}.")
         if not self._can_access_dataset(trans, data):
             return trans.show_error_message("You are not allowed to access this dataset")
-        if data.purged or data.dataset.purged:
-            return trans.show_error_message("The dataset you are attempting to view has been purged.")
-        elif data.deleted and not (trans.user_is_admin or (data.history and trans.get_user() == data.user)):
-            return trans.show_error_message("The dataset you are attempting to view has been deleted.")
-        elif data.state == Dataset.states.UPLOAD:
-            return trans.show_error_message(
-                "Please wait until this dataset finishes uploading before attempting to view it."
-            )
-        elif data.state == Dataset.states.DISCARDED:
-            return trans.show_error_message("The dataset you are attempting to view has been discarded.")
-        elif data.state == Dataset.states.DEFERRED:
-            return trans.show_error_message(
-                "The dataset you are attempting to view has deferred data. You can only use this dataset as input for jobs."
-            )
-        elif data.state == Dataset.states.PAUSED:
-            return trans.show_error_message(
-                "The dataset you are attempting to view is in paused state. One of the inputs for the job that creates this dataset has failed."
-            )
+        self.app.hda_manager.ensure_dataset_on_disk(trans, data)
         return data
 
     @web.expose
     def display(
         self, trans, dataset_id=None, preview=False, filename=None, to_ext=None, offset=None, ck_size=None, **kwd
     ):
         data = self._check_dataset(trans, dataset_id)
-        if not isinstance(data, trans.app.model.DatasetInstance):
-            return data
         if "hdca" in kwd:
             raise RequestParameterInvalidException("Invalid request parameter 'hdca' encountered.")
         if hdca_id := kwd.get("hdca_id", None):
@@ -478,8 +458,6 @@ def imp(self, trans, dataset_id=None, **kwd):
     def display_by_username_and_slug(self, trans, username, slug, filename=None, preview=True, **kwargs):
         """Display dataset by username and slug; because datasets do not yet have slugs, the slug is the dataset's id."""
         dataset = self._check_dataset(trans, slug)
-        if not isinstance(dataset, trans.app.model.DatasetInstance):
-            return dataset
         # Filename used for composite types.
         if filename:
             return self.display(trans, dataset_id=slug, filename=filename)