qinglong/packages/ql3-local-execution/test/dispatch.test.cjs

const assert = require('node:assert/strict');
const fs = require('node:fs');
const os = require('node:os');
const path = require('node:path');
const { test } = require('node:test');
const {
  createLocalExecutionContextRecipe,
  createLocalTaskExecutionRevision,
} = require('@qinglong/runtime-core/local-dispatch');
const {
  migrateLocalSqlitePath,
  openLocalSqliteRuntimeDatabase,
} = require('@qinglong/local-sqlite');
const { LocalExecutionCoordinator } = require('../dist/execution');
const { createLocalProcessDurableHandle } = require('@qinglong/local-process');
const {
  LocalArtifactCapacityUnavailableError,
  LocalDispatchPlanMaterializer,
  LocalFileArtifactAllocator,
  LocalRunDispatcher,
  localArtifactCapacityPolicyForProfile,
} = require('../dist/dispatch');

const RUN_ID = '019f7120-0000-7000-8000-000000000001';
const ATTEMPT_ID = '019f7120-0000-7000-8000-000000000002';

function eventIdFactory() {
  let value = 0;
  return () => `dispatch-event-${++value}`;
}

function processHandle() {
  const identity = {
    platform: 'linux',
    bootId: '11111111-2222-3333-4444-555555555555',
    pid: 4321,
    processGroupId: 4321,
    startTimeTicks: '123456',
  };
  return Object.freeze({
    handleId: 'dispatch-handle',
    pid: 4321,
    durableHandle: createLocalProcessDurableHandle('dispatch-handle', identity),
    startedAtMs: 10,
    completion: Promise.resolve({ exitCode: 0, signal: null }),
  });
}

async function fixture(t) {
  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'ql3-dispatch-'));
  const databasePath = path.join(directory, 'qinglong3.sqlite');
  const artifactRoot = path.join(directory, 'artifacts');
  await migrateLocalSqlitePath({ databasePath, profile: 'edge' });
  const runtime = await openLocalSqliteRuntimeDatabase({
    databasePath,
    profile: 'edge',
  });
  t.after(async () => {
    await runtime.close();
    fs.rmSync(directory, { recursive: true, force: true });
  });
  const recipe = createLocalExecutionContextRecipe({
    environment: [
      { name: 'PUBLIC_VALUE', kind: 'public', value: 'public-value' },
      { name: 'SECRET_VALUE', kind: 'secret', secretRef: 'secret-ref-1' },
    ],
    createdAtMs: 1,
  });
  assert.equal(
    await runtime.localDispatch.appendLocalExecutionContextRecipe(recipe),
    'inserted',
  );
  assert.equal(
    await runtime.localDispatch.appendLocalExecutionContextRecipe(recipe),
    'existing',
  );
  const revision = createLocalTaskExecutionRevision({
    projectId: 'default',
    taskId: 'task-1',
    taskRevision: 'revision-1',
    executorType: 'local_process',
    command: { kind: 'argv', file: '/bin/echo', args: ['hello'] },
    timeoutMs: 1_000,
    contextRef: recipe.contextRef,
    createdAtMs: 1,
  });
  assert.equal(
    await runtime.localDispatch.appendLocalTaskExecutionRevision(revision),
    'inserted',
  );
  assert.equal(
    await runtime.localDispatch.appendLocalTaskExecutionRevision(revision),
    'existing',
  );
  await runtime.runRepository.transaction(async (transaction) => {
    await transaction.insertRun({
      id: RUN_ID,
      projectId: 'default',
      taskId: 'task-1',
      taskRevision: 'revision-1',
      triggerType: 'manual',
      executionOrigin: 'manual',
      executionOwner: 'runtime',
      status: 'queued',
      version: 0,
      eventSequence: 0,
      priority: 10,
      createdAtMs: 1,
      queuedAtMs: 1,
    });
    await transaction.insertAttempt({
      id: ATTEMPT_ID,
      runId: RUN_ID,
      attempt: 1,
      status: 'claimed',
      executorType: 'local_process',
      callbackSequence: 0,
      createdAtMs: 1,
    });
  });
  return { directory, databasePath, artifactRoot, runtime };
}

test('materializes pinned context and atomically admits one real SQLite candidate', async (t) => {
  const value = await fixture(t);
  let launchRequest;
  const completionNotifications = [];
  const execution = new LocalExecutionCoordinator(
    value.runtime.runRepository,
    {
      async start(request) {
        launchRequest = request;
        assert.equal(request.environment.PUBLIC_VALUE, 'public-value');
        assert.equal(request.environment.SECRET_VALUE, 'top-secret');
        assert.equal(request.output.maximumBytes, 4 * 1024 * 1024);
        assert.match(request.output.logArtifactId, /^local-[0-9a-f]{30}$/);
        return processHandle();
      },
    },
    { stop: async () => assert.fail('valid launch must not be stopped') },
    {
      clock: { now: () => 10 },
      createEventId: eventIdFactory(),
      createCallbackToken: () => 'A'.repeat(32),
    },
  );
  const dispatcher = new LocalRunDispatcher(
    value.runtime.localDispatch,
    new LocalDispatchPlanMaterializer(
      value.runtime.localDispatch,
      new LocalFileArtifactAllocator(
        value.artifactRoot,
        localArtifactCapacityPolicyForProfile('edge'),
      ),
      {
        async resolveLocalSecretEnvironment(request) {
          assert.deepEqual(request.secretRefs, ['secret-ref-1']);
          assert.equal(request.candidate.projectId, 'default');
          return ['top-secret'];
        },
      },
    ),
    execution,
    {
      pageSize: 4,
      maxPages: 1,
      onCompletion: (attemptId) => completionNotifications.push(attemptId),
    },
  );

  const result = await dispatcher.dispatchOnce();
  assert.equal(result.status, 'activated');
  assert.equal(result.runId, RUN_ID);
  assert.equal(result.attemptId, ATTEMPT_ID);
  await new Promise((resolve) => setImmediate(resolve));
  assert.deepEqual(completionNotifications, [ATTEMPT_ID]);
  assert.equal(result.stats.candidatesScanned, 1);
  assert.equal(
    (await value.runtime.runRepository.findRunById(RUN_ID)).status,
    'running',
  );
  const attempt = await value.runtime.runRepository.findAttemptById(ATTEMPT_ID);
  assert.equal(attempt.status, 'running');
  assert.equal(attempt.logArtifactId, launchRequest.output.logArtifactId);
  assert.equal(fs.statSync(launchRequest.output.filePath).mode & 0o777, 0o600);
  assert.equal(fs.statSync(launchRequest.output.filePath).size, 0);
  assert.equal(
    fs.readFileSync(value.databasePath).includes(Buffer.from('top-secret')),
    false,
  );
});

test('missing Secret capability returns unavailable before Artifact allocation', async (t) => {
  const value = await fixture(t);
  const dispatcher = new LocalRunDispatcher(
    value.runtime.localDispatch,
    new LocalDispatchPlanMaterializer(
      value.runtime.localDispatch,
      new LocalFileArtifactAllocator(
        value.artifactRoot,
        localArtifactCapacityPolicyForProfile('edge'),
      ),
    ),
    { start: async () => assert.fail('unavailable plan must not activate') },
  );
  const result = await dispatcher.dispatchOnce();
  assert.equal(result.status, 'idle');
  assert.equal(result.reason, 'plans_unavailable');
  assert.equal(result.stats.plansUnavailable, 1);
  assert.equal(fs.existsSync(value.artifactRoot), false);
  assert.equal(
    (await value.runtime.runRepository.findRunById(RUN_ID)).status,
    'queued',
  );
});

test('capacity admission fails before creating an Artifact file', async (t) => {
  const value = await fixture(t);
  const [candidate] = (
    await value.runtime.localDispatch.listLocalDispatchCandidates({ limit: 1 })
  ).candidates;
  const allocator = new LocalFileArtifactAllocator(
    value.artifactRoot,
    localArtifactCapacityPolicyForProfile('edge'),
    { inspect: async () => 0n },
  );
  await assert.rejects(
    allocator.prepare(candidate),
    LocalArtifactCapacityUnavailableError,
  );
  assert.deepEqual(fs.readdirSync(value.artifactRoot), []);
});