open-source/qinglong
1
0
Fork 0
mirror of https://github.com/whyour/qinglong.git synced 2026-09-20 16:07:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(ci): restore ql3 multi-profile gates

Browse Source
This commit is contained in:
whyour
2026-08-22 13:44:19 +08:00
parent f1cca9a3ed
commit c8ad5ea368
17 changed files with 320 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ql3-cluster-postgres/src/task-start/taskStartRepository.ts
+24 -36
View File
@@ -6,7 +6,6 @@ import {
TaskStartUnavailableError,
normalizeTaskStartCommand,
normalizeTaskStartResult,
type TaskStartAllowedRole,
type TaskStartCommand,
type TaskStartRepository,
type TaskStartResult,
@@ -32,12 +31,6 @@ import {
type Row = Record<string, unknown>;
const ALLOWED_ROLES = new Set<TaskStartAllowedRole>([
'owner',
'admin',
'operator',
]);
function unavailable(options?: ErrorOptions): TaskStartUnavailableError {
return new TaskStartUnavailableError(options);
}
@@ -193,34 +186,30 @@ export class PostgresTaskStartRepository implements TaskStartRepository {
try {
await configurePostgresDefinitionTransaction(client);
began = true;
const project = await client.query<Row>(`
SELECT status AS "projectStatus", version AS "projectVersion"
FROM "ql3"."projects" WHERE id = $1 FOR UPDATE
`, [command.projectId]);
if (project.rows.length === 0) throw new TaskStartNotFoundError();
if (project.rows.length !== 1) throw unavailable();
const binding = await client.query<Row>(`
SELECT version AS "bindingVersion", state AS "bindingState",
role AS "bindingRole"
FROM "ql3"."project_role_bindings"
WHERE project_id = $1 AND subject_type = $2 AND subject_id = $3
ORDER BY version DESC LIMIT 1
FOR SHARE
`, [command.projectId, command.subject.type, command.subject.id]);
const currentProject = project.rows[0]!;
const currentBinding = binding.rows[0];
if (
text(currentProject, 'projectStatus') !== 'active' ||
integer(currentProject, 'projectVersion') !==
command.policyFence.projectVersion ||
!currentBinding ||
integer(currentBinding, 'bindingVersion') !==
command.policyFence.bindingVersion ||
text(currentBinding, 'bindingState') !== 'active' ||
!ALLOWED_ROLES.has(
text(currentBinding, 'bindingRole') as TaskStartAllowedRole,
)
) {
const authorization = await client.query<Row>(
`SELECT "ql3"."lock_run_management_policy_fence"(
$1::varchar, $2::varchar, $3::varchar, $4::integer, $5::integer
) AS "matches"`,
[
command.projectId,
command.subject.type,
command.subject.id,
command.policyFence.projectVersion,
command.policyFence.bindingVersion,
],
);
if (authorization.rows.length !== 1) throw unavailable();
if (authorization.rows[0]?.matches !== true) {
const project = await client.query<Row>(
`SELECT EXISTS (
SELECT 1 FROM "ql3"."projects" WHERE id = $1
) AS "exists"`,
[command.projectId],
);
if (project.rows.length !== 1) throw unavailable();
if (!postgresRequiredBoolean(project.rows[0]!.exists, unavailable)) {
throw new TaskStartNotFoundError();
}
throw new TaskStartFenceRejectedError('authorization_changed');
}
@@ -261,7 +250,6 @@ export class PostgresTaskStartRepository implements TaskStartRepository {
AND revision.task_id = head.task_id
AND revision.revision = head.current_revision
WHERE head.project_id = $1 AND head.task_id = $2
FOR UPDATE OF head
`, [command.projectId, command.taskId]);
if (task.rows.length === 0) throw new TaskStartNotFoundError();
if (task.rows.length !== 1) throw unavailable();
packages/ql3-cluster-postgres/test/postgres.integration.test.cjs
+25 -10
View File
@@ -626,6 +626,8 @@ const adminConnectionString =
const automationManagerConnectionString =
process.env.QL3_TEST_POSTGRES_AUTOMATION_MANAGER_URL ??
migrationConnectionString;
const runManagerConnectionString =
process.env.QL3_TEST_POSTGRES_RUN_MANAGER_URL ?? migrationConnectionString;
const packageManagerConnectionString =
process.env.QL3_TEST_POSTGRES_PACKAGE_MANAGER_URL ??
migrationConnectionString;
@@ -678,6 +680,8 @@ if (!migrationConnectionString) {
? adminConnectionString
: role === 'automation-manager'
? automationManagerConnectionString
: role === 'run-manager'
? runManagerConnectionString
: role === 'package-manager'
? packageManagerConnectionString
: role === 'package-executor'
@@ -1045,7 +1049,7 @@ if (!migrationConnectionString) {
const taskId = 'task-start-command';
const subjectId = 'usr_task_start_integration';
const migrationDatabase = await open('migration');
let runtimeDatabase;
let runManagerDatabase;
try {
await runPostgresMigrations({ pool: migrationDatabase.pool });
await migrationDatabase.pool.query(
@@ -1102,11 +1106,13 @@ if (!migrationConnectionString) {
})
).definition;
runtimeDatabase =
runtimeConnectionString === migrationConnectionString
runManagerDatabase =
runManagerConnectionString === migrationConnectionString
? migrationDatabase
: await open('runtime');
const repository = new PostgresTaskStartRepository(runtimeDatabase.pool);
: await open('run-manager');
const repository = new PostgresTaskStartRepository(
runManagerDatabase.pool,
);
const command = {
projectId,
taskId,
@@ -1158,8 +1164,8 @@ if (!migrationConnectionString) {
},
]);
} finally {
if (runtimeDatabase && runtimeDatabase !== migrationDatabase) {
await runtimeDatabase.close();
if (runManagerDatabase && runManagerDatabase !== migrationDatabase) {
await runManagerDatabase.close();
}
await migrationDatabase.close();
}
@@ -2193,6 +2199,9 @@ if (!migrationConnectionString) {
const migrationDatabase = await open('migration');
try {
await runPostgresMigrations({ pool: migrationDatabase.pool });
await migrationDatabase.pool.query(
'TRUNCATE TABLE "ql3"."runs" CASCADE',
);
await observeContractPublisherTrust(migrationDatabase.pool);
await migrationDatabase.pool.query(
`INSERT INTO "ql3"."projects" (
@@ -2577,6 +2586,9 @@ if (!migrationConnectionString) {
const migrationDatabase = await open('migration');
try {
await runPostgresMigrations({ pool: migrationDatabase.pool });
await migrationDatabase.pool.query(
'TRUNCATE TABLE "ql3"."plugin_package_installs" CASCADE',
);
await observeContractPublisherTrust(migrationDatabase.pool);
await migrationDatabase.pool.query(
`INSERT INTO "ql3"."projects" (
@@ -2589,7 +2601,7 @@ if (!migrationConnectionString) {
await migrationDatabase.close();
}
const executorDatabase = await open('package-executor');
const adminDatabase = await open('admin');
const automationManagerDatabase = await open('automation-manager');
return {
repository: new PostgresPluginPackageTaskReconciliationRepository(
executorDatabase.pool,
@@ -2607,11 +2619,14 @@ if (!migrationConnectionString) {
),
),
taskRepository: new PostgresTaskDefinitionRepository(
adminDatabase.pool,
automationManagerDatabase.pool,
fixture.registry,
),
close: async () => {
await Promise.all([executorDatabase.close(), adminDatabase.close()]);
await Promise.all([
executorDatabase.close(),
automationManagerDatabase.close(),
]);
},
};
},
packages/ql3-cluster-postgres/test/taskStartRepository.test.cjs
+15 -20
View File
@@ -136,19 +136,12 @@ function fixture(options = {}) {
normalized.startsWith('SELECT set_config') ||
normalized.startsWith('INSERT INTO')
) return { rows: [], rowCount: normalized.startsWith('INSERT') ? 1 : 0 };
if (normalized.includes('FROM "ql3"."projects"')) {
const rows = options.projectRows ?? [{
projectStatus: 'active',
projectVersion: 2,
}];
if (normalized.includes('lock_run_management_policy_fence')) {
const rows = options.authorizationRows ?? [{ matches: true }];
return { rows, rowCount: rows.length };
}
if (normalized.includes('FROM "ql3"."project_role_bindings"')) {
const rows = options.bindingRows ?? [{
bindingVersion: 3,
bindingState: 'active',
bindingRole: 'operator',
}];
if (normalized.includes('SELECT EXISTS')) {
const rows = options.projectExistenceRows ?? [{ exists: true }];
return { rows, rowCount: rows.length };
}
if (normalized.includes('FROM "ql3"."runs"')) {
@@ -203,11 +196,11 @@ test('revalidates Policy and Task/execution digests before one atomic Run aggreg
executionRevisionDigest: EXECUTION.contentDigest,
createdAtMs: 1_000,
});
const project = calls.findIndex(({ sql }) => sql.includes('FROM "ql3"."projects"'));
const binding = calls.findIndex(({ sql }) => sql.includes('project_role_bindings'));
const authorization = calls.findIndex(({ sql }) =>
sql.includes('lock_run_management_policy_fence'));
const task = calls.findIndex(({ sql }) => sql.includes('task_definitions'));
const execution = calls.findIndex(({ sql }) => sql.includes('task_execution_revisions'));
assert.ok(project < binding && binding < task && task < execution);
assert.ok(authorization < task && task < execution);
assert.equal(calls.filter(({ sql }) => sql.startsWith('INSERT INTO')).length, 4);
assert.equal(calls.some(({ sql }) => sql === 'COMMIT'), true);
});
@@ -273,15 +266,17 @@ test('returns the original durable identities for an exact replay', async () =>
test('rejects missing, authorization, definition and disabled fences', async () => {
await assert.rejects(
fixture({ projectRows: [] }).repository.startTask(command()),
fixture({
authorizationRows: [{ matches: null }],
projectExistenceRows: [{ exists: false }],
}).repository.startTask(command()),
TaskStartNotFoundError,
);
await assert.rejects(
fixture({ bindingRows: [{
bindingVersion: 4,
bindingState: 'revoked',
bindingRole: null,
}] }).repository.startTask(command()),
fixture({
authorizationRows: [{ matches: false }],
projectExistenceRows: [{ exists: true }],
}).repository.startTask(command()),
(error) =>
error instanceof TaskStartFenceRejectedError &&
error.reason === 'authorization_changed',