fix(grep): fall back to fs on empty VikingDB recall and query timeout (#2850)

openviking/storage/viking_fs.py

@@ -1086,6 +1086,14 @@ async def _grep_vikingdb_then_fs(
         # use the maximum limit to avoid truncation.
         remote_return_limit = min(node_limit * 5, 100000) if node_limit else 100000
 
+        # VikingDB BM25 can silently time out / return empty on large corpora
+        # (see #2850). Bound the query so a hung remote cannot stall grep; a
+        # timeout is treated as unreliability and falls back to fs, same as an
+        # raised exception.
+        vikingdb_timeout = float(
+            os.environ.get("OPENVIKING_GREP_VIKINGDB_TIMEOUT_SEC", "10")
+        )
+
         # Step 1: vikingdb recall candidate files
         try:
             logger.debug(
@@ -1096,11 +1104,31 @@ async def _grep_vikingdb_then_fs(
                 filter_expr,
                 ["uri"],
             )
-            result = await vector_store.search_by_keywords(
-                query=query,
-                limit=remote_return_limit,
-                filter=filter_expr,
-                output_fields=["uri"],
+            result = await asyncio.wait_for(
+                vector_store.search_by_keywords(
+                    query=query,
+                    limit=remote_return_limit,
+                    filter=filter_expr,
+                    output_fields=["uri"],
+                    ctx=ctx,
+                ),
+                timeout=vikingdb_timeout,
+            )
+        except asyncio.TimeoutError:
+            logger.warning(
+                "grep vikingdb query timed out (%.1fs) for pattern %r at %s, "
+                "falling back to fs",
+                vikingdb_timeout,
+                pattern,
+                uri,
+            )
+            return await self._grep_fs(
+                uri=uri,
+                pattern=pattern,
+                exclude_uri=exclude_uri,
+                case_insensitive=case_insensitive,
+                node_limit=node_limit,
+                level_limit=level_limit,
                 ctx=ctx,
             )
         except Exception as e:
@@ -1123,8 +1151,25 @@ async def _grep_vikingdb_then_fs(
                 if u != excluded_prefix and not u.startswith(excluded_prefix + "/")
             ]
         if not candidate_uris:
-            # BM25 returned no candidates — the index confirms no matching content
-            return {"matches": [], "count": 0, "match_count": 0, "files_scanned": 0}
+            # BM25 returned no candidates. On large/stale corpora this can be a
+            # false negative (silent timeout, index lag) rather than a true
+            # absence of matches (#2850). Retry with fs grep so a VikingDB gap
+            # does not masquerade as "no matching content".
+            logger.warning(
+                "grep vikingdb returned 0 candidates for pattern %r at %s, "
+                "retrying with fs grep",
+                pattern,
+                uri,
+            )
+            return await self._grep_fs(
+                uri=uri,
+                pattern=pattern,
+                exclude_uri=exclude_uri,
+                case_insensitive=case_insensitive,
+                node_limit=node_limit,
+                level_limit=level_limit,
+                ctx=ctx,
+            )
 
         # Step 2: local fs precise matching on candidate files
         return await self._grep_in_files(

tests/storage/test_viking_fs_grep.py

@@ -1,6 +1,7 @@
 # Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
 # SPDX-License-Identifier: AGPL-3.0
 
+import asyncio
 import time
 
 import pytest
@@ -147,13 +148,104 @@ async def fake_grep_fs(**kwargs):
     ]
 
 
+class _SlowVectorStore:
+    """Vector store whose search_by_keywords never returns within the timeout."""
+
+    async def search_by_keywords(self, **kwargs):
+        await asyncio.sleep(30)
+        return []
+
+
+@pytest.mark.asyncio
+async def test_grep_vikingdb_empty_recall_falls_back_to_fs(monkeypatch):
+    """VikingDB returning zero candidates must retry via fs grep (#2850)."""
+    fs = VikingFS(agfs=_DummyAgfs())
+    # _DummyVectorStore() defaults to empty results -> empty recall.
+    vector_store = _DummyVectorStore()
+    monkeypatch.setattr(fs, "_get_vector_store", lambda: vector_store)
+    monkeypatch.setattr(fs, "_ensure_access", lambda uri, ctx=None: None)
+
+    calls = []
+
+    async def fake_grep_fs(**kwargs):
+        calls.append(kwargs)
+        return {
+            "matches": [{"uri": "viking://resources/a.md"}],
+            "count": 1,
+            "match_count": 1,
+            "files_scanned": 1,
+        }
+
+    monkeypatch.setattr(fs, "_grep_fs", fake_grep_fs)
+
+    result = await fs._grep_vikingdb_then_fs(
+        uri="viking://resources",
+        pattern="needle",
+        exclude_uri=None,
+        case_insensitive=False,
+        node_limit=10,
+        level_limit=3,
+        ctx=None,
+    )
+
+    # fs fallback was invoked and its result surfaced (not the empty VikingDB hit).
+    assert len(calls) == 1
+    assert calls[0]["pattern"] == "needle"
+    assert result["count"] == 1
+    assert result["matches"][0]["uri"] == "viking://resources/a.md"
+
+
+@pytest.mark.asyncio
+async def test_grep_vikingdb_timeout_falls_back_to_fs(monkeypatch):
+    """A VikingDB query that exceeds the timeout must fall back to fs (#2850)."""
+    fs = VikingFS(agfs=_DummyAgfs())
+    monkeypatch.setattr(fs, "_get_vector_store", lambda: _SlowVectorStore())
+    monkeypatch.setattr(fs, "_ensure_access", lambda uri, ctx=None: None)
+    # Keep the test fast: 0.1s timeout instead of the 10s default.
+    monkeypatch.setenv("OPENVIKING_GREP_VIKINGDB_TIMEOUT_SEC", "0.1")
+
+    calls = []
+
+    async def fake_grep_fs(**kwargs):
+        calls.append(kwargs)
+        return {
+            "matches": [{"uri": "viking://resources/b.md"}],
+            "count": 1,
+            "match_count": 1,
+            "files_scanned": 1,
+        }
+
+    monkeypatch.setattr(fs, "_grep_fs", fake_grep_fs)
+
+    result = await fs._grep_vikingdb_then_fs(
+        uri="viking://resources",
+        pattern="needle",
+        exclude_uri=None,
+        case_insensitive=False,
+        node_limit=10,
+        level_limit=3,
+        ctx=None,
+    )
+
+    assert len(calls) == 1
+    assert result["count"] == 1
+    assert result["matches"][0]["uri"] == "viking://resources/b.md"
+
+
 @pytest.mark.asyncio
 async def test_grep_vikingdb_pushes_exclude_uri_to_filter(monkeypatch):
     fs = VikingFS(agfs=_DummyAgfs())
     vector_store = _DummyVectorStore()
     monkeypatch.setattr(fs, "_get_vector_store", lambda: vector_store)
     monkeypatch.setattr(fs, "_ensure_access", lambda uri, ctx=None: None)
 
+    # Empty VikingDB recall now falls back to fs grep (#2850); stub fs so the
+    # exclude_uri filter assertion stays focused on the remote query path.
+    async def fake_grep_fs(**kwargs):
+        return {"matches": [], "count": 0, "match_count": 0, "files_scanned": 0}
+
+    monkeypatch.setattr(fs, "_grep_fs", fake_grep_fs)
+
     result = await fs._grep_vikingdb_then_fs(
         uri="viking://resources",
         pattern="needle",