Add job identity to indexing log messages

habdelra · habdelra · commit 1509cca6ce8d · 2025-03-10T19:45:24.000-04:00
diff --git a/packages/host/app/components/card-prerender.gts b/packages/host/app/components/card-prerender.gts
@@ -5,7 +5,7 @@ import Component from '@glimmer/component';
 
 import { didCancel, enqueueTask } from 'ember-concurrency';
 
-import { type IndexWriter } from '@cardstack/runtime-common';
+import { type IndexWriter, type JobInfo } from '@cardstack/runtime-common';
 import { readFileAsText as _readFileAsText } from '@cardstack/runtime-common/stream';
 import {
   getReader,
@@ -100,11 +100,12 @@ export default class CardPrerender extends Component {
   });
 
   private doFromScratch = enqueueTask(async (realmURL: URL) => {
-    let { reader, indexWriter } = this.getRunnerParams(realmURL);
+    let { reader, indexWriter, jobInfo } = this.getRunnerParams(realmURL);
     let currentRun = new CurrentRun({
       realmURL,
       reader,
       indexWriter,
+      jobInfo,
       renderCard: this.renderService.renderCard,
       render: this.renderService.render,
     });
@@ -122,11 +123,12 @@ export default class CardPrerender extends Component {
       operation: 'delete' | 'update',
       ignoreData: Record<string, string>,
     ) => {
-      let { reader, indexWriter } = this.getRunnerParams(realmURL);
+      let { reader, indexWriter, jobInfo } = this.getRunnerParams(realmURL);
       let currentRun = new CurrentRun({
         realmURL,
         reader,
         indexWriter,
+        jobInfo,
         ignoreData: { ...ignoreData },
         renderCard: this.renderService.renderCard,
         render: this.renderService.render,
@@ -144,15 +146,18 @@ export default class CardPrerender extends Component {
   private getRunnerParams(realmURL: URL): {
     reader: Reader;
     indexWriter: IndexWriter;
+    jobInfo?: JobInfo;
   } {
     if (this.fastboot.isFastBoot) {
       let optsId = (globalThis as any).runnerOptsId;
       if (optsId == null) {
         throw new Error(`Runner Options Identifier was not set`);
       }
+      let { reader, indexWriter, jobInfo } = getRunnerOpts(optsId);
       return {
-        reader: getRunnerOpts(optsId).reader,
-        indexWriter: getRunnerOpts(optsId).indexWriter,
+        reader,
+        indexWriter,
+        jobInfo,
       };
     } else {
       return {
diff --git a/packages/host/app/lib/current-run.ts b/packages/host/app/lib/current-run.ts
@@ -33,6 +33,7 @@ import {
   type Relationship,
   type TextFileRef,
   type LastModifiedTimes,
+  type JobInfo,
 } from '@cardstack/runtime-common';
 import { Deferred } from '@cardstack/runtime-common/deferred';
 import {
@@ -88,6 +89,7 @@ export class CurrentRun {
   #render: Render;
   #realmURL: URL;
   #realmInfo?: RealmInfo;
+  #jobInfo?: JobInfo;
   readonly stats: Stats = {
     instancesIndexed: 0,
     modulesIndexed: 0,
@@ -105,13 +107,15 @@ export class CurrentRun {
     ignoreData = {},
     renderCard,
     render,
+    jobInfo,
   }: {
     realmURL: URL;
     reader: Reader;
     indexWriter: IndexWriter;
     ignoreData?: Record<string, string>;
     renderCard: RenderCard;
     render: Render;
+    jobInfo?: JobInfo;
   }) {
     this.#indexWriter = indexWriter;
     this.#realmPaths = new RealmPaths(realmURL);
@@ -120,44 +124,51 @@ export class CurrentRun {
     this.#ignoreData = ignoreData;
     this.#renderCard = renderCard;
     this.#render = render;
+    this.#jobInfo = jobInfo;
   }
 
   static async fromScratch(current: CurrentRun): Promise<IndexResults> {
     let start = Date.now();
-    log.debug(`starting from scratch indexing`);
+    log.debug(
+      `${jobIdentity(current.#jobInfo)} starting from scratch indexing`,
+    );
     perfLog.debug(
-      `starting from scratch indexing for realm ${current.realmURL.href}`,
+      `${jobIdentity(current.#jobInfo)} starting from scratch indexing for realm ${current.realmURL.href}`,
     );
     current.#batch = await current.#indexWriter.createBatch(current.realmURL);
     let invalidations: URL[] = [];
     let mtimesStart = Date.now();
     let mtimes = await current.batch.getModifiedTimes();
     perfLog.debug(
-      `completed getting index mtimes in ${Date.now() - mtimesStart} ms`,
+      `${jobIdentity(current.#jobInfo)} completed getting index mtimes in ${Date.now() - mtimesStart} ms`,
     );
     let invalidateStart = Date.now();
     invalidations = (
       await current.discoverInvalidations(current.realmURL, mtimes)
     ).map((href) => new URL(href));
     perfLog.debug(
-      `completed invalidations in ${Date.now() - invalidateStart} ms`,
+      `${jobIdentity(current.#jobInfo)} completed invalidations in ${Date.now() - invalidateStart} ms`,
     );
 
     await current.whileIndexing(async () => {
       let visitStart = Date.now();
       for (let invalidation of invalidations) {
         await current.tryToVisit(invalidation);
       }
-      perfLog.debug(`completed index visit in ${Date.now() - visitStart} ms`);
+      perfLog.debug(
+        `${jobIdentity(current.#jobInfo)} completed index visit in ${Date.now() - visitStart} ms`,
+      );
       let finalizeStart = Date.now();
       let { totalIndexEntries } = await current.batch.done();
       perfLog.debug(
-        `completed index finalization in ${Date.now() - finalizeStart} ms`,
+        `${jobIdentity(current.#jobInfo)} completed index finalization in ${Date.now() - finalizeStart} ms`,
       );
       current.stats.totalIndexEntries = totalIndexEntries;
-      log.debug(`completed from scratch indexing in ${Date.now() - start}ms`);
+      log.debug(
+        `${jobIdentity(current.#jobInfo)} completed from scratch indexing in ${Date.now() - start}ms`,
+      );
       perfLog.debug(
-        `completed from scratch indexing for realm ${
+        `${jobIdentity(current.#jobInfo)} completed from scratch indexing for realm ${
           current.realmURL.href
         } in ${Date.now() - start} ms`,
       );
@@ -180,7 +191,9 @@ export class CurrentRun {
     },
   ): Promise<IndexResults> {
     let start = Date.now();
-    log.debug(`starting from incremental indexing for ${url.href}`);
+    log.debug(
+      `${jobIdentity(current.#jobInfo)} starting from incremental indexing for ${url.href}`,
+    );
 
     current.#batch = await current.#indexWriter.createBatch(current.realmURL);
     let invalidations = (await current.batch.invalidate(url)).map(
@@ -200,7 +213,7 @@ export class CurrentRun {
       current.stats.totalIndexEntries = totalIndexEntries;
 
       log.debug(
-        `completed incremental indexing for ${url.href} in ${
+        `${jobIdentity(current.#jobInfo)} completed incremental indexing for ${url.href} in ${
           Date.now() - start
         }ms`,
       );
@@ -217,7 +230,9 @@ export class CurrentRun {
       await this.visitFile(url);
     } catch (err: any) {
       if (isCardError(err) && err.status === 404) {
-        log.info(`tried to visit file ${url.href}, but it no longer exists`);
+        log.info(
+          `${jobIdentity(this.#jobInfo)} tried to visit file ${url.href}, but it no longer exists`,
+        );
       } else {
         throw err;
       }
@@ -258,12 +273,16 @@ export class CurrentRun {
     url: URL,
     indexMtimes: LastModifiedTimes,
   ): Promise<string[]> {
-    log.debug(`discovering invalidations in dir ${url.href}`);
-    perfLog.debug(`discovering invalidations in dir ${url.href}`);
+    log.debug(
+      `${jobIdentity(this.#jobInfo)} discovering invalidations in dir ${url.href}`,
+    );
+    perfLog.debug(
+      `${jobIdentity(this.#jobInfo)} discovering invalidations in dir ${url.href}`,
+    );
     let mtimesStart = Date.now();
     let filesystemMtimes = await this.#reader.mtimes();
     perfLog.debug(
-      `time to get file system mtimes ${Date.now() - mtimesStart} ms`,
+      `${jobIdentity(this.#jobInfo)} time to get file system mtimes ${Date.now() - mtimesStart} ms`,
     );
 
     let ignoreFile = new URL('.gitignore', url).href;
@@ -278,7 +297,9 @@ export class CurrentRun {
         this.#ignoreData[url.href] = ignorePatterns.content;
       }
     } else {
-      perfLog.debug(`skip getting the ignore file--there is nothing to ignore`);
+      perfLog.debug(
+        `${jobIdentity(this.#jobInfo)} skip getting the ignore file--there is nothing to ignore`,
+      );
     }
 
     let invalidationList: string[] = [];
@@ -314,7 +335,7 @@ export class CurrentRun {
       await this.batch.invalidate(new URL(invalidationURL));
     }
     perfLog.debug(
-      `time to invalidate ${url} ${Date.now() - invalidationStart} ms`,
+      `${jobIdentity(this.#jobInfo)} time to invalidate ${url} ${Date.now() - invalidationStart} ms`,
     );
     return this.batch.invalidations;
   }
@@ -327,7 +348,7 @@ export class CurrentRun {
       return;
     }
     let start = Date.now();
-    log.debug(`begin visiting file ${url.href}`);
+    log.debug(`${jobIdentity(this.#jobInfo)} begin visiting file ${url.href}`);
     let localPath: string;
     try {
       localPath = this.#realmPaths.local(url);
@@ -336,7 +357,7 @@ export class CurrentRun {
       // graph cross a realm just skip over the file. it will be out
       // of date, but such is life...
       log.debug(
-        `Visit of ${url.href} cannot be performed as it is in a different realm than the realm whose contents are being invalidated (${this.realmURL.href})`,
+        `${jobIdentity(this.#jobInfo)} Visit of ${url.href} cannot be performed as it is in a different realm than the realm whose contents are being invalidated (${this.realmURL.href})`,
       );
       return;
     }
@@ -365,13 +386,15 @@ export class CurrentRun {
           let { data } = JSON.parse(content);
           resource = data;
         } catch (e) {
-          log.warn(`unable to parse ${url.href} as card JSON`);
+          log.warn(
+            `${jobIdentity(this.#jobInfo)} unable to parse ${url.href} as card JSON`,
+          );
         }
 
         if (resource && isCardResource(resource)) {
           if (lastModified == null) {
             log.warn(
-              `No lastModified date available for ${url.href}, using current time`,
+              `${jobIdentity(this.#jobInfo)} No lastModified date available for ${url.href}, using current time`,
             );
             lastModified = unixTime(Date.now());
           }
@@ -386,7 +409,9 @@ export class CurrentRun {
         }
       }
     }
-    log.debug(`completed visiting file ${url.href} in ${Date.now() - start}ms`);
+    log.debug(
+      `${jobIdentity(this.#jobInfo)} completed visiting file ${url.href} in ${Date.now() - start}ms`,
+    );
   }
 
   private async indexModule(url: URL, ref: TextFileRef): Promise<void> {
@@ -395,7 +420,7 @@ export class CurrentRun {
     } catch (err: any) {
       this.stats.moduleErrors++;
       log.warn(
-        `encountered error loading module "${url.href}": ${err.message}`,
+        `${jobIdentity(this.#jobInfo)} encountered error loading module "${url.href}": ${err.message}`,
       );
       let deps = await (
         await this.loaderService.loader.getConsumedModules(url.href)
@@ -413,7 +438,9 @@ export class CurrentRun {
     }
 
     if (ref.isShimmed) {
-      log.debug(`skipping indexing of shimmed module ${url.href}`);
+      log.debug(
+        `${jobIdentity(this.#jobInfo)} skipping indexing of shimmed module ${url.href}`,
+      );
       return;
     }
     let consumes = (
@@ -650,7 +677,7 @@ export class CurrentRun {
         throw err;
       }
       log.warn(
-        `encountered error indexing card instance ${path}: ${error.error.message}`,
+        `${jobIdentity(this.#jobInfo)} encountered error indexing card instance ${path}: ${error.error.message}`,
       );
       await this.updateEntry(instanceURL, error);
     }
@@ -811,3 +838,10 @@ function getDisplayName(card: typeof CardDef) {
     return card.displayName;
   }
 }
+
+function jobIdentity(jobInfo?: JobInfo): string {
+  if (!jobInfo) {
+    return `[no job identity]`;
+  }
+  return `[job: ${jobInfo.jobId}] [jobReservation: ${jobInfo.reservationId}]`;
+}
diff --git a/packages/postgres/pg-queue.ts b/packages/postgres/pg-queue.ts
@@ -4,6 +4,7 @@ import {
   type QueueRunner,
   type PgPrimitive,
   type Expression,
+  type JobInfo,
   param,
   separatedByCommas,
   addExplicitParens,
@@ -269,11 +270,20 @@ export class PgQueueRunner implements QueueRunner {
     }
   }
 
-  private async runJob(jobType: string, args: PgPrimitive) {
+  private async runJob(jobType: string, args: PgPrimitive, jobInfo: JobInfo) {
     let handler = this.#handlers.get(jobType);
     if (!handler) {
       throw new Error(`unknown job handler ${jobType}`);
     }
+
+    if (
+      args &&
+      typeof args === 'object' &&
+      !Array.isArray(args) &&
+      !('jobInfo' in args)
+    ) {
+      args.jobInfo = jobInfo;
+    }
     return await handler(args);
   }
 
@@ -348,7 +358,11 @@ export class PgQueueRunner implements QueueRunner {
           try {
             log.debug(`%s: running %s`, this.#workerId, jobToRun.id);
             result = await Promise.race([
-              this.runJob(jobToRun.job_type, jobToRun.args),
+              this.runJob(jobToRun.job_type, jobToRun.args, {
+                jobId: jobToRun.id,
+                workerId: this.#workerId,
+                reservationId: jobReservationId,
+              }),
               // we race the job so that it doesn't hold this worker hostage if
               // the job's promise never resolves
               new Promise<'timeout'>((r) =>
diff --git a/packages/runtime-common/worker.ts b/packages/runtime-common/worker.ts
