mirror of
https://github.com/whyour/qinglong.git
synced 2026-09-21 00:17:47 +08:00
893 lines
25 KiB
JavaScript
893 lines
25 KiB
JavaScript
require('ts-node/register/transpile-only');
|
|
|
|
const assert = require('node:assert/strict');
|
|
const { afterEach, test } = require('node:test');
|
|
const { DataTypes, QueryTypes, Sequelize } = require('sequelize');
|
|
const {
|
|
defineSchemaMigrationModel,
|
|
} = require('../../back/data/schemaMigration');
|
|
const {
|
|
legacyColumnsMigration,
|
|
} = require('../../back/migrations/0001-legacy-columns');
|
|
const {
|
|
RUN_ATTEMPT_TABLE,
|
|
RUN_EVENT_TABLE,
|
|
RUN_TABLE,
|
|
runSchemaMigration,
|
|
} = require('../../back/migrations/0002-run-schema');
|
|
const {
|
|
RUNNING_INSTANCE_ATTEMPT_INDEX,
|
|
RUNNING_INSTANCE_RUN_INDEX,
|
|
RUNNING_INSTANCE_TABLE,
|
|
runningInstanceRunReferenceMigration,
|
|
} = require('../../back/migrations/0003-running-instance-run-reference');
|
|
const {
|
|
RUN_CANCELLATION_REQUEST_INDEX,
|
|
runCancellationRequestMigration,
|
|
} = require('../../back/migrations/0004-run-cancellation-request');
|
|
const {
|
|
RUN_CANCELLATION_DISPATCH_DUE_INDEX,
|
|
RUN_CANCELLATION_DISPATCH_LEASE_INDEX,
|
|
RUN_CANCELLATION_DISPATCH_TABLE,
|
|
runCancellationDispatchMigration,
|
|
} = require('../../back/migrations/0005-run-cancellation-dispatch');
|
|
const {
|
|
RUN_ATTEMPT_DEADLINE_INDEX,
|
|
runAttemptDeadlineMigration,
|
|
} = require('../../back/migrations/0006-run-attempt-deadline');
|
|
const {
|
|
COMPLETION_RECEIPT_JOURNAL_PURGE_INDEX,
|
|
COMPLETION_RECEIPT_JOURNAL_SCAN_INDEX,
|
|
COMPLETION_RECEIPT_JOURNAL_TABLE,
|
|
completionReceiptJournalMigration,
|
|
} = require('../../back/migrations/0007-completion-receipt-journal');
|
|
const {
|
|
WORKER_REGISTRY_CAPACITY_INDEX,
|
|
WORKER_REGISTRY_LEASE_INDEX,
|
|
WORKER_REGISTRY_TABLE,
|
|
workerRegistryMigration,
|
|
} = require('../../back/migrations/0008-worker-registry');
|
|
const {
|
|
RUN_DISPATCH_LEASE_EXPIRY_INDEX,
|
|
RUN_DISPATCH_LEASE_TABLE,
|
|
RUN_DISPATCH_LEASE_TOKEN_INDEX,
|
|
RUN_DISPATCH_LEASE_WORKER_INDEX,
|
|
runDispatchLeaseMigration,
|
|
} = require('../../back/migrations/0009-run-dispatch-lease');
|
|
const { migrations } = require('../../back/migrations');
|
|
const { runMigrations } = require('../../back/migrations/runner');
|
|
|
|
const databases = [];
|
|
|
|
async function createDatabase() {
|
|
const database = new Sequelize({
|
|
dialect: 'sqlite',
|
|
storage: ':memory:',
|
|
logging: false,
|
|
});
|
|
const migrationModel = defineSchemaMigrationModel(database);
|
|
databases.push(database);
|
|
return { database, migrationModel };
|
|
}
|
|
|
|
afterEach(async () => {
|
|
await Promise.all(databases.splice(0).map((database) => database.close()));
|
|
});
|
|
|
|
test('runs a migration once and records its checksum', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
await queryInterface.createTable('Examples', {
|
|
id: { type: DataTypes.INTEGER, primaryKey: true },
|
|
});
|
|
|
|
let calls = 0;
|
|
const migrations = [
|
|
{
|
|
id: '0001-example',
|
|
checksum: 'checksum-v1',
|
|
async up({ queryInterface, transaction }) {
|
|
calls += 1;
|
|
await queryInterface.addColumn(
|
|
'Examples',
|
|
'name',
|
|
{ type: DataTypes.STRING },
|
|
{ transaction },
|
|
);
|
|
},
|
|
},
|
|
];
|
|
|
|
const options = {
|
|
database,
|
|
migrationModel,
|
|
migrations,
|
|
logger: { info() {} },
|
|
};
|
|
await runMigrations(options);
|
|
await runMigrations(options);
|
|
|
|
const description = await queryInterface.describeTable('Examples');
|
|
const applied = await migrationModel.findByPk('0001-example');
|
|
assert.ok(description.name);
|
|
assert.equal(calls, 1);
|
|
assert.equal(applied.checksum, 'checksum-v1');
|
|
});
|
|
|
|
test('rejects a changed checksum for an applied migration', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const baseOptions = {
|
|
database,
|
|
migrationModel,
|
|
logger: { info() {} },
|
|
};
|
|
|
|
await runMigrations({
|
|
...baseOptions,
|
|
migrations: [
|
|
{
|
|
id: '0001-example',
|
|
checksum: 'checksum-v1',
|
|
async up() {},
|
|
},
|
|
],
|
|
});
|
|
|
|
await assert.rejects(
|
|
runMigrations({
|
|
...baseOptions,
|
|
migrations: [
|
|
{
|
|
id: '0001-example',
|
|
checksum: 'checksum-v2',
|
|
async up() {},
|
|
},
|
|
],
|
|
}),
|
|
/Migration checksum mismatch/,
|
|
);
|
|
});
|
|
|
|
test('rolls back the migration record when migration work fails', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
|
|
await assert.rejects(
|
|
runMigrations({
|
|
database,
|
|
migrationModel,
|
|
logger: { info() {} },
|
|
migrations: [
|
|
{
|
|
id: '0001-failing',
|
|
checksum: 'checksum-v1',
|
|
async up() {
|
|
throw new Error('migration failed');
|
|
},
|
|
},
|
|
],
|
|
}),
|
|
/migration failed/,
|
|
);
|
|
|
|
assert.equal(await migrationModel.count(), 0);
|
|
});
|
|
|
|
test('rejects duplicate migration ids before touching the database', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const duplicate = {
|
|
id: '0001-duplicate',
|
|
checksum: 'checksum-v1',
|
|
async up() {},
|
|
};
|
|
|
|
await assert.rejects(
|
|
runMigrations({
|
|
database,
|
|
migrationModel,
|
|
migrations: [duplicate, duplicate],
|
|
logger: { info() {} },
|
|
}),
|
|
/Duplicate migration id/,
|
|
);
|
|
});
|
|
|
|
test('upgrades the legacy QingLong tables without swallowing schema errors', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
|
|
for (const table of ['CrontabViews', 'Subscriptions', 'Crontabs', 'Envs']) {
|
|
await queryInterface.createTable(table, {
|
|
id: { type: DataTypes.INTEGER, primaryKey: true },
|
|
});
|
|
}
|
|
|
|
const options = {
|
|
database,
|
|
migrationModel,
|
|
migrations: [legacyColumnsMigration],
|
|
logger: { info() {} },
|
|
};
|
|
await runMigrations(options);
|
|
await runMigrations(options);
|
|
|
|
const cronViews = await queryInterface.describeTable('CrontabViews');
|
|
const subscriptions = await queryInterface.describeTable('Subscriptions');
|
|
const crontabs = await queryInterface.describeTable('Crontabs');
|
|
const envs = await queryInterface.describeTable('Envs');
|
|
|
|
assert.ok(cronViews.filterRelation);
|
|
assert.ok(cronViews.type);
|
|
assert.ok(subscriptions.proxy);
|
|
assert.ok(subscriptions.autoAddCron);
|
|
assert.ok(subscriptions.autoDelCron);
|
|
assert.ok(crontabs.sub_id);
|
|
assert.ok(crontabs.extra_schedules);
|
|
assert.ok(crontabs.task_before);
|
|
assert.ok(crontabs.task_after);
|
|
assert.ok(crontabs.log_name);
|
|
assert.ok(crontabs.allow_multiple_instances);
|
|
assert.ok(crontabs.work_dir);
|
|
assert.ok(envs.isPinned);
|
|
assert.ok(envs.labels);
|
|
assert.equal(await migrationModel.count(), 1);
|
|
});
|
|
|
|
test('creates the Run aggregate schema with stable uniqueness constraints', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
const options = {
|
|
database,
|
|
migrationModel,
|
|
migrations: [runSchemaMigration],
|
|
logger: { info() {} },
|
|
};
|
|
|
|
await runMigrations(options);
|
|
await runMigrations(options);
|
|
|
|
const runs = await queryInterface.describeTable(RUN_TABLE);
|
|
const attempts = await queryInterface.describeTable(RUN_ATTEMPT_TABLE);
|
|
const events = await queryInterface.describeTable(RUN_EVENT_TABLE);
|
|
|
|
for (const column of [
|
|
'id',
|
|
'project_id',
|
|
'task_id',
|
|
'task_revision',
|
|
'legacy_cron_id',
|
|
'execution_origin',
|
|
'execution_owner',
|
|
'status',
|
|
'version',
|
|
'event_sequence',
|
|
'created_at_ms',
|
|
]) {
|
|
assert.ok(runs[column], `missing Runs.${column}`);
|
|
}
|
|
for (const column of [
|
|
'id',
|
|
'run_id',
|
|
'attempt',
|
|
'status',
|
|
'executor_type',
|
|
'callback_token_hash',
|
|
'callback_sequence',
|
|
'created_at_ms',
|
|
]) {
|
|
assert.ok(attempts[column], `missing RunAttempts.${column}`);
|
|
}
|
|
for (const column of [
|
|
'id',
|
|
'run_id',
|
|
'sequence',
|
|
'type',
|
|
'dedupe_key',
|
|
'actor_type',
|
|
'payload',
|
|
'created_at_ms',
|
|
]) {
|
|
assert.ok(events[column], `missing RunEvents.${column}`);
|
|
}
|
|
|
|
const runId = '019f70a0-0000-7000-8000-000000000001';
|
|
const attemptId = '019f70a0-0000-7000-8000-000000000002';
|
|
const baseRun = {
|
|
id: runId,
|
|
project_id: 'default',
|
|
task_id: 'legacy-cron:1',
|
|
task_revision: 'revision-1',
|
|
trigger_type: 'manual',
|
|
execution_origin: 'manual',
|
|
execution_owner: 'legacy',
|
|
status: 'created',
|
|
version: 0,
|
|
event_sequence: 0,
|
|
priority: 0,
|
|
created_at_ms: 1_750_000_000_000,
|
|
};
|
|
|
|
await queryInterface.bulkInsert(RUN_TABLE, [
|
|
{ ...baseRun, idempotency_key: 'manual-request-1' },
|
|
]);
|
|
await queryInterface.bulkInsert(RUN_ATTEMPT_TABLE, [
|
|
{
|
|
id: attemptId,
|
|
run_id: runId,
|
|
attempt: 1,
|
|
status: 'claimed',
|
|
executor_type: 'legacy_local',
|
|
callback_sequence: 0,
|
|
created_at_ms: 1_750_000_000_000,
|
|
},
|
|
]);
|
|
await queryInterface.bulkInsert(RUN_EVENT_TABLE, [
|
|
{
|
|
id: '019f70a0-0000-7000-8000-000000000003',
|
|
run_id: runId,
|
|
sequence: 1,
|
|
type: 'run.created',
|
|
dedupe_key: 'create',
|
|
actor_type: 'compatibility',
|
|
attempt_id: attemptId,
|
|
payload: JSON.stringify({ source: 'migration-test' }),
|
|
created_at_ms: 1_750_000_000_000,
|
|
},
|
|
]);
|
|
|
|
await assert.rejects(
|
|
queryInterface.bulkInsert(RUN_TABLE, [
|
|
{
|
|
...baseRun,
|
|
id: '019f70a0-0000-7000-8000-000000000004',
|
|
idempotency_key: 'manual-request-1',
|
|
},
|
|
]),
|
|
);
|
|
await assert.rejects(
|
|
queryInterface.bulkInsert(RUN_TABLE, [
|
|
{
|
|
...baseRun,
|
|
id: '019f70a0-0000-7000-8000-000000000007',
|
|
version: -1,
|
|
idempotency_key: 'manual-request-2',
|
|
},
|
|
]),
|
|
);
|
|
await assert.rejects(
|
|
queryInterface.bulkInsert(RUN_ATTEMPT_TABLE, [
|
|
{
|
|
id: '019f70a0-0000-7000-8000-000000000005',
|
|
run_id: runId,
|
|
attempt: 1,
|
|
status: 'claimed',
|
|
executor_type: 'legacy_local',
|
|
callback_sequence: 0,
|
|
created_at_ms: 1_750_000_000_000,
|
|
},
|
|
]),
|
|
);
|
|
await assert.rejects(
|
|
queryInterface.bulkInsert(RUN_EVENT_TABLE, [
|
|
{
|
|
id: '019f70a0-0000-7000-8000-000000000006',
|
|
run_id: runId,
|
|
sequence: 2,
|
|
type: 'run.created',
|
|
dedupe_key: 'create',
|
|
actor_type: 'compatibility',
|
|
payload: '{}',
|
|
created_at_ms: 1_750_000_000_001,
|
|
},
|
|
]),
|
|
);
|
|
|
|
assert.equal(await migrationModel.count(), 1);
|
|
});
|
|
|
|
test('creates a bounded completion receipt journal linked to Run Attempts', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
await runMigrations({
|
|
database,
|
|
migrationModel,
|
|
migrations: [runSchemaMigration, completionReceiptJournalMigration],
|
|
logger: { info() {} },
|
|
});
|
|
|
|
const columns = await queryInterface.describeTable(
|
|
COMPLETION_RECEIPT_JOURNAL_TABLE,
|
|
);
|
|
for (const column of [
|
|
'attempt_id',
|
|
'run_id',
|
|
'state',
|
|
'quarantine_ref',
|
|
'purge_after_ms',
|
|
'registered_at_ms',
|
|
'updated_at_ms',
|
|
]) {
|
|
assert.ok(columns[column], `missing journal.${column}`);
|
|
}
|
|
const indexes = new Set(
|
|
(await queryInterface.showIndex(COMPLETION_RECEIPT_JOURNAL_TABLE)).map(
|
|
(index) => index.name,
|
|
),
|
|
);
|
|
assert.ok(indexes.has(COMPLETION_RECEIPT_JOURNAL_SCAN_INDEX));
|
|
assert.ok(indexes.has(COMPLETION_RECEIPT_JOURNAL_PURGE_INDEX));
|
|
|
|
await assert.rejects(
|
|
queryInterface.bulkInsert(COMPLETION_RECEIPT_JOURNAL_TABLE, [
|
|
{
|
|
attempt_id: '019f70a0-0000-7000-8000-000000000099',
|
|
run_id: '019f70a0-0000-7000-8000-000000000098',
|
|
state: 'pending',
|
|
registered_at_ms: 1,
|
|
updated_at_ms: 1,
|
|
},
|
|
]),
|
|
/FOREIGN KEY|constraint/i,
|
|
);
|
|
});
|
|
|
|
test('creates the durable fenced Worker registry and bounded lookup indexes', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
await runMigrations({
|
|
database,
|
|
migrationModel,
|
|
migrations: [workerRegistryMigration],
|
|
logger: { info() {} },
|
|
});
|
|
|
|
const columns = await queryInterface.describeTable(WORKER_REGISTRY_TABLE);
|
|
for (const column of [
|
|
'id',
|
|
'session_id',
|
|
'generation',
|
|
'status',
|
|
'version',
|
|
'capabilities_json',
|
|
'capabilities_hash',
|
|
'max_concurrent_runs',
|
|
'available_slots',
|
|
'registered_at_ms',
|
|
'last_heartbeat_at_ms',
|
|
'lease_expires_at_ms',
|
|
'updated_at_ms',
|
|
]) {
|
|
assert.ok(columns[column], `missing workers.${column}`);
|
|
}
|
|
const indexes = new Set(
|
|
(await queryInterface.showIndex(WORKER_REGISTRY_TABLE)).map(
|
|
(index) => index.name,
|
|
),
|
|
);
|
|
assert.ok(indexes.has(WORKER_REGISTRY_LEASE_INDEX));
|
|
assert.ok(indexes.has(WORKER_REGISTRY_CAPACITY_INDEX));
|
|
|
|
await assert.rejects(
|
|
queryInterface.bulkInsert(WORKER_REGISTRY_TABLE, [
|
|
{
|
|
id: 'worker-invalid',
|
|
session_id: '019f7500-0000-7000-8000-000000000001',
|
|
generation: 0,
|
|
status: 'online',
|
|
version: 0,
|
|
capabilities_json: '{}',
|
|
capabilities_hash: 'a'.repeat(64),
|
|
max_concurrent_runs: 1,
|
|
available_slots: 1,
|
|
registered_at_ms: 1,
|
|
last_heartbeat_at_ms: 1,
|
|
lease_expires_at_ms: 2,
|
|
updated_at_ms: 1,
|
|
},
|
|
]),
|
|
/constraint/i,
|
|
);
|
|
});
|
|
|
|
test('creates attempt-scoped Run dispatch leases with Worker fencing indexes', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
await runMigrations({
|
|
database,
|
|
migrationModel,
|
|
migrations: [
|
|
runSchemaMigration,
|
|
workerRegistryMigration,
|
|
runDispatchLeaseMigration,
|
|
],
|
|
logger: { info() {} },
|
|
});
|
|
|
|
const columns = await queryInterface.describeTable(RUN_DISPATCH_LEASE_TABLE);
|
|
for (const column of [
|
|
'attempt_id',
|
|
'run_id',
|
|
'status',
|
|
'version',
|
|
'lease_generation',
|
|
'worker_id',
|
|
'worker_session_id',
|
|
'worker_generation',
|
|
'lease_token',
|
|
'expires_at_ms',
|
|
'completed_at_ms',
|
|
]) {
|
|
assert.ok(columns[column], `missing ${RUN_DISPATCH_LEASE_TABLE}.${column}`);
|
|
}
|
|
const indexes = await queryInterface.showIndex(RUN_DISPATCH_LEASE_TABLE);
|
|
assert.ok(
|
|
indexes.some((index) => index.name === RUN_DISPATCH_LEASE_EXPIRY_INDEX),
|
|
);
|
|
assert.ok(
|
|
indexes.some((index) => index.name === RUN_DISPATCH_LEASE_WORKER_INDEX),
|
|
);
|
|
assert.ok(
|
|
indexes.some(
|
|
(index) => index.name === RUN_DISPATCH_LEASE_TOKEN_INDEX && index.unique,
|
|
),
|
|
);
|
|
await assert.rejects(
|
|
database.query(
|
|
`INSERT INTO ${RUN_DISPATCH_LEASE_TABLE} (
|
|
attempt_id, run_id, status, version, lease_generation,
|
|
worker_id, worker_session_id, worker_generation, lease_token,
|
|
acquired_at_ms, renewed_at_ms, expires_at_ms, updated_at_ms
|
|
) VALUES (
|
|
'missing-attempt', 'missing-run', 'leased', -1, 0,
|
|
'missing-worker', '019f7800-0000-7000-8000-000000000001', 0,
|
|
'lease_token_abcdefghijklmnopqrstuvwxyz0123456789', 1, 1, 2, 1
|
|
)`,
|
|
),
|
|
);
|
|
});
|
|
|
|
test('adds stable Run references to legacy RunningInstance rows', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
await queryInterface.createTable(RUNNING_INSTANCE_TABLE, {
|
|
id: {
|
|
type: DataTypes.INTEGER,
|
|
primaryKey: true,
|
|
autoIncrement: true,
|
|
},
|
|
cron_id: { type: DataTypes.INTEGER, allowNull: false },
|
|
pid: { type: DataTypes.INTEGER, allowNull: true },
|
|
log_path: { type: DataTypes.STRING, allowNull: true },
|
|
started_at: { type: DataTypes.INTEGER, allowNull: false },
|
|
finished_at: { type: DataTypes.INTEGER, allowNull: true },
|
|
status: { type: DataTypes.INTEGER, allowNull: false },
|
|
exit_code: { type: DataTypes.INTEGER, allowNull: true },
|
|
});
|
|
await queryInterface.bulkInsert(RUNNING_INSTANCE_TABLE, [
|
|
{ cron_id: 7, started_at: 1_750_000_000, status: 1 },
|
|
]);
|
|
|
|
const options = {
|
|
database,
|
|
migrationModel,
|
|
migrations: [runningInstanceRunReferenceMigration],
|
|
logger: { info() {} },
|
|
};
|
|
await runMigrations(options);
|
|
await runMigrations(options);
|
|
|
|
const columns = await queryInterface.describeTable(RUNNING_INSTANCE_TABLE);
|
|
assert.ok(columns.run_id);
|
|
assert.ok(columns.attempt_id);
|
|
assert.equal(columns.run_id.allowNull, true);
|
|
assert.equal(columns.attempt_id.allowNull, true);
|
|
const indexNames = new Set(
|
|
(await queryInterface.showIndex(RUNNING_INSTANCE_TABLE)).map(
|
|
(index) => index.name,
|
|
),
|
|
);
|
|
assert.ok(indexNames.has(RUNNING_INSTANCE_RUN_INDEX));
|
|
assert.ok(indexNames.has(RUNNING_INSTANCE_ATTEMPT_INDEX));
|
|
|
|
const fixture = database.define(
|
|
'RunningInstanceMigrationFixture',
|
|
{
|
|
id: { type: DataTypes.INTEGER, primaryKey: true },
|
|
cron_id: DataTypes.INTEGER,
|
|
run_id: DataTypes.STRING(36),
|
|
attempt_id: DataTypes.STRING(36),
|
|
started_at: DataTypes.INTEGER,
|
|
status: DataTypes.INTEGER,
|
|
},
|
|
{ tableName: RUNNING_INSTANCE_TABLE, timestamps: false },
|
|
);
|
|
const legacyRow = await fixture.findByPk(1, { raw: true });
|
|
assert.equal(legacyRow.run_id, null);
|
|
assert.equal(legacyRow.attempt_id, null);
|
|
|
|
const runId = '019f7110-0000-7000-8000-000000000001';
|
|
const attemptId = '019f7110-0000-7000-8000-000000000002';
|
|
await queryInterface.bulkInsert(RUNNING_INSTANCE_TABLE, [
|
|
{
|
|
cron_id: 7,
|
|
run_id: runId,
|
|
attempt_id: attemptId,
|
|
started_at: 1_750_000_001,
|
|
status: 0,
|
|
},
|
|
]);
|
|
await assert.rejects(
|
|
queryInterface.bulkInsert(RUNNING_INSTANCE_TABLE, [
|
|
{
|
|
cron_id: 7,
|
|
run_id: runId,
|
|
attempt_id: attemptId,
|
|
started_at: 1_750_000_002,
|
|
status: 0,
|
|
},
|
|
]),
|
|
);
|
|
assert.equal(await migrationModel.count(), 1);
|
|
});
|
|
|
|
test('adds durable cancellation requests without rewriting legacy Run rows', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
await runMigrations({
|
|
database,
|
|
migrationModel,
|
|
migrations: [runSchemaMigration],
|
|
logger: { info() {} },
|
|
});
|
|
|
|
const runId = '019f7110-0000-7000-8000-000000000004';
|
|
await queryInterface.bulkInsert(RUN_TABLE, [
|
|
{
|
|
id: runId,
|
|
project_id: 'default',
|
|
task_id: 'legacy-cron:7',
|
|
task_revision: 'revision-7',
|
|
trigger_type: 'manual',
|
|
execution_origin: 'manual',
|
|
execution_owner: 'runtime',
|
|
status: 'running',
|
|
version: 3,
|
|
event_sequence: 3,
|
|
priority: 0,
|
|
created_at_ms: 1_750_000_000_000,
|
|
started_at_ms: 1_750_000_000_010,
|
|
},
|
|
]);
|
|
|
|
const options = {
|
|
database,
|
|
migrationModel,
|
|
migrations: [runSchemaMigration, runCancellationRequestMigration],
|
|
logger: { info() {} },
|
|
};
|
|
await runMigrations(options);
|
|
await runMigrations(options);
|
|
|
|
const columns = await queryInterface.describeTable(RUN_TABLE);
|
|
assert.ok(columns.cancel_requested_at_ms);
|
|
assert.ok(columns.cancel_reason);
|
|
assert.equal(columns.cancel_requested_at_ms.allowNull, true);
|
|
assert.equal(columns.cancel_reason.allowNull, true);
|
|
const indexNames = new Set(
|
|
(await queryInterface.showIndex(RUN_TABLE)).map((index) => index.name),
|
|
);
|
|
assert.ok(indexNames.has(RUN_CANCELLATION_REQUEST_INDEX));
|
|
|
|
const [legacyRow] = await database.query(
|
|
`SELECT cancel_requested_at_ms, cancel_reason FROM ${RUN_TABLE} WHERE id = :runId`,
|
|
{
|
|
replacements: { runId },
|
|
type: QueryTypes.SELECT,
|
|
},
|
|
);
|
|
assert.equal(legacyRow.cancel_requested_at_ms, null);
|
|
assert.equal(legacyRow.cancel_reason, null);
|
|
assert.equal(await migrationModel.count(), 2);
|
|
});
|
|
|
|
test('adds nullable Attempt deadlines without rewriting existing attempts', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
await runMigrations({
|
|
database,
|
|
migrationModel,
|
|
migrations: [runSchemaMigration],
|
|
logger: { info() {} },
|
|
});
|
|
|
|
const runId = '019f7110-0000-7000-8000-000000000014';
|
|
const attemptId = '019f7110-0000-7000-8000-000000000015';
|
|
await queryInterface.bulkInsert(RUN_TABLE, [
|
|
{
|
|
id: runId,
|
|
project_id: 'default',
|
|
task_id: 'legacy-cron:8',
|
|
task_revision: 'revision-8',
|
|
trigger_type: 'manual',
|
|
execution_origin: 'manual',
|
|
execution_owner: 'runtime',
|
|
status: 'dispatching',
|
|
version: 2,
|
|
event_sequence: 2,
|
|
priority: 0,
|
|
created_at_ms: 1_750_000_000_000,
|
|
},
|
|
]);
|
|
await queryInterface.bulkInsert(RUN_ATTEMPT_TABLE, [
|
|
{
|
|
id: attemptId,
|
|
run_id: runId,
|
|
attempt: 1,
|
|
status: 'starting',
|
|
executor_type: 'local_process',
|
|
callback_sequence: 0,
|
|
created_at_ms: 1_750_000_000_000,
|
|
},
|
|
]);
|
|
|
|
const options = {
|
|
database,
|
|
migrationModel,
|
|
migrations: [runSchemaMigration, runAttemptDeadlineMigration],
|
|
logger: { info() {} },
|
|
};
|
|
await runMigrations(options);
|
|
await runMigrations(options);
|
|
|
|
const columns = await queryInterface.describeTable(RUN_ATTEMPT_TABLE);
|
|
assert.ok(columns.deadline_at_ms);
|
|
assert.equal(columns.deadline_at_ms.allowNull, true);
|
|
const indexNames = new Set(
|
|
(await queryInterface.showIndex(RUN_ATTEMPT_TABLE)).map(
|
|
(index) => index.name,
|
|
),
|
|
);
|
|
assert.ok(indexNames.has(RUN_ATTEMPT_DEADLINE_INDEX));
|
|
const [legacyAttempt] = await database.query(
|
|
`SELECT deadline_at_ms FROM ${RUN_ATTEMPT_TABLE} WHERE id = :attemptId`,
|
|
{
|
|
replacements: { attemptId },
|
|
type: QueryTypes.SELECT,
|
|
},
|
|
);
|
|
assert.equal(legacyAttempt.deadline_at_ms, null);
|
|
assert.equal(await migrationModel.count(), 2);
|
|
});
|
|
|
|
test('creates a fenced cancellation dispatch lease bound to a Run Attempt', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
await runMigrations({
|
|
database,
|
|
migrationModel,
|
|
migrations: [
|
|
runSchemaMigration,
|
|
runCancellationRequestMigration,
|
|
runCancellationDispatchMigration,
|
|
],
|
|
logger: { info() {} },
|
|
});
|
|
|
|
const columns = await queryInterface.describeTable(
|
|
RUN_CANCELLATION_DISPATCH_TABLE,
|
|
);
|
|
for (const column of [
|
|
'run_id',
|
|
'attempt_id',
|
|
'status',
|
|
'version',
|
|
'dispatch_count',
|
|
'next_attempt_at_ms',
|
|
'lease_owner',
|
|
'lease_token',
|
|
'lease_expires_at_ms',
|
|
'last_result',
|
|
'last_dispatched_at_ms',
|
|
'created_at_ms',
|
|
'updated_at_ms',
|
|
]) {
|
|
assert.ok(columns[column], `missing cancellation dispatch.${column}`);
|
|
}
|
|
const indexNames = new Set(
|
|
(await queryInterface.showIndex(RUN_CANCELLATION_DISPATCH_TABLE)).map(
|
|
(index) => index.name,
|
|
),
|
|
);
|
|
assert.ok(indexNames.has(RUN_CANCELLATION_DISPATCH_DUE_INDEX));
|
|
assert.ok(indexNames.has(RUN_CANCELLATION_DISPATCH_LEASE_INDEX));
|
|
|
|
const runId = '019f7110-0000-7000-8000-000000000011';
|
|
const attemptId = '019f7110-0000-7000-8000-000000000012';
|
|
await queryInterface.bulkInsert(RUN_TABLE, [
|
|
{
|
|
id: runId,
|
|
project_id: 'default',
|
|
task_id: 'legacy-cron:8',
|
|
task_revision: 'revision-8',
|
|
trigger_type: 'manual',
|
|
execution_origin: 'manual',
|
|
execution_owner: 'runtime',
|
|
status: 'running',
|
|
version: 2,
|
|
event_sequence: 2,
|
|
priority: 0,
|
|
created_at_ms: 1_750_000_000_000,
|
|
cancel_requested_at_ms: 1_750_000_000_100,
|
|
cancel_reason: 'user',
|
|
},
|
|
]);
|
|
await queryInterface.bulkInsert(RUN_ATTEMPT_TABLE, [
|
|
{
|
|
id: attemptId,
|
|
run_id: runId,
|
|
attempt: 1,
|
|
status: 'running',
|
|
executor_type: 'local_process',
|
|
callback_sequence: 0,
|
|
created_at_ms: 1_750_000_000_010,
|
|
},
|
|
]);
|
|
await queryInterface.bulkInsert(RUN_CANCELLATION_DISPATCH_TABLE, [
|
|
{
|
|
run_id: runId,
|
|
attempt_id: attemptId,
|
|
status: 'pending',
|
|
version: 0,
|
|
dispatch_count: 0,
|
|
next_attempt_at_ms: 1_750_000_000_100,
|
|
created_at_ms: 1_750_000_000_100,
|
|
updated_at_ms: 1_750_000_000_100,
|
|
},
|
|
]);
|
|
await assert.rejects(
|
|
queryInterface.bulkInsert(RUN_CANCELLATION_DISPATCH_TABLE, [
|
|
{
|
|
run_id: 'missing-run',
|
|
attempt_id: attemptId,
|
|
status: 'pending',
|
|
version: 0,
|
|
dispatch_count: 0,
|
|
created_at_ms: 1_750_000_000_100,
|
|
updated_at_ms: 1_750_000_000_100,
|
|
},
|
|
]),
|
|
);
|
|
await assert.rejects(
|
|
queryInterface.bulkUpdate(
|
|
RUN_CANCELLATION_DISPATCH_TABLE,
|
|
{ version: -1 },
|
|
{ run_id: runId },
|
|
),
|
|
);
|
|
});
|
|
|
|
test('runs the registered migration chain against a legacy database fixture', async () => {
|
|
const { database, migrationModel } = await createDatabase();
|
|
const queryInterface = database.getQueryInterface();
|
|
|
|
for (const table of ['CrontabViews', 'Subscriptions', 'Crontabs', 'Envs']) {
|
|
await queryInterface.createTable(table, {
|
|
id: { type: DataTypes.INTEGER, primaryKey: true },
|
|
});
|
|
}
|
|
|
|
const options = {
|
|
database,
|
|
migrationModel,
|
|
logger: { info() {} },
|
|
};
|
|
await runMigrations(options);
|
|
await runMigrations(options);
|
|
|
|
const tables = new Set(await queryInterface.showAllTables());
|
|
assert.ok(tables.has(RUN_TABLE));
|
|
assert.ok(tables.has(RUN_ATTEMPT_TABLE));
|
|
assert.ok(tables.has(RUN_EVENT_TABLE));
|
|
assert.ok(tables.has(RUN_CANCELLATION_DISPATCH_TABLE));
|
|
assert.equal(await migrationModel.count(), migrations.length);
|
|
});
|