Repository: skypilot-org/skypilot
Pull request: [Storage] Fix `sky storage delete` for externally deleted buckets
Closes #1322.

If a bucket is deleted externally, then store.delete() now does a no-op (instead of creating and deleting like before). The logging has been updated. 

Now:
```
(base) ➜  repro git:(storage_deletionfix) ✗ SKYPILOT_DEBUG=0 sky storage delete -a
Deleting all storage objects.
# Shows spinner Deleting S3 bucket romilmybucket.
I 04-16 22:47:19 storage.py:1029] S3 bucket romilmybucket may have been deleted externally. Removing from local state.
```

Tested (run the relevant ones):

- [x] [Snippet](https://github.com/skypilot-org/skypilot/issues/1322#issuecomment-1510541254) from #1322
- [x] Storage smoke tests: `pytest tests/test_smoke.py::TestStorageWithCredentials` and with `--cloudflare`.
- [x] Backward compatibility testing by creating buckets in master, upgrading, then trying deletion and launching with existing buckets.


Problem statement:
[Storage] Confusing logging in `sky storage delete`
When I am trying to delete a storage object with `sky storage delete skypilot-filemounts-*`, the logging is quite confusing showing the bucket is detected, then the bucket is created, and finally the bucket is deleted.
```
I 10-29 00:50:56 storage.py:402] Detected existing storage object, loading Storage: skypilot-filemounts-folder-zhwu-301024cb-1
I 10-29 00:50:58 storage.py:1228] Created GCS bucket skypilot-filemounts-folder-zhwu-301024cb-1 in US-CENTRAL1 with storage class STANDARD
I 10-29 00:51:01 storage.py:1092] Deleted GCS bucket skypilot-filemounts-folder-zhwu-301024cb-1.
```

Probably, it is because the bucket has been deleted on the cloud console, but it is still in my storage table. Seems we will create a `sky.Storage` object (the creation will create a bucket during initialization) before calling `storage.delete()`

Hints from issue discussion:
I believe this is closed by #1323.

New output:
```
(base) romilb@romilbx1yoga:~$ sky storage delete -a
Deleting all storage objects.
I 10-31 16:18:09 storage.py:1186] Deleted GCS bucket sky-romilb-gcsstore.
I 10-31 16:18:25 storage.py:878] Deleted S3 bucket sky-romilb-s3store.
```
@romilbhardwaj On latest master  7602b254f2aa6d8a0 I see

```
» sky storage delete 'skypilot-filemounts-*'                                                    1 ↵
Deleting 18 storage objects.
I 10-31 08:11:16 storage.py:985] Created S3 bucket skypilot-filemounts-folder-zongheng-fbbb4384-0 in us-east-2
I 10-31 08:11:19 storage.py:840] Deleted S3 bucket skypilot-filemounts-folder-zongheng-fbbb4384-0.
I 10-31 08:11:21 storage.py:985] Created S3 bucket skypilot-filemounts-files-zongheng-f12868a1 in us-east-2
I 10-31 08:11:23 storage.py:840] Deleted S3 bucket skypilot-filemounts-files-zongheng-f12868a1.
I 10-31 08:11:24 storage.py:985] Created S3 bucket skypilot-filemounts-files-zongheng-0a462297 in us-east-2
I 10-31 08:11:26 storage.py:840] Deleted S3 bucket skypilot-filemounts-files-zongheng-0a462297.
I 10-31 08:11:27 storage.py:985] Created S3 bucket skypilot-filemounts-files-zongheng-fbbb4384 in us-east-2
I 10-31 08:11:30 storage.py:840] Deleted S3 bucket skypilot-filemounts-files-zongheng-fbbb4384.
⠼ Deleting S3 bucket skypilot-filemounts-files-zongheng-083760cb
...
```
Ah my bad, I realize this is related to buckets deleted externally. 
Reported by user again:

> How come when I am trying to delete all of my unused filemounts, it creates a bunch of buckets as well:

minimal script to reproduce:

```
# Externally deletes a bucket and tries to delete it using skypilot storage
import sky
import subprocess

bucket_name = 'romilmybucket'

# Create storage object
storage = sky.Storage(name=bucket_name)

storage.add_store(sky.StoreType.S3)

# Delete bucket using aws cli
del_cmd = 'aws s3 rb s3://{} --force'.format(bucket_name)
subprocess.call(del_cmd, shell=True)

# Delete bucket using skypilot
sky_del_cmd = 'sky storage delete {}'.format(bucket_name)
subprocess.call(sky_del_cmd, shell=True)
```

Output:

```
I 04-16 20:37:19 storage.py:1161] Created S3 bucket romilmybucket in us-east-2
remove_bucket: romilmybucket
Deleting 1 storage object.
D 04-16 20:37:22 storage.py:413] Detected existing storage object, loading Storage: romilmybucket
I 04-16 20:37:23 storage.py:1161] Created S3 bucket romilmybucket in us-east-2 # This is bad - bucket should not get created
I 04-16 20:37:26 storage.py:1000] Deleted S3 bucket romilmybucket.
```

Commit message: lint

Patch to review:
diff --git a/sky/data/storage.py b/sky/data/storage.py
index a2355bac57a..bd801a7cafd 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -53,6 +53,10 @@
     '  2. If you are trying to connect to an existing bucket: make sure '
     'your cloud credentials have access to it.')
 
+_BUCKET_EXTERNALLY_DELETED_DEBUG_MESSAGE = (
+    'Bucket {bucket_name!r} does not exist. '
+    'It may have been deleted externally.')
+
 
 class StoreType(enum.Enum):
     """Enum for the different types of stores."""
@@ -164,7 +168,8 @@ def __init__(self,
                  name: str,
                  source: Optional[SourceType],
                  region: Optional[str] = None,
-                 is_sky_managed: Optional[bool] = None):
+                 is_sky_managed: Optional[bool] = None,
+                 sync_on_reconstruction: Optional[bool] = True):
         """Initialize AbstractStore
 
         Args:
@@ -173,6 +178,10 @@ def __init__(self,
             region: Region to place the bucket in
             is_sky_managed: Whether the store is managed by Sky. If None, it
               must be populated by the implementing class during initialization.
+            sync_on_reconstruction: bool; Whether to sync data if the storage
+              object is found in the global_user_state and reconstructed from
+              there. This is set to false when the Storage object is created not
+              for direct use, e.g. for sky storage delete.
 
         Raises:
             StorageBucketCreateError: If bucket creation fails
@@ -183,6 +192,7 @@ def __init__(self,
         self.source = source
         self.region = region
         self.is_sky_managed = is_sky_managed
+        self.sync_on_reconstruction = sync_on_reconstruction
         # Whether sky is responsible for the lifecycle of the Store.
         self._validate()
         self.initialize()
@@ -198,7 +208,9 @@ def from_metadata(cls, metadata: StoreMetadata, **override_args):
                    source=override_args.get('source', metadata.source),
                    region=override_args.get('region', metadata.region),
                    is_sky_managed=override_args.get('is_sky_managed',
-                                                    metadata.is_sky_managed))
+                                                    metadata.is_sky_managed),
+                   sync_on_reconstruction=override_args.get(
+                       'sync_on_reconstruction', True))
 
     def get_metadata(self) -> StoreMetadata:
         return self.StoreMetadata(name=self.name,
@@ -416,14 +428,20 @@ def __init__(self,
                 # When initializing from global_user_state, we override the
                 # source from the YAML
                 if s_type == StoreType.S3:
-                    store = S3Store.from_metadata(s_metadata,
-                                                  source=self.source)
+                    store = S3Store.from_metadata(
+                        s_metadata,
+                        source=self.source,
+                        sync_on_reconstruction=self.sync_on_reconstruction)
                 elif s_type == StoreType.GCS:
-                    store = GcsStore.from_metadata(s_metadata,
-                                                   source=self.source)
+                    store = GcsStore.from_metadata(
+                        s_metadata,
+                        source=self.source,
+                        sync_on_reconstruction=self.sync_on_reconstruction)
                 elif s_type == StoreType.R2:
-                    store = R2Store.from_metadata(s_metadata,
-                                                  source=self.source)
+                    store = R2Store.from_metadata(
+                        s_metadata,
+                        source=self.source,
+                        sync_on_reconstruction=self.sync_on_reconstruction)
                 else:
                     with ux_utils.print_exception_no_traceback():
                         raise ValueError(f'Unknown store type: {s_type}')
@@ -676,7 +694,10 @@ def add_store(self, store_type: Union[str, StoreType]) -> AbstractStore:
 
         # Initialize store object and get/create bucket
         try:
-            store = store_cls(name=self.name, source=self.source)
+            store = store_cls(
+                name=self.name,
+                source=self.source,
+                sync_on_reconstruction=self.sync_on_reconstruction)
         except exceptions.StorageBucketCreateError:
             # Creation failed, so this must be sky managed store. Add failure
             # to state.
@@ -861,10 +882,12 @@ def __init__(self,
                  name: str,
                  source: str,
                  region: Optional[str] = 'us-east-2',
-                 is_sky_managed: Optional[bool] = None):
+                 is_sky_managed: Optional[bool] = None,
+                 sync_on_reconstruction: bool = True):
         self.client: 'boto3.client.Client'
         self.bucket: 'StorageHandle'
-        super().__init__(name, source, region, is_sky_managed)
+        super().__init__(name, source, region, is_sky_managed,
+                         sync_on_reconstruction)
 
     def _validate(self):
         if self.source is not None and isinstance(self.source, str):
@@ -996,8 +1019,13 @@ def upload(self):
                 f'Upload failed for store {self.name}') from e
 
     def delete(self) -> None:
-        self._delete_s3_bucket(self.name)
-        logger.info(f'{colorama.Fore.GREEN}Deleted S3 bucket {self.name}.'
+        deleted_by_skypilot = self._delete_s3_bucket(self.name)
+        if deleted_by_skypilot:
+            msg_str = f'Deleted S3 bucket {self.name}.'
+        else:
+            msg_str = f'S3 bucket {self.name} may have been deleted ' \
+                      f'externally. Removing from local state.'
+        logger.info(f'{colorama.Fore.GREEN}{msg_str}'
                     f'{colorama.Style.RESET_ALL}')
 
     def get_handle(self) -> StorageHandle:
@@ -1065,13 +1093,15 @@ def _transfer_to_s3(self) -> None:
         elif self.source.startswith('r2://'):
             data_transfer.r2_to_s3(self.name, self.name)
 
-    def _get_bucket(self) -> Tuple[StorageHandle, bool]:
+    def _get_bucket(self) -> Tuple[Optional[StorageHandle], bool]:
         """Obtains the S3 bucket.
 
-        If the bucket exists, this method will connect to the bucket.
-        If the bucket does not exist, there are two cases:
+        If the bucket exists, this method will return the bucket.
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with s3://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
+          2) Create and return a new bucket otherwise
 
         Raises:
             StorageBucketCreateError: If creating the bucket fails
@@ -1106,9 +1136,13 @@ def _get_bucket(self) -> Tuple[StorageHandle, bool]:
                     f'{self.source}` to debug.')
 
         # If bucket cannot be found in both private and public settings,
-        # the bucket is created by Sky.
-        bucket = self._create_s3_bucket(self.name)
-        return bucket, True
+        # the bucket is to be created by Sky. However, skip creation if
+        # Store object is being reconstructed for deletion.
+        if self.sync_on_reconstruction:
+            bucket = self._create_s3_bucket(self.name)
+            return bucket, True
+        else:
+            return None, False
 
     def _download_file(self, remote_path: str, local_path: str) -> None:
         """Downloads file from remote to local on s3 bucket
@@ -1166,11 +1200,14 @@ def _create_s3_bucket(self,
                     f'{self.name} but failed.') from e
         return aws.resource('s3').Bucket(bucket_name)
 
-    def _delete_s3_bucket(self, bucket_name: str) -> None:
+    def _delete_s3_bucket(self, bucket_name: str) -> bool:
         """Deletes S3 bucket, including all objects in bucket
 
         Args:
           bucket_name: str; Name of bucket
+
+        Returns:
+         bool; True if bucket was deleted, False if it was deleted externally.
         """
         # Deleting objects is very slow programatically
         # (i.e. bucket.objects.all().delete() is slow).
@@ -1183,16 +1220,24 @@ def _delete_s3_bucket(self, bucket_name: str) -> None:
         try:
             with log_utils.safe_rich_status(
                     f'[bold cyan]Deleting S3 bucket {bucket_name}[/]'):
-                subprocess.check_output(remove_command.split(' '))
+                subprocess.check_output(remove_command.split(' '),
+                                        stderr=subprocess.STDOUT)
         except subprocess.CalledProcessError as e:
-            logger.error(e.output)
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.StorageBucketDeleteError(
-                    f'Failed to delete S3 bucket {bucket_name}.')
+            if 'NoSuchBucket' in e.output.decode('utf-8'):
+                logger.debug(
+                    _BUCKET_EXTERNALLY_DELETED_DEBUG_MESSAGE.format(
+                        bucket_name=bucket_name))
+                return False
+            else:
+                logger.error(e.output)
+                with ux_utils.print_exception_no_traceback():
+                    raise exceptions.StorageBucketDeleteError(
+                        f'Failed to delete S3 bucket {bucket_name}.')
 
         # Wait until bucket deletion propagates on AWS servers
         while data_utils.verify_s3_bucket(bucket_name):
             time.sleep(0.1)
+        return True
 
 
 class GcsStore(AbstractStore):
@@ -1207,10 +1252,12 @@ def __init__(self,
                  name: str,
                  source: str,
                  region: Optional[str] = 'us-central1',
-                 is_sky_managed: Optional[bool] = None):
+                 is_sky_managed: Optional[bool] = None,
+                 sync_on_reconstruction: Optional[bool] = True):
         self.client: 'storage.Client'
         self.bucket: StorageHandle
-        super().__init__(name, source, region, is_sky_managed)
+        super().__init__(name, source, region, is_sky_managed,
+                         sync_on_reconstruction)
 
     def _validate(self):
         if self.source is not None:
@@ -1349,8 +1396,13 @@ def upload(self):
                 f'Upload failed for store {self.name}') from e
 
     def delete(self) -> None:
-        self._delete_gcs_bucket(self.name)
-        logger.info(f'{colorama.Fore.GREEN}Deleted GCS bucket {self.name}.'
+        deleted_by_skypilot = self._delete_gcs_bucket(self.name)
+        if deleted_by_skypilot:
+            msg_str = f'Deleted GCS bucket {self.name}.'
+        else:
+            msg_str = f'GCS bucket {self.name} may have been deleted ' \
+                      f'externally. Removing from local state.'
+        logger.info(f'{colorama.Fore.GREEN}{msg_str}'
                     f'{colorama.Style.RESET_ALL}')
 
     def get_handle(self) -> StorageHandle:
@@ -1455,9 +1507,11 @@ def _get_bucket(self) -> Tuple[StorageHandle, bool]:
         """Obtains the GCS bucket.
         If the bucket exists, this method will connect to the bucket.
 
-        If the bucket does not exist, there are two cases:
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with gs://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
+          2) Create and return a new bucket otherwise
 
         Raises:
             StorageBucketCreateError: If creating the bucket fails
@@ -1473,8 +1527,15 @@ def _get_bucket(self) -> Tuple[StorageHandle, bool]:
                         'Attempted to connect to a non-existent bucket: '
                         f'{self.source}') from e
             else:
-                bucket = self._create_gcs_bucket(self.name)
-                return bucket, True
+
+                # If bucket cannot be found (i.e., does not exist), it is to be
+                # created by Sky. However, skip creation if Store object is
+                # being reconstructed for deletion.
+                if self.sync_on_reconstruction:
+                    bucket = self._create_gcs_bucket(self.name)
+                    return bucket, True
+                else:
+                    return None, False
         except gcp.forbidden_exception():
             # Try public bucket to see if bucket exists
             logger.info(
@@ -1550,33 +1611,44 @@ def _create_gcs_bucket(self,
             f'with storage class {new_bucket.storage_class}')
         return new_bucket
 
-    def _delete_gcs_bucket(self, bucket_name: str) -> None:
+    def _delete_gcs_bucket(self, bucket_name: str) -> bool:
         """Deletes GCS bucket, including all objects in bucket
 
         Args:
           bucket_name: str; Name of bucket
+
+        Returns:
+         bool; True if bucket was deleted, False if it was deleted externally.
         """
-        try:
-            self.client.get_bucket(bucket_name)
-        except gcp.forbidden_exception() as e:
-            # Try public bucket to see if bucket exists
-            with ux_utils.print_exception_no_traceback():
-                raise PermissionError(
-                    'External Bucket detected. User not allowed to delete '
-                    'external bucket.') from e
 
-        try:
-            with log_utils.safe_rich_status(
-                    f'[bold cyan]Deleting GCS bucket {bucket_name}[/]'):
+        with log_utils.safe_rich_status(
+                f'[bold cyan]Deleting GCS bucket {bucket_name}[/]'):
+            try:
+                self.client.get_bucket(bucket_name)
+            except gcp.forbidden_exception() as e:
+                # Try public bucket to see if bucket exists
+                with ux_utils.print_exception_no_traceback():
+                    raise PermissionError(
+                        'External Bucket detected. User not allowed to delete '
+                        'external bucket.') from e
+            except gcp.not_found_exception():
+                # If bucket does not exist, it may have been deleted externally.
+                # Do a no-op in that case.
+                logger.debug(
+                    _BUCKET_EXTERNALLY_DELETED_DEBUG_MESSAGE.format(
+                        bucket_name=bucket_name))
+                return False
+            try:
                 remove_obj_command = ('gsutil -m rm -r'
                                       f' gs://{bucket_name}')
                 subprocess.check_output(remove_obj_command.split(' '),
                                         stderr=subprocess.STDOUT)
-        except subprocess.CalledProcessError as e:
-            logger.error(e.output)
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.StorageBucketDeleteError(
-                    f'Failed to delete GCS bucket {bucket_name}.')
+                return True
+            except subprocess.CalledProcessError as e:
+                logger.error(e.output)
+                with ux_utils.print_exception_no_traceback():
+                    raise exceptions.StorageBucketDeleteError(
+                        f'Failed to delete GCS bucket {bucket_name}.')
 
 
 class R2Store(AbstractStore):
@@ -1590,10 +1662,12 @@ def __init__(self,
                  name: str,
                  source: str,
                  region: Optional[str] = 'auto',
-                 is_sky_managed: Optional[bool] = None):
+                 is_sky_managed: Optional[bool] = None,
+                 sync_on_reconstruction: Optional[bool] = True):
         self.client: 'boto3.client.Client'
         self.bucket: 'StorageHandle'
-        super().__init__(name, source, region, is_sky_managed)
+        super().__init__(name, source, region, is_sky_managed,
+                         sync_on_reconstruction)
 
     def _validate(self):
         if self.source is not None and isinstance(self.source, str):
@@ -1665,8 +1739,13 @@ def upload(self):
                 f'Upload failed for store {self.name}') from e
 
     def delete(self) -> None:
-        self._delete_r2_bucket(self.name)
-        logger.info(f'{colorama.Fore.GREEN}Deleted R2 bucket {self.name}.'
+        deleted_by_skypilot = self._delete_r2_bucket(self.name)
+        if deleted_by_skypilot:
+            msg_str = f'Deleted R2 bucket {self.name}.'
+        else:
+            msg_str = f'R2 bucket {self.name} may have been deleted ' \
+                      f'externally. Removing from local state.'
+        logger.info(f'{colorama.Fore.GREEN}{msg_str}'
                     f'{colorama.Style.RESET_ALL}')
 
     def get_handle(self) -> StorageHandle:
@@ -1743,10 +1822,12 @@ def _transfer_to_r2(self) -> None:
     def _get_bucket(self) -> Tuple[StorageHandle, bool]:
         """Obtains the R2 bucket.
 
-        If the bucket exists, this method will connect to the bucket.
-        If the bucket does not exist, there are two cases:
+        If the bucket exists, this method will return the bucket.
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with s3://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
+          2) Create and return a new bucket otherwise
 
         Raises:
             StorageBucketCreateError: If creating the bucket fails
@@ -1786,9 +1867,13 @@ def _get_bucket(self) -> Tuple[StorageHandle, bool]:
                     'to debug.')
 
         # If bucket cannot be found in both private and public settings,
-        # the bucket is created by Sky.
-        bucket = self._create_r2_bucket(self.name)
-        return bucket, True
+        # the bucket is to be created by Sky. However, skip creation if
+        # Store object is being reconstructed for deletion.
+        if self.sync_on_reconstruction:
+            bucket = self._create_r2_bucket(self.name)
+            return bucket, True
+        else:
+            return None, False
 
     def _download_file(self, remote_path: str, local_path: str) -> None:
         """Downloads file from remote to local on r2 bucket
@@ -1849,11 +1934,14 @@ def _create_r2_bucket(self,
                     f'{self.name} but failed.') from e
         return cloudflare.resource('s3').Bucket(bucket_name)
 
-    def _delete_r2_bucket(self, bucket_name: str) -> None:
+    def _delete_r2_bucket(self, bucket_name: str) -> bool:
         """Deletes R2 bucket, including all objects in bucket
 
         Args:
           bucket_name: str; Name of bucket
+
+        Returns:
+         bool; True if bucket was deleted, False if it was deleted externally.
         """
         # Deleting objects is very slow programatically
         # (i.e. bucket.objects.all().delete() is slow).
@@ -1869,13 +1957,21 @@ def _delete_r2_bucket(self, bucket_name: str) -> None:
         try:
             with log_utils.safe_rich_status(
                     f'[bold cyan]Deleting R2 bucket {bucket_name}[/]'):
-                subprocess.check_output(remove_command.split(' '))
+                subprocess.check_output(remove_command.split(' '),
+                                        stderr=subprocess.STDOUT)
         except subprocess.CalledProcessError as e:
-            logger.error(e.output)
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.StorageBucketDeleteError(
-                    f'Failed to delete R2 bucket {bucket_name}.')
+            if 'NoSuchBucket' in e.output.decode('utf-8'):
+                logger.debug(
+                    _BUCKET_EXTERNALLY_DELETED_DEBUG_MESSAGE.format(
+                        bucket_name=bucket_name))
+                return False
+            else:
+                logger.error(e.output)
+                with ux_utils.print_exception_no_traceback():
+                    raise exceptions.StorageBucketDeleteError(
+                        f'Failed to delete R2 bucket {bucket_name}.')
 
         # Wait until bucket deletion propagates on AWS servers
         while data_utils.verify_r2_bucket(bucket_name):
             time.sleep(0.1)
+        return True
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 130849a6189..b7d0afa08c3 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -1781,6 +1781,47 @@ class TestStorageWithCredentials:
         'abc_',  # ends with an underscore
     ]
 
+    @staticmethod
+    def cli_delete_cmd(store_type, bucket_name):
+        if store_type == storage_lib.StoreType.S3:
+            url = f's3://{bucket_name}'
+            return ['aws', 's3', 'rb', url, '--force']
+        if store_type == storage_lib.StoreType.GCS:
+            url = f'gs://{bucket_name}'
+            return ['gsutil', '-m', 'rm', '-r', url]
+        if store_type == storage_lib.StoreType.R2:
+            endpoint_url = cloudflare.create_endpoint()
+            url = f's3://{bucket_name}'
+            return [
+                'aws', 's3', 'rb', url, '--force', '--endpoint', endpoint_url,
+                '--profile=r2'
+            ]
+
+    @staticmethod
+    def cli_ls_cmd(store_type, bucket_name, suffix=''):
+        if store_type == storage_lib.StoreType.S3:
+            if suffix:
+                url = f's3://{bucket_name}/{suffix}'
+            else:
+                url = f's3://{bucket_name}'
+            return ['aws', 's3', 'ls', url]
+        if store_type == storage_lib.StoreType.GCS:
+            if suffix:
+                url = f'gs://{bucket_name}/{suffix}'
+            else:
+                url = f'gs://{bucket_name}'
+            return ['gsutil', 'ls', url]
+        if store_type == storage_lib.StoreType.R2:
+            endpoint_url = cloudflare.create_endpoint()
+            if suffix:
+                url = f's3://{bucket_name}/{suffix}'
+            else:
+                url = f's3://{bucket_name}'
+            return [
+                'aws', 's3', 'ls', url, '--endpoint', endpoint_url,
+                '--profile=r2'
+            ]
+
     @pytest.fixture
     def tmp_source(self, tmp_path):
         # Creates a temporary directory with a file in it
@@ -1916,6 +1957,34 @@ def test_new_bucket_creation_and_deletion(self, tmp_local_storage_obj,
         out = subprocess.check_output(['sky', 'storage', 'ls'])
         assert tmp_local_storage_obj.name not in out.decode('utf-8')
 
+    @pytest.mark.parametrize('store_type', [
+        storage_lib.StoreType.S3, storage_lib.StoreType.GCS,
+        pytest.param(storage_lib.StoreType.R2, marks=pytest.mark.cloudflare)
+    ])
+    def test_bucket_external_deletion(self, tmp_scratch_storage_obj,
+                                      store_type):
+        # Creates a bucket, deletes it externally using cloud cli commands
+        # and then tries to delete it using sky storage delete.
+        tmp_scratch_storage_obj.add_store(store_type)
+
+        # Run sky storage ls to check if storage object exists in the output
+        out = subprocess.check_output(['sky', 'storage', 'ls'])
+        assert tmp_scratch_storage_obj.name in out.decode('utf-8')
+
+        # Delete bucket externally
+        cmd = self.cli_delete_cmd(store_type, tmp_scratch_storage_obj.name)
+        subprocess.check_output(cmd)
+
+        # Run sky storage delete to delete the storage object
+        out = subprocess.check_output(
+            ['sky', 'storage', 'delete', tmp_scratch_storage_obj.name])
+        # Make sure bucket was not created during deletion (see issue #1322)
+        assert 'created' not in out.decode('utf-8').lower()
+
+        # Run sky storage ls to check if storage object is deleted
+        out = subprocess.check_output(['sky', 'storage', 'ls'])
+        assert tmp_scratch_storage_obj.name not in out.decode('utf-8')
+
     @pytest.mark.parametrize('store_type', [
         storage_lib.StoreType.S3, storage_lib.StoreType.GCS,
         pytest.param(storage_lib.StoreType.R2, marks=pytest.mark.cloudflare)
@@ -2024,31 +2093,6 @@ def test_private_bucket(self, private_bucket):
                     name=private_bucket_name)):
             storage_obj = storage_lib.Storage(source=private_bucket)
 
-    @staticmethod
-    def cli_ls_cmd(store_type, bucket_name, suffix=''):
-        if store_type == storage_lib.StoreType.S3:
-            if suffix:
-                url = f's3://{bucket_name}/{suffix}'
-            else:
-                url = f's3://{bucket_name}'
-            return ['aws', 's3', 'ls', url]
-        if store_type == storage_lib.StoreType.GCS:
-            if suffix:
-                url = f'gs://{bucket_name}/{suffix}'
-            else:
-                url = f'gs://{bucket_name}'
-            return ['gsutil', 'ls', url]
-        if store_type == storage_lib.StoreType.R2:
-            endpoint_url = cloudflare.create_endpoint()
-            if suffix:
-                url = f's3://{bucket_name}/{suffix}'
-            else:
-                url = f's3://{bucket_name}'
-            return [
-                'aws', 's3', 'ls', url, '--endpoint', endpoint_url,
-                '--profile=r2'
-            ]
-
     @pytest.mark.parametrize('ext_bucket_fixture, store_type',
                              [('tmp_awscli_bucket', storage_lib.StoreType.S3),
                               ('tmp_gsutil_bucket', storage_lib.StoreType.GCS),


Code review comments:
diff_hunk: @@ -1455,9 +1507,11 @@ def _get_bucket(self) -> Tuple[StorageHandle, bool]:
         """Obtains the GCS bucket.
         If the bucket exists, this method will connect to the bucket.
 
-        If the bucket does not exist, there are two cases:
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with gs://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
path: sky/data/storage.py
text: @user1:
```suggestion
             sync_on_reconstruction is False
```

diff_hunk: @@ -1065,13 +1093,15 @@ def _transfer_to_s3(self) -> None:
         elif self.source.startswith('r2://'):
             data_transfer.r2_to_s3(self.name, self.name)
 
-    def _get_bucket(self) -> Tuple[StorageHandle, bool]:
+    def _get_bucket(self) -> Tuple[Optional[StorageHandle], bool]:
         """Obtains the S3 bucket.
 
-        If the bucket exists, this method will connect to the bucket.
-        If the bucket does not exist, there are two cases:
+        If the bucket exists, this method will return the bucket.
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with s3://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
+          2) Create and return a new bucket otherwise
path: sky/data/storage.py
text: @user1:
```suggestion
          3) Create and return a new bucket otherwise
```

diff_hunk: @@ -1743,10 +1822,12 @@ def _transfer_to_r2(self) -> None:
     def _get_bucket(self) -> Tuple[StorageHandle, bool]:
         """Obtains the R2 bucket.
 
-        If the bucket exists, this method will connect to the bucket.
-        If the bucket does not exist, there are two cases:
+        If the bucket exists, this method will return the bucket.
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with s3://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
+          2) Create and return a new bucket otherwise
path: sky/data/storage.py
text: @user1:
```suggestion
          3) Create and return a new bucket otherwise
```

diff_hunk: @@ -1743,10 +1822,12 @@ def _transfer_to_r2(self) -> None:
     def _get_bucket(self) -> Tuple[StorageHandle, bool]:
         """Obtains the R2 bucket.
 
-        If the bucket exists, this method will connect to the bucket.
-        If the bucket does not exist, there are two cases:
+        If the bucket exists, this method will return the bucket.
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with s3://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
path: sky/data/storage.py
text: @user1:
```suggestion
             sync_on_reconstruction is False
```

diff_hunk: @@ -1455,9 +1507,11 @@ def _get_bucket(self) -> Tuple[StorageHandle, bool]:
         """Obtains the GCS bucket.
         If the bucket exists, this method will connect to the bucket.
 
-        If the bucket does not exist, there are two cases:
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with gs://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
+          2) Create and return a new bucket otherwise
path: sky/data/storage.py
text: @user1:
```suggestion
          3) Create and return a new bucket otherwise
```

diff_hunk: @@ -1065,13 +1093,15 @@ def _transfer_to_s3(self) -> None:
         elif self.source.startswith('r2://'):
             data_transfer.r2_to_s3(self.name, self.name)
 
-    def _get_bucket(self) -> Tuple[StorageHandle, bool]:
+    def _get_bucket(self) -> Tuple[Optional[StorageHandle], bool]:
         """Obtains the S3 bucket.
 
-        If the bucket exists, this method will connect to the bucket.
-        If the bucket does not exist, there are two cases:
+        If the bucket exists, this method will return the bucket.
+        If the bucket does not exist, there are three cases:
           1) Raise an error if the bucket source starts with s3://
-          2) Create a new bucket otherwise
+          2) Return None if bucket has been externally deleted and
+             sync_on_reconstruction if False
path: sky/data/storage.py
text: @user1:
```suggestion
             sync_on_reconstruction is False
```