From 64b0684f67442c031b426a22ac50e7fb6510d200 Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Thu, 15 Oct 2026 16:30:56 +0800 Subject: [PATCH] fix(storage): make TTL deletion error propagation opt-in --- .../storage/vectordb/collection/collection.py | 23 +++++- .../collection/volcengine_collection.py | 10 ++- openviking/storage/vectordb_adapters/base.py | 10 ++- .../vectordb_adapters/opengauss_adapter.py | 1 + openviking/storage/viking_fs/_ops.py | 11 ++- openviking/storage/viking_fs/_vector.py | 10 ++- .../storage/viking_vector_index_backend.py | 14 +++- tests/storage/test_transfer_merge_binding.py | 77 ++++++++++++++++++- tests/storage/test_volcengine_clients.py | 50 ------------ 9 files changed, 140 insertions(+), 66 deletions(-) diff --git a/openviking/storage/vectordb/collection/collection.py b/openviking/storage/vectordb/collection/collection.py index 8840736efe..2495878800 100644 --- a/openviking/storage/vectordb/collection/collection.py +++ b/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): diff --git a/openviking/storage/vectordb/collection/volcengine_collection.py b/openviking/storage/vectordb/collection/volcengine_collection.py index 7ff1b569ab..e53e2e6625 100644 --- a/openviking/storage/vectordb/collection/volcengine_collection.py +++ b/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( diff --git a/openviking/storage/vectordb_adapters/base.py b/openviking/storage/vectordb_adapters/base.py index 7563e8bc3f..1665d69eb3 100644 --- a/openviking/storage/vectordb_adapters/base.py +++ b/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 diff --git a/openviking/storage/vectordb_adapters/opengauss_adapter.py b/openviking/storage/vectordb_adapters/opengauss_adapter.py index af6121b767..7120565e5b 100644 --- a/openviking/storage/vectordb_adapters/opengauss_adapter.py +++ b/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) diff --git a/openviking/storage/viking_fs/_ops.py b/openviking/storage/viking_fs/_ops.py index 214829b218..a7746b8ab1 100644 --- a/openviking/storage/viking_fs/_ops.py +++ b/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, diff --git a/openviking/storage/viking_fs/_vector.py b/openviking/storage/viking_fs/_vector.py index 532dc27a39..d023f34916 100644 --- a/openviking/storage/viking_fs/_vector.py +++ b/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: diff --git a/openviking/storage/viking_vector_index_backend.py b/openviking/storage/viking_vector_index_backend.py index 067d61d1ab..ff9f61ed36 100644 --- a/openviking/storage/viking_vector_index_backend.py +++ b/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,7 +2553,9 @@ 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), @@ -2559,7 +2563,9 @@ async def delete_uris(self, ctx: RequestContext, uris: List[str]) -> None: ] 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)] diff --git a/tests/storage/test_transfer_merge_binding.py b/tests/storage/test_transfer_merge_binding.py index c011a54005..770124d86a 100644 --- a/tests/storage/test_transfer_merge_binding.py +++ b/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 diff --git a/tests/storage/test_volcengine_clients.py b/tests/storage/test_volcengine_clients.py index b5fb1fc133..f83a974b2b 100644 --- a/tests/storage/test_volcengine_clients.py +++ b/tests/storage/test_volcengine_clients.py @@ -326,56 +326,6 @@ def _fake_do_req(method, path=None, req_params=None, req_body=None): } -@pytest.mark.parametrize("failure_path", ["/search/scalar", "/delete"]) -@pytest.mark.parametrize("response_kind", [429, 500, "invalid_json", "success"]) -def test_volcengine_delete_propagates_request_failures(monkeypatch, failure_path, response_kind): - collection = VolcengineCollection( - ak="test-ak", - sk="test-sk", - region="cn-beijing", - meta_data={"ProjectName": "default", "CollectionName": "context"}, - ) - adapter = VolcengineCollectionAdapter.from_config( - VectorDBBackendConfig( - backend="volcengine", - name="context", - volcengine=VolcengineConfig(ak="test-ak", sk="test-sk", region="cn-beijing"), - ) - ) - adapter._collection = Collection(collection) - deleted_ids = [] - - 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":[]}}' - ) - else: - assert path.endswith("/delete") - deleted_ids.extend(req_body["ids"]) - response._content = b'{"result":{}}' - 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) - scope = {"op": "must", "field": "uri", "conds": ["/resources/expired"]} - if response_kind == "success": - assert adapter.delete(filter=scope) == 1 - assert deleted_ids == ["doc-1"] - else: - message = "invalid JSON" if response_kind == "invalid_json" else str(response_kind) - with pytest.raises(ConnectionError, match=message): - adapter.delete(filter=scope) - if failure_path == "/search/scalar": - assert deleted_ids == [] - - def test_volcengine_collection_update_data_sanitizes_uri_fields(monkeypatch): captured = {}