Add validation for external volumes with auto_cleanup_duration

Author: haydnli-shopify

src/dstack/_internal/server/services/volumes.py

@@ -400,6 +400,23 @@ def _validate_volume_configuration(configuration: VolumeConfiguration):
         )
     if configuration.name is not None:
         validate_dstack_resource_name(configuration.name)
+    # External volumes (with volume_id) cannot have auto_cleanup_duration
+    if configuration.volume_id is not None and configuration.auto_cleanup_duration is not None:
+        if (
+            isinstance(configuration.auto_cleanup_duration, int)
+            and configuration.auto_cleanup_duration >= 0
+        ):
+            raise ServerClientError(
+                "External volumes (with volume_id) do not support auto_cleanup_duration. "
+                "Auto-cleanup only works for volumes created and managed by dstack."
+            )
+        elif isinstance(
+            configuration.auto_cleanup_duration, str
+        ) and configuration.auto_cleanup_duration not in ("off", "-1"):
+            raise ServerClientError(
+                "External volumes (with volume_id) do not support auto_cleanup_duration. "
+                "Auto-cleanup only works for volumes created and managed by dstack."
+            )
 
 
 async def _delete_volume(session: AsyncSession, project: ProjectModel, volume_model: VolumeModel):

src/tests/_internal/server/services/test_volumes.py

@@ -3,11 +3,78 @@
 import pytest
 from freezegun import freeze_time
 
-from dstack._internal.core.models.volumes import VolumeStatus
-from dstack._internal.server.services.volumes import _get_volume_cost
+from dstack._internal.core.errors import ServerClientError
+from dstack._internal.core.models.backends.base import BackendType
+from dstack._internal.core.models.volumes import VolumeConfiguration, VolumeStatus
+from dstack._internal.server.services.volumes import (
+    _get_volume_cost,
+    _validate_volume_configuration,
+)
 from dstack._internal.server.testing.common import get_volume, get_volume_provisioning_data
 
 
+class TestValidateVolumeConfiguration:
+    def test_external_volume_with_auto_cleanup_duration_raises_error(self):
+        """External volumes (with volume_id) should not allow auto_cleanup_duration"""
+        config = VolumeConfiguration(
+            backend=BackendType.AWS,
+            region="us-east-1",
+            volume_id="vol-123456",
+            auto_cleanup_duration="1h",
+        )
+        with pytest.raises(
+            ServerClientError, match="External volumes.*do not support auto_cleanup_duration"
+        ):
+            _validate_volume_configuration(config)
+
+    def test_external_volume_with_auto_cleanup_duration_int_raises_error(self):
+        """External volumes with integer auto_cleanup_duration should also raise error"""
+        config = VolumeConfiguration(
+            backend=BackendType.AWS,
+            region="us-east-1",
+            volume_id="vol-123456",
+            auto_cleanup_duration=3600,
+        )
+        with pytest.raises(
+            ServerClientError, match="External volumes.*do not support auto_cleanup_duration"
+        ):
+            _validate_volume_configuration(config)
+
+    def test_external_volume_with_auto_cleanup_disabled_succeeds(self):
+        """External volumes with auto_cleanup_duration='off' or -1 should be allowed"""
+        config1 = VolumeConfiguration(
+            backend=BackendType.AWS,
+            region="us-east-1",
+            volume_id="vol-123456",
+            auto_cleanup_duration="off",
+        )
+        config2 = VolumeConfiguration(
+            backend=BackendType.AWS,
+            region="us-east-1",
+            volume_id="vol-123456",
+            auto_cleanup_duration=-1,
+        )
+        # Should not raise any errors
+        _validate_volume_configuration(config1)
+        _validate_volume_configuration(config2)
+
+    def test_external_volume_without_auto_cleanup_succeeds(self):
+        """External volumes without auto_cleanup_duration should be allowed"""
+        config = VolumeConfiguration(
+            backend=BackendType.AWS, region="us-east-1", volume_id="vol-123456"
+        )
+        # Should not raise any errors
+        _validate_volume_configuration(config)
+
+    def test_new_volume_with_auto_cleanup_duration_succeeds(self):
+        """New volumes (without volume_id) with auto_cleanup_duration should be allowed"""
+        config = VolumeConfiguration(
+            backend=BackendType.AWS, region="us-east-1", size=100, auto_cleanup_duration="1h"
+        )
+        # Should not raise any errors
+        _validate_volume_configuration(config)
+
+
 class TestGetVolumeCost:
     def test_returns_0_when_no_provisioning_data(self):
         volume = get_volume(provisioning_data=None)