From 74f7101d19fd749941bd83c500dd3dda5c641c08 Mon Sep 17 00:00:00 2001 From: Kuchizu Date: Mon, 14 Sep 2026 16:32:43 +0300 Subject: [PATCH 1/5] Stop limiter from opening every events collection each hour --- workers/limiter/src/dbHelper.ts | 55 +++++++++++++--------- workers/limiter/tests/dbHelper.test.ts | 65 +++++++++++++++++++++++++- workers/limiter/tests/index.test.ts | 7 +++ 3 files changed, 103 insertions(+), 24 deletions(-) diff --git a/workers/limiter/src/dbHelper.ts b/workers/limiter/src/dbHelper.ts index 984dd335..e0059df4 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 @@ -195,35 +196,43 @@ export class DbHelper { 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 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' }, + const [ counters ] = await dailyEventsCollection + .aggregate<{ boundaryDay: number; fullDays: number }>([ + { $match: { groupingTimestamp: { $gte: boundaryDayTimestamp } } }, + { + $group: { + _id: null, + boundaryDay: { + $sum: { $cond: [ { $lt: ['$groupingTimestamp', firstFullDayTimestamp] }, '$count', 0] }, + }, + fullDays: { + $sum: { $cond: [ { $gte: ['$groupingTimestamp', firstFullDayTimestamp] }, '$count', 0] }, }, }, - ]) - .toArray(), - ]); + }, + ], { + /** one table per project instead of racing all groupingTimestamp indexes */ + hint: { $natural: 1 }, + }) + .toArray(); + + if (!counters) { + return 0; + } - const fullDaysCount = dailyCounters.length > 0 ? dailyCounters[0].count : 0; + const boundaryDayCount = counters.boundaryDay > 0 + ? await this.getRawEventsCountByProject(project, { + timestamp: { + $gt: since, + $lt: firstFullDayTimestamp, + }, + }) + : 0; - return boundaryDayCount + fullDaysCount; + return boundaryDayCount + counters.fullDays; } catch (e) { HawkCatcher.send(e); throw new CriticalError(e); diff --git a/workers/limiter/tests/dbHelper.test.ts b/workers/limiter/tests/dbHelper.test.ts index dd26baa7..bf9aa46c 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,55 @@ 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('getEventsCountByProjectsUsingDailyEvents', () => { diff --git a/workers/limiter/tests/index.test.ts b/workers/limiter/tests/index.test.ts index 3fd54a5a..f49c987b 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 () => { From 104f180f6bb26287fd5c457875a33e1a89445cb1 Mon Sep 17 00:00:00 2001 From: Kuchizu Date: Wed, 16 Sep 2026 14:53:09 +0300 Subject: [PATCH 2/5] Comment limiter dailyEvents aggregation sections --- workers/limiter/src/dbHelper.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/workers/limiter/src/dbHelper.ts b/workers/limiter/src/dbHelper.ts index e0059df4..e6948e06 100644 --- a/workers/limiter/src/dbHelper.ts +++ b/workers/limiter/src/dbHelper.ts @@ -201,13 +201,16 @@ export class DbHelper { 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] }, }, @@ -219,10 +222,12 @@ export class DbHelper { }) .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: { From 1315f342121c2ea229d2d372bfadcaa2c330304a Mon Sep 17 00:00:00 2001 From: Kuchizu Date: Wed, 16 Sep 2026 18:33:29 +0300 Subject: [PATCH 3/5] Report sampled limiter counter validation to Telegram --- .env.sample | 5 +- workers/limiter/src/dbHelper.ts | 23 ++++++-- workers/limiter/src/index.ts | 64 ++++++++++++++++++++ workers/limiter/tests/dbHelper.test.ts | 26 ++++++++ workers/limiter/tests/index.test.ts | 82 ++++++++++++++++++++++++++ 5 files changed, 193 insertions(+), 7 deletions(-) diff --git a/.env.sample b/.env.sample index f2df9be2..73f86f02 100644 --- a/.env.sample +++ b/.env.sample @@ -55,5 +55,8 @@ HAWK_CATCHER_TOKEN= ## If true, Grouper worker will send messages about new events to Notifier worker IS_NOTIFIER_WORKER_ENABLED=false -## Url for telegram notifications about workspace blocks and unblocks +## Url for telegram notifications about workspace blocks and unblocks TELEGRAM_LIMITER_CHAT_URL= + +## Share of workspaces the limiter recounts with raw boundary day and reports to telegram, 0 disables +LIMITER_COUNTER_VALIDATION_RATE=0.05 diff --git a/workers/limiter/src/dbHelper.ts b/workers/limiter/src/dbHelper.ts index e6948e06..93c957fd 100644 --- a/workers/limiter/src/dbHelper.ts +++ b/workers/limiter/src/dbHelper.ts @@ -188,10 +188,12 @@ export class DbHelper { * * @param project - project to check * @param since - timestamp of the time from which we count the events + * @param alwaysCountBoundaryDay - count the boundary day from raw collections even if its bucket is empty */ public async getEventsCountByProjectUsingDailyEvents( project: ProjectDBScheme, - since: number + since: number, + alwaysCountBoundaryDay = false ): Promise { try { const projectId = project._id.toString(); @@ -223,12 +225,16 @@ export class DbHelper { .toArray(); /** no buckets in the billing period */ - if (!counters) { + if (!counters && !alwaysCountBoundaryDay) { return 0; } + const shouldCountBoundaryDay = alwaysCountBoundaryDay + ? since < firstFullDayTimestamp + : counters.boundaryDay > 0; + /** the bucket spans the whole day, so the part after `since` is counted from raw events */ - const boundaryDayCount = counters.boundaryDay > 0 + const boundaryDayCount = shouldCountBoundaryDay ? await this.getRawEventsCountByProject(project, { timestamp: { $gt: since, @@ -237,7 +243,7 @@ export class DbHelper { }) : 0; - return boundaryDayCount + counters.fullDays; + return boundaryDayCount + (counters?.fullDays ?? 0); } catch (e) { HawkCatcher.send(e); throw new CriticalError(e); @@ -250,12 +256,17 @@ export class DbHelper { * * @param projects - projects to calculate for * @param since - timestamp of the time from which we count the events + * @param alwaysCountBoundaryDay - count the boundary day from raw collections even if its bucket is empty */ - public async getEventsCountByProjectsUsingDailyEvents(projects: ProjectDBScheme[], since: number): Promise { + public async getEventsCountByProjectsUsingDailyEvents( + projects: ProjectDBScheme[], + since: number, + alwaysCountBoundaryDay = false + ): 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.getEventsCountByProjectUsingDailyEvents(project, since, alwaysCountBoundaryDay) )) .then(sum); } diff --git a/workers/limiter/src/index.ts b/workers/limiter/src/index.ts index 9df8c970..5c85c803 100644 --- a/workers/limiter/src/index.ts +++ b/workers/limiter/src/index.ts @@ -30,6 +30,21 @@ const NOTIFY_ABOUT_LIMIT = [ 0.95, ]; +/** + * Share of workspaces recounted with raw boundary day if LIMITER_COUNTER_VALIDATION_RATE is not set + */ +const DEFAULT_COUNTER_VALIDATION_RATE = 0.05; + +/** + * Sampled comparison with the raw boundary day count + */ +interface CounterValidation { + workspaces: number; + projects: number; + /** report lines for undercounted workspaces */ + undercounted: string[]; +} + /** * Worker for checking current total events count in workspaces and limits events receiving if workspace exceed the limit */ @@ -196,6 +211,14 @@ export default class LimiterWorker extends Worker { const updatedWorkspaces: WorkspaceWithTariffPlan[] = []; + /** share of workspaces recounted with raw boundary day, 0 disables */ + const validationRate = Number(process.env.LIMITER_COUNTER_VALIDATION_RATE ?? DEFAULT_COUNTER_VALIDATION_RATE); + const validation: CounterValidation = { + workspaces: 0, + projects: 0, + undercounted: [], + }; + for await (const workspace of workspaces) { /** * If workspace is already blocked - do nothing @@ -206,10 +229,30 @@ export default class LimiterWorker extends Worker { const workspaceProjects = await this.dbHelper.getProjects(workspace._id.toString()); + /** before the regular count, so new events can't look like an undercount */ + const referenceCount = workspace.lastChargeDate && Math.random() < validationRate + ? await this.dbHelper.getEventsCountByProjectsUsingDailyEvents( + workspaceProjects, + Math.floor(new Date(workspace.lastChargeDate).getTime() / MS_IN_SEC), + true + ) + : null; + const { shouldBeBlockedByQuota, updatedWorkspace, projectsToUpdate } = await this.prepareWorkspaceUsageUpdate(workspace, workspaceProjects); updatedWorkspaces.push(updatedWorkspace); + if (referenceCount !== null) { + validation.workspaces++; + validation.projects += workspaceProjects.length; + + if (updatedWorkspace.billingPeriodEventsCount < referenceCount) { + validation.undercounted.push( + `• ${workspace.name} (id: ${workspace._id}): ${updatedWorkspace.billingPeriodEventsCount} instead of ${referenceCount}` + ); + } + } + /** * If there are no projects to update - move on to next workspace */ @@ -234,6 +277,7 @@ export default class LimiterWorker extends Worker { await this.dbHelper.updateWorkspacesEventsCountAndIsBlocked(updatedWorkspaces); this.sendRegularReport(message); + this.sendCounterValidationReport(validation); } /** @@ -431,6 +475,26 @@ export default class LimiterWorker extends Worker { telegram.sendMessage(`${message}`, telegram.TelegramBotURLs.Limiter); } + /** + * Sends counter validation result to Telegram + * + * @param validation - validation result of the regular check + */ + private sendCounterValidationReport(validation: CounterValidation): void { + if (validation.workspaces === 0) { + return; + } + + const undercounted = validation.undercounted.length > 0 ? `\n${validation.undercounted.join('\n')}` : ' none'; + + telegram.sendMessage( + `[ Limiter / Validation ]\n` + + `Checked ${validation.workspaces} workspaces (${validation.projects} projects)\n` + + `Undercounted:${undercounted}`, + telegram.TelegramBotURLs.Limiter + ); + } + /** * Method that sends regular workspace check report ti tg chat with telegram util * diff --git a/workers/limiter/tests/dbHelper.test.ts b/workers/limiter/tests/dbHelper.test.ts index bf9aa46c..976748d9 100644 --- a/workers/limiter/tests/dbHelper.test.ts +++ b/workers/limiter/tests/dbHelper.test.ts @@ -842,6 +842,32 @@ describe('DbHelper', () => { expect(count).toBe(0); }); + + test('Should count raw boundary-day events without a bucket when alwaysCountBoundaryDay is set', async () => { + /** + * Arrange + */ + const project = createProjectMock({ workspaceId: new ObjectId() }); + const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC); + + await fillDatabaseWithMockedData({ + project, + eventsToMock: 0, + }); + await db.collection(`events:${project._id.toString()}`).insertMany([createEventMock(), createEventMock()]); + + /** + * Act + */ + const gatedCount = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); + const referenceCount = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since, true); + + /** + * Assert + */ + expect(gatedCount).toBe(0); + expect(referenceCount).toBe(2); + }); }); describe('getEventsCountByProjectsUsingDailyEvents', () => { diff --git a/workers/limiter/tests/index.test.ts b/workers/limiter/tests/index.test.ts index f49c987b..1c538339 100644 --- a/workers/limiter/tests/index.test.ts +++ b/workers/limiter/tests/index.test.ts @@ -168,6 +168,7 @@ describe('Limiter worker', () => { beforeEach(async () => { jest.clearAllMocks(); + process.env.LIMITER_COUNTER_VALIDATION_RATE = '0'; await redisClient.flushAll(); await projectCollection.deleteMany({}); await workspaceCollection.deleteMany({}); @@ -388,6 +389,87 @@ describe('Limiter worker', () => { expect(telegram.sendMessage).not.toHaveBeenCalled(); }); + describe('counter validation', () => { + test('Should report workspaces counted lower than with the raw boundary day count', async () => { + /** + * Arrange + */ + process.env.LIMITER_COUNTER_VALIDATION_RATE = '1'; + + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10000, + billingPeriodEventsCount: 0, + lastChargeDate: LAST_CHARGE_DATE, + }); + const project = createProjectMock({ workspaceId: workspace._id }); + + await fillDatabaseWithMockedData({ + workspace, + project, + eventsToMock: 5, + }); + + /** + * Without a bucket the regular count skips boundary-day events + */ + await db.collection(`dailyEvents:${project._id.toString()}`).deleteMany({}); + + /** + * Act + */ + const worker = new LimiterWorker(); + + await worker.start(); + await worker.handle(REGULAR_WORKSPACES_CHECK_EVENT); + await worker.finish(); + + /** + * Assert + */ + expect(telegram.sendMessage).toHaveBeenCalledTimes(1); + + const reportMessage = (telegram.sendMessage as jest.Mock).mock.calls[0][0]; + + expect(reportMessage).toContain('Checked 1 workspaces (1 projects)'); + expect(reportMessage).toContain(`${workspace._id}): 0 instead of 5`); + }); + + test('Should report no undercount when counts match', async () => { + /** + * Arrange + */ + process.env.LIMITER_COUNTER_VALIDATION_RATE = '1'; + + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10000, + billingPeriodEventsCount: 0, + lastChargeDate: LAST_CHARGE_DATE, + }); + const project = createProjectMock({ workspaceId: workspace._id }); + + await fillDatabaseWithMockedData({ + workspace, + project, + eventsToMock: 5, + }); + + /** + * Act + */ + const worker = new LimiterWorker(); + + await worker.start(); + await worker.handle(REGULAR_WORKSPACES_CHECK_EVENT); + await worker.finish(); + + /** + * Assert + */ + expect(telegram.sendMessage).toHaveBeenCalledTimes(1); + expect((telegram.sendMessage as jest.Mock).mock.calls[0][0]).toContain('Undercounted: none'); + }); + }); + test('Should not send a report when no projects are blocked or unblocked', async () => { /** * Arrange From 7ed862692d530e1e2bbdb37791fb6b83cc11cffb Mon Sep 17 00:00:00 2001 From: Kuchizu Date: Wed, 16 Sep 2026 21:04:04 +0300 Subject: [PATCH 4/5] Compare old and new limiter algo for listed workspaces instead of sampling --- .env.sample | 6 +- workers/limiter/src/index.ts | 104 ++++++++++---------------- workers/limiter/tests/index.test.ts | 111 ++++++++++------------------ 3 files changed, 83 insertions(+), 138 deletions(-) diff --git a/.env.sample b/.env.sample index 73f86f02..bc9d55df 100644 --- a/.env.sample +++ b/.env.sample @@ -55,8 +55,8 @@ HAWK_CATCHER_TOKEN= ## If true, Grouper worker will send messages about new events to Notifier worker IS_NOTIFIER_WORKER_ENABLED=false -## Url for telegram notifications about workspace blocks and unblocks +## Url for telegram notifications about workspace blocks and unblocks TELEGRAM_LIMITER_CHAT_URL= -## Share of workspaces the limiter recounts with raw boundary day and reports to telegram, 0 disables -LIMITER_COUNTER_VALIDATION_RATE=0.05 +## Comma-separated workspace ids the limiter counts with old and new algorithms and reports to telegram +LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS= diff --git a/workers/limiter/src/index.ts b/workers/limiter/src/index.ts index 5c85c803..db7d829e 100644 --- a/workers/limiter/src/index.ts +++ b/workers/limiter/src/index.ts @@ -30,21 +30,6 @@ const NOTIFY_ABOUT_LIMIT = [ 0.95, ]; -/** - * Share of workspaces recounted with raw boundary day if LIMITER_COUNTER_VALIDATION_RATE is not set - */ -const DEFAULT_COUNTER_VALIDATION_RATE = 0.05; - -/** - * Sampled comparison with the raw boundary day count - */ -interface CounterValidation { - workspaces: number; - projects: number; - /** report lines for undercounted workspaces */ - undercounted: string[]; -} - /** * Worker for checking current total events count in workspaces and limits events receiving if workspace exceed the limit */ @@ -211,14 +196,6 @@ export default class LimiterWorker extends Worker { const updatedWorkspaces: WorkspaceWithTariffPlan[] = []; - /** share of workspaces recounted with raw boundary day, 0 disables */ - const validationRate = Number(process.env.LIMITER_COUNTER_VALIDATION_RATE ?? DEFAULT_COUNTER_VALIDATION_RATE); - const validation: CounterValidation = { - workspaces: 0, - projects: 0, - undercounted: [], - }; - for await (const workspace of workspaces) { /** * If workspace is already blocked - do nothing @@ -229,30 +206,10 @@ export default class LimiterWorker extends Worker { const workspaceProjects = await this.dbHelper.getProjects(workspace._id.toString()); - /** before the regular count, so new events can't look like an undercount */ - const referenceCount = workspace.lastChargeDate && Math.random() < validationRate - ? await this.dbHelper.getEventsCountByProjectsUsingDailyEvents( - workspaceProjects, - Math.floor(new Date(workspace.lastChargeDate).getTime() / MS_IN_SEC), - true - ) - : null; - const { shouldBeBlockedByQuota, updatedWorkspace, projectsToUpdate } = await this.prepareWorkspaceUsageUpdate(workspace, workspaceProjects); updatedWorkspaces.push(updatedWorkspace); - if (referenceCount !== null) { - validation.workspaces++; - validation.projects += workspaceProjects.length; - - if (updatedWorkspace.billingPeriodEventsCount < referenceCount) { - validation.undercounted.push( - `• ${workspace.name} (id: ${workspace._id}): ${updatedWorkspace.billingPeriodEventsCount} instead of ${referenceCount}` - ); - } - } - /** * If there are no projects to update - move on to next workspace */ @@ -277,7 +234,6 @@ export default class LimiterWorker extends Worker { await this.dbHelper.updateWorkspacesEventsCountAndIsBlocked(updatedWorkspaces); this.sendRegularReport(message); - this.sendCounterValidationReport(validation); } /** @@ -310,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}`); @@ -372,6 +328,44 @@ export default class LimiterWorker extends Worker { }; } + /** + * Returns workspace events count. For workspaces listed in LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS + * it also counts them the old way, with raw boundary day for every project, and reports both to Telegram + * + * @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); + } + + /** old algo goes first, so events arriving in between can only raise the new count */ + const oldAlgoStartedAt = Date.now(); + const oldAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since, true); + 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 // // /** @@ -475,26 +469,6 @@ export default class LimiterWorker extends Worker { telegram.sendMessage(`${message}`, telegram.TelegramBotURLs.Limiter); } - /** - * Sends counter validation result to Telegram - * - * @param validation - validation result of the regular check - */ - private sendCounterValidationReport(validation: CounterValidation): void { - if (validation.workspaces === 0) { - return; - } - - const undercounted = validation.undercounted.length > 0 ? `\n${validation.undercounted.join('\n')}` : ' none'; - - telegram.sendMessage( - `[ Limiter / Validation ]\n` + - `Checked ${validation.workspaces} workspaces (${validation.projects} projects)\n` + - `Undercounted:${undercounted}`, - telegram.TelegramBotURLs.Limiter - ); - } - /** * Method that sends regular workspace check report ti tg chat with telegram util * diff --git a/workers/limiter/tests/index.test.ts b/workers/limiter/tests/index.test.ts index 1c538339..489054b4 100644 --- a/workers/limiter/tests/index.test.ts +++ b/workers/limiter/tests/index.test.ts @@ -168,7 +168,6 @@ describe('Limiter worker', () => { beforeEach(async () => { jest.clearAllMocks(); - process.env.LIMITER_COUNTER_VALIDATION_RATE = '0'; await redisClient.flushAll(); await projectCollection.deleteMany({}); await workspaceCollection.deleteMany({}); @@ -389,85 +388,57 @@ describe('Limiter worker', () => { expect(telegram.sendMessage).not.toHaveBeenCalled(); }); - describe('counter validation', () => { - test('Should report workspaces counted lower than with the raw boundary day count', async () => { - /** - * Arrange - */ - process.env.LIMITER_COUNTER_VALIDATION_RATE = '1'; - - const workspace = createWorkspaceMock({ - plan: mockedPlans.eventsLimit10000, - billingPeriodEventsCount: 0, - lastChargeDate: LAST_CHARGE_DATE, - }); - const project = createProjectMock({ workspaceId: workspace._id }); - - await fillDatabaseWithMockedData({ - workspace, - project, - eventsToMock: 5, - }); - - /** - * Without a bucket the regular count skips boundary-day events - */ - await db.collection(`dailyEvents:${project._id.toString()}`).deleteMany({}); - - /** - * Act - */ - const worker = new LimiterWorker(); - - await worker.start(); - await worker.handle(REGULAR_WORKSPACES_CHECK_EVENT); - await worker.finish(); + test('Should report old and new algo counts for workspaces listed for comparison', async () => { + /** + * Arrange + */ + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10000, + billingPeriodEventsCount: 0, + lastChargeDate: LAST_CHARGE_DATE, + }); + const project = createProjectMock({ workspaceId: workspace._id }); - /** - * Assert - */ - expect(telegram.sendMessage).toHaveBeenCalledTimes(1); + await fillDatabaseWithMockedData({ + workspace, + project, + eventsToMock: 5, + }); - const reportMessage = (telegram.sendMessage as jest.Mock).mock.calls[0][0]; + /** + * Without a bucket the new algo skips boundary-day events + */ + await db.collection(`dailyEvents:${project._id.toString()}`).deleteMany({}); - expect(reportMessage).toContain('Checked 1 workspaces (1 projects)'); - expect(reportMessage).toContain(`${workspace._id}): 0 instead of 5`); - }); + process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS = workspace._id.toString(); - test('Should report no undercount when counts match', async () => { - /** - * Arrange - */ - process.env.LIMITER_COUNTER_VALIDATION_RATE = '1'; - - const workspace = createWorkspaceMock({ - plan: mockedPlans.eventsLimit10000, - billingPeriodEventsCount: 0, - lastChargeDate: LAST_CHARGE_DATE, - }); - const project = createProjectMock({ workspaceId: workspace._id }); - - await fillDatabaseWithMockedData({ - workspace, - project, - eventsToMock: 5, - }); - - /** - * Act - */ - const worker = new LimiterWorker(); + /** + * Act + */ + 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; + } - /** - * Assert - */ - expect(telegram.sendMessage).toHaveBeenCalledTimes(1); - expect((telegram.sendMessage as jest.Mock).mock.calls[0][0]).toContain('Undercounted: none'); + /** + * Assert + */ + 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 () => { From d03183f07ad378d807675c54860f850306a8cd5f Mon Sep 17 00:00:00 2001 From: Kuchizu Date: Thu, 17 Sep 2026 12:34:41 +0300 Subject: [PATCH 5/5] Use the previous dailyEvents query for old algo comparison --- .env.sample | 2 +- workers/limiter/src/dbHelper.ts | 84 +++++++++++++++++++++----- workers/limiter/src/index.ts | 6 +- workers/limiter/tests/dbHelper.test.ts | 27 ++++----- workers/limiter/tests/index.test.ts | 12 ---- 5 files changed, 84 insertions(+), 47 deletions(-) diff --git a/.env.sample b/.env.sample index bc9d55df..5c19ff64 100644 --- a/.env.sample +++ b/.env.sample @@ -58,5 +58,5 @@ IS_NOTIFIER_WORKER_ENABLED=false ## Url for telegram notifications about workspace blocks and unblocks TELEGRAM_LIMITER_CHAT_URL= -## Comma-separated workspace ids the limiter counts with old and new algorithms and reports to telegram +## 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 93c957fd..0d0e71e5 100644 --- a/workers/limiter/src/dbHelper.ts +++ b/workers/limiter/src/dbHelper.ts @@ -188,12 +188,10 @@ export class DbHelper { * * @param project - project to check * @param since - timestamp of the time from which we count the events - * @param alwaysCountBoundaryDay - count the boundary day from raw collections even if its bucket is empty */ public async getEventsCountByProjectUsingDailyEvents( project: ProjectDBScheme, - since: number, - alwaysCountBoundaryDay = false + since: number ): Promise { try { const projectId = project._id.toString(); @@ -225,16 +223,12 @@ export class DbHelper { .toArray(); /** no buckets in the billing period */ - if (!counters && !alwaysCountBoundaryDay) { + if (!counters) { return 0; } - const shouldCountBoundaryDay = alwaysCountBoundaryDay - ? since < firstFullDayTimestamp - : counters.boundaryDay > 0; - /** the bucket spans the whole day, so the part after `since` is counted from raw events */ - const boundaryDayCount = shouldCountBoundaryDay + const boundaryDayCount = counters.boundaryDay > 0 ? await this.getRawEventsCountByProject(project, { timestamp: { $gt: since, @@ -243,7 +237,7 @@ export class DbHelper { }) : 0; - return boundaryDayCount + (counters?.fullDays ?? 0); + return boundaryDayCount + counters.fullDays; } catch (e) { HawkCatcher.send(e); throw new CriticalError(e); @@ -256,17 +250,75 @@ export class DbHelper { * * @param projects - projects to calculate for * @param since - timestamp of the time from which we count the events - * @param alwaysCountBoundaryDay - count the boundary day from raw collections even if its bucket is empty */ - public async getEventsCountByProjectsUsingDailyEvents( - projects: ProjectDBScheme[], - since: number, - alwaysCountBoundaryDay = false + 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(); + 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.getEventsCountByProjectUsingDailyEvents(project, since, alwaysCountBoundaryDay) + project => this.getEventsCountByProjectUsingDailyEventsOld(project, since) )) .then(sum); } diff --git a/workers/limiter/src/index.ts b/workers/limiter/src/index.ts index db7d829e..1b3ca6cd 100644 --- a/workers/limiter/src/index.ts +++ b/workers/limiter/src/index.ts @@ -329,8 +329,7 @@ export default class LimiterWorker extends Worker { } /** - * Returns workspace events count. For workspaces listed in LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS - * it also counts them the old way, with raw boundary day for every project, and reports both to Telegram + * 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 @@ -347,9 +346,8 @@ export default class LimiterWorker extends Worker { return this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since); } - /** old algo goes first, so events arriving in between can only raise the new count */ const oldAlgoStartedAt = Date.now(); - const oldAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEvents(projects, since, true); + const oldAlgoCount = await this.dbHelper.getEventsCountByProjectsUsingDailyEventsOld(projects, since); const oldAlgoTook = (Date.now() - oldAlgoStartedAt) / MS_IN_SEC; const newAlgoStartedAt = Date.now(); diff --git a/workers/limiter/tests/dbHelper.test.ts b/workers/limiter/tests/dbHelper.test.ts index 976748d9..210d486f 100644 --- a/workers/limiter/tests/dbHelper.test.ts +++ b/workers/limiter/tests/dbHelper.test.ts @@ -842,31 +842,30 @@ describe('DbHelper', () => { expect(count).toBe(0); }); + }); - test('Should count raw boundary-day events without a bucket when alwaysCountBoundaryDay is set', async () => { - /** - * Arrange - */ + 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()]); - /** - * Act - */ - const gatedCount = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); - const referenceCount = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since, true); + const oldCount = await dbHelper.getEventsCountByProjectUsingDailyEventsOld(project, since); + const newCount = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); - /** - * Assert - */ - expect(gatedCount).toBe(0); - expect(referenceCount).toBe(2); + expect(oldCount).toBe(5); + expect(newCount).toBe(3); }); }); diff --git a/workers/limiter/tests/index.test.ts b/workers/limiter/tests/index.test.ts index 489054b4..d233a06f 100644 --- a/workers/limiter/tests/index.test.ts +++ b/workers/limiter/tests/index.test.ts @@ -389,9 +389,6 @@ describe('Limiter worker', () => { }); test('Should report old and new algo counts for workspaces listed for comparison', async () => { - /** - * Arrange - */ const workspace = createWorkspaceMock({ plan: mockedPlans.eventsLimit10000, billingPeriodEventsCount: 0, @@ -405,16 +402,10 @@ describe('Limiter worker', () => { eventsToMock: 5, }); - /** - * Without a bucket the new algo skips boundary-day events - */ await db.collection(`dailyEvents:${project._id.toString()}`).deleteMany({}); process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS = workspace._id.toString(); - /** - * Act - */ const worker = new LimiterWorker(); try { @@ -425,9 +416,6 @@ describe('Limiter worker', () => { delete process.env.LIMITER_COMPARE_COUNTERS_WORKSPACE_IDS; } - /** - * Assert - */ const workspaceInDatabase = await workspaceCollection.findOne({ _id: workspace._id, });