Merge pull request #7106 from shin-core/fix/qdrant-vectorize-fetch-timeout-7072

fix(selfhost): bound every Qdrant REST call with an AbortSignal timeout

Author: JSONbored

src/selfhost/qdrant-vectorize.ts

@@ -28,6 +28,10 @@ import type {
 
 const DEFAULT_COLLECTION = "loopover";
 const DEFAULT_DIM = 1024; // bge-m3 / mxbai-embed-large (1024-d); set QDRANT_DIM to override
+// Per-request ceiling for every Qdrant REST call (#7072). This is a self-host Node.js adapter with no
+// platform-level subrequest limit, so an unresponsive/partitioned Qdrant instance would otherwise hang these
+// calls indefinitely -- matching the bounded-fetch convention of the other self-host adapters (e.g. ai.ts).
+const QDRANT_FETCH_TIMEOUT_MS = 15_000;
 
 interface QdrantSearchResult {
   result: Array<{ id: string; score: number; payload: Record<string, unknown> }>;
@@ -73,12 +77,16 @@ export async function initQdrantCollection(
     method: "PUT",
     headers: qdrantHeaders(),
     body: JSON.stringify({ vectors: { size: dim, distance: "Cosine" } }),
+    signal: AbortSignal.timeout(QDRANT_FETCH_TIMEOUT_MS),
   });
   if (!res.ok && res.status !== 409) {
     throw new Error(`Qdrant collection init failed: HTTP ${res.status}`);
   }
   if (res.status === 409) {
-    const existing = await fetch(`${base}/collections/${collection}`, { headers: qdrantHeaders() });
+    const existing = await fetch(`${base}/collections/${collection}`, {
+      headers: qdrantHeaders(),
+      signal: AbortSignal.timeout(QDRANT_FETCH_TIMEOUT_MS),
+    });
     if (!existing.ok) throw new Error(`Qdrant collection lookup failed: HTTP ${existing.status}`);
     const info = (await existing.json()) as QdrantCollectionInfo;
     const existingDim = info.result.config.params.vectors.size;
@@ -103,6 +111,7 @@ export function createQdrantVectorize(url: string, collection = DEFAULT_COLLECTI
         method: "PUT",
         headers: qdrantHeaders(),
         body: JSON.stringify({ points }),
+        signal: AbortSignal.timeout(QDRANT_FETCH_TIMEOUT_MS),
       });
       if (!res.ok) {
         incr("loopover_qdrant_errors_total", { op: "upsert" });
@@ -123,6 +132,7 @@ export function createQdrantVectorize(url: string, collection = DEFAULT_COLLECTI
           method: "POST",
           headers: qdrantHeaders(),
           body: JSON.stringify(body),
+          signal: AbortSignal.timeout(QDRANT_FETCH_TIMEOUT_MS),
         });
       } catch {
         // Qdrant unreachable — degrade gracefully (RAG returns no context rather than crashing)
@@ -150,6 +160,7 @@ export function createQdrantVectorize(url: string, collection = DEFAULT_COLLECTI
         method: "POST",
         headers: qdrantHeaders(),
         body: JSON.stringify({ points }),
+        signal: AbortSignal.timeout(QDRANT_FETCH_TIMEOUT_MS),
       });
       if (!res.ok) {
         incr("loopover_qdrant_errors_total", { op: "delete" });

test/unit/selfhost-qdrant-vectorize.test.ts

@@ -302,3 +302,52 @@ describe("createQdrantVectorize (#1217 Qdrant adapter)", () => {
     expect(url).not.toContain("//collections");
   });
 });
+
+describe("Qdrant REST calls are bounded by an AbortSignal timeout (#7072)", () => {
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  function assertEveryFetchTimed(fake: ReturnType<typeof mockFetch>) {
+    expect(fake.mock.calls.length).toBeGreaterThan(0);
+    for (const call of fake.mock.calls) {
+      const init = (call as unknown as [string, RequestInit | undefined])[1];
+      expect(init?.signal).toBeInstanceOf(AbortSignal);
+    }
+  }
+
+  it("bounds initQdrantCollection's PUT (fresh create) and its GET dimension-check (pre-existing collection)", async () => {
+    // Fresh create: the PUT returns 200 and no GET follows.
+    const created = mockFetch(200);
+    vi.stubGlobal("fetch", created);
+    await initQdrantCollection(BASE);
+    assertEveryFetchTimed(created);
+
+    // Pre-existing collection: the PUT 409s, so the GET dimension-check runs too -- both must be timed.
+    const existing = vi
+      .fn()
+      .mockResolvedValueOnce(new Response("conflict", { status: 409 }))
+      .mockResolvedValueOnce(new Response(JSON.stringify(collectionInfo(1024)), { status: 200 }));
+    vi.stubGlobal("fetch", existing);
+    await initQdrantCollection(BASE);
+    assertEveryFetchTimed(existing as unknown as ReturnType<typeof mockFetch>);
+    expect(existing).toHaveBeenCalledTimes(2);
+  });
+
+  it("bounds upsert, query, and deleteByIds", async () => {
+    const upsertFake = mockFetch(200, { status: "ok" });
+    vi.stubGlobal("fetch", upsertFake);
+    await createQdrantVectorize(BASE).upsert([{ id: "r/f:1", values: [0.1, 0.2] }]);
+    assertEveryFetchTimed(upsertFake);
+
+    const queryFake = mockFetch(200, { result: [] });
+    vi.stubGlobal("fetch", queryFake);
+    await createQdrantVectorize(BASE).query([0.5, 0.5], { topK: 5 });
+    assertEveryFetchTimed(queryFake);
+
+    const deleteFake = mockFetch(200, { status: "ok" });
+    vi.stubGlobal("fetch", deleteFake);
+    await createQdrantVectorize(BASE).deleteByIds(["r/f:1"]);
+    assertEveryFetchTimed(deleteFake);
+  });
+});