From 8149ae098c765e4088d38f154e3bdee7319a209b Mon Sep 17 00:00:00 2001 From: Peter Date: Wed, 29 Jul 2026 18:54:47 +0300 Subject: [PATCH 1/6] fix(grouper): fix some ts errors (#581) --- workers/grouper/src/data-filter.ts | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/workers/grouper/src/data-filter.ts b/workers/grouper/src/data-filter.ts index a7c58c04..e55c492e 100644 --- a/workers/grouper/src/data-filter.ts +++ b/workers/grouper/src/data-filter.ts @@ -36,14 +36,14 @@ const MAX_CODE_LINE_LENGTH = 140; * @param callback - Function to call on each iteration */ function forAll(obj: Record, callback: (path: string[], key: string, obj: Record) => void): void { - const visit = (current, path: string[]): void => { + const visit = (current: Record, path: string[]): void => { for (const key in current) { if (!Object.prototype.hasOwnProperty.call(current, key)) { continue; } const value = current[key]; - if (!(typeof value === 'object' && !Array.isArray(value))) { + if (!(typeof value === 'object' && value !== null && !Array.isArray(value))) { callback(path, key, current); } else { /** @@ -51,7 +51,7 @@ function forAll(obj: Record, callback: (path: string[], key: st * This reduces GC pressure and memory usage for deeply nested objects */ const newPath = path.length < MAX_TRAVERSAL_DEPTH ? path.concat(key) : path; - visit(value, newPath); + visit(value as Record, newPath); } } }; @@ -264,11 +264,12 @@ export default class DataFilter { * * @param field - any object to iterate */ - private processField(field): void { - if (typeof field === 'string') { + private processField(field: unknown): void { + if (typeof field !== 'object' || field === null || Array.isArray(field)) { return; } - forAll(field, (_path, key, obj) => { + + forAll(field as Record, (_path, key, obj) => { obj[key] = this.filterPanNumbers(obj[key]); obj[key] = this.filterSensitiveData(key, obj[key]); }); From f5af8bc545191fddc425e6b649122bcd00651b8d Mon Sep 17 00:00:00 2001 From: Kuchizu <70284260+Kuchizu@users.noreply.github.com> Date: Wed, 26 Aug 2026 20:57:33 +0300 Subject: [PATCH 2/6] Use dailyEvents counter for all workspaces in limiter (#584) --- .env.sample | 3 - workers/limiter/src/index.ts | 128 ++++++++++++++-------------- workers/limiter/tests/index.test.ts | 33 +++---- 3 files changed, 77 insertions(+), 87 deletions(-) diff --git a/.env.sample b/.env.sample index a5ecb4dc..f2df9be2 100644 --- a/.env.sample +++ b/.env.sample @@ -55,8 +55,5 @@ HAWK_CATCHER_TOKEN= ## If true, Grouper worker will send messages about new events to Notifier worker IS_NOTIFIER_WORKER_ENABLED=false -## Comma-separated workspace ids that should use dailyEvents counters in Limiter quota checks. Use * for all workspaces. -LIMITER_DAILY_EVENTS_COUNTER_WORKSPACE_IDS= - ## Url for telegram notifications about workspace blocks and unblocks TELEGRAM_LIMITER_CHAT_URL= diff --git a/workers/limiter/src/index.ts b/workers/limiter/src/index.ts index cc38d768..9df8c970 100644 --- a/workers/limiter/src/index.ts +++ b/workers/limiter/src/index.ts @@ -266,7 +266,7 @@ export default class LimiterWorker extends Worker { const since = Math.floor(new Date(workspace.lastChargeDate).getTime() / MS_IN_SEC); - const workspaceEventsCount = await this.getWorkspaceEventsCount(workspace, projects, since); + const workspaceEventsCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); this.logger.info(`workspace ${workspace._id} events count since last charge date: ${workspaceEventsCount}`); @@ -328,68 +328,70 @@ export default class LimiterWorker extends Worker { }; } - /** - * Returns workspace events count using the default raw counter or the - * dailyEvents-based counter when it is explicitly enabled for the workspace. - * - * For enabled workspaces both counters are computed and their results with - * timings are reported to Telegram to compare the algorithms during the - * testing period. The old counter is used as a fallback if the new one fails. - * - * @param workspace - workspace to count events for - * @param projects - workspace projects - * @param since - timestamp of the time from which we count the events - */ - private async getWorkspaceEventsCount( - workspace: WorkspaceWithTariffPlan, - projects: ProjectDBScheme[], - since: number - ): Promise { - if (!this.shouldUseDailyEventsCounter(workspace._id.toString())) { - return this.dbHelper.getEventsCountByProjects(projects, since); - } - - const oldAlgoStartedAt = Date.now(); - const oldAlgoCount = await this.dbHelper.getEventsCountByProjects(projects, since); - const oldAlgoTook = (Date.now() - oldAlgoStartedAt) / MS_IN_SEC; - - try { - const newAlgoStartedAt = Date.now(); - const newAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); - const newAlgoTook = (Date.now() - newAlgoStartedAt) / MS_IN_SEC; - - telegram.sendMessage( - `Workspace ${workspace.name} event count:\n` + - `Old algo: ${oldAlgoCount}, took ${oldAlgoTook}sec\n` + - `New algo: ${newAlgoCount}, took ${newAlgoTook}sec`, - telegram.TelegramBotURLs.Limiter - ); - - return newAlgoCount; - } catch (error) { - HawkCatcher.send(error, { - workspaceId: workspace._id.toString(), - }); - - return oldAlgoCount; - } - } - - /** - * Checks whether dailyEvents-based quota counting is enabled for the workspace - * via LIMITER_DAILY_EVENTS_COUNTER_WORKSPACE_IDS environment variable — - * comma-separated workspace ids or `*` to enable it for every workspace. - * - * @param workspaceId - workspace id - */ - private shouldUseDailyEventsCounter(workspaceId: string): boolean { - const enabledWorkspaceIds = (process.env.LIMITER_DAILY_EVENTS_COUNTER_WORKSPACE_IDS || '') - .split(',') - .map(id => id.trim()) - .filter(Boolean); - - return enabledWorkspaceIds.includes('*') || enabledWorkspaceIds.includes(workspaceId); - } + // Old raw counter with the opt-in switch, kept in case we need to roll back + // + // /** + // * Returns workspace events count using the default raw counter or the + // * dailyEvents-based counter when it is explicitly enabled for the workspace. + // * + // * For enabled workspaces both counters are computed and their results with + // * timings are reported to Telegram to compare the algorithms during the + // * testing period. The old counter is used as a fallback if the new one fails. + // * + // * @param workspace - workspace to count events for + // * @param projects - workspace projects + // * @param since - timestamp of the time from which we count the events + // */ + // private async getWorkspaceEventsCount( + // workspace: WorkspaceWithTariffPlan, + // projects: ProjectDBScheme[], + // since: number + // ): Promise { + // if (!this.shouldUseDailyEventsCounter(workspace._id.toString())) { + // return this.dbHelper.getEventsCountByProjects(projects, since); + // } + // + // const oldAlgoStartedAt = Date.now(); + // const oldAlgoCount = await this.dbHelper.getEventsCountByProjects(projects, since); + // const oldAlgoTook = (Date.now() - oldAlgoStartedAt) / MS_IN_SEC; + // + // try { + // const newAlgoStartedAt = Date.now(); + // const newAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); + // const newAlgoTook = (Date.now() - newAlgoStartedAt) / MS_IN_SEC; + // + // telegram.sendMessage( + // `Workspace ${workspace.name} event count:\n` + + // `Old algo: ${oldAlgoCount}, took ${oldAlgoTook}sec\n` + + // `New algo: ${newAlgoCount}, took ${newAlgoTook}sec`, + // telegram.TelegramBotURLs.Limiter + // ); + // + // return newAlgoCount; + // } catch (error) { + // HawkCatcher.send(error, { + // workspaceId: workspace._id.toString(), + // }); + // + // return oldAlgoCount; + // } + // } + // + // /** + // * Checks whether dailyEvents-based quota counting is enabled for the workspace + // * via LIMITER_DAILY_EVENTS_COUNTER_WORKSPACE_IDS environment variable — + // * comma-separated workspace ids or `*` to enable it for every workspace. + // * + // * @param workspaceId - workspace id + // */ + // private shouldUseDailyEventsCounter(workspaceId: string): boolean { + // const enabledWorkspaceIds = (process.env.LIMITER_DAILY_EVENTS_COUNTER_WORKSPACE_IDS || '') + // .split(',') + // .map(id => id.trim()) + // .filter(Boolean); + // + // return enabledWorkspaceIds.includes('*') || enabledWorkspaceIds.includes(workspaceId); + // } /** * Method that formats project list to html used in report messages diff --git a/workers/limiter/tests/index.test.ts b/workers/limiter/tests/index.test.ts index 1c6c7648..3fd54a5a 100644 --- a/workers/limiter/tests/index.test.ts +++ b/workers/limiter/tests/index.test.ts @@ -331,7 +331,7 @@ describe('Limiter worker', () => { expect(reportMessage).toContain(`${project1.name} (id: ${project1._id})`); }); - test('Should compute both counters and report the comparison to Telegram when dailyEvents counter is enabled', async () => { + test('Should count events via dailyEvents counters for every workspace', async () => { /** * Arrange */ @@ -349,7 +349,7 @@ describe('Limiter worker', () => { }); /** - * Bucket for the day after the boundary day — counted only by the new algorithm + * Bucket for the day after the boundary day — counted via dailyEvents */ await db.collection(`dailyEvents:${project._id.toString()}`).insertOne({ groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62', @@ -357,23 +357,17 @@ describe('Limiter worker', () => { count: 7, }); - process.env.LIMITER_DAILY_EVENTS_COUNTER_WORKSPACE_IDS = workspace._id.toString(); - /** * Act */ - try { - const worker = new LimiterWorker(); - - await worker.start(); - await worker.handle(REGULAR_WORKSPACES_CHECK_EVENT); - await worker.finish(); - } finally { - delete process.env.LIMITER_DAILY_EVENTS_COUNTER_WORKSPACE_IDS; - } + const worker = new LimiterWorker(); + + await worker.start(); + await worker.handle(REGULAR_WORKSPACES_CHECK_EVENT); + await worker.finish(); /** - * Assert — the new counter result is saved, both results are reported with timings + * Assert */ const workspaceInDatabase = await workspaceCollection.findOne({ _id: workspace._id, @@ -381,13 +375,10 @@ describe('Limiter worker', () => { expect(workspaceInDatabase.billingPeriodEventsCount).toBe(12); // 5 boundary-day events + 7 from dailyEvents - const comparisonMessage = (telegram.sendMessage as jest.Mock).mock.calls - .map(call => call[0]) - .find(message => message.includes('Old algo')); - - expect(comparisonMessage).toContain(`Workspace ${workspace.name} event count:`); - expect(comparisonMessage).toMatch(/Old algo: 5, took [\d.]+sec/); - expect(comparisonMessage).toMatch(/New algo: 12, took [\d.]+sec/); + /** + * Counters comparison is not reported to Telegram anymore + */ + expect(telegram.sendMessage).not.toHaveBeenCalled(); }); test('Should not send a report when no projects are blocked or unblocked', async () => { From 39ae9b7036ffe58d4d386c50c15acd34e963dc06 Mon Sep 17 00:00:00 2001 From: Kuchizu <70284260+Kuchizu@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:05:41 +0300 Subject: [PATCH 3/6] Stop limiter from opening every events collection each hour (#587) * Stop limiter from opening every events collection each hour * Comment limiter dailyEvents aggregation sections * Report sampled limiter counter validation to Telegram * Compare old and new limiter algo for listed workspaces instead of sampling * Use the previous dailyEvents query for old algo comparison --- .env.sample | 3 + workers/limiter/src/dbHelper.ts | 85 ++++++++++++++++++++++-- workers/limiter/src/index.ts | 38 ++++++++++- workers/limiter/tests/dbHelper.test.ts | 90 +++++++++++++++++++++++++- workers/limiter/tests/index.test.ts | 48 ++++++++++++++ 5 files changed, 258 insertions(+), 6 deletions(-) diff --git a/.env.sample b/.env.sample index f2df9be2..5c19ff64 100644 --- a/.env.sample +++ b/.env.sample @@ -57,3 +57,6 @@ IS_NOTIFIER_WORKER_ENABLED=false ## Url for telegram notifications about workspace blocks and unblocks TELEGRAM_LIMITER_CHAT_URL= + +## Workspace ids to compare old and new limiter counts in telegram +LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS= diff --git a/workers/limiter/src/dbHelper.ts b/workers/limiter/src/dbHelper.ts index 984dd335..0d0e71e5 100644 --- a/workers/limiter/src/dbHelper.ts +++ b/workers/limiter/src/dbHelper.ts @@ -184,6 +184,7 @@ export class DbHelper { * increments `count` for originals and repetitions alike); only the * partial day containing `since` is counted from the raw collections, * since dailyEvents buckets have day granularity and lastChargeDate does not. + * Raw collections are skipped if that day's bucket is empty. * * @param project - project to check * @param since - timestamp of the time from which we count the events @@ -191,6 +192,83 @@ export class DbHelper { public async getEventsCountByProjectUsingDailyEvents( project: ProjectDBScheme, since: number + ): Promise { + try { + const projectId = project._id.toString(); + const dailyEventsCollection = this.eventsDbConnection.collection('dailyEvents:' + projectId); + const boundaryDayTimestamp = since - (since % SEC_IN_DAY); + const firstFullDayTimestamp = this.getFirstFullDailyEventsTimestamp(since); + + const [ counters ] = await dailyEventsCollection + .aggregate<{ boundaryDay: number; fullDays: number }>([ + /** buckets from the day containing `since` onwards */ + { $match: { groupingTimestamp: { $gte: boundaryDayTimestamp } } }, + { + $group: { + _id: null, + /** whole boundary day, only gates the raw count below */ + boundaryDay: { + $sum: { $cond: [ { $lt: ['$groupingTimestamp', firstFullDayTimestamp] }, '$count', 0] }, + }, + /** days after the boundary day */ + fullDays: { + $sum: { $cond: [ { $gte: ['$groupingTimestamp', firstFullDayTimestamp] }, '$count', 0] }, + }, + }, + }, + ], { + /** one table per project instead of racing all groupingTimestamp indexes */ + hint: { $natural: 1 }, + }) + .toArray(); + + /** no buckets in the billing period */ + if (!counters) { + return 0; + } + + /** the bucket spans the whole day, so the part after `since` is counted from raw events */ + const boundaryDayCount = counters.boundaryDay > 0 + ? await this.getRawEventsCountByProject(project, { + timestamp: { + $gt: since, + $lt: firstFullDayTimestamp, + }, + }) + : 0; + + return boundaryDayCount + counters.fullDays; + } catch (e) { + HawkCatcher.send(e); + throw new CriticalError(e); + } + } + + /** + * Calculates total events count for all provided projects since the specific date + * using dailyEvents counters for full days. + * + * @param projects - projects to calculate for + * @param since - timestamp of the time from which we count the events + */ + public async getEventsCountByProjectsUsingDailyEvents(projects: ProjectDBScheme[], since: number): Promise { + const sum = (array: number[]): number => array.reduce((acc, val) => acc + val, 0); + + return Promise.all(projects.map( + project => this.getEventsCountByProjectUsingDailyEvents(project, since) + )) + .then(sum); + } + + /** + * Previous query, kept for rollout comparison + * + * @param project - project to check + * @param since - timestamp of the time from which we count the events + */ + public async getEventsCountByProjectUsingDailyEventsOld( + project: ProjectDBScheme, + since: number ): Promise { try { const projectId = project._id.toString(); @@ -231,17 +309,16 @@ export class DbHelper { } /** - * Calculates total events count for all provided projects since the specific date - * using dailyEvents counters for full days. + * Previous query, kept for rollout comparison * * @param projects - projects to calculate for * @param since - timestamp of the time from which we count the events */ - public async getEventsCountByProjectsUsingDailyEvents(projects: ProjectDBScheme[], since: number): Promise { + public async getEventsCountByProjectsUsingDailyEventsOld(projects: ProjectDBScheme[], since: number): Promise { const sum = (array: number[]): number => array.reduce((acc, val) => acc + val, 0); return Promise.all(projects.map( - project => this.getEventsCountByProjectUsingDailyEvents(project, since) + project => this.getEventsCountByProjectUsingDailyEventsOld(project, since) )) .then(sum); } diff --git a/workers/limiter/src/index.ts b/workers/limiter/src/index.ts index 9df8c970..1b3ca6cd 100644 --- a/workers/limiter/src/index.ts +++ b/workers/limiter/src/index.ts @@ -266,7 +266,7 @@ export default class LimiterWorker extends Worker { const since = Math.floor(new Date(workspace.lastChargeDate).getTime() / MS_IN_SEC); - const workspaceEventsCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); + const workspaceEventsCount = await this.getWorkspaceEventsCount(workspace, projects, since); this.logger.info(`workspace ${workspace._id} events count since last charge date: ${workspaceEventsCount}`); @@ -328,6 +328,42 @@ export default class LimiterWorker extends Worker { }; } + /** + * Counts workspace events, comparing with the previous query for LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS + * + * @param workspace - workspace to count events for + * @param projects - workspace projects + * @param since - timestamp of the time from which we count the events + */ + private async getWorkspaceEventsCount( + workspace: WorkspaceWithTariffPlan, + projects: ProjectDBScheme[], + since: number + ): Promise { + const compareWorkspaceIds = (process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS || '').split(',').map(id => id.trim()); + + if (!compareWorkspaceIds.includes(workspace._id.toString())) { + return this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); + } + + const oldAlgoStartedAt = Date.now(); + const oldAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEventsOld(projects, since); + const oldAlgoTook = (Date.now() - oldAlgoStartedAt) / MS_IN_SEC; + + const newAlgoStartedAt = Date.now(); + const newAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); + const newAlgoTook = (Date.now() - newAlgoStartedAt) / MS_IN_SEC; + + telegram.sendMessage( + `Workspace ${workspace.name} event count:\n` + + `Old algo: ${oldAlgoCount}, took ${oldAlgoTook}sec\n` + + `New algo: ${newAlgoCount}, took ${newAlgoTook}sec`, + telegram.TelegramBotURLs.Limiter + ); + + return newAlgoCount; + } + // Old raw counter with the opt-in switch, kept in case we need to roll back // // /** diff --git a/workers/limiter/tests/dbHelper.test.ts b/workers/limiter/tests/dbHelper.test.ts index dd26baa7..210d486f 100644 --- a/workers/limiter/tests/dbHelper.test.ts +++ b/workers/limiter/tests/dbHelper.test.ts @@ -23,6 +23,9 @@ const BOUNDARY_DAY_TIMESTAMP = 1585756800; */ const NEXT_MIDNIGHT_AFTER_LAST_CHARGE = 1585785600; +/** 2020-04-01T00:00:00Z */ +const BOUNDARY_DAY_MIDNIGHT = NEXT_MIDNIGHT_AFTER_LAST_CHARGE - 86400; + describe('DbHelper', () => { let connection: MongoClient; let db: Db; @@ -132,6 +135,17 @@ describe('DbHelper', () => { await repetitionsCollection.insertMany(mockedEvents); } + /** as grouper does */ + const boundaryDayEventsCount = parameters.eventsToMock + (parameters.repetitionsToMock ?? 0); + + if (boundaryDayEventsCount > 0) { + await dailyEventsCollection.insertOne({ + groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62', + groupingTimestamp: BOUNDARY_DAY_MIDNIGHT, + count: boundaryDayEventsCount, + }); + } + if (parameters.dailyEventsToMock?.length > 0) { await dailyEventsCollection.insertMany(parameters.dailyEventsToMock.map(bucket => ({ groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62', @@ -711,7 +725,7 @@ describe('DbHelper', () => { dailyEventsToMock: [ /** bucket of the boundary day itself must not be counted */ { - groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE - 86400, + groupingTimestamp: BOUNDARY_DAY_MIDNIGHT, count: 100, }, ], @@ -779,6 +793,80 @@ describe('DbHelper', () => { */ expect(count).toBe(7); }); + + test('Should not query raw collections when the boundary day bucket is empty', async () => { + /** + * Arrange + */ + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10, + billingPeriodEventsCount: 0, + lastChargeDate: new Date(), + }); + const project = createProjectMock({ workspaceId: workspace._id }); + const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC); + + await fillDatabaseWithMockedData({ + workspace, + project, + eventsToMock: 0, + dailyEventsToMock: [ + { + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE, + count: 4, + }, + ], + }); + + const collectionSpy = jest.spyOn(db, 'collection'); + + /** + * Act + */ + const count = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); + + /** + * Assert + */ + expect(count).toBe(4); + expect(collectionSpy.mock.calls.map(([ name ]) => name)).toEqual([ `dailyEvents:${project._id.toString()}` ]); + + collectionSpy.mockRestore(); + }); + + test('Should return zero for a project without dailyEvents collection', async () => { + const project = createProjectMock({ workspaceId: new ObjectId() }); + const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC); + + const count = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); + + expect(count).toBe(0); + }); + }); + + describe('getEventsCountByProjectUsingDailyEventsOld', () => { + test('Should count raw boundary-day events even without a dailyEvents bucket', async () => { + const project = createProjectMock({ workspaceId: new ObjectId() }); + const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC); + + await fillDatabaseWithMockedData({ + project, + eventsToMock: 0, + dailyEventsToMock: [ + { + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE, + count: 3, + }, + ], + }); + await db.collection(`events:${project._id.toString()}`).insertMany([createEventMock(), createEventMock()]); + + const oldCount = await dbHelper.getEventsCountByProjectUsingDailyEventsOld(project, since); + const newCount = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); + + expect(oldCount).toBe(5); + expect(newCount).toBe(3); + }); }); describe('getEventsCountByProjectsUsingDailyEvents', () => { diff --git a/workers/limiter/tests/index.test.ts b/workers/limiter/tests/index.test.ts index 3fd54a5a..d233a06f 100644 --- a/workers/limiter/tests/index.test.ts +++ b/workers/limiter/tests/index.test.ts @@ -138,6 +138,13 @@ describe('Limiter worker', () => { } await repetitionsCollection.insertMany(mockedEvents); } + + /** as grouper does */ + await db.collection(`dailyEvents:${parameters.project._id.toString()}`).insertOne({ + groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62', + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE - 86400, + count: parameters.eventsToMock + (parameters.repetitionsToMock ?? 0), + }); }; beforeAll(async () => { @@ -381,6 +388,47 @@ describe('Limiter worker', () => { expect(telegram.sendMessage).not.toHaveBeenCalled(); }); + test('Should report old and new algo counts for workspaces listed for comparison', async () => { + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10000, + billingPeriodEventsCount: 0, + lastChargeDate: LAST_CHARGE_DATE, + }); + const project = createProjectMock({ workspaceId: workspace._id }); + + await fillDatabaseWithMockedData({ + workspace, + project, + eventsToMock: 5, + }); + + await db.collection(`dailyEvents:${project._id.toString()}`).deleteMany({}); + + process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS = workspace._id.toString(); + + const worker = new LimiterWorker(); + + try { + await worker.start(); + await worker.handle(REGULAR_WORKSPACES_CHECK_EVENT); + await worker.finish(); + } finally { + delete process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS; + } + + const workspaceInDatabase = await workspaceCollection.findOne({ + _id: workspace._id, + }); + + expect(workspaceInDatabase.billingPeriodEventsCount).toBe(0); + expect(telegram.sendMessage).toHaveBeenCalledTimes(1); + + const reportMessage = (telegram.sendMessage as jest.Mock).mock.calls[0][0]; + + expect(reportMessage).toContain('Old algo: 5'); + expect(reportMessage).toContain('New algo: 0'); + }); + test('Should not send a report when no projects are blocked or unblocked', async () => { /** * Arrange From a6b17578b80a114a0578454bb8d1a04132d439eb Mon Sep 17 00:00:00 2001 From: Kuchizu <70284260+Kuchizu@users.noreply.github.com> Date: Thu, 17 Sep 2026 22:26:06 +0300 Subject: [PATCH 4/6] Remove old limiter algo comparison (#591) --- .env.sample | 3 -- workers/limiter/src/dbHelper.ts | 63 -------------------------- workers/limiter/src/index.ts | 38 +--------------- workers/limiter/tests/dbHelper.test.ts | 25 ---------- workers/limiter/tests/index.test.ts | 41 ----------------- 5 files changed, 1 insertion(+), 169 deletions(-) diff --git a/.env.sample b/.env.sample index 5c19ff64..f2df9be2 100644 --- a/.env.sample +++ b/.env.sample @@ -57,6 +57,3 @@ IS_NOTIFIER_WORKER_ENABLED=false ## Url for telegram notifications about workspace blocks and unblocks TELEGRAM_LIMITER_CHAT_URL= - -## Workspace ids to compare old and new limiter counts in telegram -LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS= diff --git a/workers/limiter/src/dbHelper.ts b/workers/limiter/src/dbHelper.ts index 0d0e71e5..e6948e06 100644 --- a/workers/limiter/src/dbHelper.ts +++ b/workers/limiter/src/dbHelper.ts @@ -260,69 +260,6 @@ export class DbHelper { .then(sum); } - /** - * Previous query, kept for rollout comparison - * - * @param project - project to check - * @param since - timestamp of the time from which we count the events - */ - public async getEventsCountByProjectUsingDailyEventsOld( - project: ProjectDBScheme, - since: number - ): Promise { - try { - const projectId = project._id.toString(); - const dailyEventsCollection = this.eventsDbConnection.collection('dailyEvents:' + projectId); - const firstFullDayTimestamp = this.getFirstFullDailyEventsTimestamp(since); - - const boundaryDayQuery = { - timestamp: { - $gt: since, - $lt: firstFullDayTimestamp, - }, - }; - - const [boundaryDayCount, dailyCounters] = await Promise.all([ - since < firstFullDayTimestamp - ? this.getRawEventsCountByProject(project, boundaryDayQuery) - : 0, - dailyEventsCollection - .aggregate<{ count: number }>([ - { $match: { groupingTimestamp: { $gte: firstFullDayTimestamp } } }, - { - $group: { - _id: null, - count: { $sum: '$count' }, - }, - }, - ]) - .toArray(), - ]); - - const fullDaysCount = dailyCounters.length > 0 ? dailyCounters[0].count : 0; - - return boundaryDayCount + fullDaysCount; - } catch (e) { - HawkCatcher.send(e); - throw new CriticalError(e); - } - } - - /** - * Previous query, kept for rollout comparison - * - * @param projects - projects to calculate for - * @param since - timestamp of the time from which we count the events - */ - public async getEventsCountByProjectsUsingDailyEventsOld(projects: ProjectDBScheme[], since: number): Promise { - const sum = (array: number[]): number => array.reduce((acc, val) => acc + val, 0); - - return Promise.all(projects.map( - project => this.getEventsCountByProjectUsingDailyEventsOld(project, since) - )) - .then(sum); - } - /** * Returns all projects from Database or projects of the specified workspace * diff --git a/workers/limiter/src/index.ts b/workers/limiter/src/index.ts index 1b3ca6cd..9df8c970 100644 --- a/workers/limiter/src/index.ts +++ b/workers/limiter/src/index.ts @@ -266,7 +266,7 @@ export default class LimiterWorker extends Worker { const since = Math.floor(new Date(workspace.lastChargeDate).getTime() / MS_IN_SEC); - const workspaceEventsCount = await this.getWorkspaceEventsCount(workspace, projects, since); + const workspaceEventsCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); this.logger.info(`workspace ${workspace._id} events count since last charge date: ${workspaceEventsCount}`); @@ -328,42 +328,6 @@ export default class LimiterWorker extends Worker { }; } - /** - * Counts workspace events, comparing with the previous query for LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS - * - * @param workspace - workspace to count events for - * @param projects - workspace projects - * @param since - timestamp of the time from which we count the events - */ - private async getWorkspaceEventsCount( - workspace: WorkspaceWithTariffPlan, - projects: ProjectDBScheme[], - since: number - ): Promise { - const compareWorkspaceIds = (process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS || '').split(',').map(id => id.trim()); - - if (!compareWorkspaceIds.includes(workspace._id.toString())) { - return this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); - } - - const oldAlgoStartedAt = Date.now(); - const oldAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEventsOld(projects, since); - const oldAlgoTook = (Date.now() - oldAlgoStartedAt) / MS_IN_SEC; - - const newAlgoStartedAt = Date.now(); - const newAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); - const newAlgoTook = (Date.now() - newAlgoStartedAt) / MS_IN_SEC; - - telegram.sendMessage( - `Workspace ${workspace.name} event count:\n` + - `Old algo: ${oldAlgoCount}, took ${oldAlgoTook}sec\n` + - `New algo: ${newAlgoCount}, took ${newAlgoTook}sec`, - telegram.TelegramBotURLs.Limiter - ); - - return newAlgoCount; - } - // Old raw counter with the opt-in switch, kept in case we need to roll back // // /** diff --git a/workers/limiter/tests/dbHelper.test.ts b/workers/limiter/tests/dbHelper.test.ts index 210d486f..bf9aa46c 100644 --- a/workers/limiter/tests/dbHelper.test.ts +++ b/workers/limiter/tests/dbHelper.test.ts @@ -844,31 +844,6 @@ describe('DbHelper', () => { }); }); - describe('getEventsCountByProjectUsingDailyEventsOld', () => { - test('Should count raw boundary-day events even without a dailyEvents bucket', async () => { - const project = createProjectMock({ workspaceId: new ObjectId() }); - const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC); - - await fillDatabaseWithMockedData({ - project, - eventsToMock: 0, - dailyEventsToMock: [ - { - groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE, - count: 3, - }, - ], - }); - await db.collection(`events:${project._id.toString()}`).insertMany([createEventMock(), createEventMock()]); - - const oldCount = await dbHelper.getEventsCountByProjectUsingDailyEventsOld(project, since); - const newCount = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); - - expect(oldCount).toBe(5); - expect(newCount).toBe(3); - }); - }); - describe('getEventsCountByProjectsUsingDailyEvents', () => { test('Should count events, repetitions and dailyEvents for multiple projects', async () => { /** diff --git a/workers/limiter/tests/index.test.ts b/workers/limiter/tests/index.test.ts index d233a06f..f49c987b 100644 --- a/workers/limiter/tests/index.test.ts +++ b/workers/limiter/tests/index.test.ts @@ -388,47 +388,6 @@ describe('Limiter worker', () => { expect(telegram.sendMessage).not.toHaveBeenCalled(); }); - test('Should report old and new algo counts for workspaces listed for comparison', async () => { - const workspace = createWorkspaceMock({ - plan: mockedPlans.eventsLimit10000, - billingPeriodEventsCount: 0, - lastChargeDate: LAST_CHARGE_DATE, - }); - const project = createProjectMock({ workspaceId: workspace._id }); - - await fillDatabaseWithMockedData({ - workspace, - project, - eventsToMock: 5, - }); - - await db.collection(`dailyEvents:${project._id.toString()}`).deleteMany({}); - - process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS = workspace._id.toString(); - - const worker = new LimiterWorker(); - - try { - await worker.start(); - await worker.handle(REGULAR_WORKSPACES_CHECK_EVENT); - await worker.finish(); - } finally { - delete process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS; - } - - const workspaceInDatabase = await workspaceCollection.findOne({ - _id: workspace._id, - }); - - expect(workspaceInDatabase.billingPeriodEventsCount).toBe(0); - expect(telegram.sendMessage).toHaveBeenCalledTimes(1); - - const reportMessage = (telegram.sendMessage as jest.Mock).mock.calls[0][0]; - - expect(reportMessage).toContain('Old algo: 5'); - expect(reportMessage).toContain('New algo: 0'); - }); - test('Should not send a report when no projects are blocked or unblocked', async () => { /** * Arrange From 9d9c98a632be04d37d59684152d2a0080e58cb2d Mon Sep 17 00:00:00 2001 From: Kuchizu <70284260+Kuchizu@users.noreply.github.com> Date: Sat, 26 Sep 2026 19:57:49 +0300 Subject: [PATCH 5/6] Exit runner when a worker fails to start (#594) --- runner.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/runner.ts b/runner.ts index 1c19f97b..e5f20d56 100644 --- a/runner.ts +++ b/runner.ts @@ -154,7 +154,9 @@ class WorkerRunner { utils.sendReport(worker.constructor.name + ' failed to start'); - await this.stopWorker(worker); + await this.finishAll(); + + process.exit(1); } }) ); From 84e0d9393f32e157fd2fac96ed22d16bcc342f79 Mon Sep 17 00:00:00 2001 From: alisawavezen12 Date: Wed, 14 Oct 2026 18:41:49 +0300 Subject: [PATCH 6/6] Add release validator and regression tracking (#590) * Add release validator worker * Track event regressions by release * chore * Support repeated resolution and regression cycles * validate Events Batch * Add release validator to Docker Compose * chore * chore * chore * chore * Align release validation with retention policy Use MAX_DAYS_NUMBER to define the candidate release window and fall back to event timestamps when original release records are archived. * Isolate release validation failures by project * Clear regression when resolving event again * Track latest event occurrence before validation * tests fix * chore * chore --- .gitignore | 2 + docker-compose.dev.yml | 15 + docker-compose.prod.yml | 12 +- package.json | 4 +- .../grouper/src/check-and-mark-regression.ts | 55 +++ workers/grouper/src/index.ts | 22 + workers/grouper/tests/index.test.ts | 120 +++++ workers/release-validator/README.md | 15 + workers/release-validator/jest.config.js | 18 + workers/release-validator/jest.setup.js | 1 + workers/release-validator/package.json | 10 + workers/release-validator/src/index.ts | 46 ++ .../src/utils/build-event-release-map.ts | 45 ++ .../src/utils/group-releases-by-project.ts | 19 + .../src/validate-releases.ts | 397 +++++++++++++++ .../utils/build-event-release-map.test.ts | 97 ++++ .../utils/group-releases-by-project.test.ts | 54 +++ .../tests/validate-releases.test.ts | 452 ++++++++++++++++++ workers/task-manager/src/index.ts | 4 +- .../types/project-task-manager-config.ts | 19 + yarn.lock | 8 +- 21 files changed, 1407 insertions(+), 8 deletions(-) create mode 100644 workers/grouper/src/check-and-mark-regression.ts create mode 100644 workers/release-validator/README.md create mode 100644 workers/release-validator/jest.config.js create mode 100644 workers/release-validator/jest.setup.js create mode 100644 workers/release-validator/package.json create mode 100644 workers/release-validator/src/index.ts create mode 100644 workers/release-validator/src/utils/build-event-release-map.ts create mode 100644 workers/release-validator/src/utils/group-releases-by-project.ts create mode 100644 workers/release-validator/src/validate-releases.ts create mode 100644 workers/release-validator/tests/utils/build-event-release-map.test.ts create mode 100644 workers/release-validator/tests/utils/group-releases-by-project.test.ts create mode 100644 workers/release-validator/tests/validate-releases.test.ts create mode 100644 workers/task-manager/types/project-task-manager-config.ts diff --git a/.gitignore b/.gitignore index b4b45798..2507214f 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,8 @@ globalConfig.json !jest.setup.mongo-repl-set.js !jest.setup.redis-mock.js !jest-mongodb-config.js +!workers/release-validator/jest.config.js +!workers/release-validator/jest.setup.js !migrate-mongo-config.js !/env.js !convertors/**/*.js diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 29766b23..cc131d23 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -62,6 +62,7 @@ services: # # System workers: # - archiver + # - release-validator # - limiter # - paymaster # @@ -79,6 +80,20 @@ services: - ./:/usr/src/app - workers-deps:/usr/src/app/node_modules + hawk-worker-release-validator: + build: + dockerfile: "dev.Dockerfile" + context: . + env_file: + - .env + environment: + - SIMULTANEOUS_TASKS=1 + restart: unless-stopped + entrypoint: yarn run-release-validator + volumes: + - ./:/usr/src/app + - workers-deps:/usr/src/app/node_modules + hawk-worker-limiter: build: dockerfile: "dev.Dockerfile" diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 88a6e0fa..a702aefc 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -43,7 +43,7 @@ services: entrypoint: /usr/local/bin/node runner.js hawk-worker-grouper # - # System workers: archiver + # System workers: archiver, release-validator # hawk-worker-archiver: @@ -55,6 +55,16 @@ services: restart: unless-stopped entrypoint: /usr/local/bin/node runner.js hawk-worker-archiver + hawk-worker-release-validator: + image: "codexteamuser/hawk-workers:prod" + network_mode: host + env_file: + - .env + environment: + - SIMULTANEOUS_TASKS=1 + restart: unless-stopped + entrypoint: /usr/local/bin/node runner.js hawk-worker-release-validator + # # Notification workers: notifier, email, telegram # diff --git a/package.json b/package.json index f2e1e154..b5bb807e 100644 --- a/package.json +++ b/package.json @@ -25,6 +25,7 @@ "test:sentry": "jest workers/sentry --config workers/sentry/jest.config.js", "test:javascript": "jest workers/javascript", "test:release": "jest workers/release", + "test:release-validator": "jest --config workers/release-validator/jest.config.js", "test:slack": "jest workers/slack", "test:loop": "jest workers/loop", "test:limiter": "jest workers/limiter --runInBand", @@ -47,6 +48,7 @@ "run-paymaster": "yarn worker hawk-worker-paymaster", "run-notifier": "yarn worker hawk-worker-notifier", "run-release": "yarn worker hawk-worker-release", + "run-release-validator": "yarn worker hawk-worker-release-validator", "run-email": "yarn worker hawk-worker-email", "run-telegram": "yarn worker hawk-worker-telegram", "run-limiter": "yarn worker hawk-worker-limiter", @@ -57,7 +59,7 @@ "@babel/parser": "^7.26.9", "@babel/traverse": "7.26.9", "@hawk.so/nodejs": "^3.1.1", - "@hawk.so/types": "^0.5.9", + "@hawk.so/types": "^0.9.0", "@types/amqplib": "^0.8.2", "@types/jest": "^29.5.14", "@types/mongodb": "^3.5.15", diff --git a/workers/grouper/src/check-and-mark-regression.ts b/workers/grouper/src/check-and-mark-regression.ts new file mode 100644 index 00000000..53c59689 --- /dev/null +++ b/workers/grouper/src/check-and-mark-regression.ts @@ -0,0 +1,55 @@ +import { Db } from 'mongodb'; +import type { ReleaseDBScheme } from '@hawk.so/types'; + +type ReleaseRecordPart = Pick; + +/** + * Mark an event as regressed if it reoccurs in the resolved or a newer release. + * + * The update is atomic: only the first repetition after resolution sets the + * regression release, and later repetitions do not overwrite it. + * + * @param db - events database connection + * @param projectId - project identifier + * @param groupHash - original event group hash + * @param release - release in which the event occurred again + * @param resolvedInRelease - release in which the event was resolved + */ +export async function checkAndMarkRegression( + db: Db, + projectId: string, + groupHash: string, + release: string, + resolvedInRelease: string +): Promise { + const releases = await db.collection('releases').find({ + projectId, + release: { + $in: [resolvedInRelease, release], + }, + }) + .toArray(); + const resolvedRelease = releases.find(item => item.release === resolvedInRelease); + const repetitionRelease = releases.find(item => item.release === release); + + if (!resolvedRelease || !repetitionRelease) { + return; + } + + const resolvedReleaseId = resolvedRelease._id.toHexString(); + const isResolvedOrNewerRelease = repetitionRelease._id.toHexString() >= resolvedReleaseId; + + if (!isResolvedOrNewerRelease) { + return; + } + + await db.collection(`events:${projectId}`).updateOne({ + groupHash, + resolvedInRelease, + regressionInRelease: { $exists: false }, + }, { + $set: { + regressionInRelease: release, + }, + }); +} diff --git a/workers/grouper/src/index.ts b/workers/grouper/src/index.ts index 6750d768..60e9cb25 100644 --- a/workers/grouper/src/index.ts +++ b/workers/grouper/src/index.ts @@ -24,6 +24,7 @@ import GrouperMetrics from './metrics/grouperMetrics'; import GrouperMemoryMonitor from './metrics/memoryMonitor'; import SlowHandleDiagnostics, { SlowHandleSession } from './metrics/slowHandleDiagnostics'; import { grouperDiagnosticsConfig, grouperMemoryConfig } from './metrics/config'; +import { checkAndMarkRegression } from './check-and-mark-regression'; /** * eslint does not count decorators as a variable usage @@ -343,10 +344,31 @@ export default class GrouperWorker extends Worker { timestamp: task.timestamp, } as RepetitionDBScheme; + if (task.payload.release) { + newRepetition.release = task.payload.release; + } + repetitionId = await session.measureStep('saveRepetition', () => { return this.saveRepetition(task.projectId, newRepetition); }); + if (task.payload.release && existedEvent.resolvedInRelease && !existedEvent.regressionInRelease) { + try { + await checkAndMarkRegression( + this.eventsDb.getConnection(), + task.projectId, + uniqueEventHash, + task.payload.release, + existedEvent.resolvedInRelease + ); + } catch (error) { + this.logger.error( + `[checkAndMarkRegression] project=${task.projectId} groupHash=${uniqueEventHash} release=${task.payload.release}`, + error + ); + } + } + /** * Clear the large event payload references to allow garbage collection * This prevents memory leaks from retaining full event objects after delta is computed diff --git a/workers/grouper/tests/index.test.ts b/workers/grouper/tests/index.test.ts index e4f2b3bd..c71d03ea 100644 --- a/workers/grouper/tests/index.test.ts +++ b/workers/grouper/tests/index.test.ts @@ -158,6 +158,7 @@ describe('GrouperWorker', () => { await eventsCollection.deleteMany({}); await dailyEventsCollection.deleteMany({}); await repetitionsCollection.deleteMany({}); + await connection.db().collection('releases').deleteMany({ projectId: projectIdMock }); }); afterEach(async () => { @@ -391,6 +392,125 @@ describe('GrouperWorker', () => { }).toArray()).length).toBe(2); }); + test('Should save repetition release as a separate field', async () => { + await worker.handle(generateTask({ release: 'release-a' })); + await worker.handle(generateTask({ release: 'release-b' })); + + const savedRepetition = await repetitionsCollection.findOne({}); + + expect(savedRepetition.release).toBe('release-b'); + }); + + test('Should not save repetition release when event has no release', async () => { + await worker.handle(generateTask()); + await worker.handle(generateTask()); + + const savedRepetition = await repetitionsCollection.findOne({}); + + expect(savedRepetition.release).toBeUndefined(); + }); + + describe('Regression marking', () => { + test('Should mark a resolved event as regressed in a newer repetition release', async () => { + await connection.db().collection('releases').insertMany([ + { + _id: mongodb.ObjectID.createFromTime(1), + projectId: projectIdMock, + release: 'release-b', + }, + { + _id: mongodb.ObjectID.createFromTime(2), + projectId: projectIdMock, + release: 'release-c', + }, + ]); + await worker.handle(generateTask({ release: 'release-a' })); + await eventsCollection.updateOne({}, { + $set: { + resolvedInRelease: 'release-b', + }, + }); + + await worker.handle(generateTask({ release: 'release-c' })); + + expect((await eventsCollection.findOne({})).regressionInRelease).toBe('release-c'); + }); + + test('Should mark as regressed if we later encounter this event with a release that is considered a resolving release', async () => { + await connection.db().collection('releases').insertOne({ + _id: mongodb.ObjectID.createFromTime(1), + projectId: projectIdMock, + release: 'release-b', + }); + await worker.handle(generateTask({ release: 'release-a' })); + await eventsCollection.updateOne({}, { + $set: { + resolvedInRelease: 'release-b', + }, + }); + + await worker.handle(generateTask({ release: 'release-b' })); + + expect((await eventsCollection.findOne({})).regressionInRelease).toBe('release-b'); + }); + + test('Should not mark regression if we encounter an event from one of the old releases', async () => { + await connection.db().collection('releases').insertMany([ + { + _id: mongodb.ObjectID.createFromTime(1), + projectId: projectIdMock, + release: 'release-a', + }, + { + _id: mongodb.ObjectID.createFromTime(2), + projectId: projectIdMock, + release: 'release-b', + }, + ]); + await worker.handle(generateTask({ release: 'release-b' })); + await eventsCollection.updateOne({}, { + $set: { + resolvedInRelease: 'release-b', + }, + }); + + await worker.handle(generateTask({ release: 'release-a' })); + + expect((await eventsCollection.findOne({})).regressionInRelease).toBeUndefined(); + }); + + test('Should not overwrite the first marked regression if we continue receiving this event in newer releases', async () => { + await connection.db().collection('releases').insertMany([ + { + _id: mongodb.ObjectID.createFromTime(1), + projectId: projectIdMock, + release: 'release-b', + }, + { + _id: mongodb.ObjectID.createFromTime(2), + projectId: projectIdMock, + release: 'release-c', + }, + { + _id: mongodb.ObjectID.createFromTime(3), + projectId: projectIdMock, + release: 'release-d', + }, + ]); + await worker.handle(generateTask({ release: 'release-a' })); + await eventsCollection.updateOne({}, { + $set: { + resolvedInRelease: 'release-b', + regressionInRelease: 'release-c', + }, + }); + + await worker.handle(generateTask({ release: 'release-d' })); + + expect((await eventsCollection.findOne({})).regressionInRelease).toBe('release-c'); + }); + }); + test('Should stringify delta', async () => { const generatedTask = generateTask(); diff --git a/workers/release-validator/README.md b/workers/release-validator/README.md new file mode 100644 index 00000000..a64e9665 --- /dev/null +++ b/workers/release-validator/README.md @@ -0,0 +1,15 @@ +# Release Validator Worker + +Checks releases after a 24-hour observation period and marks original events that no longer occur as resolved. + +The worker processes releases from oldest to newest, stores the first release without an event in `resolvedInRelease`, and marks successfully processed releases with `fixChecked: true`. + +The worker only uses the fields required for matching releases and event groups. Records without these fields are ignored. Candidate releases must be older than 24 hours and no older than the `MAX_DAYS_NUMBER` retention period, while all available project releases are still used to compare event history. If an event's original release was archived, the original event timestamp is used as its chronological boundary. + +Queue: `cron-tasks/release-validator` + +Run locally: + +```sh +yarn run-release-validator +``` diff --git a/workers/release-validator/jest.config.js b/workers/release-validator/jest.config.js new file mode 100644 index 00000000..2b8182ed --- /dev/null +++ b/workers/release-validator/jest.config.js @@ -0,0 +1,18 @@ +const baseConfig = require('../../jest.config'); + +module.exports = { + ...baseConfig, + rootDir: '../..', + setupFiles: [ + '/jest.setup.js', + '/workers/release-validator/jest.setup.js', + ], + setupFilesAfterEnv: [ '/jest.setup.mongo-repl-set.js' ], + globalTeardown: '/jest.global-teardown.js', + roots: [ + '/workers/release-validator', + '/lib', + ], + testMatch: [ '/workers/release-validator/**/*.test.ts' ], + moduleFileExtensions: ['ts', 'tsx', 'js', 'json', 'node'], +}; diff --git a/workers/release-validator/jest.setup.js b/workers/release-validator/jest.setup.js new file mode 100644 index 00000000..d0c9bf1e --- /dev/null +++ b/workers/release-validator/jest.setup.js @@ -0,0 +1 @@ +process.env.MAX_DAYS_NUMBER = process.env.MAX_DAYS_NUMBER || '30'; diff --git a/workers/release-validator/package.json b/workers/release-validator/package.json new file mode 100644 index 00000000..3cd8c88f --- /dev/null +++ b/workers/release-validator/package.json @@ -0,0 +1,10 @@ +{ + "name": "hawk-worker-release-validator", + "version": "0.0.1", + "description": "Detects events fixed by a release", + "main": "src/index.ts", + "author": "CodeX", + "license": "UNLICENSED", + "private": true, + "workerType": "cron-tasks/release-validator" +} diff --git a/workers/release-validator/src/index.ts b/workers/release-validator/src/index.ts new file mode 100644 index 00000000..ac268f6a --- /dev/null +++ b/workers/release-validator/src/index.ts @@ -0,0 +1,46 @@ +import { DatabaseController } from '../../../lib/db/controller'; +import { Worker } from '../../../lib/worker'; +import * as pkg from '../package.json'; +import { validateReleases } from './validate-releases'; + +/** + * Worker that detects events fixed by a release. + */ +export default class ReleaseValidatorWorker extends Worker { + /** + * Worker type. + */ + public readonly type: string = pkg.workerType; + + /** + * Events database controller. + */ + private eventsDb = new DatabaseController(process.env.MONGO_EVENTS_DATABASE_URI); + + /** + * Connect to the events database and start consuming tasks. + */ + public async start(): Promise { + await this.eventsDb.connect(); + await super.start(); + } + + /** + * Stop consuming tasks and close the database connection. + */ + public async finish(): Promise { + await super.finish(); + await this.eventsDb.close(); + } + + /** + * Handle a scheduled release validation task. + */ + public async handle(): Promise { + this.logger.info('Release validation started'); + + await validateReleases(this.eventsDb.getConnection()); + + this.logger.info('Release validation finished'); + } +} diff --git a/workers/release-validator/src/utils/build-event-release-map.ts b/workers/release-validator/src/utils/build-event-release-map.ts new file mode 100644 index 00000000..13db2382 --- /dev/null +++ b/workers/release-validator/src/utils/build-event-release-map.ts @@ -0,0 +1,45 @@ +import type { GroupedEventDBScheme, RepetitionDBScheme } from '@hawk.so/types'; + +type RepetitionRelease = Pick; + +/** + * Build a lookup set of releases in which each event occurred. + * + * Repetitions are deduplicated by event and release in MongoDB before this + * function is called. Therefore, memory usage depends on the number of unique + * releases per event rather than the potentially much larger repetition count. + * Release ordering is handled separately by the validation flow. + * + * @param events - original events from the current validation batch + * @param repetitions - unique event and release pairs from MongoDB + */ +export function buildEventReleaseMap( + events: GroupedEventDBScheme[], + repetitions: RepetitionRelease[] +): Map> { + const releasesByGroupHash = new Map>(); + + for (const event of events) { + const releases = new Set(); + + if (event.payload?.release) { + releases.add(event.payload.release); + } + + releasesByGroupHash.set(event.groupHash, releases); + } + + for (const repetition of repetitions) { + if (!repetition.release) { + continue; + } + + const releases = releasesByGroupHash.get(repetition.groupHash); + + if (releases) { + releases.add(repetition.release); + } + } + + return releasesByGroupHash; +} diff --git a/workers/release-validator/src/utils/group-releases-by-project.ts b/workers/release-validator/src/utils/group-releases-by-project.ts new file mode 100644 index 00000000..efd30ae3 --- /dev/null +++ b/workers/release-validator/src/utils/group-releases-by-project.ts @@ -0,0 +1,19 @@ +import type { ReleaseDBScheme } from '@hawk.so/types'; + +/** + * Group chronologically ordered releases by project. + * + * @param releases - releases ordered from oldest to newest + */ +export function groupReleasesByProject(releases: ReleaseDBScheme[]): Map { + const releasesByProject = new Map(); + + for (const release of releases) { + const projectReleases = releasesByProject.get(release.projectId) || []; + + projectReleases.push(release); + releasesByProject.set(release.projectId, projectReleases); + } + + return releasesByProject; +} diff --git a/workers/release-validator/src/validate-releases.ts b/workers/release-validator/src/validate-releases.ts new file mode 100644 index 00000000..f8c4910a --- /dev/null +++ b/workers/release-validator/src/validate-releases.ts @@ -0,0 +1,397 @@ +import { Collection, Db, ObjectID } from 'mongodb'; +import type { GroupedEventDBScheme, ReleaseDBScheme, RepetitionDBScheme } from '@hawk.so/types'; +import { HOURS_IN_DAY, MINUTES_IN_HOUR, MS_IN_SEC, SECONDS_IN_MINUTE } from '../../../lib/utils/consts'; +import createLogger from '../../../lib/logger'; +import { buildEventReleaseMap } from './utils/build-event-release-map'; +import { groupReleasesByProject } from './utils/group-releases-by-project'; + +type ReleaseHistoryEntry = Pick; + +const logger = createLogger(); + +/** + * Time allowed for repetitions to arrive before a release is checked for + * resolved events. + */ +const RELEASE_OBSERVATION_PERIOD_SECONDS = HOURS_IN_DAY * MINUTES_IN_HOUR * SECONDS_IN_MINUTE; + +/** + * Maximum number of events processed in one validation batch. + * + * A bounded batch keeps the repetitions `$in` query below MongoDB document + * limits and prevents large projects from loading all events and repetitions + * into worker memory at once. + */ +const EVENTS_BATCH_SIZE = 50; + +/** + * Find unchecked releases older than the observation period. + * + * @param db - events database connection + * @param now - current time + */ +async function findReleasesToCheck(db: Db, now: Date): Promise { + const nowSeconds = Math.floor(now.getTime() / MS_IN_SEC); + const releaseMaxAgeDays = Number(process.env.MAX_DAYS_NUMBER); + + if (!Number.isFinite(releaseMaxAgeDays) || releaseMaxAgeDays <= 0) { + throw new Error('MAX_DAYS_NUMBER must be a positive number'); + } + + /** + * Use the archiver retention period as the candidate window so releases are + * validated only while their records are expected to remain available. + */ + const releaseMaxAgeSeconds = releaseMaxAgeDays * RELEASE_OBSERVATION_PERIOD_SECONDS; + const oldestReleaseId = ObjectID.createFromTime(nowSeconds - releaseMaxAgeSeconds); + const newestReleaseId = ObjectID.createFromTime(nowSeconds - RELEASE_OBSERVATION_PERIOD_SECONDS); + + return db.collection('releases') + .find({ + _id: { + $gte: oldestReleaseId, + $lt: newestReleaseId, + }, + release: { + $type: 'string', + $ne: '', + }, + fixChecked: { $ne: true }, + }) + .sort({ _id: 1 }) + .toArray(); +} + +/** + * Validate a bounded portion of project events. + * + * Only the repetition fields required to build the occurrence lookup are + * projected. This avoids transferring serialized deltas and payloads that are + * not used by release validation. + * + * @param events - current event portion + * @param eventsCollection - project events collection + * @param repetitionsCollection - project repetitions collection + * @param releasesToCheck - ready releases ordered from oldest to newest + * @param allProjectReleases - complete project release history + * @param releasesByName - project releases indexed by release name + */ +async function validateEventsBatch( + events: GroupedEventDBScheme[], + eventsCollection: Collection, + repetitionsCollection: Collection, + releasesToCheck: ReleaseDBScheme[], + allProjectReleases: ReleaseHistoryEntry[], + releasesByName: Map +): Promise { + const eventGroupHashes = events.map(event => event.groupHash); + const repetitions = await repetitionsCollection.aggregate>([ + { + $match: { + groupHash: { $in: eventGroupHashes }, + release: { + $type: 'string', + $ne: '', + }, + }, + }, + { + $group: { + _id: { + groupHash: '$groupHash', + release: '$release', + }, + }, + }, + { + $project: { + _id: 0, + groupHash: '$_id.groupHash', + release: '$_id.release', + }, + }, + ], { + hint: 'groupHashAndRelease', + }).toArray(); + const eventReleases = buildEventReleaseMap(events, repetitions); + + /** + * Resolve each event in the first eligible release after its latest known + * occurrence. A later occurrence blocks resolution in an older release. + */ + for (const event of events) { + /** + * The original event release is the initial occurrence boundary. + */ + const originalRelease = releasesByName.get(event.payload.release); + let lastOccurrenceRelease = originalRelease; + + /** + * An event can be resolved, reappear in a later release, and then stop + * occurring again. In that case, search for the next resolving release + * after the regression rather than after the event's original occurrence. + */ + if (event.resolvedInRelease && event.regressionInRelease) { + const resolvedRelease = releasesByName.get(event.resolvedInRelease); + const regressionRelease = releasesByName.get(event.regressionInRelease); + + /** + * Without both release records their chronological relation is unknown, + * so the event cannot be safely resolved again. + */ + if (!resolvedRelease || !regressionRelease) { + continue; + } + + const isCurrentlyRegressed = regressionRelease._id.toHexString() >= resolvedRelease._id.toHexString(); + + /** + * A regression older than the current resolution belongs to a previous + * cycle and does not make the event eligible for another resolution. + */ + if (!isCurrentlyRegressed) { + continue; + } + + lastOccurrenceRelease = regressionRelease; + } + + /** + * The archiver can remove the original release while its event is still + * active. In that case, use the original event occurrence time as the + * chronological boundary instead of skipping the event forever. + */ + let lastOccurrenceId = lastOccurrenceRelease?._id; + + if (!lastOccurrenceId) { + if (!Number.isFinite(event.timestamp)) { + continue; + } + + lastOccurrenceId = ObjectID.createFromTime(Math.floor(event.timestamp)); + } + + const releasesWithEvent = eventReleases.get(event.groupHash) || new Set(); + + /** + * Find the latest known release in which the event occurred once per event. + * This includes releases younger than 24 hours, which must still prevent an + * older candidate from resolving the event. + */ + for (const projectRelease of allProjectReleases) { + if ( + projectRelease._id.toHexString() > lastOccurrenceId.toHexString() && + releasesWithEvent.has(projectRelease.release) + ) { + lastOccurrenceId = projectRelease._id; + } + } + + for (const release of releasesToCheck) { + const isNewerThanLastOccurrence = release._id.toHexString() > lastOccurrenceId.toHexString(); + + if (!isNewerThanLastOccurrence) { + continue; + } + + /** + * Resolve only the state that was evaluated in this batch: an unresolved + * event must still be unresolved, while a regressed event must retain the + * same resolution and regression releases. Including that state in the + * update also makes the transition atomic, so a concurrent Grouper update + * cannot be overwritten. + */ + const eventState = event.regressionInRelease + ? { + resolvedInRelease: event.resolvedInRelease, + regressionInRelease: event.regressionInRelease, + } + : { + $or: [ + { resolvedInRelease: { $exists: false } }, + { resolvedInRelease: null }, + ], + }; + + await eventsCollection.updateOne({ + _id: event._id, + ...eventState, + }, { + $set: { + resolvedInRelease: release.release, + }, + ...(event.regressionInRelease + ? { $unset: { regressionInRelease: '' } } + : {}), + }); + + /** + * Releases are ordered from oldest to newest, so the first matching + * candidate is the release in which the event became likely fixed. + */ + break; + } + } +} + +/** + * Validate all ready releases of one project. + * + * @param db - events database connection + * @param projectId - project identifier + * @param releasesToCheck - ready releases ordered from oldest to newest + */ +async function validateProject(db: Db, projectId: string, releasesToCheck: ReleaseDBScheme[]): Promise { + const releasesCollection = db.collection('releases'); + const eventsCollection = db.collection(`events:${projectId}`); + const repetitionsCollection = db.collection(`repetitions:${projectId}`); + + /** + * Load the project's release timeline once. Releases outside the validation + * window are required because an old original release or a newer occurrence + * can change whether a candidate release resolved an event. Only identifiers + * and names are loaded; source maps and commit data are not used here. + */ + const allProjectReleases = await releasesCollection + .find({ + projectId, + release: { + $type: 'string', + $ne: '', + }, + }) + .project({ + _id: 1, + release: 1, + }) + .sort({ _id: 1 }) + .toArray(); + + /** + * 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. + */ + const releasesByName = new Map(); + + for (const release of allProjectReleases) { + releasesByName.set(release.release, release); + } + + /** + * Find events whose resolution state needs evaluation. An event must have a + * group hash and an original release, and must either be unresolved or have + * both resolution and regression releases for a subsequent resolution cycle. + * + * Stream the result instead of materializing all project events. The cursor + * and application batch use the same limit so the worker holds at most one + * bounded portion of event documents in memory. + */ + const eventsCursor = eventsCollection.find({ + groupHash: { + $type: 'string', + $ne: '', + }, + 'payload.release': { + $type: 'string', + $ne: '', + }, + $or: [ + { resolvedInRelease: { $exists: false } }, + { resolvedInRelease: null }, + { + resolvedInRelease: { + $type: 'string', + $ne: '', + }, + regressionInRelease: { + $type: 'string', + $ne: '', + }, + }, + ], + }, { + projection: { + _id: 1, + groupHash: 1, + 'payload.release': 1, + timestamp: 1, + resolvedInRelease: 1, + regressionInRelease: 1, + }, + }).batchSize(EVENTS_BATCH_SIZE); + let eventsBatch: GroupedEventDBScheme[] = []; + + while (await eventsCursor.hasNext()) { + const event = await eventsCursor.next(); + + if (!event) { + break; + } + + eventsBatch.push(event); + + if (eventsBatch.length < EVENTS_BATCH_SIZE) { + continue; + } + + await validateEventsBatch( + eventsBatch, + eventsCollection, + repetitionsCollection, + releasesToCheck, + allProjectReleases, + releasesByName + ); + eventsBatch = []; + } + + /** + * Process the final partial portion left after the cursor is exhausted. + */ + if (eventsBatch.length > 0) { + await validateEventsBatch( + eventsBatch, + eventsCollection, + repetitionsCollection, + releasesToCheck, + allProjectReleases, + releasesByName + ); + } + + /** + * Mark releases only after every project batch finishes successfully. + */ + await releasesCollection.updateMany({ + _id: { + $in: releasesToCheck.map(release => release._id), + }, + }, { + $set: { + fixChecked: true, + }, + }); +} + +/** + * Validate all releases whose observation period has elapsed. + * + * @param db - events database connection + * @param now - current time + */ +export async function validateReleases(db: Db, now = new Date()): Promise { + /** + * 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); + + for (const [projectId, projectReleases] of releasesByProject) { + try { + await validateProject(db, projectId, projectReleases); + } catch (error) { + logger.error(`Failed to validate releases for project ${projectId}`, error); + } + } +} diff --git a/workers/release-validator/tests/utils/build-event-release-map.test.ts b/workers/release-validator/tests/utils/build-event-release-map.test.ts new file mode 100644 index 00000000..a9d6025f --- /dev/null +++ b/workers/release-validator/tests/utils/build-event-release-map.test.ts @@ -0,0 +1,97 @@ +import { ObjectID } from 'mongodb'; +import type { GroupedEventDBScheme, RepetitionDBScheme } from '@hawk.so/types'; +import { buildEventReleaseMap } from '../../src/utils/build-event-release-map'; + +/** + * Create an original event. + * + * @param groupHash - event group hash + * @param release - release in which the event first occurred + */ +function createEvent(groupHash: string, release?: string): GroupedEventDBScheme { + return { + _id: new ObjectID(), + groupHash, + payload: { + title: groupHash, + ...(release ? { release } : {}), + }, + totalCount: 1, + catcherType: 'errors/default', + usersAffected: 0, + visitedBy: [], + timestamp: 1, + }; +} + +/** + * Create an event repetition. + * + * @param groupHash - event group hash + * @param release - release in which the event occurred + */ +function createRepetition(groupHash: string, release?: string): RepetitionDBScheme { + return { + groupHash, + timestamp: 1, + ...(release ? { release } : {}), + }; +} + +describe('buildEventReleaseMap', () => { + test('should build the release map from original events and repetitions', () => { + // Arrange + const events = [ + createEvent('error-1', 'a'), + createEvent('error-2', 'a'), + createEvent('error-3', 'a'), + ]; + const repetitions = [ + createRepetition('error-1', 'b'), + createRepetition('error-1', 'c'), + createRepetition('error-2', 'b'), + createRepetition('error-3', 'b'), + createRepetition('error-3', 'c'), + createRepetition('error-3', 'd'), + createRepetition('error-3', 'e'), + ]; + + // Act + const result = buildEventReleaseMap(events, repetitions); + + // Assert + expect(Array.from(result.entries()).map(([groupHash, releases]) => { + return [groupHash, Array.from(releases)]; + })).toEqual([ + ['error-1', ['a', 'b', 'c'] ], + ['error-2', ['a', 'b'] ], + ['error-3', ['a', 'b', 'c', 'd', 'e'] ], + ]); + }); + + test('should ignore duplicate releases and repetitions without matching events', () => { + // Arrange + const events = [ + createEvent('error-1', 'a'), + createEvent('error-without-release'), + ]; + const repetitions = [ + createRepetition('error-1', 'a'), + createRepetition('error-1', 'b'), + createRepetition('error-1', 'b'), + createRepetition('error-1'), + createRepetition('unknown-error', 'c'), + ]; + + // Act + const result = buildEventReleaseMap(events, repetitions); + + // Assert + expect(Array.from(result.entries()).map(([groupHash, releases]) => { + return [groupHash, Array.from(releases)]; + })).toEqual([ + ['error-1', ['a', 'b'] ], + ['error-without-release', [] ], + ]); + }); +}); diff --git a/workers/release-validator/tests/utils/group-releases-by-project.test.ts b/workers/release-validator/tests/utils/group-releases-by-project.test.ts new file mode 100644 index 00000000..0e53fe78 --- /dev/null +++ b/workers/release-validator/tests/utils/group-releases-by-project.test.ts @@ -0,0 +1,54 @@ +import { ObjectID } from 'mongodb'; +import type { ReleaseDBScheme } from '@hawk.so/types'; +import { groupReleasesByProject } from '../../src/utils/group-releases-by-project'; + +/** + * Create a release record with a predictable id. + * + * @param projectId - project identifier + * @param release - release name + * @param createdAtSeconds - release creation time + */ +function createRelease(projectId: string, release: string, createdAtSeconds: number): ReleaseDBScheme { + return { + _id: ObjectID.createFromTime(createdAtSeconds), + projectId, + release, + commits: [], + }; +} + +describe('groupReleasesByProject', () => { + test('should group releases by project and preserve their input order', () => { + // Arrange + const releases = [ + createRelease('project-a', 'a', 1), + createRelease('project-b', 'x', 2), + createRelease('project-a', 'b', 3), + createRelease('project-b', 'y', 4), + createRelease('project-a', 'c', 5), + ]; + + // Act + const result = groupReleasesByProject(releases); + + // Assert + expect(Array.from(result.entries()).map(([projectId, projectReleases]) => { + return [projectId, projectReleases.map(release => release.release)]; + })).toEqual([ + ['project-a', ['a', 'b', 'c'] ], + ['project-b', ['x', 'y'] ], + ]); + }); + + test('should return an empty map for an empty release list', () => { + // Arrange + const releases: ReleaseDBScheme[] = []; + + // Act + const result = groupReleasesByProject(releases); + + // Assert + expect(Array.from(result.entries())).toEqual([]); + }); +}); diff --git a/workers/release-validator/tests/validate-releases.test.ts b/workers/release-validator/tests/validate-releases.test.ts new file mode 100644 index 00000000..2d2d980e --- /dev/null +++ b/workers/release-validator/tests/validate-releases.test.ts @@ -0,0 +1,452 @@ +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'; + +const PROJECT_ID = 'release-validator-project'; +const HOUR_IN_SECONDS = 60 * 60; +const DEFAULT_MAX_DAYS_NUMBER = '30'; +const NOW_SECONDS = Math.floor(new Date('2026-09-17T12:00:00.000Z').getTime() / 1000); +const NOW = new Date(NOW_SECONDS * 1000); + +/** + * Create a release ObjectId with a predictable creation time. + * + * @param hoursBeforeNow - release age in hours + */ +function releaseId(hoursBeforeNow: number): ObjectID { + return ObjectID.createFromTime(NOW_SECONDS - hoursBeforeNow * HOUR_IN_SECONDS); +} + +/** + * Create a release record for the test project. + * + * @param release - release name + * @param hoursBeforeNow - release age in hours + * @param fixChecked - whether the release has already been checked + */ +function createRelease(release: string, hoursBeforeNow: number, fixChecked = false): ReleaseDBScheme { + return { + _id: releaseId(hoursBeforeNow), + projectId: PROJECT_ID, + release, + commits: [], + fixChecked, + }; +} + +/** + * Create an original event. + * + * @param groupHash - event group hash + * @param release - release in which the event first occurred + * @param resolvedInRelease - existing resolved release + */ +function createEvent(groupHash: string, release: string, resolvedInRelease?: string): GroupedEventDBScheme { + const event: GroupedEventDBScheme = { + _id: new ObjectID(), + groupHash, + payload: { + title: groupHash, + release, + }, + totalCount: 1, + catcherType: 'errors/default', + usersAffected: 0, + visitedBy: [], + timestamp: NOW_SECONDS, + }; + + if (resolvedInRelease !== undefined) { + event.resolvedInRelease = resolvedInRelease; + } + + return event; +} + +/** + * Create an event repetition. + * + * @param groupHash - event group hash + * @param release - release in which the event occurred + */ +function createRepetition(groupHash: string, release: string): RepetitionDBScheme { + return { + groupHash, + release, + timestamp: NOW_SECONDS, + }; +} + +describe('validateReleases', () => { + let connection: MongoClient; + let db: Db; + let releases: Collection; + let events: Collection; + let repetitions: Collection; + + beforeAll(async () => { + connection = await MongoClient.connect(process.env.MONGO_EVENTS_DATABASE_URI, { + useNewUrlParser: true, + useUnifiedTopology: true, + }); + db = connection.db(); + releases = db.collection('releases'); + events = db.collection(`events:${PROJECT_ID}`); + repetitions = db.collection(`repetitions:${PROJECT_ID}`); + await repetitions.createIndex({ + groupHash: 1, + release: 1, + }, { + name: 'groupHashAndRelease', + }); + }); + + beforeEach(async () => { + process.env.MAX_DAYS_NUMBER = DEFAULT_MAX_DAYS_NUMBER; + await releases.deleteMany({ projectId: PROJECT_ID }); + await events.deleteMany({}); + await repetitions.deleteMany({}); + }); + + afterAll(async () => { + await releases.deleteMany({ projectId: PROJECT_ID }); + await events.drop().catch(() => undefined); + await repetitions.drop().catch(() => undefined); + await connection.close(); + }); + + test('should not resolve an event that first appeared in the checked release', async () => { + await releases.insertMany([ + createRelease('a', 72, true), + createRelease('b', 48), + ]); + await events.insertOne(createEvent('error-1', 'b')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-1' })).resolvedInRelease).toBeUndefined(); + expect((await releases.findOne({ release: 'b' })).fixChecked).toBe(true); + }); + + test('should not resolve an event that occurred in the checked release', async () => { + await releases.insertMany([ + createRelease('a', 72, true), + createRelease('b', 48), + ]); + await events.insertOne(createEvent('error-2', 'a')); + await repetitions.insertOne(createRepetition('error-2', 'b')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-2' })).resolvedInRelease).toBeUndefined(); + }); + + test('should not resolve an event that reappeared in a newer release within 24 hours', async () => { + await releases.insertMany([ + createRelease('a', 72, true), + createRelease('b', 48), + createRelease('c', 2), + ]); + await events.insertOne(createEvent('error-3', 'a')); + await repetitions.insertOne(createRepetition('error-3', 'c')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-3' })).resolvedInRelease).toBeUndefined(); + expect((await releases.findOne({ release: 'b' })).fixChecked).toBe(true); + expect((await releases.findOne({ release: 'c' })).fixChecked).toBe(false); + }); + + test('should resolve an event in the checked release when it never appears again', async () => { + await releases.insertMany([ + createRelease('a', 72, true), + createRelease('b', 48), + createRelease('c', 2), + ]); + await events.insertOne(createEvent('error-4', 'a')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-4' })).resolvedInRelease).toBe('b'); + }); + + test('should resolve an event in the first checked release after its last occurrence', async () => { + await releases.insertMany([ + createRelease('a', 120, true), + createRelease('b', 96), + createRelease('c', 72), + createRelease('d', 48), + ]); + await events.insertOne(createEvent('error-5', 'a')); + await repetitions.insertOne(createRepetition('error-5', 'c')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-5' })).resolvedInRelease).toBe('d'); + }); + + test('should not check a release until its 24-hour observation period has elapsed', async () => { + await releases.insertMany([ + createRelease('a', 48, true), + createRelease('b', 23), + ]); + await events.insertOne(createEvent('error-6', 'a')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-6' })).resolvedInRelease).toBeUndefined(); + expect((await releases.findOne({ release: 'b' })).fixChecked).toBe(false); + }); + + test('should use the original event timestamp when its release record was archived', async () => { + await releases.insertOne(createRelease('b', 48)); + const event = createEvent('error-7', 'archived-release'); + + event.timestamp = NOW_SECONDS - 72 * HOUR_IN_SECONDS; + await events.insertOne(event); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-7' })).resolvedInRelease).toBe('b'); + expect((await releases.findOne({ release: 'b' })).fixChecked).toBe(true); + }); + + test('should not resolve an event before its timestamp when its release record is missing', async () => { + await releases.insertOne(createRelease('b', 48)); + const event = createEvent('error-after-release', 'missing-release'); + + event.timestamp = NOW_SECONDS - 24 * HOUR_IN_SECONDS; + await events.insertOne(event); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-after-release' })).resolvedInRelease).toBeUndefined(); + }); + + test('should not overwrite an existing resolved release on repeated validation', async () => { + await releases.insertMany([ + createRelease('a', 72, true), + createRelease('b', 48), + ]); + await events.insertOne(createEvent('error-8', 'a', 'previous-release')); + + await validateReleases(db, NOW); + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-8' })).resolvedInRelease).toBe('previous-release'); + }); + + test('should skip a malformed release and continue with the next valid release', async () => { + await releases.insertMany([ + createRelease('a', 96, true), + { + _id: releaseId(72), + projectId: PROJECT_ID, + release: 123 as unknown as string, + commits: [], + }, + createRelease('d', 48), + ]); + await events.insertOne(createEvent('error-9', 'a')); + + await expect(validateReleases(db, NOW)).resolves.toBeUndefined(); + + expect((await events.findOne({ groupHash: 'error-9' })).resolvedInRelease).toBe('d'); + expect((await releases.findOne({ release: 123 as unknown as string })).fixChecked).toBeUndefined(); + expect((await releases.findOne({ release: 'd' })).fixChecked).toBe(true); + }); + + test('should skip a malformed event without stopping valid events', async () => { + await releases.insertMany([ + createRelease('a', 72, true), + createRelease('b', 48), + ]); + await events.insertMany([ + { + _id: new ObjectID(), + groupHash: 'broken-event', + payload: { + title: 'broken-event', + }, + totalCount: 1, + catcherType: 'errors/default', + usersAffected: 0, + visitedBy: [], + timestamp: NOW_SECONDS, + }, + createEvent('valid-event', 'a'), + ]); + + await expect(validateReleases(db, NOW)).resolves.toBeUndefined(); + + expect((await events.findOne({ groupHash: 'broken-event' })).resolvedInRelease).toBeUndefined(); + expect((await events.findOne({ groupHash: 'valid-event' })).resolvedInRelease).toBe('b'); + }); + + test('should ignore a repetition without a release', async () => { + await releases.insertMany([ + createRelease('a', 72, true), + createRelease('b', 48), + ]); + await events.insertOne(createEvent('error-10', 'a')); + await repetitions.insertOne({ + groupHash: 'error-10', + timestamp: NOW_SECONDS, + }); + + await expect(validateReleases(db, NOW)).resolves.toBeUndefined(); + + expect((await events.findOne({ groupHash: 'error-10' })).resolvedInRelease).toBe('b'); + }); + + test('should ignore a repetition from an unknown release', async () => { + await releases.insertMany([ + createRelease('a', 72, true), + createRelease('b', 48), + ]); + await events.insertOne(createEvent('error-11', 'a')); + await repetitions.insertOne(createRepetition('error-11', 'unknown')); + + await expect(validateReleases(db, NOW)).resolves.toBeUndefined(); + + expect((await events.findOne({ groupHash: 'error-11' })).resolvedInRelease).toBe('b'); + }); + + test('should resolve an event again after its regression stops occurring', async () => { + await releases.insertMany([ + createRelease('a', 120, true), + createRelease('b', 96, true), + createRelease('c', 72, true), + createRelease('d', 48), + ]); + const event = createEvent('error-cycle', 'a', 'b'); + + event.regressionInRelease = 'c'; + await events.insertOne(event); + await repetitions.insertOne(createRepetition('error-cycle', 'c')); + + await validateReleases(db, NOW); + + const updatedEvent = await events.findOne({ groupHash: 'error-cycle' }); + + expect(updatedEvent.resolvedInRelease).toBe('d'); + expect(updatedEvent.regressionInRelease).toBeUndefined(); + }); + + test('should not resolve an event again before a release newer than its regression', async () => { + await releases.insertMany([ + createRelease('a', 96, true), + createRelease('b', 72, true), + createRelease('c', 48), + ]); + const event = createEvent('error-active-regression', 'a', 'b'); + + event.regressionInRelease = 'c'; + await events.insertOne(event); + await repetitions.insertOne(createRepetition('error-active-regression', 'c')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-active-regression' })).resolvedInRelease).toBe('b'); + }); + + test('should continue validating other projects when one project fails', async () => { + const failedProjectId = 'failed-release-validator-project'; + + await releases.insertMany([ + createRelease('a', 72, true), + { + ...createRelease('b', 49), + projectId: failedProjectId, + }, + createRelease('c', 48), + ]); + await db.createCollection(`events:${failedProjectId}`, { + validator: { + $jsonSchema: { + properties: { + resolvedInRelease: { + bsonType: 'int', + }, + }, + }, + }, + validationLevel: 'strict', + validationAction: 'error', + }); + await db.collection(`events:${failedProjectId}`).insertOne({ + _id: new ObjectID(), + groupHash: 'failed-project-event', + payload: { + title: 'failed-project-event', + release: 'a', + }, + totalCount: 1, + catcherType: 'errors/default', + usersAffected: 0, + visitedBy: [], + timestamp: NOW_SECONDS - 96 * HOUR_IN_SECONDS, + }); + await events.insertOne(createEvent('successful-project-event', 'a')); + + try { + await expect(validateReleases(db, NOW)).resolves.toBeUndefined(); + + expect((await releases.findOne({ + projectId: failedProjectId, + release: 'b', + })).fixChecked).toBe(false); + expect((await events.findOne({ groupHash: 'successful-project-event' })).resolvedInRelease).toBe('c'); + expect((await releases.findOne({ + projectId: PROJECT_ID, + release: 'c', + })).fixChecked).toBe(true); + } finally { + await db.dropCollection(`events:${failedProjectId}`).catch(() => undefined); + await db.dropCollection(`repetitions:${failedProjectId}`).catch(() => undefined); + await releases.deleteMany({ projectId: failedProjectId }); + } + }); + + test('should not select a release older than the configured retention period as a candidate', async () => { + const originalMaxDaysNumber = process.env.MAX_DAYS_NUMBER; + + process.env.MAX_DAYS_NUMBER = '10'; + + try { + await releases.insertMany([ + createRelease('a', 480, true), + createRelease('b', 264), + ]); + await events.insertOne(createEvent('error-12', 'a')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-12' })).resolvedInRelease).toBeUndefined(); + expect((await releases.findOne({ release: 'b' })).fixChecked).toBe(false); + } finally { + process.env.MAX_DAYS_NUMBER = originalMaxDaysNumber; + } + }); + + test('should use a release older than the retention period as event history', async () => { + const originalMaxDaysNumber = process.env.MAX_DAYS_NUMBER; + + process.env.MAX_DAYS_NUMBER = '10'; + + try { + await releases.insertMany([ + createRelease('a', 480, true), + createRelease('b', 48), + ]); + await events.insertOne(createEvent('error-13', 'a')); + + await validateReleases(db, NOW); + + expect((await events.findOne({ groupHash: 'error-13' })).resolvedInRelease).toBe('b'); + } finally { + process.env.MAX_DAYS_NUMBER = originalMaxDaysNumber; + } + }); +}); diff --git a/workers/task-manager/src/index.ts b/workers/task-manager/src/index.ts index d6dec032..cda3f338 100644 --- a/workers/task-manager/src/index.ts +++ b/workers/task-manager/src/index.ts @@ -6,9 +6,9 @@ import * as pkg from '../package.json'; import type { TaskManagerWorkerTask } from '../types/task-manager-worker-task'; import type { ProjectDBScheme, - GroupedEventDBScheme, - ProjectTaskManagerConfig + GroupedEventDBScheme } from '@hawk.so/types'; +import type { ProjectTaskManagerConfig } from '../types/project-task-manager-config'; import type { TaskManagerItem } from '@hawk.so/types/src/base/event/taskManagerItem'; import HawkCatcher from '@hawk.so/nodejs'; import { decodeUnsafeFields } from '../../../lib/utils/unsafeFields'; diff --git a/workers/task-manager/types/project-task-manager-config.ts b/workers/task-manager/types/project-task-manager-config.ts new file mode 100644 index 00000000..a255450f --- /dev/null +++ b/workers/task-manager/types/project-task-manager-config.ts @@ -0,0 +1,19 @@ +import type { ProjectTaskManagerConfig as ProjectTaskManagerConfigType } from '@hawk.so/types'; + +interface LegacyDelegatedUser { + accessToken: string; + accessTokenExpiresAt: Date | null; + refreshToken: string; + refreshTokenExpiresAt: Date | null; + status: 'active' | 'revoked' | 'missing'; +} + +/** + * Project task manager config with the legacy delegated user data still used + * by the API and task-manager worker. + */ +export type ProjectTaskManagerConfig = ProjectTaskManagerConfigType & { + config: ProjectTaskManagerConfigType['config'] & { + delegatedUser?: LegacyDelegatedUser; + }; +}; diff --git a/yarn.lock b/yarn.lock index 3172b2bc..31792312 100644 --- a/yarn.lock +++ b/yarn.lock @@ -402,10 +402,10 @@ dependencies: "@types/mongodb" "^3.5.34" -"@hawk.so/types@^0.5.9": - version "0.5.9" - resolved "https://registry.yarnpkg.com/@hawk.so/types/-/types-0.5.9.tgz#817e8b26283d0367371125f055f2e37a274797bc" - integrity sha512-86aE0Bdzvy8C+Dqd1iZpnDho44zLGX/t92SGuAv2Q52gjSJ7SHQdpGDWtM91FXncfT5uzAizl9jYMuE6Qrtm0Q== +"@hawk.so/types@^0.9.0": + version "0.9.0" + resolved "https://registry.yarnpkg.com/@hawk.so/types/-/types-0.9.0.tgz#19b68065e9bafa0f4ee14f22a71cbb946f502cbc" + integrity sha512-zY9yw83Dzw3RcbTtMR9AKj8yNEPwD1sZZCeLlw5onZr1+GdjfnR1AhLvoPeZXBbRPRLx3MzyeuM85lAl0do+DA== dependencies: bson "^7.0.0"