fix(storage): make TTL deletion error propagation opt-in

openviking/storage/vectordb/collection/collection.py

@@ -544,6 +544,8 @@ def search_by_scalar(
         offset: int = 0,
         filters: Optional[Dict[str, Any]] = None,
         output_fields: Optional[List[str]] = None,
+        *,
+        raise_on_error: bool = False,
     ):
         """Retrieve documents sorted by a scalar field value.
 
@@ -567,6 +569,20 @@ def search_by_scalar(
         """
         if self.__collection is None:
             raise RuntimeError("Collection is closed")
+        if raise_on_error:
+            from .volcengine_collection import VolcengineCollection
+
+            if isinstance(self.__collection, VolcengineCollection):
+                return self.__collection.search_by_scalar(
+                    index_name,
+                    field,
+                    order,
+                    limit,
+                    offset,
+                    filters,
+                    output_fields,
+                    raise_on_error=True,
+                )
         return self.__collection.search_by_scalar(
             index_name, field, order, limit, offset, filters, output_fields
         )
@@ -674,7 +690,7 @@ def fetch_data(self, primary_keys: List[Any]) -> List[Dict[str, Any]]:
             raise RuntimeError("Collection is closed")
         return self.__collection.fetch_data(primary_keys)
 
-    def delete_data(self, primary_keys: List[Any]):
+    def delete_data(self, primary_keys: List[Any], *, raise_on_error: bool = False):
         """
         Delete specific data documents from the collection using their primary keys.
 
@@ -683,6 +699,11 @@ def delete_data(self, primary_keys: List[Any]):
         """
         if self.__collection is None:
             raise RuntimeError("Collection is closed")
+        if raise_on_error:
+            from .volcengine_collection import VolcengineCollection
+
+            if isinstance(self.__collection, VolcengineCollection):
+                return self.__collection.delete_data(primary_keys, raise_on_error=True)
         return self.__collection.delete_data(primary_keys)
 
     def delete_all_data(self):

openviking/storage/vectordb/collection/volcengine_collection.py

@@ -446,14 +446,14 @@ def fetch_data(self, primary_keys: List[Any]) -> FetchDataInCollectionResult:
         # print(resp_data)
         return self._parse_fetch_result(resp_data)
 
-    def delete_data(self, primary_keys: List[Any]):
+    def delete_data(self, primary_keys: List[Any], *, raise_on_error: bool = False):
         path = "/api/vikingdb/data/delete"
         data = {
             "project": self.project_name,
             "collection_name": self.collection_name,
             "ids": primary_keys,
         }
-        return self._data_post(path, data, raise_for_status=True)
+        return self._data_post(path, data, raise_for_status=raise_on_error)
 
     def delete_all_data(self):
         path = "/api/vikingdb/data/delete"
@@ -462,7 +462,7 @@ def delete_all_data(self):
             "collection_name": self.collection_name,
             "del_all": True,
         }
-        return self._data_post(path, data, raise_for_status=True)
+        return self._data_post(path, data)
 
     def _parse_fetch_result(self, data: Dict[str, Any]) -> FetchDataInCollectionResult:
         result = FetchDataInCollectionResult()
@@ -633,6 +633,8 @@ def search_by_scalar(
         offset: int = 0,
         filters: Optional[Dict[str, Any]] = None,
         output_fields: Optional[List[str]] = None,
+        *,
+        raise_on_error: bool = False,
     ) -> SearchResult:
         path = "/api/vikingdb/data/search/scalar"
         data = {
@@ -647,7 +649,7 @@ def search_by_scalar(
             "offset": offset,
             "ignore_unknown_fields": True,
         }
-        resp_data = self._data_post(path, data, raise_for_status=True)
+        resp_data = self._data_post(path, data, raise_for_status=raise_on_error)
         return self._parse_search_result(resp_data)
 
     def _parse_aggregate_result(

openviking/storage/vectordb_adapters/base.py

@@ -485,6 +485,7 @@ def query(
         order_by: Optional[str] = None,
         order_desc: bool = False,
         advance: Optional[Dict[str, Any]] = None,
+        raise_on_error: bool = False,
     ) -> list[Dict[str, Any]]:
         decay_request = (advance or {}).get("time_decay")
         if decay_request is not None and self.mode != "opengauss":
@@ -531,6 +532,7 @@ def query(
                 return_detail_info=decay_request is not None,
             )
         elif order_by:
+            error_options: Dict[str, Any] = {"raise_on_error": True} if raise_on_error else {}
             result = coll.search_by_scalar(
                 index_name=self._index_name,
                 field=order_by,
@@ -539,6 +541,7 @@ def query(
                 offset=offset,
                 filters=vectordb_filter,
                 output_fields=output_fields,
+                **error_options,
             )
         else:
             # Approximate random sampling with a client-generated random
@@ -611,13 +614,15 @@ def delete(
         *,
         ids: Optional[list[str]] = None,
         filter: Optional[Dict[str, Any] | FilterExpr] = None,
+        raise_on_error: bool = False,
     ) -> int:
         """Submit IDs for deletion in batches, without waiting for index visibility."""
         coll = self.get_collection()
+        error_options: Dict[str, Any] = {"raise_on_error": True} if raise_on_error else {}
         batch_size = self._DATA_BATCH_SIZE or 100
         if ids is not None:
             for start in range(0, len(ids), batch_size):
-                coll.delete_data(ids[start : start + batch_size])
+                coll.delete_data(ids[start : start + batch_size], **error_options)
             return len(ids)
         if filter is None:
             return 0
@@ -634,6 +639,7 @@ def delete(
                     output_fields=["id"],
                     order_by="updated_at",
                     order_desc=False,
+                    **error_options,
                 )
                 if not matched:
                     break
@@ -642,7 +648,7 @@ def delete(
 
             pending_ids.seek(0)
             for batch in pending_ids:
-                coll.delete_data(json.loads(batch))
+                coll.delete_data(json.loads(batch), **error_options)
             return offset
 
     @staticmethod

openviking/storage/vectordb_adapters/opengauss_adapter.py

@@ -274,6 +274,7 @@ def delete(
         ids: Optional[list[str]] = None,
         filter: Optional[Dict[str, Any] | FilterExpr] = None,
         limit: Optional[int] = None,
+        raise_on_error: bool = False,
     ) -> int:
         if ids is not None or filter is None:
             return super().delete(ids=ids, filter=filter)

openviking/storage/viking_fs/_ops.py

@@ -219,6 +219,8 @@ async def rm(
         ctx: Optional[RequestContext] = None,
         lease_ref: Dict[str, Any] | None = None,
         auto_pathlock: bool = True,
+        *,
+        raise_on_error: bool = False,
     ) -> Dict[str, Any]:
         """Delete file/directory + recursively update vector index.
 
@@ -235,6 +237,10 @@ async def rm(
         target is not concurrently mutated (e.g. best-effort shared upload
         cleanup that only deletes already-expired directories).
 
+        Internal TTL callers can set ``raise_on_error=True`` to propagate
+        Volcengine request failures without waiting for index visibility.
+        The default preserves the existing backend error handling.
+
         Returns:
             Dict with 'estimated_deleted_count' indicating the estimated number
             of nodes deleted from vector index.
@@ -246,6 +252,7 @@ async def rm(
         await self._ensure_access(uri, ctx, action=AclAction.WRITE)
         path = self._uri_to_path(uri, ctx=ctx)
         target_uri = self._path_to_uri(path, ctx=ctx)
+        delete_options = {"raise_on_error": True} if raise_on_error else {}
 
         async def _estimate_deleted_count(target_path: str, real_ctx: RequestContext) -> int:
             """Estimate number of nodes to be deleted using vector index."""
@@ -277,7 +284,7 @@ async def _estimate_deleted_count(target_path: str, real_ctx: RequestContext) ->
             uris_to_delete.append(target_uri)
             real_ctx = self._ctx_or_default(ctx)
             estimated_count = await _estimate_deleted_count(path, real_ctx)
-            await self._delete_from_vector_store(uris_to_delete, ctx=ctx)
+            await self._delete_from_vector_store(uris_to_delete, ctx=ctx, **delete_options)
             logger.info(f"[VikingFS] rm target not found, cleaned orphan index: {uri}")
             return {"estimated_deleted_count": estimated_count}
 
@@ -320,7 +327,7 @@ async def _estimate_deleted_count(target_path: str, real_ctx: RequestContext) ->
                 await self._ensure_access_many(uris_to_delete, ctx, action=AclAction.MANAGE)
             real_ctx = self._ctx_or_default(ctx)
             estimated_count = await _estimate_deleted_count(path, real_ctx)
-            await self._delete_from_vector_store(uris_to_delete, ctx=ctx)
+            await self._delete_from_vector_store(uris_to_delete, ctx=ctx, **delete_options)
             try:
                 result = await self._async_agfs.rm(
                     path,

openviking/storage/viking_fs/_vector.py

@@ -16,7 +16,11 @@ class _VectorMixin:
     """Vector store integration: delete/update URIs, get store/embedder."""
 
     async def _delete_from_vector_store(
-        self, uris: List[str], ctx: Optional[RequestContext] = None
+        self,
+        uris: List[str],
+        ctx: Optional[RequestContext] = None,
+        *,
+        raise_on_error: bool = False,
     ) -> None:
         """Delete records with specified URIs from vector store.
 
@@ -28,7 +32,9 @@ async def _delete_from_vector_store(
         real_ctx = self._ctx_or_default(ctx)
 
         try:
-            await vector_store.delete_uris(real_ctx, uris)
+            await vector_store.delete_uris(
+                real_ctx, uris, **({"raise_on_error": True} if raise_on_error else {})
+            )
             for uri in uris:
                 logger.debug(f"[VikingFS] Deleted from vector store: {uri}")
         except Exception as e:

openviking/storage/viking_vector_index_backend.py

@@ -828,10 +828,12 @@ async def delete(self, ids: List[str]) -> int:
             raise
 
     @_backend_operation
-    async def delete_by_filter(self, filter: FilterExpr) -> int:
+    async def delete_by_filter(self, filter: FilterExpr, *, raise_on_error: bool = False) -> int:
         """Root-only: 直接通过 filter 删除"""
         try:
-            return await self._async_adapter.call("delete", filter=filter)
+            return await self._async_adapter.call(
+                "delete", filter=filter, **({"raise_on_error": True} if raise_on_error else {})
+            )
         except Exception as e:
             logger.error("Error deleting by filter: %s", e)
             raise
@@ -2551,15 +2553,19 @@ async def delete_user_data(
             And([Eq("account_id", account_id), Eq("owner_user_id", user_id)])
         )
 
-    async def delete_uris(self, ctx: RequestContext, uris: List[str]) -> None:
+    async def delete_uris(
+        self, ctx: RequestContext, uris: List[str], *, raise_on_error: bool = False
+    ) -> None:
         for uri in uris:
             conds: List[FilterExpr] = [
                 Eq("account_id", ctx.account_id),
                 Or([Eq("uri", uri), In("uri", [f"{uri}/"])]),
             ]
 
             backend = await self._get_backend_for_context(ctx)
-            await backend.delete_by_filter(And(conds))
+            await backend.delete_by_filter(
+                And(conds), **({"raise_on_error": True} if raise_on_error else {})
+            )
 
     def _uri_transfer_filter(self, ctx: RequestContext, uri: str, *, recursive: bool) -> FilterExpr:
         scopes: List[FilterExpr] = [Eq("uri", uri)]

tests/storage/test_transfer_merge_binding.py

@@ -8,20 +8,24 @@
 
 import pytest
 import pytest_asyncio
+import requests
 
 from openviking.pyagfs import get_binding_client
 from openviking.resource.watch_manager import WatchManager
 from openviking.server.identity import RequestContext, Role
 from openviking.storage.acl import AclAction, AclManager
 from openviking.storage.collection_schemas import CollectionSchemas
+from openviking.storage.errors import ConnectionError
 from openviking.storage.vector_ids import vector_record_id
 from openviking.storage.vectordb import engine as vectordb_engine
+from openviking.storage.vectordb.collection.collection import Collection
+from openviking.storage.vectordb.collection.volcengine_collection import VolcengineCollection
 from openviking.storage.viking_fs import VikingFS
 from openviking.utils.agfs_utils import RagfsBindingConfig, mount_agfs_backend
 from openviking_cli.exceptions import InvalidArgumentError, PermissionDeniedError
 from openviking_cli.session.user_id import UserIdentifier
 from openviking_cli.utils.config.agfs_config import AGFSConfig
-from openviking_cli.utils.config.vectordb_config import VectorDBBackendConfig
+from openviking_cli.utils.config.vectordb_config import VectorDBBackendConfig, VolcengineConfig
 
 
 class _EnabledAclConfig:
@@ -78,6 +82,76 @@ async def seed_vector(backend, uri, content):
     )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("raise_on_error", [False, True])
+@pytest.mark.parametrize("failure_path", ["/search/scalar", "/delete"])
+@pytest.mark.parametrize("response_kind", [429, 500, "invalid_json", "success"])
+async def test_rm_cloud_errors_are_opt_in(
+    binding_fs, vector_backend_factory, monkeypatch, raise_on_error, failure_path, response_kind
+):
+    fs, ctx = binding_fs, root_ctx()
+    uri = "viking://resources/expired.md"
+    await fs.write_file_bytes(uri, b"keep on failure", ctx=ctx)
+    backend = vector_backend_factory(
+        config=VectorDBBackendConfig(
+            backend="volcengine",
+            name="context",
+            volcengine=VolcengineConfig(ak="test-ak", sk="test-sk", region="cn-beijing"),
+        )
+    )
+    collection = VolcengineCollection(
+        ak="test-ak",
+        sk="test-sk",
+        region="cn-beijing",
+        meta_data={"ProjectName": "default", "CollectionName": "context"},
+    )
+    backend._shared_adapter._collection = Collection(collection)
+    fs.vector_store = backend
+
+    def fake_request(method, path, req_body):
+        response = requests.Response()
+        response.status_code = 200
+        if path.endswith("/search/scalar"):
+            response._content = (
+                b'{"result":{"data":[{"id":"doc-1"}]}}'
+                if req_body["offset"] == 0
+                else b'{"result":{"data":[]}}'
+            )
+        elif path.endswith("/delete"):
+            response._content = b'{"result":{}}'
+        else:
+            assert path.endswith("/agg")  # Existing pre-delete estimate only.
+            response._content = b'{"result":{"_total":0}}'
+        if path.endswith(failure_path) and response_kind != "success":
+            response.status_code = response_kind if isinstance(response_kind, int) else 200
+            response._content = b"invalid JSON" if response_kind == "invalid_json" else b"{}"
+        return response
+
+    monkeypatch.setattr(collection.data_client, "do_req", fake_request)
+    options = {"raise_on_error": True} if raise_on_error else {}
+    try:
+        if raise_on_error and response_kind != "success":
+            message = "invalid JSON" if response_kind == "invalid_json" else str(response_kind)
+            with pytest.raises(ConnectionError, match=message):
+                await fs.rm(uri, ctx=ctx, **options)
+            assert await fs.read_file_bytes(uri, ctx=ctx) == b"keep on failure"
+            await fs.remove_files(uri, ctx=ctx)
+            with pytest.raises(ConnectionError, match=message):
+                await fs.rm(uri, ctx=ctx, **options)
+        else:
+            await fs.rm(uri, ctx=ctx, **options)
+            assert not await fs.exists(uri, ctx=ctx)
+            await fs.rm(uri, ctx=ctx, **options)
+        # Opting in for deletion must not change the shared collection's defaults.
+        if response_kind != "success":
+            if failure_path == "/search/scalar":
+                assert collection.search_by_scalar("default", "updated_at").data == []
+            else:
+                assert collection.delete_all_data() == {}
+    finally:
+        await backend.close()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("operation", ["cp", "mv"])
 @pytest.mark.parametrize("directory", [False, True])
@@ -625,6 +699,7 @@ async def test_acl_failure_restores_only_moved_vectors_after_directory_merge(
     await seed_vector(backend, f"{source}/file.txt", "new")
     await seed_vector(backend, f"{target}/only.txt", "keep")
     monkeypatch.setattr(fs, "_ensure_access", AsyncMock())
+
     async def acl_enabled(_account_id):
         return True