Add progress logging to release validator

workers/release-validator/src/index.ts

@@ -21,26 +21,32 @@ export default class ReleaseValidatorWorker extends Worker {
    * Connect to the events database and start consuming tasks.
    */
   public async start(): Promise<void> {
+    this.logger.info('Connecting to events database');
     await this.eventsDb.connect();
+    this.logger.info('Events database connected; starting task consumer');
     await super.start();
+    this.logger.info('Ready to consume release validation tasks');
   }
 
   /**
    * Stop consuming tasks and close the database connection.
    */
   public async finish(): Promise<void> {
+    this.logger.info('Stopping release validator');
     await super.finish();
     await this.eventsDb.close();
+    this.logger.info('Release validator stopped');
   }
 
   /**
    * Handle a scheduled release validation task.
    */
   public async handle(): Promise<void> {
-    this.logger.info('Release validation started');
-
-    await validateReleases(this.eventsDb.getConnection());
-
-    this.logger.info('Release validation finished');
+    try {
+      await validateReleases(this.eventsDb.getConnection());
+    } catch (error) {
+      this.logger.error('Release validation failed', error);
+      throw error;
+    }
   }
 }

workers/release-validator/src/validate-releases.ts

@@ -46,6 +46,8 @@ async function findReleasesToCheck(db: Db, now: Date): Promise<ReleaseDBScheme[]
   const oldestReleaseId = ObjectID.createFromTime(nowSeconds - releaseMaxAgeSeconds);
   const newestReleaseId = ObjectID.createFromTime(nowSeconds - RELEASE_OBSERVATION_PERIOD_SECONDS);
 
+  logger.info(`Searching unchecked releases: from=${oldestReleaseId.getTimestamp().toISOString()} to=${newestReleaseId.getTimestamp().toISOString()}`);
+
   return db.collection<ReleaseDBScheme>('releases')
     .find({
       _id: {
@@ -75,6 +77,7 @@ async function findReleasesToCheck(db: Db, now: Date): Promise<ReleaseDBScheme[]
  * @param releasesToCheck - ready releases ordered from oldest to newest
  * @param allProjectReleases - complete project release history
  * @param releasesByName - project releases indexed by release name
+ * @returns {number} number of events actually updated
  */
 async function validateEventsBatch(
   events: GroupedEventDBScheme[],
@@ -83,7 +86,12 @@ async function validateEventsBatch(
   releasesToCheck: ReleaseDBScheme[],
   allProjectReleases: ReleaseHistoryEntry[],
   releasesByName: Map<string, ReleaseHistoryEntry>
-): Promise<void> {
+): Promise<number> {
+  const startedAt = Date.now();
+  let resolvedEvents = 0;
+
+  logger.info(`Batch started: collection=${eventsCollection.collectionName} events=${events.length}`);
+
   const eventGroupHashes = events.map(event => event.groupHash);
   const repetitions = await repetitionsCollection.aggregate<Pick<RepetitionDBScheme, 'groupHash' | 'release'>>([
     {
@@ -115,6 +123,8 @@ async function validateEventsBatch(
   }).toArray();
   const eventReleases = buildEventReleaseMap(events, repetitions);
 
+  logger.debug(`Repetition lookup loaded: collection=${eventsCollection.collectionName} releasePairs=${repetitions.length}`);
+
   /**
    * Resolve each event in the first eligible release after its latest known
    * occurrence. A later occurrence blocks resolution in an older release.
@@ -213,7 +223,7 @@ async function validateEventsBatch(
           ],
         };
 
-      await eventsCollection.updateOne({
+      const result = await eventsCollection.updateOne({
         _id: event._id,
         ...eventState,
       }, {
@@ -225,13 +235,21 @@ async function validateEventsBatch(
           : {}),
       });
 
+      resolvedEvents += result.modifiedCount;
+
+      logger.debug(`Event resolution evaluated: eventId=${event._id} release=${release.release} modified=${result.modifiedCount}`);
+
       /**
        * Releases are ordered from oldest to newest, so the first matching
        * candidate is the release in which the event became likely fixed.
        */
       break;
     }
   }
+
+  logger.info(`Batch finished: collection=${eventsCollection.collectionName} events=${events.length} resolvedEvents=${resolvedEvents} durationMs=${Date.now() - startedAt}`);
+
+  return resolvedEvents;
 }
 
 /**
@@ -242,6 +260,12 @@ async function validateEventsBatch(
  * @param releasesToCheck - ready releases ordered from oldest to newest
  */
 async function validateProject(db: Db, projectId: string, releasesToCheck: ReleaseDBScheme[]): Promise<void> {
+  const startedAt = Date.now();
+  let processedEvents = 0;
+  let resolvedEvents = 0;
+
+  logger.info(`Project validation started: projectId=${projectId} releases=${releasesToCheck.length}`);
+
   const releasesCollection = db.collection<ReleaseDBScheme>('releases');
   const eventsCollection = db.collection<GroupedEventDBScheme>(`events:${projectId}`);
   const repetitionsCollection = db.collection<RepetitionDBScheme>(`repetitions:${projectId}`);
@@ -267,6 +291,8 @@ async function validateProject(db: Db, projectId: string, releasesToCheck: Relea
     .sort({ _id: 1 })
     .toArray();
 
+  logger.info(`Release history loaded: projectId=${projectId} releases=${allProjectReleases.length}`);
+
   /**
    * Index the timeline by release name for event-field lookups. The map key is
    * only a lookup key; chronology is still determined by each value's ObjectId.
@@ -334,31 +360,35 @@ async function validateProject(db: Db, projectId: string, releasesToCheck: Relea
       continue;
     }
 
-    await validateEventsBatch(
+    resolvedEvents += await validateEventsBatch(
       eventsBatch,
       eventsCollection,
       repetitionsCollection,
       releasesToCheck,
       allProjectReleases,
       releasesByName
     );
+    processedEvents += eventsBatch.length;
     eventsBatch = [];
   }
 
   /**
    * Process the final partial portion left after the cursor is exhausted.
    */
   if (eventsBatch.length > 0) {
-    await validateEventsBatch(
+    resolvedEvents += await validateEventsBatch(
       eventsBatch,
       eventsCollection,
       repetitionsCollection,
       releasesToCheck,
       allProjectReleases,
       releasesByName
     );
+    processedEvents += eventsBatch.length;
   }
 
+  logger.info(`Marking releases as checked: projectId=${projectId} releases=${releasesToCheck.length}`);
+
   /**
    * Mark releases only after every project batch finishes successfully.
    */
@@ -371,6 +401,8 @@ async function validateProject(db: Db, projectId: string, releasesToCheck: Relea
       fixChecked: true,
     },
   });
+
+  logger.info(`Project validation finished: projectId=${projectId} releases=${releasesToCheck.length} processedEvents=${processedEvents} resolvedEvents=${resolvedEvents} durationMs=${Date.now() - startedAt}`);
 }
 
 /**
@@ -380,18 +412,34 @@ async function validateProject(db: Db, projectId: string, releasesToCheck: Relea
  * @param now - current time
  */
 export async function validateReleases(db: Db, now = new Date()): Promise<void> {
+  const startedAt = Date.now();
+  let failedProjects = 0;
+
+  logger.info('Release validation started');
+
   /**
    * Candidate releases are globally ordered and then grouped so each project's
    * release history, events, and repetitions are loaded only once per run.
    */
   const releasesToCheck = await findReleasesToCheck(db, now);
   const releasesByProject = groupReleasesByProject(releasesToCheck);
 
+  logger.info(`Candidates loaded: releases=${releasesToCheck.length} projects=${releasesByProject.size}`);
+
   for (const [projectId, projectReleases] of releasesByProject) {
     try {
       await validateProject(db, projectId, projectReleases);
     } catch (error) {
-      logger.error(`Failed to validate releases for project ${projectId}`, error);
+      failedProjects++;
+      logger.error(`Project validation failed: projectId=${projectId}`, error);
     }
   }
+
+  const summary = `Release validation finished: projects=${releasesByProject.size} failedProjects=${failedProjects} durationMs=${Date.now() - startedAt}`;
+
+  if (failedProjects > 0) {
+    logger.warn(summary);
+  } else {
+    logger.info(summary);
+  }
 }

workers/release-validator/tests/validate-releases.test.ts

@@ -2,6 +2,20 @@ import '../../../env-test';
 import { Collection, Db, MongoClient, ObjectID } from 'mongodb';
 import type { GroupedEventDBScheme, ReleaseDBScheme, RepetitionDBScheme } from '@hawk.so/types';
 import { validateReleases } from '../src/validate-releases';
+import createLogger from '../../../lib/logger';
+
+jest.mock('../../../lib/logger', () => {
+  const logger = {
+    info: jest.fn(),
+    debug: jest.fn(),
+    warn: jest.fn(),
+    error: jest.fn(),
+  };
+
+  return jest.fn(() => logger);
+});
+
+const logger = createLogger();
 
 const PROJECT_ID = 'release-validator-project';
 const HOUR_IN_SECONDS = 60 * 60;
@@ -103,6 +117,7 @@ describe('validateReleases', () => {
   });
 
   beforeEach(async () => {
+    jest.clearAllMocks();
     process.env.MAX_DAYS_NUMBER = DEFAULT_MAX_DAYS_NUMBER;
     await releases.deleteMany({ projectId: PROJECT_ID });
     await events.deleteMany({});
@@ -169,6 +184,29 @@ describe('validateReleases', () => {
     await validateReleases(db, NOW);
 
     expect((await events.findOne({ groupHash: 'error-4' })).resolvedInRelease).toBe('b');
+    expect(logger.info).toHaveBeenCalledWith(expect.stringContaining(
+      `Project validation finished: projectId=${PROJECT_ID} releases=1 processedEvents=1 resolvedEvents=1`
+    ));
+  });
+
+  test('should log progress and actual resolved totals across event batches', async () => {
+    await releases.insertMany([
+      createRelease('a', 72, true),
+      createRelease('b', 48),
+    ]);
+    await events.insertMany(Array.from({ length: 51 }, (_, index) => createEvent(`batch-event-${index}`, 'a')));
+
+    await validateReleases(db, NOW);
+
+    expect(logger.info).toHaveBeenCalledWith(expect.stringContaining(
+      `Batch finished: collection=events:${PROJECT_ID} events=50 resolvedEvents=50`
+    ));
+    expect(logger.info).toHaveBeenCalledWith(expect.stringContaining(
+      `Batch finished: collection=events:${PROJECT_ID} events=1 resolvedEvents=1`
+    ));
+    expect(logger.info).toHaveBeenCalledWith(expect.stringContaining(
+      `Project validation finished: projectId=${PROJECT_ID} releases=1 processedEvents=51 resolvedEvents=51`
+    ));
   });
 
   test('should resolve an event in the first checked release after its last occurrence', async () => {
@@ -197,6 +235,10 @@ describe('validateReleases', () => {
 
     expect((await events.findOne({ groupHash: 'error-6' })).resolvedInRelease).toBeUndefined();
     expect((await releases.findOne({ release: 'b' })).fixChecked).toBe(false);
+    expect(logger.info).toHaveBeenCalledWith('Candidates loaded: releases=0 projects=0');
+    expect(logger.info).toHaveBeenCalledWith(expect.stringContaining(
+      'Release validation finished: projects=0 failedProjects=0'
+    ));
   });
 
   test('should use the original event timestamp when its release record was archived', async () => {
@@ -402,6 +444,13 @@ describe('validateReleases', () => {
         projectId: PROJECT_ID,
         release: 'c',
       })).fixChecked).toBe(true);
+      expect(logger.error).toHaveBeenCalledWith(
+        `Project validation failed: projectId=${failedProjectId}`,
+        expect.any(Error)
+      );
+      expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining(
+        'Release validation finished: projects=2 failedProjects=1'
+      ));
     } finally {
       await db.dropCollection(`events:${failedProjectId}`).catch(() => undefined);
       await db.dropCollection(`repetitions:${failedProjectId}`).catch(() => undefined);