fix(gateway): stop Qdrant upsert aborts (semantic cache was being skipped)
Qdrant upserts were failing with 'This operation was aborted' ~32x/2h, so semantic moderation cache entries were silently dropped. Root cause: upsertQdrantPoint ran ensureQdrantCollection() on EVERY call — a GET (and sometimes DELETE+PUT) round-trip — while the request AbortController had only a 10s timeout. Under moderation load Qdrant is busy (the gmw_text_moderation collection is not yet HNSW-indexed, so searches are full-scans), the extra round-trips pushed the upsert past 10s, and the client aborted it. - Memoise ensureQdrantCollection() at module scope so the collection is verified exactly once per process (resetQdrantCollectionCache() for tests / config reload). - Bump the upsert request timeout 10s -> 30s so a transiently busy Qdrant no longer aborts the write. Qdrant server itself is healthy (<100ms for direct upsert; collection is green), so no server-side change is needed. Semantic cache should now populate reliably.
This commit is contained in:
@@ -17,6 +17,18 @@ import { config } from "../../shared/config/config.js";
|
||||
|
||||
const log = createChildLogger("qdrant");
|
||||
|
||||
// ensureQdrantCollection performs a network round-trip (GET, possibly
|
||||
// DELETE+PUT). Running it on every upsert adds 1-3 HTTP calls per
|
||||
// moderation verdict, which under Qdrant load pushes the upsert past the
|
||||
// request timeout and aborts it ("This operation was aborted"). Memoise the
|
||||
// result so the collection is only verified once per process lifetime.
|
||||
let ensureCollectionPromise: Promise<boolean> | null = null;
|
||||
|
||||
/** Reset the memoised ensure result (used by tests / config reload). */
|
||||
export function resetQdrantCollectionCache(): void {
|
||||
ensureCollectionPromise = null;
|
||||
}
|
||||
|
||||
export interface QdrantVerdictPayload {
|
||||
text: string;
|
||||
flags: string; // JSON string of the full moderation result
|
||||
@@ -97,46 +109,53 @@ export function qdrantPointId(cacheKey: string): number {
|
||||
export async function ensureQdrantCollection(
|
||||
vectorSize: number,
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
// 404 = collection doesn't exist yet → create it.
|
||||
let existing: {
|
||||
result?: { config?: { params?: { vectors?: { size?: number } } } };
|
||||
} | null = null;
|
||||
if (ensureCollectionPromise) return ensureCollectionPromise;
|
||||
ensureCollectionPromise = (async () => {
|
||||
try {
|
||||
existing = (await request("GET", `/collections/${collectionName()}`)) as {
|
||||
// 404 = collection doesn't exist yet → create it.
|
||||
let existing: {
|
||||
result?: { config?: { params?: { vectors?: { size?: number } } } };
|
||||
};
|
||||
} catch (error) {
|
||||
if (!(error instanceof Error) || !error.message.includes("-> 404")) {
|
||||
throw error;
|
||||
} | null = null;
|
||||
try {
|
||||
existing = (await request(
|
||||
"GET",
|
||||
`/collections/${collectionName()}`,
|
||||
)) as {
|
||||
result?: { config?: { params?: { vectors?: { size?: number } } } };
|
||||
};
|
||||
} catch (error) {
|
||||
if (!(error instanceof Error) || !error.message.includes("-> 404")) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const size = existing?.result?.config?.params?.vectors?.size;
|
||||
if (size === vectorSize) return true;
|
||||
const size = existing?.result?.config?.params?.vectors?.size;
|
||||
if (size === vectorSize) return true;
|
||||
|
||||
if (size !== undefined && size !== vectorSize) {
|
||||
log.warn(
|
||||
{ collection: collectionName(), oldSize: size, newSize: vectorSize },
|
||||
"Qdrant collection vector size changed — recreating collection",
|
||||
if (size !== undefined && size !== vectorSize) {
|
||||
log.warn(
|
||||
{ collection: collectionName(), oldSize: size, newSize: vectorSize },
|
||||
"Qdrant collection vector size changed — recreating collection",
|
||||
);
|
||||
await request("DELETE", `/collections/${collectionName()}`);
|
||||
}
|
||||
|
||||
await request("PUT", `/collections/${collectionName()}`, {
|
||||
vectors: { size: vectorSize, distance: "Cosine" },
|
||||
});
|
||||
return true;
|
||||
} catch (error) {
|
||||
log.error(
|
||||
{
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
collection: collectionName(),
|
||||
},
|
||||
"Failed to ensure Qdrant collection",
|
||||
);
|
||||
await request("DELETE", `/collections/${collectionName()}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
await request("PUT", `/collections/${collectionName()}`, {
|
||||
vectors: { size: vectorSize, distance: "Cosine" },
|
||||
});
|
||||
return true;
|
||||
} catch (error) {
|
||||
log.error(
|
||||
{
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
collection: collectionName(),
|
||||
},
|
||||
"Failed to ensure Qdrant collection",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
})();
|
||||
return ensureCollectionPromise;
|
||||
}
|
||||
|
||||
/** Upsert one embedding + verdict payload point. Returns false on failure. */
|
||||
@@ -147,10 +166,15 @@ export async function upsertQdrantPoint(
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
if (!(await ensureQdrantCollection(vector.length))) return false;
|
||||
await request("PUT", `/collections/${collectionName()}/points`, {
|
||||
points: [{ id: qdrantPointId(cacheKey), vector, payload }],
|
||||
wait: true,
|
||||
});
|
||||
await request(
|
||||
"PUT",
|
||||
`/collections/${collectionName()}/points`,
|
||||
{
|
||||
points: [{ id: qdrantPointId(cacheKey), vector, payload }],
|
||||
wait: true,
|
||||
},
|
||||
30_000,
|
||||
);
|
||||
return true;
|
||||
} catch (error) {
|
||||
log.warn(
|
||||
|
||||
Reference in New Issue
Block a user