diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index 179728a5..c036c14f 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -11,6 +11,20 @@ 最新增量证据（2026-08-19）： +- D-365/ADR-0458（已接受；聚合指标与产品 UI 待完成）：在既有隔离 `cluster-admin` Run management plane 上增加 + `run.cancellation.inspect` 与 `run.cancellation.rearm`，不新建 package、服务、端口、timer、连接池或 Kubernetes 对象。inspect 要求强认证 User + 与 `run.read`，viewer 可读取 Run/dispatch 的固定低敏投影，但永不返回 lease owner、raw token 或 token digest；rearm 要求 `run.stop`，只接受 + `blocked + expected dispatch version + expected blocking result` 的精确 CAS，retry delay 固定为 1 秒至 24 小时并由 PostgreSQL + `transaction_timestamp()` 计算绝对到期时间。成功事务同时推进 Run version/event sequence、保留 blocking result 作为历史、追加 + `run.cancel_dispatch_rearmed` 与 allowed audit；mutation replay 返回同一不可变 receipt，stale result/version、终态 Run、非活跃 Attempt 或 authorization + drift 均失败关闭。`pg-0067-cancellation-dispatch-management` 把 capability 提升至 v66，只给 `ql3_run_manager` 增加 dispatch SELECT 与 + `status/version/next_attempt_at_ms/updated_at_ms` 四列 UPDATE。真实 HA 首跑发现 `FOR KEY SHARE` 会隐含要求 Attempt UPDATE 权限，最终以既有 + SERIALIZABLE 事务、Run 行锁和普通 Attempt 读取修复，没有扩权。完整 backend `1,487 pass / 0 fail / 2 conditional skip`，18-package clean + build/test、四项架构审计与 `14/14` Local artifact audit 通过；workspace 仍为 18 包，`cluster-postgres` 为 172 个 source（171 nested），基础 + Edge/Standalone 仍为 `2,589,998 / 2,590,076` bytes，Application+AI 为 `4,493,151 / 4,493,283` bytes。PostgreSQL 18.6 arm64 HA + `144/144` 真实执行 blocked→低敏 inspect→CAS rearm→not-due→生产交付→WAL→promotion；timeline `1→2`，报告 SHA-256 为 + `58f43327f426c286aaa1764aa6dd9f5963a462304716e09b933f326991442b8a`。 + - D-364/ADR-0457（已接受；运维可见性与实机容量门待完成）：把 PostgreSQL CancellationDispatch 接入既有 caller-driven Remote Worker lease-control 生产路径，没有新建扫描 timer 或第二调度 authority。只有已有 Session/RunDispatchLease/Attempt fence 产生 `stop_requested` 后才 claim； `claimed` 必须先以 `termination_requested` 原子结算 dispatch 与 `run.cancel_dispatched`，验证 durable 结果后才向 Worker 释放停止响应；`dispatched` diff --git a/docs/adr/ADR-0005-durable-cancellation-dispatch.md b/docs/adr/ADR-0005-durable-cancellation-dispatch.md index 8e5f49f7..ab4174e0 100644 --- a/docs/adr/ADR-0005-durable-cancellation-dispatch.md +++ b/docs/adr/ADR-0005-durable-cancellation-dispatch.md @@ -1,6 +1,6 @@ # ADR-0005：Durable Cancellation Dispatch、Lease 与 Fencing -- 状态：Accepted（Local 与 PostgreSQL Repository 已实现；Cluster 生产启动拓扑待接入） +- 状态：Accepted（Local、PostgreSQL Repository、Cluster 生产交付与人工处置入口已实现） - 日期：2026-07-18 - 决策范围：跨进程取消派发、崩溃恢复、并发 Worker、退避和审计事件 - 关联：QL-RFC-0001、ADR-0001、ADR-0003、ADR-0004 @@ -151,13 +151,13 @@ PID 可复用，可能终止无关进程，禁止。 ## 5. 当前孵化边界 -`next` 已实现 profile-neutral canonical contract、`0005-run-cancellation-dispatch`、legacy Sequelize/SQLite adapter、PostgreSQL `pg-0066-cancellation-dispatch`/capability v65 adapter、lease expiry 接管、fencing、退避、结果事件、Dispatcher、有界 Supervisor 和默认惰性的 lifecycle runner。PostgreSQL 结果事务按 Run→Attempt→dispatch 锁序完成 dispatch 更新、Run version CAS 与 RunEvent 追加；runtime 角色只取得新表的 SELECT/INSERT/UPDATE。Cluster 生产交付已由 ADR-0457 接入既有 caller-driven Worker lease-control：Run-level stop 必须先结算 durable dispatch 再返回，Workflow Task timeout 则保持 `untracked`，不伪造父 Run 取消。 +`next` 已实现 profile-neutral canonical contract、`0005-run-cancellation-dispatch`、legacy Sequelize/SQLite adapter、PostgreSQL `pg-0066-cancellation-dispatch`/capability v65 adapter、lease expiry 接管、fencing、退避、结果事件、Dispatcher、有界 Supervisor 和默认惰性的 lifecycle runner。PostgreSQL 结果事务按 Run→Attempt→dispatch 锁序完成 dispatch 更新、Run version CAS 与 RunEvent 追加；runtime 角色只取得新表的 SELECT/INSERT/UPDATE。Cluster 生产交付已由 ADR-0457 接入既有 caller-driven Worker lease-control：Run-level stop 必须先结算 durable dispatch 再返回，Workflow Task timeout 则保持 `untracked`，不伪造父 Run 取消。ADR-0458 又在既有隔离 Run management plane 上增加低敏 inspect 与 exact-CAS rearm：viewer 只有 `run.read` 诊断权，rearm 继续要求 `run.stop`；任何接口都不暴露 lease owner、raw token 或 token digest。 HTTP worker 已通过默认关闭的 manual-only manifest bootstrap 接入 Local Supervisor：只有 accepted 且全部 gate 通过时才启动，失败或 shutdown 时有界停止。以下工作仍未完成，因此它仍只允许显式 canary，不得扩大到默认生产流量： -- 用户可见的运行指标、blocked 诊断和处置入口。 +- 用户可见的聚合运行指标、告警与产品 UI；私有 operator 诊断和处置协议已经实现。 - 固定 edge 设备的数据库写放大、RSS、时延和磁盘基准。 -- cluster-control 对 PostgreSQL CancellationDispatch 的用户可见 availability/blocked 指标、诊断与人工处置入口。 +- cluster-control 对 PostgreSQL CancellationDispatch 的聚合 availability/blocked 指标与告警出口。 - 首次真实目标实例完整激活/回滚仪式与共享 config 多写者 authority。 ## 6. 验证门禁 @@ -175,3 +175,5 @@ HTTP worker 已通过默认关闭的 manual-only manifest bootstrap 接入 Local 11. PostgreSQL 双连接只能产生一个 claim winner，raw token 不落库，数据库时间决定 lease/retry 到期。 12. v65 事实经 WAL 到达 standby，提升为新 Primary 后仍可读取；旧 owner/token/version 继续被 fencing。 13. Cluster Worker lease-control 只在 durable dispatch 已结算或已重放时释放 Run-level stop；Workflow Task timeout 不写父 Run 取消事实。 +14. 强认证 viewer 只能读取无 lease capability 的低敏诊断；只有 `run.stop` authority 可按 blocked/version/result 精确 rearm。 +15. rearm、RunEvent 与 allowed audit 原子提交，数据库时间决定 retry due；stale fence、授权漂移与 mutation drift 均失败关闭。 diff --git a/docs/adr/ADR-0456-database-timed-postgresql-cancellation-dispatch.md b/docs/adr/ADR-0456-database-timed-postgresql-cancellation-dispatch.md index 5e6fc5b1..a996ebf1 100644 --- a/docs/adr/ADR-0456-database-timed-postgresql-cancellation-dispatch.md +++ b/docs/adr/ADR-0456-database-timed-postgresql-cancellation-dispatch.md @@ -61,4 +61,4 @@ QingLong 3.0 还必须同时服务低配路由和集群节点。公共协议需 ## 后续 -ADR-0457 已把 PostgreSQL CancellationDispatch Repository 接入既有 caller-driven Worker lease-control 生产 composition：它刻意不新增扫描 cadence，并复用 ingress drain 与数据库 shutdown 顺序。尚需用户可见的 availability/blocked 指标、诊断和人工处置面，以及 CloudNativePG live failover、多副本压力、固定 x64/arm64 资源门。Local 侧如需消除 legacy raw-token 存量，另开兼容迁移 ADR，不与 Cluster rollout 混合。 +ADR-0457 已把 PostgreSQL CancellationDispatch Repository 接入既有 caller-driven Worker lease-control 生产 composition：它刻意不新增扫描 cadence，并复用 ingress drain 与数据库 shutdown 顺序。ADR-0458 又通过 v66 migration 和既有 Run management plane 实现无 lease capability 的诊断与 exact-CAS rearm；没有新增连接、进程或部署对象。尚需用户可见的聚合 availability/blocked 指标、告警 UI，以及 CloudNativePG live failover、多副本压力、固定 x64/arm64 资源门。Local 侧如需消除 legacy raw-token 存量，另开兼容迁移 ADR，不与 Cluster rollout 混合。 diff --git a/docs/adr/ADR-0457-worker-pull-cluster-cancellation-delivery.md b/docs/adr/ADR-0457-worker-pull-cluster-cancellation-delivery.md index 7602898f..3d8e3fa7 100644 --- a/docs/adr/ADR-0457-worker-pull-cluster-cancellation-delivery.md +++ b/docs/adr/ADR-0457-worker-pull-cluster-cancellation-delivery.md @@ -59,4 +59,4 @@ Run 取消与 Workflow Task timeout 还存在语义差异：前者具有 `Run.ca ## 后续 -仍需用户可见的 blocked/availability 指标、诊断与人工处置入口，以及 CloudNativePG live failover、多副本容量压力、固定 Linux x64/arm64 与物理 Edge 资源门。Local legacy raw-token 存量迁移继续保持独立议题。 +ADR-0458 已在既有 Run management plane 实现低敏 blocked 诊断与人工 rearm，不新增第二个 cancellation worker。仍需用户可见的聚合 blocked/availability 指标、告警与产品 UI，以及 CloudNativePG live failover、多副本容量压力、固定 Linux x64/arm64 与物理 Edge 资源门。Local legacy raw-token 存量迁移继续保持独立议题。 diff --git a/docs/adr/ADR-0458-least-privilege-cancellation-diagnostics-and-rearm.md b/docs/adr/ADR-0458-least-privilege-cancellation-diagnostics-and-rearm.md new file mode 100644 index 00000000..5139df91 --- /dev/null +++ b/docs/adr/ADR-0458-least-privilege-cancellation-diagnostics-and-rearm.md @@ -0,0 +1,65 @@ +# ADR-0458：最小权限 CancellationDispatch 诊断与人工 Rearm + +- 状态：Accepted +- 日期：2026-08-19 +- 关联 RFC：QL-RFC-0001 D-365、PR-5、PR-7 +- 关联 ADR：ADR-0005、ADR-0383、ADR-0456、ADR-0457 +- Amends：ADR-0005 的 blocked 人工处置边界、ADR-0456 的 PostgreSQL Run manager authority + +## 上下文 + +ADR-0456/0457 已让 Cluster Run cancellation 具备数据库计时、跨副本 fencing 和真实 Worker 交付，但 `identity_mismatch`、`pid_mismatch`、`unsupported`、`invalid` 会有意进入 `blocked`，禁止自动重试。没有受权诊断与人工 rearm 时，operator 只能直接查询或修改数据库；这既容易泄露 lease capability，也绕过 Project Policy、强认证、RunEvent 和安全审计。 + +QingLong 同时面向低配路由器与集群。该能力只属于 Cluster operator plane，不能扩大 Local Profile 闭包，也不应为了两个 operation 新建微包、HTTP 服务、扫描 timer 或数据库连接池。 + +## 决策 + +1. 复用既有隔离 `cluster-admin` Run management HTTPS/mTLS process、固定 `/api/v3/runs/management` 路径、OIDC assertion、速率限制、连接池和 shutdown lifecycle。新增 operation 为 `run.cancellation.inspect` 与 `run.cancellation.rearm`，不新增服务或部署对象。 +2. inspect 只接受强认证 User，授权使用既有 `run.read`。因此 viewer 可诊断，agent、弱认证、过期认证和 authorization fence drift 均失败关闭。 +3. diagnostic 只投影 Project/Run 状态、取消意图、Attempt ID、dispatch status/version/count、受限时间戳、固定 last result 与 `none|wait|rearm` 建议。lease owner、raw token、token digest、PID、handle、命令、环境、Secret 和错误原文永不进入 SQL projection、transport schema 或客户端结果。 +4. rearm 继续使用 `run.stop`，不新建更宽 permission。请求必须精确匹配 `blocked`、expected dispatch version 和 expected blocking result，并提供 1 秒至 24 小时的 retry delay；服务端生成 RunEvent ID，调用方不能塑造 event identity。 +5. Repository 使用 SERIALIZABLE 短事务和 PostgreSQL `transaction_timestamp()`。它重新锁定 Project Policy fence 与 Run，验证非终态 Run、既有 cancellation、同 Run active Attempt 和精确 dispatch CAS；绝对 `next_attempt_at_ms` 只能由数据库计算。 +6. 成功 rearm 在同一事务中推进 Run version/event sequence，把 dispatch 切换为 `retry_wait` 并递增 version，追加 `run.cancel_dispatch_rearmed` 和 allowed security audit。事件 payload 只包含 mutation、前后 version、固定 blocking result、delay、due time 与 Run version。 +7. mutation replay 先读取 immutable RunEvent，只有 actor、mutation、version、result、delay 和 receipt 全部相同时返回原 receipt；任何漂移返回 conflict，不重复更新 dispatch。 +8. rearm 保留原 blocking `last_result` 作为历史，因此 canonical/SQL 状态约束允许 `leased|retry_wait` 携带 blocking 或 retryable history；新 claim/result 仍按当前 status 与精确 fence 决定，不把历史结果当作新的 controller verdict。 +9. `pg-0067-cancellation-dispatch-management` 把 capability 提升至 v66。`ql3_run_manager` 只新增 dispatch SELECT 与四列 UPDATE：`status`、`version`、`next_attempt_at_ms`、`updated_at_ms`；没有 INSERT、DELETE、TRUNCATE、lease 字段 UPDATE 或 schema authority。 +10. Attempt active check 在已有 SERIALIZABLE 事务与 Run 行锁后执行普通 SELECT。PostgreSQL 的 `FOR KEY SHARE` 也隐含要求至少一列 UPDATE 权限；为行锁扩大 Run manager 的 Attempt UPDATE authority 被拒绝。并发终态变化由 Run 锁与 serializable ordering 收敛。 +11. 新 repository 留在现有 `cluster-postgres/run-management` 子域，协议留在现有 `cluster-admin/run-management` 子域。新增两个文件不会成为拆分 package 的理由；workspace 继续精确为 18 包。 + +## 被拒绝的替代方案 + +### 直接暴露数据库记录 + +拒绝。原始表包含 lease owner/token digest 等 capability-adjacent 字段，也没有逐请求 Policy、强认证、审计和响应 shape 验证。 + +### blocked 自动重试 + +拒绝。身份、PID 或协议不匹配不是瞬时可用性问题；自动循环可能向错误执行实例重复交付停止。 + +### 新建 cancellation-admin 服务或 package + +拒绝。既有 Run management plane 已拥有完全相同的用户身份、Project Policy、mTLS、连接和生命周期边界。另建服务会增加低价值微包、端口、Secret 和部署成本。 + +### 给 Run manager 增加 RunAttempt UPDATE 权限 + +拒绝。该权限只为满足 PostgreSQL 行锁语法，而不是业务 mutation。SERIALIZABLE + Run lock + 普通读取足以表达合法顺序，不能为了实现细节扩大 authority。 + +## 资源、安全与部署影响 + +- Edge/Standalone package closure 不包含 `cluster-admin`、`cluster-postgres` 或 `pg`；14 个 Local Profile 预算不变化。 +- Cluster 复用既有 caller-driven management request，没有扫描 timer、后台队列、缓存、额外连接或空闲写入。 +- 每次 inspect/rearm 最多一个 5 秒 statement timeout 的 SERIALIZABLE 短事务；rearm 只在人工明确操作时写一条 RunEvent 和一条 audit。 +- response 与客户端都做 exact-shape 验证；未知字段、目标漂移、version/result/delay 漂移和 lease capability 泄漏失败关闭。 + +## 验证 + +- Repository 聚焦测试 `5/5`，service `6/6`，transport/client `12/12`；覆盖低敏 inspect、viewer/read 与 operator/stop 分权、exact replay、stale fence、失败审计和 secret-field tampering。 +- `cluster-postgres` 完整包 `339 pass / 0 fail / 3 conditional skip`；`cluster-admin` 完整包 `394 pass / 0 fail / 3 conditional skip`。 +- 完整 backend `1,487 pass / 0 fail / 2 conditional skip`；18-package clean build/test 退出 0。 +- package boundary、cluster dependency、Edge import、cluster deployment 四项审计零 finding；workspace 仍为 18 包，Cluster PostgreSQL 为 172 source/171 nested，根入口计数不变。 +- `14/14` Local artifact audit 通过：基础 Edge/Standalone `2,589,998 / 2,590,076` bytes，Application+AI `4,493,151 / 4,493,283` bytes，均不包含 PostgreSQL closure。 +- PostgreSQL 18.6 arm64 HA `144/144`：真实 `ql3_run_manager` 执行 blocked→inspect→rearm，验证 retry not-due、最终 production dispatch、同步 WAL、主库隔离与 standby promotion；timeline `1→2`，报告 SHA-256 `58f43327f426c286aaa1764aa6dd9f5963a462304716e09b933f326991442b8a`。 + +## 后续 + +私有 operator 协议已经闭合，但面向产品的聚合 blocked/availability 指标、告警和 UI 仍待实现。CloudNativePG live failover、多副本容量压力、固定 Linux x64/arm64 与物理 Edge 资源证据继续作为发布最终化门，不由单机 Docker HA 结果替代。 diff --git a/packages/ql3-cluster-admin/src/run-management/runManagement.ts b/packages/ql3-cluster-admin/src/run-management/runManagement.ts index 8a09ddce..24703b4b 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagement.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagement.ts @@ -1,12 +1,21 @@ import { randomUUID } from 'node:crypto'; import { + InvalidRunCancellationDispatchManagementError, PostgresClusterRunCancellationRepository, PostgresProjectPolicyRepository, + PostgresRunCancellationDispatchManagementRepository, PostgresRunManualRetryRepository, PostgresSecurityAuditRepository, + RunCancellationDispatchManagementConflictError, + RunCancellationDispatchManagementNotFoundError, + RunCancellationDispatchManagementUnavailableError, + type BlockingCancellationDispatchResult, + type RunCancellationDispatchDiagnostic, + type RunCancellationDispatchRearmReceipt, } from '@qinglong/cluster-postgres/run-manager'; import type { PostgresPool } from '@qinglong/runtime-core'; +import { CANCELLATION_DISPATCH_BLOCKING_RESULTS } from '@qinglong/runtime-core/cancellation-dispatch'; import { ProjectPolicyEngine } from '@qinglong/runtime-core/project-policy'; import { ClusterRunCancellationFenceRejectedError, @@ -57,6 +66,23 @@ export interface ClusterRunManagementStopRequest { readonly principal: Readonly; } +export interface ClusterRunManagementCancellationInspectRequest { + readonly projectId: string; + readonly runId: string; + readonly requestId: string; + readonly auditEventId: string; + readonly failureAuditEventId: string; + readonly principal: Readonly; +} + +export interface ClusterRunManagementCancellationRearmRequest + extends ClusterRunManagementCancellationInspectRequest { + readonly mutationId: string; + readonly expectedDispatchVersion: number; + readonly expectedLastResult: BlockingCancellationDispatchResult; + readonly retryDelayMs: number; +} + export interface ClusterRunManagementService { retry( request: Readonly, @@ -64,6 +90,12 @@ export interface ClusterRunManagementService { stop( request: Readonly, ): Promise>; + inspectCancellation( + request: Readonly, + ): Promise>; + rearmCancellation( + request: Readonly, + ): Promise>; } export interface ClusterRunManagementOptions { @@ -178,6 +210,56 @@ function exactStopRequest( } } +function exactCancellationInspectRequest( + value: unknown, +): asserts value is Readonly { + if ( + !value || + typeof value !== 'object' || + Array.isArray(value) || + Object.keys(value).sort().join('\0') !== + [ + 'auditEventId', + 'failureAuditEventId', + 'principal', + 'projectId', + 'requestId', + 'runId', + ] + .sort() + .join('\0') + ) { + throw new ClusterRunManagementRequestError(); + } +} + +function exactCancellationRearmRequest( + value: unknown, +): asserts value is Readonly { + if ( + !value || + typeof value !== 'object' || + Array.isArray(value) || + Object.keys(value).sort().join('\0') !== + [ + 'auditEventId', + 'expectedDispatchVersion', + 'expectedLastResult', + 'failureAuditEventId', + 'mutationId', + 'principal', + 'projectId', + 'requestId', + 'retryDelayMs', + 'runId', + ] + .sort() + .join('\0') + ) { + throw new ClusterRunManagementRequestError(); + } +} + function validUuid(value: unknown): value is string { return typeof value === 'string' && UUID_PATTERN.test(value); } @@ -195,6 +277,12 @@ function failureReason(error: unknown): string { if (error instanceof ClusterRunCancellationFenceRejectedError) { return error.reason; } + if (error instanceof RunCancellationDispatchManagementNotFoundError) { + return 'run_not_found'; + } + if (error instanceof RunCancellationDispatchManagementConflictError) { + return error.reason; + } return 'management_unavailable'; } @@ -227,6 +315,8 @@ export function createClusterRunManagementService( const cancellations = new PostgresClusterRunCancellationRepository( options.pool, ); + const cancellationDispatches = + new PostgresRunCancellationDispatchManagementRepository(options.pool); const audit = new PostgresSecurityAuditRepository(options.pool); return Object.freeze({ @@ -412,5 +502,188 @@ export function createClusterRunManagementService( throw new ClusterRunManagementUnavailableError({ cause: error }); } }, + async inspectCancellation( + requestValue: Readonly, + ) { + exactCancellationInspectRequest(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.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.read', + ); + fence = decision.fence; + if ( + decision.effect !== 'allow' || + !fence || + fence.bindingVersion === null + ) { + throw new ClusterRunManagementAuthorizationError(); + } + return await cancellationDispatches.inspect({ + projectId: requestValue.projectId, + runId: requestValue.runId, + requestId: requestValue.requestId, + auditEventId: requestValue.auditEventId, + principal, + policyFence: fence, + }); + } catch (error) { + try { + await audit.record( + normalizeSecurityAuditRecord({ + eventId: requestValue.failureAuditEventId, + requestId: requestValue.requestId, + operationId: 'run.cancellation.inspect', + 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 InvalidRunCancellationDispatchManagementError) { + throw new ClusterRunManagementRequestError(); + } + if (error instanceof RunCancellationDispatchManagementNotFoundError) { + throw new ClusterRunManagementTargetUnavailableError(); + } + if (error instanceof RunCancellationDispatchManagementConflictError) { + throw new ClusterRunManagementConflictError(); + } + if (error instanceof RunCancellationDispatchManagementUnavailableError) { + throw new ClusterRunManagementUnavailableError({ cause: error }); + } + throw new ClusterRunManagementUnavailableError({ cause: error }); + } + }, + async rearmCancellation( + requestValue: Readonly, + ) { + exactCancellationRearmRequest(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 || + !Number.isSafeInteger(requestValue.expectedDispatchVersion) || + requestValue.expectedDispatchVersion < 1 || + requestValue.expectedDispatchVersion >= 2_147_483_647 || + !CANCELLATION_DISPATCH_BLOCKING_RESULTS.includes( + requestValue.expectedLastResult, + ) || + !Number.isSafeInteger(requestValue.retryDelayMs) || + requestValue.retryDelayMs < 1_000 || + requestValue.retryDelayMs > 24 * 60 * 60_000 + ) { + 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 cancellationDispatches.rearm({ + projectId: requestValue.projectId, + runId: requestValue.runId, + requestId: requestValue.requestId, + auditEventId: requestValue.auditEventId, + principal, + policyFence: fence, + mutationId: requestValue.mutationId, + eventId: createId(), + expectedDispatchVersion: requestValue.expectedDispatchVersion, + expectedLastResult: requestValue.expectedLastResult, + retryDelayMs: requestValue.retryDelayMs, + }); + } catch (error) { + try { + await audit.record( + normalizeSecurityAuditRecord({ + eventId: requestValue.failureAuditEventId, + requestId: requestValue.requestId, + operationId: 'run.cancellation.rearm', + 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 InvalidRunCancellationDispatchManagementError) { + throw new ClusterRunManagementRequestError(); + } + if (error instanceof RunCancellationDispatchManagementNotFoundError) { + throw new ClusterRunManagementTargetUnavailableError(); + } + if (error instanceof RunCancellationDispatchManagementConflictError) { + throw new ClusterRunManagementConflictError(); + } + if (error instanceof RunCancellationDispatchManagementUnavailableError) { + 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 32e91a77..556d2c1d 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts @@ -6,6 +6,11 @@ import { RUN_CANCELLATION_SCHEMA, normalizeRunCancellationResult, } from '@qinglong/runtime-core/run-cancellation'; +import { RUN_STATUSES } from '@qinglong/runtime-core/run'; +import { + CANCELLATION_DISPATCH_RESULTS, + CANCELLATION_DISPATCH_STATUSES, +} from '@qinglong/runtime-core/cancellation-dispatch'; import { ClusterPluginPackageManagementClientRequestError, executeClusterAuthenticatedManagementClient, @@ -14,6 +19,8 @@ import { type ClusterPluginPackageManagementClientPaths, } from '../management-support/pluginPackageManagementClient'; import { + RUN_CANCELLATION_DISPATCH_DIAGNOSTIC_SCHEMA, + RUN_CANCELLATION_DISPATCH_REARM_RECEIPT_SCHEMA, normalizeClusterRunManagementCommand, type ClusterRunManagementCommand, type ClusterRunManagementTransportResult, @@ -48,6 +55,14 @@ function exact( return value as Record; } +function safeInteger(value: unknown, minimum = 0): value is number { + return ( + typeof value === 'number' && + Number.isSafeInteger(value) && + value >= minimum + ); +} + export function validateClusterRunManagementClientResult( value: unknown, command: Readonly, @@ -100,6 +115,151 @@ export function validateClusterRunManagementClientResult( envelope as unknown as ClusterRunManagementTransportResult, ); } + if (command.operation === 'run.cancellation.inspect') { + const envelope = exact(value, [ + 'schemaVersion', + 'operation', + 'diagnostic', + ]); + if ( + envelope.schemaVersion !== 1 || + envelope.operation !== command.operation + ) { + invalid(); + } + const diagnostic = exact(envelope.diagnostic, [ + 'schema', + 'projectId', + 'runId', + 'runStatus', + 'runVersion', + 'eventSequence', + ...(Object.hasOwn(envelope.diagnostic as object, 'cancelRequestedAtMs') + ? ['cancelRequestedAtMs', 'cancelReason'] + : []), + 'operatorAction', + 'dispatch', + ]); + if ( + diagnostic.schema !== RUN_CANCELLATION_DISPATCH_DIAGNOSTIC_SCHEMA || + diagnostic.projectId !== command.request.projectId || + diagnostic.runId !== command.request.runId || + !RUN_STATUSES.includes(diagnostic.runStatus as never) || + !safeInteger(diagnostic.runVersion, 1) || + !safeInteger(diagnostic.eventSequence) || + !['none', 'wait', 'rearm'].includes(diagnostic.operatorAction as string) || + (Object.hasOwn(diagnostic, 'cancelRequestedAtMs') && + (!safeInteger(diagnostic.cancelRequestedAtMs) || + !['user', 'policy', 'shutdown', 'reconcile', 'timeout'].includes( + diagnostic.cancelReason as string, + ))) + ) { + invalid(); + } + let dispatchStatus: string | null = null; + if (diagnostic.dispatch !== null) { + const dispatch = exact(diagnostic.dispatch, [ + 'attemptId', + 'status', + 'version', + 'dispatchCount', + ...(Object.hasOwn(diagnostic.dispatch as object, 'nextAttemptAtMs') + ? ['nextAttemptAtMs'] + : []), + ...(Object.hasOwn(diagnostic.dispatch as object, 'leaseExpiresAtMs') + ? ['leaseExpiresAtMs'] + : []), + ...(Object.hasOwn(diagnostic.dispatch as object, 'lastResult') + ? ['lastResult'] + : []), + ...(Object.hasOwn( + diagnostic.dispatch as object, + 'lastDispatchedAtMs', + ) + ? ['lastDispatchedAtMs'] + : []), + 'createdAtMs', + 'updatedAtMs', + ]); + if ( + typeof dispatch.attemptId !== 'string' || + dispatch.attemptId.length < 1 || + !CANCELLATION_DISPATCH_STATUSES.includes(dispatch.status as never) || + !safeInteger(dispatch.version) || + !safeInteger(dispatch.dispatchCount) || + !safeInteger(dispatch.createdAtMs) || + !safeInteger(dispatch.updatedAtMs) || + (Object.hasOwn(dispatch, 'nextAttemptAtMs') && + !safeInteger(dispatch.nextAttemptAtMs)) || + (Object.hasOwn(dispatch, 'leaseExpiresAtMs') && + !safeInteger(dispatch.leaseExpiresAtMs)) || + (Object.hasOwn(dispatch, 'lastDispatchedAtMs') && + !safeInteger(dispatch.lastDispatchedAtMs)) || + (Object.hasOwn(dispatch, 'lastResult') && + !CANCELLATION_DISPATCH_RESULTS.includes(dispatch.lastResult as never)) + ) { + invalid(); + } + dispatchStatus = dispatch.status as string; + } + const expectedOperatorAction = + dispatchStatus === 'blocked' + ? 'rearm' + : dispatchStatus !== null && dispatchStatus !== 'dispatched' + ? 'wait' + : dispatchStatus === null && + Object.hasOwn(diagnostic, 'cancelRequestedAtMs') + ? 'wait' + : 'none'; + if (diagnostic.operatorAction !== expectedOperatorAction) invalid(); + return Object.freeze( + envelope as unknown as ClusterRunManagementTransportResult, + ); + } + if (command.operation === 'run.cancellation.rearm') { + const envelope = exact(value, ['schemaVersion', 'operation', 'rearm']); + if ( + envelope.schemaVersion !== 1 || + envelope.operation !== command.operation + ) { + invalid(); + } + const rearm = exact(envelope.rearm, [ + 'schema', + 'status', + 'projectId', + 'runId', + 'attemptId', + 'previousDispatchVersion', + 'dispatchVersion', + 'previousResult', + 'retryDelayMs', + 'nextAttemptAtMs', + 'runVersion', + 'eventSequence', + ]); + if ( + rearm.schema !== RUN_CANCELLATION_DISPATCH_REARM_RECEIPT_SCHEMA || + rearm.status !== 'rearmed' || + rearm.projectId !== command.request.projectId || + rearm.runId !== command.request.runId || + typeof rearm.attemptId !== 'string' || + rearm.attemptId.length < 1 || + rearm.previousDispatchVersion !== + command.request.body.expectedDispatchVersion || + rearm.dispatchVersion !== rearm.previousDispatchVersion + 1 || + rearm.previousResult !== command.request.body.expectedLastResult || + rearm.retryDelayMs !== command.request.body.retryDelayMs || + !safeInteger(rearm.nextAttemptAtMs) || + !safeInteger(rearm.runVersion, 1) || + !safeInteger(rearm.eventSequence, 1) + ) { + invalid(); + } + return Object.freeze( + envelope as unknown as ClusterRunManagementTransportResult, + ); + } const envelope = exact(value, ['schemaVersion', 'operation', 'stop']); if ( envelope.schemaVersion !== 1 || diff --git a/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts b/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts index fa1b6dee..a0ac9b63 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts @@ -12,6 +12,7 @@ import { normalizeSecurityPrincipal, type SecurityPrincipal, } from '@qinglong/runtime-core/security'; +import { CANCELLATION_DISPATCH_BLOCKING_RESULTS } from '@qinglong/runtime-core/cancellation-dispatch'; import type { ClusterRunManagementService } from './runManagement'; const IDENTIFIER_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; @@ -19,6 +20,15 @@ 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 const RUN_CANCELLATION_DISPATCH_INSPECT_REQUEST_SCHEMA = + 'qinglong/run-cancellation-dispatch-inspect@v1'; +export const RUN_CANCELLATION_DISPATCH_DIAGNOSTIC_SCHEMA = + 'qinglong/run-cancellation-dispatch-diagnostic@v1'; +export const RUN_CANCELLATION_DISPATCH_REARM_REQUEST_SCHEMA = + 'qinglong/run-cancellation-dispatch-rearm-request@v1'; +export const RUN_CANCELLATION_DISPATCH_REARM_RECEIPT_SCHEMA = + 'qinglong/run-cancellation-dispatch-rearm-receipt@v1'; + export type ClusterRunManagementRetryCommand = Readonly<{ schemaVersion: 1; operation: 'run.retry'; @@ -53,9 +63,49 @@ export type ClusterRunManagementStopCommand = Readonly<{ }>; }>; +export type ClusterRunManagementCancellationInspectCommand = Readonly<{ + schemaVersion: 1; + operation: 'run.cancellation.inspect'; + request: Readonly<{ + projectId: string; + runId: string; + requestId: string; + auditEventId: string; + failureAuditEventId: string; + body: Readonly<{ + schema: typeof RUN_CANCELLATION_DISPATCH_INSPECT_REQUEST_SCHEMA; + }>; + }>; +}>; + +export type ClusterRunManagementCancellationRearmCommand = Readonly<{ + schemaVersion: 1; + operation: 'run.cancellation.rearm'; + request: Readonly<{ + projectId: string; + runId: string; + requestId: string; + auditEventId: string; + failureAuditEventId: string; + body: Readonly<{ + schema: typeof RUN_CANCELLATION_DISPATCH_REARM_REQUEST_SCHEMA; + mutationId: string; + expectedDispatchVersion: number; + expectedLastResult: + | 'identity_mismatch' + | 'pid_mismatch' + | 'unsupported' + | 'invalid'; + retryDelayMs: number; + }>; + }>; +}>; + export type ClusterRunManagementCommand = | ClusterRunManagementRetryCommand - | ClusterRunManagementStopCommand; + | ClusterRunManagementStopCommand + | ClusterRunManagementCancellationInspectCommand + | ClusterRunManagementCancellationRearmCommand; export type ClusterRunManagementRetryTransportResult = Readonly<{ schemaVersion: 1; @@ -69,9 +119,31 @@ export type ClusterRunManagementStopTransportResult = Readonly<{ stop: Readonly; }>; +export type ClusterRunManagementCancellationInspectTransportResult = Readonly<{ + schemaVersion: 1; + operation: 'run.cancellation.inspect'; + diagnostic: Readonly< + Awaited> & { + schema: typeof RUN_CANCELLATION_DISPATCH_DIAGNOSTIC_SCHEMA; + } + >; +}>; + +export type ClusterRunManagementCancellationRearmTransportResult = Readonly<{ + schemaVersion: 1; + operation: 'run.cancellation.rearm'; + rearm: Readonly< + Awaited> & { + schema: typeof RUN_CANCELLATION_DISPATCH_REARM_RECEIPT_SCHEMA; + } + >; +}>; + export type ClusterRunManagementTransportResult = | ClusterRunManagementRetryTransportResult - | ClusterRunManagementStopTransportResult; + | ClusterRunManagementStopTransportResult + | ClusterRunManagementCancellationInspectTransportResult + | ClusterRunManagementCancellationRearmTransportResult; export interface ClusterRunManagementAuthentication { authenticate(): Promise | null>; @@ -152,7 +224,14 @@ export function normalizeClusterRunManagementCommand( const envelope = exact(value, ['schemaVersion', 'operation', 'request']); if (envelope.schemaVersion !== 1) invalid(); const operation = envelope.operation; - if (operation !== 'run.retry' && operation !== 'run.stop') invalid(); + if ( + operation !== 'run.retry' && + operation !== 'run.stop' && + operation !== 'run.cancellation.inspect' && + operation !== 'run.cancellation.rearm' + ) { + invalid(); + } const request = exact( envelope.request, operation === 'run.retry' @@ -196,6 +275,70 @@ export function normalizeClusterRunManagementCommand( }), }); } + if (operation === 'run.cancellation.inspect') { + const body = exact(request.body, ['schema']); + if (body.schema !== RUN_CANCELLATION_DISPATCH_INSPECT_REQUEST_SCHEMA) { + invalid(); + } + return Object.freeze({ + schemaVersion: 1, + operation, + request: Object.freeze({ + projectId: identifier(request.projectId), + runId: identifier(request.runId), + requestId: identifier(request.requestId), + auditEventId, + failureAuditEventId, + body: Object.freeze({ + schema: RUN_CANCELLATION_DISPATCH_INSPECT_REQUEST_SCHEMA, + }), + }), + }); + } + if (operation === 'run.cancellation.rearm') { + const body = exact(request.body, [ + 'schema', + 'mutationId', + 'expectedDispatchVersion', + 'expectedLastResult', + 'retryDelayMs', + ]); + if ( + body.schema !== RUN_CANCELLATION_DISPATCH_REARM_REQUEST_SCHEMA || + !CANCELLATION_DISPATCH_BLOCKING_RESULTS.includes( + body.expectedLastResult as never, + ) || + typeof body.expectedDispatchVersion !== 'number' || + !Number.isSafeInteger(body.expectedDispatchVersion) || + body.expectedDispatchVersion < 1 || + body.expectedDispatchVersion >= 2_147_483_647 || + typeof body.retryDelayMs !== 'number' || + !Number.isSafeInteger(body.retryDelayMs) || + body.retryDelayMs < 1_000 || + body.retryDelayMs > 24 * 60 * 60_000 + ) { + invalid(); + } + return Object.freeze({ + schemaVersion: 1, + operation, + request: Object.freeze({ + projectId: identifier(request.projectId), + runId: identifier(request.runId), + requestId: identifier(request.requestId), + auditEventId, + failureAuditEventId, + body: Object.freeze({ + schema: RUN_CANCELLATION_DISPATCH_REARM_REQUEST_SCHEMA, + mutationId: uuid(body.mutationId), + expectedDispatchVersion: body.expectedDispatchVersion, + expectedLastResult: body.expectedLastResult as + ClusterRunManagementCancellationRearmCommand['request']['body']['expectedLastResult'], + retryDelayMs: body.retryDelayMs, + }), + }), + }); + } let body: ReturnType; try { body = parseRunCancellationRequestBody(request.body); @@ -231,6 +374,8 @@ export function createClusterRunManagementTransport( !options.service || typeof options.service.retry !== 'function' || typeof options.service.stop !== 'function' || + typeof options.service.inspectCancellation !== 'function' || + typeof options.service.rearmCancellation !== 'function' || (options.now !== undefined && typeof options.now !== 'function') ) { throw new ClusterRunManagementTransportConfigurationError(); @@ -290,6 +435,47 @@ export function createClusterRunManagementTransport( retry: createRunManualRetryResponseBody(result), }); } + if (command.operation === 'run.cancellation.inspect') { + const result = await options.service.inspectCancellation({ + projectId: command.request.projectId, + runId: command.request.runId, + requestId: command.request.requestId, + auditEventId: command.request.auditEventId, + failureAuditEventId: command.request.failureAuditEventId, + principal, + }); + return Object.freeze({ + schemaVersion: 1, + operation: command.operation, + diagnostic: Object.freeze({ + schema: RUN_CANCELLATION_DISPATCH_DIAGNOSTIC_SCHEMA, + ...result, + }), + }); + } + if (command.operation === 'run.cancellation.rearm') { + const result = await options.service.rearmCancellation({ + projectId: command.request.projectId, + runId: command.request.runId, + requestId: command.request.requestId, + auditEventId: command.request.auditEventId, + failureAuditEventId: command.request.failureAuditEventId, + principal, + mutationId: command.request.body.mutationId, + expectedDispatchVersion: + command.request.body.expectedDispatchVersion, + expectedLastResult: command.request.body.expectedLastResult, + retryDelayMs: command.request.body.retryDelayMs, + }); + return Object.freeze({ + schemaVersion: 1, + operation: command.operation, + rearm: Object.freeze({ + schema: RUN_CANCELLATION_DISPATCH_REARM_RECEIPT_SCHEMA, + ...result, + }), + }); + } const result = await options.service.stop({ projectId: command.request.projectId, runId: command.request.runId, diff --git a/packages/ql3-cluster-admin/test/runManagement.test.cjs b/packages/ql3-cluster-admin/test/runManagement.test.cjs index 637b4d8c..9222ca9e 100644 --- a/packages/ql3-cluster-admin/test/runManagement.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagement.test.cjs @@ -61,7 +61,7 @@ function policyRow(role = 'operator') { }; } -function fixture(role = 'operator') { +function fixture(role = 'operator', options = {}) { const calls = []; const pool = { async query(sql, params = []) { @@ -86,13 +86,30 @@ function fixture(role = 'operator') { text.startsWith('SELECT set_config') ) return { rows: [], rowCount: 0 }; - if (text.includes('statement_timestamp()')) { + if ( + text.includes('statement_timestamp()') || + text.includes('transaction_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')) { + if (!text.includes('cancel_reason AS "cancelReason"')) { + return { + rows: [ + { + projectId: 'project-1', + runStatus: 'running', + runVersion: 6, + eventSequence: 8, + cancelRequestedAtMs: NOW - 2_000, + }, + ], + rowCount: 1, + }; + } return { rows: [ { @@ -123,12 +140,37 @@ function fixture(role = 'operator') { }; } if ( - text.startsWith('INSERT INTO "ql3"."security_audit_events"') && - text.includes('RETURNING event_id') + text.includes('FROM "ql3"."runs" WHERE id = $1') && + !text.includes('FOR UPDATE') ) { - return { rows: [{ eventId: request().auditEventId }], rowCount: 1 }; + return { + rows: [ + { + projectId: 'project-1', + runStatus: 'running', + runVersion: 6, + eventSequence: 8, + cancelRequestedAtMs: NOW - 2_000, + cancelReason: 'user', + }, + ], + rowCount: 1, + }; + } + if ( + text.includes('FROM "ql3"."run_events"') && + text.includes('dedupe_key = $2') + ) { + return { rows: [], rowCount: 0 }; + } + if ( + text.startsWith('SELECT attempt_id AS "attemptId"') && + text.includes('FROM "ql3"."run_cancellation_dispatches"') && + !text.includes('dispatchStatus') && + !text.includes('FOR UPDATE') + ) { + return { rows: [{ attemptId: 'attempt-1' }], rowCount: 1 }; } - if (text.includes('idempotency_key = $2')) return { rows: [] }; if (text.includes('WHERE run.id = $1')) { return { rows: [ @@ -150,6 +192,58 @@ function fixture(role = 'operator') { ], }; } + if (text.includes('FROM "ql3"."run_attempts"')) { + return { rows: [{ attemptStatus: 'running' }], rowCount: 1 }; + } + if ( + text.includes('FROM "ql3"."run_cancellation_dispatches"') && + text.includes('FOR UPDATE') + ) { + return { + rows: [ + { + attemptId: 'attempt-1', + dispatchStatus: 'blocked', + dispatchVersion: 3, + lastResult: options.lastResult ?? 'identity_mismatch', + }, + ], + rowCount: 1, + }; + } + if (text.includes('FROM "ql3"."run_cancellation_dispatches"')) { + return { + rows: [ + { + attemptId: 'attempt-1', + dispatchStatus: 'blocked', + dispatchVersion: 3, + dispatchCount: 1, + nextAttemptAtMs: null, + leaseExpiresAtMs: null, + lastResult: options.lastResult ?? 'identity_mismatch', + lastDispatchedAtMs: NOW - 1_500, + dispatchCreatedAtMs: NOW - 1_900, + dispatchUpdatedAtMs: NOW - 1_500, + }, + ], + rowCount: 1, + }; + } + if ( + text.startsWith( + 'UPDATE "ql3"."run_cancellation_dispatches"', + ) + ) { + return { rows: [], 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('FROM "ql3"."task_definitions"')) { return { rows: [{ enabled: true }] }; } @@ -252,3 +346,104 @@ test('authorizes run.stop and commits intent plus allowed audit together', async calls.findIndex(({ sql }) => sql === 'COMMIT'), ); }); + +test('allows a viewer to inspect only low-sensitive cancellation state', async () => { + const { calls, service } = fixture('viewer'); + const inspectRequest = { + projectId: 'project-1', + runId: 'run-1', + requestId: 'request-inspect-1', + auditEventId: '019f9500-0000-4000-8000-000000000031', + failureAuditEventId: '019f9500-0000-4000-8000-000000000032', + principal: request().principal, + }; + const result = await service.inspectCancellation(inspectRequest); + assert.equal(result.operatorAction, 'rearm'); + assert.equal(result.dispatch.lastResult, 'identity_mismatch'); + assert.equal(JSON.stringify(result).includes('leaseOwner'), false); + assert.equal(JSON.stringify(result).includes('leaseToken'), false); + const audit = calls.find( + ({ sql, params }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"') && + params[2] === 'run.cancellation.inspect', + ); + assert.equal(audit.params[0], inspectRequest.auditEventId); +}); + +test('authorizes exact cancellation rearm and keeps the event identity server-side', async () => { + const { calls, service } = fixture(); + const rearmRequest = { + projectId: 'project-1', + runId: 'run-1', + mutationId: '019f9500-0000-4000-8000-000000000041', + expectedDispatchVersion: 3, + expectedLastResult: 'identity_mismatch', + retryDelayMs: 5_000, + requestId: 'request-rearm-1', + auditEventId: '019f9500-0000-4000-8000-000000000042', + failureAuditEventId: '019f9500-0000-4000-8000-000000000043', + principal: request().principal, + }; + const result = await service.rearmCancellation(rearmRequest); + assert.equal(result.status, 'rearmed'); + assert.equal(result.dispatchVersion, 4); + const event = calls.find(({ sql }) => + sql.startsWith('INSERT INTO "ql3"."run_events"'), + ); + assert.equal(event.params[0], GENERATED[0]); + assert.equal(event.params.includes(rearmRequest.mutationId), false); + const allowedAudit = calls.find( + ({ sql, params }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"') && + params[2] === 'run.cancellation.rearm', + ); + assert.equal(allowedAudit.params[0], rearmRequest.auditEventId); + assert.ok( + calls.indexOf(allowedAudit) < calls.findIndex(({ sql }) => sql === 'COMMIT'), + ); +}); + +test('denies viewer rearm and records stale dispatch conflicts outside the transaction', async () => { + const rearmRequest = { + projectId: 'project-1', + runId: 'run-1', + mutationId: '019f9500-0000-4000-8000-000000000051', + expectedDispatchVersion: 3, + expectedLastResult: 'identity_mismatch', + retryDelayMs: 5_000, + requestId: 'request-rearm-conflict-1', + auditEventId: '019f9500-0000-4000-8000-000000000052', + failureAuditEventId: '019f9500-0000-4000-8000-000000000053', + principal: request().principal, + }; + + const viewer = fixture('viewer'); + await assert.rejects( + viewer.service.rearmCancellation(rearmRequest), + ClusterRunManagementAuthorizationError, + ); + assert.equal( + viewer.calls.some(({ scope }) => scope === 'client'), + false, + ); + + const stale = fixture('operator', { lastResult: 'pid_mismatch' }); + await assert.rejects( + stale.service.rearmCancellation(rearmRequest), + { code: 'CLUSTER_RUN_MANAGEMENT_CONFLICT' }, + ); + const failureAudit = stale.calls.find( + ({ scope, sql }) => + scope === 'pool' && + sql.startsWith('INSERT INTO "ql3"."security_audit_events"'), + ); + assert.equal(failureAudit.params[0], rearmRequest.failureAuditEventId); + assert.equal( + failureAudit.params.some( + (value) => + typeof value === 'string' && + value.includes('dispatch_result_changed'), + ), + true, + ); +}); diff --git a/packages/ql3-cluster-admin/test/runManagementClient.test.cjs b/packages/ql3-cluster-admin/test/runManagementClient.test.cjs index 16a2d223..bfd2a5a1 100644 --- a/packages/ql3-cluster-admin/test/runManagementClient.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagementClient.test.cjs @@ -117,6 +117,40 @@ const stopCommand = normalizeClusterRunManagementCommand({ }, }); +const inspectCommand = normalizeClusterRunManagementCommand({ + schemaVersion: 1, + operation: 'run.cancellation.inspect', + request: { + projectId: 'project-1', + runId: 'run-1', + requestId: 'request-inspect-1', + auditEventId: '019f9400-0000-4000-8000-000000000031', + failureAuditEventId: '019f9400-0000-4000-8000-000000000032', + body: { + schema: 'qinglong/run-cancellation-dispatch-inspect@v1', + }, + }, +}); + +const rearmCommand = normalizeClusterRunManagementCommand({ + schemaVersion: 1, + operation: 'run.cancellation.rearm', + request: { + projectId: 'project-1', + runId: 'run-1', + requestId: 'request-rearm-1', + auditEventId: '019f9400-0000-4000-8000-000000000041', + failureAuditEventId: '019f9400-0000-4000-8000-000000000042', + body: { + schema: 'qinglong/run-cancellation-dispatch-rearm-request@v1', + mutationId: '019f9400-0000-4000-8000-000000000043', + expectedDispatchVersion: 3, + expectedLastResult: 'identity_mismatch', + retryDelayMs: 5_000, + }, + }, +}); + function response(overrides = {}) { return { schemaVersion: 1, @@ -214,3 +248,105 @@ test('validates one low-sensitive stop response against the request target', () ); } }); + +test('validates a low-sensitive cancellation diagnostic and rejects capability leakage', () => { + const value = { + schemaVersion: 1, + operation: 'run.cancellation.inspect', + diagnostic: { + schema: 'qinglong/run-cancellation-dispatch-diagnostic@v1', + projectId: 'project-1', + runId: 'run-1', + runStatus: 'running', + runVersion: 6, + eventSequence: 8, + cancelRequestedAtMs: 999_000, + cancelReason: 'user', + operatorAction: 'rearm', + dispatch: { + attemptId: 'attempt-1', + status: 'blocked', + version: 3, + dispatchCount: 1, + lastResult: 'identity_mismatch', + createdAtMs: 999_100, + updatedAtMs: 999_200, + }, + }, + }; + assert.deepEqual( + validateClusterRunManagementClientResult(value, inspectCommand), + value, + ); + for (const drift of [ + { ...value, diagnostic: { ...value.diagnostic, projectId: 'project-2' } }, + { ...value, diagnostic: { ...value.diagnostic, runId: 'run-2' } }, + { + ...value, + diagnostic: { ...value.diagnostic, operatorAction: 'none' }, + }, + { + ...value, + diagnostic: { + ...value.diagnostic, + dispatch: { ...value.diagnostic.dispatch, leaseOwner: 'worker-1' }, + }, + }, + { + ...value, + diagnostic: { + ...value.diagnostic, + dispatch: { + ...value.diagnostic.dispatch, + leaseTokenDigest: 'a'.repeat(64), + }, + }, + }, + ]) { + assert.throws( + () => validateClusterRunManagementClientResult(drift, inspectCommand), + ClusterPluginPackageManagementClientRequestError, + ); + } +}); + +test('binds a rearm receipt to the exact dispatch version, result and delay fences', () => { + const value = { + schemaVersion: 1, + operation: 'run.cancellation.rearm', + rearm: { + schema: 'qinglong/run-cancellation-dispatch-rearm-receipt@v1', + status: 'rearmed', + projectId: 'project-1', + runId: 'run-1', + attemptId: 'attempt-1', + previousDispatchVersion: 3, + dispatchVersion: 4, + previousResult: 'identity_mismatch', + retryDelayMs: 5_000, + nextAttemptAtMs: 1_005_000, + runVersion: 7, + eventSequence: 9, + }, + }; + assert.deepEqual( + validateClusterRunManagementClientResult(value, rearmCommand), + value, + ); + for (const rearm of [ + { ...value.rearm, previousDispatchVersion: 4 }, + { ...value.rearm, dispatchVersion: 5 }, + { ...value.rearm, previousResult: 'pid_mismatch' }, + { ...value.rearm, retryDelayMs: 6_000 }, + { ...value.rearm, runId: 'run-2' }, + ]) { + assert.throws( + () => + validateClusterRunManagementClientResult( + { ...value, rearm }, + rearmCommand, + ), + ClusterPluginPackageManagementClientRequestError, + ); + } +}); diff --git a/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs b/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs index 83c98957..e22dee5b 100644 --- a/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs @@ -97,6 +97,84 @@ function stopResult() { }; } +function diagnosticResult() { + return { + projectId: 'project-1', + runId: 'run-1', + runStatus: 'running', + runVersion: 6, + eventSequence: 8, + cancelRequestedAtMs: NOW - 1_000, + cancelReason: 'user', + operatorAction: 'rearm', + dispatch: { + attemptId: 'attempt-1', + status: 'blocked', + version: 3, + dispatchCount: 1, + lastResult: 'identity_mismatch', + createdAtMs: NOW - 900, + updatedAtMs: NOW - 800, + }, + }; +} + +function rearmResult() { + return { + status: 'rearmed', + projectId: 'project-1', + runId: 'run-1', + attemptId: 'attempt-1', + previousDispatchVersion: 3, + dispatchVersion: 4, + previousResult: 'identity_mismatch', + retryDelayMs: 5_000, + nextAttemptAtMs: NOW + 5_000, + runVersion: 7, + eventSequence: 9, + }; +} + +function inspectCommand(overrides = {}) { + return { + schemaVersion: 1, + operation: 'run.cancellation.inspect', + request: { + projectId: 'project-1', + runId: 'run-1', + requestId: 'request-inspect-1', + auditEventId: '019f9300-0000-4000-8000-000000000031', + failureAuditEventId: '019f9300-0000-4000-8000-000000000032', + body: { + schema: 'qinglong/run-cancellation-dispatch-inspect@v1', + }, + ...overrides, + }, + }; +} + +function rearmCommand(overrides = {}) { + return { + schemaVersion: 1, + operation: 'run.cancellation.rearm', + request: { + projectId: 'project-1', + runId: 'run-1', + requestId: 'request-rearm-1', + auditEventId: '019f9300-0000-4000-8000-000000000041', + failureAuditEventId: '019f9300-0000-4000-8000-000000000042', + body: { + schema: 'qinglong/run-cancellation-dispatch-rearm-request@v1', + mutationId: '019f9300-0000-4000-8000-000000000043', + expectedDispatchVersion: 3, + expectedLastResult: 'identity_mismatch', + retryDelayMs: 5_000, + }, + ...overrides, + }, + }; +} + test('routes one exact strong User retry and emits the shared response', async () => { const calls = []; const transport = createClusterRunManagementTransport({ @@ -109,6 +187,12 @@ test('routes one exact strong User retry and emits the shared response', async ( async stop() { return stopResult(); }, + async inspectCancellation() { + return diagnosticResult(); + }, + async rearmCancellation() { + return rearmResult(); + }, }, }); const result = await transport.execute(command(), { @@ -139,6 +223,12 @@ test('routes one exact strong User stop and emits the shared response', async () calls.push(request); return stopResult(); }, + async inspectCancellation() { + return diagnosticResult(); + }, + async rearmCancellation() { + return rearmResult(); + }, }, }); const result = await transport.execute(stopCommand(), { @@ -169,6 +259,12 @@ test('rejects weak or non-User identity before service authority', async () => { async stop() { return stopResult(); }, + async inspectCancellation() { + return diagnosticResult(); + }, + async rearmCancellation() { + return rearmResult(); + }, }, }); await assert.rejects( @@ -187,6 +283,65 @@ test('rejects weak or non-User identity before service authority', async () => { assert.equal(called, false); }); +test('routes bounded cancellation inspection without lease capability data', async () => { + const calls = []; + const transport = createClusterRunManagementTransport({ + now: () => NOW, + service: { + async retry() { return retryResult(); }, + async stop() { return stopResult(); }, + async inspectCancellation(request) { + calls.push(request); + return diagnosticResult(); + }, + async rearmCancellation() { return rearmResult(); }, + }, + }); + const result = await transport.execute(inspectCommand(), { + authenticate: async () => principal(), + }); + assert.equal(calls[0].runId, 'run-1'); + assert.deepEqual(result, { + schemaVersion: 1, + operation: 'run.cancellation.inspect', + diagnostic: { + schema: 'qinglong/run-cancellation-dispatch-diagnostic@v1', + ...diagnosticResult(), + }, + }); + assert.equal(JSON.stringify(result).includes('leaseOwner'), false); + assert.equal(JSON.stringify(result).includes('leaseToken'), false); +}); + +test('routes an exact blocked cancellation rearm receipt', async () => { + const calls = []; + const transport = createClusterRunManagementTransport({ + now: () => NOW, + service: { + async retry() { return retryResult(); }, + async stop() { return stopResult(); }, + async inspectCancellation() { return diagnosticResult(); }, + async rearmCancellation(request) { + calls.push(request); + return rearmResult(); + }, + }, + }); + const result = await transport.execute(rearmCommand(), { + authenticate: async () => principal({ assurance: 'hardware' }), + }); + assert.equal(calls[0].expectedDispatchVersion, 3); + assert.equal(calls[0].expectedLastResult, 'identity_mismatch'); + assert.deepEqual(result, { + schemaVersion: 1, + operation: 'run.cancellation.rearm', + rearm: { + schema: 'qinglong/run-cancellation-dispatch-rearm-receipt@v1', + ...rearmResult(), + }, + }); +}); + test('rejects widened commands and ambiguous audit identity', () => { assert.throws( () => diff --git a/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts b/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts index 1cd35c4d..77800a28 100644 --- a/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts +++ b/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts @@ -1,4 +1,16 @@ export { PostgresRunManualRetryRepository } from '../run-management/runManualRetryRepository'; +export { + InvalidRunCancellationDispatchManagementError, + PostgresRunCancellationDispatchManagementRepository, + RunCancellationDispatchManagementConflictError, + RunCancellationDispatchManagementNotFoundError, + RunCancellationDispatchManagementUnavailableError, + type BlockingCancellationDispatchResult, + type PostgresRunCancellationDispatchInspectCommand, + type PostgresRunCancellationDispatchRearmCommand, + type RunCancellationDispatchDiagnostic, + type RunCancellationDispatchRearmReceipt, +} from '../run-management/runCancellationDispatchManagementRepository'; export { PostgresClusterRunCancellationRepository, type PostgresRunManagementCancellationCommand, diff --git a/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts b/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts index 5613ddb8..8b6a4a2e 100644 --- a/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts +++ b/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts @@ -338,5 +338,10 @@ export const postgresqlMainMigrationManifest: MigrationStreamManifest = checksum: 'b6d7ac81b5f75530df05f8ef05878fa30aa0f4418363973ded89d14ffce151b2', }), + Object.freeze({ + id: 'pg-0067-cancellation-dispatch-management', + checksum: + 'e78e24a06dc4c4dbdd859685f28b4bc837a8cfb279eb3512e0a57dc6d27eaaaa', + }), ]), }); diff --git a/packages/ql3-cluster-postgres/src/migrations/index.ts b/packages/ql3-cluster-postgres/src/migrations/index.ts index c32a0d88..bb231e12 100644 --- a/packages/ql3-cluster-postgres/src/migrations/index.ts +++ b/packages/ql3-cluster-postgres/src/migrations/index.ts @@ -69,6 +69,7 @@ import { pg0063PluginPackageSecretBindingTransitionReceiptsMigration } from './p import { pg0064PluginPackageSecretBindingTransitionApprovalPlansMigration } from './pg-0064-plugin-package-secret-binding-transition-approval-plans'; import { pg0065ApprovedActionManualRecoveryMigration } from '../approved-action/pg-0065-approved-action-manual-recovery'; import { pg0066CancellationDispatchMigration } from '../run/migrations/pg-0066-cancellation-dispatch'; +import { pg0067CancellationDispatchManagementMigration } from '../run-management/pg-0067-cancellation-dispatch-management'; export const postgresqlMainMigrationStream: MigrationStreamDefinition = Object.freeze({ @@ -143,5 +144,6 @@ export const postgresqlMainMigrationStream: MigrationStreamDefinition; + +export type BlockingCancellationDispatchResult = + (typeof CANCELLATION_DISPATCH_BLOCKING_RESULTS)[number]; + +export type RunCancellationDispatchDiagnostic = Readonly<{ + projectId: string; + runId: string; + runStatus: RunStatus; + runVersion: number; + eventSequence: number; + cancelRequestedAtMs?: number; + cancelReason?: 'user' | 'policy' | 'shutdown' | 'reconcile' | 'timeout'; + operatorAction: 'none' | 'wait' | 'rearm'; + dispatch: Readonly<{ + attemptId: string; + status: CancellationDispatchStatus; + version: number; + dispatchCount: number; + nextAttemptAtMs?: number; + leaseExpiresAtMs?: number; + lastResult?: CancellationDispatchResult; + lastDispatchedAtMs?: number; + createdAtMs: number; + updatedAtMs: number; + }> | null; +}>; + +export type RunCancellationDispatchRearmReceipt = Readonly<{ + status: 'rearmed'; + projectId: string; + runId: string; + attemptId: string; + previousDispatchVersion: number; + dispatchVersion: number; + previousResult: BlockingCancellationDispatchResult; + retryDelayMs: number; + nextAttemptAtMs: number; + runVersion: number; + eventSequence: number; +}>; + +interface ManagementAuthority { + readonly projectId: string; + readonly runId: string; + readonly requestId: string; + readonly auditEventId: string; + readonly principal: Readonly; + readonly policyFence: Readonly; +} + +export interface PostgresRunCancellationDispatchInspectCommand + extends ManagementAuthority {} + +export interface PostgresRunCancellationDispatchRearmCommand + extends ManagementAuthority { + readonly mutationId: string; + readonly eventId: string; + readonly expectedDispatchVersion: number; + readonly expectedLastResult: BlockingCancellationDispatchResult; + readonly retryDelayMs: number; +} + +export class InvalidRunCancellationDispatchManagementError extends TypeError { + readonly code = 'RUN_CANCELLATION_DISPATCH_MANAGEMENT_INVALID'; + constructor() { + super('Run cancellation dispatch management input is invalid'); + this.name = 'InvalidRunCancellationDispatchManagementError'; + } +} + +export class RunCancellationDispatchManagementNotFoundError extends Error { + readonly code = 'RUN_CANCELLATION_DISPATCH_MANAGEMENT_NOT_FOUND'; + constructor() { + super('Run cancellation dispatch management target is unavailable'); + this.name = 'RunCancellationDispatchManagementNotFoundError'; + } +} + +export class RunCancellationDispatchManagementConflictError extends Error { + readonly code = 'RUN_CANCELLATION_DISPATCH_MANAGEMENT_CONFLICT'; + constructor( + readonly reason: + | 'authorization_changed' + | 'run_terminal' + | 'cancellation_missing' + | 'dispatch_missing' + | 'dispatch_not_blocked' + | 'dispatch_version_changed' + | 'dispatch_result_changed' + | 'attempt_not_active' + | 'mutation_conflict', + ) { + super(`Run cancellation dispatch management conflict: ${reason}`); + this.name = 'RunCancellationDispatchManagementConflictError'; + } +} + +export class RunCancellationDispatchManagementUnavailableError extends Error { + readonly code = 'RUN_CANCELLATION_DISPATCH_MANAGEMENT_UNAVAILABLE'; + constructor(options?: ErrorOptions) { + super('Run cancellation dispatch management is unavailable', options); + this.name = 'RunCancellationDispatchManagementUnavailableError'; + } +} + +const IDENTIFIER_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u; +const UUID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/u; +const STRONG_ASSURANCES = new Set(['multi_factor', 'hardware']); +const ACTIVE_RUN_STATUSES = new Set([ + 'created', + 'queued', + 'dispatching', + 'running', + 'waiting_approval', + 'retry_wait', + 'lost', +]); +const ACTIVE_ATTEMPT_STATUSES = new Set(['claimed', 'starting', 'running']); +const CANCEL_REASONS = new Set([ + 'user', + 'policy', + 'shutdown', + 'reconcile', + 'timeout', +]); +const MAX_AUTHENTICATION_AGE_MS = 5 * 60_000; +const MIN_MANUAL_RETRY_DELAY_MS = 1_000; +const REARM_SCHEMA = 'qinglong/run-cancellation-dispatch-rearm@v1'; + +function invalid(): never { + throw new InvalidRunCancellationDispatchManagementError(); +} + +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(); + if ( + actual.length !== expected.length || + actual.some((key, index) => key !== expected[index]) + ) { + invalid(); + } + return value as Record; +} + +function text(row: Row, key: string): string { + const value = row[key]; + if (typeof value !== 'string' || value.length < 1) { + throw new TypeError(`PostgreSQL cancellation management ${key} is invalid`); + } + return value; +} + +function integer(row: Row, key: string): number { + const raw = row[key]; + const value = + typeof raw === 'string' && /^(0|[1-9]\d*)$/u.test(raw) + ? Number(raw) + : raw; + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) { + throw new TypeError(`PostgreSQL cancellation management ${key} is invalid`); + } + return value; +} + +function optionalInteger(row: Row, key: string): number | undefined { + return row[key] === null || row[key] === undefined + ? undefined + : integer(row, key); +} + +function optionalText(row: Row, key: string): string | undefined { + return row[key] === null || row[key] === undefined + ? undefined + : text(row, key); +} + +function identifier(value: unknown): string { + if (typeof value !== 'string' || !IDENTIFIER_PATTERN.test(value)) invalid(); + return value; +} + +function uuid(value: unknown): string { + if (typeof value !== 'string' || !UUID_PATTERN.test(value)) invalid(); + return value; +} + +function boundedInteger( + value: unknown, + minimum: number, + maximum = Number.MAX_SAFE_INTEGER, +): number { + if ( + typeof value !== 'number' || + !Number.isSafeInteger(value) || + value < minimum || + value > maximum + ) { + invalid(); + } + return value; +} + +function storedInteger( + value: unknown, + name: string, + minimum = 0, + maximum = Number.MAX_SAFE_INTEGER, +): number { + if ( + typeof value !== 'number' || + !Number.isSafeInteger(value) || + value < minimum || + value > maximum + ) { + throw new TypeError(`PostgreSQL cancellation management ${name} is invalid`); + } + return value; +} + +function normalizeAuthority( + value: unknown, + extraKeys: readonly string[], +): Readonly & Record { + const input = exact(value, [ + 'projectId', + 'runId', + 'requestId', + 'auditEventId', + 'principal', + 'policyFence', + ...extraKeys, + ]); + const principal = exact(input.principal, [ + 'subject', + 'authenticationId', + 'authenticatedAtMs', + 'expiresAtMs', + 'assurance', + ]) as unknown as SecurityPrincipal; + const fence = exact(input.policyFence, [ + 'projectVersion', + 'bindingVersion', + ]); + return Object.freeze({ + ...input, + projectId: identifier(input.projectId), + runId: identifier(input.runId), + requestId: identifier(input.requestId), + auditEventId: uuid(input.auditEventId), + principal, + policyFence: Object.freeze({ + projectVersion: boundedInteger(fence.projectVersion, 1, 2_147_483_647), + bindingVersion: boundedInteger(fence.bindingVersion, 1, 2_147_483_647), + }), + }); +} + +function normalizeInspectCommand( + value: Readonly, +): Readonly { + return normalizeAuthority(value, []); +} + +function normalizeRearmCommand( + value: Readonly, +): Readonly { + const input = normalizeAuthority(value, [ + 'mutationId', + 'eventId', + 'expectedDispatchVersion', + 'expectedLastResult', + 'retryDelayMs', + ]); + if ( + !CANCELLATION_DISPATCH_BLOCKING_RESULTS.includes( + input.expectedLastResult as BlockingCancellationDispatchResult, + ) + ) { + invalid(); + } + const eventId = uuid(input.eventId); + if (eventId === input.auditEventId) invalid(); + return Object.freeze({ + projectId: input.projectId, + runId: input.runId, + requestId: input.requestId, + auditEventId: input.auditEventId, + principal: input.principal, + policyFence: input.policyFence, + mutationId: uuid(input.mutationId), + eventId, + expectedDispatchVersion: boundedInteger( + input.expectedDispatchVersion, + 1, + 2_147_483_646, + ), + expectedLastResult: + input.expectedLastResult as BlockingCancellationDispatchResult, + retryDelayMs: boundedInteger( + input.retryDelayMs, + MIN_MANUAL_RETRY_DELAY_MS, + MAX_CANCELLATION_DISPATCH_RETRY_DELAY_MS, + ), + }); +} + +async function begin(client: PostgresClient): Promise { + await client.query('BEGIN ISOLATION LEVEL SERIALIZABLE'); + await client.query(`SELECT set_config('statement_timeout', $1, true)`, [ + '5000ms', + ]); + await client.query(`SELECT set_config('lock_timeout', $1, true)`, ['1000ms']); + await client.query( + `SELECT set_config('idle_in_transaction_session_timeout', $1, true)`, + ['10000ms'], + ); +} + +async function rollback(client: PostgresClient): Promise { + try { + await client.query('ROLLBACK'); + } catch { + // Preserve the transaction failure. + } +} + +async function databaseNow(client: PostgresClient): Promise { + const result = await client.query(` + SELECT floor(extract(epoch FROM transaction_timestamp()) * 1000)::bigint + AS "nowMs" + `); + if (result.rows.length !== 1) { + throw new TypeError('PostgreSQL cancellation management clock is invalid'); + } + return integer(result.rows[0]!, 'nowMs'); +} + +function strongPrincipal( + value: Readonly, + observedAtMs: number, +): Readonly { + let principal: Readonly; + try { + principal = normalizeSecurityPrincipal(value, observedAtMs); + } catch { + throw new RunCancellationDispatchManagementConflictError( + 'authorization_changed', + ); + } + if ( + principal.subject.type !== 'user' || + !STRONG_ASSURANCES.has(principal.assurance) || + principal.authenticatedAtMs > observedAtMs || + observedAtMs - principal.authenticatedAtMs > MAX_AUTHENTICATION_AGE_MS + ) { + throw new RunCancellationDispatchManagementConflictError( + '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.principal.subject.type, + command.principal.subject.id, + command.policyFence.projectVersion, + command.policyFence.bindingVersion, + ], + ); + if (result.rows.length !== 1 || result.rows[0]?.matches !== true) { + throw new RunCancellationDispatchManagementConflictError( + 'authorization_changed', + ); + } +} + +async function recordAllowedAudit( + client: PostgresClient, + command: Readonly, + operationId: 'run.cancellation.inspect' | 'run.cancellation.rearm', + 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, $3, $4, 'user', $5, $6, 'allowed', $7::jsonb, + $8, $9, $10) + ON CONFLICT (event_id) DO NOTHING RETURNING event_id AS "eventId"`, + [ + command.auditEventId, + command.requestId, + operationId, + command.projectId, + command.principal.subject.id, + command.principal.authenticationId, + JSON.stringify(['role_grant', 'strong_authentication']), + command.policyFence.projectVersion, + command.policyFence.bindingVersion, + observedAtMs, + ], + ); + if (inserted.rows.length === 1) return; + 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`, + [command.auditEventId], + ); + const row = replay.rows[0]; + if ( + replay.rows.length !== 1 || + !row || + row.requestId !== command.requestId || + row.operationId !== operationId || + row.projectId !== command.projectId || + row.subjectType !== 'user' || + row.subjectId !== command.principal.subject.id || + row.authenticationId !== command.principal.authenticationId || + row.outcome !== 'allowed' || + !Array.isArray(row.reasons) || + row.reasons.join('\0') !== 'role_grant\0strong_authentication' || + integer(row, 'projectVersion') !== command.policyFence.projectVersion || + integer(row, 'bindingVersion') !== command.policyFence.bindingVersion + ) { + throw new RunCancellationDispatchManagementConflictError( + 'mutation_conflict', + ); + } +} + +function runStatus(row: Row): RunStatus { + const value = text(row, 'runStatus') as RunStatus; + if (!RUN_STATUSES.includes(value)) { + throw new TypeError('PostgreSQL cancellation management Run status is invalid'); + } + return value; +} + +function dispatchProjection(row: Row): NonNullable { + const status = text(row, 'dispatchStatus') as CancellationDispatchStatus; + const lastResult = optionalText(row, 'lastResult') as + | CancellationDispatchResult + | undefined; + if ( + !CANCELLATION_DISPATCH_STATUSES.includes(status) || + (lastResult !== undefined && + !CANCELLATION_DISPATCH_RESULTS.includes(lastResult)) + ) { + throw new TypeError('PostgreSQL cancellation management dispatch is invalid'); + } + const nextAttemptAtMs = optionalInteger(row, 'nextAttemptAtMs'); + const leaseExpiresAtMs = optionalInteger(row, 'leaseExpiresAtMs'); + const lastDispatchedAtMs = optionalInteger(row, 'lastDispatchedAtMs'); + return Object.freeze({ + attemptId: text(row, 'attemptId'), + status, + version: integer(row, 'dispatchVersion'), + dispatchCount: integer(row, 'dispatchCount'), + ...(nextAttemptAtMs === undefined ? {} : { nextAttemptAtMs }), + ...(leaseExpiresAtMs === undefined ? {} : { leaseExpiresAtMs }), + ...(lastResult === undefined ? {} : { lastResult }), + ...(lastDispatchedAtMs === undefined ? {} : { lastDispatchedAtMs }), + createdAtMs: integer(row, 'dispatchCreatedAtMs'), + updatedAtMs: integer(row, 'dispatchUpdatedAtMs'), + }); +} + +function diagnostic( + command: Readonly, + run: Row, + dispatchRow?: Row, +): Readonly { + const cancelRequestedAtMs = optionalInteger(run, 'cancelRequestedAtMs'); + const cancelReason = optionalText(run, 'cancelReason'); + if ( + (cancelRequestedAtMs === undefined) !== (cancelReason === undefined) || + (cancelReason !== undefined && !CANCEL_REASONS.has(cancelReason)) + ) { + throw new TypeError('PostgreSQL cancellation management intent is invalid'); + } + const dispatch = dispatchRow ? dispatchProjection(dispatchRow) : null; + return Object.freeze({ + projectId: command.projectId, + runId: command.runId, + runStatus: runStatus(run), + runVersion: integer(run, 'runVersion'), + eventSequence: integer(run, 'eventSequence'), + ...(cancelRequestedAtMs === undefined + ? {} + : { + cancelRequestedAtMs, + cancelReason: cancelReason as NonNullable< + RunCancellationDispatchDiagnostic['cancelReason'] + >, + }), + operatorAction: + dispatch?.status === 'blocked' + ? 'rearm' + : dispatch && dispatch.status !== 'dispatched' + ? 'wait' + : cancelRequestedAtMs !== undefined && !dispatch + ? 'wait' + : 'none', + dispatch, + }); +} + +function rearmReceiptFromEvent( + command: Readonly, + event: Row, +): Readonly { + const payload = exact(event.payload, [ + 'schema', + 'mutation_id', + 'previous_dispatch_version', + 'dispatch_version', + 'previous_result', + 'retry_delay_ms', + 'next_attempt_at_ms', + 'run_version', + ]); + if ( + text(event, 'eventId') !== command.eventId || + text(event, 'eventType') !== 'run.cancel_dispatch_rearmed' || + text(event, 'actorType') !== 'user' || + text(event, 'actorId') !== command.principal.subject.id || + payload.schema !== REARM_SCHEMA || + payload.mutation_id !== command.mutationId || + payload.previous_dispatch_version !== command.expectedDispatchVersion || + payload.previous_result !== command.expectedLastResult || + payload.retry_delay_ms !== command.retryDelayMs + ) { + throw new RunCancellationDispatchManagementConflictError( + 'mutation_conflict', + ); + } + return Object.freeze({ + status: 'rearmed', + projectId: command.projectId, + runId: command.runId, + attemptId: text(event, 'attemptId'), + previousDispatchVersion: storedInteger( + payload.previous_dispatch_version, + 'previousDispatchVersion', + 1, + ), + dispatchVersion: storedInteger( + payload.dispatch_version, + 'dispatchVersion', + 2, + ), + previousResult: + payload.previous_result as BlockingCancellationDispatchResult, + retryDelayMs: storedInteger( + payload.retry_delay_ms, + 'retryDelayMs', + MIN_MANUAL_RETRY_DELAY_MS, + MAX_CANCELLATION_DISPATCH_RETRY_DELAY_MS, + ), + nextAttemptAtMs: storedInteger( + payload.next_attempt_at_ms, + 'nextAttemptAtMs', + ), + runVersion: storedInteger(payload.run_version, 'runVersion', 1), + eventSequence: integer(event, 'eventSequence'), + }); +} + +export class PostgresRunCancellationDispatchManagementRepository { + constructor(private readonly pool: PostgresPool) { + if (!pool || typeof pool.connect !== 'function') { + throw new InvalidRunCancellationDispatchManagementError(); + } + } + + inspect( + value: Readonly, + ): Promise> { + const command = normalizeInspectCommand(value); + return this.transaction(async (client) => { + const observedAtMs = await databaseNow(client); + const authorized = Object.freeze({ + ...command, + principal: strongPrincipal(command.principal, observedAtMs), + }); + await confirmAuthorization(client, authorized); + 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`, + [command.runId], + ); + if ( + run.rows.length !== 1 || + run.rows[0]?.projectId !== command.projectId + ) { + throw new RunCancellationDispatchManagementNotFoundError(); + } + const dispatch = await client.query( + `SELECT attempt_id AS "attemptId", status AS "dispatchStatus", + version AS "dispatchVersion", dispatch_count AS "dispatchCount", + next_attempt_at_ms AS "nextAttemptAtMs", + lease_expires_at_ms AS "leaseExpiresAtMs", + last_result AS "lastResult", + last_dispatched_at_ms AS "lastDispatchedAtMs", + created_at_ms AS "dispatchCreatedAtMs", + updated_at_ms AS "dispatchUpdatedAtMs" + FROM "ql3"."run_cancellation_dispatches" WHERE run_id = $1`, + [command.runId], + ); + if (dispatch.rows.length > 1) { + throw new TypeError('PostgreSQL cancellation management dispatch duplicated'); + } + await recordAllowedAudit( + client, + authorized, + 'run.cancellation.inspect', + observedAtMs, + ); + return diagnostic(authorized, run.rows[0]!, dispatch.rows[0]); + }); + } + + rearm( + value: Readonly, + ): Promise> { + const command = normalizeRearmCommand(value); + return this.transaction(async (client) => { + const observedAtMs = await databaseNow(client); + const authorized = Object.freeze({ + ...command, + principal: strongPrincipal(command.principal, observedAtMs), + }); + await confirmAuthorization(client, authorized); + 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" + FROM "ql3"."runs" WHERE id = $1 FOR UPDATE`, + [command.runId], + ); + if ( + run.rows.length !== 1 || + run.rows[0]?.projectId !== command.projectId + ) { + throw new RunCancellationDispatchManagementNotFoundError(); + } + const dedupeKey = `cancel-dispatch-rearm:${command.mutationId}`; + const replay = await client.query( + `SELECT id AS "eventId", sequence AS "eventSequence", + type AS "eventType", actor_type AS "actorType", + actor_id AS "actorId", attempt_id AS "attemptId", payload + FROM "ql3"."run_events" + WHERE run_id = $1 AND dedupe_key = $2`, + [command.runId, dedupeKey], + ); + if (replay.rows.length === 1) { + const receipt = rearmReceiptFromEvent(authorized, replay.rows[0]!); + await recordAllowedAudit( + client, + authorized, + 'run.cancellation.rearm', + observedAtMs, + ); + return receipt; + } + if (replay.rows.length !== 0) { + throw new RunCancellationDispatchManagementConflictError( + 'mutation_conflict', + ); + } + if (!ACTIVE_RUN_STATUSES.has(runStatus(run.rows[0]!))) { + throw new RunCancellationDispatchManagementConflictError('run_terminal'); + } + if (optionalInteger(run.rows[0]!, 'cancelRequestedAtMs') === undefined) { + throw new RunCancellationDispatchManagementConflictError( + 'cancellation_missing', + ); + } + const candidate = await client.query( + `SELECT attempt_id AS "attemptId" + FROM "ql3"."run_cancellation_dispatches" WHERE run_id = $1`, + [command.runId], + ); + if (candidate.rows.length !== 1) { + throw new RunCancellationDispatchManagementConflictError( + 'dispatch_missing', + ); + } + const attemptId = text(candidate.rows[0]!, 'attemptId'); + const attempt = await client.query( + `SELECT status AS "attemptStatus" FROM "ql3"."run_attempts" + WHERE run_id = $1 AND id = $2`, + [command.runId, attemptId], + ); + if ( + attempt.rows.length !== 1 || + !ACTIVE_ATTEMPT_STATUSES.has(text(attempt.rows[0]!, 'attemptStatus')) + ) { + throw new RunCancellationDispatchManagementConflictError( + 'attempt_not_active', + ); + } + const dispatch = await client.query( + `SELECT attempt_id AS "attemptId", status AS "dispatchStatus", + version AS "dispatchVersion", last_result AS "lastResult" + FROM "ql3"."run_cancellation_dispatches" + WHERE run_id = $1 FOR UPDATE`, + [command.runId], + ); + if (dispatch.rows.length !== 1) { + throw new RunCancellationDispatchManagementConflictError( + 'dispatch_missing', + ); + } + const current = dispatch.rows[0]!; + if ( + text(current, 'attemptId') !== attemptId || + text(current, 'dispatchStatus') !== 'blocked' + ) { + throw new RunCancellationDispatchManagementConflictError( + 'dispatch_not_blocked', + ); + } + if (integer(current, 'dispatchVersion') !== command.expectedDispatchVersion) { + throw new RunCancellationDispatchManagementConflictError( + 'dispatch_version_changed', + ); + } + if (text(current, 'lastResult') !== command.expectedLastResult) { + throw new RunCancellationDispatchManagementConflictError( + 'dispatch_result_changed', + ); + } + const nextAttemptAtMs = observedAtMs + command.retryDelayMs; + const runVersion = integer(run.rows[0]!, 'runVersion'); + const eventSequence = integer(run.rows[0]!, 'eventSequence'); + if ( + !Number.isSafeInteger(nextAttemptAtMs) || + runVersion >= 2_147_483_647 || + eventSequence >= 2_147_483_647 + ) { + throw new TypeError('PostgreSQL cancellation management counter overflowed'); + } + const dispatchVersion = command.expectedDispatchVersion + 1; + const nextRunVersion = runVersion + 1; + const nextEventSequence = eventSequence + 1; + const runUpdated = await client.query( + `UPDATE "ql3"."runs" + SET version = $2, event_sequence = $3 + WHERE id = $1 AND version = $4`, + [command.runId, nextRunVersion, nextEventSequence, runVersion], + ); + if (runUpdated.rowCount !== 1) { + throw new RunCancellationDispatchManagementConflictError( + 'dispatch_version_changed', + ); + } + const dispatchUpdated = await client.query( + `UPDATE "ql3"."run_cancellation_dispatches" + SET status = 'retry_wait', version = $2, + next_attempt_at_ms = $3, updated_at_ms = $4 + WHERE run_id = $1 AND attempt_id = $5 AND status = 'blocked' + AND version = $6 AND last_result = $7`, + [ + command.runId, + dispatchVersion, + nextAttemptAtMs, + observedAtMs, + attemptId, + command.expectedDispatchVersion, + command.expectedLastResult, + ], + ); + if (dispatchUpdated.rowCount !== 1) { + throw new RunCancellationDispatchManagementConflictError( + 'dispatch_version_changed', + ); + } + const payload = Object.freeze({ + schema: REARM_SCHEMA, + mutation_id: command.mutationId, + previous_dispatch_version: command.expectedDispatchVersion, + dispatch_version: dispatchVersion, + previous_result: command.expectedLastResult, + retry_delay_ms: command.retryDelayMs, + next_attempt_at_ms: nextAttemptAtMs, + run_version: nextRunVersion, + }); + 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_dispatch_rearmed', $4, + 'user', $5, $6, NULL, $7::jsonb, $8)`, + [ + command.eventId, + command.runId, + nextEventSequence, + dedupeKey, + authorized.principal.subject.id, + attemptId, + JSON.stringify(payload), + observedAtMs, + ], + ); + await recordAllowedAudit( + client, + authorized, + 'run.cancellation.rearm', + observedAtMs, + ); + return Object.freeze({ + status: 'rearmed', + projectId: command.projectId, + runId: command.runId, + attemptId, + previousDispatchVersion: command.expectedDispatchVersion, + dispatchVersion, + previousResult: command.expectedLastResult, + retryDelayMs: command.retryDelayMs, + nextAttemptAtMs, + runVersion: nextRunVersion, + eventSequence: nextEventSequence, + }); + }); + } + + private async transaction( + operation: (client: PostgresClient) => Promise, + ): Promise { + let client: PostgresClient | undefined; + try { + client = await this.pool.connect(); + await begin(client); + const result = await operation(client); + await client.query('COMMIT'); + return result; + } catch (error) { + if (client) await rollback(client); + if ( + error instanceof InvalidRunCancellationDispatchManagementError || + error instanceof RunCancellationDispatchManagementNotFoundError || + error instanceof RunCancellationDispatchManagementConflictError + ) { + throw error; + } + throw new RunCancellationDispatchManagementUnavailableError({ + cause: error, + }); + } finally { + client?.release(); + } + } +} diff --git a/packages/ql3-cluster-postgres/src/schema/schemaContract.ts b/packages/ql3-cluster-postgres/src/schema/schemaContract.ts index 734594e2..7fd1e01d 100644 --- a/packages/ql3-cluster-postgres/src/schema/schemaContract.ts +++ b/packages/ql3-cluster-postgres/src/schema/schemaContract.ts @@ -21,13 +21,14 @@ export interface PostgresSchemaContractTrigger { export interface PostgresSchemaContract { readonly schema: 'ql3'; readonly contractName: 'control-core'; - readonly contractVersion: 65; - readonly migrationId: 'pg-0066-cancellation-dispatch'; + readonly contractVersion: 66; + readonly migrationId: 'pg-0067-cancellation-dispatch-management'; readonly minimumServerMajor: 16; readonly maximumServerMajor: 18; readonly capabilities: Readonly<{ run_core: 1; run_cancellation_dispatch: 1; + run_cancellation_dispatch_management: 1; run_attempt_log_retention: 1; run_management_boundary: 1; run_management_stop: 1; @@ -119,8 +120,8 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = Object.freeze({ schema: 'ql3', contractName: 'control-core', - contractVersion: 65, - migrationId: 'pg-0066-cancellation-dispatch', + contractVersion: 66, + migrationId: 'pg-0067-cancellation-dispatch-management', minimumServerMajor: 16, maximumServerMajor: 18, capabilities: Object.freeze({ @@ -169,6 +170,7 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = project_tool_definition_snapshot: 1, run_core: 1, run_cancellation_dispatch: 1, + run_cancellation_dispatch_management: 1, run_attempt_log_retention: 1, run_management_boundary: 1, run_management_stop: 1, diff --git a/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts b/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts index e16edce4..8274e67d 100644 --- a/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts +++ b/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts @@ -1485,6 +1485,8 @@ const REQUIRED_RUN_MANAGER_PRIVILEGES: RequiredPrivileges = Object.freeze( name === 'run_events' || name === 'security_audit_events' ? { ...NO_TABLE_PRIVILEGES, select: true, insert: true } + : name === 'run_cancellation_dispatches' + ? { ...NO_TABLE_PRIVILEGES, select: true } : name === 'plugin_package_identity_keyset_ledger' ? { ...NO_TABLE_PRIVILEGES, @@ -2237,6 +2239,14 @@ async function assertRunManagerColumnPrivileges( 'missing-runs-contract', ]); } + const dispatch = contract.tables.find( + ({ name }) => name === 'run_cancellation_dispatches', + ); + if (!dispatch) { + throw new PostgresSchemaReadinessError('run_manager_role_invalid', [ + 'missing-run-cancellation-dispatch-contract', + ]); + } const result = await queryable.query( ` SELECT @@ -2269,6 +2279,41 @@ ORDER BY requested.column_name if (actual.size !== run.columns.length) { findings.push('column-privilege-row-count:runs'); } + const dispatchResult = await queryable.query( + ` +SELECT + requested.column_name AS "columnName", + has_column_privilege( + current_user, + format('%I.%I', $1::text, 'run_cancellation_dispatches'), + requested.column_name, + 'UPDATE' + ) AS "updateAllowed" +FROM unnest($2::text[]) AS requested(column_name) +ORDER BY requested.column_name + `.trim(), + [contract.schema, dispatch.columns], + ); + const allowedDispatch = new Set([ + 'status', + 'version', + 'next_attempt_at_ms', + 'updated_at_ms', + ]); + const actualDispatch = new Map( + dispatchResult.rows.map((row) => [row.columnName, row]), + ); + for (const columnName of dispatch.columns) { + const row = actualDispatch.get(columnName); + if (!row || row.updateAllowed !== allowedDispatch.has(columnName)) { + findings.push( + `column-update-privilege:run_cancellation_dispatches.${columnName}`, + ); + } + } + if (actualDispatch.size !== dispatch.columns.length) { + findings.push('column-privilege-row-count:run_cancellation_dispatches'); + } if (findings.length > 0) { throw new PostgresSchemaReadinessError( 'run_manager_role_invalid', diff --git a/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs b/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs index f6227871..dce51475 100644 --- a/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs @@ -117,6 +117,7 @@ test('defines the immutable PostgreSQL capability and Run core stream', async () 'pg-0064-plugin-package-secret-binding-transition-approval-plans', 'pg-0065-approved-action-manual-recovery', 'pg-0066-cancellation-dispatch', + 'pg-0067-cancellation-dispatch-management', ], ); for (const migration of postgresqlMainMigrationStream.migrations) { @@ -585,6 +586,11 @@ test('freezes every published PostgreSQL migration checksum', () => { checksum: 'b6d7ac81b5f75530df05f8ef05878fa30aa0f4418363973ded89d14ffce151b2', }, + { + id: 'pg-0067-cancellation-dispatch-management', + checksum: + 'e78e24a06dc4c4dbdd859685f28b4bc837a8cfb279eb3512e0a57dc6d27eaaaa', + }, ]; assert.deepEqual( postgresqlMainMigrationStream.migrations.map(({ id, checksum }) => ({ @@ -2329,3 +2335,41 @@ test('advances capability v65 with database-timed fenced cancellation dispatch', /migration_id = 'pg-0065-approved-action-manual-recovery'/, ); }); + +test('advances capability v66 with least-privilege cancellation diagnostics and rearm', async () => { + const migration = migrationById( + 'pg-0067-cancellation-dispatch-management', + ); + const statements = []; + await migration.up({ + async query(statement) { + statements.push(statement); + return { rows: [] }; + }, + }); + const sql = statements.join('\n'); + assert.match( + sql, + /DROP CONSTRAINT ql3_run_cancellation_dispatch_result_state_check/, + ); + assert.match( + sql, + /status IN \('leased', 'retry_wait'\)[\s\S]+identity_mismatch[\s\S]+dispatch_error/, + ); + assert.match( + sql, + /GRANT SELECT ON "ql3"\."run_cancellation_dispatches" TO ql3_run_manager/, + ); + assert.match( + sql, + /GRANT UPDATE \(status, version, next_attempt_at_ms, updated_at_ms\) ON "ql3"\."run_cancellation_dispatches" TO ql3_run_manager/, + ); + assert.doesNotMatch( + sql, + /GRANT (?:INSERT|DELETE|TRUNCATE)[^;]+run_cancellation_dispatches[^;]+ql3_run_manager/, + ); + assert.match(sql, /contract_version = 66/); + assert.match(sql, /"run_cancellation_dispatch_management":1/); + assert.match(sql, /contract_version = 65/); + assert.match(sql, /migration_id = 'pg-0066-cancellation-dispatch'/); +}); diff --git a/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs b/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs index a4a88c6e..133be40d 100644 --- a/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs @@ -527,6 +527,7 @@ function runManagerPrivileges() { 'runs', 'run_attempts', 'run_events', + 'run_cancellation_dispatches', 'security_audit_events', 'plugin_package_identity_keyset_ledger', ]); @@ -781,19 +782,36 @@ function queryable(overrides = {}) { }; } if (text.includes('has_column_privilege')) { - assert.match(text, /format\('%I\.%I', \$1::text, 'runs'\)/); + const dispatchManagement = text.includes( + "format('%I.%I', $1::text, 'run_cancellation_dispatches')", + ); + const tableName = dispatchManagement + ? 'run_cancellation_dispatches' + : 'runs'; + assert.match( + text, + new RegExp( + `format\\('%I\\.%I', \\$1::text, '${tableName}'\\)`, + ), + ); const columns = contract.tables.find( - ({ name }) => name === 'runs', + ({ name }) => name === tableName, ).columns; - const allowed = new Set([ - 'cancel_requested_at_ms', - 'cancel_reason', - 'version', - 'event_sequence', - ]); + const allowed = new Set( + dispatchManagement + ? ['status', 'version', 'next_attempt_at_ms', 'updated_at_ms'] + : [ + 'cancel_requested_at_ms', + 'cancel_reason', + 'version', + 'event_sequence', + ], + ); return { rows: - overrides.runManagerColumnPrivileges ?? + (dispatchManagement + ? overrides.runManagerDispatchColumnPrivileges + : overrides.runManagerColumnPrivileges) ?? columns.map((columnName) => ({ columnName, updateAllowed: allowed.has(columnName), @@ -817,7 +835,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro serverMajor: 16, currentUser: 'ql3_runtime', contractName: 'control-core', - contractVersion: 65, + contractVersion: 66, migrationIds: [ 'pg-0001-schema-capability', 'pg-0002-run-core', @@ -885,6 +903,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro 'pg-0064-plugin-package-secret-binding-transition-approval-plans', 'pg-0065-approved-action-manual-recovery', 'pg-0066-cancellation-dispatch', + 'pg-0067-cancellation-dispatch-management', ], }); }); @@ -915,10 +934,10 @@ test('accepts the exact schema and isolated least-privilege admin role', async ( }), ); assert.equal(report.currentUser, 'ql3_admin'); - assert.equal(report.contractVersion, 65); + assert.equal(report.contractVersion, 66); assert.equal( report.migrationIds.at(-1), - 'pg-0066-cancellation-dispatch', + 'pg-0067-cancellation-dispatch-management', ); }); @@ -931,10 +950,10 @@ test('accepts the isolated least-privilege automation manager role', async () => }), ); assert.equal(report.currentUser, 'ql3_automation_manager'); - assert.equal(report.contractVersion, 65); + assert.equal(report.contractVersion, 66); assert.equal( report.migrationIds.at(-1), - 'pg-0066-cancellation-dispatch', + 'pg-0067-cancellation-dispatch-management', ); const widened = automationManagerPrivileges(); @@ -963,10 +982,10 @@ test('accepts the isolated least-privilege human Approval manager role', async ( }), ); assert.equal(report.currentUser, 'ql3_approval_manager'); - assert.equal(report.contractVersion, 65); + assert.equal(report.contractVersion, 66); assert.equal( report.migrationIds.at(-1), - 'pg-0066-cancellation-dispatch', + 'pg-0067-cancellation-dispatch-management', ); const widened = approvalManagerPrivileges(); @@ -997,10 +1016,10 @@ test('accepts the isolated least-privilege Run manager role', async () => { }), ); assert.equal(report.currentUser, 'ql3_run_manager'); - assert.equal(report.contractVersion, 65); + assert.equal(report.contractVersion, 66); assert.equal( report.migrationIds.at(-1), - 'pg-0066-cancellation-dispatch', + 'pg-0067-cancellation-dispatch-management', ); const widened = runManagerPrivileges(); @@ -1045,6 +1064,35 @@ test('accepts the isolated least-privilege Run manager role', async () => { error.code === 'run_manager_role_invalid' && error.facts.includes('column-update-privilege:runs.status'), ); + + const widenedDispatchColumns = postgresqlControlSchemaContract.tables + .find(({ name }) => name === 'run_cancellation_dispatches') + .columns.map((columnName) => ({ + columnName, + updateAllowed: [ + 'status', + 'version', + 'next_attempt_at_ms', + 'updated_at_ms', + 'lease_token_digest', + ].includes(columnName), + })); + await assert.rejects( + assertPostgresRunManagerSchemaReady( + queryable({ + currentUser: 'ql3_run_manager', + privileges: runManagerPrivileges(), + functionMode: 'run-manager', + runManagerDispatchColumnPrivileges: widenedDispatchColumns, + }), + ), + (error) => + error instanceof PostgresSchemaReadinessError && + error.code === 'run_manager_role_invalid' && + error.facts.includes( + 'column-update-privilege:run_cancellation_dispatches.lease_token_digest', + ), + ); }); test('accepts isolated Package manager and executor roles', async () => { @@ -1132,10 +1180,10 @@ test('accepts the exact schema and isolated Worker ingress role', async () => { }), ); assert.equal(report.currentUser, 'ql3_worker_ingress'); - assert.equal(report.contractVersion, 65); + assert.equal(report.contractVersion, 66); assert.equal( report.migrationIds.at(-1), - 'pg-0066-cancellation-dispatch', + 'pg-0067-cancellation-dispatch-management', ); }); diff --git a/packages/ql3-cluster-postgres/test/runCancellationDispatchManagementRepository.test.cjs b/packages/ql3-cluster-postgres/test/runCancellationDispatchManagementRepository.test.cjs new file mode 100644 index 00000000..274268f4 --- /dev/null +++ b/packages/ql3-cluster-postgres/test/runCancellationDispatchManagementRepository.test.cjs @@ -0,0 +1,291 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { + InvalidRunCancellationDispatchManagementError, + PostgresRunCancellationDispatchManagementRepository, + RunCancellationDispatchManagementConflictError, +} = require('@qinglong/cluster-postgres/run-manager'); + +const NOW = 1_000_000; + +function command(overrides = {}) { + return { + projectId: 'project-1', + runId: 'run-1', + requestId: 'request-1', + auditEventId: '019f9600-0000-4000-8000-000000000001', + principal: { + subject: { type: 'user', id: 'operator-1' }, + authenticationId: 'oidc:run-management-1', + authenticatedAtMs: NOW - 1_000, + expiresAtMs: NOW + 60_000, + assurance: 'multi_factor', + }, + policyFence: { projectVersion: 2, bindingVersion: 3 }, + ...overrides, + }; +} + +function rearmCommand(overrides = {}) { + return command({ + requestId: 'request-rearm-1', + auditEventId: '019f9600-0000-4000-8000-000000000011', + mutationId: '019f9600-0000-4000-8000-000000000012', + eventId: '019f9600-0000-4000-8000-000000000013', + expectedDispatchVersion: 3, + expectedLastResult: 'identity_mismatch', + retryDelayMs: 5_000, + ...overrides, + }); +} + +function runRow() { + return { + projectId: 'project-1', + runStatus: 'running', + runVersion: 6, + eventSequence: 8, + cancelRequestedAtMs: NOW - 2_000, + cancelReason: 'user', + }; +} + +function dispatchRow(overrides = {}) { + return { + attemptId: 'attempt-1', + dispatchStatus: 'blocked', + dispatchVersion: 3, + dispatchCount: 1, + nextAttemptAtMs: null, + leaseExpiresAtMs: null, + lastResult: 'identity_mismatch', + lastDispatchedAtMs: NOW - 1_500, + dispatchCreatedAtMs: NOW - 1_900, + dispatchUpdatedAtMs: NOW - 1_500, + ...overrides, + }; +} + +function fixture(options = {}) { + const calls = []; + const client = { + async query(sql, params = []) { + const text = sql.replace(/\s+/g, ' ').trim(); + calls.push({ sql: text, params }); + if ( + text === 'BEGIN ISOLATION LEVEL SERIALIZABLE' || + text === 'COMMIT' || + text === 'ROLLBACK' || + text.startsWith('SELECT set_config') + ) { + return { rows: [], rowCount: 0 }; + } + if (text.includes('transaction_timestamp()')) { + return { rows: [{ nowMs: NOW }], rowCount: 1 }; + } + if (text.includes('lock_run_management_policy_fence')) { + return { + rows: [{ matches: options.authorized !== false }], + rowCount: 1, + }; + } + if (text.includes('FROM "ql3"."runs" WHERE id = $1 FOR UPDATE')) { + return { rows: [runRow()], rowCount: 1 }; + } + if (text.includes('FROM "ql3"."runs" WHERE id = $1')) { + return { rows: [runRow()], rowCount: 1 }; + } + if ( + text.includes('FROM "ql3"."run_events"') && + text.includes('dedupe_key = $2') + ) { + return { rows: options.replay ? [options.replay] : [], rowCount: 0 }; + } + if ( + text.startsWith('SELECT attempt_id AS "attemptId"') && + !text.includes('dispatchStatus') && + !text.includes('FOR UPDATE') + ) { + return { rows: [{ attemptId: 'attempt-1' }], rowCount: 1 }; + } + if (text.includes('FROM "ql3"."run_attempts"')) { + return { + rows: [{ attemptStatus: options.attemptStatus ?? 'running' }], + rowCount: 1, + }; + } + if ( + text.includes('FROM "ql3"."run_cancellation_dispatches"') && + text.includes('FOR UPDATE') + ) { + return { + rows: [ + dispatchRow({ + lastResult: options.lastResult ?? 'identity_mismatch', + }), + ], + rowCount: 1, + }; + } + if (text.includes('FROM "ql3"."run_cancellation_dispatches"')) { + return { rows: [dispatchRow()], rowCount: 1 }; + } + if (text.startsWith('UPDATE "ql3"."runs"')) { + return { rows: [], rowCount: 1 }; + } + if (text.startsWith('UPDATE "ql3"."run_cancellation_dispatches"')) { + return { rows: [], rowCount: 1 }; + } + if (text.startsWith('INSERT INTO "ql3"."run_events"')) { + return { rows: [], rowCount: 1 }; + } + if (text.startsWith('INSERT INTO "ql3"."security_audit_events"')) { + return { rows: [{ eventId: params[0] }], rowCount: 1 }; + } + throw new Error(`unexpected query: ${text}`); + }, + release() { + calls.push({ sql: 'RELEASE', params: [] }); + }, + }; + const pool = { async connect() { return client; } }; + return { + calls, + repository: new PostgresRunCancellationDispatchManagementRepository(pool), + }; +} + +test('inspects one low-sensitive blocked dispatch under run.read authority', async () => { + const { calls, repository } = fixture(); + const result = await repository.inspect(command()); + assert.equal(result.operatorAction, 'rearm'); + assert.equal(result.dispatch.status, 'blocked'); + assert.equal(result.dispatch.lastResult, 'identity_mismatch'); + const dispatchRead = calls.find(({ sql }) => + sql.includes('FROM "ql3"."run_cancellation_dispatches"'), + ); + assert.equal(dispatchRead.sql.includes('lease_owner'), false); + assert.equal(dispatchRead.sql.includes('lease_token_digest'), false); + assert.equal( + calls.some( + ({ sql }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"') && + sql.includes('$3'), + ), + true, + ); +}); + +test('rearms an exact blocked dispatch with one event and allowed audit', async () => { + const { calls, repository } = fixture(); + const result = await repository.rearm(rearmCommand()); + assert.deepEqual(result, { + status: 'rearmed', + projectId: 'project-1', + runId: 'run-1', + attemptId: 'attempt-1', + previousDispatchVersion: 3, + dispatchVersion: 4, + previousResult: 'identity_mismatch', + retryDelayMs: 5_000, + nextAttemptAtMs: NOW + 5_000, + runVersion: 7, + eventSequence: 9, + }); + const update = calls.find(({ sql }) => + sql.startsWith('UPDATE "ql3"."run_cancellation_dispatches"'), + ); + assert.deepEqual(update.params, [ + 'run-1', + 4, + NOW + 5_000, + NOW, + 'attempt-1', + 3, + 'identity_mismatch', + ]); + const attemptRead = calls.find(({ sql }) => + sql.includes('FROM "ql3"."run_attempts"'), + ); + assert.equal(attemptRead.sql.includes('FOR KEY SHARE'), false); + const event = calls.find(({ sql }) => + sql.startsWith('INSERT INTO "ql3"."run_events"'), + ); + assert.equal(event.params[0], rearmCommand().eventId); + assert.equal(JSON.parse(event.params[6]).previous_result, 'identity_mismatch'); + assert.ok( + calls.findIndex(({ sql }) => + sql.startsWith('UPDATE "ql3"."run_cancellation_dispatches"'), + ) < calls.findIndex(({ sql }) => sql === 'COMMIT'), + ); +}); + +test('exact mutation replay returns the immutable receipt without another update', async () => { + const replay = { + eventId: rearmCommand().eventId, + eventSequence: 9, + eventType: 'run.cancel_dispatch_rearmed', + actorType: 'user', + actorId: 'operator-1', + attemptId: 'attempt-1', + payload: { + schema: 'qinglong/run-cancellation-dispatch-rearm@v1', + mutation_id: rearmCommand().mutationId, + previous_dispatch_version: 3, + dispatch_version: 4, + previous_result: 'identity_mismatch', + retry_delay_ms: 5_000, + next_attempt_at_ms: NOW + 5_000, + run_version: 7, + }, + }; + const { calls, repository } = fixture({ replay }); + assert.equal((await repository.rearm(rearmCommand())).dispatchVersion, 4); + assert.equal(calls.some(({ sql }) => sql.startsWith('UPDATE')), false); +}); + +test('stale result and authorization changes fail closed before mutation', async () => { + const stale = fixture({ lastResult: 'pid_mismatch' }); + await assert.rejects( + stale.repository.rearm(rearmCommand()), + (error) => + error instanceof RunCancellationDispatchManagementConflictError && + error.reason === 'dispatch_result_changed', + ); + assert.equal( + stale.calls.some(({ sql }) => sql.startsWith('UPDATE')), + false, + ); + + const unauthorized = fixture({ authorized: false }); + await assert.rejects( + unauthorized.repository.inspect(command()), + (error) => + error instanceof RunCancellationDispatchManagementConflictError && + error.reason === 'authorization_changed', + ); + assert.equal( + unauthorized.calls.some(({ sql }) => + sql.includes('FROM "ql3"."runs"'), + ), + false, + ); +}); + +test('rejects malformed management authority before opening PostgreSQL', async () => { + let opened = false; + const repository = new PostgresRunCancellationDispatchManagementRepository({ + async connect() { + opened = true; + throw new Error('must not open'); + }, + }); + assert.throws( + () => repository.rearm(rearmCommand({ retryDelayMs: 0 })), + InvalidRunCancellationDispatchManagementError, + ); + assert.equal(opened, false); +}); diff --git a/packages/ql3-runtime-core/src/run/cancellation-dispatch/cancellationDispatch.ts b/packages/ql3-runtime-core/src/run/cancellation-dispatch/cancellationDispatch.ts index 62168da1..b82828d2 100644 --- a/packages/ql3-runtime-core/src/run/cancellation-dispatch/cancellationDispatch.ts +++ b/packages/ql3-runtime-core/src/run/cancellation-dispatch/cancellationDispatch.ts @@ -40,6 +40,13 @@ export const CANCELLATION_DISPATCH_BLOCKING_RESULTS = Object.freeze([ 'invalid', ] as const satisfies readonly CancellationDispatchResult[]); +const CANCELLATION_DISPATCH_RETRY_HISTORY_RESULTS = new Set< + CancellationDispatchResult +>([ + ...CANCELLATION_DISPATCH_RETRYABLE_RESULTS, + ...CANCELLATION_DISPATCH_BLOCKING_RESULTS, +]); + export const MAX_CANCELLATION_DISPATCH_LEASE_MS = 5 * 60_000; export const MAX_CANCELLATION_DISPATCH_RETRY_DELAY_MS = 24 * 60 * 60_000; @@ -421,13 +428,10 @@ export function normalizeCancellationDispatchRecord( version < dispatchCount || (status === 'leased' && lastResult !== undefined && - !CANCELLATION_DISPATCH_RETRYABLE_RESULTS.includes( - lastResult as (typeof CANCELLATION_DISPATCH_RETRYABLE_RESULTS)[number], - )) || + !CANCELLATION_DISPATCH_RETRY_HISTORY_RESULTS.has(lastResult)) || (status === 'retry_wait' && - !CANCELLATION_DISPATCH_RETRYABLE_RESULTS.includes( - lastResult as (typeof CANCELLATION_DISPATCH_RETRYABLE_RESULTS)[number], - )) || + (lastResult === undefined || + !CANCELLATION_DISPATCH_RETRY_HISTORY_RESULTS.has(lastResult))) || (status === 'dispatched' && lastResult !== 'termination_requested' && lastResult !== 'already_exited') || diff --git a/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs b/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs index 0f41f9b4..850b3204 100644 --- a/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs +++ b/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs @@ -4,6 +4,9 @@ const { createPostgresDatabaseOpener, PostgresCancellationDispatchRepository, } = require('../packages/ql3-cluster-postgres/dist/entrypoints/runtime.js'); +const { + PostgresRunCancellationDispatchManagementRepository, +} = require('../packages/ql3-cluster-postgres/dist/entrypoints/runManager.js'); const { CancellationDispatchFenceRejectedError, digestCancellationDispatchLeaseToken, @@ -13,12 +16,19 @@ const { } = require('../packages/ql3-cluster-control/dist/remote-execution/remoteWorkerCancellationDispatchControl.js'); const FIXTURE = Object.freeze({ + projectId: 'ha-cancel-project-d365', + actorId: 'ha-cancel-operator-d365', runId: 'ha-cancel-run-d363', attemptId: 'ha-cancel-attempt-d363', requestedAtMs: 1_750_000_000_100, retryEventId: 'ha-cancel-retry-event-d363', terminalEventId: 'ha-cancel-terminal-event-d363', settledEventId: 'ha-cancel-settled-event-d363', + blockedEventId: 'ha-cancel-blocked-event-d365', + inspectAuditEventId: '019f9700-0000-4000-8000-000000000001', + rearmAuditEventId: '019f9700-0000-4000-8000-000000000002', + rearmMutationId: '019f9700-0000-4000-8000-000000000003', + rearmEventId: '019f9700-0000-4000-8000-000000000004', }); async function openRuntime(connectionString, applicationName) { @@ -32,6 +42,17 @@ async function openRuntime(connectionString, applicationName) { })(); } +async function openRunManager(connectionString, applicationName) { + return createPostgresDatabaseOpener({ + role: 'run-manager', + connection: { connectionString, tls: { mode: 'disable' } }, + pool: { maxConnections: 1, applicationName }, + onPoolError(error) { + throw error; + }, + })(); +} + async function cancellationDispatchFacts(pool) { const result = await pool.query( `SELECT dispatch.status, dispatch.version, @@ -53,21 +74,45 @@ async function cancellationDispatchFacts(pool) { } async function persistCancellationDispatchHaFixture(options) { - const { migrationPool, runtimeConnectionString } = options; + const { + migrationPool, + runtimeConnectionString, + runManagerConnectionString, + } = options; const beforeClock = await migrationPool.query( `SELECT floor(extract(epoch FROM clock_timestamp()) * 1000)::bigint AS "nowMs"`, ); + const observedAtMs = Number(beforeClock.rows[0].nowMs); + await migrationPool.query( + `INSERT INTO "ql3"."projects" ( + id, name, slug, status, version, created_at_ms, updated_at_ms + ) VALUES ($1, 'HA cancellation dispatch', $1, 'active', 1, $2, $2)`, + [FIXTURE.projectId, observedAtMs], + ); + await migrationPool.query( + `INSERT INTO "ql3"."project_role_bindings" ( + project_id, subject_type, subject_id, version, state, role, + mutation_id, changed_by_type, changed_by_id, created_at_ms + ) VALUES ($1, 'user', $2, 1, 'active', 'operator', + 'ha-cancel-binding-d365', 'system', 'ha-contract', $3)`, + [FIXTURE.projectId, FIXTURE.actorId, observedAtMs], + ); await migrationPool.query( `INSERT INTO "ql3"."runs" ( id, project_id, task_id, task_revision, trigger_type, execution_origin, execution_owner, status, version, event_sequence, created_at_ms, started_at_ms, cancel_requested_at_ms, cancel_reason ) VALUES ( - $1, 'default', 'ha-cancel-task', 'v1', 'manual', 'api', 'runtime', - 'running', 2, 0, $2, $2, $3, 'user' + $1, $2, 'ha-cancel-task', 'v1', 'manual', 'api', 'runtime', + 'running', 2, 0, $3, $3, $4, 'user' )`, - [FIXTURE.runId, FIXTURE.requestedAtMs - 100, FIXTURE.requestedAtMs], + [ + FIXTURE.runId, + FIXTURE.projectId, + FIXTURE.requestedAtMs - 100, + FIXTURE.requestedAtMs, + ], ); await migrationPool.query( `INSERT INTO "ql3"."run_attempts" ( @@ -174,9 +219,90 @@ async function persistCancellationDispatchHaFixture(options) { SET next_attempt_at_ms = 0 WHERE run_id = $1`, [FIXTURE.runId], ); + const blockingClaim = await second.claim({ + ...candidate, + owner: 'ha-cancel-blocker', + leaseToken: 'ha-cancel-blocker-token', + }); + assert.equal(blockingClaim.status, 'claimed'); + assert.equal(blockingClaim.dispatch.version, 4); + const blocked = await second.recordResult({ + runId: FIXTURE.runId, + attemptId: FIXTURE.attemptId, + owner: 'ha-cancel-blocker', + leaseToken: 'ha-cancel-blocker-token', + expectedVersion: blockingClaim.dispatch.version, + result: 'identity_mismatch', + eventId: FIXTURE.blockedEventId, + }); + assert.equal(blocked.dispatch.status, 'blocked'); + assert.equal(blocked.dispatch.version, 5); + + const runManagerDatabase = await openRunManager( + runManagerConnectionString, + 'ql3-ha-cancel-run-manager', + ); + try { + const management = + new PostgresRunCancellationDispatchManagementRepository( + runManagerDatabase.pool, + ); + const principal = Object.freeze({ + subject: Object.freeze({ type: 'user', id: FIXTURE.actorId }), + authenticationId: 'oidc:ha-cancel-d365', + authenticatedAtMs: observedAtMs - 1_000, + expiresAtMs: observedAtMs + 5 * 60_000, + assurance: 'multi_factor', + }); + const authority = Object.freeze({ + projectId: FIXTURE.projectId, + runId: FIXTURE.runId, + requestId: 'ha-cancel-inspect-d365', + auditEventId: FIXTURE.inspectAuditEventId, + principal, + policyFence: Object.freeze({ projectVersion: 1, bindingVersion: 1 }), + }); + const diagnostic = await management.inspect(authority); + assert.equal(diagnostic.operatorAction, 'rearm'); + assert.equal(diagnostic.dispatch?.status, 'blocked'); + assert.equal(diagnostic.dispatch?.version, 5); + assert.equal(diagnostic.dispatch?.lastResult, 'identity_mismatch'); + assert.equal(JSON.stringify(diagnostic).includes('leaseOwner'), false); + assert.equal(JSON.stringify(diagnostic).includes('leaseToken'), false); + const rearm = await management.rearm({ + ...authority, + requestId: 'ha-cancel-rearm-d365', + auditEventId: FIXTURE.rearmAuditEventId, + mutationId: FIXTURE.rearmMutationId, + eventId: FIXTURE.rearmEventId, + expectedDispatchVersion: 5, + expectedLastResult: 'identity_mismatch', + retryDelayMs: 60_000, + }); + assert.equal(rearm.status, 'rearmed'); + assert.equal(rearm.dispatchVersion, 6); + assert.equal(rearm.previousResult, 'identity_mismatch'); + assert.equal(rearm.runVersion, 5); + assert.equal(rearm.eventSequence, 3); + } finally { + await runManagerDatabase.close(); + } + assert.equal( + (await first.claim({ + ...candidate, + owner: 'ha-cancel-rearm-early', + leaseToken: 'ha-cancel-rearm-early-token', + })).status, + 'not_due', + ); + await migrationPool.query( + `UPDATE "ql3"."run_cancellation_dispatches" + SET next_attempt_at_ms = 0 WHERE run_id = $1`, + [FIXTURE.runId], + ); const stopRequested = Object.freeze({ status: 'stop_requested', - projectId: 'default', + projectId: FIXTURE.projectId, runId: FIXTURE.runId, attemptId: FIXTURE.attemptId, offerId: 'ha-cancel-offer-d364', @@ -236,13 +362,13 @@ async function persistCancellationDispatchHaFixture(options) { const beforePromotion = await cancellationDispatchFacts(migrationPool); assert.deepEqual(beforePromotion, { status: 'dispatched', - version: 5, - dispatchCount: 3, + version: 8, + dispatchCount: 4, leaseTokenDigest: null, lastResult: 'termination_requested', - runVersion: 5, - eventSequence: 3, - eventCount: 3, + runVersion: 7, + eventSequence: 5, + eventCount: 5, }); return { fixture: FIXTURE, @@ -253,6 +379,9 @@ async function persistCancellationDispatchHaFixture(options) { expiredLeaseTakenOver: true, staleLeaseFenced: true, retryDeferredUntilDue: true, + operatorDiagnosticLowSensitive: true, + manualBlockedRearmExact: true, + manualRearmDeferredUntilDue: true, productionDeliverySettledBeforeStop: true, replicatedBeforePromotion: false, survivedPromotion: false, @@ -278,7 +407,7 @@ async function verifyPromotedCancellationDispatchHaFixture(options) { runtimeDatabase.pool, ).findByRunId(FIXTURE.runId); assert.equal(dispatch?.status, 'dispatched'); - assert.equal(dispatch?.dispatchCount, 3); + assert.equal(dispatch?.dispatchCount, 4); assert.equal(dispatch?.leaseTokenDigest, undefined); } finally { await runtimeDatabase.close(); diff --git a/scripts/ql3-postgres-ha-contract.cjs b/scripts/ql3-postgres-ha-contract.cjs index 24da821c..c2c7d4c6 100644 --- a/scripts/ql3-postgres-ha-contract.cjs +++ b/scripts/ql3-postgres-ha-contract.cjs @@ -11785,6 +11785,11 @@ async function main(argv = process.argv.slice(2)) { RUNTIME_PASSWORD, primaryPort, ), + runManagerConnectionString: databaseUrl( + RUN_MANAGER_USER, + RUN_MANAGER_PASSWORD, + primaryPort, + ), }); timeline.push({ state: 'cancellation_dispatch_fenced_on_primary', diff --git a/test/back/ql3PackageBoundaryAudit.test.cjs b/test/back/ql3PackageBoundaryAudit.test.cjs index 6feca1ea..b8050e46 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: 170, + sourceFiles: 172, rootSourceFiles: 1, rootSourceLines: 126, - nestedSourceFiles: 169, + nestedSourceFiles: 171, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, );