mirror of
https://github.com/whyour/qinglong.git
synced 2026-09-20 16:07:11 +08:00
502 lines
15 KiB
JavaScript
502 lines
15 KiB
JavaScript
const assert = require('node:assert/strict');
|
|
const { test } = require('node:test');
|
|
|
|
const {
|
|
compileClusterCommandTaskDefinition,
|
|
} = require('@qinglong/runtime-core/cluster-execution-revision');
|
|
const {
|
|
createInitialPluginPackageAutomationPublication,
|
|
} = require('@qinglong/runtime-core/plugin-package-automation-publication');
|
|
const {
|
|
planPluginPackageTaskReconciliation,
|
|
pluginPackageTaskReconciliationTaskIds,
|
|
} = require('@qinglong/runtime-core/plugin-package-task-reconciliation');
|
|
const {
|
|
createPluginPackageWorkflowAdmissionBundle,
|
|
createPluginPackageWorkflowExecutionPlan,
|
|
} = require('@qinglong/runtime-core/plugin-package-workflow-execution-plan');
|
|
const {
|
|
InvalidPluginPackageWorkflowTaskAttemptAdmissionError,
|
|
PluginPackageWorkflowTaskAttemptAdmissionConflictError,
|
|
} = require('@qinglong/runtime-core/plugin-package-workflow-task-attempt-admission');
|
|
const {
|
|
pluginPackageTaskReconciliationFixture,
|
|
} = require('../../../test/contracts/pluginPackageTaskReconciliationRepositoryContract.cjs');
|
|
const {
|
|
PostgresPluginPackageWorkflowTaskAttemptAdmissionRepository,
|
|
} = require('../dist/plugin-package/workflow/pluginPackageWorkflowTaskAttemptAdmissionRepository');
|
|
|
|
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-attempt-plan-${namespace}`,
|
|
runId: `wf-attempt-run-${namespace}`,
|
|
workflowId: 'daily',
|
|
stepRunIds: {
|
|
collect: `wf-attempt-collect-${namespace}`,
|
|
summarize: `wf-attempt-summary-${namespace}`,
|
|
},
|
|
publication,
|
|
revision: value.revision,
|
|
taskSpecSemanticRegistry: value.registry,
|
|
plannedAtMs: 3_000,
|
|
});
|
|
const admission = createPluginPackageWorkflowAdmissionBundle(plan);
|
|
const reconciliationPlan = planPluginPackageTaskReconciliation({
|
|
revision: value.revision,
|
|
previousReceipt: null,
|
|
facts: pluginPackageTaskReconciliationTaskIds(
|
|
value.revision,
|
|
null,
|
|
value.registry,
|
|
).map((taskId) => ({
|
|
taskId,
|
|
packageName: null,
|
|
current: null,
|
|
})),
|
|
committedAtMs: 2_500,
|
|
taskSpecSemanticRegistry: value.registry,
|
|
});
|
|
const taskDefinition = reconciliationPlan.writes.find(
|
|
({ definition }) =>
|
|
definition.taskId === `pkg:${value.packageName}:alpha`,
|
|
).definition;
|
|
const execution = compileClusterCommandTaskDefinition(
|
|
taskDefinition,
|
|
value.registry,
|
|
);
|
|
const stepRun = admission.stepMutations.find(
|
|
({ stepRun: candidate }) => candidate.stepKey === 'collect',
|
|
).stepRun;
|
|
return {
|
|
...value,
|
|
plan,
|
|
admission,
|
|
reconciliation: reconciliationPlan.receipt,
|
|
execution,
|
|
stepRun,
|
|
};
|
|
}
|
|
|
|
function runRow(run) {
|
|
return {
|
|
id: run.id,
|
|
projectId: run.projectId,
|
|
taskId: run.taskId,
|
|
taskRevision: run.taskRevision,
|
|
taskName: run.taskName ?? null,
|
|
taskSnapshotRef: run.taskSnapshotRef ?? null,
|
|
legacyCronId: run.legacyCronId ?? null,
|
|
parentRunId: run.parentRunId ?? null,
|
|
retryOfRunId: run.retryOfRunId ?? null,
|
|
triggerId: run.triggerId ?? null,
|
|
triggerType: run.triggerType,
|
|
executionOrigin: run.executionOrigin,
|
|
executionOwner: run.executionOwner,
|
|
triggeredBy: run.triggeredBy ?? null,
|
|
requestId: run.requestId ?? null,
|
|
scheduledForMs: run.scheduledForMs ?? null,
|
|
status: run.status,
|
|
version: run.version,
|
|
eventSequence: run.eventSequence,
|
|
priority: run.priority,
|
|
idempotencyKey: run.idempotencyKey ?? null,
|
|
inputRef: run.inputRef ?? null,
|
|
outputRef: run.outputRef ?? null,
|
|
createdAtMs: run.createdAtMs,
|
|
queuedAtMs: run.queuedAtMs ?? null,
|
|
startedAtMs: run.startedAtMs ?? null,
|
|
finishedAtMs: run.finishedAtMs ?? null,
|
|
cancelRequestedAtMs: run.cancelRequestedAtMs ?? null,
|
|
cancelReason: run.cancelReason ?? null,
|
|
errorCode: run.errorCode ?? null,
|
|
errorSummary: run.errorSummary ?? null,
|
|
};
|
|
}
|
|
|
|
function stepRunRow(stepRun) {
|
|
return {
|
|
id: stepRun.id,
|
|
runId: stepRun.runId,
|
|
parentStepRunId: stepRun.parentStepRunId ?? null,
|
|
stepKey: stepRun.stepKey,
|
|
kind: stepRun.kind,
|
|
definitionRef: stepRun.definitionRef,
|
|
definitionDigest: stepRun.definitionDigest,
|
|
required: stepRun.required,
|
|
status: stepRun.status,
|
|
version: stepRun.version,
|
|
attemptCount: stepRun.attemptCount,
|
|
inputRef: stepRun.inputRef ?? null,
|
|
outputRef: stepRun.outputRef ?? null,
|
|
approvalRequestId: stepRun.approvalRequestId ?? null,
|
|
readyAtMs: stepRun.readyAtMs ?? null,
|
|
startedAtMs: stepRun.startedAtMs ?? null,
|
|
finishedAtMs: stepRun.finishedAtMs ?? null,
|
|
resultCode: stepRun.resultCode ?? null,
|
|
errorSummary: stepRun.errorSummary ?? null,
|
|
createdAtMs: stepRun.createdAtMs,
|
|
updatedAtMs: stepRun.updatedAtMs,
|
|
lastMutationId: stepRun.lastMutationId,
|
|
stepRunDigest: stepRun.stepRunDigest,
|
|
stepRunJson: stepRun,
|
|
};
|
|
}
|
|
|
|
function snapshotRow(value) {
|
|
const execution = value.execution;
|
|
return {
|
|
planJson: value.plan,
|
|
reconciliationJson: value.reconciliation,
|
|
executionProjectId: execution.projectId,
|
|
executionTaskId: execution.taskId,
|
|
executionSourceRevision: execution.sourceRevision,
|
|
executionTaskRevision: execution.taskRevision,
|
|
executionSourceContentDigest: execution.sourceContentDigest,
|
|
executionExecutorType: execution.executorType,
|
|
executionPlanSchema: execution.planSchema,
|
|
executionPlanJson: {
|
|
command: execution.command,
|
|
environment: execution.environment,
|
|
...(execution.workingDirectory === undefined
|
|
? {}
|
|
: { workingDirectory: execution.workingDirectory }),
|
|
...(execution.timeoutMs === undefined
|
|
? {}
|
|
: { timeoutMs: execution.timeoutMs }),
|
|
...(execution.placement === undefined
|
|
? {}
|
|
: { placement: execution.placement }),
|
|
},
|
|
executionContentDigest: execution.contentDigest,
|
|
executionCreatedAtMs: execution.createdAtMs,
|
|
};
|
|
}
|
|
|
|
function mockPool(value, options = {}) {
|
|
const queries = [];
|
|
let connections = 0;
|
|
let releases = 0;
|
|
return {
|
|
pool: {
|
|
async connect() {
|
|
connections += 1;
|
|
const connection = connections;
|
|
let retryFailure =
|
|
connection <= (options.serializationFailures ?? 0);
|
|
return {
|
|
async query(text, values) {
|
|
queries.push({ connection, text, values });
|
|
if (
|
|
retryFailure &&
|
|
text.includes(
|
|
'plugin_package_workflow_task_attempt_snapshot',
|
|
)
|
|
) {
|
|
retryFailure = false;
|
|
throw Object.assign(new Error('serialization failure'), {
|
|
code: '40001',
|
|
});
|
|
}
|
|
if (
|
|
text.includes('FROM "ql3"."step_runs" AS current')
|
|
) {
|
|
return {
|
|
rows: options.candidateRows ?? [
|
|
{
|
|
runId: value.plan.runId,
|
|
stepRunId: value.stepRun.id,
|
|
readyAtMs: value.stepRun.readyAtMs,
|
|
planDigest: value.plan.planDigest,
|
|
},
|
|
],
|
|
};
|
|
}
|
|
if (text.includes('FROM "ql3"."runs"')) {
|
|
return {
|
|
rows: [
|
|
runRow(options.run ?? value.admission.run),
|
|
],
|
|
};
|
|
}
|
|
if (
|
|
text.includes('FROM "ql3"."step_runs"') &&
|
|
text.includes('FOR UPDATE')
|
|
) {
|
|
return { rows: [stepRunRow(value.stepRun)] };
|
|
}
|
|
if (
|
|
text.includes(
|
|
'plugin_package_workflow_task_attempt_admissions',
|
|
) &&
|
|
text.includes('receipt_json') &&
|
|
text.includes('WHERE run_id')
|
|
) {
|
|
return {
|
|
rows: options.existingReceipt
|
|
? [
|
|
{
|
|
receiptDigest:
|
|
options.existingReceipt.receiptDigest,
|
|
receiptJson: options.existingReceipt,
|
|
},
|
|
]
|
|
: [],
|
|
};
|
|
}
|
|
if (
|
|
text.includes(
|
|
'plugin_package_workflow_task_attempt_snapshot',
|
|
)
|
|
) {
|
|
return {
|
|
rows: options.snapshotMissing
|
|
? []
|
|
: [snapshotRow(value)],
|
|
};
|
|
}
|
|
if (text.includes('transaction_timestamp()')) {
|
|
return {
|
|
rows: [{ admittedAtMs: options.admittedAtMs ?? 4_000 }],
|
|
};
|
|
}
|
|
if (text.includes('MAX(attempt)')) {
|
|
return {
|
|
rows: [
|
|
{ attemptNumber: options.attemptNumber ?? 1 },
|
|
],
|
|
};
|
|
}
|
|
return { rows: [], rowCount: 1 };
|
|
},
|
|
release() {
|
|
releases += 1;
|
|
queries.push({ connection, text: 'RELEASE' });
|
|
},
|
|
};
|
|
},
|
|
},
|
|
queries,
|
|
get connections() {
|
|
return connections;
|
|
},
|
|
get releases() {
|
|
return releases;
|
|
},
|
|
};
|
|
}
|
|
|
|
test('bounds and keyset-pages only ready uncancelled Task attempts', async () => {
|
|
const value = fixture('pg-wta-page');
|
|
const db = mockPool(value, {
|
|
candidateRows: [
|
|
{
|
|
runId: value.plan.runId,
|
|
stepRunId: value.stepRun.id,
|
|
readyAtMs: value.stepRun.readyAtMs,
|
|
planDigest: value.plan.planDigest,
|
|
},
|
|
{
|
|
runId: 'wf-attempt-run-next',
|
|
stepRunId: 'wf-attempt-step-next',
|
|
readyAtMs: value.stepRun.readyAtMs + 1,
|
|
planDigest: 'f'.repeat(64),
|
|
},
|
|
],
|
|
});
|
|
const repository =
|
|
new PostgresPluginPackageWorkflowTaskAttemptAdmissionRepository(
|
|
db.pool,
|
|
);
|
|
|
|
assert.deepEqual(
|
|
await repository.listCandidates({
|
|
limit: 1,
|
|
after: {
|
|
readyAtMs: value.stepRun.readyAtMs - 1,
|
|
stepRunId: 'previous-step',
|
|
},
|
|
}),
|
|
{
|
|
candidates: [
|
|
{
|
|
runId: value.plan.runId,
|
|
stepRunId: value.stepRun.id,
|
|
readyAtMs: value.stepRun.readyAtMs,
|
|
planDigest: value.plan.planDigest,
|
|
},
|
|
],
|
|
truncated: true,
|
|
next: {
|
|
readyAtMs: value.stepRun.readyAtMs,
|
|
stepRunId: value.stepRun.id,
|
|
},
|
|
},
|
|
);
|
|
const query = db.queries[0];
|
|
assert.deepEqual(query.values, [
|
|
'previous-step',
|
|
value.stepRun.readyAtMs - 1,
|
|
2,
|
|
]);
|
|
assert.match(query.text, /run\.cancel_requested_at_ms IS NULL/);
|
|
assert.match(query.text, /current\.status = 'ready'/);
|
|
assert.match(query.text, /NOT EXISTS/);
|
|
assert.match(query.text, /LIMIT \$3/);
|
|
await assert.rejects(
|
|
repository.listCandidates({ limit: 65 }),
|
|
InvalidPluginPackageWorkflowTaskAttemptAdmissionError,
|
|
);
|
|
});
|
|
|
|
test('atomically admits the exact cluster execution and replays the epoch', async () => {
|
|
const value = fixture('pg-wta-create');
|
|
const db = mockPool(value);
|
|
const repository =
|
|
new PostgresPluginPackageWorkflowTaskAttemptAdmissionRepository(
|
|
db.pool,
|
|
);
|
|
|
|
const created = await repository.admit(
|
|
value.plan.runId,
|
|
value.stepRun.id,
|
|
);
|
|
assert.equal(created.status, 'created');
|
|
assert.equal(created.receipt.resourceTaskId, 'alpha');
|
|
assert.equal(
|
|
created.receipt.taskId,
|
|
`pkg:${value.packageName}:alpha`,
|
|
);
|
|
assert.equal(created.receipt.executorType, 'remote_worker');
|
|
assert.equal(created.receipt.executionDigest, value.execution.contentDigest);
|
|
assert.equal(created.receipt.attemptNumber, 1);
|
|
assert.equal(created.receipt.runVersion, value.admission.run.version + 1);
|
|
|
|
const sql = db.queries.map(({ text }) => text);
|
|
assert.equal(sql[0], 'BEGIN ISOLATION LEVEL SERIALIZABLE');
|
|
assert.match(
|
|
sql.find((text) => text.includes('FROM "ql3"."runs"')),
|
|
/FOR UPDATE/,
|
|
);
|
|
assert.match(
|
|
sql.find(
|
|
(text) =>
|
|
text.includes('FROM "ql3"."step_runs"') &&
|
|
text.includes('FOR UPDATE'),
|
|
),
|
|
/FOR UPDATE/,
|
|
);
|
|
assert.ok(
|
|
sql.some((text) =>
|
|
text.includes('plugin_package_workflow_task_attempt_snapshot'),
|
|
),
|
|
);
|
|
assert.equal(
|
|
sql.filter((text) => text.includes('UPDATE "ql3"."runs"')).length,
|
|
1,
|
|
);
|
|
assert.ok(
|
|
sql.some((text) => text.includes('INSERT INTO "ql3"."run_attempts"')),
|
|
);
|
|
assert.ok(
|
|
sql.some((text) => text.includes('INSERT INTO "ql3"."run_events"')),
|
|
);
|
|
assert.ok(
|
|
sql.some((text) =>
|
|
text.includes(
|
|
'"ql3"."plugin_package_workflow_task_attempt_admissions"',
|
|
),
|
|
),
|
|
);
|
|
assert.equal(sql.at(-2), 'COMMIT');
|
|
assert.equal(sql.at(-1), 'RELEASE');
|
|
|
|
const replayDb = mockPool(value, {
|
|
existingReceipt: created.receipt,
|
|
run: {
|
|
...value.admission.run,
|
|
version: created.receipt.runVersion,
|
|
eventSequence: created.receipt.runEventSequence,
|
|
},
|
|
});
|
|
const replay =
|
|
new PostgresPluginPackageWorkflowTaskAttemptAdmissionRepository(
|
|
replayDb.pool,
|
|
);
|
|
assert.deepEqual(
|
|
await replay.admit(value.plan.runId, value.stepRun.id),
|
|
{ status: 'existing', receipt: created.receipt },
|
|
);
|
|
assert.equal(
|
|
replayDb.queries.some(({ text }) =>
|
|
text.includes('plugin_package_workflow_task_attempt_snapshot'),
|
|
),
|
|
false,
|
|
);
|
|
});
|
|
|
|
test('retries serialization once and fails closed without a snapshot', async () => {
|
|
const value = fixture('pg-wta-retry');
|
|
const retried = mockPool(value, { serializationFailures: 1 });
|
|
const repository =
|
|
new PostgresPluginPackageWorkflowTaskAttemptAdmissionRepository(
|
|
retried.pool,
|
|
);
|
|
assert.equal(
|
|
(
|
|
await repository.admit(value.plan.runId, value.stepRun.id)
|
|
).status,
|
|
'created',
|
|
);
|
|
assert.equal(retried.connections, 2);
|
|
assert.equal(retried.releases, 2);
|
|
assert.ok(retried.queries.some(({ text }) => text === 'ROLLBACK'));
|
|
|
|
const missing = mockPool(value, { snapshotMissing: true });
|
|
await assert.rejects(
|
|
new PostgresPluginPackageWorkflowTaskAttemptAdmissionRepository(
|
|
missing.pool,
|
|
).admit(value.plan.runId, value.stepRun.id),
|
|
PluginPackageWorkflowTaskAttemptAdmissionConflictError,
|
|
);
|
|
assert.ok(missing.queries.some(({ text }) => text === 'ROLLBACK'));
|
|
assert.equal(
|
|
missing.queries.some(({ text }) =>
|
|
text.includes('INSERT INTO "ql3"."run_attempts"'),
|
|
),
|
|
false,
|
|
);
|
|
});
|
|
|
|
test('is available only through the explicit package subpath', () => {
|
|
assert.match(
|
|
require.resolve(
|
|
'@qinglong/cluster-postgres/plugin-package-workflow-task-attempt-admission',
|
|
),
|
|
/pluginPackageWorkflowTaskAttemptAdmissionRepository\.js$/,
|
|
);
|
|
const root = require('@qinglong/cluster-postgres');
|
|
assert.equal(
|
|
root.PostgresPluginPackageWorkflowTaskAttemptAdmissionRepository,
|
|
undefined,
|
|
);
|
|
});
|