From 5a979e510e86cee8687603c38cad421eef503dec Mon Sep 17 00:00:00 2001 From: whyour Date: Wed, 19 Aug 2026 08:29:45 +0800 Subject: [PATCH] feat(ql3): expose cancellation availability summary --- docs/QINGLONG_3_0_ARCHITECTURE_RFC.md | 13 +- .../ADR-0005-durable-cancellation-dispatch.md | 7 +- ...lege-cancellation-diagnostics-and-rearm.md | 2 +- ...coped-cancellation-availability-summary.md | 61 +++++ .../src/run-management/runManagement.ts | 113 +++++++++ .../src/run-management/runManagementClient.ts | 113 +++++++++ .../run-management/runManagementTransport.ts | 78 ++++++- .../test/runManagement.test.cjs | 55 +++++ .../test/runManagementClient.test.cjs | 75 ++++++ .../test/runManagementTransport.test.cjs | 86 +++++++ .../src/entrypoints/runManager.ts | 2 + ...ancellationDispatchManagementRepository.ts | 220 +++++++++++++++++- ...ationDispatchManagementRepository.test.cjs | 117 ++++++++++ ...tgres-ha-cancellation-dispatch-fixture.cjs | 18 ++ scripts/ql3-postgres-ha-contract.cjs | 2 + 15 files changed, 952 insertions(+), 10 deletions(-) create mode 100644 docs/adr/ADR-0459-project-scoped-cancellation-availability-summary.md diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index c036c14f..d58b8690 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -11,7 +11,18 @@ 最新增量证据（2026-08-19）： -- D-365/ADR-0458（已接受；聚合指标与产品 UI 待完成）：在既有隔离 `cluster-admin` Run management plane 上增加 +- D-366/ADR-0459（已接受；产品视觉入口待完成）：在既有 Run management plane 增加 `run.cancellation.summary`，由强认证 User 以 + `run.read` 按需读取 Project 级 PostgreSQL 快照。响应只有五态 dispatch 计数、due/expired-lease 信号、四种 blocking-result 计数、最早 blocked + 时间和 `clear|converging|attention_required`/`none|wait|inspect` 固定结论，不返回 Run/Attempt/Worker identity 或 lease capability。blocked 触发 + `attention_required`，但不错误撤回整个 Cluster readiness；due/expired 只作为 caller-driven 收敛信号。查询与 allowed audit 位于同一 5 秒 + SERIALIZABLE 短事务，复用 v66 Run manager SELECT、既有单连接池和通用 mTLS 客户端，不新增 migration、权限、package、依赖、服务、端口、timer、 + queue、cache 或 Kubernetes 对象。Edge/Standalone 闭包不变化；代码继续位于已有 `run-management` 子域，没有新增微包或 `src` 根目录平铺。完整 + backend `1,487 pass / 0 fail / 2 conditional skip`，18-package clean build/test、四项架构审计与 `14/14` Local artifact audit 通过；基础 + Edge/Standalone 保持 `2,589,998 / 2,590,076` bytes，Application+AI 保持 `4,493,151 / 4,493,283` bytes。PostgreSQL 18.6 arm64 HA + `145/145` 执行 blocked summary→inspect→rearm→production delivery→WAL→promotion，timeline `1→2`，报告 SHA-256 为 + `d763157b3a781e305add3c6f0c5080820b1d65b5feefe60be6fa7006c0050107`。 + +- D-365/ADR-0458（已接受；聚合出口由 D-366 完成，产品视觉入口待完成）：在既有隔离 `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 diff --git a/docs/adr/ADR-0005-durable-cancellation-dispatch.md b/docs/adr/ADR-0005-durable-cancellation-dispatch.md index ab4174e0..678d0898 100644 --- a/docs/adr/ADR-0005-durable-cancellation-dispatch.md +++ b/docs/adr/ADR-0005-durable-cancellation-dispatch.md @@ -153,11 +153,13 @@ PID 可复用，可能终止无关进程，禁止。 `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。 +ADR-0459 进一步增加 Project-scoped、caller-driven summary，以固定计数和三态 assessment 提供 blocked/availability 告警出口，不改变全局 readiness，也不新增采集器。 + HTTP worker 已通过默认关闭的 manual-only manifest bootstrap 接入 Local Supervisor：只有 accepted 且全部 gate 通过时才启动，失败或 shutdown 时有界停止。以下工作仍未完成，因此它仍只允许显式 canary，不得扩大到默认生产流量： -- 用户可见的聚合运行指标、告警与产品 UI；私有 operator 诊断和处置协议已经实现。 +- 产品控制台中的 Project 状态卡、告警路由和有界 blocked drill-down；Project 聚合出口与私有 operator 处置协议已经实现。 - 固定 edge 设备的数据库写放大、RSS、时延和磁盘基准。 -- cluster-control 对 PostgreSQL CancellationDispatch 的聚合 availability/blocked 指标与告警出口。 +- 固定 Project allowlist 的外部时序指标适配；数据库事实驱动的按需 availability/blocked 汇总已经实现。 - 首次真实目标实例完整激活/回滚仪式与共享 config 多写者 authority。 ## 6. 验证门禁 @@ -177,3 +179,4 @@ HTTP worker 已通过默认关闭的 manual-only manifest bootstrap 接入 Local 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 均失败关闭。 +16. Project summary 的五态与 blocking-result 计数必须交叉守恒，blocked 只产生 `attention_required` 告警而不撤回全局 readiness，且响应不包含 Run/Attempt/lease identity。 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 index 5139df91..ce144d97 100644 --- a/docs/adr/ADR-0458-least-privilege-cancellation-diagnostics-and-rearm.md +++ b/docs/adr/ADR-0458-least-privilege-cancellation-diagnostics-and-rearm.md @@ -62,4 +62,4 @@ QingLong 同时面向低配路由器与集群。该能力只属于 Cluster opera ## 后续 -私有 operator 协议已经闭合，但面向产品的聚合 blocked/availability 指标、告警和 UI 仍待实现。CloudNativePG live failover、多副本容量压力、固定 Linux x64/arm64 与物理 Edge 资源证据继续作为发布最终化门，不由单机 Docker HA 结果替代。 +ADR-0459 已用既有管理面完成数据库事实驱动的 Project 级 blocked/availability 汇总和固定告警结论；产品视觉入口、告警路由和有界 drill-down 仍待实现。CloudNativePG live failover、多副本容量压力、固定 Linux x64/arm64 与物理 Edge 资源证据继续作为发布最终化门，不由单机 Docker HA 结果替代。 diff --git a/docs/adr/ADR-0459-project-scoped-cancellation-availability-summary.md b/docs/adr/ADR-0459-project-scoped-cancellation-availability-summary.md new file mode 100644 index 00000000..9a2fb772 --- /dev/null +++ b/docs/adr/ADR-0459-project-scoped-cancellation-availability-summary.md @@ -0,0 +1,61 @@ +# ADR-0459：按需 Project CancellationDispatch 可用性汇总 + +- 状态：Accepted +- 日期：2026-08-19 +- 关联 RFC：QL-RFC-0001 D-366、PR-5、PR-7 +- 关联 ADR：ADR-0005、ADR-0456、ADR-0457、ADR-0458 +- Amends：ADR-0005 的 blocked 指标出口、ADR-0458 的产品可见运维边界 + +## 上下文 + +ADR-0458 已提供单 Run 的低敏 inspect 和人工 rearm，但 operator 仍必须预先知道哪个 Run 被 blocked。直接依赖每个 `cluster-control` 副本的进程内计数会在重启、扩缩容和 failover 后丢失事实，也无法形成 Project 一致视图。新增 Prometheus server、扫描 timer 或聚合 daemon 又会制造第二个 cancellation cadence，并让低配部署为空闲可观测性持续付费。 + +QingLong 同时面对小型路由设备和多副本 Cluster。Local/Edge 不应引入 PostgreSQL、管理面或指标闭包；Cluster 则需要数据库事实驱动、可审计、可供 CLI/UI/告警共同消费的稳定出口。 + +## 决策 + +1. 在既有 `cluster-admin` Run management 协议增加 `run.cancellation.summary`，复用原 HTTPS/mTLS process、OIDC、固定路由、请求上限、连接池和 shutdown lifecycle，不新增 package、服务、端口、timer、队列、缓存或 Kubernetes 对象。 +2. 请求只包含 Project、request/audit identity 和固定空 body schema，不接受 Run、Attempt、时间窗口、分页、状态过滤、指标标签或调用方时钟。transport 只注入强认证 User；service 使用既有 `run.read`，因此 viewer 可读取汇总，弱认证、agent 和 Policy fence drift 失败关闭。 +3. Repository 在一个最长 5 秒的 SERIALIZABLE 短事务中，以 PostgreSQL `transaction_timestamp()` 观察 Project 当前快照，并在提交前原子写 allowed audit。它只使用 v66 已授予 Run manager 的 `runs` 与 `run_cancellation_dispatches` SELECT，不增加 migration 或数据库权限。 +4. 响应只包含固定低基数投影：五个状态计数、due 与 expired-lease 两个信号、四个 blocking-result 计数、最早 blocked 时间，以及 `clear|converging|attention_required` assessment 和 `none|wait|inspect` 建议。永不返回 Run/Attempt/Worker identity、lease owner/token/digest、PID、handle、命令、环境、Secret、日志或错误原文。 +5. assessment 由服务端确定：存在 blocked 即 `attention_required/inspect`；没有 blocked 但仍有 pending/leased/retry_wait 即 `converging/wait`；否则为 `clear/none`。due 或 lease expiry 是 caller-driven 交付可接管信号，不单独把整个服务 readiness 标记为不可用。 +6. `total` 必须等于五态之和，blocked 必须等于四种 blocking result 之和，due 不得超过 pending+retry_wait，expired lease 不得超过 leased，最早 blocked 时间必须与 blocked 是否存在一致且不晚于数据库观察时间。Repository 和客户端分别做 exact-shape 与交叉不变量校验。 +7. Project ID 只用于请求作用域和响应绑定，不被设计为常驻时序指标标签。CLI、产品 UI 或告警适配器可以按需请求该固定快照；若未来导出 Prometheus，只能由部署侧选择 Project allowlist，不能让本协议隐式制造无界 label cardinality。 +8. 代码继续内聚在现有 `cluster-postgres/run-management` 与 `cluster-admin/run-management` 子域。聚合、inspect、rearm 共享同一个 management repository 与 authority，不为一个查询再拆微包，也不把文件铺到 package 的 `src` 根目录。 + +## 被拒绝的替代方案 + +### 每个 cluster-control 副本维护内存指标 + +拒绝。副本重启会清零，多副本结果无法精确合并，blocked durable state 与进程 counter 可能永久分叉。 + +### 新建扫描器或 Prometheus 服务 + +拒绝。扫描 cadence 会与 caller-driven Worker delivery 竞争，并增加空闲 CPU、连接、部署对象和低配运维成本。 + +### 把 blocked 直接映射到 `/readyz` 失败 + +拒绝。blocked 是一个 Run 的身份或协议处置事实，不代表数据库、管理面或其他 Project 不可服务。全局 readiness 失败会触发无效滚动并掩盖真正处置目标。 + +### 返回 blocked Run 列表 + +拒绝。列表需要分页、稳定 cursor、更多 identity 泄漏与更宽 UI 状态。本阶段先提供固定聚合；operator 从 `inspect` 精确诊断已知 Run，后续产品 drill-down 必须另行设计有界索引契约。 + +## 资源、安全与部署影响 + +- Edge/Standalone 不依赖 `cluster-admin`、`cluster-postgres` 或 `pg`，默认与启用态均为零新增常驻开销。 +- Cluster 只在 operator/产品调用时使用既有单连接 Run manager pool；没有后台采集、写放大或 idle I/O。 +- Project 查询可利用既有 `runs(project_id, created_at_ms, id)` 与 dispatch `run_id` 索引，并受 5 秒 statement timeout、1 秒 lock timeout 和事务 idle timeout 约束；超限失败关闭，不降级为部分或伪造的健康结果。 +- summary 只有 allowed/denied audit 写入，不修改 Run、Attempt、dispatch 或 RunEvent，不改变 `cluster-control` readiness。 + +## 验证 + +- Repository、service、transport 和客户端聚焦测试覆盖三态 assessment、计数交叉不变量、viewer `run.read`、强认证、原子 audit、Project 绑定、未知字段与 capability 泄漏拒绝。 +- `cluster-postgres` 为 `341 pass / 0 fail / 3 conditional skip`，`cluster-admin` 为 `397 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 包，`singleSourcePackages=[]`、`shallowSourcePackages=[]`。 +- `14/14` Local artifact audit 通过且字节与 D-365 相同：基础 Edge/Standalone `2,589,998 / 2,590,076`，Application+AI `4,493,151 / 4,493,283`。 +- PostgreSQL 18.6 arm64 HA `145/145`：真实 `ql3_run_manager` 在 blocked 状态读取 Project summary，再继续 inspect、CAS rearm、production delivery、WAL replication 与 promotion；timeline `1→2`，报告 SHA-256 `d763157b3a781e305add3c6f0c5080820b1d65b5feefe60be6fa7006c0050107`，独立证据审计零 finding。 + +## 后续 + +产品控制台可把该 summary 渲染为 Project 级状态卡和告警入口，并通过单 Run inspect/rearm 完成 drill-down。CloudNativePG live failover、多副本容量压力、固定 Linux x64/arm64 和物理 Edge 资源证据仍是独立发布门；本 ADR 不把 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 24703b4b..0ed00efd 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagement.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagement.ts @@ -13,6 +13,7 @@ import { type BlockingCancellationDispatchResult, type RunCancellationDispatchDiagnostic, type RunCancellationDispatchRearmReceipt, + type RunCancellationDispatchSummary, } from '@qinglong/cluster-postgres/run-manager'; import type { PostgresPool } from '@qinglong/runtime-core'; import { CANCELLATION_DISPATCH_BLOCKING_RESULTS } from '@qinglong/runtime-core/cancellation-dispatch'; @@ -75,6 +76,14 @@ export interface ClusterRunManagementCancellationInspectRequest { readonly principal: Readonly; } +export interface ClusterRunManagementCancellationSummaryRequest { + readonly projectId: string; + readonly requestId: string; + readonly auditEventId: string; + readonly failureAuditEventId: string; + readonly principal: Readonly; +} + export interface ClusterRunManagementCancellationRearmRequest extends ClusterRunManagementCancellationInspectRequest { readonly mutationId: string; @@ -90,6 +99,9 @@ export interface ClusterRunManagementService { stop( request: Readonly, ): Promise>; + summarizeCancellation( + request: Readonly, + ): Promise>; inspectCancellation( request: Readonly, ): Promise>; @@ -233,6 +245,28 @@ function exactCancellationInspectRequest( } } +function exactCancellationSummaryRequest( + 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', + ] + .sort() + .join('\0') + ) { + throw new ClusterRunManagementRequestError(); + } +} + function exactCancellationRearmRequest( value: unknown, ): asserts value is Readonly { @@ -502,6 +536,85 @@ export function createClusterRunManagementService( throw new ClusterRunManagementUnavailableError({ cause: error }); } }, + async summarizeCancellation( + requestValue: Readonly, + ) { + exactCancellationSummaryRequest(requestValue); + const observedAtMs = now(); + let principal: Readonly; + if ( + !Number.isSafeInteger(observedAtMs) || + observedAtMs < 0 || + !IDENTIFIER_PATTERN.test(requestValue.projectId) || + !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.summary({ + projectId: requestValue.projectId, + 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.summary', + 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 RunCancellationDispatchManagementConflictError) { + throw new ClusterRunManagementConflictError(); + } + if (error instanceof RunCancellationDispatchManagementUnavailableError) { + throw new ClusterRunManagementUnavailableError({ cause: error }); + } + throw new ClusterRunManagementUnavailableError({ cause: error }); + } + }, async inspectCancellation( requestValue: Readonly, ) { diff --git a/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts b/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts index 556d2c1d..6870b4f6 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagementClient.ts @@ -21,6 +21,7 @@ import { import { RUN_CANCELLATION_DISPATCH_DIAGNOSTIC_SCHEMA, RUN_CANCELLATION_DISPATCH_REARM_RECEIPT_SCHEMA, + RUN_CANCELLATION_DISPATCH_SUMMARY_SCHEMA, normalizeClusterRunManagementCommand, type ClusterRunManagementCommand, type ClusterRunManagementTransportResult, @@ -115,6 +116,118 @@ export function validateClusterRunManagementClientResult( envelope as unknown as ClusterRunManagementTransportResult, ); } + if (command.operation === 'run.cancellation.summary') { + const envelope = exact(value, ['schemaVersion', 'operation', 'summary']); + if ( + envelope.schemaVersion !== 1 || + envelope.operation !== command.operation + ) { + invalid(); + } + const summary = exact(envelope.summary, [ + 'schema', + 'projectId', + 'observedAtMs', + 'assessment', + 'operatorAction', + 'dispatches', + 'signals', + 'blockingResults', + ...(Object.hasOwn(envelope.summary as object, 'oldestBlockedAtMs') + ? ['oldestBlockedAtMs'] + : []), + ]); + const dispatches = exact(summary.dispatches, [ + 'total', + 'pending', + 'leased', + 'retryWait', + 'dispatched', + 'blocked', + ]); + const signals = exact(summary.signals, ['due', 'expiredLease']); + const blockingResults = exact(summary.blockingResults, [ + 'identityMismatch', + 'pidMismatch', + 'unsupported', + 'invalid', + ]); + const dispatchCounts = [ + dispatches.total, + dispatches.pending, + dispatches.leased, + dispatches.retryWait, + dispatches.dispatched, + dispatches.blocked, + ]; + const blockingCounts = [ + blockingResults.identityMismatch, + blockingResults.pidMismatch, + blockingResults.unsupported, + blockingResults.invalid, + ]; + if ( + summary.schema !== RUN_CANCELLATION_DISPATCH_SUMMARY_SCHEMA || + summary.projectId !== command.request.projectId || + !safeInteger(summary.observedAtMs) || + !['clear', 'converging', 'attention_required'].includes( + summary.assessment as string, + ) || + !['none', 'wait', 'inspect'].includes(summary.operatorAction as string) || + dispatchCounts.some((count) => !safeInteger(count)) || + !safeInteger(signals.due) || + !safeInteger(signals.expiredLease) || + blockingCounts.some((count) => !safeInteger(count)) || + dispatches.total !== + (dispatches.pending as number) + + (dispatches.leased as number) + + (dispatches.retryWait as number) + + (dispatches.dispatched as number) + + (dispatches.blocked as number) || + dispatches.blocked !== + (blockingResults.identityMismatch as number) + + (blockingResults.pidMismatch as number) + + (blockingResults.unsupported as number) + + (blockingResults.invalid as number) || + (signals.due as number) > + (dispatches.pending as number) + (dispatches.retryWait as number) || + (signals.expiredLease as number) > (dispatches.leased as number) || + (Object.hasOwn(summary, 'oldestBlockedAtMs') && + (!safeInteger(summary.oldestBlockedAtMs) || + (summary.oldestBlockedAtMs as number) > + (summary.observedAtMs as number))) || + ((dispatches.blocked as number) === 0) !== + !Object.hasOwn(summary, 'oldestBlockedAtMs') + ) { + invalid(); + } + const active = + (dispatches.pending as number) + + (dispatches.leased as number) + + (dispatches.retryWait as number) + + (dispatches.blocked as number); + const expectedAssessment = + (dispatches.blocked as number) > 0 + ? 'attention_required' + : active > 0 + ? 'converging' + : 'clear'; + const expectedOperatorAction = + (dispatches.blocked as number) > 0 + ? 'inspect' + : active > 0 + ? 'wait' + : 'none'; + if ( + summary.assessment !== expectedAssessment || + summary.operatorAction !== expectedOperatorAction + ) { + invalid(); + } + return Object.freeze( + envelope as unknown as ClusterRunManagementTransportResult, + ); + } if (command.operation === 'run.cancellation.inspect') { const envelope = exact(value, [ 'schemaVersion', diff --git a/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts b/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts index a0ac9b63..2aac3807 100644 --- a/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts +++ b/packages/ql3-cluster-admin/src/run-management/runManagementTransport.ts @@ -22,6 +22,10 @@ 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_SUMMARY_REQUEST_SCHEMA = + 'qinglong/run-cancellation-dispatch-summary-request@v1'; +export const RUN_CANCELLATION_DISPATCH_SUMMARY_SCHEMA = + 'qinglong/run-cancellation-dispatch-summary@v1'; export const RUN_CANCELLATION_DISPATCH_DIAGNOSTIC_SCHEMA = 'qinglong/run-cancellation-dispatch-diagnostic@v1'; export const RUN_CANCELLATION_DISPATCH_REARM_REQUEST_SCHEMA = @@ -78,6 +82,20 @@ export type ClusterRunManagementCancellationInspectCommand = Readonly<{ }>; }>; +export type ClusterRunManagementCancellationSummaryCommand = Readonly<{ + schemaVersion: 1; + operation: 'run.cancellation.summary'; + request: Readonly<{ + projectId: string; + requestId: string; + auditEventId: string; + failureAuditEventId: string; + body: Readonly<{ + schema: typeof RUN_CANCELLATION_DISPATCH_SUMMARY_REQUEST_SCHEMA; + }>; + }>; +}>; + export type ClusterRunManagementCancellationRearmCommand = Readonly<{ schemaVersion: 1; operation: 'run.cancellation.rearm'; @@ -104,6 +122,7 @@ export type ClusterRunManagementCancellationRearmCommand = Readonly<{ export type ClusterRunManagementCommand = | ClusterRunManagementRetryCommand | ClusterRunManagementStopCommand + | ClusterRunManagementCancellationSummaryCommand | ClusterRunManagementCancellationInspectCommand | ClusterRunManagementCancellationRearmCommand; @@ -129,6 +148,16 @@ export type ClusterRunManagementCancellationInspectTransportResult = Readonly<{ >; }>; +export type ClusterRunManagementCancellationSummaryTransportResult = Readonly<{ + schemaVersion: 1; + operation: 'run.cancellation.summary'; + summary: Readonly< + Awaited> & { + schema: typeof RUN_CANCELLATION_DISPATCH_SUMMARY_SCHEMA; + } + >; +}>; + export type ClusterRunManagementCancellationRearmTransportResult = Readonly<{ schemaVersion: 1; operation: 'run.cancellation.rearm'; @@ -142,6 +171,7 @@ export type ClusterRunManagementCancellationRearmTransportResult = Readonly<{ export type ClusterRunManagementTransportResult = | ClusterRunManagementRetryTransportResult | ClusterRunManagementStopTransportResult + | ClusterRunManagementCancellationSummaryTransportResult | ClusterRunManagementCancellationInspectTransportResult | ClusterRunManagementCancellationRearmTransportResult; @@ -227,6 +257,7 @@ export function normalizeClusterRunManagementCommand( if ( operation !== 'run.retry' && operation !== 'run.stop' && + operation !== 'run.cancellation.summary' && operation !== 'run.cancellation.inspect' && operation !== 'run.cancellation.rearm' ) { @@ -243,7 +274,15 @@ export function normalizeClusterRunManagementCommand( 'failureAuditEventId', 'body', ] - : [ + : operation === 'run.cancellation.summary' + ? [ + 'projectId', + 'requestId', + 'auditEventId', + 'failureAuditEventId', + 'body', + ] + : [ 'projectId', 'runId', 'requestId', @@ -275,6 +314,25 @@ export function normalizeClusterRunManagementCommand( }), }); } + if (operation === 'run.cancellation.summary') { + const body = exact(request.body, ['schema']); + if (body.schema !== RUN_CANCELLATION_DISPATCH_SUMMARY_REQUEST_SCHEMA) { + invalid(); + } + return Object.freeze({ + schemaVersion: 1, + operation, + request: Object.freeze({ + projectId: identifier(request.projectId), + requestId: identifier(request.requestId), + auditEventId, + failureAuditEventId, + body: Object.freeze({ + schema: RUN_CANCELLATION_DISPATCH_SUMMARY_REQUEST_SCHEMA, + }), + }), + }); + } if (operation === 'run.cancellation.inspect') { const body = exact(request.body, ['schema']); if (body.schema !== RUN_CANCELLATION_DISPATCH_INSPECT_REQUEST_SCHEMA) { @@ -374,6 +432,7 @@ export function createClusterRunManagementTransport( !options.service || typeof options.service.retry !== 'function' || typeof options.service.stop !== 'function' || + typeof options.service.summarizeCancellation !== 'function' || typeof options.service.inspectCancellation !== 'function' || typeof options.service.rearmCancellation !== 'function' || (options.now !== undefined && typeof options.now !== 'function') @@ -435,6 +494,23 @@ export function createClusterRunManagementTransport( retry: createRunManualRetryResponseBody(result), }); } + if (command.operation === 'run.cancellation.summary') { + const result = await options.service.summarizeCancellation({ + projectId: command.request.projectId, + requestId: command.request.requestId, + auditEventId: command.request.auditEventId, + failureAuditEventId: command.request.failureAuditEventId, + principal, + }); + return Object.freeze({ + schemaVersion: 1, + operation: command.operation, + summary: Object.freeze({ + schema: RUN_CANCELLATION_DISPATCH_SUMMARY_SCHEMA, + ...result, + }), + }); + } if (command.operation === 'run.cancellation.inspect') { const result = await options.service.inspectCancellation({ projectId: command.request.projectId, diff --git a/packages/ql3-cluster-admin/test/runManagement.test.cjs b/packages/ql3-cluster-admin/test/runManagement.test.cjs index 9222ca9e..6aeda9f4 100644 --- a/packages/ql3-cluster-admin/test/runManagement.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagement.test.cjs @@ -163,6 +163,30 @@ function fixture(role = 'operator', options = {}) { ) { return { rows: [], rowCount: 0 }; } + if ( + text.includes('FROM "ql3"."run_cancellation_dispatches" AS dispatch') + ) { + return { + rows: [ + { + total: '5', + pending: '1', + leased: '1', + retryWait: '1', + dispatched: '1', + blocked: '1', + due: '1', + expiredLease: '1', + identityMismatch: '1', + pidMismatch: '0', + unsupported: '0', + invalid: '0', + oldestBlockedAtMs: String(NOW - 1_500), + }, + ], + rowCount: 1, + }; + } if ( text.startsWith('SELECT attempt_id AS "attemptId"') && text.includes('FROM "ql3"."run_cancellation_dispatches"') && @@ -370,6 +394,37 @@ test('allows a viewer to inspect only low-sensitive cancellation state', async ( assert.equal(audit.params[0], inspectRequest.auditEventId); }); +test('allows a viewer to summarize Project cancellation availability atomically', async () => { + const { calls, service } = fixture('viewer'); + const summaryRequest = { + projectId: 'project-1', + requestId: 'request-summary-1', + auditEventId: '019f9500-0000-4000-8000-000000000061', + failureAuditEventId: '019f9500-0000-4000-8000-000000000062', + principal: request().principal, + }; + const result = await service.summarizeCancellation(summaryRequest); + assert.equal(result.assessment, 'attention_required'); + assert.equal(result.operatorAction, 'inspect'); + assert.equal(result.dispatches.blocked, 1); + assert.equal(result.blockingResults.identityMismatch, 1); + assert.equal(Object.hasOwn(result, 'runId'), false); + assert.equal(JSON.stringify(result).includes('attemptId'), false); + const aggregate = calls.find(({ sql }) => + sql.includes('FROM "ql3"."run_cancellation_dispatches" AS dispatch'), + ); + assert.deepEqual(aggregate.params, ['project-1', NOW]); + const audit = calls.find( + ({ sql, params }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"') && + params[2] === 'run.cancellation.summary', + ); + assert.equal(audit.params[0], summaryRequest.auditEventId); + assert.ok( + calls.indexOf(audit) < calls.findIndex(({ sql }) => sql === 'COMMIT'), + ); +}); + test('authorizes exact cancellation rearm and keeps the event identity server-side', async () => { const { calls, service } = fixture(); const rearmRequest = { diff --git a/packages/ql3-cluster-admin/test/runManagementClient.test.cjs b/packages/ql3-cluster-admin/test/runManagementClient.test.cjs index bfd2a5a1..3d85bc22 100644 --- a/packages/ql3-cluster-admin/test/runManagementClient.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagementClient.test.cjs @@ -132,6 +132,20 @@ const inspectCommand = normalizeClusterRunManagementCommand({ }, }); +const summaryCommand = normalizeClusterRunManagementCommand({ + schemaVersion: 1, + operation: 'run.cancellation.summary', + request: { + projectId: 'project-1', + requestId: 'request-summary-1', + auditEventId: '019f9400-0000-4000-8000-000000000051', + failureAuditEventId: '019f9400-0000-4000-8000-000000000052', + body: { + schema: 'qinglong/run-cancellation-dispatch-summary-request@v1', + }, + }, +}); + const rearmCommand = normalizeClusterRunManagementCommand({ schemaVersion: 1, operation: 'run.cancellation.rearm', @@ -310,6 +324,67 @@ test('validates a low-sensitive cancellation diagnostic and rejects capability l } }); +test('validates the fixed low-sensitive Project cancellation summary', () => { + const value = { + schemaVersion: 1, + operation: 'run.cancellation.summary', + summary: { + schema: 'qinglong/run-cancellation-dispatch-summary@v1', + projectId: 'project-1', + observedAtMs: 1_000_000, + assessment: 'attention_required', + operatorAction: 'inspect', + dispatches: { + total: 5, + pending: 1, + leased: 1, + retryWait: 1, + dispatched: 1, + blocked: 1, + }, + signals: { due: 1, expiredLease: 1 }, + blockingResults: { + identityMismatch: 1, + pidMismatch: 0, + unsupported: 0, + invalid: 0, + }, + oldestBlockedAtMs: 999_200, + }, + }; + assert.deepEqual( + validateClusterRunManagementClientResult(value, summaryCommand), + value, + ); + for (const summary of [ + { ...value.summary, projectId: 'project-2' }, + { ...value.summary, assessment: 'clear' }, + { ...value.summary, operatorAction: 'wait' }, + { + ...value.summary, + dispatches: { ...value.summary.dispatches, total: 6 }, + }, + { + ...value.summary, + blockingResults: { + ...value.summary.blockingResults, + identityMismatch: 0, + }, + }, + { ...value.summary, oldestBlockedAtMs: 1_000_001 }, + { ...value.summary, runId: 'run-1' }, + ]) { + assert.throws( + () => + validateClusterRunManagementClientResult( + { ...value, summary }, + summaryCommand, + ), + ClusterPluginPackageManagementClientRequestError, + ); + } +}); + test('binds a rearm receipt to the exact dispatch version, result and delay fences', () => { const value = { schemaVersion: 1, diff --git a/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs b/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs index e22dee5b..68e0837c 100644 --- a/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs +++ b/packages/ql3-cluster-admin/test/runManagementTransport.test.cjs @@ -119,6 +119,31 @@ function diagnosticResult() { }; } +function summaryResult() { + return { + projectId: 'project-1', + observedAtMs: NOW, + assessment: 'attention_required', + operatorAction: 'inspect', + dispatches: { + total: 5, + pending: 1, + leased: 1, + retryWait: 1, + dispatched: 1, + blocked: 1, + }, + signals: { due: 1, expiredLease: 1 }, + blockingResults: { + identityMismatch: 1, + pidMismatch: 0, + unsupported: 0, + invalid: 0, + }, + oldestBlockedAtMs: NOW - 800, + }; +} + function rearmResult() { return { status: 'rearmed', @@ -153,6 +178,23 @@ function inspectCommand(overrides = {}) { }; } +function summaryCommand(overrides = {}) { + return { + schemaVersion: 1, + operation: 'run.cancellation.summary', + request: { + projectId: 'project-1', + requestId: 'request-summary-1', + auditEventId: '019f9300-0000-4000-8000-000000000051', + failureAuditEventId: '019f9300-0000-4000-8000-000000000052', + body: { + schema: 'qinglong/run-cancellation-dispatch-summary-request@v1', + }, + ...overrides, + }, + }; +} + function rearmCommand(overrides = {}) { return { schemaVersion: 1, @@ -187,6 +229,9 @@ test('routes one exact strong User retry and emits the shared response', async ( async stop() { return stopResult(); }, + async summarizeCancellation() { + return summaryResult(); + }, async inspectCancellation() { return diagnosticResult(); }, @@ -223,6 +268,9 @@ test('routes one exact strong User stop and emits the shared response', async () calls.push(request); return stopResult(); }, + async summarizeCancellation() { + return summaryResult(); + }, async inspectCancellation() { return diagnosticResult(); }, @@ -259,6 +307,9 @@ test('rejects weak or non-User identity before service authority', async () => { async stop() { return stopResult(); }, + async summarizeCancellation() { + return summaryResult(); + }, async inspectCancellation() { return diagnosticResult(); }, @@ -290,6 +341,7 @@ test('routes bounded cancellation inspection without lease capability data', asy service: { async retry() { return retryResult(); }, async stop() { return stopResult(); }, + async summarizeCancellation() { return summaryResult(); }, async inspectCancellation(request) { calls.push(request); return diagnosticResult(); @@ -313,6 +365,39 @@ test('routes bounded cancellation inspection without lease capability data', asy assert.equal(JSON.stringify(result).includes('leaseToken'), false); }); +test('routes one Project-scoped cancellation summary without Run identity', async () => { + const calls = []; + const transport = createClusterRunManagementTransport({ + now: () => NOW, + service: { + async retry() { return retryResult(); }, + async stop() { return stopResult(); }, + async summarizeCancellation(request) { + calls.push(request); + return summaryResult(); + }, + async inspectCancellation() { return diagnosticResult(); }, + async rearmCancellation() { return rearmResult(); }, + }, + }); + const result = await transport.execute(summaryCommand(), { + authenticate: async () => principal(), + }); + assert.equal(calls.length, 1); + assert.equal(calls[0].projectId, 'project-1'); + assert.equal(Object.hasOwn(calls[0], 'runId'), false); + assert.deepEqual(result, { + schemaVersion: 1, + operation: 'run.cancellation.summary', + summary: { + schema: 'qinglong/run-cancellation-dispatch-summary@v1', + ...summaryResult(), + }, + }); + assert.equal(JSON.stringify(result).includes('attemptId'), false); + assert.equal(JSON.stringify(result).includes('leaseOwner'), false); +}); + test('routes an exact blocked cancellation rearm receipt', async () => { const calls = []; const transport = createClusterRunManagementTransport({ @@ -320,6 +405,7 @@ test('routes an exact blocked cancellation rearm receipt', async () => { service: { async retry() { return retryResult(); }, async stop() { return stopResult(); }, + async summarizeCancellation() { return summaryResult(); }, async inspectCancellation() { return diagnosticResult(); }, async rearmCancellation(request) { calls.push(request); diff --git a/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts b/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts index 77800a28..dcf40d1a 100644 --- a/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts +++ b/packages/ql3-cluster-postgres/src/entrypoints/runManager.ts @@ -8,8 +8,10 @@ export { type BlockingCancellationDispatchResult, type PostgresRunCancellationDispatchInspectCommand, type PostgresRunCancellationDispatchRearmCommand, + type PostgresRunCancellationDispatchSummaryCommand, type RunCancellationDispatchDiagnostic, type RunCancellationDispatchRearmReceipt, + type RunCancellationDispatchSummary, } from '../run-management/runCancellationDispatchManagementRepository'; export { PostgresClusterRunCancellationRepository, diff --git a/packages/ql3-cluster-postgres/src/run-management/runCancellationDispatchManagementRepository.ts b/packages/ql3-cluster-postgres/src/run-management/runCancellationDispatchManagementRepository.ts index c471a070..6762a430 100644 --- a/packages/ql3-cluster-postgres/src/run-management/runCancellationDispatchManagementRepository.ts +++ b/packages/ql3-cluster-postgres/src/run-management/runCancellationDispatchManagementRepository.ts @@ -57,15 +57,47 @@ export type RunCancellationDispatchRearmReceipt = Readonly<{ eventSequence: number; }>; -interface ManagementAuthority { +export type RunCancellationDispatchSummary = Readonly<{ + projectId: string; + observedAtMs: number; + assessment: 'clear' | 'converging' | 'attention_required'; + operatorAction: 'none' | 'wait' | 'inspect'; + dispatches: Readonly<{ + total: number; + pending: number; + leased: number; + retryWait: number; + dispatched: number; + blocked: number; + }>; + signals: Readonly<{ + due: number; + expiredLease: number; + }>; + blockingResults: Readonly<{ + identityMismatch: number; + pidMismatch: number; + unsupported: number; + invalid: number; + }>; + oldestBlockedAtMs?: number; +}>; + +interface ProjectManagementAuthority { readonly projectId: string; - readonly runId: string; readonly requestId: string; readonly auditEventId: string; readonly principal: Readonly; readonly policyFence: Readonly; } +interface ManagementAuthority extends ProjectManagementAuthority { + readonly runId: string; +} + +export interface PostgresRunCancellationDispatchSummaryCommand + extends ProjectManagementAuthority {} + export interface PostgresRunCancellationDispatchInspectCommand extends ManagementAuthority {} @@ -282,6 +314,39 @@ function normalizeInspectCommand( return normalizeAuthority(value, []); } +function normalizeSummaryCommand( + value: Readonly, +): Readonly { + const input = exact(value, [ + 'projectId', + 'requestId', + 'auditEventId', + 'principal', + 'policyFence', + ]); + const principal = exact(input.principal, [ + 'subject', + 'authenticationId', + 'authenticatedAtMs', + 'expiresAtMs', + 'assurance', + ]) as unknown as SecurityPrincipal; + const fence = exact(input.policyFence, [ + 'projectVersion', + 'bindingVersion', + ]); + return Object.freeze({ + projectId: identifier(input.projectId), + 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 normalizeRearmCommand( value: Readonly, ): Readonly { @@ -383,7 +448,7 @@ function strongPrincipal( async function confirmAuthorization( client: PostgresClient, - command: Readonly, + command: Readonly, ): Promise { const result = await client.query( `SELECT "ql3"."lock_run_management_policy_fence"( @@ -406,8 +471,11 @@ async function confirmAuthorization( async function recordAllowedAudit( client: PostgresClient, - command: Readonly, - operationId: 'run.cancellation.inspect' | 'run.cancellation.rearm', + command: Readonly, + operationId: + | 'run.cancellation.summary' + | 'run.cancellation.inspect' + | 'run.cancellation.rearm', observedAtMs: number, ): Promise { const inserted = await client.query( @@ -463,6 +531,78 @@ async function recordAllowedAudit( } } +function summaryProjection( + projectId: string, + observedAtMs: number, + row: Row, +): Readonly { + const dispatches = Object.freeze({ + total: integer(row, 'total'), + pending: integer(row, 'pending'), + leased: integer(row, 'leased'), + retryWait: integer(row, 'retryWait'), + dispatched: integer(row, 'dispatched'), + blocked: integer(row, 'blocked'), + }); + const signals = Object.freeze({ + due: integer(row, 'due'), + expiredLease: integer(row, 'expiredLease'), + }); + const blockingResults = Object.freeze({ + identityMismatch: integer(row, 'identityMismatch'), + pidMismatch: integer(row, 'pidMismatch'), + unsupported: integer(row, 'unsupported'), + invalid: integer(row, 'invalid'), + }); + if ( + dispatches.total !== + dispatches.pending + + dispatches.leased + + dispatches.retryWait + + dispatches.dispatched + + dispatches.blocked || + dispatches.blocked !== + blockingResults.identityMismatch + + blockingResults.pidMismatch + + blockingResults.unsupported + + blockingResults.invalid || + signals.due > dispatches.pending + dispatches.retryWait || + signals.expiredLease > dispatches.leased + ) { + throw new TypeError('PostgreSQL cancellation management summary is invalid'); + } + const oldestBlockedAtMs = optionalInteger(row, 'oldestBlockedAtMs'); + if ( + (dispatches.blocked === 0) !== (oldestBlockedAtMs === undefined) || + (oldestBlockedAtMs !== undefined && oldestBlockedAtMs > observedAtMs) + ) { + throw new TypeError( + 'PostgreSQL cancellation management blocked summary is invalid', + ); + } + const active = + dispatches.pending + + dispatches.leased + + dispatches.retryWait + + dispatches.blocked; + return Object.freeze({ + projectId, + observedAtMs, + assessment: + dispatches.blocked > 0 + ? 'attention_required' + : active > 0 + ? 'converging' + : 'clear', + operatorAction: + dispatches.blocked > 0 ? 'inspect' : active > 0 ? 'wait' : 'none', + dispatches, + signals, + blockingResults, + ...(oldestBlockedAtMs === undefined ? {} : { oldestBlockedAtMs }), + }); +} + function runStatus(row: Row): RunStatus { const value = text(row, 'runStatus') as RunStatus; if (!RUN_STATUSES.includes(value)) { @@ -608,6 +748,76 @@ export class PostgresRunCancellationDispatchManagementRepository { } } + summary( + value: Readonly, + ): Promise> { + const command = normalizeSummaryCommand(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 result = await client.query( + `SELECT count(*)::bigint AS total, + count(*) FILTER (WHERE dispatch.status = 'pending')::bigint + AS pending, + count(*) FILTER (WHERE dispatch.status = 'leased')::bigint + AS leased, + count(*) FILTER (WHERE dispatch.status = 'retry_wait')::bigint + AS "retryWait", + count(*) FILTER (WHERE dispatch.status = 'dispatched')::bigint + AS dispatched, + count(*) FILTER (WHERE dispatch.status = 'blocked')::bigint + AS blocked, + count(*) FILTER ( + WHERE dispatch.status IN ('pending', 'retry_wait') + AND dispatch.next_attempt_at_ms <= $2 + )::bigint AS due, + count(*) FILTER ( + WHERE dispatch.status = 'leased' + AND dispatch.lease_expires_at_ms <= $2 + )::bigint AS "expiredLease", + count(*) FILTER ( + WHERE dispatch.status = 'blocked' + AND dispatch.last_result = 'identity_mismatch' + )::bigint AS "identityMismatch", + count(*) FILTER ( + WHERE dispatch.status = 'blocked' + AND dispatch.last_result = 'pid_mismatch' + )::bigint AS "pidMismatch", + count(*) FILTER ( + WHERE dispatch.status = 'blocked' + AND dispatch.last_result = 'unsupported' + )::bigint AS unsupported, + count(*) FILTER ( + WHERE dispatch.status = 'blocked' + AND dispatch.last_result = 'invalid' + )::bigint AS invalid, + min(dispatch.updated_at_ms) FILTER ( + WHERE dispatch.status = 'blocked' + ) AS "oldestBlockedAtMs" + FROM "ql3"."run_cancellation_dispatches" AS dispatch + JOIN "ql3"."runs" AS run ON run.id = dispatch.run_id + WHERE run.project_id = $1`, + [command.projectId, observedAtMs], + ); + if (result.rows.length !== 1) { + throw new TypeError( + 'PostgreSQL cancellation management summary row is invalid', + ); + } + await recordAllowedAudit( + client, + authorized, + 'run.cancellation.summary', + observedAtMs, + ); + return summaryProjection(command.projectId, observedAtMs, result.rows[0]!); + }); + } + inspect( value: Readonly, ): Promise> { diff --git a/packages/ql3-cluster-postgres/test/runCancellationDispatchManagementRepository.test.cjs b/packages/ql3-cluster-postgres/test/runCancellationDispatchManagementRepository.test.cjs index 274268f4..8d55f322 100644 --- a/packages/ql3-cluster-postgres/test/runCancellationDispatchManagementRepository.test.cjs +++ b/packages/ql3-cluster-postgres/test/runCancellationDispatchManagementRepository.test.cjs @@ -42,6 +42,11 @@ function rearmCommand(overrides = {}) { }); } +function summaryCommand(overrides = {}) { + const { runId: _runId, ...authority } = command(); + return { ...authority, ...overrides }; +} + function runRow() { return { projectId: 'project-1', @@ -104,6 +109,30 @@ function fixture(options = {}) { ) { return { rows: options.replay ? [options.replay] : [], rowCount: 0 }; } + if ( + text.includes('FROM "ql3"."run_cancellation_dispatches" AS dispatch') + ) { + return { + rows: [ + options.summary ?? { + total: '5', + pending: '1', + leased: '1', + retryWait: '1', + dispatched: '1', + blocked: '1', + due: '1', + expiredLease: '1', + identityMismatch: '1', + pidMismatch: '0', + unsupported: '0', + invalid: '0', + oldestBlockedAtMs: String(NOW - 1_500), + }, + ], + rowCount: 1, + }; + } if ( text.startsWith('SELECT attempt_id AS "attemptId"') && !text.includes('dispatchStatus') && @@ -179,6 +208,94 @@ test('inspects one low-sensitive blocked dispatch under run.read authority', asy ); }); +test('summarizes one Project without exposing Run, Attempt or lease identity', async () => { + const { calls, repository } = fixture(); + const result = await repository.summary(summaryCommand()); + assert.deepEqual(result, { + projectId: 'project-1', + observedAtMs: NOW, + assessment: 'attention_required', + operatorAction: 'inspect', + dispatches: { + total: 5, + pending: 1, + leased: 1, + retryWait: 1, + dispatched: 1, + blocked: 1, + }, + signals: { due: 1, expiredLease: 1 }, + blockingResults: { + identityMismatch: 1, + pidMismatch: 0, + unsupported: 0, + invalid: 0, + }, + oldestBlockedAtMs: NOW - 1_500, + }); + const read = calls.find(({ sql }) => + sql.includes('FROM "ql3"."run_cancellation_dispatches" AS dispatch'), + ); + assert.deepEqual(read.params, ['project-1', NOW]); + for (const forbidden of [ + 'attempt_id AS', + 'run.id AS', + 'lease_owner', + 'lease_token', + 'lease_token_digest', + ]) { + assert.equal(read.sql.includes(forbidden), false); + } + const audit = calls.find( + ({ sql, params }) => + sql.startsWith('INSERT INTO "ql3"."security_audit_events"') && + params[2] === 'run.cancellation.summary', + ); + assert.equal(audit.params[0], summaryCommand().auditEventId); +}); + +test('derives clear and converging assessments from fixed status counts', async () => { + const clear = fixture({ + summary: { + total: '1', + pending: '0', + leased: '0', + retryWait: '0', + dispatched: '1', + blocked: '0', + due: '0', + expiredLease: '0', + identityMismatch: '0', + pidMismatch: '0', + unsupported: '0', + invalid: '0', + oldestBlockedAtMs: null, + }, + }); + assert.equal((await clear.repository.summary(summaryCommand())).assessment, 'clear'); + + const converging = fixture({ + summary: { + total: '1', + pending: '0', + leased: '0', + retryWait: '1', + dispatched: '0', + blocked: '0', + due: '1', + expiredLease: '0', + identityMismatch: '0', + pidMismatch: '0', + unsupported: '0', + invalid: '0', + oldestBlockedAtMs: null, + }, + }); + const result = await converging.repository.summary(summaryCommand()); + assert.equal(result.assessment, 'converging'); + assert.equal(result.operatorAction, 'wait'); +}); + test('rearms an exact blocked dispatch with one event and allowed audit', async () => { const { calls, repository } = fixture(); const result = await repository.rearm(rearmCommand()); diff --git a/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs b/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs index 850b3204..f3e01dd0 100644 --- a/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs +++ b/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs @@ -25,6 +25,7 @@ const FIXTURE = Object.freeze({ terminalEventId: 'ha-cancel-terminal-event-d363', settledEventId: 'ha-cancel-settled-event-d363', blockedEventId: 'ha-cancel-blocked-event-d365', + summaryAuditEventId: '019f9700-0000-4000-8000-000000000005', inspectAuditEventId: '019f9700-0000-4000-8000-000000000001', rearmAuditEventId: '019f9700-0000-4000-8000-000000000002', rearmMutationId: '019f9700-0000-4000-8000-000000000003', @@ -262,6 +263,22 @@ async function persistCancellationDispatchHaFixture(options) { principal, policyFence: Object.freeze({ projectVersion: 1, bindingVersion: 1 }), }); + const summary = await management.summary({ + projectId: FIXTURE.projectId, + requestId: 'ha-cancel-summary-d366', + auditEventId: FIXTURE.summaryAuditEventId, + principal, + policyFence: authority.policyFence, + }); + assert.equal(summary.assessment, 'attention_required'); + assert.equal(summary.operatorAction, 'inspect'); + assert.equal(summary.dispatches.blocked, 1); + assert.equal(summary.blockingResults.identityMismatch, 1); + assert.equal(summary.signals.due, 0); + assert.equal(summary.signals.expiredLease, 0); + assert.equal(Object.hasOwn(summary, 'runId'), false); + assert.equal(JSON.stringify(summary).includes('attemptId'), false); + assert.equal(JSON.stringify(summary).includes('leaseOwner'), false); const diagnostic = await management.inspect(authority); assert.equal(diagnostic.operatorAction, 'rearm'); assert.equal(diagnostic.dispatch?.status, 'blocked'); @@ -380,6 +397,7 @@ async function persistCancellationDispatchHaFixture(options) { staleLeaseFenced: true, retryDeferredUntilDue: true, operatorDiagnosticLowSensitive: true, + operatorSummaryLowSensitiveAndActionable: true, manualBlockedRearmExact: true, manualRearmDeferredUntilDue: true, productionDeliverySettledBeforeStop: true, diff --git a/scripts/ql3-postgres-ha-contract.cjs b/scripts/ql3-postgres-ha-contract.cjs index c2c7d4c6..a46cd1ed 100644 --- a/scripts/ql3-postgres-ha-contract.cjs +++ b/scripts/ql3-postgres-ha-contract.cjs @@ -13985,6 +13985,8 @@ async function main(argv = process.argv.slice(2)) { cancellationDispatch.expiredLeaseTakenOver && cancellationDispatch.staleLeaseFenced && cancellationDispatch.retryDeferredUntilDue, + cancellationDispatchSummaryIsLowSensitiveAndActionable: + cancellationDispatch.operatorSummaryLowSensitiveAndActionable, cancellationDispatchReplicatesAndSurvivesPromotion: cancellationDispatch.replicatedBeforePromotion && cancellationDispatch.survivedPromotion,