[CORL-620] Persisted Query Improvements (#2567)

* fix: resolves #2566

* fix: improved retry logic

src/core/common/helpers/index.ts

@@ -0,0 +1,3 @@
+export { default as waitFor } from "./waitFor";
+export * from "./validate";
+export * from "./i18n";

src/core/common/helpers/waitFor.ts

@@ -0,0 +1,7 @@
+export default function waitFor(timeout: number): Promise<void> {
+  return new Promise(resolve => {
+    setTimeout(() => {
+      resolve();
+    }, timeout);
+  });
+}

src/core/server/index.ts

@@ -99,6 +99,10 @@ class Server {
   // signingConfig is the server reference to the signing configuration.
   private signingConfig: JWTSigningConfig;
 
+  // persistedQueryCache is the cache of persisted queries used by the GraphQL
+  // server to handle persisted queries.
+  private persistedQueryCache: PersistedQueryCache;
+
   constructor(options: ServerOptions) {
     this.parentApp = express();
 
@@ -149,6 +153,9 @@ class Server {
       config
     );
 
+    // Load and upsert the persisted queries.
+    this.persistedQueryCache = new PersistedQueryCache({ mongo: this.mongo });
+
     // Prime the tenant cache so it'll be ready to serve now.
     await this.tenantCache.primeAll();
 
@@ -190,6 +197,9 @@ class Server {
       logger.warn("mongodb autoindexing is disabled, skipping indexing");
     }
 
+    // Prime the queries in the database.
+    await this.persistedQueryCache.prime();
+
     // Launch all of the job processors.
     this.tasks.mailer.process();
     this.tasks.scraper.process();
@@ -282,12 +292,6 @@ class Server {
     // Webpack Dev Server.
     const disableClientRoutes = this.config.get("disable_client_routes");
 
-    // Load and upsert the persisted queries.
-    const persistedQueryCache = new PersistedQueryCache({ mongo: this.mongo });
-
-    // Prime the queries in the database.
-    await persistedQueryCache.prime();
-
     const options: AppOptions = {
       parent,
       pubsub: this.pubsub,
@@ -302,7 +306,7 @@ class Server {
       scraperQueue: this.tasks.scraper,
       notifierQueue: this.tasks.notifier,
       disableClientRoutes,
-      persistedQueryCache,
+      persistedQueryCache: this.persistedQueryCache,
       persistedQueriesRequired:
         this.config.get("env") === "production" &&
         !this.config.get("enable_graphiql"),

src/core/server/models/queries/queries.ts

@@ -1,6 +1,8 @@
 import { OperationTypeNode } from "graphql";
-import { Db } from "mongodb";
+import { Db, MongoError } from "mongodb";
 
+import { waitFor } from "coral-common/helpers";
+import logger from "coral-server/logger";
 import {
   createCollection,
   createIndexFactory,
@@ -24,7 +26,11 @@ export async function createQueriesIndexes(mongo: Db) {
   await createIndex({ id: 1 }, { unique: true });
 }
 
-export async function primeQueries(mongo: Db, queries: PersistedQuery[]) {
+export async function primeQueries(
+  mongo: Db,
+  queries: PersistedQuery[],
+  tries = 1
+) {
   // Setup persisting these queries.
   const bulk = collection(mongo).initializeUnorderedBulkOp({});
 
@@ -39,8 +45,31 @@ export async function primeQueries(mongo: Db, queries: PersistedQuery[]) {
       .replaceOne(query);
   }
 
-  // Execute the bulk operations.
-  await bulk.execute();
+  try {
+    // Execute the bulk operations.
+    await bulk.execute({ w: "majority" });
+
+    return;
+  } catch (err) {
+    if (err instanceof MongoError && err.code === 11000) {
+      if (tries > 2) {
+        logger.warn(
+          { err, tries },
+          "duplicate error on inserting persisted queries after maximum tries reached"
+        );
+        return;
+      }
+
+      // Wait for 500ms before trying again.
+      await waitFor(500);
+
+      // Retry the priming operation.
+      await primeQueries(mongo, queries, tries + 1);
+    }
+
+    // An error unrelated to duplicate indexes was thrown, just rethrow it.
+    throw err;
+  }
 }
 
 export async function getQueries(mongo: Db, ids: string[]) {