Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion src/selfhost/qdrant-vectorize.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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> }>;
Expand Down Expand Up @@ -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;
Expand All @@ -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" });
Expand All @@ -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)
Expand Down Expand Up @@ -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" });
Expand Down
49 changes: 49 additions & 0 deletions test/unit/selfhost-qdrant-vectorize.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
});
});
Loading