feat(ql3): operationalize cancellation rearm

This commit is contained in:
whyour
2026-08-19 07:54:02 +08:00
parent 0b5f3bcb39
commit 5261c41828
25 changed files with 2758 additions and 58 deletions
+14
View File
@@ -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 digestrearm 要求 `run.stop`,只接受
`blocked + expected dispatch version + expected blocking result` 的精确 CASretry delay 固定为 1 秒至 24 小时并由 PostgreSQL
`transaction_timestamp()` 计算绝对到期时间。成功事务同时推进 Run version/event sequence、保留 blocking result 作为历史、追加
`run.cancel_dispatch_rearmed` 与 allowed auditmutation replay 返回同一不可变 receiptstale 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 个 source171 nested),基础
Edge/Standalone 仍为 `2,589,998 / 2,590,076` bytesApplication+AI 为 `4,493,151 / 4,493,283` bytes。PostgreSQL 18.6 arm64 HA
`144/144` 真实执行 blocked→低敏 inspect→CAS rearm→not-due→生产交付→WAL→promotiontimeline `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`
@@ -1,6 +1,6 @@
# ADR-0005Durable Cancellation Dispatch、Lease 与 Fencing
- 状态:AcceptedLocalPostgreSQL Repository 已实现;Cluster 生产启动拓扑待接入
- 状态:AcceptedLocalPostgreSQL RepositoryCluster 生产交付与人工处置入口已实现
- 日期: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-controlRun-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-controlRun-level stop 必须先结算 durable dispatch 再返回,Workflow Task timeout 则保持 `untracked`,不伪造父 Run 取消。ADR-0458 又在既有隔离 Run management plane 上增加低敏 inspect 与 exact-CAS rearmviewer 只有 `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 winnerraw token 不落库,数据库时间决定 lease/retry 到期。
12. v65 事实经 WAL 到达 standby,提升为新 Primary 后仍可读取;旧 owner/token/version 继续被 fencing。
13. Cluster Worker lease-control 只在 durable dispatch 已结算或已重放时释放 Run-level stopWorkflow Task timeout 不写父 Run 取消事实。
14. 强认证 viewer 只能读取无 lease capability 的低敏诊断;只有 `run.stop` authority 可按 blocked/version/result 精确 rearm。
15. rearm、RunEvent 与 allowed audit 原子提交,数据库时间决定 retry duestale fence、授权漂移与 mutation drift 均失败关闭。
@@ -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 混合。
@@ -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 存量迁移继续保持独立议题。
@@ -0,0 +1,65 @@
# ADR-0458:最小权限 CancellationDispatch 诊断与人工 Rearm
- 状态:Accepted
- 日期:2026-08-19
- 关联 RFCQL-RFC-0001 D-365、PR-5、PR-7
- 关联 ADRADR-0005、ADR-0383、ADR-0456、ADR-0457
- AmendsADR-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 四项审计零 findingworkspace 仍为 18 包,Cluster PostgreSQL 为 172 source/171 nested,根入口计数不变。
- `14/14` Local artifact audit 通过:基础 Edge/Standalone `2,589,998 / 2,590,076` bytesApplication+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 promotiontimeline `1→2`,报告 SHA-256 `58f43327f426c286aaa1764aa6dd9f5963a462304716e09b933f326991442b8a`
## 后续
私有 operator 协议已经闭合,但面向产品的聚合 blocked/availability 指标、告警和 UI 仍待实现。CloudNativePG live failover、多副本容量压力、固定 Linux x64/arm64 与物理 Edge 资源证据继续作为发布最终化门,不由单机 Docker HA 结果替代。
@@ -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<SecurityPrincipal>;
}
export interface ClusterRunManagementCancellationInspectRequest {
readonly projectId: string;
readonly runId: string;
readonly requestId: string;
readonly auditEventId: string;
readonly failureAuditEventId: string;
readonly principal: Readonly<SecurityPrincipal>;
}
export interface ClusterRunManagementCancellationRearmRequest
extends ClusterRunManagementCancellationInspectRequest {
readonly mutationId: string;
readonly expectedDispatchVersion: number;
readonly expectedLastResult: BlockingCancellationDispatchResult;
readonly retryDelayMs: number;
}
export interface ClusterRunManagementService {
retry(
request: Readonly<ClusterRunManagementRetryRequest>,
@@ -64,6 +90,12 @@ export interface ClusterRunManagementService {
stop(
request: Readonly<ClusterRunManagementStopRequest>,
): Promise<Readonly<ClusterRunCancellationResult>>;
inspectCancellation(
request: Readonly<ClusterRunManagementCancellationInspectRequest>,
): Promise<Readonly<RunCancellationDispatchDiagnostic>>;
rearmCancellation(
request: Readonly<ClusterRunManagementCancellationRearmRequest>,
): Promise<Readonly<RunCancellationDispatchRearmReceipt>>;
}
export interface ClusterRunManagementOptions {
@@ -178,6 +210,56 @@ function exactStopRequest(
}
}
function exactCancellationInspectRequest(
value: unknown,
): asserts value is Readonly<ClusterRunManagementCancellationInspectRequest> {
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<ClusterRunManagementCancellationRearmRequest> {
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<ClusterRunManagementCancellationInspectRequest>,
) {
exactCancellationInspectRequest(requestValue);
const observedAtMs = now();
let principal: Readonly<SecurityPrincipal>;
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<SecurityPolicyFence> | 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<ClusterRunManagementCancellationRearmRequest>,
) {
exactCancellationRearmRequest(requestValue);
const observedAtMs = now();
let principal: Readonly<SecurityPrincipal>;
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<SecurityPolicyFence> | 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 });
}
},
});
}
@@ -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<string, unknown>;
}
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<ClusterRunManagementCommand>,
@@ -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 ||
@@ -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<RunCancellationResponseBody>;
}>;
export type ClusterRunManagementCancellationInspectTransportResult = Readonly<{
schemaVersion: 1;
operation: 'run.cancellation.inspect';
diagnostic: Readonly<
Awaited<ReturnType<ClusterRunManagementService['inspectCancellation']>> & {
schema: typeof RUN_CANCELLATION_DISPATCH_DIAGNOSTIC_SCHEMA;
}
>;
}>;
export type ClusterRunManagementCancellationRearmTransportResult = Readonly<{
schemaVersion: 1;
operation: 'run.cancellation.rearm';
rearm: Readonly<
Awaited<ReturnType<ClusterRunManagementService['rearmCancellation']>> & {
schema: typeof RUN_CANCELLATION_DISPATCH_REARM_RECEIPT_SCHEMA;
}
>;
}>;
export type ClusterRunManagementTransportResult =
| ClusterRunManagementRetryTransportResult
| ClusterRunManagementStopTransportResult;
| ClusterRunManagementStopTransportResult
| ClusterRunManagementCancellationInspectTransportResult
| ClusterRunManagementCancellationRearmTransportResult;
export interface ClusterRunManagementAuthentication {
authenticate(): Promise<Readonly<SecurityPrincipal> | 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<typeof parseRunCancellationRequestBody>;
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,
@@ -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,
);
});
@@ -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,
);
}
});
@@ -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(
() =>
@@ -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,
@@ -338,5 +338,10 @@ export const postgresqlMainMigrationManifest: MigrationStreamManifest =
checksum:
'b6d7ac81b5f75530df05f8ef05878fa30aa0f4418363973ded89d14ffce151b2',
}),
Object.freeze({
id: 'pg-0067-cancellation-dispatch-management',
checksum:
'e78e24a06dc4c4dbdd859685f28b4bc837a8cfb279eb3512e0a57dc6d27eaaaa',
}),
]),
});
@@ -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<PostgresMigrationContext> =
Object.freeze({
@@ -143,5 +144,6 @@ export const postgresqlMainMigrationStream: MigrationStreamDefinition<PostgresMi
pg0064PluginPackageSecretBindingTransitionApprovalPlansMigration,
pg0065ApprovedActionManualRecoveryMigration,
pg0066CancellationDispatchMigration,
pg0067CancellationDispatchManagementMigration,
]),
});
@@ -0,0 +1,35 @@
import { CAPABILITIES_V65 } from '../run/migrations/pg-0066-cancellation-dispatch';
import { definePostgresSqlMigration } from '../migrations/sqlMigration';
export const CAPABILITIES_V66 = CAPABILITIES_V65.replace(
'"run_cancellation_dispatch":1,',
'"run_cancellation_dispatch":1,"run_cancellation_dispatch_management":1,',
);
export const pg0067CancellationDispatchManagementMigration =
definePostgresSqlMigration({
id: 'pg-0067-cancellation-dispatch-management',
statements: [
`ALTER TABLE "ql3"."run_cancellation_dispatches" DROP CONSTRAINT ql3_run_cancellation_dispatch_result_state_check`,
`
ALTER TABLE "ql3"."run_cancellation_dispatches"
ADD CONSTRAINT ql3_run_cancellation_dispatch_result_state_check CHECK (
(status = 'pending' AND last_result IS NULL) OR
(status IN ('leased', 'retry_wait') AND last_result IN (
'identity_mismatch', 'pid_mismatch', 'unsupported', 'invalid',
'controller_missing', 'handle_missing', 'dispatch_error'
)) OR
(status = 'leased' AND last_result IS NULL) OR
(status = 'dispatched' AND last_result IN (
'termination_requested', 'already_exited'
)) OR
(status = 'blocked' AND last_result IN (
'identity_mismatch', 'pid_mismatch', 'unsupported', 'invalid'
))
)
`.trim(),
`GRANT SELECT ON "ql3"."run_cancellation_dispatches" TO ql3_run_manager`,
`GRANT UPDATE (status, version, next_attempt_at_ms, updated_at_ms) ON "ql3"."run_cancellation_dispatches" TO ql3_run_manager`,
`DO $ql3$ BEGIN UPDATE "ql3"."schema_capabilities" SET contract_version = 66, migration_id = 'pg-0067-cancellation-dispatch-management', capabilities = '${CAPABILITIES_V66}'::jsonb, updated_at_ms = floor(extract(epoch FROM transaction_timestamp()) * 1000)::bigint WHERE contract_name = 'control-core' AND contract_version = 65 AND migration_id = 'pg-0066-cancellation-dispatch' AND capabilities = '${CAPABILITIES_V65}'::jsonb; IF NOT FOUND THEN RAISE EXCEPTION 'control-core capability is not at version 65' USING ERRCODE = 'check_violation'; END IF; END $ql3$`,
],
});
@@ -0,0 +1,892 @@
import type { PostgresClient, PostgresPool } from '@qinglong/runtime-core';
import {
RUN_STATUSES,
type RunStatus,
type SecurityPolicyFence,
type SecurityPrincipal,
} from '@qinglong/runtime-core';
import {
CANCELLATION_DISPATCH_BLOCKING_RESULTS,
CANCELLATION_DISPATCH_RESULTS,
CANCELLATION_DISPATCH_STATUSES,
MAX_CANCELLATION_DISPATCH_RETRY_DELAY_MS,
type CancellationDispatchResult,
type CancellationDispatchStatus,
} from '@qinglong/runtime-core/cancellation-dispatch';
import { normalizeSecurityPrincipal } from '@qinglong/runtime-core/security';
type Row = Record<string, unknown>;
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<SecurityPrincipal>;
readonly policyFence: Readonly<SecurityPolicyFence>;
}
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<RunStatus>([
'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<string, unknown> {
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<string, unknown>;
}
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<ManagementAuthority> & Record<string, unknown> {
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<PostgresRunCancellationDispatchInspectCommand>,
): Readonly<PostgresRunCancellationDispatchInspectCommand> {
return normalizeAuthority(value, []);
}
function normalizeRearmCommand(
value: Readonly<PostgresRunCancellationDispatchRearmCommand>,
): Readonly<PostgresRunCancellationDispatchRearmCommand> {
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<void> {
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<void> {
try {
await client.query('ROLLBACK');
} catch {
// Preserve the transaction failure.
}
}
async function databaseNow(client: PostgresClient): Promise<number> {
const result = await client.query<Row>(`
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<SecurityPrincipal>,
observedAtMs: number,
): Readonly<SecurityPrincipal> {
let principal: Readonly<SecurityPrincipal>;
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<ManagementAuthority>,
): Promise<void> {
const result = await client.query<Row>(
`SELECT "ql3"."lock_run_management_policy_fence"(
$1::varchar, $2::varchar, $3::varchar, $4::integer, $5::integer
) AS "matches"`,
[
command.projectId,
command.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<ManagementAuthority>,
operationId: 'run.cancellation.inspect' | 'run.cancellation.rearm',
observedAtMs: number,
): Promise<void> {
const inserted = await client.query<Row>(
`INSERT INTO "ql3"."security_audit_events" (
event_id, request_id, operation_id, project_id,
subject_type, subject_id, authentication_id, outcome, reasons,
project_version, binding_version, occurred_at_ms
) VALUES ($1, $2, $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<Row>(
`SELECT request_id AS "requestId", operation_id AS "operationId",
project_id AS "projectId", subject_type AS "subjectType",
subject_id AS "subjectId", authentication_id AS "authenticationId",
outcome, reasons, project_version AS "projectVersion",
binding_version AS "bindingVersion"
FROM "ql3"."security_audit_events" WHERE event_id = $1`,
[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<RunCancellationDispatchDiagnostic['dispatch']> {
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<ManagementAuthority>,
run: Row,
dispatchRow?: Row,
): Readonly<RunCancellationDispatchDiagnostic> {
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<PostgresRunCancellationDispatchRearmCommand>,
event: Row,
): Readonly<RunCancellationDispatchRearmReceipt> {
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<PostgresRunCancellationDispatchInspectCommand>,
): Promise<Readonly<RunCancellationDispatchDiagnostic>> {
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<Row>(
`SELECT project_id AS "projectId", status AS "runStatus",
version AS "runVersion", event_sequence AS "eventSequence",
cancel_requested_at_ms AS "cancelRequestedAtMs",
cancel_reason AS "cancelReason"
FROM "ql3"."runs" WHERE id = $1`,
[command.runId],
);
if (
run.rows.length !== 1 ||
run.rows[0]?.projectId !== command.projectId
) {
throw new RunCancellationDispatchManagementNotFoundError();
}
const dispatch = await client.query<Row>(
`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<PostgresRunCancellationDispatchRearmCommand>,
): Promise<Readonly<RunCancellationDispatchRearmReceipt>> {
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<Row>(
`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<Row>(
`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<Row>(
`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<Row>(
`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<Row>(
`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<T>(
operation: (client: PostgresClient) => Promise<T>,
): Promise<T> {
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();
}
}
}
@@ -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,
@@ -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<ColumnPrivilegeRow>(
`
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<ColumnPrivilegeRow>(
`
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',
@@ -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'/);
});
@@ -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',
);
});
@@ -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);
});
@@ -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') ||
@@ -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();
+5
View File
@@ -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',
+2 -2
View File
@@ -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' },
},
);