handle gcdata jobs in parse queue

howardchung · howardchung · commit c96b87fb991d · 2025-12-22T23:16:20.000Z
diff --git a/dev/transferQueue.ts b/dev/transferQueue.ts
diff --git a/ecosystem.config.js b/ecosystem.config.js
@@ -103,10 +103,6 @@ let arr = [
     name: "monitor",
     group: "backend",
   },
-  {
-    name: "gcdata",
-    group: "backend",
-  },
   {
     name: "buildsets",
     group: "backend",
diff --git a/global.d.ts b/global.d.ts
@@ -279,11 +279,6 @@ type ProfileJob = {
   account_id: number;
 };
 
-type GcDataJob = {
-  match_id: number;
-  pgroup: PGroup;
-};
-
 type CountsJob = ApiData;
 type ScenariosJob = {
   match_id: number;
@@ -295,6 +290,7 @@ type CacheJob = {
 type ParseJob = {
   match_id: number;
   origin?: DataOrigin;
+  gcDataOnly?: boolean;
 };
 
 type QueueJob = QueueInput["data"];
@@ -320,10 +316,6 @@ type QueueInput =
       name: "parse";
       data: ParseJob;
     }
-  | {
-      name: "gcQueue";
-      data: GcDataJob;
-    }
   | {
       name: "cacheQueue";
       data: CacheJob;
diff --git a/svc/gcdata.ts b/svc/gcdata.ts
diff --git a/svc/monitor.ts b/svc/monitor.ts
@@ -17,7 +17,6 @@ const health: Record<string, () => Promise<Metric>> = {
   seqNumDelay,
   parseDelay,
   fhDelay,
-  gcDelay,
   mmrDelay,
   cacheDelay,
   scenariosDelay,
@@ -92,15 +91,6 @@ async function parseDelay() {
     limit: 10000,
   };
 }
-async function gcDelay() {
-  const result = await db.raw(
-    "select count(*) from queue where type = 'gcQueue'",
-  );
-  return {
-    metric: result.rows[0]?.count,
-    limit: 100000,
-  };
-}
 async function fhDelay() {
   const result = await db.raw(
     "select count(*) from queue where type = 'fhQueue'",
@@ -170,11 +160,11 @@ WHERE keyspace_name = 'yasp';
   };
 }
 async function diskUsage() {
-  const result = await statfs('/');
+  const result = await statfs("/");
   return {
     metric: (result.blocks - result.bavail) * result.bsize,
     limit: result.blocks * result.bsize,
-  }
+  };
 }
 async function redisUsage() {
   const info = await redis.info();
diff --git a/svc/parser.ts b/svc/parser.ts
@@ -30,7 +30,7 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
   let parseTime = 0;
   try {
     redisCount("parser_job");
-    redis.publish(
+    await redis.publish(
       String(metadata.jobId),
       c.magenta(
         `Starting [job: ${metadata.jobId}] [match: ${job.match_id}] after ${Date.now() - Number(metadata.timestamp)}ms`,
@@ -50,7 +50,7 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
     }
 
     // Fetch the API data
-    redis.publish(String(metadata.jobId), c.blue(`Fetching API data...`));
+    await redis.publish(String(metadata.jobId), c.blue(`Fetching API data...`));
     const apiStart = Date.now();
     // The pgroup is used to update player_caches on insert.
     // Since currently gcdata and parse data have no knowledge of anonymity, we pass it from API data
@@ -73,7 +73,7 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
       return false;
     }
     apiTime = Date.now() - apiStart;
-    redis.publish(
+    await redis.publish(
       String(metadata.jobId),
       c.green(`Fetched API data in ${apiTime}ms`),
     );
@@ -90,7 +90,10 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
     // }
 
     // Fetch the gcdata and construct a replay URL
-    redis.publish(String(metadata.jobId), c.blue(`Fetching replay data...`));
+    await redis.publish(
+      String(metadata.jobId),
+      c.blue(`Fetching replay data...`),
+    );
     const gcStart = Date.now();
     const { data: gcMatch, error: gcError } =
       await gcFetcher.getOrFetchDataWithRetry(
@@ -107,17 +110,23 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
       return false;
     }
     gcTime = Date.now() - gcStart;
-    redis.publish(
+    await redis.publish(
       String(metadata.jobId),
       c.green(`Fetched replay data in ${gcTime}ms`),
     );
+
+    if (job.gcDataOnly) {
+      await log("skip", "Fetching replay data only, skipping parse");
+      return true;
+    }
+
     let url = buildReplayUrl(
       gcMatch.match_id,
       gcMatch.cluster,
       gcMatch.replay_salt,
     );
 
-    redis.publish(String(metadata.jobId), c.blue(`Parsing replay...`));
+    await redis.publish(String(metadata.jobId), c.blue(`Parsing replay...`));
     const parseStart = Date.now();
     const { error: parseError, skipped } = await parsedFetcher.getOrFetchData(
       matchId,
@@ -132,7 +141,7 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
       },
     );
     parseTime = Date.now() - parseStart;
-    redis.publish(
+    await redis.publish(
       String(metadata.jobId),
       c.green(`Parsed replay in ${parseTime}ms`),
     );
@@ -182,8 +191,8 @@ async function parseProcessor(job: ParseJob, metadata: JobMetadata) {
         job.match_id
       } ${displayMsg ?? ""}`,
     );
-    redis.publish("parsed", message);
-    redis.publish(
+    await redis.publish("parsed", message);
+    await redis.publish(
       String(metadata.jobId),
       c[colors[type]](
         `[${type}] Finished [job: ${metadata.jobId}] [match: ${
diff --git a/svc/retriever.ts b/svc/retriever.ts
@@ -57,7 +57,7 @@ const CMsgGCMatchDetailsResponse = builder.lookupType(
 
 setInterval(() => {
   const shouldRestart =
-    matchRequests >= clientMap.size * matchesPerAccount || 
+    matchRequests >= clientMap.size * matchesPerAccount ||
     profileRequests - profileSuccesses > 100000 ||
     noneReady();
   if (config.NODE_ENV !== "development" && shouldRestart && getUptime() > 60) {
diff --git a/svc/store/queue.ts b/svc/store/queue.ts
@@ -138,8 +138,6 @@ export async function addReliableJob(
     jobKey = `${name}:${data.match_id}`;
   } else if (name === "fhQueue") {
     jobKey = `${name}:${data.account_id}`;
-  } else if (name === "gcQueue") {
-    jobKey = `${name}:${data.match_id}`;
   } else if (name === "scenariosQueue") {
     jobKey = `${name}:${data.match_id}`;
   } else if (name === "profileQueue") {
@@ -182,7 +180,7 @@ export async function addReliableJob(
         name === "parse" ? data.match_id : ""
       }`,
     );
-    redis.publish("queue", message);
+    await redis.publish("queue", message);
   }
   // This might be undefined if a job with the same key already exists. Try to find it
   // May not exist anymore if the job finished in the meantime
diff --git a/svc/util/insert.ts b/svc/util/insert.ts
@@ -56,6 +56,8 @@ export async function insertMatch(
   // Do this after removing anonymous account IDs
   const pgroup = options.pgroup ?? getPGroup(match as ApiData);
   const trx = await db.transaction();
+  let doGcData = false;
+
   try {
     let isProTier = false;
     if ("leagueid" in match && match.leagueid) {
@@ -743,28 +745,15 @@ function updateMatchups(match) {
       ),
     );
   }
+
   async function queueGcData(trx: knex.Knex.Transaction) {
     // Trigger a request for gcdata
     if (
       options.origin === "scanner" &&
       options.type === "api" &&
-      "game_mode" in match &&
-      // Don't get replay URLs for event matches
-      match.game_mode !== 19 &&
       match.match_id % 100 < Number(config.GCDATA_PERCENT)
     ) {
-      await addReliableJob(
-        {
-          name: "gcQueue",
-          data: {
-            match_id: match.match_id,
-            pgroup,
-          },
-        },
-        {
-          trx,
-        },
-      );
+      doGcData = true;
     }
   }
 
@@ -789,18 +778,7 @@ function updateMatchups(match) {
           match.game_mode,
         ],
       );
-      await addReliableJob(
-        {
-          name: "gcQueue",
-          data: {
-            match_id: match.match_id,
-            pgroup,
-          },
-        },
-        {
-          trx,
-        },
-      );
+      doGcData = true;
     }
   }
 
@@ -836,10 +814,6 @@ function updateMatchups(match) {
     if (options.skipParse) {
       return null;
     }
-    if ("game_mode" in match && match.game_mode === 19) {
-      // don't parse event matches
-      return null;
-    }
     // We only auto-parse if this is a fresh match from API
     if (!(options.origin === "scanner" && options.type === "api")) {
       return null;
@@ -855,6 +829,8 @@ function updateMatchups(match) {
     const doParse = hasTrackedPlayer || isLeagueMatch;
     if (doParse) {
       redisCount("auto_parse");
+    }
+    if (doGcData || doParse) {
       // By default, lower priority than requests
       let priority = PRIORITY.AUTO_DEFAULT;
       if (isLeagueMatch) {
@@ -863,6 +839,9 @@ function updateMatchups(match) {
       if (hasTrackedPlayer) {
         priority = PRIORITY.AUTO_TRACKED_PLAYER;
       }
+      if (doGcData && !doParse) {
+        priority = PRIORITY.AUTO_GCDATA;
+      }
       // We might have to retry since it might be too soon for the replay
       let attempts = 50;
       const job = await addReliableJob(
@@ -871,6 +850,7 @@ function updateMatchups(match) {
           data: {
             match_id: match.match_id,
             origin: options.origin,
+            gcDataOnly: doGcData && !doParse,
           },
         },
         {
diff --git a/svc/util/priority.ts b/svc/util/priority.ts
@@ -6,5 +6,6 @@ export const PRIORITY = {
   REQUEST_DEFAULT: -2,
   REQUEST_API_KEY: -1,
   AUTO_DEFAULT: 0,
+  AUTO_GCDATA: 1,
   REQUEST_ALREADY_PARSED: 9,
 };
