Handle locked volumes when attaching

Author: r4victor

src/dstack/_internal/server/background/scheduled_tasks/submitted_jobs.py

@@ -1042,18 +1042,23 @@ async def _attach_volumes(
                     )
                     job_runtime_data.volume_names.append(volume.name)
                     break  # attach next mount point
-            except (ServerClientError, BackendError) as e:
-                logger.warning("%s: failed to attached volume: %s", fmt(job_model), repr(e))
+            except ServerClientError as e:
+                logger.info("%s: failed to attach volume: %s", fmt(job_model), repr(e))
                 job_model.termination_reason = JobTerminationReason.VOLUME_ERROR
-                job_model.termination_reason_message = "Failed to attach volume"
+                job_model.termination_reason_message = f"Failed to attach volume: {e.msg}"
+                switch_job_status(session, job_model, JobStatus.TERMINATING)
+            except BackendError as e:
+                logger.warning("%s: failed to attach volume: %s", fmt(job_model), repr(e))
+                job_model.termination_reason = JobTerminationReason.VOLUME_ERROR
+                job_model.termination_reason_message = f"Failed to attach volume: {str(e)}"
                 switch_job_status(session, job_model, JobStatus.TERMINATING)
             except Exception:
                 logger.exception(
                     "%s: got exception when attaching volume",
                     fmt(job_model),
                 )
                 job_model.termination_reason = JobTerminationReason.VOLUME_ERROR
-                job_model.termination_reason_message = "Failed to attach volume"
+                job_model.termination_reason_message = "Failed to attach volume: unexpected error"
                 switch_job_status(session, job_model, JobStatus.TERMINATING)
             finally:
                 job_model.job_runtime_data = job_runtime_data.json()
@@ -1075,6 +1080,8 @@ async def _attach_volume(
         raise ServerClientError("Cannot attach a deleted volume")
     if volume_model.to_be_deleted:
         raise ServerClientError("Cannot attach a volume marked for deletion")
+    if volume_model.lock_expires_at is not None:
+        raise ServerClientError("Cannot attach a volume locked for processing")
     attachment_data = await common_utils.run_async(
         compute.attach_volume,
         volume=volume,

src/dstack/_internal/server/services/jobs/__init__.py

@@ -720,6 +720,10 @@ async def get_job_configured_volume_models(
             )
             if volume_model is None:
                 raise ResourceNotExistsError(f"Volume {mount_point.name} not found")
+            if volume_model.to_be_deleted:
+                raise ServerClientError(
+                    f"Volume {mount_point.name} is marked for deletion and cannot be attached"
+                )
             mount_point_volume_models.append(volume_model)
         volume_models.append(mount_point_volume_models)
     return volume_models
@@ -729,7 +733,7 @@ def check_can_attach_job_volumes(volumes: List[List[Volume]]):
     """
     Performs basic checks if volumes can be attached.
     This is useful to show error ASAP (when user submits the run).
-    If the attachment is to fail anyway, the error will be handled when proccessing submitted jobs.
+    If the attachment is to fail anyway, the error will be handled when processing submitted jobs.
     """
     if len(volumes) == 0:
         return