mirror of
https://github.com/whyour/qinglong.git
synced 2026-09-21 00:36:13 +08:00
316 lines
9.9 KiB
JavaScript
316 lines
9.9 KiB
JavaScript
const assert = require('node:assert/strict');
|
|
const { DatabaseSync } = require('node:sqlite');
|
|
const { test } = require('node:test');
|
|
|
|
const {
|
|
createInitialPluginPackageAutomationPublication,
|
|
} = require('@qinglong/runtime-core/plugin-package-automation-publication');
|
|
const {
|
|
createPluginPackageWorkflowExecutionPlan,
|
|
} = require('@qinglong/runtime-core/plugin-package-workflow-execution-plan');
|
|
const {
|
|
InvalidPluginPackageWorkflowFrontierError,
|
|
PluginPackageWorkflowFrontierUnavailableError,
|
|
} = require('@qinglong/runtime-core/plugin-package-workflow-frontier');
|
|
const {
|
|
transitionStepRunMutation,
|
|
} = require('@qinglong/runtime-core/step-run');
|
|
const {
|
|
activateInstall,
|
|
pluginPackageTaskReconciliationFixture,
|
|
} = require('../../../test/contracts/pluginPackageTaskReconciliationRepositoryContract.cjs');
|
|
const { migrateLocalSqliteDatabase } = require('../dist/migration/migration');
|
|
const {
|
|
LocalSqlitePluginPackageAutomationPublicationRepository,
|
|
} = require('../dist/plugin-package/pluginPackageAutomationPublicationRepository');
|
|
const {
|
|
LocalSqlitePluginPackageInstallRepository,
|
|
} = require('../dist/plugin-package/pluginPackageInstallRepository');
|
|
const {
|
|
LocalSqlitePluginPackageMaterializedRevisionRepository,
|
|
} = require('../dist/plugin-package/pluginPackageMaterializedRevisionRepository');
|
|
const {
|
|
LocalSqlitePluginPackageWorkflowAdmissionRepository,
|
|
} = require('../dist/plugin-package/workflow/pluginPackageWorkflowAdmissionRepository');
|
|
const {
|
|
LocalSqlitePluginPackageWorkflowFrontierRepository,
|
|
} = require('../dist/plugin-package/workflow/pluginPackageWorkflowFrontierRepository');
|
|
const {
|
|
LocalSqliteStepRunRepository,
|
|
} = require('../dist/run/stepRunRepository');
|
|
|
|
function fixture(namespace) {
|
|
const value = pluginPackageTaskReconciliationFixture(namespace, {
|
|
workflows: [
|
|
{
|
|
schema: 'qinglong/plugin-package-workflow-resource@v1',
|
|
id: 'daily',
|
|
name: 'Daily workflow',
|
|
enabled: true,
|
|
steps: [
|
|
{ id: 'collect', task: 'alpha', needs: [] },
|
|
{ id: 'summarize', task: 'beta', needs: ['collect'] },
|
|
],
|
|
},
|
|
],
|
|
});
|
|
const publication = createInitialPluginPackageAutomationPublication(
|
|
value.revision,
|
|
value.registry,
|
|
2_000,
|
|
);
|
|
const plan = createPluginPackageWorkflowExecutionPlan({
|
|
planId: `wf-plan-${namespace}`,
|
|
runId: `wf-run-${namespace}`,
|
|
workflowId: 'daily',
|
|
stepRunIds: {
|
|
collect: `wf-collect-${namespace}`,
|
|
summarize: `wf-summary-${namespace}`,
|
|
},
|
|
publication,
|
|
revision: value.revision,
|
|
taskSpecSemanticRegistry: value.registry,
|
|
plannedAtMs: 3_000,
|
|
});
|
|
return { ...value, publication, plan };
|
|
}
|
|
|
|
async function harness(t, namespace) {
|
|
const value = fixture(namespace);
|
|
const client = new DatabaseSync(':memory:');
|
|
client.exec('PRAGMA foreign_keys = ON');
|
|
await migrateLocalSqliteDatabase(client);
|
|
client
|
|
.prepare(
|
|
`INSERT INTO "QingLong3Projects"
|
|
(id, name, slug, status, version, created_at_ms, updated_at_ms)
|
|
VALUES (?, ?, ?, 'active', 1, 1, 1)`,
|
|
)
|
|
.run(value.projectId, value.projectId, value.projectId);
|
|
await activateInstall(
|
|
new LocalSqlitePluginPackageInstallRepository(client),
|
|
value,
|
|
);
|
|
await new LocalSqlitePluginPackageMaterializedRevisionRepository(
|
|
client,
|
|
value.registry,
|
|
).publish(value.revision);
|
|
await new LocalSqlitePluginPackageAutomationPublicationRepository(
|
|
client,
|
|
).publish(value.publication);
|
|
const admission =
|
|
new LocalSqlitePluginPackageWorkflowAdmissionRepository(client);
|
|
const admitted = await admission.admit(value.plan);
|
|
t.after(() => client.close());
|
|
return {
|
|
client,
|
|
value,
|
|
admission,
|
|
admitted,
|
|
frontier:
|
|
new LocalSqlitePluginPackageWorkflowFrontierRepository(client),
|
|
stepRuns: new LocalSqliteStepRunRepository(client),
|
|
};
|
|
}
|
|
|
|
async function transitionStep(
|
|
harnessValue,
|
|
stepKey,
|
|
to,
|
|
runVersion,
|
|
atMs,
|
|
) {
|
|
const current = await harnessValue.stepRuns.findByRunAndStepKey(
|
|
harnessValue.value.plan.runId,
|
|
stepKey,
|
|
);
|
|
assert.ok(current);
|
|
const mutation = transitionStepRunMutation(
|
|
current,
|
|
{
|
|
expectedVersion: current.version,
|
|
expectedDigest: current.stepRunDigest,
|
|
mutationId: `${stepKey}-${to}-${current.version}`,
|
|
to,
|
|
atMs,
|
|
...(to === 'failed' ? { resultCode: 'task_failed' } : {}),
|
|
},
|
|
{
|
|
expectedRunVersion: runVersion,
|
|
expectedRunEventSequence: runVersion,
|
|
eventId: `${stepKey}-${to}-event-${current.version}`,
|
|
dedupeKey: `${stepKey}-${to}-event-${current.version}`,
|
|
actor: { type: 'executor' },
|
|
},
|
|
);
|
|
return harnessValue.stepRuns.apply(mutation);
|
|
}
|
|
|
|
test('advances only an actionable dependency and terminalizes exactly once', async (t) => {
|
|
const value = await harness(t, 'frontier-ok');
|
|
assert.deepEqual(await value.frontier.listCandidates({ limit: 8 }), {
|
|
candidates: [],
|
|
truncated: false,
|
|
});
|
|
|
|
await transitionStep(value, 'collect', 'running', 3, 4_000);
|
|
await transitionStep(value, 'collect', 'succeeded', 4, 5_000);
|
|
const actionable = await value.frontier.listCandidates({ limit: 8 });
|
|
assert.deepEqual(actionable, {
|
|
candidates: [
|
|
{
|
|
runId: value.value.plan.runId,
|
|
planDigest: value.value.plan.planDigest,
|
|
admittedAtMs: 3_000,
|
|
},
|
|
],
|
|
truncated: false,
|
|
});
|
|
const advanced = await value.frontier.advance(value.value.plan.runId);
|
|
assert.equal(advanced.status, 'advanced');
|
|
assert.equal(advanced.stepMutationCount, 1);
|
|
assert.deepEqual(advanced.readyStepRunIds, [
|
|
value.value.plan.steps.find(({ stepKey }) => stepKey === 'summarize')
|
|
.stepRunId,
|
|
]);
|
|
assert.equal(advanced.runVersion, 6);
|
|
assert.deepEqual(await value.frontier.listCandidates({ limit: 8 }), {
|
|
candidates: [],
|
|
truncated: false,
|
|
});
|
|
|
|
await transitionStep(
|
|
value,
|
|
'summarize',
|
|
'running',
|
|
6,
|
|
advanced.observedAtMs + 1,
|
|
);
|
|
await transitionStep(
|
|
value,
|
|
'summarize',
|
|
'succeeded',
|
|
7,
|
|
advanced.observedAtMs + 2,
|
|
);
|
|
assert.equal(
|
|
(await value.frontier.listCandidates({ limit: 8 })).candidates.length,
|
|
1,
|
|
);
|
|
const terminal = await value.frontier.advance(value.value.plan.runId);
|
|
assert.equal(terminal.status, 'terminal');
|
|
assert.equal(terminal.terminalStatus, 'succeeded');
|
|
assert.equal(terminal.runVersion, 9);
|
|
assert.deepEqual(
|
|
{
|
|
...value.client
|
|
.prepare(
|
|
`SELECT status, version, event_sequence AS "eventSequence",
|
|
finished_at_ms IS NOT NULL AS finished,
|
|
error_code AS "errorCode"
|
|
FROM "Runs" WHERE id = ?`,
|
|
)
|
|
.get(value.value.plan.runId),
|
|
},
|
|
{
|
|
status: 'succeeded',
|
|
version: 9,
|
|
eventSequence: 9,
|
|
finished: 1,
|
|
errorCode: null,
|
|
},
|
|
);
|
|
const replay = await value.frontier.advance(value.value.plan.runId);
|
|
assert.equal(replay.status, 'settled');
|
|
assert.equal(replay.terminalStatus, 'succeeded');
|
|
assert.deepEqual(await value.admission.admit(value.value.plan), {
|
|
status: 'existing',
|
|
receipt: value.admitted.receipt,
|
|
});
|
|
});
|
|
|
|
test('skips downstream work and fails the aggregate atomically', async (t) => {
|
|
const value = await harness(t, 'frontier-fail');
|
|
await transitionStep(value, 'collect', 'running', 3, 4_000);
|
|
await transitionStep(value, 'collect', 'failed', 4, 5_000);
|
|
const result = await value.frontier.advance(value.value.plan.runId);
|
|
assert.equal(result.status, 'terminal');
|
|
assert.equal(result.stepMutationCount, 1);
|
|
assert.equal(result.terminalStatus, 'failed');
|
|
assert.equal(result.runVersion, 7);
|
|
assert.deepEqual(
|
|
{
|
|
...value.client
|
|
.prepare(
|
|
`SELECT run.status, run.version,
|
|
run.event_sequence AS "eventSequence",
|
|
run.error_code AS "errorCode",
|
|
step.status AS "stepStatus",
|
|
step.result_code AS "resultCode",
|
|
(SELECT COUNT(*) FROM "RunEvents"
|
|
WHERE run_id = run.id) AS events
|
|
FROM "Runs" AS run
|
|
JOIN "StepRuns" AS step
|
|
ON step.run_id = run.id AND step.step_key = 'summarize'
|
|
WHERE run.id = ?`,
|
|
)
|
|
.get(value.value.plan.runId),
|
|
},
|
|
{
|
|
status: 'failed',
|
|
version: 7,
|
|
eventSequence: 7,
|
|
errorCode: 'workflow_step_failed',
|
|
stepStatus: 'skipped',
|
|
resultCode: 'dependency_not_succeeded',
|
|
events: 7,
|
|
},
|
|
);
|
|
});
|
|
|
|
test('bounds paging before SQL and fails closed on current StepRun drift', async (t) => {
|
|
const value = await harness(t, 'frontier-drift');
|
|
assert.throws(
|
|
() => value.frontier.listCandidates({ limit: 65 }),
|
|
InvalidPluginPackageWorkflowFrontierError,
|
|
);
|
|
await transitionStep(value, 'collect', 'running', 3, 4_000);
|
|
await transitionStep(value, 'collect', 'succeeded', 4, 5_000);
|
|
value.client.exec('PRAGMA ignore_check_constraints = ON');
|
|
value.client
|
|
.prepare(
|
|
`UPDATE "StepRuns"
|
|
SET step_run_json = json_set(step_run_json, '$.definitionRef', 'drift')
|
|
WHERE run_id = ? AND step_key = 'summarize'`,
|
|
)
|
|
.run(value.value.plan.runId);
|
|
await assert.rejects(
|
|
value.frontier.advance(value.value.plan.runId),
|
|
PluginPackageWorkflowFrontierUnavailableError,
|
|
);
|
|
assert.deepEqual(
|
|
{
|
|
...value.client
|
|
.prepare(
|
|
`SELECT status, version, event_sequence AS "eventSequence"
|
|
FROM "Runs" WHERE id = ?`,
|
|
)
|
|
.get(value.value.plan.runId),
|
|
},
|
|
{ status: 'running', version: 5, eventSequence: 5 },
|
|
);
|
|
});
|
|
|
|
test('publishes Workflow frontier only through its explicit SQLite subpath', () => {
|
|
const subpath = require('@qinglong/local-sqlite/plugin-package-workflow-frontier');
|
|
const root = require('../dist');
|
|
assert.equal(
|
|
subpath.LocalSqlitePluginPackageWorkflowFrontierRepository,
|
|
LocalSqlitePluginPackageWorkflowFrontierRepository,
|
|
);
|
|
assert.equal(
|
|
root.LocalSqlitePluginPackageWorkflowFrontierRepository,
|
|
undefined,
|
|
);
|
|
});
|