qinglong/packages/ql3-runtime-core/test/clusterRunLostRetry.test.cjs

'use strict';

const assert = require('node:assert/strict');
const { test } = require('node:test');

const {
  ClusterRunLostRetryCoordinator,
  ClusterRunLostRetryUnavailableError,
  buildClusterRunLostRetryTransition,
  normalizeClusterRunLostRetryPageCommand,
  normalizeClusterRunLostRetryPageResult,
} = require('../dist/run/clusterRunLostRetry');

function lostRun(status = 'lost') {
  return {
    id: 'run-1',
    projectId: 'project-1',
    taskId: 'task-1',
    taskRevision: 'revision-1',
    triggerType: 'manual',
    executionOrigin: 'manual',
    executionOwner: 'runtime',
    status,
    version: 4,
    eventSequence: 4,
    priority: 0,
    createdAtMs: 100,
    queuedAtMs: 110,
    startedAtMs: 150,
    errorCode: 'CLUSTER_RECOVERY_EXECUTION_NOT_RUNNING',
    errorSummary: 'lost',
  };
}

function lostAttempt(attempt = 1) {
  return {
    id: `attempt-${attempt}`,
    runId: 'run-1',
    attempt,
    status: 'lost',
    executorType: 'remote_worker',
    callbackSequence: 0,
    createdAtMs: 120,
    startedAtMs: 150,
    finishedAtMs: 200,
    errorCode: 'CLUSTER_RECOVERY_EXECUTION_NOT_RUNNING',
    errorSummary: 'lost',
  };
}

function policy(overrides = {}) {
  return {
    runId: 'run-1',
    maxAttempts: 3,
    retryOnLost: true,
    safety: 'idempotent',
    backoffBaseMs: 1_000,
    backoffMaxMs: 8_000,
    version: 0,
    createdAtMs: 100,
    updatedAtMs: 100,
    ...overrides,
  };
}

test('schedules a safe lost Run from its admitted immutable policy', () => {
  const result = buildClusterRunLostRetryTransition({
    run: lostRun(),
    attempt: lostAttempt(),
    policy: policy(),
    observedAtMs: 300,
    runEventId: 'event-1',
  });

  assert.equal(result.disposition, 'scheduled');
  assert.equal(result.runTransitions.length, 1);
  assert.equal(result.runTransitions[0].status, 'retry_wait');
  assert.equal(result.runTransitions[0].version, 5);
  assert.equal(result.policy.nextAttemptAtMs, 1_200);
  assert.equal(result.policy.version, 1);
  assert.deepEqual(
    result.events.map((event) => [event.type, event.sequence]),
    [['run.retry_wait', 5]],
  );
});

test('requeues a due retry_wait Run with one fresh unleased Attempt', () => {
  const result = buildClusterRunLostRetryTransition({
    run: lostRun('retry_wait'),
    attempt: lostAttempt(),
    policy: policy({ nextAttemptAtMs: 250 }),
    observedAtMs: 300,
    runEventId: 'event-queued',
    attemptId: 'attempt-2',
    attemptEventId: 'event-claimed',
  });

  assert.equal(result.disposition, 'requeued');
  assert.deepEqual(
    result.runTransitions.map((run) => [run.status, run.version]),
    [
      ['queued', 5],
      ['queued', 6],
    ],
  );
  assert.deepEqual(result.attempt, {
    id: 'attempt-2',
    runId: 'run-1',
    attempt: 2,
    status: 'claimed',
    executorType: 'remote_worker',
    callbackSequence: 0,
    createdAtMs: 300,
  });
  assert.equal(
    Object.prototype.hasOwnProperty.call(result.policy, 'nextAttemptAtMs'),
    false,
  );
  assert.deepEqual(
    result.events.map((event) => [event.type, event.sequence]),
    [
      ['run.queued', 5],
      ['attempt.claimed', 6],
    ],
  );
});

test('terminalizes disabled, unsafe and exhausted policies without retrying', () => {
  const cases = [
    [null, lostAttempt(), 'failed_disabled', 'RUN_LOST_RETRY_DISABLED'],
    [
      policy({ safety: 'unknown' }),
      lostAttempt(),
      'failed_unsafe',
      'RUN_LOST_RETRY_UNSAFE',
    ],
    [
      policy({ maxAttempts: 3 }),
      lostAttempt(3),
      'failed_exhausted',
      'RUN_LOST_RETRY_EXHAUSTED',
    ],
  ];
  for (const [admittedPolicy, attempt, disposition, errorCode] of cases) {
    const result = buildClusterRunLostRetryTransition({
      run: lostRun(),
      attempt,
      policy: admittedPolicy,
      observedAtMs: 300,
      runEventId: `event-${disposition}`,
    });
    assert.equal(result.disposition, disposition);
    assert.equal(result.runTransitions[0].status, 'failed');
    assert.equal(result.runTransitions[0].errorCode, errorCode);
    assert.equal(result.runTransitions[0].finishedAtMs, 300);
  }
});

test('rejects cancellation, Workflow aggregates and early retry_wait', () => {
  assert.throws(() =>
    buildClusterRunLostRetryTransition({
      run: { ...lostRun(), cancelRequestedAtMs: 250, cancelReason: 'user' },
      attempt: lostAttempt(),
      policy: policy(),
      observedAtMs: 300,
      runEventId: 'event-cancelled',
    }),
  );
  assert.throws(() =>
    buildClusterRunLostRetryTransition({
      run: { ...lostRun(), triggerType: 'plugin_package_workflow' },
      attempt: lostAttempt(),
      policy: policy(),
      observedAtMs: 300,
      runEventId: 'event-workflow',
    }),
  );
  assert.throws(() =>
    buildClusterRunLostRetryTransition({
      run: lostRun('retry_wait'),
      attempt: lostAttempt(),
      policy: policy({ nextAttemptAtMs: 301 }),
      observedAtMs: 300,
      runEventId: 'event-early',
      attemptId: 'attempt-2',
      attemptEventId: 'event-2',
    }),
  );
});

test('normalizes one bounded page and coalesces overlapping callers', async () => {
  assert.deepEqual(normalizeClusterRunLostRetryPageCommand({ limit: 2 }), {
    limit: 2,
  });
  assert.deepEqual(
    normalizeClusterRunLostRetryPageResult(
      {
        scanned: 2,
        scheduled: 1,
        requeued: 0,
        failed: 0,
        raced: 1,
        hasMore: false,
      },
      2,
    ),
    {
      scanned: 2,
      scheduled: 1,
      requeued: 0,
      failed: 0,
      raced: 1,
      hasMore: false,
    },
  );

  let release;
  const coordinator = new ClusterRunLostRetryCoordinator(
    {
      async reconcilePage(command) {
        await new Promise((resolve) => {
          release = resolve;
        });
        return {
          scanned: command.limit,
          scheduled: command.limit,
          requeued: 0,
          failed: 0,
          raced: 0,
          hasMore: false,
        };
      },
    },
    { pageSize: 2 },
  );
  const first = coordinator.reconcile();
  const second = coordinator.reconcile();
  assert.equal(first, second);
  while (!release) await new Promise((resolve) => setImmediate(resolve));
  release();
  assert.equal((await first).scheduled, 2);
});

test('wraps malformed repository results as unavailable', async () => {
  const coordinator = new ClusterRunLostRetryCoordinator({
    async reconcilePage() {
      return {
        scanned: 1,
        scheduled: 1,
        requeued: 1,
        failed: 0,
        raced: 0,
        hasMore: false,
      };
    },
  });
  await assert.rejects(
    coordinator.reconcile(),
    ClusterRunLostRetryUnavailableError,
  );
});