qinglong/packages/ql3-cluster-postgres/test/clusterRecoverySource.test.cjs

const assert = require('node:assert/strict');
const { test } = require('node:test');
const {
  PostgresClusterControlRecoverySource,
} = require('../dist/entrypoints/runtime');

function sourceWith(rows, observations = []) {
  return new PostgresClusterControlRecoverySource({
    async query(text, values) {
      observations.push({ text, values });
      return { rows };
    },
  });
}

test('reads Run and Attempt recovery candidates through one bounded query', async () => {
  const observations = [];
  const source = sourceWith(
    [
      {
        observedAtMs: '100',
        kind: 'run',
        id: 'run-1',
        runId: 'run-1',
        status: 'running',
        createdAtMs: '10',
      },
      {
        observedAtMs: '100',
        kind: 'attempt',
        id: 'attempt-1',
        runId: 'run-1',
        status: 'starting',
        createdAtMs: 11,
      },
      {
        observedAtMs: '100',
        kind: 'run',
        id: 'run-2',
        runId: 'run-2',
        status: 'created',
        createdAtMs: 12,
      },
    ],
    observations,
  );

  assert.deepEqual(await source.listOutstanding(2), {
    observedAtMs: 100,
    candidates: [
      {
        kind: 'run',
        id: 'run-1',
        runId: 'run-1',
        status: 'running',
        createdAtMs: 10,
      },
      {
        kind: 'attempt',
        id: 'attempt-1',
        runId: 'run-1',
        status: 'starting',
        createdAtMs: 11,
      },
    ],
    hasMore: true,
  });
  assert.deepEqual(observations[0].values, [3]);
  assert.match(observations[0].text, /WITH observation AS/);
  assert.match(observations[0].text, /statement_timestamp\(\)/);
  assert.match(observations[0].text, /LIMIT \$1/);
  assert.match(observations[0].text, /execution_owner = 'runtime'/);
  assert.match(
    observations[0].text,
    /trigger_type NOT IN \([\s\S]*'plugin_package_workflow',[\s\S]*'copilot_failure_diagnosis'[\s\S]*\)/,
  );
  assert.match(observations[0].text, /attempt_candidates/);
  assert.match(
    observations[0].text,
    /plugin_package_workflow_task_attempt_admissions/,
  );
  assert.match(
    observations[0].text,
    /attempt_run\.trigger_type = 'plugin_package_workflow'/,
  );
  assert.match(observations[0].text, /workflow_task\.attempt_id = attempt\.id/);
  assert.match(
    observations[0].text,
    /lease_expires_at_ms > observation\.observed_at_ms/,
  );
  assert.match(
    observations[0].text,
    /lease_expires_at_ms <= observation\.observed_at_ms/,
  );
  assert.match(observations[0].text, /INNER JOIN "ql3"\."runs" AS attempt_run/);
  assert.match(observations[0].text, /attempt_run\.status = 'queued'/);
  assert.match(
    observations[0].text,
    /attempt\.executor_type = 'remote_worker'/,
  );
  assert.match(observations[0].text, /attempt\.callback_sequence = 0/);
  assert.match(
    observations[0].text,
    /newer_attempt\.attempt > attempt\.attempt/,
  );
});

test('rejects unbounded page sizes before touching PostgreSQL', async () => {
  let queries = 0;
  const source = sourceWith([], {
    push() {
      queries += 1;
    },
  });
  await assert.rejects(source.listOutstanding(0), /between 1 and 128/);
  await assert.rejects(source.listOutstanding(129), /between 1 and 128/);
  assert.equal(queries, 0);
});

test('fails closed on malformed or terminal PostgreSQL rows', async () => {
  await assert.rejects(
    sourceWith([
      {
        observedAtMs: '100',
        kind: 'run',
        id: 'run-1',
        runId: 'run-1',
        status: 'succeeded',
        createdAtMs: 1,
      },
    ]).listOutstanding(1),
    /kind or status is invalid/,
  );
  await assert.rejects(
    sourceWith([
      {
        observedAtMs: '100',
        kind: 'attempt',
        id: 'attempt-1',
        runId: 'run-1',
        status: 'running',
        createdAtMs: '9007199254740992',
      },
    ]).listOutstanding(1),
    /createdAtMs is invalid/,
  );
});

test('represents an empty page without losing the database observation', async () => {
  const source = sourceWith([
    {
      observedAtMs: '100',
      kind: null,
      id: null,
      runId: null,
      status: null,
      createdAtMs: null,
    },
  ]);
  assert.deepEqual(await source.listOutstanding(1), {
    observedAtMs: 100,
    candidates: [],
    hasMore: false,
  });
});