fix(vespa): handle continuation tokens in selection-based bulk deletes (#1733)

* fix(vespa): propagate delete errors instead of swallowing them

* fix(vespa): use 'contains' instead of '=' for collection_id in fast-delete YQL

The fast-delete path constructs a YQL query to resolve Vespa doc IDs
before issuing parallel direct DELETEs. The query used `= '{uuid}'` for
the collection_id filter, but in Vespa YQL `=` is the numeric equality
operator — it fails on UUID strings with HTTP 400 "not an int item
expression: Illegal embedded sign character" (the `-` in UUIDs).

This caused every fast-delete to fall back to the visitor-based
selection scan (5 schemas × 120s timeout each ≈ 300s per batch),
producing 1,100+ "update_delete_VespaDestination slow: 300.Xs"
warnings per hour in production.

Fix: change `=` to `contains` (the correct YQL string-match operator),
consistent with query_builder.py and vespa_client.py which already
use `contains` for the same field.

* fix(vespa): handle continuation token in selection-based bulk deletes

Vespa visitor-based DELETE returns a continuation token when the
result set is too large for a single pass. The previous implementation
made a single request and ignored the token, silently leaving
undeleted documents behind on large collections.

Now delete_by_selection loops: issue DELETE, check for continuation
token in response, re-issue with continuation param until Vespa
stops returning one. This affects delete_by_sync_id,
delete_by_collection_id, and the entity-level fallback path.

* style: fix ruff format for bulk delete response assignment

Author: orhanrauf

backend/airweave/platform/destinations/vespa/client.py

@@ -188,8 +188,9 @@ async def delete_by_selection(self, schema: str, selection: str) -> DeleteResult
         """Delete documents using Vespa's selection-based bulk delete API (visitor scan).
 
         Uses a visitor that walks all buckets evaluating the selection expression.
-        Appropriate for broad deletions (by sync_id, collection_id) but slow for
-        targeted deletions -- use _delete_by_doc_ids for those.
+        For large result sets Vespa returns a continuation token; this method
+        loops until no more tokens are returned so all matching documents are
+        deleted.
 
         Uses: DELETE /document/v1/{namespace}/{doctype}/docid?selection={expr}&cluster={cluster}
 
@@ -200,26 +201,51 @@ async def delete_by_selection(self, schema: str, selection: str) -> DeleteResult
         Returns:
             DeleteResult with count of deleted documents
         """
-        url = self._build_bulk_delete_url(schema, selection)
+        base_url = self._build_bulk_delete_url(schema, selection)
         self._logger.debug(f"[VespaClient] Bulk delete from {schema} with selection: {selection}")
 
         deleted_count = 0
+        continuation: Optional[str] = None
+        pass_num = 0
+
         try:
             async with httpx.AsyncClient(timeout=settings.VESPA_TIMEOUT) as client:
-                async with client.stream("DELETE", url) as response:
-                    if response.status_code == 200:
-                        deleted_count = await self._parse_bulk_delete_response(response)
-                    else:
-                        await self._log_delete_error(response)
+                while True:
+                    pass_num += 1
+                    url = base_url
+                    if continuation:
+                        url += f"&continuation={quote(continuation, safe='')}"
+
+                    async with client.stream("DELETE", url) as response:
+                        if response.status_code == 200:
+                            batch_count, continuation = await self._parse_bulk_delete_response(
+                                response
+                            )
+                            deleted_count += batch_count
+                        else:
+                            await self._log_delete_error(response)
+                            break
+
+                    if not continuation:
+                        break
+
+                    self._logger.debug(
+                        f"[VespaClient] Bulk delete pass {pass_num}: "
+                        f"{deleted_count} deleted so far, continuing..."
+                    )
         except httpx.TimeoutException:
             self._logger.error(
-                f"[VespaClient] Bulk delete timed out after {settings.VESPA_TIMEOUT}s"
+                f"[VespaClient] Bulk delete timed out after {settings.VESPA_TIMEOUT}s "
+                f"(pass {pass_num}, {deleted_count} deleted before timeout)"
             )
         except Exception as e:
             self._logger.error(f"[VespaClient] Bulk delete error: {e}")
 
         if deleted_count > 0:
-            self._logger.info(f"[VespaClient] Deleted {deleted_count} documents from {schema}")
+            self._logger.info(
+                f"[VespaClient] Deleted {deleted_count} documents from {schema} "
+                f"in {pass_num} pass(es)"
+            )
         else:
             self._logger.debug(f"[VespaClient] No documents to delete from {schema}")
 
@@ -477,14 +503,24 @@ def _build_bulk_delete_url(self, schema: str, selection: str) -> str:
             f"&cluster={settings.VESPA_CLUSTER}"
         )
 
-    async def _parse_bulk_delete_response(self, response: httpx.Response) -> int:
-        """Parse streaming response from Vespa bulk delete."""
+    async def _parse_bulk_delete_response(
+        self, response: httpx.Response
+    ) -> Tuple[int, Optional[str]]:
+        """Parse streaming response from Vespa bulk delete.
+
+        Returns:
+            (deleted_count, continuation_token) — token is None when the
+            visitor has finished and there are no more documents to process.
+        """
         count = 0
+        continuation: Optional[str] = None
         async for line in response.aiter_lines():
             if not line.strip():
                 continue
             try:
                 result = json.loads(line)
+                if "continuation" in result:
+                    continuation = result["continuation"]
                 if "documentCount" in result:
                     count += result["documentCount"]
                 elif "sessionStats" in result:
@@ -494,7 +530,7 @@ async def _parse_bulk_delete_response(self, response: httpx.Response) -> int:
                     count += 1
             except json.JSONDecodeError:
                 pass
-        return count
+        return count, continuation
 
     async def _log_delete_error(self, response: httpx.Response) -> None:
         """Log error from failed bulk delete response."""

backend/tests/unit/platform/destinations/vespa/test_client.py

@@ -1,5 +1,9 @@
 """Unit tests for VespaClient (with mocked I/O)."""
 
+import json
+from contextlib import asynccontextmanager
+
+import httpx
 import pytest
 from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import UUID
@@ -123,6 +127,131 @@ async def test_delete_by_selection_builds_url_correctly(self, client):
         assert result.deleted_count == 5
         assert result.schema_name == schema
 
+    @pytest.mark.asyncio
+    async def test_delete_by_selection_follows_continuation_token(self, client):
+        """Test that delete_by_selection loops when Vespa returns a continuation token.
+
+        Vespa's visitor-based delete returns a continuation token for large result
+        sets. The client must re-issue the DELETE with the token until Vespa stops
+        returning one.
+        """
+        captured_urls: list[str] = []
+
+        def _make_stream_response(body_lines: list[str], status: int = 200):
+            """Build a fake async streaming response."""
+            @asynccontextmanager
+            async def _stream(method, url, **kwargs):
+                captured_urls.append(url)
+                resp = MagicMock()
+                resp.status_code = status
+
+                async def aiter_lines():
+                    for line in body_lines:
+                        yield line
+
+                resp.aiter_lines = aiter_lines
+                yield resp
+            return _stream
+
+        pass_1_body = [
+            json.dumps({"documentCount": 500, "continuation": "AAAABB=="}),
+        ]
+        pass_2_body = [
+            json.dumps({"documentCount": 300, "continuation": "CCCCDD=="}),
+        ]
+        pass_3_body = [
+            json.dumps({"documentCount": 200}),
+        ]
+
+        call_count = 0
+
+        @asynccontextmanager
+        async def mock_async_client(**kwargs):
+            class FakeClient:
+                def stream(self, method, url, **kw):
+                    nonlocal call_count
+                    call_count += 1
+                    if call_count == 1:
+                        return _make_stream_response(pass_1_body)(method, url)
+                    elif call_count == 2:
+                        return _make_stream_response(pass_2_body)(method, url)
+                    else:
+                        return _make_stream_response(pass_3_body)(method, url)
+            yield FakeClient()
+
+        with patch("httpx.AsyncClient", side_effect=lambda **kw: mock_async_client(**kw)):
+            result = await client.delete_by_selection("base_entity", "field=='value'")
+
+        assert result.deleted_count == 1000
+        assert call_count == 3
+        assert "continuation=" not in captured_urls[0]
+        assert "continuation=" in captured_urls[1]
+        assert "AAAABB" in captured_urls[1]
+        assert "continuation=" in captured_urls[2]
+        assert "CCCCDD" in captured_urls[2]
+
+    @pytest.mark.asyncio
+    async def test_delete_by_selection_no_continuation_single_pass(self, client):
+        """Test that delete_by_selection completes in one pass when no continuation token."""
+        call_count = 0
+
+        @asynccontextmanager
+        async def mock_async_client(**kwargs):
+            class FakeClient:
+                def stream(self, method, url, **kw):
+                    nonlocal call_count
+                    call_count += 1
+
+                    @asynccontextmanager
+                    async def _ctx(m, u):
+                        resp = MagicMock()
+                        resp.status_code = 200
+
+                        async def aiter_lines():
+                            yield json.dumps({"documentCount": 42})
+
+                        resp.aiter_lines = aiter_lines
+                        yield resp
+
+                    return _ctx(method, url)
+            yield FakeClient()
+
+        with patch("httpx.AsyncClient", side_effect=lambda **kw: mock_async_client(**kw)):
+            result = await client.delete_by_selection("base_entity", "field=='value'")
+
+        assert result.deleted_count == 42
+        assert call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_parse_bulk_delete_response_extracts_continuation(self, client):
+        """Test _parse_bulk_delete_response returns the continuation token."""
+        response = MagicMock()
+
+        async def aiter_lines():
+            yield json.dumps({"documentCount": 150, "continuation": "TOKEN123"})
+
+        response.aiter_lines = aiter_lines
+
+        count, token = await client._parse_bulk_delete_response(response)
+
+        assert count == 150
+        assert token == "TOKEN123"
+
+    @pytest.mark.asyncio
+    async def test_parse_bulk_delete_response_none_when_no_continuation(self, client):
+        """Test _parse_bulk_delete_response returns None when no continuation."""
+        response = MagicMock()
+
+        async def aiter_lines():
+            yield json.dumps({"documentCount": 50})
+
+        response.aiter_lines = aiter_lines
+
+        count, token = await client._parse_bulk_delete_response(response)
+
+        assert count == 50
+        assert token is None
+
     @pytest.mark.asyncio
     async def test_delete_by_sync_id(self, client):
         """Test delete by sync ID across all schemas."""