diff --git a/.changeset/giant-cups-matter.md b/.changeset/giant-cups-matter.md new file mode 100644 index 0000000000..20600e2245 --- /dev/null +++ b/.changeset/giant-cups-matter.md @@ -0,0 +1,5 @@ +--- +'hive': patch +--- + +Runs metric alerts cron with a concurrency pool diff --git a/packages/services/workflows/package.json b/packages/services/workflows/package.json index 19b847f8c2..7ed39cb86a 100644 --- a/packages/services/workflows/package.json +++ b/packages/services/workflows/package.json @@ -31,6 +31,7 @@ "graphql-yoga": "5.13.3", "mjml": "4.14.0", "nodemailer": "9.0.1", + "p-limit": "6.2.0", "sendmail": "1.6.1", "zod": "3.25.76" } diff --git a/packages/services/workflows/src/tasks/evaluate-metric-alert-rules.ts b/packages/services/workflows/src/tasks/evaluate-metric-alert-rules.ts index 4c44956e64..8ea16da925 100644 --- a/packages/services/workflows/src/tasks/evaluate-metric-alert-rules.ts +++ b/packages/services/workflows/src/tasks/evaluate-metric-alert-rules.ts @@ -1,3 +1,4 @@ +import pLimit from 'p-limit'; import { z } from 'zod'; import { psql } from '@hive/postgres'; import { SpanKind, SpanStatusCode, trace } from '@hive/service-common'; @@ -175,28 +176,25 @@ export const task = implementTask(EvaluateMetricAlertRulesTask, async args => { }, async span => { try { - // Bounded parallelism over groups: each group = 1 CH query + per-rule - // state writes. allSettled (not Promise.all) so that one unexpected - // throw inside evaluateRule doesn't strand the rest of the batch's - // successful work; the throwing group's rules get re-evaluated on the - // next cron tick (60s) rather than via graphile-worker retries, which - // would otherwise re-run work that's already idempotently committed. - // With GROUP_CONCURRENCY=5 we cut wall-clock per tick by up to 5x - // without exhausting the PG pool. + // Fixed-capacity pool: keep GROUP_CONCURRENCY groups in flight at once so + // a slow group can't idle the other slots (a batch barrier would wait for + // the whole batch before starting the next). allSettled so one thrown + // group doesn't strand the rest; it re-evaluates on the next 60s tick, + // not via graphile-worker retries that would re-run already-committed work. + const limit = pLimit(GROUP_CONCURRENCY); let groupsFailed = 0; const evaluatedRuleIds: string[] = []; - for (let i = 0; i < groupList.length; i += GROUP_CONCURRENCY) { - const batch = groupList.slice(i, i + GROUP_CONCURRENCY); - const results = await Promise.allSettled(batch.map(processGroup)); - for (const r of results) { - if (r.status === 'rejected') { - groupsFailed++; - logger.error({ error: r.reason }, 'Group evaluation threw unexpectedly'); - } else if (r.value.failed) { - groupsFailed++; - } else { - evaluatedRuleIds.push(...r.value.evaluatedIds); - } + const results = await Promise.allSettled( + groupList.map(group => limit(() => processGroup(group))), + ); + for (const r of results) { + if (r.status === 'rejected') { + groupsFailed++; + logger.error({ error: r.reason }, 'Group evaluation threw unexpectedly'); + } else if (r.value.failed) { + groupsFailed++; + } else { + evaluatedRuleIds.push(...r.value.evaluatedIds); } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ae11dbc2db..c490dec7e8 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2071,6 +2071,9 @@ importers: nodemailer: specifier: 9.0.1 version: 9.0.1 + p-limit: + specifier: 6.2.0 + version: 6.2.0 sendmail: specifier: 1.6.1 version: 1.6.1