diff --git a/notebooks/tutorials/model-training/01-data-scientist-submit-code.ipynb b/notebooks/tutorials/model-training/01-data-scientist-submit-code.ipynb
index a5b6b4a4d94..6063d233753 100644
--- a/notebooks/tutorials/model-training/01-data-scientist-submit-code.ipynb
+++ b/notebooks/tutorials/model-training/01-data-scientist-submit-code.ipynb
@@ -103,7 +103,8 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "assert training_images.data is None"
+    "assert isinstance(training_images.data, sy.SyftError)\n",
+    "training_labels.data"
    ]
   },
   {
diff --git a/packages/syft/src/syft/client/datasite_client.py b/packages/syft/src/syft/client/datasite_client.py
index 9ef0638acb6..aa7c2ce2894 100644
--- a/packages/syft/src/syft/client/datasite_client.py
+++ b/packages/syft/src/syft/client/datasite_client.py
@@ -108,7 +108,7 @@ def upload_dataset(self, dataset: CreateDataset) -> SyftSuccess | SyftError:
             asset = dataset.asset_list[i]
             dataset.asset_list[i] = add_default_uploader(user, asset)
 
-        dataset._check_asset_must_contain_mock()
+        # dataset._check_asset_must_contain_mock()
         dataset_size: float = 0.0
 
         # TODO: Refactor so that object can also be passed to generate warnings
diff --git a/packages/syft/src/syft/service/action/action_service.py b/packages/syft/src/syft/service/action/action_service.py
index 7c0a65c5fcf..ab82c80a2b8 100644
--- a/packages/syft/src/syft/service/action/action_service.py
+++ b/packages/syft/src/syft/service/action/action_service.py
@@ -2,6 +2,7 @@
 import importlib
 import logging
 from typing import Any
+from typing import cast
 
 # third party
 import numpy as np
@@ -289,10 +290,6 @@ def _get(
         resolve_nested: bool = True,
     ) -> Result[ActionObject, str]:
         """Get an object from the action store"""
-        # stdlib
-
-        # relative
-
         result = self.store.get(
             uid=uid, credentials=context.credentials, has_permission=has_permission
         )
@@ -945,12 +942,114 @@ def exists(
 
     @service_method(path="action.delete", name="delete", roles=ADMIN_ROLE_LEVEL)
     def delete(
-        self, context: AuthedServiceContext, uid: UID
+        self, context: AuthedServiceContext, uid: UID, soft_delete: bool = False
     ) -> SyftSuccess | SyftError:
-        res = self.store.delete(context.credentials, uid)
-        if res.is_err():
-            return SyftError(message=res.err())
-        return SyftSuccess(message="Great Success!")
+        get_res = self.store.get(uid=uid, credentials=context.credentials)
+        if get_res.is_err():
+            return SyftError(message=get_res.err())
+        obj: ActionObject | TwinObject = get_res.ok()
+        return_msg = []
+
+        # delete any associated blob storage entry object to the action object
+        blob_del_res = self._delete_blob_storage_entry(context=context, obj=obj)
+        if isinstance(blob_del_res, SyftError):
+            return SyftError(message=blob_del_res.message)
+        return_msg.append(blob_del_res.message)
+
+        # delete the action object from the action store
+        store_del_res = self._delete_from_action_store(
+            context=context, uid=obj.id, soft_delete=soft_delete
+        )
+        if isinstance(store_del_res, SyftError):
+            return SyftError(message=store_del_res.message)
+        return_msg.append(store_del_res.message)
+
+        return SyftSuccess(message="\n".join(return_msg))
+
+    def _delete_blob_storage_entry(
+        self,
+        context: AuthedServiceContext,
+        obj: TwinObject | ActionObject,
+    ) -> SyftSuccess | SyftError:
+        deleted_blob_ids = []
+        blob_store_service = cast(
+            BlobStorageService, context.server.get_service(BlobStorageService)
+        )
+
+        if isinstance(obj, ActionObject) and obj.syft_blob_storage_entry_id:
+            blob_del_res = blob_store_service.delete(
+                context=context, uid=obj.syft_blob_storage_entry_id
+            )
+            if isinstance(blob_del_res, SyftError):
+                return SyftError(message=blob_del_res.message)
+            deleted_blob_ids.append(obj.syft_blob_storage_entry_id)
+
+        if isinstance(obj, TwinObject):
+            if obj.private.syft_blob_storage_entry_id:
+                blob_del_res = blob_store_service.delete(
+                    context=context, uid=obj.private.syft_blob_storage_entry_id
+                )
+                if isinstance(blob_del_res, SyftError):
+                    return SyftError(message=blob_del_res.message)
+                deleted_blob_ids.append(obj.private.syft_blob_storage_entry_id)
+
+            if obj.mock.syft_blob_storage_entry_id:
+                blob_del_res = blob_store_service.delete(
+                    context=context, uid=obj.mock.syft_blob_storage_entry_id
+                )
+                if isinstance(blob_del_res, SyftError):
+                    return SyftError(message=blob_del_res.message)
+                deleted_blob_ids.append(obj.mock.syft_blob_storage_entry_id)
+
+        message = f"Deleted blob storage entries: {', '.join(str(blob_id) for blob_id in deleted_blob_ids)}"
+
+        return SyftSuccess(message=message)
+
+    def _delete_from_action_store(
+        self,
+        context: AuthedServiceContext,
+        uid: UID,
+        soft_delete: bool = False,
+    ) -> SyftSuccess | SyftError:
+        if soft_delete:
+            get_res = self.store.get(uid=uid, credentials=context.credentials)
+            if get_res.is_err():
+                return SyftError(message=get_res.err())
+            obj: ActionObject | TwinObject = get_res.ok()
+
+            if isinstance(obj, TwinObject):
+                res = self._soft_delete_action_obj(
+                    context=context, action_obj=obj.private
+                )
+                if res.is_err():
+                    return SyftError(message=res.err())
+                res = self._soft_delete_action_obj(context=context, action_obj=obj.mock)
+                if res.is_err():
+                    return SyftError(message=res.err())
+
+            if isinstance(obj, ActionObject):
+                res = self._soft_delete_action_obj(context=context, action_obj=obj)
+                if res.is_err():
+                    return SyftError(message=res.err())
+        else:
+            res = self.store.delete(credentials=context.credentials, uid=uid)
+            if res.is_err():
+                return SyftError(message=res.err())
+
+        return SyftSuccess(message=f"Action object with uid '{uid}' deleted.")
+
+    def _soft_delete_action_obj(
+        self, context: AuthedServiceContext, action_obj: ActionObject
+    ) -> Result[ActionObject, str]:
+        action_obj.syft_action_data_cache = None
+        res = action_obj._save_to_blob_storage()
+        if isinstance(res, SyftError):
+            return Err(res.message)
+        set_result = self._set(
+            context=context,
+            action_object=action_obj,
+        )
+        return set_result
 
 
 def resolve_action_args(
diff --git a/packages/syft/src/syft/service/blob_storage/service.py b/packages/syft/src/syft/service/blob_storage/service.py
index dfcddbe2c66..3254e2918da 100644
--- a/packages/syft/src/syft/service/blob_storage/service.py
+++ b/packages/syft/src/syft/service/blob_storage/service.py
@@ -352,30 +352,36 @@ def mark_write_complete(
     def delete(
         self, context: AuthedServiceContext, uid: UID
     ) -> SyftSuccess | SyftError:
-        result = self.stash.get_by_uid(context.credentials, uid=uid)
-        if result.is_ok():
-            obj = result.ok()
+        get_res = self.stash.get_by_uid(context.credentials, uid=uid)
+        if get_res.is_err():
+            return SyftError(message=get_res.err())
 
-            if obj is None:
-                return SyftError(
-                    message=f"No blob storage entry exists for uid: {uid}, or you have no permissions to read it"
-                )
-
-            try:
-                with context.server.blob_storage_client.connect() as conn:
-                    file_unlinked_result = conn.delete(obj.location)
-            except Exception as e:
-                return SyftError(message=f"Failed to delete file: {e}")
+        obj = get_res.ok()
+        if obj is None:
+            return SyftError(
+                message=f"No blob storage entry exists for uid: {uid}, "
+                f"or you have no permissions to read it"
+            )
 
-            if isinstance(file_unlinked_result, SyftError):
-                return file_unlinked_result
-            blob_storage_entry_deleted = self.stash.delete(
-                context.credentials, UIDPartitionKey.with_obj(uid), has_permission=True
+        try:
+            with context.server.blob_storage_client.connect() as conn:
+                file_unlinked_result = conn.delete(obj.location)
+                if isinstance(file_unlinked_result, SyftError):
+                    return file_unlinked_result
+        except Exception as e:
+            return SyftError(
+                message=f"Failed to delete blob file with id '{uid}'. Error: {e}"
             )
-            if blob_storage_entry_deleted.is_ok():
-                return file_unlinked_result
 
-        return SyftError(message=result.err())
+        blob_entry_delete_res = self.stash.delete(
+            context.credentials, UIDPartitionKey.with_obj(uid), has_permission=True
+        )
+        if blob_entry_delete_res.is_err():
+            return SyftError(message=blob_entry_delete_res.err())
+
+        return SyftSuccess(
+            message=f"Blob storage entry with id '{uid}' deleted successfully."
+        )
 
 
 TYPE_TO_SERVICE[BlobStorageEntry] = BlobStorageEntry
diff --git a/packages/syft/src/syft/service/dataset/dataset.py b/packages/syft/src/syft/service/dataset/dataset.py
index 623e8d9f763..5ab3eb1411b 100644
--- a/packages/syft/src/syft/service/dataset/dataset.py
+++ b/packages/syft/src/syft/service/dataset/dataset.py
@@ -20,14 +20,17 @@
 from typing_extensions import Self
 
 # relative
+from ...client.api import APIRegistry
 from ...serde.serializable import serializable
 from ...store.document_store import PartitionKey
 from ...types.datetime import DateTime
 from ...types.dicttuple import DictTuple
+from ...types.syft_object import PartialSyftObject
 from ...types.syft_object import SYFT_OBJECT_VERSION_1
 from ...types.syft_object import SyftObject
 from ...types.transforms import TransformContext
 from ...types.transforms import generate_id
+from ...types.transforms import make_set_default
 from ...types.transforms import transform
 from ...types.transforms import validate_url
 from ...types.uid import UID
@@ -239,9 +242,6 @@ def __eq__(self, other: object) -> bool:
 
     @property
     def pointer(self) -> Any:
-        # relative
-        from ...client.api import APIRegistry
-
         api = APIRegistry.api_for(
             server_uid=self.server_uid,
             user_verify_key=self.syft_client_verify_key,
@@ -251,9 +251,6 @@ def pointer(self) -> Any:
 
     @property
     def mock(self) -> SyftError | Any:
-        # relative
-        from ...client.api import APIRegistry
-
         api = APIRegistry.api_for(
             server_uid=self.server_uid,
             user_verify_key=self.syft_client_verify_key,
@@ -261,6 +258,8 @@ def mock(self) -> SyftError | Any:
         if api is None:
             return SyftError(message=f"You must login to {self.server_uid}")
         result = api.services.action.get_mock(self.action_id)
+        if isinstance(result, SyftError):
+            return result
         try:
             if isinstance(result, SyftObject):
                 return result.syft_action_data
@@ -282,7 +281,6 @@ def has_permission(self, data_result: Any) -> bool:
     @property
     def data(self) -> Any:
         # relative
-        from ...client.api import APIRegistry
 
         api = APIRegistry.api_for(
             server_uid=self.server_uid,
@@ -291,6 +289,8 @@ def data(self) -> Any:
         if api is None or api.services is None:
             return None
         res = api.services.action.get(self.action_id)
+        if isinstance(res, str):
+            return SyftError(message=f"Could not access private data. {str(res)}")
         if self.has_permission(res):
             return res.syft_action_data
         else:
@@ -471,6 +471,7 @@ class Dataset(SyftObject):
     created_at: DateTime = DateTime.now()
     uploader: Contributor
     summary: str | None = None
+    to_be_deleted: bool = False
 
     __attr_searchable__ = [
         "name",
@@ -523,6 +524,8 @@ def _repr_html_(self) -> Any:
             """
         else:
             description_info_message = ""
+        if self.to_be_deleted:
+            return "This dataset has been marked for deletion. The underlying data may be not available."
         return f"""
             <style>
             {FONT_CSS}
@@ -549,8 +552,7 @@ def _repr_html_(self) -> Any:
             """
 
     def action_ids(self) -> list[UID]:
-        data = [asset.action_id for asset in self.asset_list if asset.action_id]
-        return data
+        return [asset.action_id for asset in self.asset_list if asset.action_id]
 
     @property
     def assets(self) -> DictTuple[str, Asset]:
@@ -639,9 +641,6 @@ class CreateDataset(Dataset):
 
     model_config = ConfigDict(validate_assignment=True, extra="forbid")
 
-    def _check_asset_must_contain_mock(self) -> None:
-        _check_asset_must_contain_mock(self.asset_list)
-
     @field_validator("asset_list")
     @classmethod
     def __assets_must_contain_mock(
@@ -650,6 +649,13 @@ def __assets_must_contain_mock(
         _check_asset_must_contain_mock(asset_list)
         return asset_list
 
+    @field_validator("to_be_deleted")
+    @classmethod
+    def __to_be_deleted_must_be_false(cls, v: bool) -> bool:
+        if v is True:
+            raise ValueError("to_be_deleted must be False")
+        return v
+
     def set_description(self, description: str) -> None:
         self.description = MarkdownDescription(text=description)
 
@@ -876,8 +882,13 @@ def createdataset_to_dataset() -> list[Callable]:
         validate_url,
         convert_asset,
         add_current_date,
+        make_set_default("to_be_deleted", False),  # explicitly set it to False
     ]
 
 
-class DatasetUpdate:
-    pass
+class DatasetUpdate(PartialSyftObject):
+    __canonical_name__ = "DatasetUpdate"
+    __version__ = SYFT_OBJECT_VERSION_1
+
+    name: str
+    to_be_deleted: bool
diff --git a/packages/syft/src/syft/service/dataset/dataset_service.py b/packages/syft/src/syft/service/dataset/dataset_service.py
index 3eeaaf80551..c3bc68385ad 100644
--- a/packages/syft/src/syft/service/dataset/dataset_service.py
+++ b/packages/syft/src/syft/service/dataset/dataset_service.py
@@ -1,6 +1,8 @@
 # stdlib
 from collections.abc import Collection
 from collections.abc import Sequence
+import logging
+from typing import cast
 
 # relative
 from ...serde.serializable import serializable
@@ -10,6 +12,7 @@
 from ...util.telemetry import instrument
 from ..action.action_permissions import ActionObjectPermission
 from ..action.action_permissions import ActionPermission
+from ..action.action_service import ActionService
 from ..context import AuthedServiceContext
 from ..response import SyftError
 from ..response import SyftSuccess
@@ -26,8 +29,11 @@
 from .dataset import CreateDataset
 from .dataset import Dataset
 from .dataset import DatasetPageView
+from .dataset import DatasetUpdate
 from .dataset_stash import DatasetStash
 
+logger = logging.getLogger(__name__)
+
 
 def _paginate_collection(
     collection: Collection,
@@ -123,6 +129,8 @@ def get_all(
         for dataset in datasets:
             if context.server is not None:
                 dataset.server_uid = context.server.id
+            if dataset.to_be_deleted:
+                datasets.remove(dataset)
 
         return _paginate_dataset_collection(
             datasets=datasets, page_size=page_size, page_index=page_index
@@ -143,7 +151,9 @@ def search(
             return results
 
         filtered_results = [
-            dataset for dataset_name, dataset in results.items() if name in dataset_name
+            dataset
+            for dataset_name, dataset in results.items()
+            if name in dataset_name and not dataset.to_be_deleted
         ]
 
         return _paginate_dataset_collection(
@@ -151,17 +161,16 @@ def search(
         )
 
     @service_method(path="dataset.get_by_id", name="get_by_id")
-    def get_by_id(
-        self, context: AuthedServiceContext, uid: UID
-    ) -> SyftSuccess | SyftError:
+    def get_by_id(self, context: AuthedServiceContext, uid: UID) -> Dataset | SyftError:
         """Get a Dataset"""
         result = self.stash.get_by_uid(context.credentials, uid=uid)
-        if result.is_ok():
-            dataset = result.ok()
-            if context.server is not None:
-                dataset.server_uid = context.server.id
-            return dataset
-        return SyftError(message=result.err())
+        if result.is_err():
+            return SyftError(message=result.err())
+        dataset = result.ok()
+
+        if context.server is not None:
+            dataset.server_uid = context.server.id
+        return dataset
 
     @service_method(path="dataset.get_by_action_id", name="get_by_action_id")
     def get_by_action_id(
@@ -169,13 +178,15 @@ def get_by_action_id(
     ) -> list[Dataset] | SyftError:
         """Get Datasets by an Action ID"""
         result = self.stash.search_action_ids(context.credentials, uid=uid)
-        if result.is_ok():
-            datasets = result.ok()
-            for dataset in datasets:
-                if context.server is not None:
-                    dataset.server_uid = context.server.id
-            return datasets
-        return SyftError(message=result.err())
+        if result.is_err():
+            return SyftError(message=result.err())
+        datasets = result.ok()
+
+        for dataset in datasets:
+            if context.server is not None:
+                dataset.server_uid = context.server.id
+
+        return datasets
 
     @service_method(
         path="dataset.get_assets_by_action_id",
@@ -189,6 +200,9 @@ def get_assets_by_action_id(
         datasets = self.get_by_action_id(context=context, uid=uid)
         if isinstance(datasets, SyftError):
             return datasets
+        for dataset in datasets:
+            if dataset.to_be_deleted:
+                datasets.remove(dataset)
         return [
             asset
             for dataset in datasets
@@ -197,19 +211,61 @@ def get_assets_by_action_id(
         ]
 
     @service_method(
-        path="dataset.delete_by_uid",
-        name="delete_by_uid",
+        path="dataset.delete",
+        name="delete",
         roles=DATA_OWNER_ROLE_LEVEL,
         warning=HighSideCRUDWarning(confirmation=True),
     )
-    def delete_dataset(
-        self, context: AuthedServiceContext, uid: UID
+    def delete(
+        self, context: AuthedServiceContext, uid: UID, delete_assets: bool = True
     ) -> SyftSuccess | SyftError:
-        result = self.stash.delete_by_uid(context.credentials, uid)
-        if result.is_ok():
-            return result.ok()
-        else:
+        """
+        Soft delete: keep the dataset object, only remove the blob store entries
+        After soft deleting a dataset, the user will not be able to
+        see it using the `datasets.get_all` endpoint.
+        Delete unique `dataset.name` key and leave UID, just rename it in case the
+        user upload a new dataset with the same name.
+        """
+        # check if the dataset exists
+        dataset = self.get_by_id(context=context, uid=uid)
+        if isinstance(dataset, SyftError):
+            return dataset
+
+        return_msg = []
+        if delete_assets:
+            # delete the dataset's assets
+            for asset in dataset.asset_list:
+                msg = (
+                    f"ActionObject {asset.action_id} "
+                    f"linked with Assset {asset.id} "
+                    f"in Dataset {uid}"
+                )
+
+                action_service = cast(
+                    ActionService, context.server.get_service(ActionService)
+                )
+                del_res: SyftSuccess | SyftError = action_service.delete(
+                    context=context, uid=asset.action_id, soft_delete=True
+                )
+
+                if isinstance(del_res, SyftError):
+                    del_msg = f"Failed to delete {msg}: {del_res.message}"
+                    logger.error(del_msg)
+                    return del_res
+
+                logger.info(f"Successfully deleted {msg}: {del_res.message}")
+
+                return_msg.append(f"Asset with id '{asset.id}' successfully deleted.")
+
+        # soft delete the dataset object from the store
+        dataset_update = DatasetUpdate(
+            id=uid, name=f"_deleted_{dataset.name}_{uid}", to_be_deleted=True
+        )
+        result = self.stash.update(context.credentials, dataset_update)
+        if result.is_err():
             return SyftError(message=result.err())
+        return_msg.append(f"Dataset with id '{uid}' successfully deleted.")
+        return SyftSuccess(message="\n".join(return_msg))
 
 
 TYPE_TO_SERVICE[Dataset] = DatasetService
diff --git a/packages/syft/src/syft/store/kv_document_store.py b/packages/syft/src/syft/store/kv_document_store.py
index d0f5c22d7df..9935fe646dd 100644
--- a/packages/syft/src/syft/store/kv_document_store.py
+++ b/packages/syft/src/syft/store/kv_document_store.py
@@ -377,8 +377,16 @@ def _remove_keys(
         for qk in sqks:
             pk_key, pk_value = qk.key, qk.value
             ck_col = self.searchable_keys[pk_key]
-            if pk_value in ck_col and (store_key.value in ck_col[pk_value]):
-                ck_col[pk_value].remove(store_key.value)
+            if isinstance(pk_value, list):
+                for pk_value_item in pk_value:
+                    pk_value_str = str(pk_value_item)
+                    if pk_value_str in ck_col and (
+                        store_key.value in ck_col[pk_value_str]
+                    ):
+                        ck_col[pk_value_str].remove(store_key.value)
+            else:
+                if pk_value in ck_col and (store_key.value in ck_col[pk_value]):
+                    ck_col[pk_value].remove(store_key.value)
             self.searchable_keys[pk_key] = ck_col
 
     def _find_index_or_search_keys(
@@ -547,9 +555,13 @@ def _delete_unique_keys_for(self, obj: SyftObject) -> Result[SyftSuccess, str]:
 
     def _delete_search_keys_for(self, obj: SyftObject) -> Result[SyftSuccess, str]:
         for _search_ck in self.searchable_cks:
-            qk = _search_ck.with_obj(obj)
-            search_keys = self.searchable_keys[qk.key]
-            search_keys.pop(qk.value, None)
+            qk: QueryKey = _search_ck.with_obj(obj)
+            search_keys: defaultdict = self.searchable_keys[qk.key]
+            if isinstance(qk.value, list):
+                for qk_value in qk.value:
+                    search_keys.pop(qk_value, None)
+            else:
+                search_keys.pop(qk.value, None)
             self.searchable_keys[qk.key] = search_keys
         return Ok(SyftSuccess(message="Deleted"))
 
diff --git a/packages/syft/tests/conftest.py b/packages/syft/tests/conftest.py
index dcca9d8f113..7deec7e632c 100644
--- a/packages/syft/tests/conftest.py
+++ b/packages/syft/tests/conftest.py
@@ -10,10 +10,12 @@
 
 # third party
 from faker import Faker
+import numpy as np
 import pytest
 
 # syft absolute
 import syft as sy
+from syft import Dataset
 from syft.abstract_server import ServerSideType
 from syft.client.datasite_client import DatasiteClient
 from syft.protocol.data_protocol import get_data_protocol
@@ -251,6 +253,39 @@ def patched_user(monkeypatch):
     )
 
 
+@pytest.fixture
+def small_dataset() -> Dataset:
+    dataset = Dataset(
+        name="small_dataset",
+        asset_list=[
+            sy.Asset(
+                name="small_dataset",
+                data=np.array([1, 2, 3]),
+                mock=np.array([1, 1, 1]),
+            )
+        ],
+    )
+    yield dataset
+
+
+@pytest.fixture
+def big_dataset() -> Dataset:
+    num_elements = 20 * 1024 * 1024
+    data_big = np.random.randint(0, 100, size=num_elements)
+    mock_big = np.random.randint(0, 100, size=num_elements)
+    dataset = Dataset(
+        name="big_dataset",
+        asset_list=[
+            sy.Asset(
+                name="big_dataset",
+                data=data_big,
+                mock=mock_big,
+            )
+        ],
+    )
+    yield dataset
+
+
 __all__ = [
     "mongo_store_partition",
     "mongo_document_store",
diff --git a/packages/syft/tests/syft/blob_storage/blob_storage_test.py b/packages/syft/tests/syft/blob_storage/blob_storage_test.py
index a2728c6803e..ebcaf0235eb 100644
--- a/packages/syft/tests/syft/blob_storage/blob_storage_test.py
+++ b/packages/syft/tests/syft/blob_storage/blob_storage_test.py
@@ -9,6 +9,8 @@
 # syft absolute
 import syft as sy
 from syft import ActionObject
+from syft import Dataset
+from syft import Worker
 from syft.client.datasite_client import DatasiteClient
 from syft.service.blob_storage.util import can_upload_to_blob_storage
 from syft.service.blob_storage.util import min_size_for_blob_storage_upload
@@ -117,7 +119,7 @@ def test_action_obj_send_save_to_blob_storage(worker):
 
     # big object that should be saved to blob storage
     assert min_size_for_blob_storage_upload(root_client.api.metadata) == 16
-    num_elements = 50 * 1024 * 1024
+    num_elements = 20 * 1024 * 1024
     data_big = np.random.randint(0, 100, size=num_elements)  # 4 bytes per int32
     action_obj_2 = ActionObject.from_obj(data_big)
     assert can_upload_to_blob_storage(action_obj_2, root_client.api.metadata)
@@ -135,44 +137,14 @@ def test_action_obj_send_save_to_blob_storage(worker):
     assert all(syft_retrieved_data.read() == data_big)
 
 
-def test_upload_dataset_save_to_blob_storage(worker):
+def test_upload_dataset_save_to_blob_storage(
+    worker: Worker, big_dataset: Dataset, small_dataset: Dataset
+) -> None:
     root_client: DatasiteClient = worker.root_client
-    root_authed_ctx = AuthedServiceContext(
-        server=worker, credentials=root_client.verify_key
-    )
-    dataset = sy.Dataset(
-        name="small_dataset",
-        asset_list=[
-            sy.Asset(
-                name="small_dataset",
-                data=np.array([1, 2, 3]),
-                mock=np.array([1, 1, 1]),
-            )
-        ],
-    )
-    root_client.upload_dataset(dataset)
-    blob_storage = worker.get_service("BlobStorageService")
-    assert len(blob_storage.get_all_blob_storage_entries(context=root_authed_ctx)) == 0
-
-    num_elements = 50 * 1024 * 1024
-    data_big = np.random.randint(0, 100, size=num_elements)
-    dataset_big = sy.Dataset(
-        name="big_dataset",
-        asset_list=[
-            sy.Asset(
-                name="big_dataset",
-                data=data_big,
-                mock=np.array([1, 1, 1]),
-            )
-        ],
-    )
-    root_client.upload_dataset(dataset_big)
-    # the private data should be saved to the blob storage
-    blob_entries: list = blob_storage.get_all_blob_storage_entries(
-        context=root_authed_ctx
-    )
-    assert len(blob_entries) == 1
-    data_big_retrieved: SyftObjectRetrieval = blob_storage.read(
-        context=root_authed_ctx, uid=blob_entries[0].id
-    )
-    assert all(data_big_retrieved.read() == data_big)
+    # the small dataset should not be saved to the blob storage
+    root_client.upload_dataset(small_dataset)
+    assert len(root_client.api.services.blob_storage.get_all()) == 0
+
+    # the big dataset should be saved to the blob storage
+    root_client.upload_dataset(big_dataset)
+    assert len(root_client.api.services.blob_storage.get_all()) == 2
diff --git a/packages/syft/tests/syft/service/action/action_object_test.py b/packages/syft/tests/syft/service/action/action_object_test.py
index fa0e2695542..da3ea135f5b 100644
--- a/packages/syft/tests/syft/service/action/action_object_test.py
+++ b/packages/syft/tests/syft/service/action/action_object_test.py
@@ -23,6 +23,9 @@
 from syft.service.action.action_object import propagate_server_uid
 from syft.service.action.action_object import send_action_side_effect
 from syft.service.action.action_types import action_type_for_type
+from syft.service.response import SyftError
+from syft.service.response import SyftSuccess
+from syft.store.blob_storage import SyftObjectRetrieval
 from syft.types.uid import LineageID
 from syft.types.uid import UID
 
@@ -1023,3 +1026,35 @@ def test_actionobject_syft_getattr_pandas(worker):
 
     obj.columns = ["a", "b", "c"]
     assert (obj.columns == ["a", "b", "c"]).all()
+
+
+def test_actionobject_delete(worker):
+    """
+    Test deleting action objects and their corresponding blob storage entries
+    """
+    root_client = worker.root_client
+
+    # small object with no blob store entry
+    data_small = np.random.randint(0, 100, size=3)
+    action_obj = ActionObject.from_obj(data_small)
+    action_obj.send(root_client)
+    assert action_obj.syft_blob_storage_entry_id is None
+    del_res = root_client.api.services.action.delete(uid=action_obj.id)
+    assert isinstance(del_res, SyftSuccess)
+
+    # big object with blob store entry
+    num_elements = 25 * 1024 * 1024
+    data_big = np.random.randint(0, 100, size=num_elements)  # 4 bytes per int32
+    action_obj_2 = ActionObject.from_obj(data_big)
+    action_obj_2.send(root_client)
+    assert isinstance(action_obj_2.syft_blob_storage_entry_id, UID)
+    read_res = root_client.api.services.blob_storage.read(
+        action_obj_2.syft_blob_storage_entry_id
+    )
+    assert isinstance(read_res, SyftObjectRetrieval)
+    del_res = root_client.api.services.action.delete(uid=action_obj_2.id)
+    assert isinstance(del_res, SyftSuccess)
+    read_res = root_client.api.services.blob_storage.read(
+        action_obj_2.syft_blob_storage_entry_id
+    )
+    assert isinstance(read_res, SyftError)
diff --git a/packages/syft/tests/syft/service/dataset/dataset_service_test.py b/packages/syft/tests/syft/service/dataset/dataset_service_test.py
index f9dc3045643..5fcc2a29fa1 100644
--- a/packages/syft/tests/syft/service/dataset/dataset_service_test.py
+++ b/packages/syft/tests/syft/service/dataset/dataset_service_test.py
@@ -14,13 +14,14 @@
 import syft as sy
 from syft.server.worker import Worker
 from syft.service.action.action_object import ActionObject
+from syft.service.action.action_object import TwinMode
+from syft.service.blob_storage.util import can_upload_to_blob_storage
 from syft.service.dataset.dataset import CreateAsset as Asset
 from syft.service.dataset.dataset import CreateDataset as Dataset
 from syft.service.dataset.dataset import _ASSET_WITH_NONE_MOCK_ERROR_MESSAGE
 from syft.service.response import SyftError
 from syft.service.response import SyftException
 from syft.service.response import SyftSuccess
-from syft.types.twin_object import TwinMode
 
 
 def random_hash() -> str:
@@ -297,3 +298,76 @@ def test_upload_dataset_with_assets_of_different_data_types(
     assert type(root_datasite_client.datasets[0].assets[0].mock) == type(
         different_data_types
     )
+
+
+def test_delete_small_datasets(worker: Worker, small_dataset: Dataset) -> None:
+    root_client = worker.root_client
+    assert not can_upload_to_blob_storage(small_dataset, root_client.metadata)
+    upload_res = root_client.upload_dataset(small_dataset)
+    assert isinstance(upload_res, SyftSuccess)
+
+    dataset = root_client.api.services.dataset.get_all()[0]
+    asset = dataset.asset_list[0]
+    assert isinstance(asset.data, np.ndarray)
+    assert isinstance(asset.mock, np.ndarray)
+
+    # delete the dataset without deleting its assets
+    del_res = root_client.api.services.dataset.delete(
+        uid=dataset.id, delete_assets=False
+    )
+    assert isinstance(del_res, SyftSuccess)
+    assert isinstance(asset.data, np.ndarray)
+    assert isinstance(asset.mock, np.ndarray)
+    assert len(root_client.api.services.dataset.get_all()) == 0
+    # we can still get back the deleted dataset by uid
+    deleted_dataset = root_client.api.services.dataset.get_by_id(uid=dataset.id)
+    assert deleted_dataset.name == f"_deleted_{dataset.name}_{dataset.id}"
+    assert deleted_dataset.to_be_deleted
+
+    # delete the dataset and its assets
+    del_res = root_client.api.services.dataset.delete(
+        uid=dataset.id, delete_assets=True
+    )
+    assert isinstance(del_res, SyftSuccess)
+    assert asset.data is None
+    assert isinstance(asset.mock, SyftError)
+    assert len(root_client.api.services.dataset.get_all()) == 0
+
+
+def test_delete_big_datasets(worker: Worker, big_dataset: Dataset) -> None:
+    root_client = worker.root_client
+    assert can_upload_to_blob_storage(big_dataset, root_client.metadata)
+    upload_res = root_client.upload_dataset(big_dataset)
+    assert isinstance(upload_res, SyftSuccess)
+
+    dataset = root_client.api.services.dataset.get_all()[0]
+    asset = dataset.asset_list[0]
+    assert isinstance(asset.data, np.ndarray)
+    assert isinstance(asset.mock, np.ndarray)
+    # test that the data is saved in the blob storage
+    assert len(root_client.api.services.blob_storage.get_all()) == 2
+
+    # delete the dataset without deleting its assets
+    del_res = root_client.api.services.dataset.delete(
+        uid=dataset.id, delete_assets=False
+    )
+    assert isinstance(del_res, SyftSuccess)
+    assert isinstance(asset.data, np.ndarray)
+    assert isinstance(asset.mock, np.ndarray)
+    assert len(root_client.api.services.dataset.get_all()) == 0
+    # we can still get back the deleted dataset by uid
+    deleted_dataset = root_client.api.services.dataset.get_by_id(uid=dataset.id)
+    assert deleted_dataset.name == f"_deleted_{dataset.name}_{dataset.id}"
+    assert deleted_dataset.to_be_deleted
+    # the dataset's blob entries are still there
+    assert len(root_client.api.services.blob_storage.get_all()) == 2
+
+    # delete the dataset
+    del_res = root_client.api.services.dataset.delete(
+        uid=dataset.id, delete_assets=True
+    )
+    assert isinstance(del_res, SyftSuccess)
+    assert asset.data is None
+    assert isinstance(asset.mock, SyftError)
+    assert len(root_client.api.services.blob_storage.get_all()) == 0
+    assert len(root_client.api.services.dataset.get_all()) == 0
diff --git a/packages/syft/tests/syft/service/sync/sync_resolve_single_test.py b/packages/syft/tests/syft/service/sync/sync_resolve_single_test.py
index 33427bfe765..8aa8103ef0f 100644
--- a/packages/syft/tests/syft/service/sync/sync_resolve_single_test.py
+++ b/packages/syft/tests/syft/service/sync/sync_resolve_single_test.py
@@ -10,6 +10,7 @@
 from syft.client.sync_decision import SyncDecision
 from syft.client.syncing import compare_clients
 from syft.client.syncing import resolve
+from syft.server.worker import Worker
 from syft.service.job.job_stash import Job
 from syft.service.request.request import RequestStatus
 from syft.service.response import SyftError
@@ -135,13 +136,12 @@ def compute() -> int:
     assert res == compute(syft_no_server=True)
 
 
-def test_diff_state_with_dataset(low_worker, high_worker):
+def test_diff_state_with_dataset(low_worker: Worker, high_worker: Worker):
     low_client: DatasiteClient = low_worker.root_client
     client_low_ds = get_ds_client(low_client)
     high_client: DatasiteClient = high_worker.root_client
 
     _ = create_dataset(high_client)
-    _ = create_dataset(low_client)
 
     @sy.syft_function_single_use()
     def compute_mean(data) -> int:
@@ -150,7 +150,9 @@ def compute_mean(data) -> int:
     _ = client_low_ds.code.request_code_execution(compute_mean)
 
     result = client_low_ds.code.compute_mean(blocking=False)
-    assert isinstance(result, SyftError), "DS cannot start a job on low side"
+    assert isinstance(
+        result, SyftError
+    ), "DS cannot start a job on low side since the data was not passed as an argument"
 
     diff_state_before, diff_state_after = compare_and_resolve(
         from_client=low_client, to_client=high_client
@@ -162,8 +164,14 @@ def compute_mean(data) -> int:
 
     # run_and_deposit_result(high_client)
     data_high = high_client.datasets[0].assets[0]
-    result = high_client.code.compute_mean(data=data_high, blocking=True)
-    high_client.requests[0].deposit_result(result)
+    mean_result = high_client.code.compute_mean(data=data_high, blocking=True)
+    high_client.requests[0].deposit_result(mean_result)
+
+    # the high side client delete the dataset after depositing the result
+    dataset_del_res = high_client.api.services.dataset.delete(
+        uid=high_client.datasets[0].id
+    )
+    assert isinstance(dataset_del_res, SyftSuccess)
 
     diff_state_before, diff_state_after = compare_and_resolve(
         from_client=high_client, to_client=low_client
@@ -182,11 +190,7 @@ def compute_mean(data) -> int:
     res_non_blocking = client_low_ds.code.compute_mean(blocking=False).wait()
 
     # expected_result = compute_mean(syft_no_server=True, data=)
-    assert (
-        res_blocking
-        == res_non_blocking
-        == high_client.datasets[0].assets[0].data.mean()
-    )
+    assert res_blocking == res_non_blocking == mean_result
 
 
 def test_sync_with_error(low_worker, high_worker):
diff --git a/tests/integration/network/gateway_test.py b/tests/integration/network/gateway_test.py
index 28148cfe9c8..39ddc0bad5d 100644
--- a/tests/integration/network/gateway_test.py
+++ b/tests/integration/network/gateway_test.py
@@ -222,14 +222,14 @@ def test_dataset_search(set_env_var, gateway_port: int, datasite_1_port: int) ->
     assert isinstance(dataset, Dataset)
     assert len(dataset.assets) == 1
     assert isinstance(dataset.assets[0].mock, np.ndarray)
-    assert dataset.assets[0].data is None
+    assert isinstance(dataset.assets[0].data, SyftError)
 
     # search a wrong dataset should return an empty list
     wrong_search = sy.search(_random_hash())
     assert len(wrong_search) == 0
 
     # the datasite client delete the dataset
-    datasite_client.api.services.dataset.delete_by_uid(uid=dataset.id)
+    datasite_client.api.services.dataset.delete(uid=dataset.id)
 
     # Remove existing peers
     assert isinstance(_remove_existing_peers(datasite_client), SyftSuccess)
@@ -311,7 +311,7 @@ def mock_function(asset):
     assert (final_result == input_data + 1).all()
 
     # the datasite client delete the dataset
-    datasite_client.api.services.dataset.delete_by_uid(uid=dataset.id)
+    datasite_client.api.services.dataset.delete(uid=dataset.id)
 
     # Remove existing peers
     assert isinstance(_remove_existing_peers(datasite_client), SyftSuccess)
@@ -842,7 +842,7 @@ def test_dataset_stream(set_env_var, gateway_port: int, datasite_1_port: int) ->
     assert np.all(retrieved_asset.data == input_data)
 
     # the datasite client delete the dataset
-    datasite_client.api.services.dataset.delete_by_uid(uid=retrieved_dataset.id)
+    datasite_client.api.services.dataset.delete(uid=retrieved_dataset.id)
 
     # Remove existing peers
     assert isinstance(_remove_existing_peers(datasite_client), SyftSuccess)