From b258c48040a7da088112e3ddd019c40b211182ed Mon Sep 17 00:00:00 2001
From: Oleksandr Pichak <stasselust@gmail.com>
Date: Mon, 24 Aug 2026 12:32:15 +0200
Subject: [PATCH] fix(cache): a failed disk-cache write no longer kills the
 writer thread

An exception from diskcache escaped _consume_queue, terminated the only
consumer thread, and was swallowed by the never-retrieved Future from
_threadpool.submit(). From that point _add_q had no consumer and grew one
entry per embedded item for the lifetime of the process, while the cache
stored nothing.

Also fixes _threadpool.shutdown(wait=True) being called from a worker of
that same pool, which raises RuntimeError: cannot join current thread.
---
 .../infinity_emb/inference/caching_layer.py   | 18 +++++++---
 .../unit_test/inference/test_caching_layer.py | 36 +++++++++++++++++++
 2 files changed, 49 insertions(+), 5 deletions(-)

diff --git a/libs/infinity_emb/infinity_emb/inference/caching_layer.py b/libs/infinity_emb/infinity_emb/inference/caching_layer.py
index 4a5628f58..ccc72093b 100644
--- a/libs/infinity_emb/infinity_emb/inference/caching_layer.py
+++ b/libs/infinity_emb/infinity_emb/inference/caching_layer.py
@@ -58,11 +58,19 @@ def _consume_queue(self) -> None:
                 item = self._add_q.get(timeout=0.5)
             except queue.Empty:
                 continue
-            if item is not None:
-                k, v = item
-                self._cache.add(key=self._pre_hash(k), value=v, expire=86400)
-            self._add_q.task_done()
-        self._threadpool.shutdown(wait=True)
+            try:
+                if item is not None:
+                    k, v = item
+                    self._cache.add(key=self._pre_hash(k), value=v, expire=86400)
+            except Exception as ex:
+                # a best-effort cache must never take down its own writer thread.
+                logger.warning(f"failed to write to the vector disk cache: {ex}")
+            finally:
+                self._add_q.task_done()
+        # this runs *on* a worker of self._threadpool, so wait=True would join the
+        # current thread and raise `RuntimeError: cannot join current thread` into a
+        # Future that is never retrieved.
+        self._threadpool.shutdown(wait=False)
 
     def _get(self, sentence: str) -> Union[None, EmbeddingReturnType, list[float]]:
         """sets the item.complete() and sets embedding, if in cache."""
diff --git a/libs/infinity_emb/tests/unit_test/inference/test_caching_layer.py b/libs/infinity_emb/tests/unit_test/inference/test_caching_layer.py
index 254236b34..07b4ef951 100644
--- a/libs/infinity_emb/tests/unit_test/inference/test_caching_layer.py
+++ b/libs/infinity_emb/tests/unit_test/inference/test_caching_layer.py
@@ -42,3 +42,39 @@ async def test_cache():
     finally:
         INFINITY_CACHE_VECTORS = False
         shutdown.set()
+
+
+@pytest.mark.anyio
+async def test_consumer_survives_failed_write(monkeypatch):
+    """a raising `_cache.add` must not kill the only queue consumer.
+
+    Before the fix, the exception escaped `_consume_queue`, ended the single writer
+    thread, and was swallowed by the never-retrieved Future from `_threadpool.submit`.
+    `_add_q` then had no consumer at all and grew for the lifetime of the process.
+    """
+    shutdown = threading.Event()
+    try:
+        c = caching_layer.Cache(cache_name="pytest_write_error", shutdown=shutdown)
+        seen: list[str] = []
+        real_add = c._cache.add
+
+        def flaky(**kwargs):
+            seen.append(kwargs["key"])
+            if kwargs["key"] == "boom":
+                raise RuntimeError("simulated disk failure")
+            return real_add(**kwargs)
+
+        monkeypatch.setattr(c._cache, "add", flaky)
+        c._add_q.put(("boom", [1.0]))
+        c._add_q.put(("fine", [2.0]))
+
+        for _ in range(100):
+            await asyncio.sleep(0.05)
+            if seen == ["boom", "fine"]:
+                break
+
+        # the writer processed the item *after* the one that raised
+        assert seen == ["boom", "fine"]
+        assert c._get("fine") == [2.0]
+    finally:
+        shutdown.set()