From 2d7c7f2c357440a5174974e83019ad088a0e2c49 Mon Sep 17 00:00:00 2001
From: asepharyana <superaseph@gmail.com>
Date: Sat, 15 Aug 2026 23:40:19 +0700
Subject: [PATCH] fix(gateway): stop Qdrant upsert aborts (semantic cache was
 being skipped)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 .../src/modules/ai-moderation/qdrantClient.ts | 98 ++++++++++++-------
 1 file changed, 61 insertions(+), 37 deletions(-)

diff --git a/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts b/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts
index 07ff582..f8c13a7 100644
--- a/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts
+++ b/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts
@@ -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(