8d2cc4b3 - Run long commands as jobs, starting with the account merge

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:
 

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"`);
+  }
+};

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"`);
+  }
+};

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",

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);
+  });
 });

src/config/config.ts

@@ -117,6 +117,7 @@ export class Configuration {
     paymentLinkPaymentUidPrefix: 'plp',
     paymentQuoteUidPrefix: 'plq',
     realUnitTransferUidPrefix: 'RT',
+    jobUidPrefix: 'J',
   };
 
   moderators = {

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<string, SettingSchema> = {
 
   // Compliance
   complianceClerks: 'string[]',
+
+  // Job Groups
+  jobGroups: { type: 'array', items: JobGroupSettingDto },
 };
 
 export function isArraySchema(schema: SettingSchema): schema is ArraySchema {

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);
+  });
+});

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<string, Date>();
 
   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());
+      }
     };
   }