feat(ql3): add strong cluster run stop

This commit is contained in:
whyour
2026-08-12 08:21:28 +08:00
parent c0ab62e64a
commit dd370b2842
22 changed files with 1602 additions and 321 deletions
@@ -1,4 +1,8 @@
export { PostgresRunManualRetryRepository } from '../run-management/runManualRetryRepository';
export {
PostgresClusterRunCancellationRepository,
type PostgresRunManagementCancellationCommand,
} from '../run-recovery/clusterRunCancellationRepository';
export { PostgresProjectPolicyRepository } from '../security/projectPolicyRepository';
export { PostgresSecurityAuditRepository } from '../security/securityAuditRepository';
export {
@@ -288,5 +288,10 @@ export const postgresqlMainMigrationManifest: MigrationStreamManifest =
checksum:
'7aa2b2ade67cdfa6839d4af02209906646a68adfd6c12c4dddeb854021da72b8',
}),
Object.freeze({
id: 'pg-0057-run-management-stop-boundary',
checksum:
'ab2d0eee3d85a937e1e87243b1fd1e75181529122b64026303488404162e4ba7',
}),
]),
});
@@ -59,6 +59,7 @@ import { pg0053PluginPackageWorkflowRunListIndexMigration } from './pg-0053-plug
import { pg0054ApprovalManagementBoundaryMigration } from './pg-0054-approval-management-boundary';
import { pg0055RunAttemptLogRetentionMigration } from './pg-0055-run-attempt-log-retention';
import { pg0056RunManagementBoundaryMigration } from '../run-management/pg-0056-run-management-boundary';
import { pg0057RunManagementStopBoundaryMigration } from '../run-management/pg-0057-run-management-stop-boundary';
export const postgresqlMainMigrationStream: MigrationStreamDefinition<PostgresMigrationContext> =
Object.freeze({
@@ -123,5 +124,6 @@ export const postgresqlMainMigrationStream: MigrationStreamDefinition<PostgresMi
pg0054ApprovalManagementBoundaryMigration,
pg0055RunAttemptLogRetentionMigration,
pg0056RunManagementBoundaryMigration,
pg0057RunManagementStopBoundaryMigration,
]),
});
@@ -0,0 +1,35 @@
import { CAPABILITIES_V55 } from './pg-0056-run-management-boundary';
import { definePostgresSqlMigration } from '../migrations/sqlMigration';
export const CAPABILITIES_V56 = CAPABILITIES_V55.replace(
'"run_management_boundary":1,',
'"run_management_boundary":1,"run_management_stop":1,',
);
export const pg0057RunManagementStopBoundaryMigration =
definePostgresSqlMigration({
id: 'pg-0057-run-management-stop-boundary',
statements: [
`REVOKE UPDATE ON "ql3"."runs" FROM ql3_run_manager`,
`GRANT UPDATE (cancel_requested_at_ms, cancel_reason, version, event_sequence) ON "ql3"."runs" TO ql3_run_manager`,
`
DO $ql3$
BEGIN
UPDATE "ql3"."schema_capabilities"
SET contract_version = 56,
migration_id = 'pg-0057-run-management-stop-boundary',
capabilities = '${CAPABILITIES_V56}'::jsonb,
updated_at_ms = floor(extract(epoch FROM transaction_timestamp()) * 1000)::bigint
WHERE contract_name = 'control-core'
AND contract_version = 55
AND migration_id = 'pg-0056-run-management-boundary'
AND capabilities = '${CAPABILITIES_V55}'::jsonb;
IF NOT FOUND THEN
RAISE EXCEPTION 'control-core capability is not at version 55'
USING ERRCODE = 'check_violation';
END IF;
END
$ql3$
`.trim(),
],
});
@@ -7,20 +7,25 @@ import {
InvalidClusterRunCancellationError,
normalizeClusterRunCancellationCommand,
normalizeClusterRunCancellationResult,
type ClusterRunCancellationAllowedRole,
type ClusterRunCancellationCommand,
type ClusterRunCancellationRepository,
type ClusterRunCancellationResult,
} from '@qinglong/runtime-core/cluster-run-cancellation';
import { RUN_STATUSES, type RunStatus } from '@qinglong/runtime-core';
import {
RUN_STATUSES,
type RunStatus,
type SecurityPolicyFence,
type SecurityPrincipal,
} from '@qinglong/runtime-core';
import { normalizeSecurityPrincipal } from '@qinglong/runtime-core/security';
type Row = Record<string, unknown>;
const ALLOWED_ROLES = new Set<ClusterRunCancellationAllowedRole>([
'owner',
'admin',
'operator',
]);
const STRONG_ASSURANCES = new Set(['multi_factor', 'hardware']);
const MAX_AUTHENTICATION_AGE_MS = 5 * 60 * 1_000;
const IDENTIFIER_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/;
const UUID_PATTERN =
/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
const TERMINAL = new Set<RunStatus>([
'succeeded',
'failed',
@@ -35,6 +40,23 @@ const CANCEL_REASONS = new Set([
'timeout',
]);
export interface PostgresRunManagementCancellationCommand {
readonly projectId: string;
readonly runId: string;
readonly mutationId: string;
readonly eventId: string;
readonly requestId: string;
readonly auditEventId: string;
readonly principal: Readonly<SecurityPrincipal>;
readonly policyFence: Readonly<SecurityPolicyFence>;
}
interface CancellationAudit {
readonly requestId: string;
readonly auditEventId: string;
readonly principal: Readonly<SecurityPrincipal>;
}
function text(row: Row, key: string): string {
const value = row[key];
if (typeof value !== 'string' || value.length < 1) {
@@ -45,14 +67,9 @@ function text(row: Row, key: string): string {
function integer(row: Row, key: string): number {
const raw = row[key];
const value = typeof raw === 'string' && /^(0|[1-9]\d*)$/.test(raw)
? Number(raw)
: raw;
if (
typeof value !== 'number' ||
!Number.isSafeInteger(value) ||
value < 0
) {
const value =
typeof raw === 'string' && /^(0|[1-9]\d*)$/.test(raw) ? Number(raw) : raw;
if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) {
throw new TypeError(`PostgreSQL Run cancellation ${key} is invalid`);
}
return value;
@@ -70,6 +87,94 @@ function optionalText(row: Row, key: string): string | undefined {
: text(row, key);
}
function exact(
value: unknown,
keys: readonly string[],
): Record<string, unknown> {
if (!value || typeof value !== 'object' || Array.isArray(value)) {
throw new InvalidClusterRunCancellationError(
'management command is invalid',
);
}
const actual = Object.keys(value).sort();
const expected = [...keys].sort();
if (
actual.length !== expected.length ||
actual.some((key, index) => key !== expected[index])
) {
throw new InvalidClusterRunCancellationError(
'management command shape is invalid',
);
}
return value as Record<string, unknown>;
}
function managementIdentifier(value: unknown, name: string): string {
if (typeof value !== 'string' || !IDENTIFIER_PATTERN.test(value)) {
throw new InvalidClusterRunCancellationError(`${name} is invalid`);
}
return value;
}
function managementUuid(value: unknown, name: string): string {
if (typeof value !== 'string' || !UUID_PATTERN.test(value)) {
throw new InvalidClusterRunCancellationError(`${name} is invalid`);
}
return value;
}
function normalizeManagementCommand(
value: Readonly<PostgresRunManagementCancellationCommand>,
): Readonly<{
command: Readonly<ClusterRunCancellationCommand>;
audit: Readonly<CancellationAudit>;
}> {
const input = exact(value, [
'projectId',
'runId',
'mutationId',
'eventId',
'requestId',
'auditEventId',
'principal',
'policyFence',
]);
const principalInput = exact(input.principal, [
'subject',
'authenticationId',
'authenticatedAtMs',
'expiresAtMs',
'assurance',
]) as unknown as SecurityPrincipal;
const projectId = managementIdentifier(input.projectId, 'projectId');
const runId = managementIdentifier(input.runId, 'runId');
const mutationId = managementUuid(input.mutationId, 'mutationId');
const eventId = managementUuid(input.eventId, 'eventId');
const requestId = managementIdentifier(input.requestId, 'requestId');
const auditEventId = managementUuid(input.auditEventId, 'auditEventId');
if (eventId === auditEventId) {
throw new InvalidClusterRunCancellationError(
'event and audit identity must differ',
);
}
const command = normalizeClusterRunCancellationCommand({
projectId,
runId,
mutationId,
eventId,
subject: principalInput.subject,
policyFence: input.policyFence as SecurityPolicyFence,
});
return Object.freeze({
command,
audit: Object.freeze({
requestId,
auditEventId,
principal: principalInput,
}),
});
}
function runStatus(row: Row): RunStatus {
const value = text(row, 'runStatus') as RunStatus;
if (!RUN_STATUSES.includes(value)) {
@@ -131,6 +236,122 @@ async function databaseNow(client: PostgresClient): Promise<number> {
return integer(result.rows[0]!, 'nowMs');
}
function confirmStrongAuthentication(
value: Readonly<SecurityPrincipal>,
observedAtMs: number,
): Readonly<SecurityPrincipal> {
let principal: Readonly<SecurityPrincipal>;
try {
principal = normalizeSecurityPrincipal(value, observedAtMs);
} catch {
throw new ClusterRunCancellationFenceRejectedError('authorization_changed');
}
if (
principal.subject.type !== 'user' ||
!STRONG_ASSURANCES.has(principal.assurance) ||
principal.authenticatedAtMs > observedAtMs ||
principal.expiresAtMs <= observedAtMs ||
observedAtMs - principal.authenticatedAtMs > MAX_AUTHENTICATION_AGE_MS
) {
throw new ClusterRunCancellationFenceRejectedError('authorization_changed');
}
return principal;
}
async function confirmAuthorization(
client: PostgresClient,
command: Readonly<ClusterRunCancellationCommand>,
): Promise<void> {
const result = 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 (result.rows.length !== 1 || result.rows[0]?.matches !== true) {
throw new ClusterRunCancellationFenceRejectedError('authorization_changed');
}
}
async function recordAllowedAudit(
client: PostgresClient,
command: Readonly<ClusterRunCancellationCommand>,
audit: Readonly<CancellationAudit>,
observedAtMs: number,
): Promise<void> {
const inserted = await client.query<Row>(
`
INSERT INTO "ql3"."security_audit_events" (
event_id, request_id, operation_id, project_id,
subject_type, subject_id, authentication_id, outcome, reasons,
project_version, binding_version, occurred_at_ms
) VALUES (
$1, $2, 'run.stop', $3, $4, $5, $6, 'allowed', $7::jsonb,
$8, $9, $10
)
ON CONFLICT (event_id) DO NOTHING
RETURNING event_id AS "eventId"
`,
[
audit.auditEventId,
audit.requestId,
command.projectId,
audit.principal.subject.type,
audit.principal.subject.id,
audit.principal.authenticationId,
JSON.stringify(['role_grant', 'strong_authentication']),
command.policyFence.projectVersion,
command.policyFence.bindingVersion,
observedAtMs,
],
);
if (inserted.rows.length === 1) return;
if (inserted.rows.length !== 0) {
throw new TypeError('PostgreSQL Run cancellation audit is invalid');
}
const replay = await client.query<Row>(
`
SELECT request_id AS "requestId", operation_id AS "operationId",
project_id AS "projectId", subject_type AS "subjectType",
subject_id AS "subjectId", authentication_id AS "authenticationId",
outcome, reasons, project_version AS "projectVersion",
binding_version AS "bindingVersion"
FROM "ql3"."security_audit_events"
WHERE event_id = $1
`,
[audit.auditEventId],
);
const row = replay.rows[0];
const reasons = row?.reasons;
if (
replay.rows.length !== 1 ||
!row ||
row.requestId !== audit.requestId ||
row.operationId !== 'run.stop' ||
row.projectId !== command.projectId ||
row.subjectType !== audit.principal.subject.type ||
row.subjectId !== audit.principal.subject.id ||
row.authenticationId !== audit.principal.authenticationId ||
row.outcome !== 'allowed' ||
!Array.isArray(reasons) ||
reasons.length !== 2 ||
reasons[0] !== 'role_grant' ||
reasons[1] !== 'strong_authentication' ||
integer(row, 'projectVersion') !== command.policyFence.projectVersion ||
integer(row, 'bindingVersion') !== command.policyFence.bindingVersion
) {
throw new TypeError('PostgreSQL Run cancellation audit replay drifted');
}
}
async function rollback(client: PostgresClient): Promise<void> {
try {
await client.query('ROLLBACK');
@@ -140,7 +361,8 @@ async function rollback(client: PostgresClient): Promise<void> {
}
export class PostgresClusterRunCancellationRepository
implements ClusterRunCancellationRepository {
implements ClusterRunCancellationRepository
{
constructor(private readonly pool: PostgresPool) {
if (!pool || typeof pool.connect !== 'function') {
throw new TypeError('PostgreSQL Run cancellation pool is invalid');
@@ -151,67 +373,71 @@ export class PostgresClusterRunCancellationRepository
value: Readonly<ClusterRunCancellationCommand>,
): Promise<Readonly<ClusterRunCancellationResult>> {
const command = normalizeClusterRunCancellationCommand(value);
return this.requestCancellation(command);
}
async requestUserCancellationAudited(
value: Readonly<PostgresRunManagementCancellationCommand>,
): Promise<Readonly<ClusterRunCancellationResult>> {
const normalized = normalizeManagementCommand(value);
return this.requestCancellation(normalized.command, normalized.audit);
}
private requestCancellation(
command: Readonly<ClusterRunCancellationCommand>,
audit?: Readonly<CancellationAudit>,
): Promise<Readonly<ClusterRunCancellationResult>> {
return this.transaction(async (client) => {
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 ClusterRunCancellationNotFoundError();
}
if (project.rows.length !== 1) {
throw new TypeError('PostgreSQL Run cancellation Project is invalid');
}
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
`, [
command.projectId,
command.subject.type,
command.subject.id,
]);
const currentProject = project.rows[0]!;
const currentBinding = binding.rows[0];
const observedAtMs = audit ? await databaseNow(client) : undefined;
const confirmedAudit = audit
? Object.freeze({
...audit,
principal: confirmStrongAuthentication(
audit.principal,
observedAtMs!,
),
})
: undefined;
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 ClusterRunCancellationAllowedRole,
)
confirmedAudit &&
(confirmedAudit.principal.subject.type !== command.subject.type ||
confirmedAudit.principal.subject.id !== command.subject.id)
) {
throw new ClusterRunCancellationFenceRejectedError(
'authorization_changed',
);
}
await confirmAuthorization(client, command);
const run = await client.query<Row>(`
const run = await client.query<Row>(
`
SELECT project_id AS "projectId", status AS "runStatus",
version AS "runVersion", event_sequence AS "eventSequence",
cancel_requested_at_ms AS "cancelRequestedAtMs",
cancel_reason AS "cancelReason"
FROM "ql3"."runs" WHERE id = $1 FOR UPDATE
`, [command.runId]);
if (run.rows.length === 0 || run.rows[0]?.projectId !== command.projectId) {
`,
[command.runId],
);
if (
run.rows.length === 0 ||
run.rows[0]?.projectId !== command.projectId
) {
throw new ClusterRunCancellationNotFoundError();
}
if (run.rows.length !== 1) {
throw new TypeError('PostgreSQL Run cancellation Run is invalid');
}
if (command.workflowTarget) {
const admission = await client.query<Row>(`
const admission = await client.query<Row>(
`
SELECT project_id AS "projectId", package_name AS "packageName",
workflow_id AS "workflowId"
FROM "ql3"."plugin_package_workflow_admissions"
WHERE run_id = $1
`, [command.runId]);
`,
[command.runId],
);
const target = admission.rows[0];
if (
admission.rows.length !== 1 ||
@@ -225,65 +451,82 @@ export class PostgresClusterRunCancellationRepository
}
const current = run.rows[0]!;
const currentStatus = runStatus(current);
let result: Readonly<ClusterRunCancellationResult>;
if (TERMINAL.has(currentStatus)) {
return cancellationResult('already_terminal', command, current);
}
if (optionalInteger(current, 'cancelRequestedAtMs') !== undefined) {
return cancellationResult('already_requested', command, current);
}
if (optionalText(current, 'cancelReason') !== undefined) {
result = cancellationResult('already_terminal', command, current);
} else if (
optionalInteger(current, 'cancelRequestedAtMs') !== undefined
) {
result = cancellationResult('already_requested', command, current);
} else if (optionalText(current, 'cancelReason') !== undefined) {
throw new TypeError('PostgreSQL Run cancellation intent is invalid');
} else {
const runVersion = integer(current, 'runVersion');
const eventSequence = integer(current, 'eventSequence');
if (runVersion >= 2_147_483_647 || eventSequence >= 2_147_483_647) {
throw new TypeError('PostgreSQL Run cancellation counter overflowed');
}
const mutationObservedAtMs =
observedAtMs ?? (await databaseNow(client));
const updated = await client.query<Row>(
`
UPDATE "ql3"."runs"
SET cancel_requested_at_ms = $2, cancel_reason = 'user',
version = $3, event_sequence = $4
WHERE id = $1 AND version = $5 AND cancel_requested_at_ms IS NULL
RETURNING project_id AS "projectId", status AS "runStatus",
version AS "runVersion", event_sequence AS "eventSequence",
cancel_requested_at_ms AS "cancelRequestedAtMs",
cancel_reason AS "cancelReason"
`,
[
command.runId,
mutationObservedAtMs,
runVersion + 1,
eventSequence + 1,
runVersion,
],
);
if (updated.rows.length !== 1) {
throw new ClusterRunCancellationFenceRejectedError('state_mismatch');
}
await client.query(
`
INSERT INTO "ql3"."run_events" (
id, run_id, sequence, type, dedupe_key, actor_type, actor_id,
attempt_id, step_run_id, payload, created_at_ms
) VALUES ($1, $2, $3, 'run.cancel_requested', $4, $5, $6,
NULL, NULL, $7::jsonb, $8)
`,
[
command.eventId,
command.runId,
eventSequence + 1,
`user-cancel:${command.mutationId}`,
command.subject.type,
command.subject.id,
JSON.stringify({
reason: 'user',
mutation_id: command.mutationId,
policy_fence: {
project_version: command.policyFence.projectVersion,
binding_version: command.policyFence.bindingVersion,
},
}),
mutationObservedAtMs,
],
);
result = cancellationResult('accepted', command, updated.rows[0]!);
}
const runVersion = integer(current, 'runVersion');
const eventSequence = integer(current, 'eventSequence');
if (runVersion >= 2_147_483_647 || eventSequence >= 2_147_483_647) {
throw new TypeError('PostgreSQL Run cancellation counter overflowed');
if (confirmedAudit) {
await recordAllowedAudit(
client,
command,
confirmedAudit,
observedAtMs!,
);
}
const observedAtMs = await databaseNow(client);
const updated = await client.query<Row>(`
UPDATE "ql3"."runs"
SET cancel_requested_at_ms = $2, cancel_reason = 'user',
version = $3, event_sequence = $4
WHERE id = $1 AND version = $5 AND cancel_requested_at_ms IS NULL
RETURNING project_id AS "projectId", status AS "runStatus",
version AS "runVersion", event_sequence AS "eventSequence",
cancel_requested_at_ms AS "cancelRequestedAtMs",
cancel_reason AS "cancelReason"
`, [
command.runId,
observedAtMs,
runVersion + 1,
eventSequence + 1,
runVersion,
]);
if (updated.rows.length !== 1) {
throw new ClusterRunCancellationFenceRejectedError('state_mismatch');
}
await client.query(`
INSERT INTO "ql3"."run_events" (
id, run_id, sequence, type, dedupe_key, actor_type, actor_id,
attempt_id, step_run_id, payload, created_at_ms
) VALUES ($1, $2, $3, 'run.cancel_requested', $4, $5, $6,
NULL, NULL, $7::jsonb, $8)
`, [
command.eventId,
command.runId,
eventSequence + 1,
`user-cancel:${command.mutationId}`,
command.subject.type,
command.subject.id,
JSON.stringify({
reason: 'user',
mutation_id: command.mutationId,
policy_fence: {
project_version: command.policyFence.projectVersion,
binding_version: command.policyFence.bindingVersion,
},
}),
observedAtMs,
]);
return cancellationResult('accepted', command, updated.rows[0]!);
return result;
});
}
@@ -15,14 +15,15 @@ export interface PostgresSchemaContractFunction {
export interface PostgresSchemaContract {
readonly schema: 'ql3';
readonly contractName: 'control-core';
readonly contractVersion: 55;
readonly migrationId: 'pg-0056-run-management-boundary';
readonly contractVersion: 56;
readonly migrationId: 'pg-0057-run-management-stop-boundary';
readonly minimumServerMajor: 16;
readonly maximumServerMajor: 18;
readonly capabilities: Readonly<{
run_core: 1;
run_attempt_log_retention: 1;
run_management_boundary: 1;
run_management_stop: 1;
run_dispatch_lease: 1;
run_retry_policy: 1;
project_policy: 1;
@@ -102,8 +103,8 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract =
Object.freeze({
schema: 'ql3',
contractName: 'control-core',
contractVersion: 55,
migrationId: 'pg-0056-run-management-boundary',
contractVersion: 56,
migrationId: 'pg-0057-run-management-stop-boundary',
minimumServerMajor: 16,
maximumServerMajor: 18,
capabilities: Object.freeze({
@@ -145,6 +146,7 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract =
run_core: 1,
run_attempt_log_retention: 1,
run_management_boundary: 1,
run_management_stop: 1,
run_dispatch_lease: 1,
run_retry_policy: 1,
security_audit: 1,
@@ -119,6 +119,11 @@ interface FunctionPrivilegeRow extends Record<string, unknown> {
isOwner: unknown;
}
interface ColumnPrivilegeRow extends Record<string, unknown> {
columnName: unknown;
updateAllowed: unknown;
}
const REQUIRED_RUNTIME_PRIVILEGES = Object.freeze({
schema_migrations: Object.freeze({
select: true,
@@ -1315,13 +1320,13 @@ const REQUIRED_AUTOMATION_MANAGER_PRIVILEGES: RequiredPrivileges =
update: true,
}
: name === 'security_audit_events' ||
name === 'task_definition_revisions' ||
name === 'task_execution_revisions' ||
name === 'trigger_revisions'
name === 'task_definition_revisions' ||
name === 'task_execution_revisions' ||
name === 'trigger_revisions'
? { ...NO_TABLE_PRIVILEGES, select: true, insert: true }
: name === 'task_definitions' ||
name === 'triggers' ||
name === 'trigger_schedules'
name === 'triggers' ||
name === 'trigger_schedules'
? {
...NO_TABLE_PRIVILEGES,
select: true,
@@ -1376,9 +1381,9 @@ const REQUIRED_RUN_MANAGER_PRIVILEGES: RequiredPrivileges = Object.freeze(
name === 'task_execution_revisions'
? { ...NO_TABLE_PRIVILEGES, select: true }
: name === 'runs' ||
name === 'run_attempts' ||
name === 'run_events' ||
name === 'security_audit_events'
name === 'run_attempts' ||
name === 'run_events' ||
name === 'security_audit_events'
? { ...NO_TABLE_PRIVILEGES, select: true, insert: true }
: name === 'plugin_package_identity_keyset_ledger'
? {
@@ -1415,7 +1420,7 @@ const REQUIRED_WORKER_CREDENTIAL_MANAGER_PRIVILEGES: RequiredPrivileges =
update: true,
}
: name === 'worker_credential_management_quota_buckets' ||
name === 'plugin_package_identity_keyset_ledger'
name === 'plugin_package_identity_keyset_ledger'
? {
...NO_TABLE_PRIVILEGES,
select: true,
@@ -2042,6 +2047,56 @@ ORDER BY requested.function_name
}
}
async function assertRunManagerColumnPrivileges(
queryable: PostgresMigrationQueryable,
contract: PostgresSchemaContract,
): Promise<void> {
const run = contract.tables.find(({ name }) => name === 'runs');
if (!run) {
throw new PostgresSchemaReadinessError('run_manager_role_invalid', [
'missing-runs-contract',
]);
}
const result = await queryable.query<ColumnPrivilegeRow>(
`
SELECT
requested.column_name AS "columnName",
has_column_privilege(
current_user,
format('%I.%I', $1::text, 'runs'),
requested.column_name,
'UPDATE'
) AS "updateAllowed"
FROM unnest($2::text[]) AS requested(column_name)
ORDER BY requested.column_name
`.trim(),
[contract.schema, run.columns],
);
const allowed = new Set([
'cancel_requested_at_ms',
'cancel_reason',
'version',
'event_sequence',
]);
const actual = new Map(result.rows.map((row) => [row.columnName, row]));
const findings: string[] = [];
for (const columnName of run.columns) {
const row = actual.get(columnName);
if (!row || row.updateAllowed !== allowed.has(columnName)) {
findings.push(`column-update-privilege:runs.${columnName}`);
}
}
if (actual.size !== run.columns.length) {
findings.push('column-privilege-row-count:runs');
}
if (findings.length > 0) {
throw new PostgresSchemaReadinessError(
'run_manager_role_invalid',
sorted(findings),
);
}
}
export async function assertPostgresSchemaReady(
queryable: PostgresMigrationQueryable,
contract: PostgresSchemaContract = postgresqlControlSchemaContract,
@@ -2177,6 +2232,7 @@ export async function assertPostgresRunManagerSchemaReady(
REQUIRED_RUN_MANAGER_FUNCTION_PRIVILEGES,
'run_manager_role_invalid',
);
await assertRunManagerColumnPrivileges(queryable, contract);
return Object.freeze({
ready: true,
...server,