From c45a7dbde82276586895cf4b3b4d634f0bcaf1ba Mon Sep 17 00:00:00 2001 From: Kuchizu <70284260+Kuchizu@users.noreply.github.com> Date: Wed, 3 Jun 2026 16:58:36 +0300 Subject: [PATCH 1/6] feat(grouper): add slow handle diagnostics (#549) * feat(grouper): add slow handle diagnostics * refactor(grouper): extract slow handle diagnostics into session * fix(grouper): use monotonic time and exclusive timings in slow handle diagnostics --- workers/grouper/src/index.ts | 221 ++++++++++++------ workers/grouper/src/metrics/config.ts | 22 ++ workers/grouper/src/metrics/grouperMetrics.ts | 62 ++++- .../src/metrics/slowHandleDiagnostics.ts | 144 ++++++++++++ 4 files changed, 374 insertions(+), 75 deletions(-) create mode 100644 workers/grouper/src/metrics/slowHandleDiagnostics.ts diff --git a/workers/grouper/src/index.ts b/workers/grouper/src/index.ts index 44460c8d..8203e8d3 100644 --- a/workers/grouper/src/index.ts +++ b/workers/grouper/src/index.ts @@ -28,7 +28,8 @@ import { rightTrim } from '../../../lib/utils/string'; import { hasValue } from '../../../lib/utils/hasValue'; import GrouperMetrics from './metrics/grouperMetrics'; import GrouperMemoryMonitor from './metrics/memoryMonitor'; -import { grouperMemoryConfig } from './metrics/config'; +import SlowHandleDiagnostics, { SlowHandleSession } from './metrics/slowHandleDiagnostics'; +import { grouperDiagnosticsConfig, grouperMemoryConfig } from './metrics/config'; /** * eslint does not count decorators as a variable usage @@ -101,6 +102,11 @@ export default class GrouperWorker extends Worker { */ private memoryMonitor = new GrouperMemoryMonitor(this.logger, grouperMemoryConfig); + /** + * Slow handle diagnostics helper. + */ + private slowHandleDiagnostics = new SlowHandleDiagnostics(this.logger, this.grouperMetrics, grouperDiagnosticsConfig); + /** * Interval for periodic cache cleanup to prevent memory leaks from unbounded cache growth */ @@ -180,9 +186,14 @@ export default class GrouperWorker extends Worker { * @param task - event to handle */ private async handleInternal(task: GroupWorkerTask): Promise { - const taskPayloadSize = Buffer.byteLength(JSON.stringify(task.payload)); + const session = this.slowHandleDiagnostics.startSession(); + const taskPayloadSize = await session.measureStep('payloadSize', () => { + return Buffer.byteLength(JSON.stringify(task.payload)); + }); const handledTasksCount = ++this.handledTasksCount; const memoryBeforeHandle = process.memoryUsage(); + let deltaSize = 0; + let eventType: 'new' | 'repeated' = 'new'; this.grouperMetrics.observePayloadSize(taskPayloadSize); this.memoryMonitor.logBeforeHandle(memoryBeforeHandle, handledTasksCount, taskPayloadSize, task.projectId); @@ -198,25 +209,29 @@ export default class GrouperWorker extends Worker { }; } - let uniqueEventHash = await this.getUniqueEventHash(task); + let uniqueEventHash = await session.measureStep('hash', () => this.getUniqueEventHash(task)); let existedEvent: GroupedEventDBScheme; let repetitionId = null; let incrementDailyAffectedUsers = false; - /** - * Trim source code lines to prevent memory leaks - */ - this.trimSourceCodeLines(task.payload); + await session.measureStep('preprocess', () => { + /** + * Trim source code lines to prevent memory leaks + */ + this.trimSourceCodeLines(task.payload); - /** - * Filter sensitive information - */ - this.dataFilter.processEvent(task.payload); + /** + * Filter sensitive information + */ + this.dataFilter.processEvent(task.payload); + }); /** * Find similar events by grouping pattern */ - const similarEvent = await this.findSimilarEvent(task.projectId, task.payload.title); + const similarEvent = await session.measureStep('findSimilarEvent', () => { + return this.findSimilarEvent(task.projectId, task.payload.title); + }); if (similarEvent) { this.logger.info(`[handle] project=${task.projectId} title="${task.payload.title}" similar event found, groupHash=${similarEvent.groupHash} totalCount=${similarEvent.totalCount}`); @@ -234,7 +249,9 @@ export default class GrouperWorker extends Worker { /** * Find event by group hash. */ - existedEvent = await this.getEvent(task.projectId, uniqueEventHash); + existedEvent = await session.measureStep('getEvent', () => { + return this.getEvent(task.projectId, uniqueEventHash); + }); } /** @@ -242,6 +259,8 @@ export default class GrouperWorker extends Worker { */ const isFirstOccurrence = !existedEvent && !similarEvent; + eventType = isFirstOccurrence ? 'new' : 'repeated'; + if (isFirstOccurrence) { try { const incrementAffectedUsers = !!task.payload.user; @@ -251,14 +270,16 @@ export default class GrouperWorker extends Worker { /** * Insert new event */ - await this.saveEvent(task.projectId, { - groupHash: uniqueEventHash, - totalCount: 1, - catcherType: task.catcherType, - payload: task.payload, - timestamp: task.timestamp, - usersAffected: incrementAffectedUsers ? 1 : 0, - } as GroupedEventDBScheme); + await session.measureStep('saveNewEvent', () => { + return this.saveEvent(task.projectId, { + groupHash: uniqueEventHash, + totalCount: 1, + catcherType: task.catcherType, + payload: task.payload, + timestamp: task.timestamp, + usersAffected: incrementAffectedUsers ? 1 : 0, + } as GroupedEventDBScheme); + }); const eventCacheKey = await this.getEventCacheKey(task.projectId, uniqueEventHash); @@ -288,21 +309,27 @@ export default class GrouperWorker extends Worker { throw e; } } else { - const [incrementAffectedUsers, shouldIncrementDailyAffectedUsers] = await this.shouldIncrementAffectedUsers(task, existedEvent); + const [incrementAffectedUsers, shouldIncrementDailyAffectedUsers] = await session.measureStep('affectedUsers', () => { + return this.shouldIncrementAffectedUsers(task, existedEvent, session); + }); incrementDailyAffectedUsers = shouldIncrementDailyAffectedUsers; /** * Increment existed task's counter */ - await this.incrementEventCounterAndAffectedUsers(task.projectId, { - groupHash: uniqueEventHash, - }, incrementAffectedUsers); + await session.measureStep('incrementCounter', () => { + return this.incrementEventCounterAndAffectedUsers(task.projectId, { + groupHash: uniqueEventHash, + }, incrementAffectedUsers); + }); /** * Decode existed event to calculate diffs correctly */ - decodeUnsafeFields(existedEvent); + await session.measureStep('decodeEvent', () => { + decodeUnsafeFields(existedEvent); + }); let delta: RepetitionDelta; @@ -314,14 +341,17 @@ export default class GrouperWorker extends Worker { /** * Calculate delta between original event and repetition */ - delta = computeDelta(existedEvent.payload, task.payload); + delta = await session.measureStep('computeDelta', () => { + return computeDelta(existedEvent.payload, task.payload); + }); } catch (e) { console.error(e); throw new DiffCalculationError(e, existedEvent.payload, task.payload); } const deltaStr = JSON.stringify(delta); - const deltaSize = deltaStr != null ? Buffer.byteLength(deltaStr) : 0; + + deltaSize = deltaStr != null ? Buffer.byteLength(deltaStr) : 0; this.grouperMetrics.observeDeltaSize(deltaSize); @@ -333,7 +363,9 @@ export default class GrouperWorker extends Worker { timestamp: task.timestamp, } as RepetitionDBScheme; - repetitionId = await this.saveRepetition(task.projectId, newRepetition); + repetitionId = await session.measureStep('saveRepetition', () => { + return this.saveRepetition(task.projectId, newRepetition); + }); /** * Clear the large event payload references to allow garbage collection @@ -350,13 +382,15 @@ export default class GrouperWorker extends Worker { /** * Store events counter by days */ - await this.saveDailyEvents( - task.projectId, - uniqueEventHash, - task.timestamp, - repetitionId, - incrementDailyAffectedUsers - ); + await session.measureStep('saveDailyEvents', () => { + return this.saveDailyEvents( + task.projectId, + uniqueEventHash, + task.timestamp, + repetitionId, + incrementDailyAffectedUsers + ); + }); this.memoryMonitor.logHandleCompletion( memoryBeforeHandle, @@ -373,19 +407,31 @@ export default class GrouperWorker extends Worker { const isIgnored = isFirstOccurrence ? false : !!existedEvent?.marks?.ignored; if (!isIgnored) { - await this.addTask(WorkerNames.NOTIFIER, { - projectId: task.projectId, - event: { - title: task.payload.title, - groupHash: uniqueEventHash, - isNew: isFirstOccurrence, - repetitionId: repetitionId ? repetitionId.toString() : null, - }, + await session.measureStep('enqueueNotifier', () => { + return this.addTask(WorkerNames.NOTIFIER, { + projectId: task.projectId, + event: { + title: task.payload.title, + groupHash: uniqueEventHash, + isNew: isFirstOccurrence, + repetitionId: repetitionId ? repetitionId.toString() : null, + }, + }); }); } } - await this.recordProjectMetrics(task.projectId, 'events-accepted'); + await session.measureStep('recordProjectMetrics', () => { + return this.recordProjectMetrics(task.projectId, 'events-accepted'); + }); + + session.logIfSlow({ + projectId: task.projectId, + title: task.payload.title, + type: eventType, + payloadSize: taskPayloadSize, + deltaSize, + }); } /** @@ -421,8 +467,18 @@ export default class GrouperWorker extends Worker { }; const series = [ - { key: minutelyKey, label: 'minutely', retentionMs: TimeMs.DAY, timestampMs: bucketTimestampMs('minutely') }, - { key: hourlyKey, label: 'hourly', retentionMs: TimeMs.WEEK, timestampMs: bucketTimestampMs('hourly') }, + { + key: minutelyKey, + label: 'minutely', + retentionMs: TimeMs.DAY, + timestampMs: bucketTimestampMs('minutely'), + }, + { + key: hourlyKey, + label: 'hourly', + retentionMs: TimeMs.WEEK, + timestampMs: bucketTimestampMs('hourly'), + }, { key: dailyKey, label: 'daily', @@ -491,11 +547,13 @@ export default class GrouperWorker extends Worker { * @param projectId - id of the project to find event in */ private async findFirstEventByPattern(pattern: string, projectId: string): Promise { - return await this.eventsDb.getConnection() - .collection(`events:${projectId}`) - .findOne( - { 'payload.title': { $regex: pattern } } - ); + return this.grouperMetrics.observeMongoDuration('findFirstEventByPattern', async () => { + return await this.eventsDb.getConnection() + .collection(`events:${projectId}`) + .findOne( + { 'payload.title': { $regex: pattern } } + ); + }); } /** @@ -561,11 +619,13 @@ export default class GrouperWorker extends Worker { * @returns {ProjectEventGroupingPatternsDBScheme[]} EventPatterns object with projectId and list of patterns */ private async getProjectPatterns(projectId: string): Promise { - const project = await this.accountsDb.getConnection() - .collection('projects') - .findOne({ - _id: new mongodb.ObjectId(projectId), - }); + const project = await this.grouperMetrics.observeMongoDuration('getProjectPatterns', async () => { + return this.accountsDb.getConnection() + .collection('projects') + .findOne({ + _id: new mongodb.ObjectId(projectId), + }); + }); return project?.eventGroupingPatterns || []; } @@ -575,9 +635,14 @@ export default class GrouperWorker extends Worker { * * @param task - worker task to process * @param existedEvent - original event to get its user + * @param session - current slow handle diagnostics session. * @returns {[boolean, boolean]} - whether to increment affected users for the repetition and the daily aggregation */ - private async shouldIncrementAffectedUsers(task: GroupWorkerTask, existedEvent: GroupedEventDBScheme): Promise<[boolean, boolean]> { + private async shouldIncrementAffectedUsers( + task: GroupWorkerTask, + existedEvent: GroupedEventDBScheme, + session: SlowHandleSession + ): Promise<[boolean, boolean]> { const eventUser = task.payload.user; /** @@ -609,11 +674,13 @@ export default class GrouperWorker extends Worker { */ const repetitionCacheKey = `repetitions:${task.projectId}:${existedEvent.groupHash}:${eventUser.id}`; const repetition = await this.cache.get(repetitionCacheKey, async () => { - return this.eventsDb.getConnection().collection(`repetitions:${task.projectId}`) - .findOne({ - groupHash: existedEvent.groupHash, - 'payload.user.id': eventUser.id, - }); + return this.grouperMetrics.observeMongoDuration('findUserRepetition', async () => { + return this.eventsDb.getConnection().collection(`repetitions:${task.projectId}`) + .findOne({ + groupHash: existedEvent.groupHash, + 'payload.user.id': eventUser.id, + }); + }); }); if (repetition) { @@ -643,15 +710,17 @@ export default class GrouperWorker extends Worker { */ const repetitionDailyCacheKey = `repetitions:${task.projectId}:${existedEvent.groupHash}:${eventUser.id}:${eventMidnight}`; const repetitionDaily = await this.cache.get(repetitionDailyCacheKey, async () => { - return this.eventsDb.getConnection().collection(`repetitions:${task.projectId}`) - .findOne({ - groupHash: existedEvent.groupHash, - 'payload.user.id': eventUser.id, - timestamp: { - $gte: eventMidnight, - $lt: eventNextMidnight, - }, - }); + return this.grouperMetrics.observeMongoDuration('findDailyUserRepetition', async () => { + return this.eventsDb.getConnection().collection(`repetitions:${task.projectId}`) + .findOne({ + groupHash: existedEvent.groupHash, + 'payload.user.id': eventUser.id, + timestamp: { + $gte: eventMidnight, + $lt: eventNextMidnight, + }, + }); + }); }); /** @@ -665,8 +734,12 @@ export default class GrouperWorker extends Worker { /** * Check Redis lock - if locked, don't increment either counter */ - const isEventLocked = await this.redis.checkOrSetlockEventForAffectedUsersIncrement(existedEvent.groupHash, eventUser.id); - const isDailyEventLocked = await this.redis.checkOrSetlockDailyEventForAffectedUsersIncrement(existedEvent.groupHash, eventUser.id, eventMidnight); + const [isEventLocked, isDailyEventLocked] = await session.measureStep('affectedUsersRedisLocks', async () => { + return [ + await this.redis.checkOrSetlockEventForAffectedUsersIncrement(existedEvent.groupHash, eventUser.id), + await this.redis.checkOrSetlockDailyEventForAffectedUsersIncrement(existedEvent.groupHash, eventUser.id, eventMidnight), + ]; + }); shouldIncrementRepetitionAffectedUsers = isEventLocked ? false : shouldIncrementRepetitionAffectedUsers; shouldIncrementDailyAffectedUsers = isDailyEventLocked ? false : shouldIncrementDailyAffectedUsers; diff --git a/workers/grouper/src/metrics/config.ts b/workers/grouper/src/metrics/config.ts index 7fe5abb2..f36821ba 100644 --- a/workers/grouper/src/metrics/config.ts +++ b/workers/grouper/src/metrics/config.ts @@ -23,6 +23,16 @@ export interface GrouperMemoryConfig { handleGrowthWarnMb: number; } +/** + * Parsed config for slow handle diagnostics. + */ +export interface GrouperDiagnosticsConfig { + /** + * Log handle step breakdown when total handle duration is greater than this value. + */ + slowHandleWarnMs: number; +} + /** * Default memory checkpoint interval in tasks. */ @@ -43,6 +53,11 @@ const DEFAULT_MEMORY_GROWTH_WARN_MB = 64; */ const DEFAULT_MEMORY_HANDLE_GROWTH_WARN_MB = 16; +/** + * Default slow handle warning threshold in ms. + */ +const DEFAULT_SLOW_HANDLE_WARN_MS = 5000; + /** * Histogram buckets for payload and delta sizes (bytes). */ @@ -70,3 +85,10 @@ export const grouperMemoryConfig: GrouperMemoryConfig = { growthWarnMb: asPositiveNumber(process.env.GROUPER_MEMORY_GROWTH_WARN_MB, DEFAULT_MEMORY_GROWTH_WARN_MB), handleGrowthWarnMb: asPositiveNumber(process.env.GROUPER_MEMORY_HANDLE_GROWTH_WARN_MB, DEFAULT_MEMORY_HANDLE_GROWTH_WARN_MB), }; + +/** + * Slow handle diagnostics config from environment. + */ +export const grouperDiagnosticsConfig: GrouperDiagnosticsConfig = { + slowHandleWarnMs: asPositiveNumber(process.env.GROUPER_SLOW_HANDLE_WARN_MS, DEFAULT_SLOW_HANDLE_WARN_MS), +}; diff --git a/workers/grouper/src/metrics/grouperMetrics.ts b/workers/grouper/src/metrics/grouperMetrics.ts index 4be19e87..02a36306 100644 --- a/workers/grouper/src/metrics/grouperMetrics.ts +++ b/workers/grouper/src/metrics/grouperMetrics.ts @@ -2,7 +2,39 @@ import { client, register } from '../../../../lib/metrics'; import { GROUPER_METRICS_SIZE_BUCKETS } from './config'; type EventType = 'new' | 'repeated'; -type MongoOperation = 'getEvent' | 'saveEvent' | 'saveRepetition' | 'incrementCounter' | 'saveDailyEvents'; +type MongoOperation = + 'findDailyUserRepetition' | + 'findFirstEventByPattern' | + 'findUserRepetition' | + 'getEvent' | + 'getProjectPatterns' | + 'incrementCounter' | + 'saveDailyEvents' | + 'saveEvent' | + 'saveRepetition'; + +export type GrouperStep = + 'affectedUsers' | + 'affectedUsersRedisLocks' | + 'computeDelta' | + 'decodeEvent' | + 'enqueueNotifier' | + 'findSimilarEvent' | + 'getEvent' | + 'hash' | + 'incrementCounter' | + 'payloadSize' | + 'preprocess' | + 'recordProjectMetrics' | + 'saveDailyEvents' | + 'saveNewEvent' | + 'saveRepetition'; + +// eslint-disable-next-line @typescript-eslint/no-magic-numbers +const GROUPER_HANDLE_DURATION_BUCKETS = [0.05, 0.1, 0.25, 0.5, 1, 2, 3, 5, 7.5, 10, 15, 30, 60]; + +// eslint-disable-next-line @typescript-eslint/no-magic-numbers +const GROUPER_STEP_DURATION_BUCKETS = [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 30]; /** * Reuse already registered metric by name, or create one. @@ -39,6 +71,18 @@ export default class GrouperMetrics { () => new client.Histogram({ name: 'hawk_grouper_handle_duration_seconds', help: 'Duration of handle() call in seconds', + buckets: GROUPER_HANDLE_DURATION_BUCKETS, + registers: [ register ], + }) + ); + + private readonly stepDuration = getOrCreateMetric( + 'hawk_grouper_step_duration_seconds', + () => new client.Histogram({ + name: 'hawk_grouper_step_duration_seconds', + help: 'Duration of Grouper handle step in seconds', + labelNames: [ 'step' ], + buckets: GROUPER_STEP_DURATION_BUCKETS, registers: [ register ], }) ); @@ -106,6 +150,22 @@ export default class GrouperMetrics { } } + /** + * Measure a single Grouper handle step duration. + * + * @param step - step label. + * @param callback - callback to execute under timer. + */ + public async observeStepDuration(step: GrouperStep, callback: () => Promise | T): Promise { + const endTimer = this.stepDuration.startTimer({ step }); + + try { + return await callback(); + } finally { + endTimer(); + } + } + /** * Increment events counter by event type. * diff --git a/workers/grouper/src/metrics/slowHandleDiagnostics.ts b/workers/grouper/src/metrics/slowHandleDiagnostics.ts new file mode 100644 index 00000000..dce98722 --- /dev/null +++ b/workers/grouper/src/metrics/slowHandleDiagnostics.ts @@ -0,0 +1,144 @@ +import type { GrouperDiagnosticsConfig } from './config'; +import type GrouperMetrics from './grouperMetrics'; +import type { GrouperStep } from './grouperMetrics'; + +const NS_PER_MS = 1_000_000n; + +interface LoggerLike { + warn(message: string): void; +} + +/** + * Frame on the active measurement stack. childTimeNs accumulates the duration + * of nested measureStep() calls so the parent step records exclusive time. + */ +interface StepFrame { + step: GrouperStep; + startedAt: bigint; + childTimeNs: bigint; +} + +/** + * Context describing the handled task for slow handle log line. + */ +export interface SlowHandleContext { + projectId: string; + title?: string; + type: 'new' | 'repeated'; + payloadSize: number; + deltaSize: number; +} + +/** + * Per-handle diagnostics session: owns its own start timestamp and step timings, + * measures Grouper handle steps and emits a slow handle warning on demand. + */ +export class SlowHandleSession { + private readonly startedAt: bigint = process.hrtime.bigint(); + private readonly timings: Map = new Map(); + private readonly stack: StepFrame[] = []; + private readonly logger: LoggerLike; + private readonly metrics: GrouperMetrics; + private readonly config: GrouperDiagnosticsConfig; + + /** + * @param logger - logger instance. + * @param metrics - Grouper metrics facade used to record step duration. + * @param config - slow handle diagnostics thresholds. + */ + constructor(logger: LoggerLike, metrics: GrouperMetrics, config: GrouperDiagnosticsConfig) { + this.logger = logger; + this.metrics = metrics; + this.config = config; + } + + /** + * Measure a Grouper handle step and accumulate its exclusive duration + * (time not spent inside nested measureStep calls) in the session. + * + * @param step - step name. + * @param callback - measured callback. + */ + public async measureStep(step: GrouperStep, callback: () => Promise | T): Promise { + const frame: StepFrame = { + step, + startedAt: process.hrtime.bigint(), + childTimeNs: 0n, + }; + + this.stack.push(frame); + + try { + return await this.metrics.observeStepDuration(step, callback); + } finally { + const elapsedNs = process.hrtime.bigint() - frame.startedAt; + + this.stack.pop(); + + const parent = this.stack[this.stack.length - 1]; + + if (parent) { + parent.childTimeNs += elapsedNs; + } + + const exclusiveMs = Number((elapsedNs - frame.childTimeNs) / NS_PER_MS); + const previousMs = this.timings.get(step) || 0; + + this.timings.set(step, previousMs + exclusiveMs); + } + } + + /** + * Log slow handle breakdown if total session duration exceeds the warn threshold. + * + * @param context - handled task context. + */ + public logIfSlow(context: SlowHandleContext): void { + const durationMs = Number((process.hrtime.bigint() - this.startedAt) / NS_PER_MS); + + if (durationMs < this.config.slowHandleWarnMs) { + return; + } + + const steps = Array.from(this.timings.entries()) + .sort((first, second) => second[1] - first[1]) + .map(([step, stepDurationMs]) => `${step}=${stepDurationMs}ms`) + .join(' '); + + const titleField = context.title !== undefined + ? ` title=${JSON.stringify(context.title)}` + : ''; + + this.logger.warn( + `[slowHandle] duration=${durationMs}ms project=${context.projectId} type=${context.type} ` + + `payloadSize=${context.payloadSize}b deltaSize=${context.deltaSize}b${titleField} steps="${steps}"` + ); + } +} + +/** + * Factory for per-handle slow handle diagnostics sessions. + */ +export default class SlowHandleDiagnostics { + private readonly logger: LoggerLike; + private readonly metrics: GrouperMetrics; + private readonly config: GrouperDiagnosticsConfig; + + /** + * @param logger - logger instance. + * @param metrics - Grouper metrics facade used to record step duration. + * @param config - slow handle diagnostics thresholds. + */ + constructor(logger: LoggerLike, metrics: GrouperMetrics, config: GrouperDiagnosticsConfig) { + this.logger = logger; + this.metrics = metrics; + this.config = config; + } + + /** + * Start a new per-handle diagnostics session. + */ + public startSession(): SlowHandleSession { + return new SlowHandleSession(this.logger, this.metrics, this.config); + } +} From 330cad9a946cb7de82098138a2dbbc122ca771bf Mon Sep 17 00:00:00 2001 From: Kuchizu <70284260+Kuchizu@users.noreply.github.com> Date: Fri, 12 Jun 2026 18:57:24 +0300 Subject: [PATCH 2/6] fix(db): retry initial Mongo connection to avoid worker crash-loop (#559) * fix(db): retry initial Mongo connection to avoid worker crash-loop * fix(db): clamp Mongo reconnect env vars and test retry loop * refactor(db): move positiveIntEnv to utils --- lib/db/controller.test.ts | 44 +++++++++++++++++++++++++ lib/db/controller.ts | 66 +++++++++++++++++++++++++++++-------- lib/utils/positiveIntEnv.ts | 16 +++++++++ 3 files changed, 112 insertions(+), 14 deletions(-) create mode 100644 lib/utils/positiveIntEnv.ts diff --git a/lib/db/controller.test.ts b/lib/db/controller.test.ts index 1ec55b7c..c7870b94 100644 --- a/lib/db/controller.test.ts +++ b/lib/db/controller.test.ts @@ -1,5 +1,6 @@ import * as mongodb from 'mongodb'; import { DatabaseController } from './controller'; +import { DatabaseConnectionError } from '../workerErrors'; import '../../env-test'; /** @@ -24,4 +25,47 @@ describe('Database Controller Test', () => { expect(result).toBe(true); }); }); + + describe('initial handshake retry', () => { + const mongoModule = jest.requireActual('mongodb'); + let connectSpy: jest.SpyInstance; + + beforeEach(() => { + process.env.MONGO_RECONNECT_TRIES = '5'; + process.env.MONGO_RECONNECT_INTERVAL = '1'; + jest.spyOn(console, 'warn').mockImplementation(() => undefined); + connectSpy = jest.spyOn(mongoModule, 'connect'); + }); + + afterEach(() => { + delete process.env.MONGO_RECONNECT_TRIES; + delete process.env.MONGO_RECONNECT_INTERVAL; + jest.restoreAllMocks(); + }); + + it('retries the initial connection until it succeeds', async () => { + const fakeDb = {} as mongodb.Db; + const fakeClient = { db: jest.fn().mockReturnValue(fakeDb) } as unknown as mongodb.MongoClient; + + connectSpy + .mockRejectedValueOnce(new Error('unreachable')) + .mockRejectedValueOnce(new Error('unreachable')) + .mockResolvedValueOnce(fakeClient); + + const controller = new DatabaseController('mongodb://localhost:27017/test'); + const result = await controller.connect(); + + expect(connectSpy).toHaveBeenCalledTimes(3); + expect(result).toBe(fakeDb); + }); + + it('throws DatabaseConnectionError after exhausting all retries', async () => { + connectSpy.mockRejectedValue(new Error('unreachable')); + + const controller = new DatabaseController('mongodb://localhost:27017/test'); + + await expect(controller.connect()).rejects.toBeInstanceOf(DatabaseConnectionError); + expect(connectSpy).toHaveBeenCalledTimes(5); + }); + }); }); diff --git a/lib/db/controller.ts b/lib/db/controller.ts index 220db627..07de49ac 100644 --- a/lib/db/controller.ts +++ b/lib/db/controller.ts @@ -1,5 +1,22 @@ import { GridFSBucket, MongoClient, Db, connect } from 'mongodb'; import { DatabaseConnectionError } from '../workerErrors'; +import { positiveIntEnv } from '../utils/positiveIntEnv'; + +/** + * How many times to retry the initial Mongo handshake before giving up + */ +const DEFAULT_RECONNECT_TRIES = 60; + +/** + * Delay between initial-handshake retries, in ms + */ +const DEFAULT_RECONNECT_INTERVAL_MS = 3000; + +/** + * Bounds how long a single attempt waits for an available server, so a retry + * fails fast during an outage instead of hanging on the 30s driver default + */ +const SERVER_SELECTION_TIMEOUT_MS = 10000; /** * Database connection singleton @@ -46,28 +63,49 @@ export class DatabaseController { } /** - * Connect to database - * Requires `MONGO_DSN` environment variable to be set + * Connect to the database, retrying with a fixed backoff while the server is + * unreachable so a worker booting during a Mongo outage waits instead of + * crash-looping. The driver auto-recovers already-open connections on its + * own, so this retry covers the initial handshake only. * - * @throws {Error} if `MONGO_DSN` is not set + * Tunable via MONGO_RECONNECT_TRIES (default 60) and + * MONGO_RECONNECT_INTERVAL in ms (default 3000). + * + * @throws {DatabaseConnectionError} if every attempt fails */ public async connect(): Promise { if (this.db) { - return; + return this.db; } - try { - this.connection = await connect(this.connectionUri, { - useNewUrlParser: true, - useUnifiedTopology: true, - ...(this.appName ? { appName: this.appName } : {}), - }); - this.db = await this.connection.db(); + const tries = positiveIntEnv(process.env.MONGO_RECONNECT_TRIES, DEFAULT_RECONNECT_TRIES); + const intervalMs = positiveIntEnv(process.env.MONGO_RECONNECT_INTERVAL, DEFAULT_RECONNECT_INTERVAL_MS); - return this.db; - } catch (err) { - throw new DatabaseConnectionError(err); + for (let attempt = 1; attempt <= tries; attempt++) { + try { + this.connection = await connect(this.connectionUri, { + useNewUrlParser: true, + useUnifiedTopology: true, + serverSelectionTimeoutMS: SERVER_SELECTION_TIMEOUT_MS, + ...(this.appName ? { appName: this.appName } : {}), + }); + this.db = this.connection.db(); + + return this.db; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + + console.warn(`[Mongo] connect attempt ${attempt}/${tries} failed: ${message}`); + + if (attempt >= tries) { + throw new DatabaseConnectionError(err); + } + + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } } + + throw new DatabaseConnectionError('Failed to connect to MongoDB'); } /** diff --git a/lib/utils/positiveIntEnv.ts b/lib/utils/positiveIntEnv.ts new file mode 100644 index 00000000..06aff2e9 --- /dev/null +++ b/lib/utils/positiveIntEnv.ts @@ -0,0 +1,16 @@ +/** + * Parses a positive-integer env var, using `fallback` for missing, non-numeric, + * zero or negative values + * + * @param value - raw env var value + * @param fallback - default for an invalid value + */ +export function positiveIntEnv(value: string | undefined, fallback: number): number { + const parsed = Number(value); + + if (!Number.isFinite(parsed) || parsed < 1) { + return fallback; + } + + return Math.floor(parsed); +} From 97a262c6e364ed702cd69dc67aee248533ad773d Mon Sep 17 00:00:00 2001 From: Copilot <198982749+Copilot@users.noreply.github.com> Date: Fri, 12 Jun 2026 19:11:54 +0300 Subject: [PATCH 3/6] fix(task-manager): use event._id instead of groupHash in issue event URL (#569) * Initial plan * fix: use event._id instead of groupHash in task-manager event URL --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> --- workers/task-manager/src/utils/issue.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/workers/task-manager/src/utils/issue.ts b/workers/task-manager/src/utils/issue.ts index c6298edc..ef109824 100644 --- a/workers/task-manager/src/utils/issue.ts +++ b/workers/task-manager/src/utils/issue.ts @@ -106,7 +106,7 @@ export function formatIssueFromEvent(event: GroupedEventDBScheme, project: Proje const projectId = project._id.toString(); const garageUrl = process.env.GARAGE_URL || 'https://garage.hawk.so'; - const eventUrl = `${garageUrl}/project/${projectId}/event/${event.groupHash}`; + const eventUrl = `${garageUrl}/project/${projectId}/event/${event._id}`; /** * Format title: [Hawk] ${event.payload.title} From 53bc37d3d82fe1ad9d55feccd743c5a49325d583 Mon Sep 17 00:00:00 2001 From: Kuchizu <70284260+Kuchizu@users.noreply.github.com> Date: Tue, 23 Jun 2026 18:19:57 +0300 Subject: [PATCH 4/6] fix(limiter): handle workspace without tariffPlanId (#571) * fix(limiter): handle workspace without tariffPlanId * fix(limiter): avoid toString on missing tariffPlanId in error message --- workers/limiter/src/dbHelper.ts | 9 ++++- workers/limiter/tests/dbHelper.test.ts | 54 ++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/workers/limiter/src/dbHelper.ts b/workers/limiter/src/dbHelper.ts index b4cc845f..3af60b34 100644 --- a/workers/limiter/src/dbHelper.ts +++ b/workers/limiter/src/dbHelper.ts @@ -203,6 +203,13 @@ export class DbHelper { * @param planId - id of the plan to find */ private async resolvePlan(planId: WorkspaceDBScheme['tariffPlanId']): Promise { + /** + * Workspace may have no tariff plan assigned + */ + if (!planId) { + return null; + } + let plan = this.findPlanById(planId); if (plan) { @@ -252,7 +259,7 @@ export class DbHelper { const plan = await this.resolvePlan(workspace.tariffPlanId); if (!plan) { - throw new NonCriticalError(`Tariff plan ${workspace.tariffPlanId.toString()} not found for workspace ${id}`, { + throw new NonCriticalError(`Tariff plan ${workspace.tariffPlanId?.toString()} not found for workspace ${id}`, { workspaceId: id, }); } diff --git a/workers/limiter/tests/dbHelper.test.ts b/workers/limiter/tests/dbHelper.test.ts index 3e8eba54..f57be08d 100644 --- a/workers/limiter/tests/dbHelper.test.ts +++ b/workers/limiter/tests/dbHelper.test.ts @@ -363,6 +363,60 @@ describe('DbHelper', () => { ).rejects.toThrow(NonCriticalError); }); + test('Should skip and report a workspace that has no tariffPlanId', async () => { + /** + * Arrange — a workspace document with a missing tariffPlanId + */ + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10, + billingPeriodEventsCount: 0, + lastChargeDate: new Date(), + }); + + delete (workspace as Partial).tariffPlanId; + + await workspaceCollection.insertOne(workspace); + + const hawkCatcherSpy = jest.spyOn(HawkCatcher, 'send').mockImplementation(() => undefined); + + /** + * Act + */ + const workspaces = []; + + for await (const resolved of dbHelper.getWorkspacesWithTariffPlans()) { + workspaces.push(resolved); + } + + /** + * Assert — no crash, the workspace is skipped and reported + */ + expect(workspaces).toHaveLength(0); + expect(hawkCatcherSpy).toHaveBeenCalledTimes(1); + }); + + test('Should throw NonCriticalError when a single workspace has no tariffPlanId', async () => { + /** + * Arrange + */ + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10, + billingPeriodEventsCount: 0, + lastChargeDate: new Date(), + }); + + delete (workspace as Partial).tariffPlanId; + + await workspaceCollection.insertOne(workspace); + + /** + * Act & Assert + */ + await expect( + dbHelper.getWorkspacesWithTariffPlans(workspace._id.toString()) + ).rejects.toThrow(NonCriticalError); + }); + test('fetchPlans should throw CriticalError when the plans collection is empty', async () => { /** * Arrange — a helper pointed at an empty plans collection From 2ce8984980b2a32e80d616ba31c798ea117ef454 Mon Sep 17 00:00:00 2001 From: Dobrunia Kostrigin <48620984+Dobrunia@users.noreply.github.com> Date: Sun, 19 Jul 2026 17:37:17 +0300 Subject: [PATCH 5/6] Await provider.send in sender worker (#577) * Await provider.send in sender worker * test(sender): add test for provider.send rejection handling in worker --- workers/sender/src/index.ts | 14 +++++++------- workers/sender/tests/worker.test.ts | 29 +++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/workers/sender/src/index.ts b/workers/sender/src/index.ts index 242d2da8..a52b05c5 100644 --- a/workers/sender/src/index.ts +++ b/workers/sender/src/index.ts @@ -192,7 +192,7 @@ export default abstract class SenderWorker extends Worker { this.logger.info(`Sending ${notificationType} notification to ${channel.endpoint}`); - this.provider.send(channel.endpoint, { + await this.provider.send(channel.endpoint, { type: notificationType, payload: { host: process.env.GARAGE_URL, @@ -245,7 +245,7 @@ export default abstract class SenderWorker extends Worker { return; } - this.provider.send(endpoint, { + await this.provider.send(endpoint, { type: 'assignee', payload: { host: process.env.GARAGE_URL, @@ -506,7 +506,7 @@ export default abstract class SenderWorker extends Worker { return; } - this.provider.send(endpoint, { + await this.provider.send(endpoint, { type: 'payment-failed', payload: { host: process.env.GARAGE_URL, @@ -541,7 +541,7 @@ export default abstract class SenderWorker extends Worker { return; } - this.provider.send(endpoint, { + await this.provider.send(endpoint, { type: 'payment-success', payload: { host: process.env.GARAGE_URL, @@ -560,7 +560,7 @@ export default abstract class SenderWorker extends Worker { private async handlePasswordResetTask(task: SenderWorkerPasswordResetTask): Promise { const { newPassword, endpoint } = task.payload; - this.provider.send(endpoint, { + await this.provider.send(endpoint, { type: 'password-reset', payload: { host: process.env.GARAGE_URL, @@ -578,7 +578,7 @@ export default abstract class SenderWorker extends Worker { private async handleWorkspaceInviteTask(task: SenderWorkerWorkspaceInviteTask): Promise { const { workspaceName, inviteLink, endpoint } = task.payload; - this.provider.send(endpoint, { + await this.provider.send(endpoint, { type: 'workspace-invite', payload: { host: process.env.GARAGE_URL, @@ -597,7 +597,7 @@ export default abstract class SenderWorker extends Worker { private async handleSignUpTask(task: SenderWorkerSignUpTask): Promise { const { password, endpoint } = task.payload; - this.provider.send(endpoint, { + await this.provider.send(endpoint, { type: 'sign-up', payload: { host: process.env.GARAGE_URL, diff --git a/workers/sender/tests/worker.test.ts b/workers/sender/tests/worker.test.ts index 73a9b5ae..ef5b0512 100644 --- a/workers/sender/tests/worker.test.ts +++ b/workers/sender/tests/worker.test.ts @@ -249,4 +249,33 @@ describe('Sender Worker', () => { expect(dailyEventsQueryMock).toBeCalledWith({ groupHash: 'groupHash' }); }); }); + + describe('provider.send awaiting', () => { + /** + * Without await, handle() resolves even if send() rejects — + * message would be acked and the error lost. + * The .catch on the rejected promise only prevents an unhandledRejection + * crash in the fire-and-forget case; await still observes the rejection. + */ + it('should reject handle when provider.send fails', async () => { + const worker = new ExampleSenderWorker(); + const sendError = new Error('provider send failed'); + + (worker as any).provider.send = jest.fn(() => { + const sendPromise = Promise.reject(sendError); + + sendPromise.catch(() => undefined); + + return sendPromise; + }); + + await expect(worker.handle({ + type: 'sign-up', + payload: { + password: 'secret', + endpoint: 'user@example.com', + }, + })).rejects.toThrow('provider send failed'); + }); + }); }); From 50fc9075096af719b1a0164f522f6bd90a66bd38 Mon Sep 17 00:00:00 2001 From: Kuchizu <70284260+Kuchizu@users.noreply.github.com> Date: Tue, 21 Jul 2026 14:57:03 +0300 Subject: [PATCH 6/6] Fix appName option and count billing events via dailyEvents (#575) * Fix appName option ignored by mongodb driver 3.x * Count billing events via dailyEvents daily counters * Fix limiter worker tests for boundary-day counting * Add opt-in dailyEvents limiter counter * Compare old and new limiter counters via Telegram log --- .env.sample | 3 + lib/db/controller.ts | 3 +- workers/limiter/src/dbHelper.ts | 127 +++++++++++++- workers/limiter/src/index.ts | 65 +++++++- workers/limiter/tests/dbHelper.test.ts | 219 ++++++++++++++++++++++++- workers/limiter/tests/index.test.ts | 74 ++++++++- 6 files changed, 471 insertions(+), 20 deletions(-) diff --git a/.env.sample b/.env.sample index f2df9be2..a5ecb4dc 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 +## 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/lib/db/controller.ts b/lib/db/controller.ts index 07de49ac..9cd24e18 100644 --- a/lib/db/controller.ts +++ b/lib/db/controller.ts @@ -87,7 +87,8 @@ export class DatabaseController { useNewUrlParser: true, useUnifiedTopology: true, serverSelectionTimeoutMS: SERVER_SELECTION_TIMEOUT_MS, - ...(this.appName ? { appName: this.appName } : {}), + /** driver 3.x only recognizes the lowercase `appname` option key */ + ...(this.appName ? { appname: this.appName } : {}), }); this.db = this.connection.db(); diff --git a/workers/limiter/src/dbHelper.ts b/workers/limiter/src/dbHelper.ts index 3af60b34..984dd335 100644 --- a/workers/limiter/src/dbHelper.ts +++ b/workers/limiter/src/dbHelper.ts @@ -3,6 +3,10 @@ import { PlanDBScheme, ProjectDBScheme, WorkspaceDBScheme } from '@hawk.so/types import { WorkspaceWithTariffPlan } from '../types'; import HawkCatcher from '@hawk.so/nodejs'; import { CriticalError, NonCriticalError } from '../../../lib/workerErrors'; +import { MS_IN_SEC } from '../../../lib/utils/consts'; +import TimeMs from '../../../lib/utils/time'; + +const SEC_IN_DAY = TimeMs.DAY / TimeMs.SECOND; const WORKSPACE_PROJECTION = { _id: 1, @@ -145,19 +149,13 @@ export class DbHelper { since: number ): Promise { try { - const repetitionsCollection = this.eventsDbConnection.collection('repetitions:' + project._id.toString()); - const eventsCollection = this.eventsDbConnection.collection('events:' + project._id.toString()); - const query = { timestamp: { $gt: since, }, }; - const repetitionsCount = await repetitionsCollection.countDocuments(query); - const originalEventCount = await eventsCollection.countDocuments(query); - - return repetitionsCount + originalEventCount; + return await this.getRawEventsCountByProject(project, query); } catch (e) { HawkCatcher.send(e); throw new CriticalError(e); @@ -179,6 +177,75 @@ export class DbHelper { .then(sum); } + /** + * Returns total event counts for last billing period using dailyEvents counters. + * + * Full days are summed from dailyEvents per-day counters (grouper + * 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. + * + * @param project - project to check + * @param since - timestamp of the time from which we count the events + */ + public async getEventsCountByProjectUsingDailyEvents( + 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); + } + } + + /** + * 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); + } + /** * Returns all projects from Database or projects of the specified workspace * @@ -197,6 +264,52 @@ export class DbHelper { return this.projectsCollection.find(query).toArray(); } + /** + * UTC midnight right after the given timestamp. Mirrors grouper's + * getMidnightByEventTimestamp, which fills the dailyEvents buckets. + * + * @param timestamp - unix timestamp in seconds + */ + private getNextUtcMidnight(timestamp: number): number { + const date = new Date(timestamp * MS_IN_SEC); + + date.setUTCDate(date.getUTCDate() + 1); + date.setUTCHours(0, 0, 0, 0); + + return date.getTime() / MS_IN_SEC; + } + + /** + * Returns first dailyEvents bucket that can be safely used without counting + * events before the requested timestamp. + * + * @param timestamp - unix timestamp in seconds + */ + private getFirstFullDailyEventsTimestamp(timestamp: number): number { + const midnight = timestamp - (timestamp % SEC_IN_DAY); + + return timestamp === midnight ? timestamp : this.getNextUtcMidnight(timestamp); + } + + /** + * Counts raw original events and repetitions for the passed query. + * + * @param project - project to check + * @param query - MongoDB timestamp query + */ + private async getRawEventsCountByProject(project: ProjectDBScheme, query: Record): Promise { + const projectId = project._id.toString(); + const repetitionsCollection = this.eventsDbConnection.collection('repetitions:' + projectId); + const eventsCollection = this.eventsDbConnection.collection('events:' + projectId); + + const [repetitionsCount, originalEventCount] = await Promise.all([ + repetitionsCollection.countDocuments(query), + eventsCollection.countDocuments(query), + ]); + + return repetitionsCount + originalEventCount; + } + /** * Returns plan from cache, refetches once on miss * diff --git a/workers/limiter/src/index.ts b/workers/limiter/src/index.ts index 6ed21cfe..cc38d768 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.getEventsCountByProjects(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,69 @@ 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); + } + /** * Method that formats project list to html used in report messages * diff --git a/workers/limiter/tests/dbHelper.test.ts b/workers/limiter/tests/dbHelper.test.ts index f57be08d..dd26baa7 100644 --- a/workers/limiter/tests/dbHelper.test.ts +++ b/workers/limiter/tests/dbHelper.test.ts @@ -9,9 +9,20 @@ import HawkCatcher from '@hawk.so/nodejs'; /** * Constant of last charge date in all workspaces for tests + * 2020-04-01T12:00:00Z — intentionally not midnight-aligned */ const LAST_CHARGE_DATE = new Date(1585742400 * 1000); +/** + * Timestamp inside the boundary day of LAST_CHARGE_DATE (2020-04-01T16:00:00Z) + */ +const BOUNDARY_DAY_TIMESTAMP = 1585756800; + +/** + * UTC midnight right after LAST_CHARGE_DATE (2020-04-02T00:00:00Z) + */ +const NEXT_MIDNIGHT_AFTER_LAST_CHARGE = 1585785600; + describe('DbHelper', () => { let connection: MongoClient; let db: Db; @@ -66,8 +77,10 @@ describe('DbHelper', () => { /** * Returns mocked event for tests + * + * @param timestamp - event timestamp, defaults to the boundary day of LAST_CHARGE_DATE */ - const createEventMock = (): GroupedEventDBScheme => { + const createEventMock = (timestamp: number = BOUNDARY_DAY_TIMESTAMP): GroupedEventDBScheme => { return { catcherType: '', totalCount: 0, @@ -77,7 +90,7 @@ describe('DbHelper', () => { payload: { title: 'Mocked event', }, - timestamp: 1586892935, + timestamp, }; }; @@ -89,11 +102,13 @@ describe('DbHelper', () => { const fillDatabaseWithMockedData = async (parameters: { workspace?: WorkspaceDBScheme, project: ProjectDBScheme, - eventsToMock: number + eventsToMock: number, repetitionsToMock?: number, + dailyEventsToMock?: Array<{ groupingTimestamp: number; count: number }>, }): Promise => { const eventsCollection = db.collection(`events:${parameters.project._id.toString()}`); const repetitionsCollection = db.collection(`repetitions:${parameters.project._id.toString()}`); + const dailyEventsCollection = db.collection(`dailyEvents:${parameters.project._id.toString()}`); if (parameters.workspace) { await workspaceCollection.insertOne(parameters.workspace); @@ -104,7 +119,9 @@ describe('DbHelper', () => { for (let i = 0; i < parameters.eventsToMock; i++) { mockedEvents.push(createEventMock()); } - await eventsCollection.insertMany(mockedEvents); + if (mockedEvents.length > 0) { + await eventsCollection.insertMany(mockedEvents); + } mockedEvents.length = 0; @@ -114,6 +131,14 @@ describe('DbHelper', () => { } await repetitionsCollection.insertMany(mockedEvents); } + + if (parameters.dailyEventsToMock?.length > 0) { + await dailyEventsCollection.insertMany(parameters.dailyEventsToMock.map(bucket => ({ + groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62', + groupingTimestamp: bucket.groupingTimestamp, + count: bucket.count, + }))); + } }; beforeAll(async () => { @@ -554,7 +579,7 @@ describe('DbHelper', () => { }); describe('getEventsCountByProject', () => { - test('Should count events and repetitions for a project', async () => { + test('Should count boundary-day events and repetitions for a project', async () => { /** * Arrange */ @@ -584,10 +609,180 @@ describe('DbHelper', () => { */ expect(count).toBe(10); // 5 events + 5 repetitions }); + + test('Should keep raw counting as default and include raw docs after the boundary day', 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: 1, + dailyEventsToMock: [ + { + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE, + count: 100, + }, + ], + }); + + await db.collection(`events:${project._id.toString()}`).insertOne( + createEventMock(NEXT_MIDNIGHT_AFTER_LAST_CHARGE + 100) + ); + + /** + * Act + */ + const count = await dbHelper.getEventsCountByProject(project, since); + + /** + * Assert + */ + expect(count).toBe(2); + }); + }); + + describe('getEventsCountByProjectUsingDailyEvents', () => { + test('Should add per-day counters from dailyEvents for days after the boundary day', async () => { + /** + * Arrange + */ + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10, + billingPeriodEventsCount: 0, + lastChargeDate: new Date(), + }); + const project = createProjectMock({ workspaceId: workspace._id }); + + await fillDatabaseWithMockedData({ + workspace, + project, + eventsToMock: 2, + repetitionsToMock: 3, + dailyEventsToMock: [ + { + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE, + count: 7, + }, + { + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE + 86400, + count: 3, + }, + ], + }); + + const since = Math.floor(LAST_CHARGE_DATE.getTime() / MS_IN_SEC); + + /** + * Act + */ + const count = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); + + /** + * Assert + */ + expect(count).toBe(15); // 2 events + 3 repetitions on the boundary day + 7 + 3 from dailyEvents + }); + + test('Should ignore raw docs outside the boundary day and dailyEvents buckets before it', 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: 1, + dailyEventsToMock: [ + /** bucket of the boundary day itself must not be counted */ + { + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE - 86400, + count: 100, + }, + ], + }); + + const eventsCollection = db.collection(`events:${project._id.toString()}`); + + await eventsCollection.insertMany([ + /** before lastChargeDate */ + createEventMock(since - 100), + /** after the boundary day — counted via dailyEvents, not the raw scan */ + createEventMock(NEXT_MIDNIGHT_AFTER_LAST_CHARGE + 100), + ]); + + /** + * Act + */ + const count = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); + + /** + * Assert + */ + expect(count).toBe(1); // only the single boundary-day event + }); + + test('Should count dailyEvents bucket at since when since is already UTC midnight', async () => { + /** + * Arrange + */ + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10, + billingPeriodEventsCount: 0, + lastChargeDate: new Date(), + }); + const project = createProjectMock({ workspaceId: workspace._id }); + const since = NEXT_MIDNIGHT_AFTER_LAST_CHARGE; + + await fillDatabaseWithMockedData({ + workspace, + project, + eventsToMock: 0, + dailyEventsToMock: [ + { + groupingTimestamp: since, + count: 4, + }, + { + groupingTimestamp: since + 86400, + count: 3, + }, + ], + }); + + await db.collection(`events:${project._id.toString()}`).insertOne( + createEventMock(since + 100) + ); + + /** + * Act + */ + const count = await dbHelper.getEventsCountByProjectUsingDailyEvents(project, since); + + /** + * Assert + */ + expect(count).toBe(7); + }); }); - describe('getEventsCountByProjects', () => { - test('Should count events and repetitions for multiple projects', async () => { + describe('getEventsCountByProjectsUsingDailyEvents', () => { + test('Should count events, repetitions and dailyEvents for multiple projects', async () => { /** * Arrange */ @@ -604,6 +799,12 @@ describe('DbHelper', () => { project: project1, eventsToMock: 5, repetitionsToMock: 5, + dailyEventsToMock: [ + { + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE, + count: 6, + }, + ], }); await fillDatabaseWithMockedData({ project: project2, @@ -616,12 +817,12 @@ describe('DbHelper', () => { /** * Act */ - const count = await dbHelper.getEventsCountByProjects([project1, project2], since); + const count = await dbHelper.getEventsCountByProjectsUsingDailyEvents([project1, project2], since); /** * Assert */ - expect(count).toBe(16); // (5 + 5) + (3 + 3) events and repetitions + expect(count).toBe(22); // (5 + 5 + 6) + (3 + 3) }); }); diff --git a/workers/limiter/tests/index.test.ts b/workers/limiter/tests/index.test.ts index 8a7ebf7b..1c6c7648 100644 --- a/workers/limiter/tests/index.test.ts +++ b/workers/limiter/tests/index.test.ts @@ -27,6 +27,17 @@ const REGULAR_WORKSPACES_CHECK_EVENT: RegularWorkspacesCheckEvent = { */ const LAST_CHARGE_DATE = new Date(1585742400 * 1000); +/** + * Timestamp inside the boundary day of LAST_CHARGE_DATE (2020-04-01T16:00:00Z): + * such events are counted from the raw collections, later ones — via dailyEvents + */ +const BOUNDARY_DAY_TIMESTAMP = 1585756800; + +/** + * UTC midnight right after LAST_CHARGE_DATE (2020-04-02T00:00:00Z) + */ +const NEXT_MIDNIGHT_AFTER_LAST_CHARGE = 1585785600; + describe('Limiter worker', () => { let connection: MongoClient; let db: Db; @@ -89,7 +100,7 @@ describe('Limiter worker', () => { usersAffected: 0, visitedBy: [], groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62', - timestamp: 1586892935, + timestamp: BOUNDARY_DAY_TIMESTAMP, payload: { title: 'Mocked event', }, @@ -104,7 +115,7 @@ describe('Limiter worker', () => { const fillDatabaseWithMockedData = async (parameters: { workspace: WorkspaceDBScheme, project: ProjectDBScheme, - eventsToMock: number + eventsToMock: number, repetitionsToMock?: number, }): Promise => { const eventsCollection = db.collection(`events:${parameters.project._id.toString()}`); @@ -320,6 +331,65 @@ 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 () => { + /** + * Arrange + */ + const workspace = createWorkspaceMock({ + plan: mockedPlans.eventsLimit10000, + billingPeriodEventsCount: 0, + lastChargeDate: LAST_CHARGE_DATE, + }); + const project = createProjectMock({ workspaceId: workspace._id }); + + await fillDatabaseWithMockedData({ + workspace, + project, + eventsToMock: 5, + }); + + /** + * Bucket for the day after the boundary day — counted only by the new algorithm + */ + await db.collection(`dailyEvents:${project._id.toString()}`).insertOne({ + groupHash: 'ade987831d0d0d167aeea685b49db164eb4e113fd027858eef7f69d049357f62', + groupingTimestamp: NEXT_MIDNIGHT_AFTER_LAST_CHARGE, + 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; + } + + /** + * Assert — the new counter result is saved, both results are reported with timings + */ + const workspaceInDatabase = await workspaceCollection.findOne({ + _id: workspace._id, + }); + + 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/); + }); + test('Should not send a report when no projects are blocked or unblocked', async () => { /** * Arrange