diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1a7212d758..e6e4f24f8b 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -970,7 +970,7 @@ Keep old endpoints for backward compatibility but annotate: An endpoint that does nothing but wait for someone else to act — its response time is determined solely by when another request or process triggers the event, and it performs no work of its own meanwhile — **must** carry `wait` as its own path segment, unless it is listed as an explicit exemption below. Conversely, an endpoint expected to answer quickly **must not** use `wait` as a path segment. -This does **not** cover an endpoint that starts an operation and then waits for it to finish — broadcasting a transaction and awaiting its confirmation, for example. That duration reflects work the API itself set in motion, which makes it a legitimate monitoring signal, so those endpoints stay visible and must **not** be named `wait`. Current examples: `PUT /v1/sell/paymentInfos/:id/confirm` and `PUT /v1/swap/paymentInfos/:id/confirm` (both in their `authorization` branch), `PUT /v1/realunit/sell/:id/confirm` (`eip7702` branch) and `PUT /v1/realunit/transfer/:id/confirm`. +This does **not** cover an endpoint that starts an operation and then waits for it to finish — broadcasting a transaction and awaiting its confirmation, for example. That duration reflects work the API itself set in motion, which makes it a legitimate monitoring signal, so those endpoints stay visible and must **not** be named `wait`. Current examples: `PUT /v1/sell/paymentInfos/:id/confirm` and `PUT /v1/swap/paymentInfos/:id/confirm` (both in their `authorization` branch), `PUT /v1/realunit/sell/:id/confirm` (`eip7702` branch), `PUT /v1/realunit/transfer/:id/confirm`, and `GET /v1/auth/mail/confirm`, which enqueues the account-merge job, kicks the dispatcher and waits up to 900 ms for the result before handing out a ticket — the duration measures the merge the API itself started, so it stays in the latency view. Endpoints that block by design: diff --git a/migration/1785600000000-AddJobTable.js b/migration/1785600000000-AddJobTable.js new file mode 100644 index 0000000000..9343bc3327 --- /dev/null +++ b/migration/1785600000000-AddJobTable.js @@ -0,0 +1,57 @@ +/** + * @typedef {import('typeorm').MigrationInterface} MigrationInterface + * @typedef {import('typeorm').QueryRunner} QueryRunner + */ + +/** + * Create the `job` table backing the generic command-job model: an endpoint whose work no longer fits + * inside a request persists a job, answers with its `uid` as a ticket, and a dispatcher executes it. + * + * Two indexes, both load-bearing rather than decorative: + * + * - `("status", "group", "id")` is the dispatcher's access path. Every claim attempt reads the oldest + * claimable rows of one group; without the index that is a sequential scan per sweep, once per group + * per minute. + * - `("group", "idempotencyKey")` UNIQUE carries idempotency in the database rather than in code. Two + * concurrent enqueues of the same business key cannot both create a job — the loser gets a unique + * violation and loads the winner's row. This is what makes a retried confirmation link harmless. + * + * `attempt` defaults to 0 so existing-row concerns do not arise (the table starts empty), and + * `maxAttempts` is written per job on purpose: a job carries the limit it was created under, so a + * later configuration change cannot silently re-budget work that is already in flight. + * + * The `userDataId` foreign key is nullable because a job may be triggered by an unauthenticated flow + * (the account-merge confirmation link is the first case). Where it is set, the status endpoint uses it + * to refuse a job that belongs to another account. + * + * @param {QueryRunner} queryRunner + */ +module.exports = class AddJobTable1785600000000 { + name = 'AddJobTable1785600000000'; + + /** + * @param {QueryRunner} queryRunner + */ + async up(queryRunner) { + await queryRunner.query( + `CREATE TABLE "job" ("id" SERIAL NOT NULL, "updated" TIMESTAMP NOT NULL DEFAULT now(), "created" TIMESTAMP NOT NULL DEFAULT now(), "uid" character varying(256) NOT NULL, "group" character varying(256) NOT NULL, "status" character varying(256) NOT NULL, "idempotencyKey" character varying(256) NOT NULL, "input" text NOT NULL, "output" text, "attempt" integer NOT NULL DEFAULT '0', "maxAttempts" integer NOT NULL, "claimedAt" TIMESTAMP, "claimedBy" character varying(256), "startedAt" TIMESTAMP, "finishedAt" TIMESTAMP, "nextAttemptAt" TIMESTAMP, "error" text, "traceparent" character varying(256), "userDataId" integer, CONSTRAINT "UQ_84d390c08e353b6202e6495bada" UNIQUE ("uid"), CONSTRAINT "PK_98ab1c14ff8d1cf80d18703b92f" PRIMARY KEY ("id"))`, + ); + await queryRunner.query(`CREATE INDEX "IDX_e8aa536bda884c9f342b9783ca" ON "job" ("status", "group", "id")`); + await queryRunner.query( + `CREATE UNIQUE INDEX "IDX_595111195612c6fb0889278bce" ON "job" ("group", "idempotencyKey")`, + ); + await queryRunner.query( + `ALTER TABLE "job" ADD CONSTRAINT "FK_df137d89d7c792e8a5d7d97719c" FOREIGN KEY ("userDataId") REFERENCES "user_data"("id") ON DELETE NO ACTION ON UPDATE NO ACTION`, + ); + } + + /** + * @param {QueryRunner} queryRunner + */ + async down(queryRunner) { + await queryRunner.query(`ALTER TABLE "job" DROP CONSTRAINT "FK_df137d89d7c792e8a5d7d97719c"`); + await queryRunner.query(`DROP INDEX "public"."IDX_595111195612c6fb0889278bce"`); + await queryRunner.query(`DROP INDEX "public"."IDX_e8aa536bda884c9f342b9783ca"`); + await queryRunner.query(`DROP TABLE "job"`); + } +}; diff --git a/migration/1785700000000-AddJobAttemptTable.js b/migration/1785700000000-AddJobAttemptTable.js new file mode 100644 index 0000000000..954d640c0b --- /dev/null +++ b/migration/1785700000000-AddJobAttemptTable.js @@ -0,0 +1,44 @@ +/** + * @typedef {import('typeorm').MigrationInterface} MigrationInterface + * @typedef {import('typeorm').QueryRunner} QueryRunner + */ + +/** + * Create the append-only `job_attempt` table: one immutable row per claim of a job, recording who took the + * attempt, when, and how it ended (including the full error). The `job` snapshot row alone cannot carry that + * history — every retry would overwrite `claimedAt`, `claimedBy`, `startedAt` and `error` and destroy the + * prior attempt's proof. This table is the durable event log those overwrites would otherwise erase. + * + * The unique index on `("jobId", "attempt")` ensures the same attempt of the same job can never be recorded + * twice, no matter how often claimNext, finish, abort or recoverStale run against it. `jobId` is NOT NULL + * because an attempt always belongs to exactly one job. + * + * @param {QueryRunner} queryRunner + */ +module.exports = class AddJobAttemptTable1785700000000 { + name = 'AddJobAttemptTable1785700000000'; + + /** + * @param {QueryRunner} queryRunner + */ + async up(queryRunner) { + await queryRunner.query( + `CREATE TABLE "job_attempt" ("id" SERIAL NOT NULL, "updated" TIMESTAMP NOT NULL DEFAULT now(), "created" TIMESTAMP NOT NULL DEFAULT now(), "jobId" integer NOT NULL, "attempt" integer NOT NULL, "claimedBy" character varying(256) NOT NULL, "claimedAt" TIMESTAMP NOT NULL, "finishedAt" TIMESTAMP, "outcome" character varying(256), "error" text, CONSTRAINT "PK_cae15a4b6ee8638059fb15054e2" PRIMARY KEY ("id"))`, + ); + await queryRunner.query( + `CREATE UNIQUE INDEX "IDX_b501e31b3f9bd927efe01b7a96" ON "job_attempt" ("jobId", "attempt")`, + ); + await queryRunner.query( + `ALTER TABLE "job_attempt" ADD CONSTRAINT "FK_cbf1dd429e2b8c2afe2d71a6d8e" FOREIGN KEY ("jobId") REFERENCES "job"("id") ON DELETE NO ACTION ON UPDATE NO ACTION`, + ); + } + + /** + * @param {QueryRunner} queryRunner + */ + async down(queryRunner) { + await queryRunner.query(`ALTER TABLE "job_attempt" DROP CONSTRAINT "FK_cbf1dd429e2b8c2afe2d71a6d8e"`); + await queryRunner.query(`DROP INDEX "public"."IDX_b501e31b3f9bd927efe01b7a96"`); + await queryRunner.query(`DROP TABLE "job_attempt"`); + } +}; diff --git a/package-lock.json b/package-lock.json index b293fef6e1..875bec180f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -44,7 +44,9 @@ "@noble/curves": "^1.9.7", "@opentelemetry/api": "^1.9.1", "@opentelemetry/auto-instrumentations-node": "^0.76.0", + "@opentelemetry/exporter-metrics-otlp-http": "^0.218.0", "@opentelemetry/exporter-trace-otlp-http": "^0.218.0", + "@opentelemetry/sdk-metrics": "^2.7.1", "@opentelemetry/sdk-node": "^0.218.0", "@opentelemetry/sdk-trace-base": "^2.7.1", "@railgun-community/engine": "^9.4.0", diff --git a/package.json b/package.json index 74a9f61938..cc2729a471 100644 --- a/package.json +++ b/package.json @@ -63,7 +63,9 @@ "@noble/curves": "^1.9.7", "@opentelemetry/api": "^1.9.1", "@opentelemetry/auto-instrumentations-node": "^0.76.0", + "@opentelemetry/exporter-metrics-otlp-http": "^0.218.0", "@opentelemetry/exporter-trace-otlp-http": "^0.218.0", + "@opentelemetry/sdk-metrics": "^2.7.1", "@opentelemetry/sdk-node": "^0.218.0", "@opentelemetry/sdk-trace-base": "^2.7.1", "@railgun-community/engine": "^9.4.0", diff --git a/src/__tests__/tracing.spec.ts b/src/__tests__/tracing.spec.ts index 424a36e058..315e7e39e9 100644 --- a/src/__tests__/tracing.spec.ts +++ b/src/__tests__/tracing.spec.ts @@ -9,8 +9,15 @@ jest.mock('@opentelemetry/auto-instrumentations-node', () => ({ jest.mock('@opentelemetry/exporter-trace-otlp-http', () => ({ OTLPTraceExporter: jest.fn(), })); +jest.mock('@opentelemetry/exporter-metrics-otlp-http', () => ({ + OTLPMetricExporter: jest.fn(), +})); +jest.mock('@opentelemetry/sdk-metrics', () => ({ + PeriodicExportingMetricReader: jest.fn(), +})); import { SpanKind, SpanStatusCode } from '@opentelemetry/api'; +import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics'; import { ReadableSpan } from '@opentelemetry/sdk-trace-base'; import { ClientErrorSpanProcessor, isClientError, startTracing } from '../tracing'; @@ -73,6 +80,7 @@ describe('startTracing', () => { afterEach(() => { mockStart.mockClear(); + (PeriodicExportingMetricReader as unknown as jest.Mock).mockClear(); if (original === undefined) delete process.env.OTEL_EXPORTER_OTLP_ENDPOINT; else process.env.OTEL_EXPORTER_OTLP_ENDPOINT = original; }); @@ -82,10 +90,39 @@ describe('startTracing', () => { expect(startTracing()).toBeUndefined(); }); + it('returns undefined and does not build a metric reader without OTEL_EXPORTER_OTLP_ENDPOINT', () => { + delete process.env.OTEL_EXPORTER_OTLP_ENDPOINT; + (PeriodicExportingMetricReader as unknown as jest.Mock).mockClear(); + + expect(startTracing()).toBeUndefined(); + expect(PeriodicExportingMetricReader).not.toHaveBeenCalled(); + }); + it('starts the SDK when an endpoint is configured', () => { process.env.OTEL_EXPORTER_OTLP_ENDPOINT = 'http://localhost:4318'; const sdk = startTracing(); expect(sdk).toBeDefined(); expect(mockStart).toHaveBeenCalledTimes(1); }); + + it('constructs PeriodicExportingMetricReader exactly once with exportIntervalMillis 15000 when endpoint is set', async () => { + process.env.OTEL_EXPORTER_OTLP_ENDPOINT = 'http://localhost:4318'; + (PeriodicExportingMetricReader as unknown as jest.Mock).mockClear(); + + // Re-import in isolation so the module-level sdk cache is empty and the + // module-level startTracing() runs with the endpoint set. + let isolatedStart: typeof startTracing; + await jest.isolateModulesAsync(async () => { + isolatedStart = (await import('../tracing')).startTracing; + }); + + expect(PeriodicExportingMetricReader).toHaveBeenCalledTimes(1); + expect(PeriodicExportingMetricReader).toHaveBeenCalledWith( + expect.objectContaining({ exportIntervalMillis: 15000 }), + ); + + // Calling again must not construct a second reader (singleton). + isolatedStart!(); + expect(PeriodicExportingMetricReader).toHaveBeenCalledTimes(1); + }); }); diff --git a/src/config/config.ts b/src/config/config.ts index a297cdffc3..cf2dbb05bd 100644 --- a/src/config/config.ts +++ b/src/config/config.ts @@ -117,6 +117,7 @@ export class Configuration { paymentLinkPaymentUidPrefix: 'plp', paymentQuoteUidPrefix: 'plq', realUnitTransferUidPrefix: 'RT', + jobUidPrefix: 'J', }; moderators = { diff --git a/src/shared/models/setting/setting-schema.registry.ts b/src/shared/models/setting/setting-schema.registry.ts index be2c130db7..15b7535238 100644 --- a/src/shared/models/setting/setting-schema.registry.ts +++ b/src/shared/models/setting/setting-schema.registry.ts @@ -1,4 +1,5 @@ import { Type } from '@nestjs/common'; +import { JobGroupSettingDto } from 'src/subdomains/supporting/job/dto/job-group-setting.dto'; import { CustomSignUpFeesDto } from './dto/custom-sign-up-fees.dto'; import { ManualLogPositionDto } from './dto/manual-log-position.dto'; import { SupportClerkAccountDto } from './dto/support-clerk-account.dto'; @@ -37,6 +38,9 @@ export const SettingSchemaRegistry: Record = { // Compliance complianceClerks: 'string[]', + + // Job Groups + jobGroups: { type: 'array', items: JobGroupSettingDto }, }; export function isArraySchema(schema: SettingSchema): schema is ArraySchema { diff --git a/src/shared/services/__tests__/metric.service.spec.ts b/src/shared/services/__tests__/metric.service.spec.ts new file mode 100644 index 0000000000..f489d0964f --- /dev/null +++ b/src/shared/services/__tests__/metric.service.spec.ts @@ -0,0 +1,72 @@ +import { metrics } from '@opentelemetry/api'; +import { MetricService } from '../metric.service'; + +describe('MetricService', () => { + const counterAdd = jest.fn(); + const histogramRecord = jest.fn(); + const gaugeAddCallback = jest.fn(); + const createCounter = jest.fn(() => ({ add: counterAdd })); + const createHistogram = jest.fn(() => ({ record: histogramRecord })); + const createObservableGauge = jest.fn(() => ({ addCallback: gaugeAddCallback })); + + beforeEach(() => { + jest.spyOn(metrics, 'getMeter').mockReturnValue({ + createCounter, + createHistogram, + createObservableGauge, + } as never); + createCounter.mockClear(); + createHistogram.mockClear(); + createObservableGauge.mockClear(); + counterAdd.mockClear(); + histogramRecord.mockClear(); + gaugeAddCallback.mockClear(); + }); + + afterEach(() => jest.restoreAllMocks()); + + it('creates one counter per name and increments by 1 with attributes', () => { + const service = new MetricService(); + const attributes = { job: 'Foo::bar' }; + + service.increment('dfx_cron_started', attributes); + service.increment('dfx_cron_started', attributes); + + expect(createCounter).toHaveBeenCalledTimes(1); + expect(createCounter).toHaveBeenCalledWith('dfx_cron_started'); + expect(counterAdd).toHaveBeenCalledTimes(2); + expect(counterAdd).toHaveBeenNthCalledWith(1, 1, attributes); + expect(counterAdd).toHaveBeenNthCalledWith(2, 1, attributes); + }); + + it('creates one histogram per name and records value with unit and attributes', () => { + const service = new MetricService(); + const attributes = { job: 'Foo::bar' }; + + service.record('dfx_cron_run_seconds', 1.5, 's', attributes); + service.record('dfx_cron_run_seconds', 2.25, 's', attributes); + + expect(createHistogram).toHaveBeenCalledTimes(1); + expect(createHistogram).toHaveBeenCalledWith('dfx_cron_run_seconds', { unit: 's' }); + expect(histogramRecord).toHaveBeenCalledTimes(2); + expect(histogramRecord).toHaveBeenNthCalledWith(1, 1.5, attributes); + expect(histogramRecord).toHaveBeenNthCalledWith(2, 2.25, attributes); + }); + + it('registers a gauge callback and rejects a second registration of the same name', () => { + const service = new MetricService(); + const observe = jest.fn(); + + service.registerGauge('dfx_cron_seconds_since_last_run', 's', observe); + + expect(createObservableGauge).toHaveBeenCalledTimes(1); + expect(createObservableGauge).toHaveBeenCalledWith('dfx_cron_seconds_since_last_run', { unit: 's' }); + expect(gaugeAddCallback).toHaveBeenCalledTimes(1); + expect(gaugeAddCallback).toHaveBeenCalledWith(observe); + + expect(() => service.registerGauge('dfx_cron_seconds_since_last_run', 's', observe)).toThrow( + "Gauge 'dfx_cron_seconds_since_last_run' is already registered", + ); + expect(createObservableGauge).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/shared/services/dfx-cron.service.ts b/src/shared/services/dfx-cron.service.ts index af3f63c068..a35c28636d 100644 --- a/src/shared/services/dfx-cron.service.ts +++ b/src/shared/services/dfx-cron.service.ts @@ -9,6 +9,7 @@ import { LockClass } from 'src/shared/utils/lock'; import { Util } from 'src/shared/utils/util'; import { CustomCronExpression } from '../utils/custom-cron-expression'; import { DfxLogger } from './dfx-logger'; +import { MetricService } from './metric.service'; interface CronJobData { instance: object; @@ -20,14 +21,18 @@ interface CronJobData { @Injectable() export class DfxCronService implements OnModuleInit { private readonly logger = new DfxLogger(DfxCronService); + // Seeded at registration with process start so a freshly started process does + // not report an implausibly high age for jobs that have not finished a run yet. + private readonly lastRunAt = new Map(); constructor( private readonly discovery: DiscoveryService, private readonly metadataScanner: MetadataScanner, private readonly schedulerRegisty: SchedulerRegistry, + private readonly metricService: MetricService, ) {} - onModuleInit() { + onModuleInit(): void { this.discovery .getProviders() .filter((wrapper) => wrapper.isDependencyTreeStatic()) @@ -48,33 +53,53 @@ export class DfxCronService implements OnModuleInit { .filter((data) => data.params) .forEach((data) => this.addCronJob(data)); }); + + this.metricService.registerGauge('dfx_cron_seconds_since_last_run', 's', (result) => { + for (const [job, lastRunAt] of this.lastRunAt.entries()) { + result.observe(Util.secondsDiff(lastRunAt), { job }); + } + }); } - private addCronJob(data: CronJobData) { + private addCronJob(data: CronJobData): void { const lock = LockClass.create(data.params.timeout ?? Infinity); const context = { target: data.instance.constructor.name, method: data.methodName }; const cronJob = new CronJob(data.params.expression, () => lock(this.wrapFunction(data), context)); const cronJobName = `${context.target}::${context.method}`; + this.lastRunAt.set(cronJobName, new Date()); this.schedulerRegisty.addCronJob(cronJobName, cronJob); cronJob.start(); } private wrapFunction(data: CronJobData) { const context = { target: data.instance.constructor.name, method: data.methodName }; + const job = `${context.target}::${context.method}`; return async (...args: any) => { if (data.params.process && DisabledProcess(data.params.process)) { this.logger.verbose( `Skipping ${context.target}::${context.method} - process ${data.params.process} is disabled`, ); + this.metricService.increment('dfx_cron_skipped', { job, reason: 'disabled' }); return; } if (data.params.useDelay ?? true) await this.cronJobDelay(data.params.expression); - await data.methodRef.apply(data.instance, args); + this.metricService.increment('dfx_cron_started', { job }); + const startedAt = new Date(); + try { + await data.methodRef.apply(data.instance, args); + this.metricService.increment('dfx_cron_finished', { job, outcome: 'success' }); + } catch (e) { + this.metricService.increment('dfx_cron_finished', { job, outcome: 'failed' }); + throw e; + } finally { + this.metricService.record('dfx_cron_run_seconds', Util.secondsDiff(startedAt), 's', { job }); + this.lastRunAt.set(job, new Date()); + } }; } diff --git a/src/shared/services/metric.service.ts b/src/shared/services/metric.service.ts new file mode 100644 index 0000000000..adf89570c8 --- /dev/null +++ b/src/shared/services/metric.service.ts @@ -0,0 +1,42 @@ +// Thin facade over the OpenTelemetry Metrics API. +// Instrument names use underscores because the Prometheus exporter takes them +// as-is (and appends `_total` to counters). + +import { Injectable } from '@nestjs/common'; +import { Attributes, Counter, Histogram, metrics, ObservableResult } from '@opentelemetry/api'; + +@Injectable() +export class MetricService { + // OTel returns a no-op meter when no SDK is registered, so this service does + // not check whether metrics are enabled and has no fallback path. + private readonly meter = metrics.getMeter('dfx-api'); + private readonly counters = new Map(); + private readonly histograms = new Map(); + private readonly gaugeNames = new Set(); + + increment(name: string, attributes: Attributes): void { + let counter = this.counters.get(name); + if (!counter) { + counter = this.meter.createCounter(name); + this.counters.set(name, counter); + } + counter.add(1, attributes); + } + + record(name: string, value: number, unit: string, attributes: Attributes): void { + let histogram = this.histograms.get(name); + if (!histogram) { + histogram = this.meter.createHistogram(name, { unit }); + this.histograms.set(name, histogram); + } + histogram.record(value, attributes); + } + + registerGauge(name: string, unit: string, observe: (result: ObservableResult) => void | Promise): void { + if (this.gaugeNames.has(name)) { + throw new Error(`Gauge '${name}' is already registered`); + } + this.gaugeNames.add(name); + this.meter.createObservableGauge(name, { unit }).addCallback(observe); + } +} diff --git a/src/shared/services/process.service.ts b/src/shared/services/process.service.ts index 2c17e4fefd..a0401a150a 100644 --- a/src/shared/services/process.service.ts +++ b/src/shared/services/process.service.ts @@ -116,6 +116,10 @@ export enum Process { LEDGER_MARK_TO_MARKET = 'LedgerMarkToMarket', LEDGER_CUTOVER = 'LedgerCutover', LEDGER_COA_BOOTSTRAP = 'LedgerCoaBootstrap', + // one kill-switch per job group so a single group can be paused without taking others down; + // JOB_METRICS toggles the observation cron + JOB_ACCOUNT_MERGE = 'JobAccountMerge', + JOB_METRICS = 'JobMetrics', } const safetyProcesses: Process[] = [ diff --git a/src/shared/shared.module.ts b/src/shared/shared.module.ts index 371f467022..297025ffd0 100644 --- a/src/shared/shared.module.ts +++ b/src/shared/shared.module.ts @@ -37,6 +37,7 @@ import { SettingService } from './models/setting/setting.service'; import { RepositoryFactory } from './repositories/repository.factory'; import { DfxCronService } from './services/dfx-cron.service'; import { HttpService } from './services/http.service'; +import { MetricService } from './services/metric.service'; import { PaymentInfoService } from './services/payment-info.service'; import { ProcessService } from './services/process.service'; @@ -73,6 +74,7 @@ import { ProcessService } from './services/process.service'; IpLogService, ProcessService, DfxCronService, + MetricService, ], exports: [ RepositoryFactory, @@ -89,6 +91,7 @@ import { ProcessService } from './services/process.service'; PaymentInfoService, IpLogService, ProcessService, + MetricService, ], }) export class SharedModule {} diff --git a/src/subdomains/generic/user/models/account-merge/__tests__/account-merge-job.handler.spec.ts b/src/subdomains/generic/user/models/account-merge/__tests__/account-merge-job.handler.spec.ts new file mode 100644 index 0000000000..63ab52dea7 --- /dev/null +++ b/src/subdomains/generic/user/models/account-merge/__tests__/account-merge-job.handler.spec.ts @@ -0,0 +1,63 @@ +import { ConflictException } from '@nestjs/common'; +import { JobDeadLetterException } from 'src/subdomains/supporting/job/exceptions/job-dead-letter.exception'; +import { JobService } from 'src/subdomains/supporting/job/services/job.service'; +import { UserData } from '../../user-data/user-data.entity'; +import { AccountMerge } from '../account-merge.entity'; +import { AccountMergeJobHandler } from '../account-merge-job.handler'; +import { AccountMergeService } from '../account-merge.service'; + +describe('AccountMergeJobHandler', () => { + let handler: AccountMergeJobHandler; + let accountMergeService: jest.Mocked>; + let jobService: jest.Mocked>; + + beforeEach(() => { + accountMergeService = { executeMerge: jest.fn() }; + jobService = { registerHandler: jest.fn() }; + + handler = new AccountMergeJobHandler( + jobService as unknown as JobService, + accountMergeService as unknown as AccountMergeService, + ); + }); + + describe('execute', () => { + it('calls executeMerge with the code and maps master id and kycHash', async () => { + const request = Object.assign(new AccountMerge(), { + master: Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }), + }); + accountMergeService.executeMerge.mockResolvedValue(request); + + const result = await handler.execute({ code: 'merge-code' }); + + expect(accountMergeService.executeMerge).toHaveBeenCalledWith('merge-code'); + expect(result).toEqual({ masterUserDataId: 1, kycHash: 'hash-1' }); + }); + + it('does not include an accessToken field on the result', async () => { + const request = Object.assign(new AccountMerge(), { + master: Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }), + }); + accountMergeService.executeMerge.mockResolvedValue(request); + + const result = await handler.execute({ code: 'merge-code' }); + + expect(result).not.toHaveProperty('accessToken'); + expect(Object.keys(result)).toEqual(['masterUserDataId', 'kycHash']); + }); + + it('maps ConflictException to JobDeadLetterException', async () => { + accountMergeService.executeMerge.mockRejectedValue(new ConflictException('Merge request is already completed')); + + await expect(handler.execute({ code: 'merge-code' })).rejects.toThrow(JobDeadLetterException); + await expect(handler.execute({ code: 'merge-code' })).rejects.toThrow('Merge request is already completed'); + }); + + it('rethrows a generic Error unchanged', async () => { + const error = new Error('transient failure'); + accountMergeService.executeMerge.mockRejectedValue(error); + + await expect(handler.execute({ code: 'merge-code' })).rejects.toBe(error); + }); + }); +}); diff --git a/src/subdomains/generic/user/models/account-merge/account-merge-job.handler.ts b/src/subdomains/generic/user/models/account-merge/account-merge-job.handler.ts new file mode 100644 index 0000000000..b63a46d79e --- /dev/null +++ b/src/subdomains/generic/user/models/account-merge/account-merge-job.handler.ts @@ -0,0 +1,49 @@ +import { BadRequestException, ConflictException, Injectable, NotFoundException, OnModuleInit } from '@nestjs/common'; +import { JobGroup } from 'src/subdomains/supporting/job/enums'; +import { JobDeadLetterException } from 'src/subdomains/supporting/job/exceptions/job-dead-letter.exception'; +import { JobHandler } from 'src/subdomains/supporting/job/interfaces/job-handler.interface'; +import { JobService } from 'src/subdomains/supporting/job/services/job.service'; +import { AccountMergeService } from './account-merge.service'; + +export interface AccountMergeJobInput { + code: string; +} + +export interface AccountMergeJobOutput { + masterUserDataId: number; + kycHash: string; +} + +@Injectable() +export class AccountMergeJobHandler implements JobHandler, OnModuleInit { + readonly group = JobGroup.ACCOUNT_MERGE; + + constructor( + private readonly jobService: JobService, + private readonly accountMergeService: AccountMergeService, + ) {} + + // Registered from the outside so the job module stays free of domain dependencies. + onModuleInit(): void { + this.jobService.registerHandler(this); + } + + async execute(input: AccountMergeJobInput): Promise { + try { + const request = await this.accountMergeService.executeMerge(input.code); + + // No access token here: it is bound to the request context (address, IP, 2FA marker) and + // would otherwise sit in the job table as a stored credential. Minted fresh in the HTTP + // layer when the caller picks up the result. + return { masterUserDataId: request.master.id, kycHash: request.master.kycHash }; + } catch (e) { + // These preconditions can never turn true by retrying, so the job must not consume attempt + // budget on them. + if (e instanceof NotFoundException || e instanceof BadRequestException || e instanceof ConflictException) { + throw new JobDeadLetterException(e.message); + } + + throw e; + } + } +} diff --git a/src/subdomains/generic/user/models/account-merge/account-merge.service.ts b/src/subdomains/generic/user/models/account-merge/account-merge.service.ts index 564dee8db4..e04c8c5e99 100644 --- a/src/subdomains/generic/user/models/account-merge/account-merge.service.ts +++ b/src/subdomains/generic/user/models/account-merge/account-merge.service.ts @@ -115,6 +115,31 @@ export class AccountMergeService { return true; } + /** + * Validates a merge request without changing anything. Synchronous front door before the merge + * is enqueued as a job: the 404/400 below are a contract with existing clients and must come + * back synchronously, not as a job outcome. executeMerge() re-checks found/expired/completed + * on its own right before it runs, since time passes in between. + */ + async validateForExecution(code: string): Promise { + const request = await this.accountMergeRepo.findOne({ + where: { code }, + relations: { master: true, slave: true }, + }); + if (!request) throw new NotFoundException('Account merge information not found'); + + if (request.isExpired) throw new BadRequestException('Merge request is expired'); + // A running or already completed merge is not an error here: the caller decides based on the + // associated job (or, for pre-job legacy rows with no job, raises its own conflict). + return request; + } + + // Exposes the master/slave ordering to callers outside this service instead of having them + // rebuild the sort themselves. + getMaster(request: AccountMerge): UserData { + return AccountMergeService.masterFirst([request.master, request.slave])[0]; + } + async executeMerge(code: string): Promise { const request = await this.accountMergeRepo.findOne({ where: { code }, relations: { master: true, slave: true } }); if (!request) throw new NotFoundException('Account merge information not found'); diff --git a/src/subdomains/generic/user/models/auth/__tests__/auth.controller.spec.ts b/src/subdomains/generic/user/models/auth/__tests__/auth.controller.spec.ts new file mode 100644 index 0000000000..904ee1f34b --- /dev/null +++ b/src/subdomains/generic/user/models/auth/__tests__/auth.controller.spec.ts @@ -0,0 +1,279 @@ +import { createMock } from '@golevelup/ts-jest'; +import { ConflictException, HttpStatus } from '@nestjs/common'; +import { Test, TestingModule } from '@nestjs/testing'; +import { ThrottlerModule } from '@nestjs/throttler'; +import { Response } from 'express'; +import { JwtPayload } from 'src/shared/auth/jwt-payload.interface'; +import { IpLogService } from 'src/shared/models/ip-log/ip-log.service'; +import { TestSharedModule } from 'src/shared/utils/test.shared.module'; +import { TfaService } from 'src/subdomains/generic/kyc/services/tfa.service'; +import { Job } from 'src/subdomains/supporting/job/entities/job.entity'; +import { JobGroup, JobStatus } from 'src/subdomains/supporting/job/enums'; +import { JOB_GROUP_DEFAULTS } from 'src/subdomains/supporting/job/job-group.config'; +import { JobDispatcherService } from 'src/subdomains/supporting/job/services/job-dispatcher.service'; +import { JobService } from 'src/subdomains/supporting/job/services/job.service'; +import { AccountMerge } from '../../account-merge/account-merge.entity'; +import { AccountMergeService } from '../../account-merge/account-merge.service'; +import { UserData } from '../../user-data/user-data.entity'; +import { UserDataService } from '../../user-data/user-data.service'; +import { UserRepository } from '../../user/user.repository'; +import { AuthAlbyService } from '../auth-alby.service'; +import { AuthController } from '../auth.controller'; +import { AuthService } from '../auth.service'; + +// AuthController pulls TfaService in purely as a DI token — no test here calls check/setup/verify. +// Importing it drags in the KYC entity chain, which is circular at import time and breaks when an +// isolated spec is the first to resolve it (step-log.entity.ts: `class StepLog extends KycLog` +// runs while KycLog is still undefined). Replacing the module keeps that chain out of this suite. +jest.mock('src/subdomains/generic/kyc/services/tfa.service', () => ({ + TfaLevel: { BASIC: 'Basic', STRICT: 'Strict' }, + TfaService: class TfaService {}, +})); + +describe('AuthController', () => { + let controller: AuthController; + + let authService: AuthService; + let albyService: AuthAlbyService; + let mergeService: AccountMergeService; + let userRepo: UserRepository; + let userDataService: UserDataService; + let tfaService: TfaService; + let jobService: JobService; + let jobDispatcher: JobDispatcherService; + let ipLogService: IpLogService; + + beforeEach(async () => { + authService = createMock(); + albyService = createMock(); + mergeService = createMock(); + userRepo = createMock(); + userDataService = createMock(); + tfaService = createMock(); + jobService = createMock(); + jobDispatcher = createMock(); + // AuthController's routes carry IpCountryGuard, which injects IpLogService — without it Nest + // cannot resolve the guard and the whole module fails to compile. + ipLogService = createMock(); + + const module: TestingModule = await Test.createTestingModule({ + // AuthController's routes carry RateLimitGuard, which needs the throttler options. + imports: [TestSharedModule, ThrottlerModule.forRoot()], + providers: [ + AuthController, + { provide: AuthService, useValue: authService }, + { provide: AuthAlbyService, useValue: albyService }, + { provide: AccountMergeService, useValue: mergeService }, + { provide: UserRepository, useValue: userRepo }, + { provide: UserDataService, useValue: userDataService }, + { provide: TfaService, useValue: tfaService }, + { provide: JobService, useValue: jobService }, + { provide: JobDispatcherService, useValue: jobDispatcher }, + { provide: IpLogService, useValue: ipLogService }, + ], + }).compile(); + + controller = module.get(AuthController); + }); + + it('should be defined', () => { + expect(controller).toBeDefined(); + }); + + describe('mail/confirm', () => { + const code = 'merge-code'; + const ip = '1.2.3.4'; + const traceparent = '00-trace-01'; + const jwt = { + account: 1, + address: '0xabc', + ip: '1.2.3.4', + tfaRequired: false, + } as JwtPayload; + + function createRes(): Response { + return { status: jest.fn() } as unknown as Response; + } + + it('returns kycHash and accessToken when the job completes immediately', async () => { + const master = Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }); + const request = Object.assign(new AccountMerge(), { id: 10, master, code }); + const completedJob = Object.assign(new Job(), { + uid: 'Jmerge1', + group: JobGroup.ACCOUNT_MERGE, + status: JobStatus.COMPLETE, + created: new Date(), + output: JSON.stringify({ masterUserDataId: 1, kycHash: 'hash-1' }), + }); + const res = createRes(); + + jest.spyOn(mergeService, 'validateForExecution').mockResolvedValue(request); + jest.spyOn(mergeService, 'getMaster').mockReturnValue(master); + // createMock returns a truthy proxy for un-mocked methods, so the lookup has to be + // stubbed explicitly — otherwise the controller takes the 'job already exists' branch. + jest.spyOn(jobService, 'findByIdempotencyKey').mockResolvedValue(undefined); + jest.spyOn(jobService, 'enqueue').mockResolvedValue(completedJob); + jest.spyOn(userRepo, 'findOne').mockResolvedValue(null); + jest.spyOn(authService, 'generateAccountToken').mockReturnValue('token-1'); + + const result = await controller.executeMerge(jwt, code, ip, traceparent, res); + + expect(result).toEqual({ kycHash: 'hash-1', accessToken: 'token-1' }); + expect(res.status).toHaveBeenCalledWith(HttpStatus.OK); + expect(jobDispatcher.kick).toHaveBeenCalledWith(JobGroup.ACCOUNT_MERGE); + }); + + it('returns a JobDto with ACCEPTED when the job is still pending after the poll window', async () => { + const master = Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }); + const request = Object.assign(new AccountMerge(), { id: 11, master, code }); + const pendingJob = Object.assign(new Job(), { + uid: 'Jmerge-pending', + group: JobGroup.ACCOUNT_MERGE, + status: JobStatus.PENDING, + created: new Date(), + }); + const res = createRes(); + + jest.spyOn(mergeService, 'validateForExecution').mockResolvedValue(request); + jest.spyOn(mergeService, 'getMaster').mockReturnValue(master); + // createMock returns a truthy proxy for un-mocked methods, so the lookup has to be + // stubbed explicitly — otherwise the controller takes the 'job already exists' branch. + jest.spyOn(jobService, 'findByIdempotencyKey').mockResolvedValue(undefined); + jest.spyOn(jobService, 'enqueue').mockResolvedValue(pendingJob); + jest.spyOn(jobService, 'getByUid').mockResolvedValue(pendingJob); + jest.spyOn(jobService, 'getConfig').mockResolvedValue(JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE]); + + const result = await controller.executeMerge(jwt, code, ip, traceparent, res); + + expect(result).toEqual(expect.objectContaining({ uid: 'Jmerge-pending' })); + expect(res.status).toHaveBeenCalledWith(HttpStatus.ACCEPTED); + }, 15000); + + it('propagates ConflictException from validateForExecution and does not enqueue', async () => { + const res = createRes(); + jest + .spyOn(mergeService, 'validateForExecution') + .mockRejectedValue(new ConflictException('Merge request is already completed')); + + await expect(controller.executeMerge(jwt, code, ip, traceparent, res)).rejects.toThrow(ConflictException); + expect(jobService.enqueue).not.toHaveBeenCalled(); + }); + + it('returns kycHash and accessToken on a second call when a COMPLETE job already exists', async () => { + const master = Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }); + const request = Object.assign(new AccountMerge(), { id: 12, master, code, isCompleted: true }); + const completedJob = Object.assign(new Job(), { + uid: 'Jmerge-existing', + group: JobGroup.ACCOUNT_MERGE, + status: JobStatus.COMPLETE, + created: new Date(), + finishedAt: new Date(), + output: JSON.stringify({ masterUserDataId: 1, kycHash: 'hash-1' }), + }); + const res = createRes(); + + jest.spyOn(mergeService, 'validateForExecution').mockResolvedValue(request); + jest.spyOn(mergeService, 'getMaster').mockReturnValue(master); + jest.spyOn(jobService, 'findByIdempotencyKey').mockResolvedValue(completedJob); + jest.spyOn(userRepo, 'findOne').mockResolvedValue(null); + jest.spyOn(authService, 'generateAccountToken').mockReturnValue('token-2'); + + const result = await controller.executeMerge(jwt, code, ip, traceparent, res); + + expect(result).toEqual({ kycHash: 'hash-1', accessToken: 'token-2' }); + expect(res.status).toHaveBeenCalledWith(HttpStatus.OK); + expect(jobService.enqueue).not.toHaveBeenCalled(); + expect(jobDispatcher.kick).not.toHaveBeenCalled(); + }); + + it('returns kycHash and accessToken when a COMPLETE job finished within the result window', async () => { + const master = Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }); + const request = Object.assign(new AccountMerge(), { id: 20, master, code, isCompleted: true }); + const completedJob = Object.assign(new Job(), { + uid: 'Jmerge-within-window', + group: JobGroup.ACCOUNT_MERGE, + status: JobStatus.COMPLETE, + created: new Date(), + finishedAt: new Date(Date.now() - 2 * 60 * 1000), + output: JSON.stringify({ masterUserDataId: 1, kycHash: 'hash-1' }), + }); + const res = createRes(); + + jest.spyOn(mergeService, 'validateForExecution').mockResolvedValue(request); + jest.spyOn(mergeService, 'getMaster').mockReturnValue(master); + jest.spyOn(jobService, 'findByIdempotencyKey').mockResolvedValue(completedJob); + jest.spyOn(userRepo, 'findOne').mockResolvedValue(null); + jest.spyOn(authService, 'generateAccountToken').mockReturnValue('token-within'); + + const result = await controller.executeMerge(jwt, code, ip, traceparent, res); + + expect(result).toEqual({ kycHash: 'hash-1', accessToken: 'token-within' }); + expect(res.status).toHaveBeenCalledWith(HttpStatus.OK); + expect(jobService.enqueue).not.toHaveBeenCalled(); + }); + + it('throws ConflictException when a COMPLETE job finished outside the result window', async () => { + const master = Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }); + const request = Object.assign(new AccountMerge(), { id: 21, master, code, isCompleted: true }); + const completedJob = Object.assign(new Job(), { + uid: 'Jmerge-outside-window', + group: JobGroup.ACCOUNT_MERGE, + status: JobStatus.COMPLETE, + created: new Date(), + finishedAt: new Date(Date.now() - 20 * 60 * 1000), + output: JSON.stringify({ masterUserDataId: 1, kycHash: 'hash-1' }), + }); + const res = createRes(); + + jest.spyOn(mergeService, 'validateForExecution').mockResolvedValue(request); + jest.spyOn(mergeService, 'getMaster').mockReturnValue(master); + jest.spyOn(jobService, 'findByIdempotencyKey').mockResolvedValue(completedJob); + + await expect(controller.executeMerge(jwt, code, ip, traceparent, res)).rejects.toThrow( + new ConflictException('Merge request is already completed'), + ); + expect(jobService.enqueue).not.toHaveBeenCalled(); + expect(authService.generateAccountToken).not.toHaveBeenCalled(); + expect(authService.generateUserToken).not.toHaveBeenCalled(); + }); + + it('throws ConflictException when merge is completed and no job exists', async () => { + const master = Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }); + const request = Object.assign(new AccountMerge(), { id: 13, master, code, isCompleted: true }); + const res = createRes(); + + jest.spyOn(mergeService, 'validateForExecution').mockResolvedValue(request); + jest.spyOn(mergeService, 'getMaster').mockReturnValue(master); + jest.spyOn(jobService, 'findByIdempotencyKey').mockResolvedValue(undefined); + + await expect(controller.executeMerge(jwt, code, ip, traceparent, res)).rejects.toThrow( + new ConflictException('Merge request is already completed'), + ); + expect(jobService.enqueue).not.toHaveBeenCalled(); + }); + + it('returns ACCEPTED JobDto when an existing job is FAILED', async () => { + const master = Object.assign(new UserData(), { id: 1, kycHash: 'hash-1' }); + const request = Object.assign(new AccountMerge(), { id: 14, master, code, isCompleted: false }); + const failedJob = Object.assign(new Job(), { + uid: 'Jmerge-failed', + group: JobGroup.ACCOUNT_MERGE, + status: JobStatus.FAILED, + created: new Date(), + error: 'internal connection refused', + }); + const res = createRes(); + + jest.spyOn(mergeService, 'validateForExecution').mockResolvedValue(request); + jest.spyOn(mergeService, 'getMaster').mockReturnValue(master); + jest.spyOn(jobService, 'findByIdempotencyKey').mockResolvedValue(failedJob); + jest.spyOn(jobService, 'getConfig').mockResolvedValue(JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE]); + + const result = await controller.executeMerge(jwt, code, ip, traceparent, res); + + expect(result).toEqual(expect.objectContaining({ status: JobStatus.FAILED, uid: 'Jmerge-failed' })); + expect(res.status).toHaveBeenCalledWith(HttpStatus.ACCEPTED); + expect(jobService.enqueue).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/src/subdomains/generic/user/models/auth/auth.controller.ts b/src/subdomains/generic/user/models/auth/auth.controller.ts index dad1b9c19b..8357452200 100644 --- a/src/subdomains/generic/user/models/auth/auth.controller.ts +++ b/src/subdomains/generic/user/models/auth/auth.controller.ts @@ -1,6 +1,26 @@ -import { Body, Controller, Get, Param, Post, Query, Req, Res, UseGuards } from '@nestjs/common'; +import { + Body, + ConflictException, + Controller, + Get, + Headers, + HttpStatus, + Param, + Post, + Query, + Req, + Res, + UseGuards, +} from '@nestjs/common'; import { AuthGuard } from '@nestjs/passport'; -import { ApiBearerAuth, ApiCreatedResponse, ApiExcludeEndpoint, ApiOkResponse, ApiTags } from '@nestjs/swagger'; +import { + ApiAcceptedResponse, + ApiBearerAuth, + ApiCreatedResponse, + ApiExcludeEndpoint, + ApiOkResponse, + ApiTags, +} from '@nestjs/swagger'; import { Throttle } from '@nestjs/throttler'; import { Request, Response } from 'express'; import { AllowTfaPending } from 'src/shared/auth/allow-tfa-pending.decorator'; @@ -13,10 +33,18 @@ import { RateLimitGuard } from 'src/shared/auth/rate-limit.guard'; import { RoleGuard } from 'src/shared/auth/role.guard'; import { UserActiveGuard } from 'src/shared/auth/user-active.guard'; import { UserRole } from 'src/shared/auth/user-role.enum'; +import { Util } from 'src/shared/utils/util'; import { Start2faDto } from 'src/subdomains/generic/kyc/dto/input/start-2fa.dto'; import { Verify2faDto } from 'src/subdomains/generic/kyc/dto/input/verify-2fa.dto'; import { Setup2faDto } from 'src/subdomains/generic/kyc/dto/output/setup-2fa.dto'; import { TfaService } from 'src/subdomains/generic/kyc/services/tfa.service'; +import { JobDtoMapper } from 'src/subdomains/supporting/job/dto/job-dto.mapper'; +import { JobDto } from 'src/subdomains/supporting/job/dto/job.dto'; +import { Job } from 'src/subdomains/supporting/job/entities/job.entity'; +import { JobGroup, JobStatus } from 'src/subdomains/supporting/job/enums'; +import { JobDispatcherService } from 'src/subdomains/supporting/job/services/job-dispatcher.service'; +import { JobService } from 'src/subdomains/supporting/job/services/job.service'; +import { AccountMergeJobOutput } from '../account-merge/account-merge-job.handler'; import { AccountMergeService } from '../account-merge/account-merge.service'; import { UserDataService } from '../user-data/user-data.service'; import { UserData } from '../user-data/user-data.entity'; @@ -33,6 +61,13 @@ import { RedirectResponseDto } from './dto/redirect-response.dto'; import { SignMessageDto } from './dto/sign-message.dto'; import { VerifySignMessageDto } from './dto/verify-sign-message.dto'; +// A confirmation link is a one-time ticket, not a standing credential: once the merge is complete, +// its result (and the fresh access token that comes with it) may only be collected within this +// window. A polling client re-hits this endpoint every few seconds and gives up after at most ten +// minutes; fifteen minutes covers that fully. Past the window the link behaves exactly as it did +// before this asynchronous path existed — already completed, nothing more to hand out. +const MERGE_RESULT_WINDOW_MINUTES = 15; + @ApiTags('Auth') @Controller('auth') export class AuthController { @@ -43,6 +78,8 @@ export class AuthController { private readonly userRepo: UserRepository, private readonly userDataService: UserDataService, private readonly tfaService: TfaService, + private readonly jobService: JobService, + private readonly jobDispatcher: JobDispatcherService, ) {} @Post() @@ -129,21 +166,84 @@ export class AuthController { @UseGuards(OptionalJwtAuthGuard) @ApiExcludeEndpoint() @ApiOkResponse({ type: MergeResponseDto }) + @ApiAcceptedResponse({ type: JobDto }) async executeMerge( @GetJwt() jwt: JwtPayload | undefined, @Query('code') code: string, @RealIP() ip: string, - ): Promise { - const { master } = await this.mergeService.executeMerge(code); + @Headers('traceparent') traceparent: string | undefined, + @Res({ passthrough: true }) res: Response, + ): Promise { + // 404/400 are a contract with existing clients and must come back synchronously, before any + // job is enqueued. + const request = await this.mergeService.validateForExecution(code); + const master = this.mergeService.getMaster(request); + + // The request id is the idempotency key: the same confirmation link never enqueues twice, and + // a polling client always hits the same job. + const idempotencyKey = `merge:${request.id}`; + const existingJob = await this.jobService.findByIdempotencyKey(JobGroup.ACCOUNT_MERGE, idempotencyKey); + if (existingJob) { + if ( + existingJob.status === JobStatus.COMPLETE && + Util.minutesDiff(existingJob.finishedAt) > MERGE_RESULT_WINDOW_MINUTES + ) { + throw new ConflictException('Merge request is already completed'); + } + + return this.waitForJobResult(existingJob, master, jwt, ip, res); + } + + // Completed before this job mechanism existed: no job row was ever written, so there is no + // result to return — a genuine conflict for that legacy case. + if (request.isCompleted) { + throw new ConflictException('Merge request is already completed'); + } + + const job = await this.jobService.enqueue( + JobGroup.ACCOUNT_MERGE, + idempotencyKey, + { code }, + { userData: master, traceparent }, + ); + this.jobDispatcher.kick(JobGroup.ACCOUNT_MERGE); - const accessToken = jwt - ? await this.createAccessTokenAfterMerge(master, jwt.address, ip, jwt.tfaRequired) - : undefined; + return this.waitForJobResult(job, master, jwt, ip, res); + } + + private async waitForJobResult( + job: Job, + master: UserData, + jwt: JwtPayload | undefined, + ip: string, + res: Response, + ): Promise { + // Bounded well under the second that is the response-time target: most merges still get + // today's synchronous-looking answer, only the long tail turns into a ticket. + const pollDeadline = Date.now() + 900; + let current = job; + while (!current.isFinished && Date.now() < pollDeadline) { + await Util.delay(100); + const found = await this.jobService.getByUid(job.uid); + if (!found) throw new Error(`Job ${job.uid} not found during poll`); + current = found; + } + + if (current.status === JobStatus.COMPLETE) { + const output = current.outputData as AccountMergeJobOutput; + const accessToken = jwt + ? await this.createAccessTokenAfterMerge(master, jwt.address, ip, jwt.tfaRequired) + : undefined; + + res.status(HttpStatus.OK); + return { kycHash: output.kycHash, accessToken }; + } - return { - kycHash: master.kycHash, - accessToken, - }; + // Still running, or ended FAILED/DEAD_LETTER: the status code marks that the job was + // accepted, the DTO's own status field carries the outcome. + const config = await this.jobService.getConfig(JobGroup.ACCOUNT_MERGE); + res.status(HttpStatus.ACCEPTED); + return JobDtoMapper.mapJob(current, config); } private async createAccessTokenAfterMerge( diff --git a/src/subdomains/generic/user/user.module.ts b/src/subdomains/generic/user/user.module.ts index 5fce80a978..c6dea99c9c 100644 --- a/src/subdomains/generic/user/user.module.ts +++ b/src/subdomains/generic/user/user.module.ts @@ -16,7 +16,9 @@ import { PaymentModule } from 'src/subdomains/supporting/payment/payment.module' import { TransactionModule } from 'src/subdomains/supporting/payment/transaction.module'; import { SupportIssueModule } from 'src/subdomains/supporting/support-issue/support-issue.module'; import { KycModule } from '../kyc/kyc.module'; +import { JobModule } from 'src/subdomains/supporting/job/job.module'; import { AccountMerge } from './models/account-merge/account-merge.entity'; +import { AccountMergeJobHandler } from './models/account-merge/account-merge-job.handler'; import { AccountMergeRepository } from './models/account-merge/account-merge.repository'; import { AccountMergeService } from './models/account-merge/account-merge.service'; import { AuthAlbyService } from './models/auth/auth-alby.service'; @@ -84,6 +86,7 @@ import { WebhookService } from './services/webhook/webhook.service'; forwardRef(() => SupportIssueModule), forwardRef(() => TransactionModule), forwardRef(() => CustodyModule), + JobModule, ], controllers: [ UserV2Controller, @@ -120,6 +123,7 @@ import { WebhookService } from './services/webhook/webhook.service'; UserDataNotificationService, UserDataRelationService, AccountMergeService, + AccountMergeJobHandler, CustodyProviderService, CustodyProviderRepository, OrganizationService, diff --git a/src/subdomains/supporting/job/__tests__/job.controller.spec.ts b/src/subdomains/supporting/job/__tests__/job.controller.spec.ts new file mode 100644 index 0000000000..1abe86f713 --- /dev/null +++ b/src/subdomains/supporting/job/__tests__/job.controller.spec.ts @@ -0,0 +1,111 @@ +import { createMock } from '@golevelup/ts-jest'; +import { NotFoundException } from '@nestjs/common'; +import { Test, TestingModule } from '@nestjs/testing'; +import { JwtPayload } from 'src/shared/auth/jwt-payload.interface'; +import { UserRole } from 'src/shared/auth/user-role.enum'; +import { TestSharedModule } from 'src/shared/utils/test.shared.module'; +import { UserData } from 'src/subdomains/generic/user/models/user-data/user-data.entity'; +import { Job } from '../entities/job.entity'; +import { JobGroup, JobStatus } from '../enums'; +import { JobController } from '../job.controller'; +import { JOB_GROUP_DEFAULTS, JobGroupConfig } from '../job-group.config'; +import { JobService } from '../services/job.service'; + +describe('JobController', () => { + let controller: JobController; + let jobService: JobService; + + beforeEach(async () => { + jobService = createMock(); + + const module: TestingModule = await Test.createTestingModule({ + imports: [TestSharedModule], + controllers: [JobController], + providers: [{ provide: JobService, useValue: jobService }], + }).compile(); + + controller = module.get(JobController); + }); + + function createJob(partial: Partial & { userData?: UserData } = {}): Job { + const job = new Job(); + job.id = partial.id ?? 1; + job.uid = partial.uid ?? 'Jabc'; + job.group = partial.group ?? JobGroup.ACCOUNT_MERGE; + job.status = partial.status ?? JobStatus.COMPLETE; + job.idempotencyKey = partial.idempotencyKey ?? 'key-1'; + job.attempt = partial.attempt ?? 1; + job.maxAttempts = partial.maxAttempts ?? 3; + job.input = partial.input ?? JSON.stringify({ masterId: 1 }); + job.created = partial.created ?? new Date('2026-01-01T00:00:00.000Z'); + if (partial.output !== undefined) job.output = partial.output; + if (partial.startedAt !== undefined) job.startedAt = partial.startedAt; + if (partial.finishedAt !== undefined) job.finishedAt = partial.finishedAt; + if (partial.error !== undefined) job.error = partial.error; + if (partial.userData !== undefined) job.userData = partial.userData; + return job; + } + + function jwtForAccount(account: number): JwtPayload { + return { account, role: UserRole.USER, ip: '127.0.0.1' }; + } + + it('throws NotFoundException when the job belongs to a different account', async () => { + const owner = { id: 10 } as UserData; + const job = createJob({ uid: 'Jsecret', userData: owner }); + jest.spyOn(jobService, 'getByUid').mockResolvedValue(job); + + await expect(controller.getJob(jwtForAccount(99), 'Jsecret')).rejects.toThrow(NotFoundException); + await expect(controller.getJob(jwtForAccount(99), 'Jsecret')).rejects.toThrow('Job not found'); + expect(jobService.getConfig).not.toHaveBeenCalled(); + }); + + it('omits result when exposeResult is false and includes it when true and status is COMPLETE', async () => { + const job = createJob({ + uid: 'Jresult', + status: JobStatus.COMPLETE, + output: JSON.stringify({ masterId: 7 }), + finishedAt: new Date('2026-01-01T00:01:00.000Z'), + }); + jest.spyOn(jobService, 'getByUid').mockResolvedValue(job); + + const hiddenConfig: JobGroupConfig = { + ...JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE], + exposeResult: false, + }; + jest.spyOn(jobService, 'getConfig').mockResolvedValue(hiddenConfig); + + const hidden = await controller.getJob(undefined, 'Jresult'); + expect(hidden).toEqual( + expect.objectContaining({ + uid: 'Jresult', + group: JobGroup.ACCOUNT_MERGE, + status: JobStatus.COMPLETE, + expectedSeconds: hiddenConfig.maxWaitSeconds + hiddenConfig.maxRunSeconds, + }), + ); + // The serialized response is checked, because that is what the client sees: the class + // property `result` exists structurally (declared fields become own properties with value + // `undefined` when `new JobDto()` runs under target es2023), but carries no value here and + // is dropped by `JSON.stringify`. + expect(JSON.parse(JSON.stringify(hidden))).not.toHaveProperty('result'); + expect(hidden.result).toBeUndefined(); + + const exposedConfig: JobGroupConfig = { + ...JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE], + exposeResult: true, + }; + jest.spyOn(jobService, 'getConfig').mockResolvedValue(exposedConfig); + + const exposed = await controller.getJob(undefined, 'Jresult'); + expect(JSON.parse(JSON.stringify(exposed))).toHaveProperty('result'); + expect(exposed.result).toEqual({ masterId: 7 }); + expect(exposed).toEqual( + expect.objectContaining({ + uid: 'Jresult', + status: JobStatus.COMPLETE, + result: { masterId: 7 }, + }), + ); + }); +}); diff --git a/src/subdomains/supporting/job/dto/__tests__/job-dto.mapper.spec.ts b/src/subdomains/supporting/job/dto/__tests__/job-dto.mapper.spec.ts new file mode 100644 index 0000000000..91d455f068 --- /dev/null +++ b/src/subdomains/supporting/job/dto/__tests__/job-dto.mapper.spec.ts @@ -0,0 +1,51 @@ +import { Job } from '../../entities/job.entity'; +import { JobGroup, JobStatus } from '../../enums'; +import { JOB_GROUP_DEFAULTS } from '../../job-group.config'; +import { JobDtoMapper } from '../job-dto.mapper'; + +describe('JobDtoMapper', () => { + function createJob(partial: Partial = {}): Job { + const job = new Job(); + job.id = partial.id ?? 1; + job.uid = partial.uid ?? 'Jabc'; + job.group = partial.group ?? JobGroup.ACCOUNT_MERGE; + job.status = partial.status ?? JobStatus.PENDING; + job.idempotencyKey = partial.idempotencyKey ?? 'key-1'; + job.attempt = partial.attempt ?? 1; + job.maxAttempts = partial.maxAttempts ?? 3; + job.input = partial.input ?? JSON.stringify({ masterId: 1 }); + job.created = partial.created ?? new Date('2026-01-01T00:00:00.000Z'); + if (partial.error !== undefined) job.error = partial.error; + return job; + } + + const config = JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE]; + + it('replaces FAILED internal errors with a uid-referencing public message', () => { + const job = createJob({ + uid: 'Jfail1', + status: JobStatus.FAILED, + error: 'ECONNREFUSED: connection to database refused at 10.0.0.5:5432', + }); + + const dto = JobDtoMapper.mapJob(job, config); + + expect(dto.error).toBeDefined(); + expect(dto.error).not.toContain('ECONNREFUSED'); + expect(dto.error).not.toContain('10.0.0.5'); + expect(dto.error).toContain('Jfail1'); + }); + + it('exposes DEAD_LETTER domain errors unchanged', () => { + const domainError = 'Merge request is expired'; + const job = createJob({ + uid: 'Jdead1', + status: JobStatus.DEAD_LETTER, + error: domainError, + }); + + const dto = JobDtoMapper.mapJob(job, config); + + expect(dto.error).toBe(domainError); + }); +}); diff --git a/src/subdomains/supporting/job/dto/job-dto.mapper.ts b/src/subdomains/supporting/job/dto/job-dto.mapper.ts new file mode 100644 index 0000000000..25f1be6e6f --- /dev/null +++ b/src/subdomains/supporting/job/dto/job-dto.mapper.ts @@ -0,0 +1,40 @@ +import { Job } from '../entities/job.entity'; +import { JobStatus } from '../enums'; +import { JobGroupConfig } from '../job-group.config'; +import { JobDto } from './job.dto'; + +export class JobDtoMapper { + static mapJob(job: Job, config: JobGroupConfig): JobDto { + // Total time a client may reasonably wait for — queueing time plus run time, not run time alone. + const expectedSeconds = config.maxWaitSeconds + config.maxRunSeconds; + + // Fail-closed — groups whose result is sensitive only expose it through their own domain-specific endpoint. + const includeResult = config.exposeResult && job.status === JobStatus.COMPLETE; + + // Dead-letter messages are domain statements about the job (safe for the client). Failed + // messages are internal diagnostics (DB/provider details) and must not leave the system raw. + let error: string | undefined; + if (job.status === JobStatus.DEAD_LETTER && job.error != null) { + error = job.error; + } else if (job.status === JobStatus.FAILED) { + error = `Job ${job.uid} failed, contact support if this persists.`; + } + // Any other state deliberately reports no error at all. A job in RETRY carries the raw message + // of its last attempt, which is internal diagnostics — and the job is not finished, so there is + // nothing final to tell the client yet. + + const dto: JobDto = { + uid: job.uid, + group: job.group, + status: job.status, + created: job.created, + expectedSeconds, + ...(job.startedAt != null ? { started: job.startedAt } : {}), + ...(job.finishedAt != null ? { finished: job.finishedAt } : {}), + ...(includeResult ? { result: job.outputData } : {}), + ...(error != null ? { error } : {}), + }; + + return Object.assign(new JobDto(), dto); + } +} diff --git a/src/subdomains/supporting/job/dto/job-group-setting.dto.ts b/src/subdomains/supporting/job/dto/job-group-setting.dto.ts new file mode 100644 index 0000000000..3e58780f3a --- /dev/null +++ b/src/subdomains/supporting/job/dto/job-group-setting.dto.ts @@ -0,0 +1,27 @@ +import { IsBoolean, IsEnum, IsInt, IsNotEmpty, IsOptional, IsPositive } from 'class-validator'; +import { JobGroup } from '../enums'; + +export class JobGroupSettingDto { + @IsNotEmpty() + @IsEnum(JobGroup) + group: JobGroup; + + @IsOptional() + @IsInt() + @IsPositive() + maxWaitSeconds?: number; + + @IsOptional() + @IsInt() + @IsPositive() + maxRunSeconds?: number; + + @IsOptional() + @IsInt() + @IsPositive() + maxAttempts?: number; + + @IsOptional() + @IsBoolean() + exposeResult?: boolean; +} diff --git a/src/subdomains/supporting/job/dto/job.dto.ts b/src/subdomains/supporting/job/dto/job.dto.ts new file mode 100644 index 0000000000..9773c9e65e --- /dev/null +++ b/src/subdomains/supporting/job/dto/job.dto.ts @@ -0,0 +1,31 @@ +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { JobGroup, JobStatus } from '../enums'; + +export class JobDto { + @ApiProperty() + uid: string; + + @ApiProperty({ enum: JobGroup }) + group: JobGroup; + + @ApiProperty({ enum: JobStatus }) + status: JobStatus; + + @ApiProperty() + created: Date; + + @ApiPropertyOptional() + started?: Date; + + @ApiPropertyOptional() + finished?: Date; + + @ApiProperty() + expectedSeconds: number; + + @ApiPropertyOptional() + result?: unknown; + + @ApiPropertyOptional() + error?: string; +} diff --git a/src/subdomains/supporting/job/entities/job-attempt.entity.ts b/src/subdomains/supporting/job/entities/job-attempt.entity.ts new file mode 100644 index 0000000000..fc3591c19b --- /dev/null +++ b/src/subdomains/supporting/job/entities/job-attempt.entity.ts @@ -0,0 +1,36 @@ +import { IEntity } from 'src/shared/models/entity'; +import { Column, Entity, Index, ManyToOne } from 'typeorm'; +import { JobAttemptOutcome } from '../enums'; +import { Job } from './job.entity'; + +// Append-only record of a single job attempt. Rows here are never updated after the attempt they describe has +// finished, and never deleted. They are the proof of who claimed which attempt, when, and — if it did not +// succeed — the full error it failed with. The `job` row itself only ever carries the current snapshot; this +// table is where the history that snapshot would otherwise destroy on every retry survives. +// +// The unique index on (job, attempt) guarantees that the same attempt of the same job can never be recorded +// twice, no matter how many times claimNext, finish, abort or recoverStale run against it. +@Entity() +@Index((a: JobAttempt) => [a.job, a.attempt], { unique: true }) +export class JobAttempt extends IEntity { + @ManyToOne(() => Job, { nullable: false }) + job: Job; + + @Column({ type: 'int' }) + attempt: number; + + @Column({ length: 256 }) + claimedBy: string; + + @Column({ type: 'timestamp' }) + claimedAt: Date; + + @Column({ type: 'timestamp', nullable: true }) + finishedAt?: Date; + + @Column({ length: 256, nullable: true }) + outcome?: JobAttemptOutcome; + + @Column({ type: 'text', nullable: true }) + error?: string; +} diff --git a/src/subdomains/supporting/job/entities/job.entity.ts b/src/subdomains/supporting/job/entities/job.entity.ts new file mode 100644 index 0000000000..bd9bcf7c79 --- /dev/null +++ b/src/subdomains/supporting/job/entities/job.entity.ts @@ -0,0 +1,149 @@ +import { IEntity, UpdateResult } from 'src/shared/models/entity'; +import { Util } from 'src/shared/utils/util'; +import { UserData } from 'src/subdomains/generic/user/models/user-data/user-data.entity'; +import { Column, Entity, Index, ManyToOne } from 'typeorm'; +import { JobGroup, JobStatus } from '../enums'; + +@Entity() +@Index((j: Job) => [j.status, j.group, j.id]) +@Index((j: Job) => [j.group, j.idempotencyKey], { unique: true }) +export class Job extends IEntity { + @Column({ length: 256, unique: true }) + uid: string; + + @Column({ length: 256 }) + group: JobGroup; + + @Column({ length: 256 }) + status: JobStatus; + + @Column({ length: 256 }) + idempotencyKey: string; + + @Column({ type: 'text' }) + input: string; + + @Column({ type: 'text', nullable: true }) + output?: string; + + @Column({ type: 'int', default: 0 }) + attempt: number; + + @Column({ type: 'int' }) + maxAttempts: number; + + @Column({ type: 'timestamp', nullable: true }) + claimedAt?: Date; + + @Column({ length: 256, nullable: true }) + claimedBy?: string; + + @Column({ type: 'timestamp', nullable: true }) + startedAt?: Date; + + @Column({ type: 'timestamp', nullable: true }) + finishedAt?: Date; + + @Column({ type: 'timestamp', nullable: true }) + nextAttemptAt?: Date; + + @Column({ type: 'text', nullable: true }) + error?: string; + + @Column({ length: 256, nullable: true }) + traceparent?: string; + + @ManyToOne(() => UserData, { nullable: true }) + userData?: UserData; + + // --- JSON accessors (raw columns must not leak into business logic) --- // + + get inputData(): unknown { + return JSON.parse(this.input); + } + + set inputData(data: unknown) { + this.input = JSON.stringify(data); + } + + get outputData(): unknown { + return this.output ? JSON.parse(this.output) : undefined; + } + + set outputData(data: unknown) { + this.output = JSON.stringify(data); + } + + // --- state transitions --- // + + claim(owner: string): UpdateResult { + const now = new Date(); + const update: Partial = { + status: JobStatus.PROCESSING, + claimedAt: now, + claimedBy: owner, + startedAt: now, + attempt: this.attempt + 1, + }; + + Object.assign(this, update); + + return [this.id, update]; + } + + complete(output: unknown): UpdateResult { + this.outputData = output; + + const update: Partial = { + status: JobStatus.COMPLETE, + finishedAt: new Date(), + output: this.output, + }; + + Object.assign(this, update); + + return [this.id, update]; + } + + fail(error: string, nextAttemptAt: Date | undefined): UpdateResult { + const update: Partial = nextAttemptAt + ? { status: JobStatus.RETRY, error, nextAttemptAt } + : { status: JobStatus.FAILED, error, finishedAt: new Date() }; + + Object.assign(this, update); + + return [this.id, update]; + } + + deadLetter(error: string): UpdateResult { + const update: Partial = { + status: JobStatus.DEAD_LETTER, + error, + finishedAt: new Date(), + }; + + Object.assign(this, update); + + return [this.id, update]; + } + + get isFinished(): boolean { + return [JobStatus.COMPLETE, JobStatus.FAILED, JobStatus.DEAD_LETTER].includes(this.status); + } + + get waitSeconds(): number | undefined { + if (!this.startedAt) return undefined; + return Util.secondsDiff(this.created, this.startedAt); + } + + get runSeconds(): number | undefined { + if (!this.finishedAt) return undefined; + return Util.secondsDiff(this.startedAt, this.finishedAt); + } + + // Time the caller experiences — from acceptance of the job until the result. + get totalSeconds(): number | undefined { + if (!this.finishedAt) return undefined; + return Util.secondsDiff(this.created, this.finishedAt); + } +} diff --git a/src/subdomains/supporting/job/enums/index.ts b/src/subdomains/supporting/job/enums/index.ts new file mode 100644 index 0000000000..cb9f5011be --- /dev/null +++ b/src/subdomains/supporting/job/enums/index.ts @@ -0,0 +1,28 @@ +export enum JobStatus { + PENDING = 'Pending', + PROCESSING = 'Processing', + // Successful terminal state. + COMPLETE = 'Complete', + // Temporary: failed but attempts remain; waiting for the next run. + RETRY = 'Retry', + // Terminal after attempts are exhausted. + FAILED = 'Failed', + // Terminal and not retryable (broken input / missing precondition). Kept separate so an + // unworkable job does not consume attempts yet stays visible. + DEAD_LETTER = 'DeadLetter', +} + +export enum JobGroup { + ACCOUNT_MERGE = 'AccountMerge', +} + +export enum JobPhase { + WAIT = 'Wait', + RUN = 'Run', +} + +export enum JobAttemptOutcome { + COMPLETE = 'Complete', + FAILED = 'Failed', + DEAD_LETTER = 'DeadLetter', +} diff --git a/src/subdomains/supporting/job/exceptions/job-dead-letter.exception.ts b/src/subdomains/supporting/job/exceptions/job-dead-letter.exception.ts new file mode 100644 index 0000000000..ed6287f8b4 --- /dev/null +++ b/src/subdomains/supporting/job/exceptions/job-dead-letter.exception.ts @@ -0,0 +1,3 @@ +// Thrown by a handler when the job is unworkable from its input alone (broken payload, +// missing precondition). Any other error is treated as transient and retried. +export class JobDeadLetterException extends Error {} diff --git a/src/subdomains/supporting/job/interfaces/job-handler.interface.ts b/src/subdomains/supporting/job/interfaces/job-handler.interface.ts new file mode 100644 index 0000000000..2c206b0bfd --- /dev/null +++ b/src/subdomains/supporting/job/interfaces/job-handler.interface.ts @@ -0,0 +1,8 @@ +import { JobGroup } from '../enums'; + +// Handlers are registered from the outside (JobService.registerHandler) so the job module +// does not import domain modules and no circular dependency forms. +export interface JobHandler { + readonly group: JobGroup; + execute(input: TInput): Promise; +} diff --git a/src/subdomains/supporting/job/job-group.config.ts b/src/subdomains/supporting/job/job-group.config.ts new file mode 100644 index 0000000000..c115d5f873 --- /dev/null +++ b/src/subdomains/supporting/job/job-group.config.ts @@ -0,0 +1,29 @@ +import { Process } from 'src/shared/services/process.service'; +import { JobGroup } from './enums'; + +export interface JobGroupConfig { + maxWaitSeconds: number; + maxRunSeconds: number; + maxAttempts: number; + exposeResult: boolean; +} + +// Defaults are the source of truth when the settings table has no override (key `jobGroups`). +// Runtime overrides are partial per group and merge field-wise onto these values. +export const JOB_GROUP_DEFAULTS: Record = { + [JobGroup.ACCOUNT_MERGE]: { + // Sync merge p95 is ~16.4s today; 60s leaves roughly 3.5× headroom. + maxWaitSeconds: 5, + maxRunSeconds: 60, + maxAttempts: 3, + // Merge output carries an account identifier. The generic status endpoint must not expose + // result data for this group — only the domain endpoint may. Fail-closed: new groups set this deliberately. + exposeResult: false, + }, +}; + +// The single place mapping a job group to the Process that gates it, so drain() and the cron +// wrapper both check the same kill-switch flag instead of the mapping being duplicated or drifting. +export const JOB_GROUP_PROCESS: Record = { + [JobGroup.ACCOUNT_MERGE]: Process.JOB_ACCOUNT_MERGE, +}; diff --git a/src/subdomains/supporting/job/job.controller.ts b/src/subdomains/supporting/job/job.controller.ts new file mode 100644 index 0000000000..23d3c8ed9a --- /dev/null +++ b/src/subdomains/supporting/job/job.controller.ts @@ -0,0 +1,38 @@ +// Account ownership on the JWT payload uses `account` (user data ID), as defined in +// src/shared/auth/jwt-payload.interface.ts — not `id`. +import { Controller, Get, NotFoundException, Param, UseGuards } from '@nestjs/common'; +import { ApiBearerAuth, ApiOkResponse, ApiTags } from '@nestjs/swagger'; +import { GetJwt } from 'src/shared/auth/get-jwt.decorator'; +import { JwtPayload } from 'src/shared/auth/jwt-payload.interface'; +import { OptionalJwtAuthGuard } from 'src/shared/auth/optional.guard'; +import { JobDtoMapper } from './dto/job-dto.mapper'; +import { JobDto } from './dto/job.dto'; +import { JobService } from './services/job.service'; + +@ApiTags('Job') +@Controller('job') +export class JobController { + constructor(private readonly jobService: JobService) {} + + @Get(':uid') + @ApiBearerAuth() + @UseGuards(OptionalJwtAuthGuard) + @ApiOkResponse({ type: JobDto }) + async getJob(@GetJwt() jwt: JwtPayload | undefined, @Param('uid') uid: string): Promise { + const job = await this.jobService.getByUid(uid); + if (!job) throw new NotFoundException('Job not found'); + + // Avoids confirming the existence of another account's job to an unauthorized caller + // (same message, same status, whether the job doesn't exist or belongs to someone else). + if (job.userData && jwt !== undefined && jwt.account !== job.userData.id) { + throw new NotFoundException('Job not found'); + } + + // If no JWT is present, the uid itself is treated as sufficient proof of ownership (it is a + // random value known only to whoever triggered the job) — same trust level as the confirmation + // link that originates the job. + + const config = await this.jobService.getConfig(job.group); + return JobDtoMapper.mapJob(job, config); + } +} diff --git a/src/subdomains/supporting/job/job.module.ts b/src/subdomains/supporting/job/job.module.ts new file mode 100644 index 0000000000..012a1b357b --- /dev/null +++ b/src/subdomains/supporting/job/job.module.ts @@ -0,0 +1,19 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { SharedModule } from 'src/shared/shared.module'; +import { JobAttempt } from './entities/job-attempt.entity'; +import { Job } from './entities/job.entity'; +import { JobController } from './job.controller'; +import { JobAttemptRepository } from './repositories/job-attempt.repository'; +import { JobRepository } from './repositories/job.repository'; +import { JobDispatcherService } from './services/job-dispatcher.service'; +import { JobMetricService } from './services/job-metric.service'; +import { JobService } from './services/job.service'; + +@Module({ + imports: [TypeOrmModule.forFeature([Job, JobAttempt]), SharedModule], + controllers: [JobController], + providers: [JobRepository, JobAttemptRepository, JobService, JobDispatcherService, JobMetricService], + exports: [JobService, JobDispatcherService], +}) +export class JobModule {} diff --git a/src/subdomains/supporting/job/repositories/job-attempt.repository.ts b/src/subdomains/supporting/job/repositories/job-attempt.repository.ts new file mode 100644 index 0000000000..ecb67b2465 --- /dev/null +++ b/src/subdomains/supporting/job/repositories/job-attempt.repository.ts @@ -0,0 +1,11 @@ +import { Injectable } from '@nestjs/common'; +import { BaseRepository } from 'src/shared/repositories/base.repository'; +import { EntityManager } from 'typeorm'; +import { JobAttempt } from '../entities/job-attempt.entity'; + +@Injectable() +export class JobAttemptRepository extends BaseRepository { + constructor(manager: EntityManager) { + super(JobAttempt, manager); + } +} diff --git a/src/subdomains/supporting/job/repositories/job.repository.ts b/src/subdomains/supporting/job/repositories/job.repository.ts new file mode 100644 index 0000000000..56c7bb95fd --- /dev/null +++ b/src/subdomains/supporting/job/repositories/job.repository.ts @@ -0,0 +1,11 @@ +import { Injectable } from '@nestjs/common'; +import { BaseRepository } from 'src/shared/repositories/base.repository'; +import { EntityManager } from 'typeorm'; +import { Job } from '../entities/job.entity'; + +@Injectable() +export class JobRepository extends BaseRepository { + constructor(manager: EntityManager) { + super(Job, manager); + } +} diff --git a/src/subdomains/supporting/job/services/__tests__/job-dispatcher.service.spec.ts b/src/subdomains/supporting/job/services/__tests__/job-dispatcher.service.spec.ts new file mode 100644 index 0000000000..4c25419bdb --- /dev/null +++ b/src/subdomains/supporting/job/services/__tests__/job-dispatcher.service.spec.ts @@ -0,0 +1,235 @@ +import { createMock } from '@golevelup/ts-jest'; +import { Test, TestingModule } from '@nestjs/testing'; +import { DfxLogger } from 'src/shared/services/dfx-logger'; +import { MetricService } from 'src/shared/services/metric.service'; +import * as processServiceModule from 'src/shared/services/process.service'; +import { TestSharedModule } from 'src/shared/utils/test.shared.module'; +import { Job } from '../../entities/job.entity'; +import { JobGroup, JobStatus } from '../../enums'; +import { JobHandler } from '../../interfaces/job-handler.interface'; +import { JOB_GROUP_DEFAULTS } from '../../job-group.config'; +import { JobDispatcherService } from '../job-dispatcher.service'; +import { JobService } from '../job.service'; + +describe('JobDispatcherService', () => { + let service: JobDispatcherService; + let jobService: JobService; + let metricService: MetricService; + + beforeEach(async () => { + jobService = createMock(); + metricService = createMock(); + + jest.spyOn(processServiceModule, 'DisabledProcess').mockReturnValue(false); + jest.spyOn(jobService, 'getConfig').mockResolvedValue(JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE]); + jest.spyOn(jobService, 'recoverStale').mockResolvedValue(0); + jest.spyOn(metricService, 'registerGauge').mockImplementation(); + + const module: TestingModule = await Test.createTestingModule({ + imports: [TestSharedModule], + providers: [ + JobDispatcherService, + { provide: JobService, useValue: jobService }, + { provide: MetricService, useValue: metricService }, + ], + }).compile(); + + service = module.get(JobDispatcherService); + }); + + function createJob(partial: Partial = {}): Job { + const job = new Job(); + job.id = partial.id ?? 1; + job.uid = partial.uid ?? 'Jabc'; + job.group = partial.group ?? JobGroup.ACCOUNT_MERGE; + job.status = partial.status ?? JobStatus.PROCESSING; + job.idempotencyKey = partial.idempotencyKey ?? 'key-1'; + job.attempt = partial.attempt ?? 1; + job.maxAttempts = partial.maxAttempts ?? JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE].maxAttempts; + job.input = partial.input ?? JSON.stringify({ masterId: 1 }); + job.created = partial.created ?? new Date(Date.now() - 5000); + job.startedAt = partial.startedAt ?? new Date(Date.now() - 1000); + if (partial.finishedAt !== undefined) job.finishedAt = partial.finishedAt; + if (partial.traceparent !== undefined) job.traceparent = partial.traceparent; + return job; + } + + it('drain executes a claimed job and calls finish with the handler result', async () => { + const job = createJob({ id: 1, uid: 'J1' }); + const handler: JobHandler = { + group: JobGroup.ACCOUNT_MERGE, + execute: jest.fn().mockResolvedValue({ masterId: 42 }), + }; + + jest.spyOn(jobService, 'getHandler').mockReturnValue(handler); + jest.spyOn(jobService, 'claimNext').mockResolvedValueOnce(job).mockResolvedValueOnce(undefined); + jest.spyOn(jobService, 'finish').mockImplementation(async (j, output) => { + j.complete(output); + }); + + await service.drain(JobGroup.ACCOUNT_MERGE); + + expect(handler.execute).toHaveBeenCalledWith({ masterId: 1 }); + expect(jobService.finish).toHaveBeenCalledWith(job, { masterId: 42 }); + expect(jobService.finish).toHaveBeenCalledTimes(1); + expect(jobService.claimNext).toHaveBeenCalledTimes(2); + }); + + it('handler failure calls abort and continues with the next job', async () => { + const first = createJob({ id: 1, uid: 'J1' }); + const second = createJob({ id: 2, uid: 'J2', input: JSON.stringify({ masterId: 2 }) }); + const handlerError = new Error('transient'); + const handler: JobHandler = { + group: JobGroup.ACCOUNT_MERGE, + execute: jest.fn().mockRejectedValueOnce(handlerError).mockResolvedValueOnce({ ok: true }), + }; + + jest.spyOn(jobService, 'getHandler').mockReturnValue(handler); + jest + .spyOn(jobService, 'claimNext') + .mockResolvedValueOnce(first) + .mockResolvedValueOnce(second) + .mockResolvedValueOnce(undefined); + jest.spyOn(jobService, 'abort').mockImplementation(async (j) => { + j.fail('transient', new Date(Date.now() + 10000)); + }); + jest.spyOn(jobService, 'finish').mockImplementation(async (j, output) => { + j.complete(output); + }); + + await service.drain(JobGroup.ACCOUNT_MERGE); + + expect(jobService.abort).toHaveBeenCalledWith(first, handlerError); + expect(jobService.finish).toHaveBeenCalledWith(second, { ok: true }); + expect(handler.execute).toHaveBeenCalledTimes(2); + expect(jobService.claimNext).toHaveBeenCalledTimes(3); + }); + + it('missing handler does not claim jobs and logs an error', async () => { + jest.spyOn(jobService, 'getHandler').mockReturnValue(undefined); + const claimNext = jest.spyOn(jobService, 'claimNext'); + const errorSpy = jest.spyOn(DfxLogger.prototype, 'error').mockImplementation(); + + await service.drain(JobGroup.ACCOUNT_MERGE); + + expect(claimNext).not.toHaveBeenCalled(); + expect(errorSpy).toHaveBeenCalledWith(`No handler registered for job group ${JobGroup.ACCOUNT_MERGE}`); + errorSpy.mockRestore(); + }); + + it('drain calls recoverStale with double maxRunSeconds', async () => { + const config = JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE]; + jest.spyOn(jobService, 'getHandler').mockReturnValue({ + group: JobGroup.ACCOUNT_MERGE, + execute: async () => undefined, + }); + jest.spyOn(jobService, 'claimNext').mockResolvedValue(undefined); + + await service.drain(JobGroup.ACCOUNT_MERGE); + + expect(jobService.recoverStale).toHaveBeenCalledWith(JobGroup.ACCOUNT_MERGE, config.maxRunSeconds * 2); + }); + + it('drain claims nothing and does not call the handler when the group process is disabled', async () => { + jest.spyOn(processServiceModule, 'DisabledProcess').mockReturnValue(true); + const handler: JobHandler = { + group: JobGroup.ACCOUNT_MERGE, + execute: jest.fn(), + }; + jest.spyOn(jobService, 'getHandler').mockReturnValue(handler); + const claimNext = jest.spyOn(jobService, 'claimNext'); + const recoverStale = jest.spyOn(jobService, 'recoverStale'); + const getConfig = jest.spyOn(jobService, 'getConfig'); + + await service.drain(JobGroup.ACCOUNT_MERGE); + + expect(claimNext).not.toHaveBeenCalled(); + expect(handler.execute).not.toHaveBeenCalled(); + expect(recoverStale).not.toHaveBeenCalled(); + expect(getConfig).not.toHaveBeenCalled(); + }); + + it('stops at 100 jobs and logs a warning when the drain cap is hit', async () => { + const handler: JobHandler = { + group: JobGroup.ACCOUNT_MERGE, + execute: jest.fn().mockResolvedValue({ ok: true }), + }; + jest.spyOn(jobService, 'getHandler').mockReturnValue(handler); + jest.spyOn(jobService, 'claimNext').mockImplementation(async () => createJob()); + jest.spyOn(jobService, 'finish').mockImplementation(async (j, output) => { + j.complete(output); + }); + const warnSpy = jest.spyOn(DfxLogger.prototype, 'warn').mockImplementation(); + + await service.drain(JobGroup.ACCOUNT_MERGE); + + expect(jobService.claimNext).toHaveBeenCalledTimes(100); + expect(handler.execute).toHaveBeenCalledTimes(100); + expect(warnSpy).toHaveBeenCalledWith( + `Job drain cap reached for group ${JobGroup.ACCOUNT_MERGE}: processed 100 jobs in one drain`, + ); + warnSpy.mockRestore(); + }); + + it('counts dfx_job_finished complete when finished, and does not count failed when abort leaves RETRY', async () => { + const completeJob = createJob({ id: 1, uid: 'Jcomplete' }); + const retryJob = createJob({ id: 2, uid: 'Jretry' }); + const handler: JobHandler = { + group: JobGroup.ACCOUNT_MERGE, + execute: jest.fn().mockResolvedValueOnce({ done: true }).mockRejectedValueOnce(new Error('retry me')), + }; + + jest.spyOn(jobService, 'getHandler').mockReturnValue(handler); + jest + .spyOn(jobService, 'claimNext') + .mockResolvedValueOnce(completeJob) + .mockResolvedValueOnce(retryJob) + .mockResolvedValueOnce(undefined); + jest.spyOn(jobService, 'finish').mockImplementation(async (j, output) => { + j.complete(output); + }); + jest.spyOn(jobService, 'abort').mockImplementation(async (j, error) => { + j.fail(error.message, new Date(Date.now() + 10000)); + }); + + await service.drain(JobGroup.ACCOUNT_MERGE); + + expect(metricService.increment).toHaveBeenCalledWith('dfx_job_finished', { + group: JobGroup.ACCOUNT_MERGE, + outcome: 'complete', + }); + expect(metricService.increment).not.toHaveBeenCalledWith('dfx_job_finished', { + group: JobGroup.ACCOUNT_MERGE, + outcome: 'failed', + }); + expect(metricService.increment).not.toHaveBeenCalledWith('dfx_job_finished', { + group: JobGroup.ACCOUNT_MERGE, + outcome: 'dead_letter', + }); + expect(metricService.increment).toHaveBeenCalledTimes(1); + }); + + it('propagates finish persistence failures without counting dfx_job_finished complete', async () => { + const job = createJob({ id: 3, uid: 'Jpersist' }); + const handler: JobHandler = { + group: JobGroup.ACCOUNT_MERGE, + execute: jest.fn().mockResolvedValue({ done: true }), + }; + const persistError = new Error('connection lost'); + + jest.spyOn(jobService, 'getHandler').mockReturnValue(handler); + jest.spyOn(jobService, 'claimNext').mockResolvedValueOnce(job).mockResolvedValueOnce(undefined); + jest.spyOn(jobService, 'finish').mockRejectedValue(persistError); + const errorSpy = jest.spyOn(DfxLogger.prototype, 'error').mockImplementation(); + + await expect(service.drain(JobGroup.ACCOUNT_MERGE)).rejects.toThrow('connection lost'); + + expect(metricService.increment).not.toHaveBeenCalledWith('dfx_job_finished', { + group: JobGroup.ACCOUNT_MERGE, + outcome: 'complete', + }); + expect(errorSpy).toHaveBeenCalledWith(`Failed to persist result for job ${job.uid}:`, persistError); + + errorSpy.mockRestore(); + }); +}); diff --git a/src/subdomains/supporting/job/services/__tests__/job-metric.service.spec.ts b/src/subdomains/supporting/job/services/__tests__/job-metric.service.spec.ts new file mode 100644 index 0000000000..9a835af647 --- /dev/null +++ b/src/subdomains/supporting/job/services/__tests__/job-metric.service.spec.ts @@ -0,0 +1,128 @@ +import { createMock } from '@golevelup/ts-jest'; +import { Test, TestingModule } from '@nestjs/testing'; +import { ObservableResult } from '@opentelemetry/api'; +import { MetricService } from 'src/shared/services/metric.service'; +import { TestSharedModule } from 'src/shared/utils/test.shared.module'; +import { JobGroup, JobPhase, JobStatus } from '../../enums'; +import { JOB_GROUP_DEFAULTS } from '../../job-group.config'; +import { JobRepository } from '../../repositories/job.repository'; +import { JobMetricService } from '../job-metric.service'; +import { JobService } from '../job.service'; + +describe('JobMetricService', () => { + let service: JobMetricService; + let jobRepo: JobRepository; + let metricService: MetricService; + let jobService: JobService; + const gaugeCallbacks = new Map void | Promise>(); + + beforeEach(async () => { + jobRepo = createMock(); + metricService = createMock(); + jobService = createMock(); + gaugeCallbacks.clear(); + + jest.spyOn(metricService, 'registerGauge').mockImplementation((name, _unit, observe) => { + gaugeCallbacks.set(name, observe); + }); + jest.spyOn(jobService, 'getConfig').mockImplementation(async (group) => JOB_GROUP_DEFAULTS[group]); + + const module: TestingModule = await Test.createTestingModule({ + imports: [TestSharedModule], + providers: [ + JobMetricService, + { provide: JobRepository, useValue: jobRepo }, + { provide: MetricService, useValue: metricService }, + { provide: JobService, useValue: jobService }, + ], + }).compile(); + + service = module.get(JobMetricService); + service.onModuleInit(); + }); + + it('updateSnapshot counts overSlaRun for PROCESSING jobs older than maxRunSeconds', async () => { + const config = JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE]; + jest.spyOn(jobRepo, 'countBy').mockImplementation(async (where) => { + const criteria = Array.isArray(where) ? where[0] : where; + if (criteria && 'status' in criteria && criteria.status === JobStatus.PROCESSING && 'startedAt' in criteria) { + return 3; + } + if (criteria && 'status' in criteria && criteria.status === JobStatus.DEAD_LETTER) { + return 0; + } + return 0; + }); + jest.spyOn(jobRepo, 'findOne').mockResolvedValue(null); + + await service.updateSnapshot(); + + const overSlaCalls = (jobRepo.countBy as jest.Mock).mock.calls.filter((call) => { + const where = call[0] as { status?: JobStatus; startedAt?: unknown }; + return where && where.status === JobStatus.PROCESSING && where.startedAt !== undefined; + }); + expect(overSlaCalls).toHaveLength(1); + expect(overSlaCalls[0][0]).toEqual( + expect.objectContaining({ + group: JobGroup.ACCOUNT_MERGE, + status: JobStatus.PROCESSING, + }), + ); + expect(overSlaCalls[0][0].startedAt).toBeDefined(); + + const observe = jest.fn(); + const overSlaGauge = gaugeCallbacks.get('dfx_job_over_sla'); + expect(overSlaGauge).toBeDefined(); + await overSlaGauge!({ observe } as unknown as ObservableResult); + + expect(observe).toHaveBeenCalledWith(3, { + group: JobGroup.ACCOUNT_MERGE, + phase: JobPhase.RUN, + }); + expect(observe).toHaveBeenCalledWith(0, { + group: JobGroup.ACCOUNT_MERGE, + phase: JobPhase.WAIT, + }); + + // Sanity: SLA configured limit matches defaults used for the cutoff. + expect(config.maxRunSeconds).toBe(60); + }); + + it('gauge callbacks report the snapshot without querying the database again', async () => { + jest.spyOn(jobRepo, 'countBy').mockResolvedValue(7); + jest.spyOn(jobRepo, 'findOne').mockResolvedValue(null); + + await service.updateSnapshot(); + + const countByCallsAfterSnapshot = (jobRepo.countBy as jest.Mock).mock.calls.length; + const findOneCallsAfterSnapshot = (jobRepo.findOne as jest.Mock).mock.calls.length; + expect(countByCallsAfterSnapshot).toBeGreaterThan(0); + expect(findOneCallsAfterSnapshot).toBeGreaterThan(0); + + const observe = jest.fn(); + for (const callback of gaugeCallbacks.values()) { + await callback({ observe } as unknown as ObservableResult); + } + + expect(jobRepo.countBy).toHaveBeenCalledTimes(countByCallsAfterSnapshot); + expect(jobRepo.findOne).toHaveBeenCalledTimes(findOneCallsAfterSnapshot); + + expect(observe).toHaveBeenCalledWith(7, { group: JobGroup.ACCOUNT_MERGE }); + expect(observe).toHaveBeenCalledWith(7, { + group: JobGroup.ACCOUNT_MERGE, + phase: JobPhase.WAIT, + }); + expect(observe).toHaveBeenCalledWith(7, { + group: JobGroup.ACCOUNT_MERGE, + phase: JobPhase.RUN, + }); + expect(observe).toHaveBeenCalledWith(JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE].maxWaitSeconds, { + group: JobGroup.ACCOUNT_MERGE, + phase: JobPhase.WAIT, + }); + expect(observe).toHaveBeenCalledWith(JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE].maxRunSeconds, { + group: JobGroup.ACCOUNT_MERGE, + phase: JobPhase.RUN, + }); + }); +}); diff --git a/src/subdomains/supporting/job/services/__tests__/job.service.spec.ts b/src/subdomains/supporting/job/services/__tests__/job.service.spec.ts new file mode 100644 index 0000000000..2603f17dd9 --- /dev/null +++ b/src/subdomains/supporting/job/services/__tests__/job.service.spec.ts @@ -0,0 +1,553 @@ +import { createMock } from '@golevelup/ts-jest'; +import { Test, TestingModule } from '@nestjs/testing'; +import { SettingService } from 'src/shared/models/setting/setting.service'; +import { DfxLogger } from 'src/shared/services/dfx-logger'; +import { MetricService } from 'src/shared/services/metric.service'; +import { TestSharedModule } from 'src/shared/utils/test.shared.module'; +import { TestUtil } from 'src/shared/utils/test.util'; +import { IsNull } from 'typeorm'; +import { JobAttempt } from '../../entities/job-attempt.entity'; +import { Job } from '../../entities/job.entity'; +import { JobAttemptOutcome, JobGroup, JobStatus } from '../../enums'; +import { JobDeadLetterException } from '../../exceptions/job-dead-letter.exception'; +import { JobHandler } from '../../interfaces/job-handler.interface'; +import { JOB_GROUP_DEFAULTS } from '../../job-group.config'; +import { JobAttemptRepository } from '../../repositories/job-attempt.repository'; +import { JobRepository } from '../../repositories/job.repository'; +import { JobService } from '../job.service'; + +describe('JobService', () => { + let service: JobService; + let jobRepo: JobRepository; + let jobAttemptRepo: JobAttemptRepository; + let settingService: SettingService; + let metricService: MetricService; + + beforeEach(async () => { + jobRepo = createMock(); + jobAttemptRepo = createMock(); + settingService = createMock(); + metricService = createMock(); + + jest.spyOn(settingService, 'getObjCached').mockResolvedValue(undefined); + + const module: TestingModule = await Test.createTestingModule({ + imports: [TestSharedModule], + providers: [ + JobService, + { provide: JobRepository, useValue: jobRepo }, + { provide: JobAttemptRepository, useValue: jobAttemptRepo }, + { provide: SettingService, useValue: settingService }, + { provide: MetricService, useValue: metricService }, + TestUtil.provideConfig(), + ], + }).compile(); + + service = module.get(JobService); + }); + + function createJob(partial: Partial = {}): Job { + const job = new Job(); + job.id = partial.id ?? 1; + job.uid = partial.uid ?? 'Jabc'; + job.group = partial.group ?? JobGroup.ACCOUNT_MERGE; + job.status = partial.status ?? JobStatus.PENDING; + job.idempotencyKey = partial.idempotencyKey ?? 'key-1'; + job.attempt = partial.attempt ?? 0; + job.maxAttempts = partial.maxAttempts ?? JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE].maxAttempts; + job.input = partial.input ?? JSON.stringify({ masterId: 1 }); + if (partial.output !== undefined) job.output = partial.output; + if (partial.claimedAt !== undefined) job.claimedAt = partial.claimedAt; + if (partial.claimedBy !== undefined) job.claimedBy = partial.claimedBy; + if (partial.startedAt !== undefined) job.startedAt = partial.startedAt; + if (partial.finishedAt !== undefined) job.finishedAt = partial.finishedAt; + if (partial.nextAttemptAt !== undefined) job.nextAttemptAt = partial.nextAttemptAt; + if (partial.error !== undefined) job.error = partial.error; + return job; + } + + it('should be defined', () => { + expect(service).toBeDefined(); + }); + + describe('getConfig', () => { + it('overrides only the provided field and keeps remaining defaults', async () => { + jest.spyOn(settingService, 'getObjCached').mockResolvedValue([{ group: JobGroup.ACCOUNT_MERGE, maxAttempts: 7 }]); + + const config = await service.getConfig(JobGroup.ACCOUNT_MERGE); + const defaults = JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE]; + + expect(config).toEqual({ + maxWaitSeconds: defaults.maxWaitSeconds, + maxRunSeconds: defaults.maxRunSeconds, + maxAttempts: 7, + exposeResult: defaults.exposeResult, + }); + }); + }); + + describe('enqueue', () => { + it('creates a PENDING job with uid and maxAttempts from config when none exists', async () => { + jest.spyOn(jobRepo, 'findOne').mockResolvedValue(undefined); + jest.spyOn(jobRepo, 'create').mockImplementation((entity: Partial) => Object.assign(new Job(), entity)); + jest.spyOn(jobRepo, 'save').mockImplementation(async (entity: Job) => { + entity.id = 42; + return entity; + }); + + const result = await service.enqueue(JobGroup.ACCOUNT_MERGE, 'merge-1-2', { masterId: 1, slaveId: 2 }, {}); + + expect(result.status).toBe(JobStatus.PENDING); + expect(result.uid).toMatch(/^J/); + expect(result.maxAttempts).toBe(JOB_GROUP_DEFAULTS[JobGroup.ACCOUNT_MERGE].maxAttempts); + expect(result.group).toBe(JobGroup.ACCOUNT_MERGE); + expect(result.idempotencyKey).toBe('merge-1-2'); + expect(result.inputData).toEqual({ masterId: 1, slaveId: 2 }); + expect(jobRepo.save).toHaveBeenCalled(); + }); + + it('returns the existing job for the same group/idempotencyKey without saving', async () => { + const existing = createJob({ id: 7, idempotencyKey: 'merge-1-2', status: JobStatus.PROCESSING }); + jest.spyOn(jobRepo, 'findOne').mockResolvedValue(existing); + + const result = await service.enqueue(JobGroup.ACCOUNT_MERGE, 'merge-1-2', { masterId: 1 }, {}); + + expect(result).toBe(existing); + expect(jobRepo.save).not.toHaveBeenCalled(); + expect(jobRepo.create).not.toHaveBeenCalled(); + }); + + it('reloads the existing job on unique violation (23505) and does not throw', async () => { + const existing = createJob({ id: 9, idempotencyKey: 'merge-1-2' }); + jest.spyOn(jobRepo, 'findOne').mockResolvedValueOnce(undefined).mockResolvedValueOnce(existing); + jest.spyOn(jobRepo, 'create').mockImplementation((entity: Partial) => Object.assign(new Job(), entity)); + jest.spyOn(jobRepo, 'save').mockRejectedValue(Object.assign(new Error('duplicate key'), { code: '23505' })); + + const result = await service.enqueue(JobGroup.ACCOUNT_MERGE, 'merge-1-2', { masterId: 1 }, {}); + + expect(result).toBe(existing); + expect(jobRepo.findOne).toHaveBeenCalledTimes(2); + }); + + it('re-throws non-unique save errors', async () => { + jest.spyOn(jobRepo, 'findOne').mockResolvedValue(undefined); + jest.spyOn(jobRepo, 'create').mockImplementation((entity: Partial) => Object.assign(new Job(), entity)); + jest.spyOn(jobRepo, 'save').mockRejectedValue(Object.assign(new Error('connection lost'), { code: '08006' })); + + await expect(service.enqueue(JobGroup.ACCOUNT_MERGE, 'merge-1-2', { masterId: 1 }, {})).rejects.toThrow( + 'connection lost', + ); + }); + }); + + describe('claimNext', () => { + it('returns the job when the conditional update reports affected=1', async () => { + const candidate = createJob({ id: 3, status: JobStatus.PENDING, attempt: 0 }); + jest.spyOn(jobRepo, 'find').mockResolvedValue([candidate]); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + const result = await service.claimNext(JobGroup.ACCOUNT_MERGE, 'worker-1'); + + expect(result).toBe(candidate); + expect(result.status).toBe(JobStatus.PROCESSING); + expect(result.attempt).toBe(1); + expect(result.claimedBy).toBe('worker-1'); + expect(jobRepo.update).toHaveBeenCalledWith( + { id: 3, status: JobStatus.PENDING }, + expect.objectContaining({ status: JobStatus.PROCESSING, claimedBy: 'worker-1', attempt: 1 }), + ); + }); + + it('skips a first candidate with affected=0 and takes the second', async () => { + const first = createJob({ id: 1, status: JobStatus.PENDING, attempt: 0 }); + const second = createJob({ id: 2, status: JobStatus.RETRY, attempt: 1 }); + jest.spyOn(jobRepo, 'find').mockResolvedValue([first, second]); + jest + .spyOn(jobRepo, 'update') + .mockResolvedValueOnce({ affected: 0, raw: [], generatedMaps: [] }) + .mockResolvedValueOnce({ affected: 1, raw: [], generatedMaps: [] }); + + const result = await service.claimNext(JobGroup.ACCOUNT_MERGE, 'worker-1'); + + expect(result).toBe(second); + expect(result.status).toBe(JobStatus.PROCESSING); + expect(result.attempt).toBe(2); + expect(jobRepo.update).toHaveBeenCalledTimes(2); + }); + + it('returns undefined when find yields an empty list', async () => { + jest.spyOn(jobRepo, 'find').mockResolvedValue([]); + + const result = await service.claimNext(JobGroup.ACCOUNT_MERGE, 'worker-1'); + + expect(result).toBeUndefined(); + expect(jobRepo.update).not.toHaveBeenCalled(); + }); + + it('creates exactly one attempt row with the winning attempt number and claimedBy', async () => { + const candidate = createJob({ id: 30, status: JobStatus.PENDING, attempt: 0 }); + jest.spyOn(jobRepo, 'find').mockResolvedValue([candidate]); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + jest.spyOn(jobAttemptRepo, 'create').mockImplementation((entity) => Object.assign(new JobAttempt(), entity)); + const saveSpy = jest.spyOn(jobAttemptRepo, 'save').mockImplementation(async (entity) => entity as JobAttempt); + + const result = await service.claimNext(JobGroup.ACCOUNT_MERGE, 'worker-1'); + + expect(result).toBe(candidate); + expect(saveSpy).toHaveBeenCalledTimes(1); + expect(saveSpy).toHaveBeenCalledWith( + expect.objectContaining({ job: candidate, attempt: 1, claimedBy: 'worker-1' }), + ); + }); + + it('creates no attempt row for a lost claim (affected: 0)', async () => { + const candidate = createJob({ id: 31, status: JobStatus.PENDING, attempt: 0 }); + jest.spyOn(jobRepo, 'find').mockResolvedValue([candidate]); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 0, raw: [], generatedMaps: [] }); + const saveSpy = jest.spyOn(jobAttemptRepo, 'save'); + + const result = await service.claimNext(JobGroup.ACCOUNT_MERGE, 'worker-1'); + + expect(result).toBeUndefined(); + expect(saveSpy).not.toHaveBeenCalled(); + }); + + it('propagates the error and returns no job when writing the attempt row fails', async () => { + const candidate = createJob({ id: 32, status: JobStatus.PENDING, attempt: 0 }); + jest.spyOn(jobRepo, 'find').mockResolvedValue([candidate]); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + jest.spyOn(jobAttemptRepo, 'create').mockImplementation((entity) => Object.assign(new JobAttempt(), entity)); + jest.spyOn(jobAttemptRepo, 'save').mockRejectedValue(new Error('insert failed')); + + await expect(service.claimNext(JobGroup.ACCOUNT_MERGE, 'worker-1')).rejects.toThrow('insert failed'); + }); + }); + + describe('abort', () => { + it('sets DEAD_LETTER for JobDeadLetterException without scheduling a retry', async () => { + const job = createJob({ id: 5, status: JobStatus.PROCESSING, attempt: 1 }); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + await service.abort(job, new JobDeadLetterException('broken input')); + + expect(job.status).toBe(JobStatus.DEAD_LETTER); + expect(job.error).toBe('broken input'); + expect(job.finishedAt).toBeInstanceOf(Date); + expect(job.nextAttemptAt).toBeUndefined(); + expect(jobRepo.update).toHaveBeenCalledWith( + expect.objectContaining({ + id: 5, + status: JobStatus.PROCESSING, + attempt: job.attempt, + claimedBy: job.claimedBy, + }), + expect.objectContaining({ status: JobStatus.DEAD_LETTER, error: 'broken input' }), + ); + }); + + it('sets RETRY with a future nextAttemptAt when attempts remain', async () => { + const job = createJob({ + id: 6, + status: JobStatus.PROCESSING, + attempt: 1, + maxAttempts: 3, + }); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + const before = Date.now(); + await service.abort(job, new Error('transient')); + + expect(job.status).toBe(JobStatus.RETRY); + expect(job.error).toBe('transient'); + expect(job.nextAttemptAt).toBeInstanceOf(Date); + expect(job.nextAttemptAt.getTime()).toBeGreaterThan(before); + expect(job.finishedAt).toBeUndefined(); + }); + + it('sets FAILED without nextAttemptAt when attempts are exhausted', async () => { + const job = createJob({ + id: 8, + status: JobStatus.PROCESSING, + attempt: 3, + maxAttempts: 3, + }); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + await service.abort(job, new Error('still failing')); + + expect(job.status).toBe(JobStatus.FAILED); + expect(job.error).toBe('still failing'); + expect(job.finishedAt).toBeInstanceOf(Date); + expect(job.nextAttemptAt).toBeUndefined(); + expect(jobRepo.update).toHaveBeenCalledWith( + expect.objectContaining({ + id: 8, + status: JobStatus.PROCESSING, + attempt: job.attempt, + claimedBy: job.claimedBy, + }), + expect.objectContaining({ status: JobStatus.FAILED, error: 'still failing' }), + ); + }); + + it('skips the job update and does not throw when ownership is lost (affected: 0)', async () => { + const job = createJob({ + id: 50, + uid: 'Jlost-abort', + status: JobStatus.PROCESSING, + attempt: 1, + maxAttempts: 3, + claimedBy: 'worker-old', + }); + jest.spyOn(jobAttemptRepo, 'findOne').mockResolvedValue(undefined); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 0, raw: [], generatedMaps: [] }); + const warnSpy = jest.spyOn(DfxLogger.prototype, 'warn').mockImplementation(); + const errorSpy = jest.spyOn(DfxLogger.prototype, 'error').mockImplementation(); + + await expect(service.abort(job, new Error('transient'))).resolves.toBeUndefined(); + + expect(jobRepo.update).toHaveBeenCalledTimes(1); + expect(jobRepo.update).toHaveBeenCalledWith( + expect.objectContaining({ + id: 50, + status: JobStatus.PROCESSING, + attempt: job.attempt, + claimedBy: job.claimedBy, + }), + expect.anything(), + ); + expect(warnSpy).toHaveBeenCalled(); + expect(errorSpy).not.toHaveBeenCalled(); + + warnSpy.mockRestore(); + errorSpy.mockRestore(); + }); + }); + + describe('abort — attempt history', () => { + it('closes the attempt row with outcome Failed and the full message for a plain error', async () => { + const job = createJob({ id: 41, status: JobStatus.PROCESSING, attempt: 1, maxAttempts: 3 }); + const attemptRow = Object.assign(new JobAttempt(), { + id: 100, + job, + attempt: 1, + claimedBy: 'worker-1', + claimedAt: new Date(), + }); + jest.spyOn(jobAttemptRepo, 'findOne').mockResolvedValue(attemptRow); + const attemptUpdateSpy = jest + .spyOn(jobAttemptRepo, 'update') + .mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + await service.abort(job, new Error('transient failure with full detail')); + + expect(attemptUpdateSpy).toHaveBeenCalledWith( + 100, + expect.objectContaining({ outcome: JobAttemptOutcome.FAILED, error: 'transient failure with full detail' }), + ); + }); + + it('closes the attempt row with outcome DeadLetter for a JobDeadLetterException', async () => { + const job = createJob({ id: 42, status: JobStatus.PROCESSING, attempt: 1 }); + const attemptRow = Object.assign(new JobAttempt(), { + id: 101, + job, + attempt: 1, + claimedBy: 'worker-1', + claimedAt: new Date(), + }); + jest.spyOn(jobAttemptRepo, 'findOne').mockResolvedValue(attemptRow); + const attemptUpdateSpy = jest + .spyOn(jobAttemptRepo, 'update') + .mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + await service.abort(job, new JobDeadLetterException('broken input')); + + expect(attemptUpdateSpy).toHaveBeenCalledWith( + 101, + expect.objectContaining({ outcome: JobAttemptOutcome.DEAD_LETTER, error: 'broken input' }), + ); + }); + }); + + describe('finish', () => { + it('closes the attempt row with outcome Complete before writing the job snapshot', async () => { + const job = createJob({ id: 40, status: JobStatus.PROCESSING, attempt: 1 }); + const attemptRow = Object.assign(new JobAttempt(), { + id: 99, + job, + attempt: 1, + claimedBy: 'worker-1', + claimedAt: new Date(), + }); + jest.spyOn(jobAttemptRepo, 'findOne').mockResolvedValue(attemptRow); + const attemptUpdateSpy = jest + .spyOn(jobAttemptRepo, 'update') + .mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + const jobUpdateSpy = jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + await service.finish(job, { done: true }); + + expect(attemptUpdateSpy).toHaveBeenCalledWith( + 99, + expect.objectContaining({ outcome: JobAttemptOutcome.COMPLETE }), + ); + expect(jobUpdateSpy).toHaveBeenCalled(); + expect(attemptUpdateSpy.mock.invocationCallOrder[0]).toBeLessThan(jobUpdateSpy.mock.invocationCallOrder[0]); + }); + + it('skips the job update and does not throw when ownership is lost (affected: 0)', async () => { + const job = createJob({ + id: 51, + uid: 'Jlost-finish', + status: JobStatus.PROCESSING, + attempt: 1, + claimedBy: 'worker-old', + }); + jest.spyOn(jobAttemptRepo, 'findOne').mockResolvedValue(undefined); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 0, raw: [], generatedMaps: [] }); + const warnSpy = jest.spyOn(DfxLogger.prototype, 'warn').mockImplementation(); + + await expect(service.finish(job, { done: true })).resolves.toBeUndefined(); + + expect(jobRepo.update).toHaveBeenCalledTimes(1); + expect(jobRepo.update).toHaveBeenCalledWith( + expect.objectContaining({ + id: 51, + status: JobStatus.PROCESSING, + attempt: job.attempt, + claimedBy: job.claimedBy, + }), + expect.anything(), + ); + expect(warnSpy).toHaveBeenCalled(); + + warnSpy.mockRestore(); + }); + + it('does not rewrite an already closed attempt row and still updates the job snapshot', async () => { + const job = createJob({ + id: 52, + uid: 'Jclosed-attempt', + status: JobStatus.PROCESSING, + attempt: 1, + claimedBy: 'worker-1', + }); + jest.spyOn(jobAttemptRepo, 'findOne').mockResolvedValue(undefined); + const attemptUpdateSpy = jest.spyOn(jobAttemptRepo, 'update'); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + await service.finish(job, { done: true }); + + expect(jobAttemptRepo.findOne).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ finishedAt: IsNull() }), + }), + ); + expect(attemptUpdateSpy).not.toHaveBeenCalled(); + expect(jobRepo.update).toHaveBeenCalledWith( + expect.objectContaining({ + id: 52, + status: JobStatus.PROCESSING, + attempt: job.attempt, + claimedBy: job.claimedBy, + }), + expect.anything(), + ); + }); + }); + + describe('registerHandler', () => { + it('throws when a handler for the same group is already registered', () => { + const handler: JobHandler = { + group: JobGroup.ACCOUNT_MERGE, + execute: async () => undefined, + }; + + service.registerHandler(handler); + + expect(() => service.registerHandler(handler)).toThrow( + `Handler for job group ${JobGroup.ACCOUNT_MERGE} is already registered`, + ); + }); + }); + + describe('recoverStale', () => { + it('moves a PROCESSING job with remaining attempts to RETRY', async () => { + const job = createJob({ + id: 20, + status: JobStatus.PROCESSING, + attempt: 1, + maxAttempts: 3, + claimedAt: new Date(Date.now() - 60_000), + }); + jest.spyOn(jobRepo, 'find').mockResolvedValue([job]); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + const recovered = await service.recoverStale(JobGroup.ACCOUNT_MERGE, 30); + + expect(recovered).toBe(1); + expect(job.status).toBe(JobStatus.RETRY); + expect(job.nextAttemptAt).toBeInstanceOf(Date); + expect(jobRepo.update).toHaveBeenCalledWith( + expect.objectContaining({ id: job.id, status: JobStatus.PROCESSING }), + expect.objectContaining({ status: JobStatus.RETRY }), + ); + }); + + it('fails a PROCESSING job when the attempt budget is exhausted', async () => { + const job = createJob({ + id: 21, + status: JobStatus.PROCESSING, + attempt: 3, + maxAttempts: 3, + claimedAt: new Date(Date.now() - 60_000), + }); + jest.spyOn(jobRepo, 'find').mockResolvedValue([job]); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 1, raw: [], generatedMaps: [] }); + + const recovered = await service.recoverStale(JobGroup.ACCOUNT_MERGE, 30); + + expect(recovered).toBe(1); + expect(job.status).toBe(JobStatus.FAILED); + expect(job.finishedAt).toBeInstanceOf(Date); + expect(job.error).toContain('Orphaned after restart'); + expect(job.error).toContain('3/3'); + expect(jobRepo.update).toHaveBeenCalledWith( + expect.objectContaining({ id: job.id, status: JobStatus.PROCESSING }), + expect.objectContaining({ status: JobStatus.FAILED }), + ); + }); + + it('does not mutate or log when the conditional update affects zero rows', async () => { + const job = createJob({ + id: 22, + status: JobStatus.PROCESSING, + attempt: 1, + maxAttempts: 3, + claimedAt: new Date(Date.now() - 60_000), + }); + const originalStatus = job.status; + jest.spyOn(jobRepo, 'find').mockResolvedValue([job]); + jest.spyOn(jobRepo, 'update').mockResolvedValue({ affected: 0, raw: [], generatedMaps: [] }); + const warnSpy = jest.spyOn(DfxLogger.prototype, 'warn').mockImplementation(); + const errorSpy = jest.spyOn(DfxLogger.prototype, 'error').mockImplementation(); + + const recovered = await service.recoverStale(JobGroup.ACCOUNT_MERGE, 30); + + expect(recovered).toBe(0); + expect(job.status).toBe(originalStatus); + expect(warnSpy).not.toHaveBeenCalled(); + expect(errorSpy).not.toHaveBeenCalled(); + expect(jobRepo.update).toHaveBeenCalledWith( + expect.objectContaining({ id: job.id, status: JobStatus.PROCESSING }), + expect.objectContaining({ status: JobStatus.RETRY }), + ); + + warnSpy.mockRestore(); + errorSpy.mockRestore(); + }); + }); +}); diff --git a/src/subdomains/supporting/job/services/job-dispatcher.service.ts b/src/subdomains/supporting/job/services/job-dispatcher.service.ts new file mode 100644 index 0000000000..96c8fd9851 --- /dev/null +++ b/src/subdomains/supporting/job/services/job-dispatcher.service.ts @@ -0,0 +1,180 @@ +import { Injectable, OnModuleInit } from '@nestjs/common'; +import { CronExpression } from '@nestjs/schedule'; +import { ROOT_CONTEXT, SpanStatusCode, propagation, trace } from '@opentelemetry/api'; +import { hostname } from 'os'; +import { DfxLogger } from 'src/shared/services/dfx-logger'; +import { MetricService } from 'src/shared/services/metric.service'; +import { DisabledProcess, Process } from 'src/shared/services/process.service'; +import { DfxCron } from 'src/shared/utils/cron'; +import { Util } from 'src/shared/utils/util'; +import { Job } from '../entities/job.entity'; +import { JobGroup } from '../enums'; +import { JobDeadLetterException } from '../exceptions/job-dead-letter.exception'; +import { JobHandler } from '../interfaces/job-handler.interface'; +import { JOB_GROUP_PROCESS } from '../job-group.config'; +import { JobService } from './job.service'; + +const MAX_JOBS_PER_DRAIN = 100; + +@Injectable() +export class JobDispatcherService implements OnModuleInit { + private readonly logger = new DfxLogger(JobDispatcherService); + // Identifies the claim owner so an orphaned claim can be reassigned to a different runner after a restart. + private readonly owner = `${hostname()}:${process.pid}`; + private readonly lastRunAt = new Map(); + + constructor( + private readonly jobService: JobService, + private readonly metricService: MetricService, + ) { + const now = new Date(); + for (const group of Object.values(JobGroup)) { + this.lastRunAt.set(group, now); + } + } + + onModuleInit(): void { + this.metricService.registerGauge('dfx_job_seconds_since_last_run', 's', (result) => { + for (const group of Object.values(JobGroup)) { + const lastRunAt = this.lastRunAt.get(group); + if (lastRunAt === undefined) continue; + result.observe(Util.secondsDiff(lastRunAt), { group }); + } + }); + } + + // Kicking right after a job is created brings the wait time down to effectively zero; the cron + // below is the safety net in case the kick is lost (process restart, a second instance, an error + // inside the kick itself) — the delivery guarantee must NOT depend on the kick. + kick(group: JobGroup): void { + void this.drain(group).catch((e) => { + const error = e instanceof Error ? e : new Error(String(e)); + this.logger.error(`Failed to drain job group ${group}:`, error); + }); + } + + async drain(group: JobGroup): Promise { + // Named jobProcess, not process: the class uses Node's global `process` for the owner id. + const jobProcess = JOB_GROUP_PROCESS[group]; + if (DisabledProcess(jobProcess)) { + this.logger.verbose(`Skipping job drain for group ${group} - process ${jobProcess} is disabled`); + return; + } + + const config = await this.jobService.getConfig(group); + // Threshold is derived (maxRunSeconds * 2) rather than a separate config knob: one fewer + // setting to configure and keep in sync. + await this.jobService.recoverStale(group, config.maxRunSeconds * 2); + + const handler = this.jobService.getHandler(group); + if (!handler) { + // A missing handler is a deployment/wiring bug, not an unworkable task; the queue should + // visibly grow rather than silently discarding jobs. + this.logger.error(`No handler registered for job group ${group}`); + return; + } + + let processed = 0; + for (;;) { + if (processed >= MAX_JOBS_PER_DRAIN) { + this.logger.warn(`Job drain cap reached for group ${group}: processed ${MAX_JOBS_PER_DRAIN} jobs in one drain`); + break; + } + + const job = await this.jobService.claimNext(group, this.owner); + if (!job) break; + + processed += 1; + await this.executeJob(handler, job); + } + + // Heartbeat attests that the sweep ran, not that there was work to do. + this.lastRunAt.set(group, new Date()); + } + + @DfxCron(CronExpression.EVERY_MINUTE, { process: Process.JOB_ACCOUNT_MERGE, timeout: 1800 }) + async sweepAccountMerge(): Promise { + await this.drain(JobGroup.ACCOUNT_MERGE); + } + + private async executeJob(handler: JobHandler, job: Job): Promise { + // Keeps the incoming request and the later job execution in the same trace — without it, + // nothing would show what happened to a job after the request already returned 202. + const tracer = trace.getTracer('dfx-api'); + const ctx = job.traceparent ? propagation.extract(ROOT_CONTEXT, { traceparent: job.traceparent }) : ROOT_CONTEXT; + + await tracer.startActiveSpan(`job ${job.group}`, {}, ctx, async (span) => { + try { + span.setAttributes({ + 'job.uid': job.uid, + 'job.group': job.group, + 'job.attempt': job.attempt, + }); + + let handlerError: Error | undefined; + let output: unknown; + + try { + output = await handler.execute(job.inputData); + } catch (e) { + handlerError = e instanceof Error ? e : new Error(String(e)); + } + + if (handlerError) { + span.recordException(handlerError); + span.setStatus({ code: SpanStatusCode.ERROR }); + + try { + if (!job.isFinished) { + await this.jobService.abort(job, handlerError); + } + } catch (e) { + // Persistence failure: abort never landed. Re-throw so callers see an untrustworthy + // data state instead of a terminal job outcome that the database never recorded. + const abortError = e instanceof Error ? e : new Error(String(e)); + this.logger.error(`Failed to persist abort for job ${job.uid}:`, abortError); + throw abortError; + } + + if (job.isFinished) { + const outcome = handlerError instanceof JobDeadLetterException ? 'dead_letter' : 'failed'; + this.metricService.increment('dfx_job_finished', { group: job.group, outcome }); + } + } else { + try { + await this.jobService.finish(job, output); + } catch (e) { + // Persistence failure: the database does not reflect the outcome we computed. Logging + // it as a terminal metric would claim a completion the database never recorded, so it + // is logged and re-thrown. Wait/run/total metrics below are skipped because the data + // is no longer trustworthy from this point. + const error = e instanceof Error ? e : new Error(String(e)); + this.logger.error(`Failed to persist result for job ${job.uid}:`, error); + throw error; + } + + if (job.isFinished) { + this.metricService.increment('dfx_job_finished', { group: job.group, outcome: 'complete' }); + } + } + + const waitSeconds = job.waitSeconds; + if (waitSeconds !== undefined) { + this.metricService.record('dfx_job_wait_seconds', waitSeconds, 's', { group: job.group }); + } + + const runSeconds = job.runSeconds; + if (runSeconds !== undefined) { + this.metricService.record('dfx_job_run_seconds', runSeconds, 's', { group: job.group }); + } + + const totalSeconds = job.totalSeconds; + if (totalSeconds !== undefined) { + this.metricService.record('dfx_job_total_seconds', totalSeconds, 's', { group: job.group }); + } + } finally { + span.end(); + } + }); + } +} diff --git a/src/subdomains/supporting/job/services/job-metric.service.ts b/src/subdomains/supporting/job/services/job-metric.service.ts new file mode 100644 index 0000000000..77ee4fac16 --- /dev/null +++ b/src/subdomains/supporting/job/services/job-metric.service.ts @@ -0,0 +1,127 @@ +import { Injectable, OnModuleInit } from '@nestjs/common'; +import { CronExpression } from '@nestjs/schedule'; +import { MetricService } from 'src/shared/services/metric.service'; +import { Process } from 'src/shared/services/process.service'; +import { DfxCron } from 'src/shared/utils/cron'; +import { Util } from 'src/shared/utils/util'; +import { In, LessThanOrEqual } from 'typeorm'; +import { JobGroup, JobPhase, JobStatus } from '../enums'; +import { JobRepository } from '../repositories/job.repository'; +import { JobService } from './job.service'; + +interface JobGroupSnapshot { + group: JobGroup; + pending: number; + oldestPendingSeconds: number; + overSlaWait: number; + overSlaRun: number; + deadLetter: number; + maxWaitSeconds: number; + maxRunSeconds: number; +} + +const PENDING_STATUSES = [JobStatus.PENDING, JobStatus.RETRY]; + +@Injectable() +export class JobMetricService implements OnModuleInit { + // Queue health is computed from the database, but not queried from the metrics export path: a + // cron writes a snapshot into memory, and the gauge callbacks only read that snapshot. Gauge + // callbacks run on the export cadence; a DB query inside them would tie the metrics export to + // the database's response time. + private snapshots = new Map(); + + constructor( + private readonly jobRepo: JobRepository, + private readonly metricService: MetricService, + private readonly jobService: JobService, + ) {} + + onModuleInit(): void { + this.metricService.registerGauge('dfx_job_pending', '1', (result) => { + for (const snapshot of this.snapshots.values()) { + result.observe(snapshot.pending, { group: snapshot.group }); + } + }); + + this.metricService.registerGauge('dfx_job_oldest_pending_seconds', 's', (result) => { + for (const snapshot of this.snapshots.values()) { + result.observe(snapshot.oldestPendingSeconds, { group: snapshot.group }); + } + }); + + // This gauge is necessary (as opposed to a counter incremented on completion) because it + // reports a job that is CURRENTLY stuck WHILE it is stuck — a completion-time counter could + // never do that, since a stuck job by definition has not completed. + this.metricService.registerGauge('dfx_job_over_sla', '1', (result) => { + for (const snapshot of this.snapshots.values()) { + result.observe(snapshot.overSlaWait, { group: snapshot.group, phase: JobPhase.WAIT }); + result.observe(snapshot.overSlaRun, { group: snapshot.group, phase: JobPhase.RUN }); + } + }); + + this.metricService.registerGauge('dfx_job_sla_seconds', 's', (result) => { + for (const snapshot of this.snapshots.values()) { + result.observe(snapshot.maxWaitSeconds, { group: snapshot.group, phase: JobPhase.WAIT }); + result.observe(snapshot.maxRunSeconds, { group: snapshot.group, phase: JobPhase.RUN }); + } + }); + + this.metricService.registerGauge('dfx_job_dead_letter', '1', (result) => { + for (const snapshot of this.snapshots.values()) { + result.observe(snapshot.deadLetter, { group: snapshot.group }); + } + }); + } + + @DfxCron(CronExpression.EVERY_30_SECONDS, { process: Process.JOB_METRICS, timeout: 300 }) + async updateSnapshot(): Promise { + const next = new Map(); + + for (const group of Object.values(JobGroup)) { + const config = await this.jobService.getConfig(group); + const waitCutoff = Util.secondsBefore(config.maxWaitSeconds); + const runCutoff = Util.secondsBefore(config.maxRunSeconds); + + const pending = await this.jobRepo.countBy({ + group, + status: In(PENDING_STATUSES), + }); + + const oldest = await this.jobRepo.findOne({ + where: { group, status: In(PENDING_STATUSES) }, + order: { created: 'ASC' }, + }); + const oldestPendingSeconds = oldest ? Util.secondsDiff(oldest.created) : 0; + + const overSlaWait = await this.jobRepo.countBy({ + group, + status: In(PENDING_STATUSES), + created: LessThanOrEqual(waitCutoff), + }); + + const overSlaRun = await this.jobRepo.countBy({ + group, + status: JobStatus.PROCESSING, + startedAt: LessThanOrEqual(runCutoff), + }); + + const deadLetter = await this.jobRepo.countBy({ + group, + status: JobStatus.DEAD_LETTER, + }); + + next.set(group, { + group, + pending, + oldestPendingSeconds, + overSlaWait, + overSlaRun, + deadLetter, + maxWaitSeconds: config.maxWaitSeconds, + maxRunSeconds: config.maxRunSeconds, + }); + } + + this.snapshots = next; + } +} diff --git a/src/subdomains/supporting/job/services/job.service.ts b/src/subdomains/supporting/job/services/job.service.ts new file mode 100644 index 0000000000..a2f473f542 --- /dev/null +++ b/src/subdomains/supporting/job/services/job.service.ts @@ -0,0 +1,281 @@ +import { Injectable } from '@nestjs/common'; +import { Config } from 'src/config/config'; +import { SettingService } from 'src/shared/models/setting/setting.service'; +import { DfxLogger } from 'src/shared/services/dfx-logger'; +import { MetricService } from 'src/shared/services/metric.service'; +import { Util } from 'src/shared/utils/util'; +import { UserData } from 'src/subdomains/generic/user/models/user-data/user-data.entity'; +import { In, IsNull, LessThanOrEqual } from 'typeorm'; +import { JobGroupSettingDto } from '../dto/job-group-setting.dto'; +import { JobAttempt } from '../entities/job-attempt.entity'; +import { Job } from '../entities/job.entity'; +import { JobAttemptOutcome, JobGroup, JobStatus } from '../enums'; +import { JobDeadLetterException } from '../exceptions/job-dead-letter.exception'; +import { JobHandler } from '../interfaces/job-handler.interface'; +import { JOB_GROUP_DEFAULTS, JobGroupConfig } from '../job-group.config'; +import { JobAttemptRepository } from '../repositories/job-attempt.repository'; +import { JobRepository } from '../repositories/job.repository'; + +@Injectable() +export class JobService { + private readonly logger = new DfxLogger(JobService); + private readonly handlers = new Map(); + + constructor( + private readonly jobRepo: JobRepository, + private readonly jobAttemptRepo: JobAttemptRepository, + private readonly settingService: SettingService, + private readonly metricService: MetricService, + ) {} + + registerHandler(handler: JobHandler): void { + if (this.handlers.has(handler.group)) { + throw new Error(`Handler for job group ${handler.group} is already registered`); + } + + this.handlers.set(handler.group, handler); + } + + getHandler(group: JobGroup): JobHandler | undefined { + return this.handlers.get(group); + } + + async getConfig(group: JobGroup): Promise { + const settings = await this.settingService.getObjCached('jobGroups'); + const defaults = JOB_GROUP_DEFAULTS[group]; + const override = settings?.find((s) => s.group === group); + + return { + maxWaitSeconds: override?.maxWaitSeconds !== undefined ? override.maxWaitSeconds : defaults.maxWaitSeconds, + maxRunSeconds: override?.maxRunSeconds !== undefined ? override.maxRunSeconds : defaults.maxRunSeconds, + maxAttempts: override?.maxAttempts !== undefined ? override.maxAttempts : defaults.maxAttempts, + exposeResult: override?.exposeResult !== undefined ? override.exposeResult : defaults.exposeResult, + }; + } + + async findByIdempotencyKey(group: JobGroup, idempotencyKey: string): Promise { + return this.jobRepo.findOne({ where: { group, idempotencyKey } }); + } + + async enqueue( + group: JobGroup, + idempotencyKey: string, + input: unknown, + options: { userData?: UserData; traceparent?: string }, + ): Promise { + const existing = await this.findByIdempotencyKey(group, idempotencyKey); + if (existing) return existing; + + const config = await this.getConfig(group); + + const job = this.jobRepo.create({ + uid: Util.createUid(Config.prefixes.jobUidPrefix), + group, + status: JobStatus.PENDING, + idempotencyKey, + attempt: 0, + maxAttempts: config.maxAttempts, + userData: options.userData, + traceparent: options.traceparent, + }); + job.inputData = input; + + try { + const saved = await this.jobRepo.save(job); + this.metricService.increment('dfx_job_enqueued', { group }); + return saved; + } catch (e) { + // Concurrent enqueue of the same group+idempotencyKey: unique index race. + if ((e as { code?: string }).code !== '23505') throw e; + + const concurrent = await this.findByIdempotencyKey(group, idempotencyKey); + if (!concurrent) throw e; + return concurrent; + } + } + + // Conditional claim via status CAS: only one worker can win the update even across instances. + // No time-window heuristic is required because two runners can never both process the same job. + async claimNext(group: JobGroup, owner: string): Promise { + const now = new Date(); + const candidates = await this.jobRepo.find({ + where: [ + { status: In([JobStatus.PENDING, JobStatus.RETRY]), group, nextAttemptAt: IsNull() }, + { status: In([JobStatus.PENDING, JobStatus.RETRY]), group, nextAttemptAt: LessThanOrEqual(now) }, + ], + order: { id: 'ASC' }, + take: 10, + }); + + for (const candidate of candidates) { + const previousStatus = candidate.status; + const [, update] = candidate.claim(owner); + const result = await this.jobRepo.update({ id: candidate.id, status: previousStatus }, update); + + if (result.affected === 1) { + // The claim is a race that exactly one caller wins. Writing the attempt row before the conditional update + // would leave every loser with a row describing an attempt it never actually took. Writing it here, right + // after the win is confirmed, means the row always corresponds to a real attempt. + // + // If this insert fails, the error propagates (no try/catch): the `job` row already carries PROCESSING from + // the update above, and there is no attempt row to prove work started. The only safe outcome is for the + // job to sit in PROCESSING until recoverStale's staleness timeout picks it up again — not to swallow the + // error and continue as if the attempt were properly recorded. + const attemptRow: JobAttempt = this.jobAttemptRepo.create({ + job: candidate, + attempt: candidate.attempt, + claimedBy: candidate.claimedBy, + claimedAt: candidate.claimedAt, + }); + await this.jobAttemptRepo.save(attemptRow); + + return candidate; + } + } + + return undefined; + } + + async recoverStale(group: JobGroup, staleAfterSeconds: number): Promise { + const cutoff = Util.secondsBefore(staleAfterSeconds); + const staleJobs = await this.jobRepo.find({ + where: { + status: JobStatus.PROCESSING, + group, + claimedAt: LessThanOrEqual(cutoff), + }, + }); + + let recovered = 0; + + for (const job of staleJobs) { + const update: Partial = + job.attempt >= job.maxAttempts + ? { + status: JobStatus.FAILED, + finishedAt: new Date(), + error: `Orphaned after restart, attempt budget exhausted (${job.attempt}/${job.maxAttempts})`, + } + : { + status: JobStatus.RETRY, + nextAttemptAt: new Date(), + }; + + // Same SELECT-then-conditional-UPDATE pattern as claimNext: a job that finished between + // the find and this update must not be overwritten. + const result = await this.jobRepo.update( + { id: job.id, status: JobStatus.PROCESSING, claimedAt: LessThanOrEqual(cutoff) }, + update, + ); + + if (result.affected === 1) { + Object.assign(job, update); + // From this attempt's own point of view it failed by orphaning/timeout regardless of whether the job as a + // whole goes on to RETRY or ends up FAILED — there is no RETRY outcome for an individual attempt, only + // whether the job gets another one. + await this.closeAttempt( + job, + JobAttemptOutcome.FAILED, + `Orphaned: claim exceeded ${staleAfterSeconds}s without finishing, recovered by recoverStale`, + ); + if (update.status === JobStatus.FAILED) { + this.logger.error( + `Failed stale job ${job.uid} (group ${job.group}, attempt ${job.attempt}/${job.maxAttempts}): attempt budget exhausted`, + ); + } else { + this.logger.warn(`Recovered stale job ${job.uid} (group ${job.group}, attempt ${job.attempt})`); + } + recovered += 1; + } + } + + return recovered; + } + + async finish(job: Job, output: unknown): Promise { + await this.closeAttempt(job, JobAttemptOutcome.COMPLETE); + + const [, update] = job.complete(output); + await this.updateJobIfOwned(job, update); + } + + async abort(job: Job, error: Error): Promise { + if (error instanceof JobDeadLetterException) { + await this.closeAttempt(job, JobAttemptOutcome.DEAD_LETTER, error.message); + const [, update] = job.deadLetter(error.message); + if (await this.updateJobIfOwned(job, update)) { + this.logger.error(`Job ${job.id} dead-lettered:`, error); + } + return; + } + + // Whichever branch the job itself takes below (another RETRY or a terminal FAILED), this attempt is over, + // so its row is closed here, once, before either branch touches the `job` snapshot. + await this.closeAttempt(job, JobAttemptOutcome.FAILED, error.message); + + if (job.attempt < job.maxAttempts) { + const nextAttemptAt = Util.secondsAfter(this.backoff(job.attempt)); + const [, update] = job.fail(error.message, nextAttemptAt); + await this.updateJobIfOwned(job, update); + return; + } + + const [, update] = job.fail(error.message, undefined); + if (await this.updateJobIfOwned(job, update)) { + this.logger.error(`Job ${job.id} failed after ${job.attempt} attempts:`, error); + } + } + + async getByUid(uid: string): Promise { + return this.jobRepo.findOne({ where: { uid }, relations: { userData: true } }); + } + + // Deterministic backoff (seconds): attempt 1 → 10s, 2 → 30s, 3 → 120s, 4+ → 300s. + private backoff(attempt: number): number { + if (attempt === 1) return 10; + if (attempt === 2) return 30; + if (attempt === 3) return 120; + return 300; + } + + // Writes the job snapshot only if the row still belongs to the attempt this runner claimed. If + // recoverStale has since reassigned the job to another runner (new attempt, new owner), the + // affected count is 0 and this write is silently dropped — overwriting the new owner's state + // here would let the handler run twice for the same job. The lost runner backs off; the new + // owner is left to finish the job on its own. + private async updateJobIfOwned(job: Job, update: Partial): Promise { + const result = await this.jobRepo.update( + { id: job.id, status: JobStatus.PROCESSING, attempt: job.attempt, claimedBy: job.claimedBy }, + update, + ); + + if (result.affected === 0) { + this.logger.warn( + `Job ${job.uid} attempt ${job.attempt} (claimed by ${job.claimedBy}) no longer owns the job row — skipping update`, + ); + return false; + } + + return true; + } + + // Closes the attempt row for the job's current attempt number. Called before every snapshot update on `job` + // that concludes an attempt (finish, abort, recoverStale) — the immutable event must exist before the + // denormalised current-state column changes, per the auditable-mutations rule. + // + // If no row is found, the process crashed between winning the claim and writing the attempt row in + // claimNext. That gap is itself the documented outcome of a crash, not a new failure to raise here — the + // `job` snapshot is still correct and must still receive its update, so this only logs and returns. + private async closeAttempt(job: Job, outcome: JobAttemptOutcome, error?: string): Promise { + const attemptRow = await this.jobAttemptRepo.findOne({ + where: { job: { id: job.id }, attempt: job.attempt, finishedAt: IsNull() }, + }); + + if (!attemptRow) { + this.logger.warn(`No attempt row found for job ${job.uid}, attempt ${job.attempt}`); + return; + } + + await this.jobAttemptRepo.update(attemptRow.id, { finishedAt: new Date(), outcome, error }); + } +} diff --git a/src/subdomains/supporting/supporting.module.ts b/src/subdomains/supporting/supporting.module.ts index 6ead699738..d1e93c08aa 100644 --- a/src/subdomains/supporting/supporting.module.ts +++ b/src/subdomains/supporting/supporting.module.ts @@ -7,6 +7,7 @@ import { DashboardModule } from './dashboard/dashboard.module'; import { DexModule } from './dex/dex.module'; import { FiatOutputModule } from './fiat-output/fiat-output.module'; import { FiatPayInModule } from './fiat-payin/fiat-payin.module'; +import { JobModule } from './job/job.module'; import { LogModule } from './log/log.module'; import { MrosModule } from './mros/mros.module'; import { NotificationModule } from './notification/notification.module'; @@ -25,6 +26,7 @@ import { SupportIssueModule } from './support-issue/support-issue.module'; BankTxModule, DashboardModule, DexModule, + JobModule, LogModule, NotificationModule, PayInModule, @@ -39,6 +41,6 @@ import { SupportIssueModule } from './support-issue/support-issue.module'; ], controllers: [], providers: [], - exports: [], + exports: [JobModule], }) export class SupportingModule {} diff --git a/src/tracing.ts b/src/tracing.ts index 937f4f5fb1..ff58b4bbe4 100644 --- a/src/tracing.ts +++ b/src/tracing.ts @@ -1,17 +1,23 @@ import { SpanKind, SpanStatusCode } from '@opentelemetry/api'; import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node'; +import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http'; import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http'; +import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics'; import { NodeSDK } from '@opentelemetry/sdk-node'; import { BatchSpanProcessor, ReadableSpan, SpanProcessor } from '@opentelemetry/sdk-trace-base'; -// OpenTelemetry tracing for dfx-api. +// OpenTelemetry tracing and metrics for dfx-api. // // This module is imported first in main.ts so the SDK starts before any // instrumented library (http, pg/TypeORM, …) is loaded — otherwise the // auto-instrumentation cannot patch them. It replaces the previous App // Insights setup; the exporter target is supplied exclusively through // OTEL_EXPORTER_OTLP_ENDPOINT (no hardcoded collector address). When the -// variable is unset, tracing is disabled and the app boots unchanged. +// variable is unset, tracing and metrics are disabled and the app boots +// unchanged. +// +// Metrics export every 15000 ms to match the central Prometheus scrape_interval; +// a shorter interval would only produce points that nobody scrapes. // // The exported helpers are pure and unit-tested; startTracing() has the side // effect of registering the global SDK. @@ -74,6 +80,10 @@ export function startTracing(): NodeSDK | undefined { // processor so corrected statuses are what gets exported. The exporter // reads OTEL_EXPORTER_OTLP_ENDPOINT from the environment. spanProcessors: [new ClientErrorSpanProcessor(), new BatchSpanProcessor(new OTLPTraceExporter())], + metricReader: new PeriodicExportingMetricReader({ + exporter: new OTLPMetricExporter(), + exportIntervalMillis: 15000, + }), instrumentations: [ getNodeAutoInstrumentations({ // Filesystem spans are pure noise for an API service.