diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index 347ac741..16417d81 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -11,6 +11,20 @@ 最新增量证据（2026-08-12）： +- D-296/ADR-0384（已接受） + Cluster Run Management Plane 已在既有内聚领域内增加强认证 `run.stop`，不新增 package、进程、listener、Pool、timer、watcher、cache、 + sidecar 或部署 overlay。`run.retry | run.stop` 共用 mTLS、purpose-bound OIDC、五分钟内 `multi_factor|hardware` User、固定 + `/api/v3/runs/management` route 与 `operations/run-management` opt-in workload；默认 Edge、Standalone 和 Cluster base 的资源与 + 依赖闭包不变。服务端生成 canonical `qinglong/run-cancellation@v1` Event identity，并在单个 `SERIALIZABLE` 事务中重验 + Project/RoleBinding fence、写 durable cancellation intent/Event 和 `run.stop` allowed audit；相同 mutation 精确重放，不同 mutation + 不覆写，实际终态仍由既有 cancellation dispatch/fencing 收敛。PostgreSQL capability v56 / migration + `pg-0057-run-management-stop-boundary` 只向 `ql3_run_manager` 授予 Runs 的 + `cancel_requested_at_ms/cancel_reason/version/event_sequence` 四列 UPDATE，表级 UPDATE 和 `status` 列 UPDATE 均保持禁止。 + 本阶段坚持由领域内聚、依赖方向和部署生命周期决定 package 粒度，拒绝为 stop 新增浅 package。完整 18-package clean build/test + 与 backend 1,165 pass/2 conditional skip/0 fail；workspace 保持 18 package/1,071 source/1,053 nested，无 single-source 或 shallow + package；14 档 Local Profile artifact 全部 compatible，最小 Edge 仍为 53 loaded modules 且不含 Cluster/PostgreSQL。真实 PostgreSQL + 18.4 arm64 physical HA 通过 123 gates、timeline `1→2`，覆盖双 Pool exact replay、原子审计复制、cancellation convergence 和同步冗余 + 恢复后的 promotion stop；报告 SHA-256 `2e5759d3b5e62cd571f6c31450aec0d7f611fa8cafd727b7bb25471792e83c29`，离线审计零 finding。 - D-295/ADR-0383（已接受） Cluster 手动 `run.retry` 已从“仅有 PostgreSQL authority”推进到独立强认证产品面：能力内聚在既有 `@qinglong/cluster-admin` 的 `run-management/` 目录，不新增 workspace package；只有 `QL3_PROFILE=cluster-admin` 且显式启用时才读取 mTLS/CRL、 diff --git a/docs/adr/ADR-0383-strong-cluster-run-management-plane.md b/docs/adr/ADR-0383-strong-cluster-run-management-plane.md index 7005933a..30e08a60 100644 --- a/docs/adr/ADR-0383-strong-cluster-run-management-plane.md +++ b/docs/adr/ADR-0383-strong-cluster-run-management-plane.md @@ -6,6 +6,8 @@ - 前置决策：ADR-0039、ADR-0056、ADR-0356、ADR-0364、ADR-0366、ADR-0381、ADR-0382 - Supersedes：ADR-0382 中“复用 `ql3_runtime` 且不新增 role/migration”的产品装配决策；ADR-0382 的共享语义与 PostgreSQL 原子事务仍有效 +> 后续说明：ADR-0384 在不改变本 ADR 的进程、认证、部署与 package 边界下增加 `run.stop`，并 supersede 本文“`ql3_run_manager` 不更新既有 Run”的最小权限定义；该角色改为只允许更新取消意图所需的四个列。 + ## 上下文 ADR-0382 已证明 PostgreSQL 手动 Run retry 的原子语义与 HA 收敛，但现有 Cluster Control bearer 只能建立 `single_factor` User，不能承载会再次执行外部副作用的人工恢复操作。直接把 repository 接进通用 Cluster HTTP 会混合普通控制面与强人类认证 authority；继续使用 `ql3_runtime` 又会使常驻 runtime 持有本不需要的人工恢复权限。 diff --git a/docs/adr/ADR-0384-strong-cluster-run-stop-management.md b/docs/adr/ADR-0384-strong-cluster-run-stop-management.md new file mode 100644 index 00000000..870932f2 --- /dev/null +++ b/docs/adr/ADR-0384-strong-cluster-run-stop-management.md @@ -0,0 +1,74 @@ +# ADR-0384：强认证的 Cluster Run Stop Management + +- 状态：Accepted +- 日期：2026-08-12 +- 关联 RFC：QL-RFC-0001 D-296 +- 前置决策：ADR-0005、ADR-0039、ADR-0056、ADR-0372、ADR-0383 +- Supersedes：ADR-0383 中“`ql3_run_manager` 没有更新既有 Run 权限”的最小权限定义；其余进程、认证、部署与 package 决策保持有效 + +## 上下文 + +ADR-0383 已建立独立、强认证、显式启用的 Cluster Run Management Plane，但产品面只暴露 `run.retry`。Cluster Control 已有内部 cancellation repository，直接把它接到产品 route 会留下三个缺口：允许审计与取消意图不在同一事务、专用管理角色没有精确写权限、调用方可能绕过 Run Management 的 purpose-bound OIDC 与五分钟强认证。 + +QingLong 同时面向低资源路由设备和多节点集群。本能力不能给 Edge/Standalone 增加常驻成本，也不能为单个命令新增浅 package、独立进程、连接池或部署 overlay。 + +## 决策 + +### 1. 扩展既有内聚领域，不新增 package 或进程 + +`run.stop` 与 `run.retry` 复用 `@qinglong/cluster-admin/run-management` 的 service、discriminated transport、HTTPS route、client、OIDC keyset、mTLS listener、PostgreSQL Pool 和 `operations/run-management` 部署。PostgreSQL adapter 继续由 `@qinglong/cluster-postgres/run-manager` 发布。 + +默认 Edge、Standalone、Cluster base overlay 均不加载该能力；关闭 Run Management 时仍为零新增进程、listener、Pool、timer、watcher、cache、sidecar 与 Cluster dependency。这里以“领域内聚 + 独立部署生命周期”决定 package 粒度，不以文件数量决定 package 数量。 + +### 2. 固定强认证命令与服务端身份 + +transport 只接受 `operation=run.stop`、Project/Run identity、UUID mutation identity 和低敏 request/audit identity，不接受 caller 提供 Event ID、时间、取消原因、Run version 或状态。服务端生成 canonical `qinglong/run-cancellation@v1` Event ID，并固定 `cancel_reason=user`。 + +操作必须同时满足 mTLS、`run-management` purpose-bound OIDC、五分钟内 `multi_factor|hardware` User 和 `run.stop` Policy。route admission 与 PostgreSQL 事务分别重验身份和 Project/RoleBinding fence；不存在继续遮蔽，终态或 durable fence 漂移稳定映射为 conflict。 + +### 3. 取消意图与允许审计原子提交 + +PostgreSQL repository 在一个 `SERIALIZABLE` 事务中使用数据库时钟，锁定 Run，重验 `lock_run_management_policy_fence`，写入 `cancel_requested_at_ms`、`cancel_reason`、Run `version`、`event_sequence`，追加 immutable cancellation Event 和 `run.stop` allowed security audit。任一写入失败都整体回滚；拒绝或不可用由 service 写独立 failure audit。 + +相同 mutation 的重放必须返回 `already_requested`，并验证已有 Event 与 allowed audit 的精确语义；不同 mutation 不能覆写既有取消意图。terminal Run 不接受新的 stop。该操作只记录 durable intent，实际 Attempt/Run 收敛仍由既有 cancellation dispatch、lease/fencing 与 recovery authority 完成。 + +### 4. PostgreSQL 权限按列收窄 + +Migration `pg-0057-run-management-stop-boundary` 把 control-core contract 提升到 v56，并添加 `run_management_stop` capability。`ql3_run_manager` 不获得 Runs 表级 UPDATE，只获得以下四列的 column-level UPDATE： + +- `cancel_requested_at_ms` +- `cancel_reason` +- `version` +- `event_sequence` + +readiness 同时证明四列可更新、`status` 不可更新、表级 UPDATE 仍为 false。角色不得改变 Run 状态、Project/RoleBinding、Task、execution revision，且继续没有 DELETE、migration、Worker、AI 或 Approval authority。 + +## 验收 + +- service/transport/client 测试覆盖 stop、强身份、命令 shape、response drift、原子 allowed audit 与稳定错误映射； +- PostgreSQL 测试覆盖 v56 checksum/capability、精确列权限、事务顺序、exact replay、fence 漂移和回滚； +- 真实 PostgreSQL 18.4 physical HA 以两个独立 `ql3_run_manager` Pool 证明跨 Pool replay、允许审计同步复制与 promotion 后写入； +- 完整 18-package clean build/test、backend、dependency/package/deployment/Profile artifact、GitNexus change scope 全部通过后才允许阶段性提交。 + +## 被否决的替代方案 + +1. **新增 `@qinglong/run-stop` package**：没有独立依赖、制品或部署生命周期，只会制造浅 package。 +2. **新增 stop listener/Pool/overlay**：与 retry 共享同一认证和资源边界，会重复常驻成本。 +3. **授予 Runs 表级 UPDATE**：会允许管理角色修改状态与其他控制字段，超出产品操作所需。 +4. **先写 allowed audit、再写取消意图**：可能留下“审计显示成功但意图未提交”的错误事实。 +5. **由调用方提交 Event ID 或 cancel reason**：扩大重放和语义漂移表面。 + +## 验收证据（2026-08-12） + +- `@qinglong/cluster-admin`：284 pass、2 个外部集成条件 skip、0 fail；`@qinglong/cluster-postgres`：311 pass、1 个外部数据库条件 skip、0 fail；完整 18-package clean build/test 全部退出 0； +- backend 1,167 项中 1,165 pass、2 个环境条件 skip、0 fail；package/dependency/Edge import/Cluster deployment/CloudNativePG/Local image 静态门均 compatible、零 finding； +- workspace 保持 18 个 package、1,071 个 source、1,053 个 nested，`singleSourcePackages=[]`、`shallowSourcePackages=[]`；`cluster-postgres` 为 153/152，v56 migration 归入既有 `run-management` 领域，没有新增浅 package； +- 14 档 Edge/Standalone Profile artifact 全部 compatible；最小 Edge 仍仅包含 Local SQLite、Runtime Core 与 `semver`，产物 2,459,624 bytes、53 个 loaded module，未引入 Cluster/PostgreSQL； +- PostgreSQL 18.4 arm64 physical HA 通过 123 gates、timeline `1→2`，证明列级 UPDATE readiness、双 Run Manager Pool exact replay、allowed audit 同步复制、既有 cancellation convergence、旧主 fencing，以及恢复 `remote_apply` 同步冗余后的 promotion stop；报告 SHA-256 `2e5759d3b5e62cd571f6c31450aec0d7f611fa8cafd727b7bb25471792e83c29`，独立离线审计 `compatible:true`、零 finding。 + +## 影响 + +- PostgreSQL schema contract 从 v55 升至 v56；启用新版 Run Management 前必须先运行 migration 和 readiness； +- Cluster Run Management route 成为 `retry | stop` 的严格判别联合，但既有 retry envelope/response 保持兼容； +- 低配设备默认资源与依赖闭包不变；集群只在已选择的 Run Management workload 内增加同一 Pool 上的一类短事务； +- UI 后续可以复用同一 transport 展示 durable cancellation intent，但不得直接写 Run 状态或绕过强认证。 diff --git a/docs/adr/README.md b/docs/adr/README.md index e68eb637..6c24c4d9 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -387,6 +387,7 @@ | [ADR-0381](./ADR-0381-strong-local-manual-run-retry.md) | 强认证的 Local 手动 Run retry | Accepted | | [ADR-0382](./ADR-0382-postgresql-cluster-manual-run-retry-authority.md) | PostgreSQL Cluster 手动 Run retry 原子 Authority | Accepted | | [ADR-0383](./ADR-0383-strong-cluster-run-management-plane.md) | 强认证的 Cluster Run Management Plane 与专用数据库角色 | Accepted | +| [ADR-0384](./ADR-0384-strong-cluster-run-stop-management.md) | 强认证的 Cluster Run Stop Management 与列级数据库权限 | Accepted | ## 规则 diff --git a/packages/ql3-cluster-admin/src/run-management/runManagement.ts b/packages/ql3-cluster-admin/src/run-management/runManagement.ts index b4043c91..8a09ddce 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagement.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagement.ts @@ -1,12 +1,20 @@ import { randomUUID } from 'node:crypto'; import { + PostgresClusterRunCancellationRepository, PostgresProjectPolicyRepository, PostgresRunManualRetryRepository, PostgresSecurityAuditRepository, } from '@qinglong/cluster-postgres/run-manager'; import type { PostgresPool } from '@qinglong/runtime-core'; import { ProjectPolicyEngine } from '@qinglong/runtime-core/project-policy'; +import { + ClusterRunCancellationFenceRejectedError, + ClusterRunCancellationNotFoundError, + ClusterRunCancellationUnavailableError, + InvalidClusterRunCancellationError, + type ClusterRunCancellationResult, +} from '@qinglong/runtime-core/run-cancellation'; import { InvalidRunManualRetryError, RunManualRetryFenceRejectedError, @@ -39,10 +47,23 @@ export interface ClusterRunManagementRetryRequest { readonly principal: Readonly; } +export interface ClusterRunManagementStopRequest { + readonly projectId: string; + readonly runId: string; + readonly mutationId: string; + readonly requestId: string; + readonly auditEventId: string; + readonly failureAuditEventId: string; + readonly principal: Readonly; +} + export interface ClusterRunManagementService { retry( request: Readonly, ): Promise>; + stop( + request: Readonly, + ): Promise>; } export interface ClusterRunManagementOptions { @@ -107,7 +128,7 @@ export class ClusterRunManagementUnavailableError extends Error { } } -function exactRequest( +function exactRetryRequest( value: unknown, ): asserts value is Readonly { if ( @@ -133,6 +154,30 @@ function exactRequest( } } +function exactStopRequest( + value: unknown, +): asserts value is Readonly { + if ( + !value || + typeof value !== 'object' || + Array.isArray(value) || + Object.keys(value).sort().join('\0') !== + [ + 'auditEventId', + 'failureAuditEventId', + 'mutationId', + 'principal', + 'projectId', + 'requestId', + 'runId', + ] + .sort() + .join('\0') + ) { + throw new ClusterRunManagementRequestError(); + } +} + function validUuid(value: unknown): value is string { return typeof value === 'string' && UUID_PATTERN.test(value); } @@ -144,6 +189,12 @@ function failureReason(error: unknown): string { if (error instanceof RunManualRetryNotFoundError) return 'run_not_found'; if (error instanceof RunManualRetryRateLimitedError) return 'rate_limited'; if (error instanceof RunManualRetryFenceRejectedError) return error.reason; + if (error instanceof ClusterRunCancellationNotFoundError) { + return 'run_not_found'; + } + if (error instanceof ClusterRunCancellationFenceRejectedError) { + return error.reason; + } return 'management_unavailable'; } @@ -162,7 +213,8 @@ export function createClusterRunManagementService( typeof options.pool.query !== 'function' || typeof options.pool.connect !== 'function' || (options.now !== undefined && typeof options.now !== 'function') || - (options.randomUuid !== undefined && typeof options.randomUuid !== 'function') + (options.randomUuid !== undefined && + typeof options.randomUuid !== 'function') ) { throw new ClusterRunManagementConfigurationError(); } @@ -172,11 +224,14 @@ export function createClusterRunManagementService( new PostgresProjectPolicyRepository(options.pool), ); const retries = new PostgresRunManualRetryRepository(options.pool); + const cancellations = new PostgresClusterRunCancellationRepository( + options.pool, + ); const audit = new PostgresSecurityAuditRepository(options.pool); return Object.freeze({ async retry(requestValue: Readonly) { - exactRequest(requestValue); + exactRetryRequest(requestValue); const observedAtMs = now(); let principal: Readonly; if ( @@ -250,7 +305,8 @@ export function createClusterRunManagementService( } catch (auditError) { throw new ClusterRunManagementUnavailableError({ cause: auditError }); } - if (error instanceof ClusterRunManagementAuthorizationError) throw error; + if (error instanceof ClusterRunManagementAuthorizationError) + throw error; if (error instanceof InvalidRunManualRetryError) { throw new ClusterRunManagementRequestError(); } @@ -269,5 +325,92 @@ export function createClusterRunManagementService( throw new ClusterRunManagementUnavailableError({ cause: error }); } }, + async stop(requestValue: Readonly) { + exactStopRequest(requestValue); + const observedAtMs = now(); + let principal: Readonly; + if ( + !Number.isSafeInteger(observedAtMs) || + observedAtMs < 0 || + !IDENTIFIER_PATTERN.test(requestValue.projectId) || + !IDENTIFIER_PATTERN.test(requestValue.runId) || + !IDENTIFIER_PATTERN.test(requestValue.requestId) || + !validUuid(requestValue.mutationId) || + !validUuid(requestValue.auditEventId) || + !validUuid(requestValue.failureAuditEventId) || + requestValue.auditEventId === requestValue.failureAuditEventId + ) { + throw new ClusterRunManagementRequestError(); + } + try { + principal = normalizeSecurityPrincipal( + requestValue.principal, + observedAtMs, + ); + } catch { + throw new ClusterRunManagementRequestError(); + } + + let fence: Readonly | null = null; + try { + const decision = await policy.authorize( + principal, + requestValue.projectId, + 'run.stop', + ); + fence = decision.fence; + if ( + decision.effect !== 'allow' || + !fence || + fence.bindingVersion === null + ) { + throw new ClusterRunManagementAuthorizationError(); + } + return await cancellations.requestUserCancellationAudited({ + projectId: requestValue.projectId, + runId: requestValue.runId, + mutationId: requestValue.mutationId, + eventId: createId(), + requestId: requestValue.requestId, + auditEventId: requestValue.auditEventId, + principal, + policyFence: fence, + }); + } catch (error) { + try { + await audit.record( + normalizeSecurityAuditRecord({ + eventId: requestValue.failureAuditEventId, + requestId: requestValue.requestId, + operationId: 'run.stop', + projectId: requestValue.projectId, + subject: principal.subject, + authenticationId: principal.authenticationId, + outcome: 'denied', + reasons: [failureReason(error)], + fence, + occurredAtMs: observedAtMs, + }), + ); + } catch (auditError) { + throw new ClusterRunManagementUnavailableError({ cause: auditError }); + } + if (error instanceof ClusterRunManagementAuthorizationError) + throw error; + if (error instanceof InvalidClusterRunCancellationError) { + throw new ClusterRunManagementRequestError(); + } + if (error instanceof ClusterRunCancellationNotFoundError) { + throw new ClusterRunManagementTargetUnavailableError(); + } + if (error instanceof ClusterRunCancellationFenceRejectedError) { + throw new ClusterRunManagementConflictError(); + } + if (error instanceof ClusterRunCancellationUnavailableError) { + throw new ClusterRunManagementUnavailableError({ cause: error }); + } + throw new ClusterRunManagementUnavailableError({ cause: error }); + } + }, }); } diff --git a/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts b/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts index 8708e489..32e91a77 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts @@ -2,6 +2,10 @@ import { RUN_MANUAL_RETRY_SCHEMA, normalizeRunManualRetryResult, } from '@qinglong/runtime-core/run-manual-retry'; +import { + RUN_CANCELLATION_SCHEMA, + normalizeRunCancellationResult, +} from '@qinglong/runtime-core/run-cancellation'; import { ClusterPluginPackageManagementClientRequestError, executeClusterAuthenticatedManagementClient, @@ -28,7 +32,10 @@ function invalid(): never { throw new ClusterPluginPackageManagementClientRequestError(); } -function exact(value: unknown, keys: readonly string[]): Record { +function exact( + value: unknown, + keys: readonly string[], +): Record { if (!value || typeof value !== 'object' || Array.isArray(value)) invalid(); const actual = Object.keys(value as object).sort(); const expected = [...keys].sort(); @@ -45,37 +52,80 @@ export function validateClusterRunManagementClientResult( value: unknown, command: Readonly, ): Readonly { - const envelope = exact(value, ['schemaVersion', 'operation', 'retry']); - if (envelope.schemaVersion !== 1 || envelope.operation !== 'run.retry') invalid(); - const retry = exact(envelope.retry, [ + if (command.operation === 'run.retry') { + const envelope = exact(value, ['schemaVersion', 'operation', 'retry']); + if ( + envelope.schemaVersion !== 1 || + envelope.operation !== command.operation + ) { + invalid(); + } + const retry = exact(envelope.retry, [ + 'schema', + 'status', + 'projectId', + 'sourceRunId', + 'sourceRunStatus', + 'sourceRunVersion', + 'runId', + 'retryOfRunId', + 'taskId', + 'taskRevision', + 'attemptId', + 'runStatus', + 'runVersion', + 'eventSequence', + 'executorType', + 'executionRevisionDigest', + 'createdAtMs', + ]); + if (retry.schema !== RUN_MANUAL_RETRY_SCHEMA) invalid(); + try { + const { schema: _schema, ...result } = retry; + const normalized = normalizeRunManualRetryResult(result as never); + if ( + normalized.projectId !== command.request.projectId || + normalized.sourceRunId !== command.request.sourceRunId || + normalized.sourceRunVersion !== + command.request.body.expectedRunVersion || + normalized.sourceRunStatus !== command.request.body.expectedRunStatus || + normalized.executorType !== 'remote_worker' + ) { + invalid(); + } + } catch { + invalid(); + } + return Object.freeze( + envelope as unknown as ClusterRunManagementTransportResult, + ); + } + const envelope = exact(value, ['schemaVersion', 'operation', 'stop']); + if ( + envelope.schemaVersion !== 1 || + envelope.operation !== command.operation + ) { + invalid(); + } + const stop = exact(envelope.stop, [ 'schema', 'status', 'projectId', - 'sourceRunId', - 'sourceRunStatus', - 'sourceRunVersion', 'runId', - 'retryOfRunId', - 'taskId', - 'taskRevision', - 'attemptId', 'runStatus', 'runVersion', 'eventSequence', - 'executorType', - 'executionRevisionDigest', - 'createdAtMs', + ...(Object.hasOwn(envelope.stop as object, 'cancelRequestedAtMs') + ? ['cancelRequestedAtMs', 'cancelReason'] + : []), ]); - if (retry.schema !== RUN_MANUAL_RETRY_SCHEMA) invalid(); + if (stop.schema !== RUN_CANCELLATION_SCHEMA) invalid(); try { - const { schema: _schema, ...result } = retry; - const normalized = normalizeRunManualRetryResult(result as never); + const { schema: _schema, ...result } = stop; + const normalized = normalizeRunCancellationResult(result as never); if ( normalized.projectId !== command.request.projectId || - normalized.sourceRunId !== command.request.sourceRunId || - normalized.sourceRunVersion !== command.request.body.expectedRunVersion || - normalized.sourceRunStatus !== command.request.body.expectedRunStatus || - normalized.executorType !== 'remote_worker' + normalized.runId !== command.request.runId ) { invalid(); } diff --git a/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts b/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts index 68634180..fa1b6dee 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts @@ -3,6 +3,11 @@ import { parseRunManualRetryRequestBody, type RunManualRetryResponseBody, } from '@qinglong/runtime-core/run-manual-retry'; +import { + createRunCancellationResponseBody, + parseRunCancellationRequestBody, + type RunCancellationResponseBody, +} from '@qinglong/runtime-core/run-cancellation'; import { normalizeSecurityPrincipal, type SecurityPrincipal, @@ -14,7 +19,7 @@ const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; const STRONG_ASSURANCES = new Set(['multi_factor', 'hardware']); -export type ClusterRunManagementCommand = Readonly<{ +export type ClusterRunManagementRetryCommand = Readonly<{ schemaVersion: 1; operation: 'run.retry'; request: Readonly<{ @@ -32,12 +37,42 @@ export type ClusterRunManagementCommand = Readonly<{ }>; }>; -export type ClusterRunManagementTransportResult = Readonly<{ +export type ClusterRunManagementStopCommand = Readonly<{ + schemaVersion: 1; + operation: 'run.stop'; + request: Readonly<{ + projectId: string; + runId: string; + requestId: string; + auditEventId: string; + failureAuditEventId: string; + body: Readonly<{ + schema: 'qinglong/run-cancellation@v1'; + mutationId: string; + }>; + }>; +}>; + +export type ClusterRunManagementCommand = + | ClusterRunManagementRetryCommand + | ClusterRunManagementStopCommand; + +export type ClusterRunManagementRetryTransportResult = Readonly<{ schemaVersion: 1; operation: 'run.retry'; retry: Readonly; }>; +export type ClusterRunManagementStopTransportResult = Readonly<{ + schemaVersion: 1; + operation: 'run.stop'; + stop: Readonly; +}>; + +export type ClusterRunManagementTransportResult = + | ClusterRunManagementRetryTransportResult + | ClusterRunManagementStopTransportResult; + export interface ClusterRunManagementAuthentication { authenticate(): Promise | null>; } @@ -85,7 +120,10 @@ function invalid(): never { throw new ClusterRunManagementTransportRequestError(); } -function exact(value: unknown, keys: readonly string[]): Record { +function exact( + value: unknown, + keys: readonly string[], +): Record { if (!value || typeof value !== 'object' || Array.isArray(value)) invalid(); const actual = Object.keys(value as object).sort(); const expected = [...keys].sort(); @@ -112,30 +150,65 @@ export function normalizeClusterRunManagementCommand( value: unknown, ): Readonly { const envelope = exact(value, ['schemaVersion', 'operation', 'request']); - if (envelope.schemaVersion !== 1 || envelope.operation !== 'run.retry') invalid(); - const request = exact(envelope.request, [ - 'projectId', - 'sourceRunId', - 'requestId', - 'auditEventId', - 'failureAuditEventId', - 'body', - ]); - let body: ReturnType; - try { - body = parseRunManualRetryRequestBody(request.body); - } catch { - invalid(); - } + if (envelope.schemaVersion !== 1) invalid(); + const operation = envelope.operation; + if (operation !== 'run.retry' && operation !== 'run.stop') invalid(); + const request = exact( + envelope.request, + operation === 'run.retry' + ? [ + 'projectId', + 'sourceRunId', + 'requestId', + 'auditEventId', + 'failureAuditEventId', + 'body', + ] + : [ + 'projectId', + 'runId', + 'requestId', + 'auditEventId', + 'failureAuditEventId', + 'body', + ], + ); const auditEventId = uuid(request.auditEventId); const failureAuditEventId = uuid(request.failureAuditEventId); if (auditEventId === failureAuditEventId) invalid(); + if (operation === 'run.retry') { + let body: ReturnType; + try { + body = parseRunManualRetryRequestBody(request.body); + } catch { + invalid(); + } + return Object.freeze({ + schemaVersion: 1, + operation, + request: Object.freeze({ + projectId: identifier(request.projectId), + sourceRunId: identifier(request.sourceRunId), + requestId: identifier(request.requestId), + auditEventId, + failureAuditEventId, + body, + }), + }); + } + let body: ReturnType; + try { + body = parseRunCancellationRequestBody(request.body); + } catch { + invalid(); + } + uuid(body.mutationId); return Object.freeze({ schemaVersion: 1, - operation: 'run.retry', + operation, request: Object.freeze({ projectId: identifier(request.projectId), - sourceRunId: identifier(request.sourceRunId), + runId: identifier(request.runId), requestId: identifier(request.requestId), auditEventId, failureAuditEventId, @@ -144,10 +217,12 @@ export function normalizeClusterRunManagementCommand( }); } -export function createClusterRunManagementTransport(options: Readonly<{ - service: ClusterRunManagementService; - now?: () => number; -}>): Readonly { +export function createClusterRunManagementTransport( + options: Readonly<{ + service: ClusterRunManagementService; + now?: () => number; + }>, +): Readonly { if ( !options || typeof options !== 'object' || @@ -155,6 +230,7 @@ export function createClusterRunManagementTransport(options: Readonly<{ Object.keys(options).some((key) => key !== 'service' && key !== 'now') || !options.service || typeof options.service.retry !== 'function' || + typeof options.service.stop !== 'function' || (options.now !== undefined && typeof options.now !== 'function') ) { throw new ClusterRunManagementTransportConfigurationError(); @@ -183,7 +259,10 @@ export function createClusterRunManagementTransport(options: Readonly<{ } let principal: Readonly; try { - principal = normalizeSecurityPrincipal(candidate as SecurityPrincipal, now()); + principal = normalizeSecurityPrincipal( + candidate as SecurityPrincipal, + now(), + ); } catch { throw new ClusterRunManagementTransportAuthenticationError(); } @@ -193,12 +272,28 @@ export function createClusterRunManagementTransport(options: Readonly<{ ) { throw new ClusterRunManagementTransportAuthenticationError(); } - const result = await options.service.retry({ + if (command.operation === 'run.retry') { + const result = await options.service.retry({ + projectId: command.request.projectId, + sourceRunId: command.request.sourceRunId, + mutationId: command.request.body.mutationId, + expectedRunVersion: command.request.body.expectedRunVersion, + expectedRunStatus: command.request.body.expectedRunStatus, + requestId: command.request.requestId, + auditEventId: command.request.auditEventId, + failureAuditEventId: command.request.failureAuditEventId, + principal, + }); + return Object.freeze({ + schemaVersion: 1, + operation: command.operation, + retry: createRunManualRetryResponseBody(result), + }); + } + const result = await options.service.stop({ projectId: command.request.projectId, - sourceRunId: command.request.sourceRunId, + runId: command.request.runId, mutationId: command.request.body.mutationId, - expectedRunVersion: command.request.body.expectedRunVersion, - expectedRunStatus: command.request.body.expectedRunStatus, requestId: command.request.requestId, auditEventId: command.request.auditEventId, failureAuditEventId: command.request.failureAuditEventId, @@ -206,8 +301,8 @@ export function createClusterRunManagementTransport(options: Readonly<{ }); return Object.freeze({ schemaVersion: 1, - operation: 'run.retry', - retry: createRunManualRetryResponseBody(result), + operation: command.operation, + stop: createRunCancellationResponseBody(result), }); }, }); diff --git a/packages/ql3-cluster-admin/test/runManagement.test.cjs b/packages/ql3-cluster-admin/test/runManagement.test.cjs index 8072fb15..637b4d8c 100644 --- a/packages/ql3-cluster-admin/test/runManagement.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagement.test.cjs @@ -67,7 +67,8 @@ function fixture(role = 'operator') { async query(sql, params = []) { const text = sql.replace(/\s+/g, ' ').trim(); calls.push({ scope: 'pool', sql: text, params }); - if (text.includes('LEFT JOIN LATERAL')) return { rows: [policyRow(role)] }; + if (text.includes('LEFT JOIN LATERAL')) + return { rows: [policyRow(role)] }; if (text.startsWith('INSERT INTO "ql3"."security_audit_events"')) { return { rows: [], rowCount: 1 }; } @@ -83,40 +84,89 @@ function fixture(role = 'operator') { text === 'COMMIT' || text === 'ROLLBACK' || text.startsWith('SELECT set_config') - ) return { rows: [], rowCount: 0 }; + ) + return { rows: [], rowCount: 0 }; if (text.includes('statement_timestamp()')) { return { rows: [{ nowMs: NOW }], rowCount: 1 }; } if (text.includes('lock_run_management_policy_fence')) { return { rows: [{ matches: true }], rowCount: 1 }; } + if (text.includes('FROM "ql3"."runs" WHERE id = $1 FOR UPDATE')) { + return { + rows: [ + { + projectId: 'project-1', + runStatus: 'running', + runVersion: 4, + eventSequence: 6, + cancelRequestedAtMs: null, + cancelReason: null, + }, + ], + rowCount: 1, + }; + } + if (text.startsWith('UPDATE "ql3"."runs"')) { + return { + rows: [ + { + projectId: 'project-1', + runStatus: 'running', + runVersion: 5, + eventSequence: 7, + cancelRequestedAtMs: NOW, + cancelReason: 'user', + }, + ], + rowCount: 1, + }; + } + if ( + text.startsWith('INSERT INTO "ql3"."security_audit_events"') && + text.includes('RETURNING event_id') + ) { + return { rows: [{ eventId: request().auditEventId }], rowCount: 1 }; + } if (text.includes('idempotency_key = $2')) return { rows: [] }; if (text.includes('WHERE run.id = $1')) { return { - rows: [{ - projectId: 'project-1', - taskId: 'task-1', - taskRevision: TASK_REVISION, - taskName: 'Task 1', - taskSnapshotRef: TASK_REVISION, - parentRunId: null, - triggerType: 'task_start', - executionOwner: 'runtime', - inputRef: null, - priority: 1, - runStatus: 'failed', - runVersion: 7, - attemptExecutorType: 'remote_worker', - }], + rows: [ + { + projectId: 'project-1', + taskId: 'task-1', + taskRevision: TASK_REVISION, + taskName: 'Task 1', + taskSnapshotRef: TASK_REVISION, + parentRunId: null, + triggerType: 'task_start', + executionOwner: 'runtime', + inputRef: null, + priority: 1, + runStatus: 'failed', + runVersion: 7, + attemptExecutorType: 'remote_worker', + }, + ], }; } if (text.includes('FROM "ql3"."task_definitions"')) { return { rows: [{ enabled: true }] }; } if (text.includes('task_execution_revisions')) { - return { rows: [{ sourceContentDigest: SOURCE_DIGEST, contentDigest: EXECUTION_DIGEST }] }; + return { + rows: [ + { + sourceContentDigest: SOURCE_DIGEST, + contentDigest: EXECUTION_DIGEST, + }, + ], + }; } - if (text.startsWith('SELECT') && text.includes("trigger_type = 'run_manual_retry'")) { + if ( + text.startsWith('SELECT') && + text.includes("trigger_type = 'run_manual_retry'") + ) { return { rows: [] }; } if (text.startsWith('INSERT INTO')) return { rows: [], rowCount: 1 }; @@ -143,7 +193,9 @@ test('authorizes run.retry and keeps all generated aggregate identities server-s assert.equal(result.status, 'accepted'); assert.equal(result.runId, GENERATED[0]); assert.equal(result.attemptId, GENERATED[1]); - const runInsert = calls.find(({ sql }) => sql.startsWith('INSERT INTO "ql3"."runs"')); + const runInsert = calls.find(({ sql }) => + sql.startsWith('INSERT INTO "ql3"."runs"'), + ); assert.equal(runInsert.params[0], GENERATED[0]); assert.equal(runInsert.params.includes(GENERATED[2]), false); assert.equal( @@ -154,11 +206,49 @@ test('authorizes run.retry and keeps all generated aggregate identities server-s test('denied policy writes only the caller-supplied failure audit', async () => { const { calls, service } = fixture('viewer'); - await assert.rejects(service.retry(request()), ClusterRunManagementAuthorizationError); + await assert.rejects( + service.retry(request()), + ClusterRunManagementAuthorizationError, + ); const audits = calls.filter(({ sql }) => sql.startsWith('INSERT INTO "ql3"."security_audit_events"'), ); assert.equal(audits.length, 1); assert.equal(audits[0].params[0], request().failureAuditEventId); - assert.equal(calls.some(({ scope }) => scope === 'client'), false); + assert.equal( + calls.some(({ scope }) => scope === 'client'), + false, + ); +}); + +test('authorizes run.stop and commits intent plus allowed audit together', async () => { + const { calls, service } = fixture(); + const stopRequest = { + projectId: 'project-1', + runId: 'run-1', + mutationId: '019f9500-0000-4000-8000-000000000021', + requestId: 'request-stop-1', + auditEventId: '019f9500-0000-4000-8000-000000000022', + failureAuditEventId: '019f9500-0000-4000-8000-000000000023', + principal: request().principal, + }; + const result = await service.stop(stopRequest); + assert.equal(result.status, 'accepted'); + assert.equal(result.cancelRequestedAtMs, NOW); + const event = calls.find( + ({ sql }) => + sql.startsWith('INSERT INTO "ql3"."run_events"') && + sql.includes('run.cancel_requested'), + ); + assert.equal(event.params[0], GENERATED[0]); + const allowedAudit = calls.find( + ({ sql }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"') && + sql.includes("'run.stop'"), + ); + assert.equal(allowedAudit.params[0], stopRequest.auditEventId); + assert.ok( + calls.findIndex(({ sql }) => sql.includes("'run.stop'")) < + calls.findIndex(({ sql }) => sql === 'COMMIT'), + ); }); diff --git a/packages/ql3-cluster-admin/test/runManagementClient.test.cjs b/packages/ql3-cluster-admin/test/runManagementClient.test.cjs index 5c9efb75..16f86078 100644 --- a/packages/ql3-cluster-admin/test/runManagementClient.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagementClient.test.cjs @@ -31,6 +31,22 @@ const command = normalizeClusterRunManagementCommand({ }, }); +const stopCommand = normalizeClusterRunManagementCommand({ + schemaVersion: 1, + operation: 'run.stop', + request: { + projectId: 'project-1', + runId: 'run-1', + requestId: 'request-stop-1', + auditEventId: '019f9400-0000-4000-8000-000000000021', + failureAuditEventId: '019f9400-0000-4000-8000-000000000022', + body: { + schema: 'qinglong/run-cancellation@v1', + mutationId: '019f9400-0000-4000-8000-000000000023', + }, + }, +}); + function response(overrides = {}) { return { schemaVersion: 1, @@ -59,7 +75,10 @@ function response(overrides = {}) { } test('validates one low-sensitive retry response against the request fence', () => { - assert.deepEqual(validateClusterRunManagementClientResult(response(), command), response()); + assert.deepEqual( + validateClusterRunManagementClientResult(response(), command), + response(), + ); }); test('rejects response target, execution placement and shape drift', () => { @@ -75,3 +94,35 @@ test('rejects response target, execution placement and shape drift', () => { ); } }); + +test('validates one low-sensitive stop response against the request target', () => { + const value = { + schemaVersion: 1, + operation: 'run.stop', + stop: { + schema: 'qinglong/run-cancellation@v1', + status: 'accepted', + projectId: 'project-1', + runId: 'run-1', + runStatus: 'running', + runVersion: 5, + eventSequence: 7, + cancelRequestedAtMs: 1_000_000, + cancelReason: 'user', + }, + }; + assert.deepEqual( + validateClusterRunManagementClientResult(value, stopCommand), + value, + ); + for (const drift of [ + { ...value, stop: { ...value.stop, projectId: 'project-2' } }, + { ...value, stop: { ...value.stop, runId: 'run-2' } }, + { ...value, operation: 'run.retry' }, + ]) { + assert.throws( + () => validateClusterRunManagementClientResult(drift, stopCommand), + ClusterPluginPackageManagementClientRequestError, + ); + } +}); diff --git a/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs b/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs index 0f1221d7..83c98957 100644 --- a/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs @@ -65,6 +65,38 @@ function retryResult() { }; } +function stopCommand(overrides = {}) { + return { + schemaVersion: 1, + operation: 'run.stop', + request: { + projectId: 'project-1', + runId: 'run-1', + requestId: 'request-stop-1', + auditEventId: '019f9300-0000-4000-8000-000000000021', + failureAuditEventId: '019f9300-0000-4000-8000-000000000022', + body: { + schema: 'qinglong/run-cancellation@v1', + mutationId: '019f9300-0000-4000-8000-000000000023', + }, + ...overrides, + }, + }; +} + +function stopResult() { + return { + status: 'accepted', + projectId: 'project-1', + runId: 'run-1', + runStatus: 'running', + runVersion: 5, + eventSequence: 7, + cancelRequestedAtMs: NOW, + cancelReason: 'user', + }; +} + test('routes one exact strong User retry and emits the shared response', async () => { const calls = []; const transport = createClusterRunManagementTransport({ @@ -74,6 +106,9 @@ test('routes one exact strong User retry and emits the shared response', async ( calls.push(request); return retryResult(); }, + async stop() { + return stopResult(); + }, }, }); const result = await transport.execute(command(), { @@ -92,6 +127,36 @@ test('routes one exact strong User retry and emits the shared response', async ( }); }); +test('routes one exact strong User stop and emits the shared response', async () => { + const calls = []; + const transport = createClusterRunManagementTransport({ + now: () => NOW, + service: { + async retry() { + return retryResult(); + }, + async stop(request) { + calls.push(request); + return stopResult(); + }, + }, + }); + const result = await transport.execute(stopCommand(), { + authenticate: async () => principal({ assurance: 'hardware' }), + }); + assert.equal(calls.length, 1); + assert.equal(calls[0].runId, 'run-1'); + assert.equal(calls[0].mutationId, stopCommand().request.body.mutationId); + assert.deepEqual(result, { + schemaVersion: 1, + operation: 'run.stop', + stop: { + schema: 'qinglong/run-cancellation@v1', + ...stopResult(), + }, + }); +}); + test('rejects weak or non-User identity before service authority', async () => { let called = false; const transport = createClusterRunManagementTransport({ @@ -101,6 +166,9 @@ test('rejects weak or non-User identity before service authority', async () => { called = true; return retryResult(); }, + async stop() { + return stopResult(); + }, }, }); await assert.rejects( @@ -121,7 +189,11 @@ test('rejects weak or non-User identity before service authority', async () => { test('rejects widened commands and ambiguous audit identity', () => { assert.throws( - () => normalizeClusterRunManagementCommand({ ...command(), principal: principal() }), + () => + normalizeClusterRunManagementCommand({ + ...command(), + principal: principal(), + }), ClusterRunManagementTransportRequestError, ); assert.throws( @@ -134,7 +206,18 @@ test('rejects widened commands and ambiguous audit identity', () => { assert.throws( () => normalizeClusterRunManagementCommand( - command({ body: { ...command().request.body, expectedRunStatus: 'lost' } }), + command({ + body: { ...command().request.body, expectedRunStatus: 'lost' }, + }), + ), + ClusterRunManagementTransportRequestError, + ); + assert.throws( + () => + normalizeClusterRunManagementCommand( + stopCommand({ + body: { ...stopCommand().request.body, mutationId: 'weak' }, + }), ), ClusterRunManagementTransportRequestError, ); diff --git a/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts b/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts index 434ea0b5..1cd35c4d 100644 --- a/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts +++ b/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts @@ -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 { diff --git a/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts b/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts index 3c039dd5..4774d936 100644 --- a/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts +++ b/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts @@ -288,5 +288,10 @@ export const postgresqlMainMigrationManifest: MigrationStreamManifest = checksum: '7aa2b2ade67cdfa6839d4af02209906646a68adfd6c12c4dddeb854021da72b8', }), + Object.freeze({ + id: 'pg-0057-run-management-stop-boundary', + checksum: + 'ab2d0eee3d85a937e1e87243b1fd1e75181529122b64026303488404162e4ba7', + }), ]), }); diff --git a/packages/ql3-cluster-postgres/src/migrations/index.ts b/packages/ql3-cluster-postgres/src/migrations/index.ts index b0404562..45531ded 100644 --- a/packages/ql3-cluster-postgres/src/migrations/index.ts +++ b/packages/ql3-cluster-postgres/src/migrations/index.ts @@ -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 = Object.freeze({ @@ -123,5 +124,6 @@ export const postgresqlMainMigrationStream: MigrationStreamDefinition; -const ALLOWED_ROLES = new Set([ - '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([ '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; + readonly policyFence: Readonly; +} + +interface CancellationAudit { + readonly requestId: string; + readonly auditEventId: string; + readonly principal: Readonly; +} + 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 { + 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; +} + +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, +): Readonly<{ + command: Readonly; + audit: Readonly; +}> { + 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 { return integer(result.rows[0]!, 'nowMs'); } +function confirmStrongAuthentication( + value: Readonly, + observedAtMs: number, +): Readonly { + let principal: Readonly; + 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, +): Promise { + const result = await client.query( + ` + 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, + audit: Readonly, + observedAtMs: number, +): Promise { + const inserted = await client.query( + ` + 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( + ` + 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 { try { await client.query('ROLLBACK'); @@ -140,7 +361,8 @@ async function rollback(client: PostgresClient): Promise { } 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, ): Promise> { const command = normalizeClusterRunCancellationCommand(value); + return this.requestCancellation(command); + } + + async requestUserCancellationAudited( + value: Readonly, + ): Promise> { + const normalized = normalizeManagementCommand(value); + return this.requestCancellation(normalized.command, normalized.audit); + } + + private requestCancellation( + command: Readonly, + audit?: Readonly, + ): Promise> { return this.transaction(async (client) => { - const project = await client.query(` - 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(` - 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(` + const run = await client.query( + ` 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(` + const admission = await client.query( + ` 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; 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( + ` + 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(` - 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; }); } diff --git a/packages/ql3-cluster-postgres/src/schema/schemaContract.ts b/packages/ql3-cluster-postgres/src/schema/schemaContract.ts index c0936930..faa32eb7 100644 --- a/packages/ql3-cluster-postgres/src/schema/schemaContract.ts +++ b/packages/ql3-cluster-postgres/src/schema/schemaContract.ts @@ -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, diff --git a/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts b/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts index db4d11a9..ac09942f 100644 --- a/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts +++ b/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts @@ -119,6 +119,11 @@ interface FunctionPrivilegeRow extends Record { isOwner: unknown; } +interface ColumnPrivilegeRow extends Record { + 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 { + 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( + ` +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, diff --git a/packages/ql3-cluster-postgres/test/clusterRunCancellationRepository.test.cjs b/packages/ql3-cluster-postgres/test/clusterRunCancellationRepository.test.cjs index d99feecd..2a25f7ae 100644 --- a/packages/ql3-cluster-postgres/test/clusterRunCancellationRepository.test.cjs +++ b/packages/ql3-cluster-postgres/test/clusterRunCancellationRepository.test.cjs @@ -41,25 +41,38 @@ function fixture(options = {}) { const normalized = sql.replace(/\s+/g, ' ').trim(); calls.push({ sql: normalized, params }); if ( - normalized.startsWith('BEGIN') || normalized === 'COMMIT' || - normalized === 'ROLLBACK' || normalized.startsWith('SELECT set_config') - ) return { rows: [], rowCount: 0 }; + normalized.startsWith('BEGIN') || + normalized === 'COMMIT' || + normalized === 'ROLLBACK' || + normalized.startsWith('SELECT set_config') + ) + return { rows: [], rowCount: 0 }; + if (normalized.includes('lock_run_management_policy_fence')) { + return { + rows: [{ matches: options.policyMatches ?? true }], + rowCount: 1, + }; + } if (normalized.includes('FROM "ql3"."projects"')) { return { - rows: options.projectRows ?? [{ - projectStatus: 'active', - projectVersion: 2, - }], + rows: options.projectRows ?? [ + { + projectStatus: 'active', + projectVersion: 2, + }, + ], rowCount: 1, }; } if (normalized.includes('FROM "ql3"."project_role_bindings"')) { return { - rows: options.bindingRows ?? [{ - bindingVersion: 3, - bindingState: 'active', - bindingRole: 'operator', - }], + rows: options.bindingRows ?? [ + { + bindingVersion: 3, + bindingState: 'active', + bindingRole: 'operator', + }, + ], rowCount: 1, }; } @@ -73,9 +86,7 @@ function fixture(options = {}) { }; } if ( - normalized.includes( - 'FROM "ql3"."plugin_package_workflow_admissions"', - ) + normalized.includes('FROM "ql3"."plugin_package_workflow_admissions"') ) { const rows = options.workflowAdmissionRows ?? [ { @@ -91,25 +102,40 @@ function fixture(options = {}) { } if (normalized.startsWith('UPDATE "ql3"."runs"')) { return { - rows: options.updatedRows ?? [run({ - runVersion: 5, - eventSequence: 7, - cancelRequestedAtMs: options.nowMs ?? 1_000, - cancelReason: 'user', - })], + rows: options.updatedRows ?? [ + run({ + runVersion: 5, + eventSequence: 7, + cancelRequestedAtMs: options.nowMs ?? 1_000, + cancelReason: 'user', + }), + ], rowCount: options.updatedRows?.length ?? 1, }; } if (normalized.startsWith('INSERT INTO "ql3"."run_events"')) { return { rows: [], rowCount: 1 }; } + if (normalized.startsWith('INSERT INTO "ql3"."security_audit_events"')) { + return { + rows: options.auditInserted === false ? [] : [{ eventId: params[0] }], + rowCount: options.auditInserted === false ? 0 : 1, + }; + } + if (normalized.includes('FROM "ql3"."security_audit_events"')) { + return { rows: options.auditReplayRows ?? [], rowCount: 0 }; + } throw new Error(`Unexpected SQL: ${normalized}`); }, - release() { calls.push({ sql: 'RELEASE', params: [] }); }, + release() { + calls.push({ sql: 'RELEASE', params: [] }); + }, }; return { repository: new PostgresClusterRunCancellationRepository({ - async connect() { return client; }, + async connect() { + return client; + }, }), calls, }; @@ -127,23 +153,26 @@ test('revalidates policy authority and commits one database-timed intent', async cancelRequestedAtMs: 1_000, cancelReason: 'user', }); - const projectIndex = calls.findIndex(({ sql }) => - sql.includes('FROM "ql3"."projects"')); - const bindingIndex = calls.findIndex(({ sql }) => - sql.includes('FROM "ql3"."project_role_bindings"')); + const policyIndex = calls.findIndex(({ sql }) => + sql.includes('lock_run_management_policy_fence'), + ); const runIndex = calls.findIndex(({ sql }) => - sql.includes('FROM "ql3"."runs"')); - assert.ok(projectIndex < bindingIndex && bindingIndex < runIndex); - const update = calls.find(({ sql }) => - sql.startsWith('UPDATE "ql3"."runs"')); + sql.includes('FROM "ql3"."runs"'), + ); + assert.ok(policyIndex >= 0 && policyIndex < runIndex); + const update = calls.find(({ sql }) => sql.startsWith('UPDATE "ql3"."runs"')); assert.deepEqual(update.params, ['run-1', 1_000, 5, 7, 4]); const event = calls.find(({ sql }) => - sql.startsWith('INSERT INTO "ql3"."run_events"')); + sql.startsWith('INSERT INTO "ql3"."run_events"'), + ); assert.equal(event.params[0], command().eventId); assert.equal(event.params[3], 'user-cancel:mutation-1'); assert.equal(event.params[4], 'user'); assert.equal(JSON.parse(event.params[6]).reason, 'user'); - assert.equal(calls.some(({ sql }) => sql === 'COMMIT'), true); + assert.equal( + calls.some(({ sql }) => sql === 'COMMIT'), + true, + ); }); test('returns existing intent and terminal state without adding an event', async () => { @@ -154,8 +183,12 @@ test('returns existing intent and terminal state without adding an event', async (await existing.repository.requestUserCancellation(command())).status, 'already_requested', ); - assert.equal(existing.calls.some(({ sql }) => - sql.startsWith('INSERT INTO "ql3"."run_events"')), false); + assert.equal( + existing.calls.some(({ sql }) => + sql.startsWith('INSERT INTO "ql3"."run_events"'), + ), + false, + ); const terminal = fixture({ runRows: [run({ runStatus: 'succeeded', runVersion: 5 })], @@ -164,20 +197,24 @@ test('returns existing intent and terminal state without adding an event', async (await terminal.repository.requestUserCancellation(command())).status, 'already_terminal', ); - assert.equal(terminal.calls.some(({ sql }) => - sql.startsWith('UPDATE "ql3"."runs"')), false); + assert.equal( + terminal.calls.some(({ sql }) => sql.startsWith('UPDATE "ql3"."runs"')), + false, + ); }); test('accepts cancellation for a lost Run that still owns retry authority', async () => { const { repository } = fixture({ runRows: [run({ runStatus: 'lost' })], - updatedRows: [run({ - runStatus: 'lost', - runVersion: 5, - eventSequence: 7, - cancelRequestedAtMs: 1_000, - cancelReason: 'user', - })], + updatedRows: [ + run({ + runStatus: 'lost', + runVersion: 5, + eventSequence: 7, + cancelRequestedAtMs: 1_000, + cancelReason: 'user', + }), + ], }); assert.equal( (await repository.requestUserCancellation(command())).status, @@ -187,11 +224,7 @@ test('accepts cancellation for a lost Run that still owns retry authority', asyn test('rejects a revoked policy fence before locking the Run', async () => { const { repository, calls } = fixture({ - bindingRows: [{ - bindingVersion: 4, - bindingState: 'revoked', - bindingRole: null, - }], + policyMatches: false, }); await assert.rejects( repository.requestUserCancellation(command()), @@ -199,8 +232,78 @@ test('rejects a revoked policy fence before locking the Run', async () => { error instanceof ClusterRunCancellationFenceRejectedError && error.reason === 'authorization_changed', ); - assert.equal(calls.some(({ sql }) => sql.includes('FROM "ql3"."runs"')), false); - assert.equal(calls.some(({ sql }) => sql === 'ROLLBACK'), true); + assert.equal( + calls.some(({ sql }) => sql.includes('FROM "ql3"."runs"')), + false, + ); + assert.equal( + calls.some(({ sql }) => sql === 'ROLLBACK'), + true, + ); +}); + +test('atomically records strong management audit and exact audit replay', async () => { + const { repository, calls } = fixture(); + const { subject: _subject, ...baseCommand } = command(); + const managed = { + ...baseCommand, + mutationId: '019f0000-0000-4000-8000-000000000001', + requestId: 'request-stop-1', + auditEventId: '019f0000-0000-4000-8000-000000000002', + principal: { + subject: command().subject, + authenticationId: 'oidc:run-management-1', + authenticatedAtMs: 900, + expiresAtMs: 2_000, + assurance: 'hardware', + }, + }; + const result = await repository.requestUserCancellationAudited(managed); + assert.equal(result.status, 'accepted'); + const audit = calls.find(({ sql }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"'), + ); + assert.equal(audit.params[0], managed.auditEventId); + assert.equal(audit.params[1], managed.requestId); + assert.equal(audit.params[5], managed.principal.authenticationId); + assert.ok( + calls.findIndex(({ sql }) => sql.startsWith('UPDATE "ql3"."runs"')) < + calls.findIndex(({ sql }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"'), + ), + ); + assert.ok( + calls.findIndex(({ sql }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"'), + ) < calls.findIndex(({ sql }) => sql === 'COMMIT'), + ); + + const replay = fixture({ + runRows: [run({ cancelRequestedAtMs: 1_000, cancelReason: 'user' })], + auditInserted: false, + auditReplayRows: [ + { + requestId: managed.requestId, + operationId: 'run.stop', + projectId: managed.projectId, + subjectType: 'user', + subjectId: 'user-1', + authenticationId: managed.principal.authenticationId, + outcome: 'allowed', + reasons: ['role_grant', 'strong_authentication'], + projectVersion: 2, + bindingVersion: 3, + }, + ], + }); + assert.equal( + (await replay.repository.requestUserCancellationAudited(managed)).status, + 'already_requested', + ); + assert.equal( + replay.calls.some(({ sql }) => sql.startsWith('UPDATE "ql3"."runs"')), + false, + ); }); test('masks cross-Project and missing Runs', async () => { @@ -255,9 +358,7 @@ test('binds Workflow cancellation to the immutable admission target', async () = ClusterRunCancellationNotFoundError, ); assert.equal( - rejected.calls.some(({ sql }) => - sql.startsWith('UPDATE "ql3"."runs"'), - ), + rejected.calls.some(({ sql }) => sql.startsWith('UPDATE "ql3"."runs"')), false, ); } diff --git a/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs b/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs index 7d07cebe..460c2d17 100644 --- a/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs @@ -1,6 +1,8 @@ const assert = require('node:assert/strict'); const { test } = require('node:test'); -const { postgresqlControlSchemaContract } = require('../dist/schema/schemaContract'); +const { + postgresqlControlSchemaContract, +} = require('../dist/schema/schemaContract'); const { postgresqlMainMigrationManifest, } = require('../dist/migration/migrationManifest'); @@ -105,6 +107,7 @@ test('defines the immutable PostgreSQL capability and Run core stream', async () 'pg-0054-approval-management-boundary', 'pg-0055-run-attempt-log-retention', 'pg-0056-run-management-boundary', + 'pg-0057-run-management-stop-boundary', ], ); for (const migration of postgresqlMainMigrationStream.migrations) { @@ -520,6 +523,11 @@ test('freezes every published PostgreSQL migration checksum', () => { checksum: '7aa2b2ade67cdfa6839d4af02209906646a68adfd6c12c4dddeb854021da72b8', }, + { + id: 'pg-0057-run-management-stop-boundary', + checksum: + 'ab2d0eee3d85a937e1e87243b1fd1e75181529122b64026303488404162e4ba7', + }, ]; assert.deepEqual( postgresqlMainMigrationStream.migrations.map(({ id, checksum }) => ({ @@ -1588,7 +1596,10 @@ test('advances capability v45 with generation-bound Workflow Task attempts', asy /CREATE FUNCTION "ql3"\."plugin_package_workflow_task_attempt_snapshot"/, ); assert.match(sql, /SECURITY DEFINER/); - assert.match(sql, /FOR KEY SHARE OF workflow, source, reconciliation, item, execution/); + assert.match( + sql, + /FOR KEY SHARE OF workflow, source, reconciliation, item, execution/, + ); assert.match( sql, /GRANT SELECT, INSERT[\s\S]*plugin_package_workflow_task_attempt_admissions[\s\S]*TO ql3_runtime/, @@ -1611,16 +1622,11 @@ test('advances capability v45 with generation-bound Workflow Task attempts', asy sql, /migration_id\s*=\s*'pg-0045-plugin-package-workflow-admissions'/, ); - assert.match( - sql, - /"plugin_package_workflow_task_attempt_admission":1/, - ); + assert.match(sql, /"plugin_package_workflow_task_attempt_admission":1/); }); test('advances capability v46 with split Worker credential management authorities', async () => { - const migration = migrationById( - 'pg-0047-worker-credential-management-plans', - ); + const migration = migrationById('pg-0047-worker-credential-management-plans'); const statements = []; await migration.up({ async query(statement) { @@ -1629,10 +1635,7 @@ test('advances capability v46 with split Worker credential management authoritie }, }); const sql = statements.join('\n'); - assert.match( - sql, - /CREATE TABLE "ql3"\."worker_credential_management_plans"/, - ); + assert.match(sql, /CREATE TABLE "ql3"\."worker_credential_management_plans"/); assert.match(sql, /'ql3_worker_credential_manager'/); assert.match(sql, /'ql3_worker_credential_executor'/); assert.match( @@ -1676,10 +1679,7 @@ test('advances capability v47 without invalidating preapproved Worker credential }, }); const sql = statements.join('\n'); - assert.match( - sql, - /DROP CONSTRAINT ql3_worker_credentials_lifetime_check/, - ); + assert.match(sql, /DROP CONSTRAINT ql3_worker_credentials_lifetime_check/); assert.match( sql, /expires_at_ms > GREATEST\(created_at_ms, not_before_at_ms\)/, @@ -1747,7 +1747,10 @@ test('advances capability v49 with durable Worker credential management boundari }, }); const sql = statements.join('\n'); - assert.match(sql, /CREATE TABLE "ql3"\."worker_credential_management_quota_buckets"/); + assert.match( + sql, + /CREATE TABLE "ql3"\."worker_credential_management_quota_buckets"/, + ); assert.match(sql, /TO ql3_worker_credential_manager/); assert.doesNotMatch( sql, @@ -1820,10 +1823,7 @@ test('advances capability v51 with a restart-safe automation identity keyset led assert.match(sql, /contract_version = 51/); assert.match(sql, /"automation_management_identity_keyset_ledger":1/); assert.match(sql, /contract_version = 50/); - assert.match( - sql, - /migration_id = 'pg-0051-automation-management-boundary'/, - ); + assert.match(sql, /migration_id = 'pg-0051-automation-management-boundary'/); }); test('advances capability v52 with a bounded Workflow Run history index', async () => { @@ -1905,10 +1905,7 @@ test('advances capability v54 with durable Cluster log retention authority', asy assert.match(sql, /contract_version = 54/); assert.match(sql, /"run_attempt_log_retention":1/); assert.match(sql, /contract_version = 53/); - assert.match( - sql, - /migration_id = 'pg-0054-approval-management-boundary'/, - ); + assert.match(sql, /migration_id = 'pg-0054-approval-management-boundary'/); }); test('advances capability v55 with isolated strong Run management authority', async () => { @@ -1932,8 +1929,26 @@ test('advances capability v55 with isolated strong Run management authority', as assert.match(sql, /contract_version = 55/); assert.match(sql, /"run_management_boundary":1/); assert.match(sql, /contract_version = 54/); + assert.match(sql, /migration_id = 'pg-0055-run-attempt-log-retention'/); +}); + +test('advances capability v56 with column-scoped Run stop authority', async () => { + const migration = migrationById('pg-0057-run-management-stop-boundary'); + const statements = []; + await migration.up({ + async query(statement) { + statements.push(statement); + return { rows: [] }; + }, + }); + const sql = statements.join('\n'); assert.match( sql, - /migration_id = 'pg-0055-run-attempt-log-retention'/, + /GRANT UPDATE \(cancel_requested_at_ms, cancel_reason, version, event_sequence\) ON "ql3"\."runs" TO ql3_run_manager/, ); + assert.doesNotMatch(sql, /GRANT UPDATE ON "ql3"\."runs" TO ql3_run_manager/); + assert.match(sql, /contract_version = 56/); + assert.match(sql, /"run_management_stop":1/); + assert.match(sql, /contract_version = 55/); + assert.match(sql, /migration_id = 'pg-0056-run-management-boundary'/); }); diff --git a/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs b/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs index 404aab5d..16588cfc 100644 --- a/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs @@ -13,7 +13,9 @@ const { assertPostgresWorkerCredentialManagerSchemaReady, assertPostgresWorkerIngressSchemaReady, } = require('../dist/schema/schemaReadiness'); -const { postgresqlControlSchemaContract } = require('../dist/schema/schemaContract'); +const { + postgresqlControlSchemaContract, +} = require('../dist/schema/schemaContract'); const { postgresqlMainMigrationStream } = require('../dist/migrations'); function validHistory() { @@ -86,12 +88,7 @@ function validPrivileges() { plugin_package_automation_publication_heads: [true, false, false, false], plugin_package_workflow_admissions: [true, true, false, false], plugin_package_workflow_admission_steps: [true, true, false, false], - plugin_package_workflow_task_attempt_admissions: [ - true, - true, - false, - false, - ], + plugin_package_workflow_task_attempt_admissions: [true, true, false, false], plugin_package_publisher_provenance: [false, false, false, false], plugin_package_publisher_revocation_receipts: [false, false, false, false], plugin_package_publisher_revocation_impacts: [false, false, false, false], @@ -523,11 +520,11 @@ function workerCredentialPrivileges(kind) { ] : []), ...(manager - ? [] - : [ - 'approved_action_dispatches', - 'approved_action_executions', - 'worker_credentials', + ? [] + : [ + 'approved_action_dispatches', + 'approved_action_executions', + 'worker_credentials', 'worker_credential_mutations', 'worker_credential_deliveries', 'worker_credential_stage_discards', @@ -714,6 +711,26 @@ function queryable(overrides = {}) { ], }; } + if (text.includes('has_column_privilege')) { + assert.match(text, /format\('%I\.%I', \$1::text, 'runs'\)/); + const columns = contract.tables.find( + ({ name }) => name === 'runs', + ).columns; + const allowed = new Set([ + 'cancel_requested_at_ms', + 'cancel_reason', + 'version', + 'event_sequence', + ]); + return { + rows: + overrides.runManagerColumnPrivileges ?? + columns.map((columnName) => ({ + columnName, + updateAllowed: allowed.has(columnName), + })), + }; + } if (text.includes('has_table_privilege')) { return { rows: overrides.privileges ?? validPrivileges() }; } @@ -731,7 +748,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro serverMajor: 16, currentUser: 'ql3_runtime', contractName: 'control-core', - contractVersion: 55, + contractVersion: 56, migrationIds: [ 'pg-0001-schema-capability', 'pg-0002-run-core', @@ -789,6 +806,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro 'pg-0054-approval-management-boundary', 'pg-0055-run-attempt-log-retention', 'pg-0056-run-management-boundary', + 'pg-0057-run-management-stop-boundary', ], }); }); @@ -819,10 +837,10 @@ test('accepts the exact schema and isolated least-privilege admin role', async ( }), ); assert.equal(report.currentUser, 'ql3_admin'); - assert.equal(report.contractVersion, 55); + assert.equal(report.contractVersion, 56); assert.equal( report.migrationIds.at(-1), - 'pg-0056-run-management-boundary', + 'pg-0057-run-management-stop-boundary', ); }); @@ -835,10 +853,10 @@ test('accepts the isolated least-privilege automation manager role', async () => }), ); assert.equal(report.currentUser, 'ql3_automation_manager'); - assert.equal(report.contractVersion, 55); + assert.equal(report.contractVersion, 56); assert.equal( report.migrationIds.at(-1), - 'pg-0056-run-management-boundary', + 'pg-0057-run-management-stop-boundary', ); const widened = automationManagerPrivileges(); @@ -867,10 +885,10 @@ test('accepts the isolated least-privilege human Approval manager role', async ( }), ); assert.equal(report.currentUser, 'ql3_approval_manager'); - assert.equal(report.contractVersion, 55); + assert.equal(report.contractVersion, 56); assert.equal( report.migrationIds.at(-1), - 'pg-0056-run-management-boundary', + 'pg-0057-run-management-stop-boundary', ); const widened = approvalManagerPrivileges(); @@ -901,8 +919,11 @@ test('accepts the isolated least-privilege Run manager role', async () => { }), ); assert.equal(report.currentUser, 'ql3_run_manager'); - assert.equal(report.contractVersion, 55); - assert.equal(report.migrationIds.at(-1), 'pg-0056-run-management-boundary'); + assert.equal(report.contractVersion, 56); + assert.equal( + report.migrationIds.at(-1), + 'pg-0057-run-management-stop-boundary', + ); const widened = runManagerPrivileges(); widened.find(({ tableName }) => tableName === 'runs').updateAllowed = true; @@ -919,6 +940,33 @@ test('accepts the isolated least-privilege Run manager role', async () => { error.code === 'run_manager_role_invalid' && error.facts.includes('table-privileges:runs'), ); + + const widenedColumns = postgresqlControlSchemaContract.tables + .find(({ name }) => name === 'runs') + .columns.map((columnName) => ({ + columnName, + updateAllowed: [ + 'cancel_requested_at_ms', + 'cancel_reason', + 'version', + 'event_sequence', + 'status', + ].includes(columnName), + })); + await assert.rejects( + assertPostgresRunManagerSchemaReady( + queryable({ + currentUser: 'ql3_run_manager', + privileges: runManagerPrivileges(), + functionMode: 'run-manager', + runManagerColumnPrivileges: widenedColumns, + }), + ), + (error) => + error instanceof PostgresSchemaReadinessError && + error.code === 'run_manager_role_invalid' && + error.facts.includes('column-update-privilege:runs.status'), + ); }); test('accepts isolated Package manager and executor roles', async () => { @@ -1006,10 +1054,10 @@ test('accepts the exact schema and isolated Worker ingress role', async () => { }), ); assert.equal(report.currentUser, 'ql3_worker_ingress'); - assert.equal(report.contractVersion, 55); + assert.equal(report.contractVersion, 56); assert.equal( report.migrationIds.at(-1), - 'pg-0056-run-management-boundary', + 'pg-0057-run-management-stop-boundary', ); }); diff --git a/scripts/ql3-postgres-ha-contract.cjs b/scripts/ql3-postgres-ha-contract.cjs index f1a73497..7e5f9315 100644 --- a/scripts/ql3-postgres-ha-contract.cjs +++ b/scripts/ql3-postgres-ha-contract.cjs @@ -9396,7 +9396,8 @@ async function manualRunRetryFacts(pool, fixture) { (SELECT count(*)::integer FROM "ql3"."run_events" AS event JOIN "ql3"."runs" AS run ON run.id = event.run_id WHERE run.project_id = $2 - AND run.trigger_type = 'run_manual_retry') AS "eventCount", + AND run.trigger_type = 'run_manual_retry' + AND event.type IN ('run.created', 'run.queued')) AS "eventCount", (SELECT count(*)::integer FROM "ql3"."security_audit_events" WHERE project_id = $2 AND operation_id = 'run.retry' AND outcome = 'allowed') AS "allowedAuditCount", @@ -9410,6 +9411,26 @@ async function manualRunRetryFacts(pool, fixture) { return result.rows[0]; } +async function runManagementStopFacts(pool, projectId, runIds) { + const result = await pool.query( + `SELECT + (SELECT count(*)::integer FROM "ql3"."runs" + WHERE project_id = $1 AND id = ANY($2::varchar[]) + AND cancel_requested_at_ms IS NOT NULL + AND cancel_reason = 'user') AS "stoppedRunCount", + (SELECT count(*)::integer FROM "ql3"."run_events" AS event + JOIN "ql3"."runs" AS run ON run.id = event.run_id + WHERE run.project_id = $1 AND run.id = ANY($2::varchar[]) + AND event.type = 'run.cancel_requested') AS "eventCount", + (SELECT count(*)::integer FROM "ql3"."security_audit_events" + WHERE project_id = $1 AND operation_id = 'run.stop' + AND outcome = 'allowed') AS "allowedAuditCount"`, + [projectId, runIds], + ); + assert.equal(result.rowCount, 1); + return result.rows[0]; +} + async function runManualRunRetryHaEvidence(options) { const { primaryPort, primaryDatabase, standbyDatabase } = options; const suffix = `${process.pid}-${randomBytes(3).toString('hex')}`; @@ -9479,6 +9500,11 @@ async function runManualRunRetryHaEvidence(options) { databaseUrl(RUN_MANAGER_USER, RUN_MANAGER_PASSWORD, primaryPort), 'ql3-ha-manual-run-retry-b', )(); + const cancellationRuntime = await databaseOpener( + 'runtime', + databaseUrl(RUNTIME_USER, RUNTIME_PASSWORD, primaryPort), + 'ql3-ha-manual-run-stop-convergence', + )(); try { const source = await new PostgresTaskStartRepository( primaryDatabase.pool, @@ -9571,6 +9597,12 @@ async function runManualRunRetryHaEvidence(options) { const secondRepository = new PostgresRunManualRetryRepository( secondRuntime.pool, ); + const firstCancellation = new PostgresClusterRunCancellationRepository( + firstRuntime.pool, + ); + const secondCancellation = new PostgresClusterRunCancellationRepository( + secondRuntime.pool, + ); const replayCommand = retryCommand(); const exactRace = await Promise.all([ firstRepository.retryRun(replayCommand), @@ -9582,6 +9614,26 @@ async function runManualRunRetryHaEvidence(options) { ]); assert.equal(exactRace[0].runId, exactRace[1].runId); assert.equal(exactRace[0].attemptId, exactRace[1].attemptId); + const stopCommand = { + projectId: fixture.projectId, + runId: exactRace[0].runId, + mutationId: randomUUID(), + eventId: randomUUID(), + requestId: `ha-run-stop-${suffix}-before-promotion`, + auditEventId: randomUUID(), + principal: authentication, + policyFence: { projectVersion: 1, bindingVersion: 1 }, + }; + assert.equal( + (await firstCancellation.requestUserCancellationAudited(stopCommand)) + .status, + 'accepted', + ); + assert.equal( + (await secondCancellation.requestUserCancellationAudited(stopCommand)) + .status, + 'already_requested', + ); for ( let index = 1; @@ -9608,6 +9660,7 @@ async function runManualRunRetryHaEvidence(options) { true, ); assert.ok(rejected[0].reason.retryAfterMs > 0); + const promotionStopRunId = accepted[0].value.runId; const expectedFacts = { sourceStatus: 'failed', @@ -9621,6 +9674,12 @@ async function runManualRunRetryHaEvidence(options) { await manualRunRetryFacts(primaryDatabase.pool, fixture), expectedFacts, ); + assert.deepEqual( + await runManagementStopFacts(primaryDatabase.pool, fixture.projectId, [ + stopCommand.runId, + ]), + { stoppedRunCount: 1, eventCount: 1, allowedAuditCount: 1 }, + ); await waitFor(async () => { const facts = await manualRunRetryFacts(standbyDatabase.pool, fixture); return facts.retryRunCount === CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT @@ -9631,8 +9690,28 @@ async function runManualRunRetryHaEvidence(options) { await manualRunRetryFacts(standbyDatabase.pool, fixture), expectedFacts, ); + assert.deepEqual( + await runManagementStopFacts(standbyDatabase.pool, fixture.projectId, [ + stopCommand.runId, + ]), + { stoppedRunCount: 1, eventCount: 1, allowedAuditCount: 1 }, + ); + assert.deepEqual( + await new PostgresClusterRunCancellationConvergenceRepository( + cancellationRuntime.pool, + ).convergePage({ limit: 1 }), + { + scanned: 1, + settledRuns: 1, + settledAttempts: 1, + blocked: 0, + hasMore: false, + }, + ); return { fixture, + authentication, + stopRunIds: [stopCommand.runId, promotionStopRunId], report: { sourceRunId: source.runId, exactConcurrentReplay: true, @@ -9643,15 +9722,25 @@ async function runManualRunRetryHaEvidence(options) { allowedAuditEvents: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, replicatedBeforePromotion: true, survivedPromotion: false, + stopExactReplay: true, + stopAllowedAuditEvents: 1, + stopReplicatedBeforePromotion: true, + stopConvergedBeforePromotion: true, + stopAcceptedAfterPromotion: false, + stopConvergedAfterPromotion: false, }, }; } finally { - await Promise.all([firstRuntime.close(), secondRuntime.close()]); + await Promise.all([ + firstRuntime.close(), + secondRuntime.close(), + cancellationRuntime.close(), + ]); } } async function verifyManualRunRetryAfterPromotion(options) { - const { promotedPool, evidence } = options; + const { promotedPort, promotedPool, evidence } = options; assert.deepEqual(await manualRunRetryFacts(promotedPool, evidence.fixture), { sourceStatus: 'failed', retryRunCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, @@ -9660,7 +9749,74 @@ async function verifyManualRunRetryAfterPromotion(options) { allowedAuditCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, retryPolicyCount: 0, }); + assert.deepEqual( + await runManagementStopFacts(promotedPool, evidence.fixture.projectId, [ + evidence.stopRunIds[0], + ]), + { stoppedRunCount: 1, eventCount: 1, allowedAuditCount: 1 }, + ); + const clock = await promotedPool.query( + `SELECT floor(extract(epoch FROM statement_timestamp()) * 1000)::bigint + AS "nowMs"`, + ); + const nowMs = Number(clock.rows[0].nowMs); + const promotedRunManager = await databaseOpener( + 'run-manager', + databaseUrl(RUN_MANAGER_USER, RUN_MANAGER_PASSWORD, promotedPort), + 'ql3-ha-run-stop-promoted', + )(); + const promotedRuntime = await databaseOpener( + 'runtime', + databaseUrl(RUNTIME_USER, RUNTIME_PASSWORD, promotedPort), + 'ql3-ha-run-stop-promoted-convergence', + )(); + try { + const result = await new PostgresClusterRunCancellationRepository( + promotedRunManager.pool, + ).requestUserCancellationAudited({ + projectId: evidence.fixture.projectId, + runId: evidence.stopRunIds[1], + mutationId: randomUUID(), + eventId: randomUUID(), + requestId: `ha-run-stop-promoted-${process.pid}`, + auditEventId: randomUUID(), + principal: { + subject: { type: 'user', id: evidence.fixture.actorId }, + authenticationId: `oidc:mfa-promoted-${process.pid}`, + authenticatedAtMs: nowMs, + expiresAtMs: nowMs + 60 * 60_000, + assurance: 'hardware', + }, + policyFence: { projectVersion: 1, bindingVersion: 1 }, + }); + assert.equal(result.status, 'accepted'); + assert.deepEqual( + await new PostgresClusterRunCancellationConvergenceRepository( + promotedRuntime.pool, + ).convergePage({ limit: 1 }), + { + scanned: 1, + settledRuns: 1, + settledAttempts: 1, + blocked: 0, + hasMore: false, + }, + ); + } finally { + await Promise.all([promotedRunManager.close(), promotedRuntime.close()]); + } + assert.deepEqual( + await runManagementStopFacts( + promotedPool, + evidence.fixture.projectId, + evidence.stopRunIds, + ), + { stoppedRunCount: 2, eventCount: 2, allowedAuditCount: 2 }, + ); evidence.report.survivedPromotion = true; + evidence.report.stopAllowedAuditEvents = 2; + evidence.report.stopAcceptedAfterPromotion = true; + evidence.report.stopConvergedAfterPromotion = true; } async function main(argv = process.argv.slice(2)) { @@ -11168,14 +11324,6 @@ async function main(argv = process.argv.slice(2)) { promotedPool: promotedDatabase.pool, report: modelProviderCredentialTestConnection, }); - await verifyManualRunRetryAfterPromotion({ - promotedPool: promotedDatabase.pool, - evidence: manualRunRetry, - }); - timeline.push({ - state: 'manual_run_retry_survived_promotion', - atMs: Number((performance.now() - startedAt).toFixed(3)), - }); timeline.push({ state: 'optional_ai_feature_schema_survived_promotion', atMs: Number((performance.now() - startedAt).toFixed(3)), @@ -11474,6 +11622,15 @@ async function main(argv = process.argv.slice(2)) { await promotedDatabase.pool.query( `SET synchronous_commit = 'remote_apply'`, ); + await verifyManualRunRetryAfterPromotion({ + promotedPort: standbyPort, + promotedPool: promotedDatabase.pool, + evidence: manualRunRetry, + }); + timeline.push({ + state: 'manual_run_retry_survived_promotion', + atMs: Number((performance.now() - startedAt).toFixed(3)), + }); await promotedDatabase.pool.query( `INSERT INTO "ql3"."projects" ( id, name, slug, status, version, created_at_ms, updated_at_ms @@ -11908,7 +12065,7 @@ async function main(argv = process.argv.slice(2)) { FROM "ql3"."worker_credential_deliveries") AS "credentialDeliveries"`, ); assert.deepEqual(sideEffects.rows, [ - { runs: 76, runEvents: 170, credentialDeliveries: 4 }, + { runs: 76, runEvents: 176, credentialDeliveries: 4 }, ]); timeline.push({ state: 'two_fresh_control_replicas_ready', @@ -12344,6 +12501,16 @@ async function main(argv = process.argv.slice(2)) { manualRunRetry.report.replicatedBeforePromotion, manualRunRetrySurvivesPromotion: manualRunRetry.report.survivedPromotion, + runManagementStopExactlyReplaysAcrossPools: + manualRunRetry.report.stopExactReplay, + runManagementStopAuditReplicatesAtomically: + manualRunRetry.report.stopReplicatedBeforePromotion && + manualRunRetry.report.stopAllowedAuditEvents === 2, + runManagementStopWritesAfterPromotion: + manualRunRetry.report.stopAcceptedAfterPromotion, + runManagementStopFeedsCancellationLifecycle: + manualRunRetry.report.stopConvergedBeforePromotion && + manualRunRetry.report.stopConvergedAfterPromotion, physicalStreaming: true, oldPrimaryFencedBeforePromotion: true, bothOldReplicasNotReady: true, diff --git a/test/back/ql3PackageBoundaryAudit.test.cjs b/test/back/ql3PackageBoundaryAudit.test.cjs index 0d9cecf1..2d471dae 100644 --- a/test/back/ql3PackageBoundaryAudit.test.cjs +++ b/test/back/ql3PackageBoundaryAudit.test.cjs @@ -421,10 +421,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( rootSourceFileRoles: clusterPostgres.rootSourceFileRoles, }, { - sourceFiles: 152, + sourceFiles: 153, rootSourceFiles: 1, rootSourceLines: 125, - nestedSourceFiles: 151, + nestedSourceFiles: 152, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, );