diff --git a/back/runtime/adapters/legacy-sequelize/cancellationDispatchRepository.ts b/back/runtime/adapters/legacy-sequelize/cancellationDispatchRepository.ts index e52353e4..fcb297bb 100644 --- a/back/runtime/adapters/legacy-sequelize/cancellationDispatchRepository.ts +++ b/back/runtime/adapters/legacy-sequelize/cancellationDispatchRepository.ts @@ -1,3 +1,4 @@ +import { createHash } from 'crypto'; import { DataTypes, Model, @@ -55,6 +56,19 @@ const BLOCKING_RESULTS: readonly CancellationDispatchResult[] = [ 'unsupported', 'invalid', ]; +const MAX_LEASE_DURATION_MS = 5 * 60_000; +const MAX_RETRY_DELAY_MS = 24 * 60 * 60_000; + +export interface LegacySequelizeCancellationDispatchRepositoryOptions { + clock?: () => number; +} + +function digestLeaseToken(value: string): string { + return createHash('sha256') + .update('qinglong.cancellation-dispatch-lease.v1\0', 'utf8') + .update(value, 'utf8') + .digest('hex'); +} interface CancellationDispatchRow { runId: string; @@ -287,18 +301,13 @@ function assertClaim(command: ClaimCancellationDispatchCommand): void { assertId('owner', command.owner, 128); assertId('leaseToken', command.leaseToken, 128); assertTimestamp('requestedAtMs', command.requestedAtMs); - assertTimestamp('nowMs', command.nowMs); if ( !Number.isSafeInteger(command.leaseDurationMs) || - command.leaseDurationMs < 1 + command.leaseDurationMs < 1 || + command.leaseDurationMs > MAX_LEASE_DURATION_MS ) { throw new InvalidCancellationDispatchCommandError( - 'leaseDurationMs must be a positive safe integer', - ); - } - if (!Number.isSafeInteger(command.nowMs + command.leaseDurationMs)) { - throw new InvalidCancellationDispatchCommandError( - 'lease expiry exceeds the supported range', + `leaseDurationMs must be between 1 and ${MAX_LEASE_DURATION_MS}`, ); } } @@ -311,7 +320,6 @@ function assertRecordResult( assertId('owner', command.owner, 128); assertId('leaseToken', command.leaseToken, 128); assertId('eventId', command.eventId); - assertTimestamp('atMs', command.atMs); if ( !Number.isSafeInteger(command.expectedVersion) || command.expectedVersion < 1 @@ -327,17 +335,18 @@ function assertRecordResult( } if (RETRYABLE_RESULTS.includes(command.result)) { if ( - command.nextAttemptAtMs === undefined || - !Number.isSafeInteger(command.nextAttemptAtMs) || - command.nextAttemptAtMs <= command.atMs + command.retryDelayMs === undefined || + !Number.isSafeInteger(command.retryDelayMs) || + command.retryDelayMs < 1 || + command.retryDelayMs > MAX_RETRY_DELAY_MS ) { throw new InvalidCancellationDispatchCommandError( - 'retryable results require nextAttemptAtMs greater than atMs', + `retryable results require retryDelayMs between 1 and ${MAX_RETRY_DELAY_MS}`, ); } - } else if (command.nextAttemptAtMs !== undefined) { + } else if (command.retryDelayMs !== undefined) { throw new InvalidCancellationDispatchCommandError( - 'terminal results must not include nextAttemptAtMs', + 'terminal results must not include retryDelayMs', ); } } @@ -417,7 +426,9 @@ function rowToDispatch( ? {} : { nextAttemptAtMs: Number(row.nextAttemptAtMs) }), ...(row.leaseOwner === null ? {} : { leaseOwner: row.leaseOwner }), - ...(row.leaseToken === null ? {} : { leaseToken: row.leaseToken }), + ...(row.leaseToken === null + ? {} + : { leaseTokenDigest: digestLeaseToken(row.leaseToken) }), ...(row.leaseExpiresAtMs === null ? {} : { leaseExpiresAtMs: Number(row.leaseExpiresAtMs) }), @@ -447,12 +458,15 @@ function withoutScheduleAndLease( dispatch: CancellationDispatchRecord, ): Omit< CancellationDispatchRecord, - 'nextAttemptAtMs' | 'leaseOwner' | 'leaseToken' | 'leaseExpiresAtMs' + | 'nextAttemptAtMs' + | 'leaseOwner' + | 'leaseTokenDigest' + | 'leaseExpiresAtMs' > { const { nextAttemptAtMs: _nextAttemptAtMs, leaseOwner: _leaseOwner, - leaseToken: _leaseToken, + leaseTokenDigest: _leaseTokenDigest, leaseExpiresAtMs: _leaseExpiresAtMs, ...rest } = dispatch; @@ -466,12 +480,17 @@ export class LegacySequelizeCancellationDispatchRepository private readonly run: ModelStatic; private readonly attempt: ModelStatic; private readonly event: ModelStatic; + private readonly clock: () => number; - constructor(private readonly database: Sequelize) { + constructor( + private readonly database: Sequelize, + options: LegacySequelizeCancellationDispatchRepositoryOptions = {}, + ) { this.dispatch = defineDispatchModel(database); this.run = defineRunModel(database); this.attempt = defineAttemptModel(database); this.event = defineEventModel(database); + this.clock = options.clock ?? Date.now; } async findByRunId(runId: string): Promise { @@ -489,6 +508,7 @@ export class LegacySequelizeCancellationDispatchRepository return this.database.transaction( { type: Transaction.TYPES.IMMEDIATE }, async (transaction) => { + const nowMs = this.now(); const [run, attempt] = await Promise.all([ this.run.findByPk(command.runId, { raw: true, transaction }), this.attempt.findByPk(command.attemptId, { raw: true, transaction }), @@ -530,8 +550,8 @@ export class LegacySequelizeCancellationDispatchRepository leaseExpiresAtMs: null, lastResult: null, lastDispatchedAtMs: null, - createdAtMs: command.nowMs, - updatedAtMs: command.nowMs, + createdAtMs: nowMs, + updatedAtMs: nowMs, }, { transaction }, ); @@ -556,21 +576,26 @@ export class LegacySequelizeCancellationDispatchRepository if ( dispatch.status === 'leased' && dispatch.leaseExpiresAtMs !== undefined && - dispatch.leaseExpiresAtMs > command.nowMs + dispatch.leaseExpiresAtMs > nowMs ) { return { status: 'leased', dispatch }; } if ( dispatch.status !== 'leased' && dispatch.nextAttemptAtMs !== undefined && - dispatch.nextAttemptAtMs > command.nowMs + dispatch.nextAttemptAtMs > nowMs ) { return { status: 'not_due', dispatch }; } const nextVersion = dispatch.version + 1; const nextCount = dispatch.dispatchCount + 1; - const leaseExpiresAtMs = command.nowMs + command.leaseDurationMs; + const leaseExpiresAtMs = nowMs + command.leaseDurationMs; + if (!Number.isSafeInteger(leaseExpiresAtMs)) { + throw new CancellationDispatchRepositoryError( + new Error('Cancellation dispatch lease expiry overflowed'), + ); + } const [affected] = await this.dispatch.update( { status: 'leased', @@ -580,7 +605,7 @@ export class LegacySequelizeCancellationDispatchRepository leaseOwner: command.owner, leaseToken: command.leaseToken, leaseExpiresAtMs, - updatedAtMs: command.nowMs, + updatedAtMs: nowMs, }, { where: { runId: command.runId, version: dispatch.version }, @@ -592,15 +617,16 @@ export class LegacySequelizeCancellationDispatchRepository } return { status: 'claimed', + leaseToken: command.leaseToken, dispatch: { ...withoutScheduleAndLease(dispatch), status: 'leased', version: nextVersion, dispatchCount: nextCount, leaseOwner: command.owner, - leaseToken: command.leaseToken, + leaseTokenDigest: digestLeaseToken(command.leaseToken), leaseExpiresAtMs, - updatedAtMs: command.nowMs, + updatedAtMs: nowMs, }, }; }, @@ -614,6 +640,7 @@ export class LegacySequelizeCancellationDispatchRepository return this.database.transaction( { type: Transaction.TYPES.IMMEDIATE }, async (transaction) => { + const atMs = this.now(); const row = (await this.dispatch.findByPk(command.runId, { raw: true, transaction, @@ -644,6 +671,18 @@ export class LegacySequelizeCancellationDispatchRepository ].includes(command.result); const nextVersion = row.version + 1; const nextSequence = Number(run.eventSequence) + 1; + const nextAttemptAtMs = + command.retryDelayMs === undefined + ? undefined + : atMs + command.retryDelayMs; + if ( + nextAttemptAtMs !== undefined && + !Number.isSafeInteger(nextAttemptAtMs) + ) { + throw new CancellationDispatchRepositoryError( + new Error('Cancellation dispatch retry deadline overflowed'), + ); + } const [runAffected] = await this.run.update( { version: Number(run.version) + 1, eventSequence: nextSequence }, { where: { id: command.runId, version: run.version }, transaction }, @@ -655,15 +694,15 @@ export class LegacySequelizeCancellationDispatchRepository { status: state.status, version: nextVersion, - nextAttemptAtMs: command.nextAttemptAtMs ?? null, + nextAttemptAtMs: nextAttemptAtMs ?? null, leaseOwner: null, leaseToken: null, leaseExpiresAtMs: null, lastResult: command.result, lastDispatchedAtMs: controllerInvoked - ? command.atMs + ? atMs : row.lastDispatchedAtMs, - updatedAtMs: command.atMs, + updatedAtMs: atMs, }, { where: { @@ -694,7 +733,7 @@ export class LegacySequelizeCancellationDispatchRepository dispatch_count: row.dispatchCount, result: command.result, }, - createdAtMs: command.atMs, + createdAtMs: atMs, }; await this.event.create( { @@ -716,20 +755,30 @@ export class LegacySequelizeCancellationDispatchRepository ...withoutScheduleAndLease(rowToDispatch(row)), status: state.status, version: nextVersion, - ...(command.nextAttemptAtMs === undefined + ...(nextAttemptAtMs === undefined ? {} - : { nextAttemptAtMs: command.nextAttemptAtMs }), + : { nextAttemptAtMs }), lastResult: command.result, ...(controllerInvoked - ? { lastDispatchedAtMs: command.atMs } + ? { lastDispatchedAtMs: atMs } : row.lastDispatchedAtMs === null ? {} : { lastDispatchedAtMs: Number(row.lastDispatchedAtMs) }), - updatedAtMs: command.atMs, + updatedAtMs: atMs, }, event, }; }, ); } + + private now(): number { + const nowMs = this.clock(); + if (!Number.isSafeInteger(nowMs) || nowMs < 0) { + throw new CancellationDispatchRepositoryError( + new Error('Cancellation dispatch repository clock is invalid'), + ); + } + return nowMs; + } } diff --git a/back/runtime/application/primaryCancellationDispatcher.ts b/back/runtime/application/primaryCancellationDispatcher.ts index 68d4b15b..928a822a 100644 --- a/back/runtime/application/primaryCancellationDispatcher.ts +++ b/back/runtime/application/primaryCancellationDispatcher.ts @@ -35,7 +35,6 @@ export interface PrimaryCancellationDispatcherOptions { leaseDurationMs?: number; retryBaseMs?: number; retryMaxMs?: number; - clock?: () => number; createId?: () => string; } @@ -55,7 +54,6 @@ export class PrimaryCancellationDispatcher { private readonly leaseDurationMs: number; private readonly retryBaseMs: number; private readonly retryMaxMs: number; - private readonly clock: () => number; private readonly createId: () => string; constructor( @@ -71,7 +69,6 @@ export class PrimaryCancellationDispatcher { this.leaseDurationMs = options.leaseDurationMs ?? DEFAULT_LEASE_DURATION_MS; this.retryBaseMs = options.retryBaseMs ?? DEFAULT_RETRY_BASE_MS; this.retryMaxMs = options.retryMaxMs ?? DEFAULT_RETRY_MAX_MS; - this.clock = options.clock ?? Date.now; this.createId = options.createId ?? uuidV7; assertPositiveInteger('leaseDurationMs', this.leaseDurationMs); assertPositiveInteger('retryBaseMs', this.retryBaseMs); @@ -130,7 +127,6 @@ export class PrimaryCancellationDispatcher { } const attempt = candidate.attempts[0]; - const claimedAtMs = this.now(); let claim; try { claim = await this.dispatches.claim({ @@ -139,7 +135,6 @@ export class PrimaryCancellationDispatcher { requestedAtMs: candidate.requestedAtMs, owner: this.owner, leaseToken: this.createId(), - nowMs: claimedAtMs, leaseDurationMs: this.leaseDurationMs, }); } catch { @@ -165,6 +160,11 @@ export class PrimaryCancellationDispatcher { summary.blocked += 1; continue; } + if (claim.status !== 'claimed') { + summary.failed += 1; + summary.pending += 1; + continue; + } summary.claimed += 1; await this.dispatchClaimed( @@ -172,6 +172,7 @@ export class PrimaryCancellationDispatcher { candidate.requestedAtMs, attempt, claim.dispatch, + claim.leaseToken, summary, ); } @@ -186,15 +187,28 @@ export class PrimaryCancellationDispatcher { Awaited>, { status: 'claimed' } >['dispatch'], + leaseToken: string, summary: PrimaryCancellationDispatchSummary, ): Promise { const controller = this.controllers.get(attempt.executorType); if (!controller) { - await this.record(attempt, dispatch, 'controller_missing', summary); + await this.record( + attempt, + dispatch, + leaseToken, + 'controller_missing', + summary, + ); return; } if (!attempt.executorHandle) { - await this.record(attempt, dispatch, 'handle_missing', summary); + await this.record( + attempt, + dispatch, + leaseToken, + 'handle_missing', + summary, + ); return; } @@ -207,7 +221,7 @@ export class PrimaryCancellationDispatcher { requestedAtMs, }, }); - await this.record(attempt, dispatch, result.status, summary); + await this.record(attempt, dispatch, leaseToken, result.status, summary); if (result.status === 'termination_requested') { summary.terminationRequested += 1; } else if (result.status === 'already_exited') { @@ -217,7 +231,13 @@ export class PrimaryCancellationDispatcher { } } catch { summary.failed += 1; - await this.record(attempt, dispatch, 'dispatch_error', summary); + await this.record( + attempt, + dispatch, + leaseToken, + 'dispatch_error', + summary, + ); } } @@ -227,10 +247,10 @@ export class PrimaryCancellationDispatcher { Awaited>, { status: 'claimed' } >['dispatch'], + leaseToken: string, result: CancellationDispatchResult, summary: PrimaryCancellationDispatchSummary, ): Promise { - const atMs = this.now(); const retryable = [ 'controller_missing', 'handle_missing', @@ -241,12 +261,11 @@ export class PrimaryCancellationDispatcher { runId: dispatch.runId, attemptId: attempt.attemptId, owner: this.owner, - leaseToken: dispatch.leaseToken!, + leaseToken, expectedVersion: dispatch.version, result, - atMs, ...(retryable - ? { nextAttemptAtMs: this.nextRetryAt(atMs, dispatch.dispatchCount) } + ? { retryDelayMs: this.nextRetryDelay(dispatch.dispatchCount) } : {}), eventId: this.createId(), }); @@ -257,17 +276,8 @@ export class PrimaryCancellationDispatcher { } } - private nextRetryAt(atMs: number, dispatchCount: number): number { + private nextRetryDelay(dispatchCount: number): number { const exponent = Math.max(0, Math.min(dispatchCount - 1, 30)); - const delay = Math.min(this.retryMaxMs, this.retryBaseMs * 2 ** exponent); - return Math.min(Number.MAX_SAFE_INTEGER, atMs + delay); - } - - private now(): number { - const nowMs = this.clock(); - if (!Number.isSafeInteger(nowMs) || nowMs < 0) { - throw new RangeError('clock must return a non-negative safe integer'); - } - return nowMs; + return Math.min(this.retryMaxMs, this.retryBaseMs * 2 ** exponent); } } diff --git a/back/runtime/domain/cancellationDispatch.ts b/back/runtime/domain/cancellationDispatch.ts index 10c39350..1c1fb64b 100644 --- a/back/runtime/domain/cancellationDispatch.ts +++ b/back/runtime/domain/cancellationDispatch.ts @@ -32,7 +32,7 @@ export interface CancellationDispatchRecord { dispatchCount: number; nextAttemptAtMs?: number; leaseOwner?: string; - leaseToken?: string; + leaseTokenDigest?: string; leaseExpiresAtMs?: number; lastResult?: CancellationDispatchResult; lastDispatchedAtMs?: number; diff --git a/back/runtime/ports/cancellationDispatchRepository.ts b/back/runtime/ports/cancellationDispatchRepository.ts index 6a54e959..b39b4c64 100644 --- a/back/runtime/ports/cancellationDispatchRepository.ts +++ b/back/runtime/ports/cancellationDispatchRepository.ts @@ -10,12 +10,15 @@ export interface ClaimCancellationDispatchCommand { requestedAtMs: number; owner: string; leaseToken: string; - nowMs: number; leaseDurationMs: number; } export type ClaimCancellationDispatchResult = - | { status: 'claimed'; dispatch: CancellationDispatchRecord } + | { + status: 'claimed'; + dispatch: CancellationDispatchRecord; + leaseToken: string; + } | { status: 'not_eligible' } | { status: 'not_due' | 'leased' | 'dispatched' | 'blocked'; @@ -29,8 +32,7 @@ export interface RecordCancellationDispatchResultCommand { leaseToken: string; expectedVersion: number; result: CancellationDispatchResult; - atMs: number; - nextAttemptAtMs?: number; + retryDelayMs?: number; eventId: string; } diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index 81ea31c4..53163f4f 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -11,6 +11,17 @@ 最新增量证据(2026-08-19): +- D-363/ADR-0456(已接受;Cluster 生产启动拓扑待接入):完成 profile-neutral CancellationDispatch canonical contract 与 PostgreSQL + `pg-0066-cancellation-dispatch`/capability v65 adapter。公共契约仅从显式子路径发布,调用方不提交当前时间、lease expiry 或绝对 retry timestamp; + PostgreSQL 以 `transaction_timestamp()` 作为 lease/retry authority,按 Run→Attempt→dispatch 固定锁序执行 claim/result。raw lease token 只随成功 claim + 返回,durable record、表、WAL 与事件只保存 domain-separated SHA-256 digest;结果事务原子完成 dispatch 更新、Run version CAS 与低敏 RunEvent,runtime + 角色对新表只有 SELECT/INSERT/UPDATE。新增能力收进 `run/cancellation-dispatch` 与 `run/migrations` 子域,没有扩大 workspace package 数、根入口、生产依赖、 + timer、连接、端口或 Kubernetes 对象;Local legacy adapter 复用 canonical outward record,但本阶段不冒充已迁移既有 SQLite raw-token 存量。完整 backend + `1,487 pass / 0 fail / 2 conditional skip`;18-package clean/build 退出 0,18-package 顺序测试单次退出 0;五项架构审计与 `14/14` artifact audit + 全部通过,基础 Edge/Standalone 仍为 `2,589,998 / 2,590,076` bytes 且不包含 PostgreSQL 闭包。PostgreSQL 18.6 arm64 HA 门 + `144/144`,覆盖双连接单 claim、数据库时间、digest-only token、租约接管、stale fence、retry due、事务回滚、WAL standby 与 promotion 后读取;timeline + `1→2`,报告 SHA-256 为 `b168b25023f7aad623153d22e41cccfe5f511a6985dc75c9e9e20073f980d5cb`,临时容器已清理。 + - D-362/ADR-0455(已接受;首次真实目标实例执行待运维):manual Primary 默认 bootstrap 在 router 安装后、`activated` 审计前原子发布 `qinglong/manual-primary-runtime-receipt@v1` 当前状态,并在清理前后推进 `active → stopping → stopped`;激活或停止失败收敛为 `failed`。receipt 固定为 config root 内一个 `0600`、8 KiB 上限、有 domain-separated 自摘要的 observed-state projection,绑定 Profile、manifest revision 与原始 SHA-256,但不成为第二 rollout authority。 @@ -9214,7 +9225,7 @@ flowchart LR | PR-2 Run 状态机 | Incubating | 纯转换表、终态/时间/错误/执行器元数据规则、Run version 与 event sequence CAS、事务性 RunCommandService、回滚测试 | 重复 Worker callback/fencing、并发数据库压力测试、Primary 执行链接入 | | PR-3 Executor 端口 | Incubating | ADR-0003、ExecutionSpec/Context/Handle/Result、Executor port、LocalProcessExecutor、进程组取消/超时升级、流式背压、Legacy Cron spec builder、真实进程 contract tests、可复现 edge 基准入口 | 固定 edge/多架构设备基线、Legacy builder 与 makeCommand 差异审计、Primary 生产流量接入 | | PR-4 Shadow Run | Incubating | origin 三态策略;默认关闭的 `QL3_SHADOW_ORIGINS`;manual、scheduled_node、boot、subscription、system 与 script 现有 ChildProcess 旁路观察;system crond 显式 origin marker、Shell execution ID、finish-only 准入、确定性 Run/Attempt 与 exact replay;`@once` 保持 manual、gRPC transport 不冒充 origin 的准入裁决;每个 worker 懒加载;Run/Attempt/Event 影子生命周期;稳定且不复制 caller 原文的 task identity/revision 与有界日志引用;同 worker 有界注册表和跨 worker 持久化候选关联;stop all/stop instance、Shell callback、乱序/迟到/歧义处理;监听前一次性、Profile-aware 的 keyset Startup Reconciler,终态证据补齐、lost/abandoned/pending 分流与 terminal Attempt response-loss 修复;origin-bounded 且逐级守恒的版本化 startup difference report、固定字段 metric batch 与一次性 collector;显式、只读、闭合窗口且 Profile-bounded 的 Shadow→Legacy 终态差异审计;128/256 MiB Linux arm64 资源门、SQLite 零增长与 Shadow enabled→off 进程重启回滚;process-epoch Legacy admission/capture/failure/pending 守恒;clean-shutdown `0600` no-replace capture+startup exporter;manual Edge 8/Standalone 32–128 canary;capture/terminal/resource 自包含 Primary bundle;rollout v2 loader 重算 source digest 与 eligibility;不可变 prepare/observe/resource/qualify 目标实例仪式、独立只读 audit;失败开放和契约测试 | 首次真实目标实例完整 canary 与 bootstrap activated 记录、其他 origin 独立 capture/Primary gate、固定物理 edge/flash/断电证据 | -| PR-5 Primary LocalExecutor | Incubating(默认不激活,仅 manifest-gated manual) | runtime-owned Run 创建器;持久化先于 spawn;Run/Attempt 完整成功、失败、取消、超时与 lost 闭环;Executor handle 身份校验;spawn 后激活写失败的 stop+lost 补偿;completion rejection 安全收敛;独立 Primary 幂等查询与唯一索引竞态裁决;durable `run.cancel_requested`、stop-before-signal、首次请求幂等、晚到完成裁决与待取消有界恢复查询;最多 64 条一页的 cross-worker cancellation source;独立 CancellationDispatch Repository 原子 claim/result、lease expiry 接管、owner/token/version fencing、指数退避与结果 RunEvent;最多 64 页的单周期 cancel supervisor;显式 start/stop、无重叠、错误隔离、停止等待有上限且 timer unref 的 lifecycle runner;Linux durable handle 的 PID/boot/start ticks/process-group 复验与 TERM/KILL controller;完整有界分页且 fail-closed 的 startup Reconcile supervisor;RunningInstance nullable `run_id/attempt_id` 关联;Primary 专用组合 Repository 在同一 SQLite 事务提交前投影 Crontab/RunningInstance,失败整体回滚;有界且防穿越的 legacy log output ref;manual owner seam、真实本机装配、单 spawn/fail-closed;严格 manual-only rollout manifest loader、短期审批/gate、配置哈希审计;HTTP worker 已接轻量 lazy bootstrap,accepted 后按 receipt-first reconcile→completion receipt lifecycle→timeout intent lifecycle→cancel dispatch lifecycle→router→durable active receipt 顺序激活,失败撤销,监听失败和 shutdown 以 stopping→有界清理→stopped/failed 失效;receipt 固定为单文件 observed-state projection,Linux 以 boot/PID/process-group/start ticks 复验,独立 auditor 支持 active 且 off/rolled-back 拒绝 live runtime;Primary timeout 在 spawn 前持久化绝对 deadline,有界 source/requester/supervisor 只提交 timeout 意图并复用 CancellationDispatch;代码级 edge/standalone Profile 为各 lifecycle 提供不同 cadence 与页上限,cluster-control/worker 拒绝误装本机 SQLite Primary;统一 CompletionService 原子提交 Attempt/Run/双 Event,spawn 前保存 callback token hash、终态推进 sequence,实时回调与 receipt consumer 共享入口并覆盖两个清理 crash window;manual Primary 已接入受限 POSIX launcher、`0600` direct-file stdout/stderr、父进程退出后续写、不可覆盖 receipt 生产、回执环境清除、TERM 转发等待及 live transaction 后清理;Startup Reconciler receipt-first 双检查并在确定 exited 后执行 profile 化的单次 50/100 ms publish grace;`0007` 独立 CompletionReceiptJournal 在 spawn 前登记、为升级前 active Attempt 补登记并驱动周期扫描,使终态残留继续可发现;确定无效的已知 Attempt receipt 先持久化隔离状态,再进入确定性私有分片 quarantine;终态 missing 与 quarantine 按 edge/standalone retention 有界清理;非 Journal 文件具备只读优先、固定分片/条目上限、overflow fail-closed、显式同盘隔离的 Node 24 运维 CLI;扫描具备页上限、resume cursor、timer unref、无重叠、有界 stop 和低敏计数;ENOSPC 与 launcher receipt 存储失败有代码门禁;显式最长 24 小时 approve 写入、`primary_selected` 只读状态、selection receipt、approval-expiry off 与 intent/completion crash-replay rollback | 首次真实目标实例完整激活/回滚仪式;PostgreSQL CancellationDispatch adapter;cluster-control 生产启动拓扑;固定 edge/Linux 多架构与真实磁盘压力基线、完整 2.x API 契约和回滚演练;共享 config 多写者 authority | +| PR-5 Primary LocalExecutor | Incubating(默认不激活,仅 manifest-gated manual) | runtime-owned Run 创建器;持久化先于 spawn;Run/Attempt 完整成功、失败、取消、超时与 lost 闭环;Executor handle 身份校验;spawn 后激活写失败的 stop+lost 补偿;completion rejection 安全收敛;独立 Primary 幂等查询与唯一索引竞态裁决;durable `run.cancel_requested`、stop-before-signal、首次请求幂等、晚到完成裁决与待取消有界恢复查询;最多 64 条一页的 cross-worker cancellation source;独立 CancellationDispatch Repository 原子 claim/result、lease expiry 接管、owner/token/version fencing、指数退避与结果 RunEvent;PostgreSQL `pg-0066`/capability v65 adapter 以数据库时间、Run→Attempt→dispatch 锁序、digest-only durable token、最小 runtime 权限和原子 RunEvent/Run version CAS 提供多副本同构实现,真实双连接与 HA promotion 门已通过;最多 64 页的单周期 cancel supervisor;显式 start/stop、无重叠、错误隔离、停止等待有上限且 timer unref 的 lifecycle runner;Linux durable handle 的 PID/boot/start ticks/process-group 复验与 TERM/KILL controller;完整有界分页且 fail-closed 的 startup Reconcile supervisor;RunningInstance nullable `run_id/attempt_id` 关联;Primary 专用组合 Repository 在同一 SQLite 事务提交前投影 Crontab/RunningInstance,失败整体回滚;有界且防穿越的 legacy log output ref;manual owner seam、真实本机装配、单 spawn/fail-closed;严格 manual-only rollout manifest loader、短期审批/gate、配置哈希审计;HTTP worker 已接轻量 lazy bootstrap,accepted 后按 receipt-first reconcile→completion receipt lifecycle→timeout intent lifecycle→cancel dispatch lifecycle→router→durable active receipt 顺序激活,失败撤销,监听失败和 shutdown 以 stopping→有界清理→stopped/failed 失效;receipt 固定为单文件 observed-state projection,Linux 以 boot/PID/process-group/start ticks 复验,独立 auditor 支持 active 且 off/rolled-back 拒绝 live runtime;Primary timeout 在 spawn 前持久化绝对 deadline,有界 source/requester/supervisor 只提交 timeout 意图并复用 CancellationDispatch;代码级 edge/standalone Profile 为各 lifecycle 提供不同 cadence 与页上限,cluster-control/worker 拒绝误装本机 SQLite Primary;统一 CompletionService 原子提交 Attempt/Run/双 Event,spawn 前保存 callback token hash、终态推进 sequence,实时回调与 receipt consumer 共享入口并覆盖两个清理 crash window;manual Primary 已接入受限 POSIX launcher、`0600` direct-file stdout/stderr、父进程退出后续写、不可覆盖 receipt 生产、回执环境清除、TERM 转发等待及 live transaction 后清理;Startup Reconciler receipt-first 双检查并在确定 exited 后执行 profile 化的单次 50/100 ms publish grace;`0007` 独立 CompletionReceiptJournal 在 spawn 前登记、为升级前 active Attempt 补登记并驱动周期扫描,使终态残留继续可发现;确定无效的已知 Attempt receipt 先持久化隔离状态,再进入确定性私有分片 quarantine;终态 missing 与 quarantine 按 edge/standalone retention 有界清理;非 Journal 文件具备只读优先、固定分片/条目上限、overflow fail-closed、显式同盘隔离的 Node 24 运维 CLI;扫描具备页上限、resume cursor、timer unref、无重叠、有界 stop 和低敏计数;ENOSPC 与 launcher receipt 存储失败有代码门禁;显式最长 24 小时 approve 写入、`primary_selected` 只读状态、selection receipt、approval-expiry off 与 intent/completion crash-replay rollback | 首次真实目标实例完整激活/回滚仪式;cluster-control 生产启动拓扑;固定 edge/Linux 多架构与真实磁盘压力基线、完整 2.x API 契约和回滚演练;共享 config 多写者 authority | | PR-7 Worker Session、Run Lease 与启动协议基础 | Incubating(默认关闭,独立入口显式 opt-in) | ADR-0012/0013/0014/0021/0057–0061/0108–0121/0231–0239/0377;有界 capability/Placement/Dispatcher;SQLite 协议孵化与 PostgreSQL v9 Session/Run Lease/credential/attestation authority;immutable revision Placement、数据库时钟 keyset candidate、认证 Worker Pull、digest-only offer recovery;versioned capability-free ExecutionSpec response、stable claim 跨重启退避、单 owner 原子 inbox 准入与 TLS 1.3 mTLS/`ql3w` HTTPS client;同一 package journal 上 revision-fenced starting/spawn/started/running/completion 状态、callback digest、tagged no-spawn 与 ambiguous recovery;PostgreSQL starting/running/start-failure/completion 数据库权威事务、精确重放与 cancellation/timeout 优先终态;batch Secret delivery 在 Attempt advisory lock 下复验 Session/Lease/revision 完整围栏并复用单 Agent,Secret-before-Artifact materializer 将同一 log ID 交给 Executor/journal/running ACK;offer-scoped `wlog-*` 私有文件 spool、Edge/Node 容量策略、append/quota/path 防护、barrier 后 output ownership、受审 POSIX Executor、truncation fact、固定内存流式 source、认证 Artifact stream、共享 immutable store port、S3-compatible SSE/checksum/条件 promotion adapter、upload-before-completion 协调,以及 Local/Cluster 同构、Profile-aware、ETag-fenced range read;用户取消 run.stop mutation 以数据库时间写 intent/Event 并在事务内复验 Project/RoleBinding fence;非执行取消 convergence lifecycle、运行期 expiry 与安全 lost retry 已接入 cluster-control 单一全局 cadence;完整 generation/version/token/Attempt fencing;独立最小权限 Worker ingress、CA/CRL 与连接 generation 热重载;offer journal、spawn barrier、receipt-first recovery;独立 `@qinglong/worker-runtime` 的本地 P-256 CSR、key/chain/trust 验证、generation + active pointer 安装和持久退避;默认关闭的 production process 已装配具体 execution graph、完整 Session heartbeat/drain/offline、direct-file bootstrap、单 Agent/单 cadence、startup reconciliation、证书 maintenance、transport fail-close/recovery 与 Edge/Node 有界预算;真实 PostgreSQL 18 + Linux Node 合约已覆盖 Run completion、credential 和 CA 双轮换且保持同一 Session;真实 K3s 合约已覆盖 TLS/credential Secret 分权、双对象 CAS、Recreate 顺序、identity generation 与单节点 PVC recovery;所有能力默认不可达且受 edge/cluster import audit 约束 | 具体 cert-manager/Vault/SPIFFE/离线 CA adapter 与模板、ingress reload controller、生产 RBAC、证书到期告警和 `ql3w` credential recovery 产品面;具体 KMS/Vault Secret provider、对象存储 credential/temporary lifecycle 与 retention/tombstone;Worker 管理 API;真实 Kubernetes 多节点 CSI/node-loss/production 360 秒 drain 与固定 edge 文件系统 suspend/时钟/断电、x64/arm64 资源门禁 | | PR-8 Project/Policy/Approval Core | Incubating(默认拒绝、无生产业务执行入口) | ADR-0028;统一六类 ActorRef 与 exact-shape 校验;`0017` ownerless default Project 和 append-only versioned RoleBinding;owner/admin/operator/viewer 固定矩阵;Project 内 mutation 幂等、expected-version CAS、双 SQLite 连接竞争门禁;archived read-only、revocation、存储损坏 fail-closed;Agent 写/Secret/Tool `require_approval`;ADR-0047 把六类 subject、role/permission matrix 与 fence 抽到 runtime-core,`pg-0004-project-policy`/capability v3 建立 ownerless PostgreSQL baseline、严格 role/state CHECK、append-only runtime 权限、SERIALIZABLE Project lock、mutation replay、双连接单 winner 和 cluster admission authorizer;ADR-0049/`pg-0005` capability v4 建立 stable IdentitySubject、append-only digest-only API credential、真实 cluster bearer authenticator、write-only durable security audit 与最小权限 runtime role,且已验证 HTTP→credential→Policy→audit→handler 纵向链路;ADR-0051 建立 `/api/v3` 认证前 peer/global 双预算、transport-peer-only、无 timer 且有界内存的 overload shield;ADR-0027 Artifact authorizer adapter;ADR-0029 `AuthenticatedPrincipal` contract、`0018` digest-only versioned challenge、CSPRNG/TTL、同事务消费 challenge + 写首 owner、精确重放与双连接竞争/崩溃回滚门禁;ADR-0030 `0019` stable identity/binding、legacy HS384 + current-session membership、logout/platform/revoke/disable、single-factor 与损坏 fail-closed 门禁;ADR-0031 `0020` digest-bound ApprovalRequest、User-only decision、Project/Role version fence、精确 expiry/重放/并发裁决及同事务 immutable dispatch;ADR-0032 `0021` execution backfill、三表原子 consume、稳定 due keyset、claim/renew/start/result fencing、pre-start takeover/post-start recovery-required、attempt budget、handler inspect/digest barrier 和 bounded dispatcher;ADR-0033/`0022` control/resolution backfill、start/renew/completion 原子联动、稳定 recovery keyset、双 resolver claim/takeover、finding/result 精确重放、自动/人工终结、迟到 completion 单 winner 和 evidence-only bounded reconciler;ADR-0034/`0023` 首个 `run.create` canonical plan、Run/Attempt/Event/receipt 同事务、幂等 collision fail-closed、renew/终态 fence、真实 SQLite handler 与 automatic evidence provider;ADR-0035/`0024` 独立 `approval.recover` 矩阵、稳定 User + 五分钟强认证、Project/RoleBinding fence、human resolution + authorization fact 原子提交、撤权竞态与回滚门禁;ADR-0036 recovery-first 单 timer lifecycle、edge/standalone 独立 cadence/页预算、跨周期 cursor、非重叠与有界 stop;ADR-0074 以新的 Node 24 SQLite v5 ownerless Project/RoleBinding/audit authority 和独立 local-secret-admin 提供强 Principal、`secret.manage`、撤权 fence、envelope+allowed audit 原子提交及不回显语义;ADR-0086 以可信 POSIX console 和 staged delivery 完成本机首 Owner 产品 ceremony | fresh database/pepper setup 与安全迁移向导;`shareStore`/Express 到 authentication core 的 production migration;credential rotation/revocation API、mTLS/Worker enrollment、恢复码;Project/Role/Approval/Secret 管理 CLI/API/UI、audit retention/query/export/alert、preview Artifact/digest/immutable plan builder、真实 MFA/hardware adapter、人工 recovery API/UI/独立 rate limit 与审计事件、handler/provider registry、lifecycle startup/shutdown/指标/admission gate;PostgreSQL action/receipt/provider/recovery-authorization 与 OPA adapter、缓存 version 失效;Tool/Package/Secret/Shell 各自的 handler/evidence contract;Secret/Run/Tool/Workflow waiting_approval 全入口装配;完整回滚演练 | diff --git a/docs/adr/ADR-0005-durable-cancellation-dispatch.md b/docs/adr/ADR-0005-durable-cancellation-dispatch.md index 4657a54e..fffe0f4f 100644 --- a/docs/adr/ADR-0005-durable-cancellation-dispatch.md +++ b/docs/adr/ADR-0005-durable-cancellation-dispatch.md @@ -1,6 +1,6 @@ # ADR-0005:Durable Cancellation Dispatch、Lease 与 Fencing -- 状态:Proposed +- 状态:Accepted(Local 与 PostgreSQL Repository 已实现;Cluster 生产启动拓扑待接入) - 日期:2026-07-18 - 决策范围:跨进程取消派发、崩溃恢复、并发 Worker、退避和审计事件 - 关联:QL-RFC-0001、ADR-0001、ADR-0003、ADR-0004 @@ -30,7 +30,7 @@ version non-negative integer dispatch_count non-negative integer next_attempt_at_ms nullable lease_owner nullable -lease_token nullable +lease_token_digest nullable lease_expires_at_ms nullable last_result nullable last_dispatched_at_ms nullable @@ -49,9 +49,9 @@ Worker 对唯一 active Attempt 执行原子 claim: 3. 不存在 dispatch 时创建 `pending`,绑定该 Attempt。 4. 既有记录绑定其他 Attempt 时 fail closed,不重新绑定或选择“最新 PID”。 5. `dispatched`、`blocked` 不再 claim;未到 `next_attempt_at_ms` 返回 not-due;未过期 lease 返回 leased。 -6. 到期或可派发时,以 version CAS 更新为 leased,递增 version 和 dispatch_count,写入新的 owner、不可预测 token 和 expiry。 +6. 到期或可派发时,以 version CAS 更新为 leased,递增 version 和 dispatch_count,写入新的 owner、domain-separated SHA-256 token digest 和 expiry;原始 token 只随成功 claim 返回给当前调用者,不进入 durable record。 -SQLite adapter 使用短 `IMMEDIATE` 事务串行化写竞争。PostgreSQL adapter 必须提供等价的行锁或条件更新语义;实现方式可以不同,行为契约不得改变。 +Repository 是租约、到期与退避的时间 authority。调用方只提交已有 `cancel_requested_at_ms` 事实、lease duration 或 retry delay,不得提交“当前时间”、lease expiry 或绝对 retry timestamp。SQLite adapter 使用注入 clock 与短 `IMMEDIATE` 事务串行化写竞争;PostgreSQL adapter 使用 `transaction_timestamp()`,按 Run→Attempt→CancellationDispatch 顺序取行锁。实现方式可以不同,行为契约不得改变。 ### 2.3 发出副作用 @@ -67,7 +67,7 @@ SQLite adapter 使用短 `IMMEDIATE` 事务串行化写竞争。PostgreSQL adapt ### 2.4 Result 事务与 fencing -结果提交必须同时匹配 run ID、attempt ID、lease owner、lease token 和 expected dispatch version。任何一项过期都拒绝写入。 +结果提交必须同时匹配 run ID、attempt ID、lease owner、原始 lease token 的 digest 和 expected dispatch version。任何一项过期都拒绝写入。 同一事务中: @@ -99,7 +99,7 @@ lease 到期后其他 Worker 可以重新 claim。重试仍必须执行完整身 ### 2.6 退避与 Supervisor -首版退避为 `min(max, base * 2^(dispatch_count-1))`,指数有上限。Repository 持久化绝对 `next_attempt_at_ms`,进程重启不会清空退避。 +首版退避为 `min(max, base * 2^(dispatch_count-1))`,指数有上限。调用方只提交有硬上限的 `retryDelayMs`,Repository 依据自身时间计算并持久化绝对 `next_attempt_at_ms`,进程重启不会清空退避。 Supervisor 一次只执行有界 cycle: @@ -151,15 +151,14 @@ PID 可复用,可能终止无关进程,禁止。 ## 5. 当前孵化边界 -`next` 已实现 `0005-run-cancellation-dispatch`、Repository 端口、临时 Sequelize/SQLite adapter、lease expiry 接管、fencing、退避、结果事件、Dispatcher、有界 Supervisor 和默认惰性的 lifecycle runner,并覆盖双 Worker、崩溃接管、事务回滚、无重叠调度和有界 shutdown 测试。 +`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。 -HTTP worker 已通过默认关闭的 manual-only manifest bootstrap 接入该 Supervisor:只有 accepted 且全部 gate 通过时才启动,失败或 shutdown 时有界停止。以下工作仍未完成,因此它仍只允许显式 canary,不得扩大到默认生产流量: +HTTP worker 已通过默认关闭的 manual-only manifest bootstrap 接入 Local Supervisor:只有 accepted 且全部 gate 通过时才启动,失败或 shutdown 时有界停止。以下工作仍未完成,因此它仍只允许显式 canary,不得扩大到默认生产流量: -- PostgreSQL adapter 与真实多连接并发压力测试。 -- ADR-0007 的 completion receipt、direct-file log、CompletionService 与周期 completion supervisor(timeout lifecycle 已接入)。 - 用户可见的运行指标、blocked 诊断和处置入口。 - 固定 edge 设备的数据库写放大、RSS、时延和磁盘基准。 -- 与 rollout manifest、回滚 runbook 和运维告警的最终接线。 +- cluster-control 对 PostgreSQL CancellationDispatch 的生产启动/停止拓扑与运维告警接线。 +- 首次真实目标实例完整激活/回滚仪式与共享 config 多写者 authority。 ## 6. 验证门禁 @@ -173,3 +172,5 @@ HTTP worker 已通过默认关闭的 manual-only manifest bootstrap 接入该 Su 8. identity/PID/process-group 不一致时零 signal。 9. page、cycle 和退避均有硬上限。 10. Event 与日志不包含 handle、命令、环境和 Secret。 +11. PostgreSQL 双连接只能产生一个 claim winner,raw token 不落库,数据库时间决定 lease/retry 到期。 +12. v65 事实经 WAL 到达 standby,提升为新 Primary 后仍可读取;旧 owner/token/version 继续被 fencing。 diff --git a/docs/adr/ADR-0456-database-timed-postgresql-cancellation-dispatch.md b/docs/adr/ADR-0456-database-timed-postgresql-cancellation-dispatch.md new file mode 100644 index 00000000..d801526c --- /dev/null +++ b/docs/adr/ADR-0456-database-timed-postgresql-cancellation-dispatch.md @@ -0,0 +1,64 @@ +# ADR-0456:数据库计时的 PostgreSQL CancellationDispatch + +- 状态:Accepted +- 日期:2026-08-19 +- 关联 RFC:QL-RFC-0001 D-363、PR-5 +- 关联 ADR:ADR-0001、ADR-0005、ADR-0041、ADR-0384 +- Amends:ADR-0005 的 PostgreSQL adapter、时间 authority 与 token 持久化边界 + +## 上下文 + +ADR-0005 已在 Local Profile 建立 durable cancellation dispatch,但 Cluster Profile 不能直接复用 SQLite 的单写者事务或进程时钟。多个 cluster-control 副本可能同时扫描同一 Run;节点时钟漂移会让租约提前接管或永久延后;把 raw lease token 持久化又会扩大数据库快照、备份与只读诊断面的能力泄漏。 + +QingLong 3.0 还必须同时服务低配路由和集群节点。公共协议需要同构,部署依赖与运行 authority 必须按 Profile 隔离:Edge 不应因 Cluster 能力引入 `pg`、连接池或常驻协调器,Cluster 也不能用进程内锁冒充多副本共识。 + +## 决策 + +1. `CancellationDispatch` 的 canonical contract 位于 `@qinglong/runtime-core/cancellation-dispatch` 显式子路径,不从 runtime-core 根入口导出。它定义 exact-shape command/record/result、硬上限、状态不变量、结果分类和 domain-separated SHA-256 token digest;不拥有数据库连接、timer、worker 或部署 Profile。 +2. claim command 只携带 Run/Attempt、已有的 `cancel_requested_at_ms`、owner、raw token 和有上限的 lease duration。result command只携带精确 fence、结果枚举、event ID,以及 retryable 结果所需的有上限 delay。调用方不得提交当前时间、lease expiry 或绝对 retry timestamp。 +3. PostgreSQL Repository 在事务中以 `transaction_timestamp()` 取得唯一时间事实。lease expiry、`updated_at_ms`、`last_dispatched_at_ms` 和 retry due time均由数据库时间计算;Local adapter 保持注入/default clock,以便低成本确定性测试和单设备运行。 +4. raw lease token 仅在成功 claim 的返回值中出现。durable record 与 PostgreSQL 表只保存 `sha256("qinglong.cancellation-dispatch-lease.v1\\0" || token)`;后续 result 在事务内重新计算 digest 比对。read/list、WAL、备份和诊断面不得恢复该 capability。 +5. `pg-0066-cancellation-dispatch` 把 PostgreSQL capability 提升到 v65,创建 `ql3.run_cancellation_dispatches`。主键为 Run ID,Attempt 通过 `(attempt_id, run_id)` 复合外键固定绑定同一 Run;CHECK 约束状态、counter、lease/retry/terminal shape,索引只支持 bounded due 与 expired-lease recovery。 +6. runtime role 对新表只有 SELECT、INSERT、UPDATE,没有 DELETE、TRUNCATE、REFERENCES、TRIGGER 或 schema create;migration owner 继续独占 DDL。readiness、Drizzle schema、reviewed SQL、migration checksum 与 catalog privilege 行必须保持锁步。 +7. claim 的锁序固定为 Run→Attempt→CancellationDispatch。先验证 runtime-owned active Run、精确 cancel timestamp 与同 Run active Attempt,再创建或锁定 dispatch;跨 Attempt 重绑定失败关闭。正常 lease 未过期、不 due、terminal 和 blocked 都不会产生新 owner。 +8. recordResult 使用同一锁序并精确验证 run/attempt/owner/token digest/expected version。在一个事务中更新 dispatch、对 Run version 做 CAS、分配 event sequence 并追加低敏 RunEvent;任一步失败全部回滚。stale fence 不能覆盖新 owner。 +9. PostgreSQL adapter 只通过 `@qinglong/cluster-postgres/cancellation-dispatch` 和受审 runtime entrypoint 发布,不从 package 根入口扩张。它不自动创建连接池、扫描器、timer、listener 或 cluster-control 进程;生产启动/停止拓扑是后续独立决策。 +10. Local legacy 表暂时保留既有列名与迁移兼容性,由 adapter 在 canonical record 边界投影 digest。D-363 不把这一点表述为 Local 数据库存量已经完成 raw-token 迁移;若要修改既有 SQLite durable layout,必须单独设计兼容迁移和回滚门。 + +## 被拒绝的替代方案 + +### 使用 cluster-control 进程时钟 + +拒绝。多副本时钟漂移会破坏 lease 与 retry 的单一到期语义,主库提升后也无法证明旧节点计算的绝对时间仍可信。 + +### 在表中保存 raw lease token + +拒绝。token 是一次短期写能力,不是诊断事实。持久化 raw capability 会让只读快照、复制链和备份获得不必要的可重放材料。 + +### 只用唯一索引或进程锁去重 + +拒绝。它们不能同时表达过期接管、owner fencing、结果原子事件与多副本崩溃恢复。 + +### 把 adapter 从包根入口导出并自动启动 + +拒绝。根入口扩张会污染轻量依赖闭包,自动启动会在未决的生产拓扑之前引入常驻扫描与连接 authority。 + +## 资源、安全与部署影响 + +- Edge/Standalone 基础产物不新增 `pg` 或 cluster package;最小 Edge artifact 仍约 2.59 MiB。 +- Cluster 新增一张当前状态表、两个恢复索引和短事务;无新 workspace package、生产依赖、Kubernetes 对象、端口、timer 或常驻进程。 +- raw lease token 不进入 canonical record、PostgreSQL row、WAL 或事件;事件仍只含 Attempt、dispatch count 与固定结果枚举。 +- 行锁顺序与 5 秒 statement timeout、1 秒 lock timeout、10 秒 idle-in-transaction timeout 共同限制锁等待;这不是无限并发压力证明,生产指标与容量门仍需完成。 + +## 验证 + +- runtime-core 契约 `5/5`,PostgreSQL schema/migration/readiness 聚焦 `75/75`;v65 checksum、CHECK/FK/index 与最小权限通过。 +- 完整 backend:`1,487 pass / 0 fail / 2 conditional skip`;18-package 最新 clean/build 退出 0,随后 18-package 顺序测试单次退出 0。 +- package boundary、Edge import、cluster dependency、cluster deployment、service-manager bridge 审计均零 finding;workspace package 精确为 18,新增实现位于明确子域而非 `src` 根平铺。 +- `14/14` Local Profile artifact audit 通过;基础 Edge/Standalone 为 `2,589,998 / 2,590,076` bytes,没有 PostgreSQL 依赖泄漏。 +- PostgreSQL 18.6 arm64 HA 门 `144/144`:双连接单 claim、数据库时钟、digest-only durable token、expired takeover、stale fence、retry due、事务回滚、WAL standby 可见和 promotion 后读取均通过;timeline `1→2`,报告 SHA-256 为 `b168b25023f7aad623153d22e41cccfe5f511a6985dc75c9e9e20073f980d5cb`。 +- HA 临时 Docker 容器在门结束后全部清理。该证据不冒充 CloudNativePG、多节点网络分区或生产容量证明。 + +## 后续 + +下一阶段把 PostgreSQL CancellationDispatch Repository 接入 cluster-control 的明确生产 composition、单一 cadence、availability withdrawal、shutdown drain、指标与 blocked 处置面;随后补 CloudNativePG live failover、多副本压力、固定 x64/arm64 资源门。Local 侧如需消除 legacy raw-token 存量,另开兼容迁移 ADR,不与 Cluster rollout 混合。 diff --git a/packages/ql3-cluster-control/test/application.test.cjs b/packages/ql3-cluster-control/test/application.test.cjs index 6d39eac6..a953cbc9 100644 --- a/packages/ql3-cluster-control/test/application.test.cjs +++ b/packages/ql3-cluster-control/test/application.test.cjs @@ -219,6 +219,7 @@ function runtimePrivileges() { plugin_package_workflow_task_attempt_admissions: [true, true, false, false], worker_execution_attestations: [true, false, false, false], run_events: [true, true, false, false], + run_cancellation_dispatches: [true, true, true, false], run_retry_policies: [true, true, true, false], }; return Object.entries(privileges).map( diff --git a/packages/ql3-cluster-control/test/bootstrap.test.cjs b/packages/ql3-cluster-control/test/bootstrap.test.cjs index 71ba6a32..aa98d8ce 100644 --- a/packages/ql3-cluster-control/test/bootstrap.test.cjs +++ b/packages/ql3-cluster-control/test/bootstrap.test.cjs @@ -133,6 +133,7 @@ function runtimePrivileges() { plugin_package_workflow_task_attempt_admissions: [true, true, false, false], worker_execution_attestations: [true, false, false, false], run_events: [true, true, false, false], + run_cancellation_dispatches: [true, true, true, false], run_retry_policies: [true, true, true, false], }; return Object.entries(privileges).map( diff --git a/packages/ql3-cluster-postgres/package.json b/packages/ql3-cluster-postgres/package.json index db98f27c..c4c06efe 100644 --- a/packages/ql3-cluster-postgres/package.json +++ b/packages/ql3-cluster-postgres/package.json @@ -85,6 +85,11 @@ "require": "./dist/entrypoints/runManager.js", "default": "./dist/entrypoints/runManager.js" }, + "./cancellation-dispatch": { + "types": "./dist/run/cancellationDispatchRepository.d.ts", + "require": "./dist/run/cancellationDispatchRepository.js", + "default": "./dist/run/cancellationDispatchRepository.js" + }, "./approval-manager": { "types": "./dist/approval-management/index.d.ts", "require": "./dist/approval-management/index.js", diff --git a/packages/ql3-cluster-postgres/src/entrypoints/runtime.ts b/packages/ql3-cluster-postgres/src/entrypoints/runtime.ts index 0b6d158f..6fe6009a 100644 --- a/packages/ql3-cluster-postgres/src/entrypoints/runtime.ts +++ b/packages/ql3-cluster-postgres/src/entrypoints/runtime.ts @@ -54,6 +54,7 @@ export { PostgresToolResultKeyCatalogReader } from '../tool-execution/toolResult export { PostgresToolResultRekeyReader } from '../tool-execution/toolResultRekeyRepository'; export * from '../run/runRepository'; +export * from '../run/cancellationDispatchRepository'; export * from '../run/runAttemptLogRetentionClaimRepository'; export * from '../security/projectPolicyRepository'; export * from '../security/apiCredentialRepository'; diff --git a/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts b/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts index 30295823..5613ddb8 100644 --- a/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts +++ b/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts @@ -333,5 +333,10 @@ export const postgresqlMainMigrationManifest: MigrationStreamManifest = checksum: '95387c5b40659490dbcb7626ecd15bacf6412360752bef88873bde57c43e0185', }), + Object.freeze({ + id: 'pg-0066-cancellation-dispatch', + checksum: + 'b6d7ac81b5f75530df05f8ef05878fa30aa0f4418363973ded89d14ffce151b2', + }), ]), }); diff --git a/packages/ql3-cluster-postgres/src/migrations/index.ts b/packages/ql3-cluster-postgres/src/migrations/index.ts index 1de87840..c32a0d88 100644 --- a/packages/ql3-cluster-postgres/src/migrations/index.ts +++ b/packages/ql3-cluster-postgres/src/migrations/index.ts @@ -68,6 +68,7 @@ import { pg0062PluginPackageSecretBindingTargetGuardMigration } from './pg-0062- import { pg0063PluginPackageSecretBindingTransitionReceiptsMigration } from './pg-0063-plugin-package-secret-binding-transition-receipts'; 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'; export const postgresqlMainMigrationStream: MigrationStreamDefinition = Object.freeze({ @@ -141,5 +142,6 @@ export const postgresqlMainMigrationStream: MigrationStreamDefinition; + +const ACTIVE_RUN_STATUSES = new Set([ + 'created', + 'queued', + 'dispatching', + 'running', + 'waiting_approval', + 'retry_wait', + 'lost', +]); +const ACTIVE_ATTEMPT_STATUSES = new Set(['claimed', 'starting', 'running']); +const CONTROLLER_NOT_INVOKED_RESULTS = new Set([ + 'controller_missing', + 'handle_missing', +]); + +function text(row: Row, key: string): string { + const value = row[key]; + if (typeof value !== 'string' || value.length < 1) { + throw new TypeError(`PostgreSQL cancellation dispatch ${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 dispatch ${key} is invalid`); + } + return value; +} + +function optionalText(row: Row, key: string): string | undefined { + return row[key] === null || row[key] === undefined + ? undefined + : text(row, key); +} + +function optionalInteger(row: Row, key: string): number | undefined { + return row[key] === null || row[key] === undefined + ? undefined + : integer(row, key); +} + +function dispatchFromRow(row: Row): Readonly { + const nextAttemptAtMs = optionalInteger(row, 'nextAttemptAtMs'); + const leaseOwner = optionalText(row, 'leaseOwner'); + const leaseTokenDigest = optionalText(row, 'leaseTokenDigest'); + const leaseExpiresAtMs = optionalInteger(row, 'leaseExpiresAtMs'); + const lastResult = optionalText(row, 'lastResult'); + const lastDispatchedAtMs = optionalInteger(row, 'lastDispatchedAtMs'); + return normalizeCancellationDispatchRecord({ + runId: text(row, 'runId'), + attemptId: text(row, 'attemptId'), + status: text(row, 'status') as CancellationDispatchRecord['status'], + version: integer(row, 'version'), + dispatchCount: integer(row, 'dispatchCount'), + ...(nextAttemptAtMs === undefined ? {} : { nextAttemptAtMs }), + ...(leaseOwner === undefined ? {} : { leaseOwner }), + ...(leaseTokenDigest === undefined ? {} : { leaseTokenDigest }), + ...(leaseExpiresAtMs === undefined ? {} : { leaseExpiresAtMs }), + ...(lastResult === undefined + ? {} + : { + lastResult: + lastResult as NonNullable, + }), + ...(lastDispatchedAtMs === undefined ? {} : { lastDispatchedAtMs }), + createdAtMs: integer(row, 'createdAtMs'), + updatedAtMs: integer(row, 'updatedAtMs'), + }); +} + +const DISPATCH_COLUMNS = ` + run_id AS "runId", attempt_id AS "attemptId", status, + version, dispatch_count AS "dispatchCount", + next_attempt_at_ms AS "nextAttemptAtMs", lease_owner AS "leaseOwner", + lease_token_digest AS "leaseTokenDigest", + lease_expires_at_ms AS "leaseExpiresAtMs", last_result AS "lastResult", + last_dispatched_at_ms AS "lastDispatchedAtMs", + created_at_ms AS "createdAtMs", updated_at_ms AS "updatedAtMs" +`; + +async function begin(client: PostgresClient): Promise { + await client.query('BEGIN'); + 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 databaseNow(client: PostgresClient): Promise { + const result = await client.query(` + SELECT floor(extract(epoch FROM transaction_timestamp()) * 1000)::bigint + AS "nowMs" + `); + if (result.rows.length !== 1) { + throw new TypeError('PostgreSQL cancellation dispatch clock is invalid'); + } + return integer(result.rows[0]!, 'nowMs'); +} + +async function rollback(client: PostgresClient): Promise { + try { + await client.query('ROLLBACK'); + } catch { + // Preserve the transaction failure. + } +} + +function repositoryFailure(error: unknown): never { + if (error instanceof CancellationDispatchError) throw error; + throw new CancellationDispatchRepositoryError(error); +} + +export class PostgresCancellationDispatchRepository + implements CancellationDispatchRepository +{ + constructor(private readonly pool: PostgresPool) { + if (!pool || typeof pool.connect !== 'function') { + throw new TypeError('PostgreSQL cancellation dispatch pool is invalid'); + } + } + + async findByRunId( + runId: string, + ): Promise | null> { + try { + const normalizedRunId = normalizeCancellationDispatchRunId(runId); + const result = await this.pool.query( + `SELECT ${DISPATCH_COLUMNS} + FROM "ql3"."run_cancellation_dispatches" + WHERE run_id = $1`, + [normalizedRunId], + ); + if (result.rows.length === 0) return null; + if (result.rows.length !== 1) { + throw new TypeError('PostgreSQL cancellation dispatch is not unique'); + } + return dispatchFromRow(result.rows[0]!); + } catch (error) { + return repositoryFailure(error); + } + } + + async claim( + value: Readonly, + ): Promise { + const command = normalizeClaimCancellationDispatchCommand(value); + return this.transaction(async (client) => { + const nowMs = await databaseNow(client); + const run = await client.query( + `SELECT execution_owner AS "executionOwner", status, + cancel_requested_at_ms AS "cancelRequestedAtMs" + FROM "ql3"."runs" WHERE id = $1 FOR UPDATE`, + [command.runId], + ); + if (run.rows.length > 1) { + throw new TypeError('PostgreSQL cancellation dispatch Run is invalid'); + } + const attempt = await client.query( + `SELECT run_id AS "runId", status + FROM "ql3"."run_attempts" WHERE id = $1 FOR UPDATE`, + [command.attemptId], + ); + if (attempt.rows.length > 1) { + throw new TypeError( + 'PostgreSQL cancellation dispatch Attempt is invalid', + ); + } + const runRow = run.rows[0]; + const attemptRow = attempt.rows[0]; + if ( + !runRow || + !attemptRow || + runRow.executionOwner !== 'runtime' || + !ACTIVE_RUN_STATUSES.has(runRow.status as string) || + optionalInteger(runRow, 'cancelRequestedAtMs') !== + command.requestedAtMs || + attemptRow.runId !== command.runId || + !ACTIVE_ATTEMPT_STATUSES.has(attemptRow.status as string) + ) { + return Object.freeze({ status: 'not_eligible' as const }); + } + + let dispatchResult = await client.query( + `SELECT ${DISPATCH_COLUMNS} + FROM "ql3"."run_cancellation_dispatches" + WHERE run_id = $1 FOR UPDATE`, + [command.runId], + ); + if (dispatchResult.rows.length === 0) { + dispatchResult = await client.query( + `INSERT INTO "ql3"."run_cancellation_dispatches" ( + run_id, attempt_id, status, version, dispatch_count, + next_attempt_at_ms, created_at_ms, updated_at_ms + ) VALUES ($1, $2, 'pending', 0, 0, $3, $4, $4) + RETURNING ${DISPATCH_COLUMNS}`, + [command.runId, command.attemptId, command.requestedAtMs, nowMs], + ); + } + if (dispatchResult.rows.length !== 1) { + throw new TypeError('PostgreSQL cancellation dispatch is invalid'); + } + const current = dispatchFromRow(dispatchResult.rows[0]!); + if (current.attemptId !== command.attemptId) { + throw new CancellationDispatchBindingConflictError( + command.runId, + command.attemptId, + ); + } + if (current.status === 'dispatched' || current.status === 'blocked') { + return Object.freeze({ status: current.status, dispatch: current }); + } + if ( + current.status === 'leased' && + current.leaseExpiresAtMs! > nowMs + ) { + return Object.freeze({ status: 'leased' as const, dispatch: current }); + } + if ( + current.status !== 'leased' && + current.nextAttemptAtMs! > nowMs + ) { + return Object.freeze({ status: 'not_due' as const, dispatch: current }); + } + if ( + current.version >= 2_147_483_647 || + current.dispatchCount >= 2_147_483_647 + ) { + throw new TypeError('PostgreSQL cancellation dispatch counter overflowed'); + } + const leaseExpiresAtMs = nowMs + command.leaseDurationMs; + if (!Number.isSafeInteger(leaseExpiresAtMs)) { + throw new TypeError('PostgreSQL cancellation dispatch lease overflowed'); + } + const leaseTokenDigest = digestCancellationDispatchLeaseToken( + command.leaseToken, + ); + const claimed = await client.query( + `UPDATE "ql3"."run_cancellation_dispatches" + SET status = 'leased', version = version + 1, + dispatch_count = dispatch_count + 1, + next_attempt_at_ms = NULL, lease_owner = $3, + lease_token_digest = $4, lease_expires_at_ms = $5, + updated_at_ms = $6 + WHERE run_id = $1 AND attempt_id = $2 AND version = $7 + RETURNING ${DISPATCH_COLUMNS}`, + [ + command.runId, + command.attemptId, + command.owner, + leaseTokenDigest, + leaseExpiresAtMs, + nowMs, + current.version, + ], + ); + if (claimed.rows.length !== 1) { + throw new CancellationDispatchFenceRejectedError(command.runId); + } + return Object.freeze({ + status: 'claimed' as const, + dispatch: dispatchFromRow(claimed.rows[0]!), + leaseToken: command.leaseToken, + }); + }); + } + + async recordResult( + value: Readonly, + ): Promise> { + const command = normalizeRecordCancellationDispatchResultCommand(value); + return this.transaction(async (client) => { + const atMs = await databaseNow(client); + const run = await client.query( + `SELECT version, event_sequence AS "eventSequence" + FROM "ql3"."runs" WHERE id = $1 FOR UPDATE`, + [command.runId], + ); + if (run.rows.length !== 1) { + throw new TypeError( + 'PostgreSQL cancellation dispatch Run disappeared', + ); + } + const dispatchResult = await client.query( + `SELECT ${DISPATCH_COLUMNS} + FROM "ql3"."run_cancellation_dispatches" + WHERE run_id = $1 FOR UPDATE`, + [command.runId], + ); + if (dispatchResult.rows.length !== 1) { + throw new CancellationDispatchFenceRejectedError(command.runId); + } + const current = dispatchFromRow(dispatchResult.rows[0]!); + if ( + current.attemptId !== command.attemptId || + current.status !== 'leased' || + current.version !== command.expectedVersion || + current.leaseOwner !== command.owner || + current.leaseTokenDigest !== + digestCancellationDispatchLeaseToken(command.leaseToken) + ) { + throw new CancellationDispatchFenceRejectedError(command.runId); + } + const runVersion = integer(run.rows[0]!, 'version'); + const eventSequence = integer(run.rows[0]!, 'eventSequence'); + if ( + current.version >= 2_147_483_647 || + runVersion >= 2_147_483_647 || + eventSequence >= 2_147_483_647 + ) { + throw new TypeError('PostgreSQL cancellation dispatch counter overflowed'); + } + const nextAttemptAtMs = + command.retryDelayMs === undefined + ? undefined + : atMs + command.retryDelayMs; + if ( + nextAttemptAtMs !== undefined && + !Number.isSafeInteger(nextAttemptAtMs) + ) { + throw new TypeError('PostgreSQL cancellation dispatch retry overflowed'); + } + const state = cancellationDispatchResultState(command.result); + const nextSequence = eventSequence + 1; + const runUpdated = await client.query( + `UPDATE "ql3"."runs" + SET version = version + 1, event_sequence = $2 + WHERE id = $1 AND version = $3`, + [command.runId, nextSequence, runVersion], + ); + if (runUpdated.rowCount !== 1) { + throw new CancellationDispatchFenceRejectedError(command.runId); + } + const dispatchUpdated = await client.query( + `UPDATE "ql3"."run_cancellation_dispatches" + SET status = $6, version = version + 1, + next_attempt_at_ms = $7, lease_owner = NULL, + lease_token_digest = NULL, lease_expires_at_ms = NULL, + last_result = $8, + last_dispatched_at_ms = CASE + WHEN $9::boolean THEN last_dispatched_at_ms ELSE $10 END, + updated_at_ms = $10 + WHERE run_id = $1 AND attempt_id = $2 AND status = 'leased' + AND version = $3 AND lease_owner = $4 + AND lease_token_digest = $5 + RETURNING ${DISPATCH_COLUMNS}`, + [ + command.runId, + command.attemptId, + command.expectedVersion, + command.owner, + current.leaseTokenDigest, + state.status, + nextAttemptAtMs ?? null, + command.result, + CONTROLLER_NOT_INVOKED_RESULTS.has(command.result), + atMs, + ], + ); + if (dispatchUpdated.rows.length !== 1) { + throw new CancellationDispatchFenceRejectedError(command.runId); + } + const event: Readonly = Object.freeze({ + id: command.eventId, + runId: command.runId, + sequence: nextSequence, + type: state.eventType, + dedupeKey: `cancel-dispatch:${command.attemptId}:${current.dispatchCount}`, + actorType: 'worker', + actorId: command.owner, + attemptId: command.attemptId, + payload: Object.freeze({ + attempt_id: command.attemptId, + dispatch_count: current.dispatchCount, + result: command.result, + }), + createdAtMs: atMs, + }); + 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, $4, $5, 'worker', $6, $7, NULL, $8::jsonb, $9)`, + [ + event.id, + event.runId, + event.sequence, + event.type, + event.dedupeKey, + event.actorId, + event.attemptId, + JSON.stringify(event.payload), + event.createdAtMs, + ], + ); + return Object.freeze({ + dispatch: dispatchFromRow(dispatchUpdated.rows[0]!), + event, + }); + }); + } + + private async transaction( + operation: (client: PostgresClient) => Promise, + ): Promise { + let client: PostgresClient | undefined; + try { + client = await this.pool.connect(); + await begin(client); + const result = await operation(client); + await client.query('COMMIT'); + return result; + } catch (error) { + if (client) await rollback(client); + return repositoryFailure(error); + } finally { + client?.release(); + } + } +} diff --git a/packages/ql3-cluster-postgres/src/run/migrations/pg-0066-cancellation-dispatch.ts b/packages/ql3-cluster-postgres/src/run/migrations/pg-0066-cancellation-dispatch.ts new file mode 100644 index 00000000..91840113 --- /dev/null +++ b/packages/ql3-cluster-postgres/src/run/migrations/pg-0066-cancellation-dispatch.ts @@ -0,0 +1,99 @@ +import { CAPABILITIES_V64 } from '../../approved-action/pg-0065-approved-action-manual-recovery'; +import { definePostgresSqlMigration } from '../../migrations/sqlMigration'; + +export const POSTGRESQL_CANCELLATION_DISPATCH_TABLE = + 'run_cancellation_dispatches'; + +export const CAPABILITIES_V65 = CAPABILITIES_V64.replace( + '"run_core":1,', + '"run_cancellation_dispatch":1,"run_core":1,', +); + +export const pg0066CancellationDispatchMigration = + definePostgresSqlMigration({ + id: 'pg-0066-cancellation-dispatch', + statements: [ + `CREATE UNIQUE INDEX ql3_run_attempts_run_id_uidx ON "ql3"."run_attempts" (run_id, id)`, + ` +CREATE TABLE "ql3"."${POSTGRESQL_CANCELLATION_DISPATCH_TABLE}" ( + run_id varchar(36) PRIMARY KEY, + attempt_id varchar(36) NOT NULL, + status varchar(32) NOT NULL, + version integer NOT NULL, + dispatch_count integer NOT NULL, + next_attempt_at_ms bigint, + lease_owner varchar(128), + lease_token_digest char(64), + lease_expires_at_ms bigint, + last_result varchar(32), + last_dispatched_at_ms bigint, + created_at_ms bigint NOT NULL, + updated_at_ms bigint NOT NULL, + CONSTRAINT ql3_run_cancellation_dispatch_run_fk + FOREIGN KEY (run_id) REFERENCES "ql3"."runs" (id) + ON DELETE CASCADE ON UPDATE RESTRICT, + CONSTRAINT ql3_run_cancellation_dispatch_attempt_fk + FOREIGN KEY (run_id, attempt_id) + REFERENCES "ql3"."run_attempts" (run_id, id) + ON DELETE CASCADE ON UPDATE RESTRICT, + CONSTRAINT ql3_run_cancellation_dispatch_status_check CHECK ( + status IN ('pending', 'leased', 'retry_wait', 'dispatched', 'blocked') + ), + CONSTRAINT ql3_run_cancellation_dispatch_result_check CHECK ( + last_result IS NULL OR last_result IN ( + 'termination_requested', 'already_exited', 'identity_mismatch', + 'pid_mismatch', 'unsupported', 'invalid', 'controller_missing', + 'handle_missing', 'dispatch_error' + ) + ), + CONSTRAINT ql3_run_cancellation_dispatch_counter_check CHECK ( + version BETWEEN 0 AND 2147483647 AND + dispatch_count BETWEEN 0 AND 2147483647 AND + version >= dispatch_count AND + ((status = 'pending' AND version = 0 AND dispatch_count = 0) OR + (status <> 'pending' AND dispatch_count >= 1)) + ), + CONSTRAINT ql3_run_cancellation_dispatch_time_check CHECK ( + (next_attempt_at_ms IS NULL OR next_attempt_at_ms >= 0) AND + (lease_expires_at_ms IS NULL OR lease_expires_at_ms >= 0) AND + (last_dispatched_at_ms IS NULL OR last_dispatched_at_ms >= 0) AND + created_at_ms >= 0 AND updated_at_ms >= created_at_ms + ), + CONSTRAINT ql3_run_cancellation_dispatch_lease_digest_check CHECK ( + lease_token_digest IS NULL OR lease_token_digest ~ '^[0-9a-f]{64}$' + ), + CONSTRAINT ql3_run_cancellation_dispatch_shape_check CHECK ( + (status = 'leased' AND next_attempt_at_ms IS NULL AND + lease_owner IS NOT NULL AND + octet_length(lease_owner) BETWEEN 1 AND 128 AND + lease_owner !~ '[[:cntrl:]]' AND + lease_token_digest IS NOT NULL AND lease_expires_at_ms IS NOT NULL) OR + (status IN ('pending', 'retry_wait') AND next_attempt_at_ms IS NOT NULL AND + lease_owner IS NULL AND lease_token_digest IS NULL AND + lease_expires_at_ms IS NULL) OR + (status IN ('dispatched', 'blocked') AND next_attempt_at_ms IS NULL AND + lease_owner IS NULL AND lease_token_digest IS NULL AND + lease_expires_at_ms IS NULL AND last_result IS NOT NULL) + ), + 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 IS NULL OR last_result IN ( + 'controller_missing', 'handle_missing', 'dispatch_error' + ))) 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(), + `CREATE INDEX ql3_run_cancellation_dispatch_due_idx ON "ql3"."${POSTGRESQL_CANCELLATION_DISPATCH_TABLE}" (next_attempt_at_ms, run_id) WHERE status IN ('pending', 'retry_wait')`, + `CREATE INDEX ql3_run_cancellation_dispatch_lease_expiry_idx ON "ql3"."${POSTGRESQL_CANCELLATION_DISPATCH_TABLE}" (lease_expires_at_ms, run_id) WHERE status = 'leased'`, + `REVOKE ALL ON "ql3"."${POSTGRESQL_CANCELLATION_DISPATCH_TABLE}" FROM PUBLIC, ql3_runtime, ql3_admin, ql3_package_manager, ql3_package_executor, ql3_worker_ingress, ql3_worker_credential_manager, ql3_worker_credential_executor, ql3_automation_manager, ql3_approval_manager, ql3_run_manager`, + `GRANT SELECT, INSERT, UPDATE ON "ql3"."${POSTGRESQL_CANCELLATION_DISPATCH_TABLE}" TO ql3_runtime`, + `DO $ql3$ BEGIN UPDATE "ql3"."schema_capabilities" SET contract_version = 65, migration_id = 'pg-0066-cancellation-dispatch', capabilities = '${CAPABILITIES_V65}'::jsonb, updated_at_ms = floor(extract(epoch FROM transaction_timestamp()) * 1000)::bigint WHERE contract_name = 'control-core' AND contract_version = 64 AND migration_id = 'pg-0065-approved-action-manual-recovery' AND capabilities = '${CAPABILITIES_V64}'::jsonb; IF NOT FOUND THEN RAISE EXCEPTION 'control-core capability is not at version 64' USING ERRCODE = 'check_violation'; END IF; END $ql3$`, + ], + }); diff --git a/packages/ql3-cluster-postgres/src/schema/schema.ts b/packages/ql3-cluster-postgres/src/schema/schema.ts index a376472d..dee52227 100644 --- a/packages/ql3-cluster-postgres/src/schema/schema.ts +++ b/packages/ql3-cluster-postgres/src/schema/schema.ts @@ -4810,6 +4810,7 @@ export const runAttempts = ql3Schema.table( table.runId, table.attempt, ), + uniqueIndex('ql3_run_attempts_run_id_uidx').on(table.runId, table.id), index('ql3_run_attempts_dispatch_candidates_idx').on( table.status, table.runId, @@ -4830,6 +4831,75 @@ export const runAttempts = ql3Schema.table( ], ); +export const runCancellationDispatches = ql3Schema.table( + 'run_cancellation_dispatches', + { + runId: varchar('run_id', { length: 36 }).primaryKey(), + attemptId: varchar('attempt_id', { length: 36 }).notNull(), + status: varchar('status', { length: 32 }).notNull(), + version: integer('version').notNull(), + dispatchCount: integer('dispatch_count').notNull(), + nextAttemptAtMs: bigint('next_attempt_at_ms', { mode: 'number' }), + leaseOwner: varchar('lease_owner', { length: 128 }), + leaseTokenDigest: char('lease_token_digest', { length: 64 }), + leaseExpiresAtMs: bigint('lease_expires_at_ms', { mode: 'number' }), + lastResult: varchar('last_result', { length: 32 }), + lastDispatchedAtMs: bigint('last_dispatched_at_ms', { mode: 'number' }), + createdAtMs: bigint('created_at_ms', { mode: 'number' }).notNull(), + updatedAtMs: bigint('updated_at_ms', { mode: 'number' }).notNull(), + }, + (table) => [ + foreignKey({ + name: 'ql3_run_cancellation_dispatch_run_fk', + columns: [table.runId], + foreignColumns: [runs.id], + }) + .onDelete('cascade') + .onUpdate('restrict'), + foreignKey({ + name: 'ql3_run_cancellation_dispatch_attempt_fk', + columns: [table.runId, table.attemptId], + foreignColumns: [runAttempts.runId, runAttempts.id], + }) + .onDelete('cascade') + .onUpdate('restrict'), + check( + 'ql3_run_cancellation_dispatch_status_check', + sql`${table.status} in ('pending', 'leased', 'retry_wait', 'dispatched', 'blocked')`, + ), + check( + 'ql3_run_cancellation_dispatch_result_check', + sql`${table.lastResult} is null or ${table.lastResult} in ('termination_requested', 'already_exited', 'identity_mismatch', 'pid_mismatch', 'unsupported', 'invalid', 'controller_missing', 'handle_missing', 'dispatch_error')`, + ), + check( + 'ql3_run_cancellation_dispatch_counter_check', + sql`${table.version} between 0 and 2147483647 and ${table.dispatchCount} between 0 and 2147483647 and ${table.version} >= ${table.dispatchCount} and ((${table.status} = 'pending' and ${table.version} = 0 and ${table.dispatchCount} = 0) or (${table.status} <> 'pending' and ${table.dispatchCount} >= 1))`, + ), + check( + 'ql3_run_cancellation_dispatch_time_check', + sql`(${table.nextAttemptAtMs} is null or ${table.nextAttemptAtMs} >= 0) and (${table.leaseExpiresAtMs} is null or ${table.leaseExpiresAtMs} >= 0) and (${table.lastDispatchedAtMs} is null or ${table.lastDispatchedAtMs} >= 0) and ${table.createdAtMs} >= 0 and ${table.updatedAtMs} >= ${table.createdAtMs}`, + ), + check( + 'ql3_run_cancellation_dispatch_lease_digest_check', + sql`${table.leaseTokenDigest} is null or ${table.leaseTokenDigest} ~ '^[0-9a-f]{64}$'`, + ), + check( + 'ql3_run_cancellation_dispatch_shape_check', + sql`(${table.status} = 'leased' and ${table.nextAttemptAtMs} is null and ${table.leaseOwner} is not null and octet_length(${table.leaseOwner}) between 1 and 128 and ${table.leaseOwner} !~ '[[:cntrl:]]' and ${table.leaseTokenDigest} is not null and ${table.leaseExpiresAtMs} is not null) or (${table.status} in ('pending', 'retry_wait') and ${table.nextAttemptAtMs} is not null and ${table.leaseOwner} is null and ${table.leaseTokenDigest} is null and ${table.leaseExpiresAtMs} is null) or (${table.status} in ('dispatched', 'blocked') and ${table.nextAttemptAtMs} is null and ${table.leaseOwner} is null and ${table.leaseTokenDigest} is null and ${table.leaseExpiresAtMs} is null and ${table.lastResult} is not null)`, + ), + check( + 'ql3_run_cancellation_dispatch_result_state_check', + sql`(${table.status} = 'pending' and ${table.lastResult} is null) or (${table.status} in ('leased', 'retry_wait') and (${table.lastResult} is null or ${table.lastResult} in ('controller_missing', 'handle_missing', 'dispatch_error'))) or (${table.status} = 'dispatched' and ${table.lastResult} in ('termination_requested', 'already_exited')) or (${table.status} = 'blocked' and ${table.lastResult} in ('identity_mismatch', 'pid_mismatch', 'unsupported', 'invalid'))`, + ), + index('ql3_run_cancellation_dispatch_due_idx') + .on(table.nextAttemptAtMs, table.runId) + .where(sql`${table.status} in ('pending', 'retry_wait')`), + index('ql3_run_cancellation_dispatch_lease_expiry_idx') + .on(table.leaseExpiresAtMs, table.runId) + .where(sql`${table.status} = 'leased'`), + ], +); + export const runAttemptLogRetentionControls = ql3Schema.table( 'run_attempt_log_retention_controls', { @@ -6211,6 +6281,7 @@ export const ql3PostgresTables = [ toolExecutionResultRekeyHeads, toolResultKeyRetirementReceipts, runAttempts, + runCancellationDispatches, runAttemptLogRetentionControls, runAttemptLogArtifactTombstones, workerSessions, diff --git a/packages/ql3-cluster-postgres/src/schema/schemaContract.ts b/packages/ql3-cluster-postgres/src/schema/schemaContract.ts index 248441c8..734594e2 100644 --- a/packages/ql3-cluster-postgres/src/schema/schemaContract.ts +++ b/packages/ql3-cluster-postgres/src/schema/schemaContract.ts @@ -21,12 +21,13 @@ export interface PostgresSchemaContractTrigger { export interface PostgresSchemaContract { readonly schema: 'ql3'; readonly contractName: 'control-core'; - readonly contractVersion: 64; - readonly migrationId: 'pg-0065-approved-action-manual-recovery'; + readonly contractVersion: 65; + readonly migrationId: 'pg-0066-cancellation-dispatch'; readonly minimumServerMajor: 16; readonly maximumServerMajor: 18; readonly capabilities: Readonly<{ run_core: 1; + run_cancellation_dispatch: 1; run_attempt_log_retention: 1; run_management_boundary: 1; run_management_stop: 1; @@ -118,8 +119,8 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = Object.freeze({ schema: 'ql3', contractName: 'control-core', - contractVersion: 64, - migrationId: 'pg-0065-approved-action-manual-recovery', + contractVersion: 65, + migrationId: 'pg-0066-cancellation-dispatch', minimumServerMajor: 16, maximumServerMajor: 18, capabilities: Object.freeze({ @@ -167,6 +168,7 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = project_policy: 1, project_tool_definition_snapshot: 1, run_core: 1, + run_cancellation_dispatch: 1, run_attempt_log_retention: 1, run_management_boundary: 1, run_management_stop: 1, @@ -1289,6 +1291,21 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = 'error_code', 'error_summary', ]), + table('run_cancellation_dispatches', [ + 'run_id', + 'attempt_id', + 'status', + 'version', + 'dispatch_count', + 'next_attempt_at_ms', + 'lease_owner', + 'lease_token_digest', + 'lease_expires_at_ms', + 'last_result', + 'last_dispatched_at_ms', + 'created_at_ms', + 'updated_at_ms', + ]), table('run_attempt_log_retention_controls', [ 'attempt_id', 'project_id', @@ -1770,9 +1787,13 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = 'ql3_result_retirement_catalog_idx', 'run_attempts_pkey', 'ql3_run_attempts_run_attempt_uidx', + 'ql3_run_attempts_run_id_uidx', 'ql3_run_attempts_dispatch_candidates_idx', 'ql3_run_attempts_recovery_idx', 'ql3_run_attempts_lease_idx', + 'run_cancellation_dispatches_pkey', + 'ql3_run_cancellation_dispatch_due_idx', + 'ql3_run_cancellation_dispatch_lease_expiry_idx', 'run_attempt_log_retention_controls_pkey', 'ql3_run_log_retention_control_artifact_key', 'ql3_run_log_retention_retry_idx', @@ -2106,6 +2127,13 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = 'ql3_runs_cancel_reason_check', 'ql3_run_attempts_attempt_check', 'ql3_run_attempts_status_check', + 'ql3_run_cancellation_dispatch_status_check', + 'ql3_run_cancellation_dispatch_result_check', + 'ql3_run_cancellation_dispatch_counter_check', + 'ql3_run_cancellation_dispatch_time_check', + 'ql3_run_cancellation_dispatch_lease_digest_check', + 'ql3_run_cancellation_dispatch_shape_check', + 'ql3_run_cancellation_dispatch_result_state_check', 'ql3_run_attempts_pid_check', 'ql3_run_attempts_lease_expiry_check', 'ql3_run_attempts_worker_session_id_check', @@ -2445,6 +2473,8 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = 'ql3_result_rekey_head_overlay_fk', 'ql3_result_retirement_catalog_fk', 'ql3_run_attempts_run_fk', + 'ql3_run_cancellation_dispatch_run_fk', + 'ql3_run_cancellation_dispatch_attempt_fk', 'ql3_run_attempts_step_run_fk', 'ql3_run_log_retention_control_attempt_fk', 'ql3_run_log_retention_control_run_fk', diff --git a/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts b/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts index c75bf708..e16edce4 100644 --- a/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts +++ b/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts @@ -588,6 +588,12 @@ const REQUIRED_RUNTIME_PRIVILEGES = Object.freeze({ update: true, delete: false, }), + run_cancellation_dispatches: Object.freeze({ + select: true, + insert: true, + update: true, + delete: false, + }), run_attempt_log_retention_controls: Object.freeze({ select: true, insert: true, @@ -1131,6 +1137,12 @@ const REQUIRED_ADMIN_PRIVILEGES = Object.freeze({ update: false, delete: false, }), + run_cancellation_dispatches: Object.freeze({ + select: false, + insert: false, + update: false, + delete: false, + }), run_attempt_log_retention_controls: Object.freeze({ select: false, insert: false, diff --git a/packages/ql3-cluster-postgres/test/postgres.integration.test.cjs b/packages/ql3-cluster-postgres/test/postgres.integration.test.cjs index 88811a45..9fe497c8 100644 --- a/packages/ql3-cluster-postgres/test/postgres.integration.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgres.integration.test.cjs @@ -33,6 +33,7 @@ const { PostgresClusterControlRecoveryResolutionRepository, PostgresClusterControlRecoverySource, PostgresClusterRunCancellationConvergenceRepository, + PostgresCancellationDispatchRepository, PostgresClusterScheduleRepository, PostgresProjectPolicyRepository, PostgresRunRepository, @@ -41,6 +42,11 @@ const { PostgresWorkerSessionRepository, PostgresRemoteWorkerAttestationEvidenceProvider, } = require('../dist/entrypoints/runtime'); +const { + CancellationDispatchBindingConflictError, + CancellationDispatchFenceRejectedError, + digestCancellationDispatchLeaseToken, +} = require('@qinglong/runtime-core/cancellation-dispatch'); const { PostgresTaskDefinitionRepository, PostgresTriggerRepository, @@ -789,6 +795,251 @@ if (!migrationConnectionString) { }, }); + test('PostgreSQL cancellation dispatch fences replicas with database time and atomic events', async () => { + const runId = '019f7300-0000-7000-8000-000000000901'; + const attemptId = '019f7300-0000-7000-8000-000000000902'; + const secondAttemptId = '019f7300-0000-7000-8000-000000000903'; + const duplicateEventId = '019f7300-0000-7000-8000-000000000904'; + const retryEventId = '019f7300-0000-7000-8000-000000000905'; + const terminalEventId = '019f7300-0000-7000-8000-000000000906'; + const requestedAtMs = 1_750_000_000_100; + const migrationDatabase = await open('migration'); + let firstDatabase; + let secondDatabase; + try { + await runPostgresMigrations({ pool: migrationDatabase.pool }); + await migrationDatabase.pool.query( + 'TRUNCATE TABLE "ql3"."run_events", "ql3"."run_retry_policies", "ql3"."run_attempts", "ql3"."runs" CASCADE', + ); + const before = await migrationDatabase.pool.query( + `SELECT floor(extract(epoch FROM clock_timestamp()) * 1000)::bigint + AS "nowMs"`, + ); + await migrationDatabase.pool.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', 'cancellation-integration', 'v1', 'manual', + 'api', 'runtime', 'running', 2, 0, $2, $2, $3, 'user' + )`, + [runId, requestedAtMs - 100, requestedAtMs], + ); + await migrationDatabase.pool.query( + `INSERT INTO "ql3"."run_attempts" ( + id, run_id, attempt, status, executor_type, callback_sequence, + created_at_ms + ) VALUES ($1, $2, 1, 'running', 'local_process', 0, $3)`, + [attemptId, runId, requestedAtMs - 50], + ); + + [firstDatabase, secondDatabase] = await Promise.all([ + open('runtime'), + open('runtime'), + ]); + const firstRepository = new PostgresCancellationDispatchRepository( + firstDatabase.pool, + ); + const secondRepository = new PostgresCancellationDispatchRepository( + secondDatabase.pool, + ); + const candidate = { + runId, + attemptId, + requestedAtMs, + leaseDurationMs: 10_000, + }; + const [firstClaim, secondClaim] = await Promise.all([ + firstRepository.claim({ + ...candidate, + owner: 'primary-a', + leaseToken: 'lease-a', + }), + secondRepository.claim({ + ...candidate, + owner: 'primary-b', + leaseToken: 'lease-b', + }), + ]); + const claimed = [firstClaim, secondClaim].find( + (result) => result.status === 'claimed', + ); + const competing = [firstClaim, secondClaim].find( + (result) => result.status !== 'claimed', + ); + assert.equal(claimed?.status, 'claimed'); + assert.equal(competing?.status, 'leased'); + assert.equal(claimed.dispatch.version, 1); + assert.equal(claimed.dispatch.dispatchCount, 1); + assert.equal(claimed.dispatch.createdAtMs >= Number(before.rows[0].nowMs), true); + const rawLeaseToken = claimed.leaseToken; + const stored = await migrationDatabase.pool.query( + `SELECT lease_token_digest AS "leaseTokenDigest", + lease_owner AS "leaseOwner", version, dispatch_count + AS "dispatchCount" + FROM "ql3"."run_cancellation_dispatches" WHERE run_id = $1`, + [runId], + ); + assert.equal( + stored.rows[0].leaseTokenDigest, + digestCancellationDispatchLeaseToken(rawLeaseToken), + ); + assert.notEqual(stored.rows[0].leaseTokenDigest, rawLeaseToken); + + await migrationDatabase.pool.query( + `UPDATE "ql3"."run_cancellation_dispatches" + SET lease_expires_at_ms = 0 WHERE run_id = $1`, + [runId], + ); + const takeover = await secondRepository.claim({ + ...candidate, + owner: 'primary-takeover', + leaseToken: 'lease-takeover', + }); + assert.equal(takeover.status, 'claimed'); + assert.equal(takeover.dispatch.version, 2); + assert.equal(takeover.dispatch.dispatchCount, 2); + await assert.rejects( + firstRepository.recordResult({ + runId, + attemptId, + owner: claimed.dispatch.leaseOwner, + leaseToken: rawLeaseToken, + expectedVersion: claimed.dispatch.version, + result: 'already_exited', + eventId: terminalEventId, + }), + CancellationDispatchFenceRejectedError, + ); + + await migrationDatabase.pool.query( + `INSERT INTO "ql3"."run_events" ( + id, run_id, sequence, type, dedupe_key, actor_type, payload, + created_at_ms + ) VALUES ($1, $2, 99, 'fixture.event', 'fixture-event', 'system', + '{}'::jsonb, $3)`, + [duplicateEventId, runId, requestedAtMs], + ); + await assert.rejects( + secondRepository.recordResult({ + runId, + attemptId, + owner: 'primary-takeover', + leaseToken: 'lease-takeover', + expectedVersion: takeover.dispatch.version, + result: 'dispatch_error', + retryDelayMs: 1_000, + eventId: duplicateEventId, + }), + ); + const rolledBack = await migrationDatabase.pool.query( + `SELECT dispatch.status, dispatch.version, run.version AS "runVersion", + run.event_sequence AS "eventSequence" + FROM "ql3"."run_cancellation_dispatches" dispatch + JOIN "ql3"."runs" run ON run.id = dispatch.run_id + WHERE dispatch.run_id = $1`, + [runId], + ); + assert.deepEqual(rolledBack.rows, [ + { status: 'leased', version: 2, runVersion: 2, eventSequence: 0 }, + ]); + + const retry = await secondRepository.recordResult({ + runId, + attemptId, + owner: 'primary-takeover', + leaseToken: 'lease-takeover', + expectedVersion: takeover.dispatch.version, + result: 'dispatch_error', + retryDelayMs: 60_000, + eventId: retryEventId, + }); + assert.equal(retry.dispatch.status, 'retry_wait'); + assert.equal(retry.event.type, 'run.cancel_dispatch_failed'); + assert.equal( + (await firstRepository.claim({ + ...candidate, + owner: 'primary-a', + leaseToken: 'lease-a-retry', + })).status, + 'not_due', + ); + await migrationDatabase.pool.query( + `UPDATE "ql3"."run_cancellation_dispatches" + SET next_attempt_at_ms = 0 WHERE run_id = $1`, + [runId], + ); + const finalLease = await firstRepository.claim({ + ...candidate, + owner: 'primary-final', + leaseToken: 'lease-final', + }); + assert.equal(finalLease.status, 'claimed'); + assert.equal(finalLease.dispatch.dispatchCount, 3); + + await migrationDatabase.pool.query( + `INSERT INTO "ql3"."run_attempts" ( + id, run_id, attempt, status, executor_type, callback_sequence, + created_at_ms + ) VALUES ($1, $2, 2, 'running', 'local_process', 0, $3)`, + [secondAttemptId, runId, requestedAtMs], + ); + await assert.rejects( + secondRepository.claim({ + ...candidate, + attemptId: secondAttemptId, + owner: 'primary-conflict', + leaseToken: 'lease-conflict', + }), + CancellationDispatchBindingConflictError, + ); + + const terminal = await firstRepository.recordResult({ + runId, + attemptId, + owner: 'primary-final', + leaseToken: 'lease-final', + expectedVersion: finalLease.dispatch.version, + result: 'already_exited', + eventId: terminalEventId, + }); + assert.equal(terminal.dispatch.status, 'dispatched'); + assert.equal(terminal.event.sequence, 2); + assert.deepEqual(terminal.event.payload, { + attempt_id: attemptId, + dispatch_count: 3, + result: 'already_exited', + }); + const durable = await migrationDatabase.pool.query( + `SELECT dispatch.status, dispatch.lease_token_digest AS "leaseDigest", + dispatch.dispatch_count AS "dispatchCount", + run.version AS "runVersion", + run.event_sequence AS "eventSequence" + FROM "ql3"."run_cancellation_dispatches" dispatch + JOIN "ql3"."runs" run ON run.id = dispatch.run_id + WHERE dispatch.run_id = $1`, + [runId], + ); + assert.deepEqual(durable.rows, [ + { + status: 'dispatched', + leaseDigest: null, + dispatchCount: 3, + runVersion: 4, + eventSequence: 2, + }, + ]); + } finally { + await Promise.allSettled([ + firstDatabase?.close(), + secondDatabase?.close(), + ]); + await migrationDatabase.close(); + } + }); + test('PostgreSQL Task Start atomically persists and exactly replays one Run aggregate', async () => { const projectId = 'task-start-integration'; const taskId = 'task-start-command'; diff --git a/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs b/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs index 7130bb77..f6227871 100644 --- a/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs @@ -116,6 +116,7 @@ test('defines the immutable PostgreSQL capability and Run core stream', async () 'pg-0063-plugin-package-secret-binding-transition-receipts', 'pg-0064-plugin-package-secret-binding-transition-approval-plans', 'pg-0065-approved-action-manual-recovery', + 'pg-0066-cancellation-dispatch', ], ); for (const migration of postgresqlMainMigrationStream.migrations) { @@ -579,6 +580,11 @@ test('freezes every published PostgreSQL migration checksum', () => { checksum: '95387c5b40659490dbcb7626ecd15bacf6412360752bef88873bde57c43e0185', }, + { + id: 'pg-0066-cancellation-dispatch', + checksum: + 'b6d7ac81b5f75530df05f8ef05878fa30aa0f4418363973ded89d14ffce151b2', + }, ]; assert.deepEqual( postgresqlMainMigrationStream.migrations.map(({ id, checksum }) => ({ @@ -2282,3 +2288,44 @@ test('advances capability v64 with atomic least-privilege manual recovery', asyn /migration_id = 'pg-0064-plugin-package-secret-binding-transition-approval-plans'/, ); }); + +test('advances capability v65 with database-timed fenced cancellation dispatch', async () => { + const migration = migrationById('pg-0066-cancellation-dispatch'); + const statements = []; + await migration.up({ + async query(statement) { + statements.push(statement); + return { rows: [] }; + }, + }); + const sql = statements.join('\n'); + assert.match( + sql, + /CREATE UNIQUE INDEX ql3_run_attempts_run_id_uidx ON "ql3"\."run_attempts" \(run_id, id\)/, + ); + assert.match( + sql, + /CREATE TABLE "ql3"\."run_cancellation_dispatches"/, + ); + assert.match( + sql, + /FOREIGN KEY \(run_id, attempt_id\)[\s\S]+REFERENCES "ql3"\."run_attempts" \(run_id, id\)/, + ); + assert.match(sql, /lease_token_digest char\(64\)/); + assert.doesNotMatch(sql, /lease_token varchar/); + assert.match( + sql, + /GRANT SELECT, INSERT, UPDATE ON "ql3"\."run_cancellation_dispatches" TO ql3_runtime/, + ); + assert.doesNotMatch( + sql, + /GRANT (?:SELECT|INSERT|UPDATE|DELETE)[^;]+run_cancellation_dispatches[^;]+ql3_admin/, + ); + assert.match(sql, /contract_version = 65/); + assert.match(sql, /"run_cancellation_dispatch":1/); + assert.match(sql, /contract_version = 64/); + assert.match( + sql, + /migration_id = 'pg-0065-approved-action-manual-recovery'/, + ); +}); diff --git a/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs b/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs index 6da6f323..a4a88c6e 100644 --- a/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs @@ -61,6 +61,7 @@ function validPrivileges() { tool_invocation_input_artifacts: [true, true, false, false], tool_invocation_preview_artifacts: [true, true, false, false], run_attempts: [true, true, true, false], + run_cancellation_dispatches: [true, true, true, false], run_attempt_log_retention_controls: [true, true, true, true], run_attempt_log_artifact_tombstones: [true, true, false, false], worker_sessions: [true, true, true, false], @@ -194,6 +195,7 @@ function validAdminPrivileges() { tool_invocation_input_artifacts: [false, false, false, false], tool_invocation_preview_artifacts: [false, false, false, false], run_attempts: [false, false, false, false], + run_cancellation_dispatches: [false, false, false, false], run_attempt_log_retention_controls: [false, false, false, false], run_attempt_log_artifact_tombstones: [false, false, false, false], worker_sessions: [false, false, false, false], @@ -815,7 +817,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro serverMajor: 16, currentUser: 'ql3_runtime', contractName: 'control-core', - contractVersion: 64, + contractVersion: 65, migrationIds: [ 'pg-0001-schema-capability', 'pg-0002-run-core', @@ -882,6 +884,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro 'pg-0063-plugin-package-secret-binding-transition-receipts', 'pg-0064-plugin-package-secret-binding-transition-approval-plans', 'pg-0065-approved-action-manual-recovery', + 'pg-0066-cancellation-dispatch', ], }); }); @@ -912,10 +915,10 @@ test('accepts the exact schema and isolated least-privilege admin role', async ( }), ); assert.equal(report.currentUser, 'ql3_admin'); - assert.equal(report.contractVersion, 64); + assert.equal(report.contractVersion, 65); assert.equal( report.migrationIds.at(-1), - 'pg-0065-approved-action-manual-recovery', + 'pg-0066-cancellation-dispatch', ); }); @@ -928,10 +931,10 @@ test('accepts the isolated least-privilege automation manager role', async () => }), ); assert.equal(report.currentUser, 'ql3_automation_manager'); - assert.equal(report.contractVersion, 64); + assert.equal(report.contractVersion, 65); assert.equal( report.migrationIds.at(-1), - 'pg-0065-approved-action-manual-recovery', + 'pg-0066-cancellation-dispatch', ); const widened = automationManagerPrivileges(); @@ -960,10 +963,10 @@ test('accepts the isolated least-privilege human Approval manager role', async ( }), ); assert.equal(report.currentUser, 'ql3_approval_manager'); - assert.equal(report.contractVersion, 64); + assert.equal(report.contractVersion, 65); assert.equal( report.migrationIds.at(-1), - 'pg-0065-approved-action-manual-recovery', + 'pg-0066-cancellation-dispatch', ); const widened = approvalManagerPrivileges(); @@ -994,10 +997,10 @@ test('accepts the isolated least-privilege Run manager role', async () => { }), ); assert.equal(report.currentUser, 'ql3_run_manager'); - assert.equal(report.contractVersion, 64); + assert.equal(report.contractVersion, 65); assert.equal( report.migrationIds.at(-1), - 'pg-0065-approved-action-manual-recovery', + 'pg-0066-cancellation-dispatch', ); const widened = runManagerPrivileges(); @@ -1129,10 +1132,10 @@ test('accepts the exact schema and isolated Worker ingress role', async () => { }), ); assert.equal(report.currentUser, 'ql3_worker_ingress'); - assert.equal(report.contractVersion, 64); + assert.equal(report.contractVersion, 65); assert.equal( report.migrationIds.at(-1), - 'pg-0065-approved-action-manual-recovery', + 'pg-0066-cancellation-dispatch', ); }); diff --git a/packages/ql3-runtime-core/package.json b/packages/ql3-runtime-core/package.json index a565828d..7bd20fcc 100644 --- a/packages/ql3-runtime-core/package.json +++ b/packages/ql3-runtime-core/package.json @@ -41,6 +41,9 @@ "run-cancellation": [ "dist/run/clusterRunCancellation.d.ts" ], + "cancellation-dispatch": [ + "dist/run/cancellation-dispatch/cancellationDispatch.d.ts" + ], "task-start": [ "dist/task-start/taskStart.d.ts" ], @@ -776,6 +779,11 @@ "require": "./dist/run/clusterRunCancellation.js", "default": "./dist/run/clusterRunCancellation.js" }, + "./cancellation-dispatch": { + "types": "./dist/run/cancellation-dispatch/cancellationDispatch.d.ts", + "require": "./dist/run/cancellation-dispatch/cancellationDispatch.js", + "default": "./dist/run/cancellation-dispatch/cancellationDispatch.js" + }, "./task-start": { "types": "./dist/task-start/taskStart.d.ts", "require": "./dist/task-start/taskStart.js", diff --git a/packages/ql3-runtime-core/src/run/cancellation-dispatch/cancellationDispatch.ts b/packages/ql3-runtime-core/src/run/cancellation-dispatch/cancellationDispatch.ts new file mode 100644 index 00000000..62168da1 --- /dev/null +++ b/packages/ql3-runtime-core/src/run/cancellation-dispatch/cancellationDispatch.ts @@ -0,0 +1,496 @@ +import { createHash } from 'node:crypto'; +import type { RunEventRecord } from '../run'; + +export const CANCELLATION_DISPATCH_STATUSES = Object.freeze([ + 'pending', + 'leased', + 'retry_wait', + 'dispatched', + 'blocked', +] as const); + +export type CancellationDispatchStatus = + (typeof CANCELLATION_DISPATCH_STATUSES)[number]; + +export const CANCELLATION_DISPATCH_RESULTS = Object.freeze([ + 'termination_requested', + 'already_exited', + 'identity_mismatch', + 'pid_mismatch', + 'unsupported', + 'invalid', + 'controller_missing', + 'handle_missing', + 'dispatch_error', +] as const); + +export type CancellationDispatchResult = + (typeof CANCELLATION_DISPATCH_RESULTS)[number]; + +export const CANCELLATION_DISPATCH_RETRYABLE_RESULTS = Object.freeze([ + 'controller_missing', + 'handle_missing', + 'dispatch_error', +] as const satisfies readonly CancellationDispatchResult[]); + +export const CANCELLATION_DISPATCH_BLOCKING_RESULTS = Object.freeze([ + 'identity_mismatch', + 'pid_mismatch', + 'unsupported', + 'invalid', +] as const satisfies readonly CancellationDispatchResult[]); + +export const MAX_CANCELLATION_DISPATCH_LEASE_MS = 5 * 60_000; +export const MAX_CANCELLATION_DISPATCH_RETRY_DELAY_MS = 24 * 60 * 60_000; + +export interface CancellationDispatchRecord { + readonly runId: string; + readonly attemptId: string; + readonly status: CancellationDispatchStatus; + readonly version: number; + readonly dispatchCount: number; + readonly nextAttemptAtMs?: number; + readonly leaseOwner?: string; + readonly leaseTokenDigest?: string; + readonly leaseExpiresAtMs?: number; + readonly lastResult?: CancellationDispatchResult; + readonly lastDispatchedAtMs?: number; + readonly createdAtMs: number; + readonly updatedAtMs: number; +} + +export interface ClaimCancellationDispatchCommand { + readonly runId: string; + readonly attemptId: string; + readonly requestedAtMs: number; + readonly owner: string; + readonly leaseToken: string; + readonly leaseDurationMs: number; +} + +export type ClaimCancellationDispatchResult = + | Readonly<{ + status: 'claimed'; + dispatch: Readonly; + leaseToken: string; + }> + | Readonly<{ status: 'not_eligible' }> + | Readonly<{ + status: 'not_due' | 'leased' | 'dispatched' | 'blocked'; + dispatch: Readonly; + }>; + +export interface RecordCancellationDispatchResultCommand { + readonly runId: string; + readonly attemptId: string; + readonly owner: string; + readonly leaseToken: string; + readonly expectedVersion: number; + readonly result: CancellationDispatchResult; + readonly retryDelayMs?: number; + readonly eventId: string; +} + +export interface RecordCancellationDispatchResult { + readonly dispatch: Readonly; + readonly event: Readonly; +} + +export interface CancellationDispatchRepository { + findByRunId( + runId: string, + ): Promise | null>; + claim( + command: Readonly, + ): Promise; + recordResult( + command: Readonly, + ): Promise>; +} + +export class CancellationDispatchError extends Error { + constructor( + message: string, + readonly code: string, + readonly retryable = false, + options?: ErrorOptions, + ) { + super(message, options); + this.name = new.target.name; + } +} + +export class InvalidCancellationDispatchCommandError extends CancellationDispatchError { + constructor(message: string) { + super(message, 'INVALID_CANCELLATION_DISPATCH_COMMAND'); + } +} + +export class CancellationDispatchBindingConflictError extends CancellationDispatchError { + constructor(runId: string, attemptId: string) { + super( + `Cancellation dispatch for Run ${runId} is already bound to another Attempt than ${attemptId}`, + 'CANCELLATION_DISPATCH_BINDING_CONFLICT', + ); + } +} + +export class CancellationDispatchFenceRejectedError extends CancellationDispatchError { + constructor(runId: string) { + super( + `Cancellation dispatch lease for Run ${runId} is stale or no longer owned by this worker`, + 'CANCELLATION_DISPATCH_FENCE_REJECTED', + ); + } +} + +export class CancellationDispatchRepositoryError extends CancellationDispatchError { + constructor(cause?: unknown) { + super( + 'Cancellation dispatch repository operation failed', + 'CANCELLATION_DISPATCH_REPOSITORY_FAILED', + false, + { cause }, + ); + } +} + +const SHA256_PATTERN = /^[a-f0-9]{64}$/u; + +function record(value: unknown, name: string): Record { + if (value === null || typeof value !== 'object' || Array.isArray(value)) { + throw new InvalidCancellationDispatchCommandError(`${name} is invalid`); + } + return value as Record; +} + +function exactKeys( + value: Record, + expected: readonly string[], + name: string, +): void { + const actual = Object.keys(value).sort(); + const canonical = [...expected].sort(); + if ( + actual.length !== canonical.length || + actual.some((key, index) => key !== canonical[index]) + ) { + throw new InvalidCancellationDispatchCommandError( + `${name} shape is invalid`, + ); + } +} + +function identifier( + value: unknown, + name: string, + maximum: number, +): string { + if ( + typeof value !== 'string' || + value.length < 1 || + value.length > maximum || + /[\u0000-\u001f\u007f]/u.test(value) + ) { + throw new InvalidCancellationDispatchCommandError(`${name} is invalid`); + } + return value; +} + +function integer( + value: unknown, + name: string, + minimum: number, + maximum = Number.MAX_SAFE_INTEGER, +): number { + if ( + typeof value !== 'number' || + !Number.isSafeInteger(value) || + value < minimum || + value > maximum + ) { + throw new InvalidCancellationDispatchCommandError(`${name} is invalid`); + } + return value; +} + +export function digestCancellationDispatchLeaseToken(value: string): string { + const token = identifier(value, 'leaseToken', 128); + return createHash('sha256') + .update('qinglong.cancellation-dispatch-lease.v1\0', 'utf8') + .update(token, 'utf8') + .digest('hex'); +} + +export function normalizeCancellationDispatchRunId(value: string): string { + return identifier(value, 'runId', 36); +} + +export function normalizeClaimCancellationDispatchCommand( + value: Readonly, +): Readonly { + const command = record(value, 'claim command'); + exactKeys( + command, + [ + 'runId', + 'attemptId', + 'requestedAtMs', + 'owner', + 'leaseToken', + 'leaseDurationMs', + ], + 'claim command', + ); + return Object.freeze({ + runId: identifier(command.runId, 'runId', 36), + attemptId: identifier(command.attemptId, 'attemptId', 36), + requestedAtMs: integer(command.requestedAtMs, 'requestedAtMs', 0), + owner: identifier(command.owner, 'owner', 128), + leaseToken: identifier(command.leaseToken, 'leaseToken', 128), + leaseDurationMs: integer( + command.leaseDurationMs, + 'leaseDurationMs', + 1, + MAX_CANCELLATION_DISPATCH_LEASE_MS, + ), + }); +} + +export function normalizeRecordCancellationDispatchResultCommand( + value: Readonly, +): Readonly { + const command = record(value, 'result command'); + const retryable = CANCELLATION_DISPATCH_RETRYABLE_RESULTS.includes( + command.result as (typeof CANCELLATION_DISPATCH_RETRYABLE_RESULTS)[number], + ); + exactKeys( + command, + [ + 'runId', + 'attemptId', + 'owner', + 'leaseToken', + 'expectedVersion', + 'result', + ...(retryable ? ['retryDelayMs'] : []), + 'eventId', + ], + 'result command', + ); + if ( + !CANCELLATION_DISPATCH_RESULTS.includes( + command.result as CancellationDispatchResult, + ) + ) { + throw new InvalidCancellationDispatchCommandError('result is invalid'); + } + return Object.freeze({ + runId: identifier(command.runId, 'runId', 36), + attemptId: identifier(command.attemptId, 'attemptId', 36), + owner: identifier(command.owner, 'owner', 128), + leaseToken: identifier(command.leaseToken, 'leaseToken', 128), + expectedVersion: integer( + command.expectedVersion, + 'expectedVersion', + 1, + 2_147_483_647, + ), + result: command.result as CancellationDispatchResult, + ...(retryable + ? { + retryDelayMs: integer( + command.retryDelayMs, + 'retryDelayMs', + 1, + MAX_CANCELLATION_DISPATCH_RETRY_DELAY_MS, + ), + } + : {}), + eventId: identifier(command.eventId, 'eventId', 36), + }); +} + +export function cancellationDispatchResultState( + result: CancellationDispatchResult, +): Readonly<{ + status: Extract< + CancellationDispatchStatus, + 'retry_wait' | 'dispatched' | 'blocked' + >; + eventType: + | 'run.cancel_dispatch_failed' + | 'run.cancel_dispatched' + | 'run.cancel_dispatch_blocked'; +}> { + if ( + CANCELLATION_DISPATCH_RETRYABLE_RESULTS.includes( + result as (typeof CANCELLATION_DISPATCH_RETRYABLE_RESULTS)[number], + ) + ) { + return Object.freeze({ + status: 'retry_wait', + eventType: 'run.cancel_dispatch_failed', + }); + } + if ( + CANCELLATION_DISPATCH_BLOCKING_RESULTS.includes( + result as (typeof CANCELLATION_DISPATCH_BLOCKING_RESULTS)[number], + ) + ) { + return Object.freeze({ + status: 'blocked', + eventType: 'run.cancel_dispatch_blocked', + }); + } + if (!CANCELLATION_DISPATCH_RESULTS.includes(result)) { + throw new InvalidCancellationDispatchCommandError('result is invalid'); + } + return Object.freeze({ + status: 'dispatched', + eventType: 'run.cancel_dispatched', + }); +} + +export function normalizeCancellationDispatchRecord( + value: CancellationDispatchRecord, +): Readonly { + const dispatch = record(value, 'dispatch record'); + const optionalKeys = [ + 'nextAttemptAtMs', + 'leaseOwner', + 'leaseTokenDigest', + 'leaseExpiresAtMs', + 'lastResult', + 'lastDispatchedAtMs', + ].filter((key) => dispatch[key] !== undefined); + exactKeys( + dispatch, + [ + 'runId', + 'attemptId', + 'status', + 'version', + 'dispatchCount', + ...optionalKeys, + 'createdAtMs', + 'updatedAtMs', + ], + 'dispatch record', + ); + if ( + !CANCELLATION_DISPATCH_STATUSES.includes( + dispatch.status as CancellationDispatchStatus, + ) + ) { + throw new InvalidCancellationDispatchCommandError('status is invalid'); + } + const status = dispatch.status as CancellationDispatchStatus; + const createdAtMs = integer(dispatch.createdAtMs, 'createdAtMs', 0); + const updatedAtMs = integer(dispatch.updatedAtMs, 'updatedAtMs', createdAtMs); + const version = integer(dispatch.version, 'version', 0, 2_147_483_647); + const dispatchCount = integer( + dispatch.dispatchCount, + 'dispatchCount', + 0, + 2_147_483_647, + ); + const lastResult = dispatch.lastResult as + | CancellationDispatchResult + | undefined; + const hasLease = + dispatch.leaseOwner !== undefined || + dispatch.leaseTokenDigest !== undefined || + dispatch.leaseExpiresAtMs !== undefined; + if ( + (status === 'leased' && + (typeof dispatch.leaseTokenDigest !== 'string' || + !SHA256_PATTERN.test(dispatch.leaseTokenDigest) || + dispatch.leaseOwner === undefined || + dispatch.leaseExpiresAtMs === undefined)) || + (status !== 'leased' && hasLease) || + ((status === 'pending' || status === 'retry_wait') && + dispatch.nextAttemptAtMs === undefined) || + ((status === 'leased' || status === 'dispatched' || status === 'blocked') && + dispatch.nextAttemptAtMs !== undefined) || + ((status === 'dispatched' || status === 'blocked') && + lastResult === undefined) || + (status === 'pending' && + (version !== 0 || dispatchCount !== 0 || lastResult !== undefined)) || + (status !== 'pending' && dispatchCount < 1) || + version < dispatchCount || + (status === 'leased' && + lastResult !== undefined && + !CANCELLATION_DISPATCH_RETRYABLE_RESULTS.includes( + lastResult as (typeof CANCELLATION_DISPATCH_RETRYABLE_RESULTS)[number], + )) || + (status === 'retry_wait' && + !CANCELLATION_DISPATCH_RETRYABLE_RESULTS.includes( + lastResult as (typeof CANCELLATION_DISPATCH_RETRYABLE_RESULTS)[number], + )) || + (status === 'dispatched' && + lastResult !== 'termination_requested' && + lastResult !== 'already_exited') || + (status === 'blocked' && + !CANCELLATION_DISPATCH_BLOCKING_RESULTS.includes( + lastResult as (typeof CANCELLATION_DISPATCH_BLOCKING_RESULTS)[number], + )) + ) { + throw new InvalidCancellationDispatchCommandError( + 'dispatch state is inconsistent', + ); + } + if ( + dispatch.lastResult !== undefined && + !CANCELLATION_DISPATCH_RESULTS.includes( + dispatch.lastResult as CancellationDispatchResult, + ) + ) { + throw new InvalidCancellationDispatchCommandError('lastResult is invalid'); + } + return Object.freeze({ + runId: identifier(dispatch.runId, 'runId', 36), + attemptId: identifier(dispatch.attemptId, 'attemptId', 36), + status, + version, + dispatchCount, + ...(dispatch.nextAttemptAtMs === undefined + ? {} + : { + nextAttemptAtMs: integer( + dispatch.nextAttemptAtMs, + 'nextAttemptAtMs', + 0, + ), + }), + ...(dispatch.leaseOwner === undefined + ? {} + : { leaseOwner: identifier(dispatch.leaseOwner, 'leaseOwner', 128) }), + ...(dispatch.leaseTokenDigest === undefined + ? {} + : { leaseTokenDigest: dispatch.leaseTokenDigest as string }), + ...(dispatch.leaseExpiresAtMs === undefined + ? {} + : { + leaseExpiresAtMs: integer( + dispatch.leaseExpiresAtMs, + 'leaseExpiresAtMs', + 0, + ), + }), + ...(lastResult === undefined + ? {} + : { lastResult }), + ...(dispatch.lastDispatchedAtMs === undefined + ? {} + : { + lastDispatchedAtMs: integer( + dispatch.lastDispatchedAtMs, + 'lastDispatchedAtMs', + 0, + ), + }), + createdAtMs, + updatedAtMs, + }); +} diff --git a/packages/ql3-runtime-core/test/cancellationDispatch.test.cjs b/packages/ql3-runtime-core/test/cancellationDispatch.test.cjs new file mode 100644 index 00000000..73e2cbd8 --- /dev/null +++ b/packages/ql3-runtime-core/test/cancellationDispatch.test.cjs @@ -0,0 +1,177 @@ +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const root = require('../dist'); +const contract = require('../dist/run/cancellation-dispatch/cancellationDispatch'); + +const RUN_ID = '019f71c0-0000-7000-8000-000000000001'; +const ATTEMPT_ID = '019f71c0-0000-7000-8000-000000000002'; +const EVENT_ID = '019f71c0-0000-7000-8000-000000000003'; +const TOKEN_DIGEST = + 'bf9dbe4700121e13b366bba7adbdfbb5a29d7e4b7a4b8d9181acd45b38c9a8bf'; + +function pendingRecord(overrides = {}) { + return { + runId: RUN_ID, + attemptId: ATTEMPT_ID, + status: 'pending', + version: 0, + dispatchCount: 0, + nextAttemptAtMs: 1_750_000_000_100, + createdAtMs: 1_750_000_000_100, + updatedAtMs: 1_750_000_000_100, + ...overrides, + }; +} + +test('normalizes database-timed claim and result commands with exact bounds', () => { + const claim = contract.normalizeClaimCancellationDispatchCommand({ + runId: RUN_ID, + attemptId: ATTEMPT_ID, + requestedAtMs: 1_750_000_000_100, + owner: 'primary-a', + leaseToken: 'lease-a', + leaseDurationMs: contract.MAX_CANCELLATION_DISPATCH_LEASE_MS, + }); + assert.equal(Object.isFrozen(claim), true); + assert.equal('nowMs' in claim, false); + + const result = contract.normalizeRecordCancellationDispatchResultCommand({ + runId: RUN_ID, + attemptId: ATTEMPT_ID, + owner: 'primary-a', + leaseToken: 'lease-a', + expectedVersion: 1, + result: 'dispatch_error', + retryDelayMs: contract.MAX_CANCELLATION_DISPATCH_RETRY_DELAY_MS, + eventId: EVENT_ID, + }); + assert.equal(Object.isFrozen(result), true); + assert.equal('atMs' in result, false); + assert.equal('nextAttemptAtMs' in result, false); + + assert.throws( + () => + contract.normalizeClaimCancellationDispatchCommand({ + ...claim, + nowMs: 1_750_000_000_100, + }), + contract.InvalidCancellationDispatchCommandError, + ); + assert.throws( + () => + contract.normalizeRecordCancellationDispatchResultCommand({ + ...result, + retryDelayMs: 0, + }), + contract.InvalidCancellationDispatchCommandError, + ); + assert.throws( + () => + contract.normalizeRecordCancellationDispatchResultCommand({ + ...result, + result: 'already_exited', + }), + contract.InvalidCancellationDispatchCommandError, + ); +}); + +test('domain-separates lease digests and never admits a raw token into records', () => { + assert.equal( + contract.digestCancellationDispatchLeaseToken('lease-a'), + TOKEN_DIGEST, + ); + assert.match(TOKEN_DIGEST, /^[0-9a-f]{64}$/); + + const { nextAttemptAtMs: _nextAttemptAtMs, ...pending } = pendingRecord(); + const leased = contract.normalizeCancellationDispatchRecord( + { + ...pending, + status: 'leased', + version: 1, + dispatchCount: 1, + leaseOwner: 'primary-a', + leaseTokenDigest: TOKEN_DIGEST, + leaseExpiresAtMs: 1_750_000_000_200, + }, + ); + assert.equal(leased.leaseTokenDigest, TOKEN_DIGEST); + assert.equal('leaseToken' in leased, false); + assert.throws( + () => + contract.normalizeCancellationDispatchRecord({ + ...leased, + leaseToken: 'lease-a', + }), + contract.InvalidCancellationDispatchCommandError, + ); +}); + +test('rejects records that violate counter, lease, retry, or terminal state invariants', () => { + for (const invalid of [ + pendingRecord({ version: 1 }), + pendingRecord({ status: 'retry_wait', version: 2, dispatchCount: 1 }), + pendingRecord({ + status: 'leased', + version: 1, + dispatchCount: 1, + nextAttemptAtMs: undefined, + leaseOwner: 'primary-a', + leaseTokenDigest: TOKEN_DIGEST, + leaseExpiresAtMs: 1_750_000_000_200, + lastResult: 'already_exited', + }), + pendingRecord({ + status: 'dispatched', + version: 2, + dispatchCount: 1, + nextAttemptAtMs: undefined, + lastResult: 'identity_mismatch', + }), + pendingRecord({ + status: 'blocked', + version: 2, + dispatchCount: 1, + nextAttemptAtMs: undefined, + lastResult: 'dispatch_error', + }), + pendingRecord({ version: 0, dispatchCount: 1 }), + ]) { + assert.throws( + () => contract.normalizeCancellationDispatchRecord(invalid), + contract.InvalidCancellationDispatchCommandError, + ); + } +}); + +test('classifies every result into one durable state and low-sensitive event', () => { + for (const result of contract.CANCELLATION_DISPATCH_RESULTS) { + const state = contract.cancellationDispatchResultState(result); + if (contract.CANCELLATION_DISPATCH_RETRYABLE_RESULTS.includes(result)) { + assert.deepEqual(state, { + status: 'retry_wait', + eventType: 'run.cancel_dispatch_failed', + }); + } else if ( + contract.CANCELLATION_DISPATCH_BLOCKING_RESULTS.includes(result) + ) { + assert.deepEqual(state, { + status: 'blocked', + eventType: 'run.cancel_dispatch_blocked', + }); + } else { + assert.deepEqual(state, { + status: 'dispatched', + eventType: 'run.cancel_dispatched', + }); + } + } +}); + +test('publishes the profile-neutral contract only through its explicit subpath', () => { + assert.equal(root.normalizeCancellationDispatchRecord, undefined); + assert.equal( + root.PostgresCancellationDispatchRepository, + undefined, + ); +}); diff --git a/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs b/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs new file mode 100644 index 00000000..52cbe376 --- /dev/null +++ b/scripts/ql3-postgres-ha-cancellation-dispatch-fixture.cjs @@ -0,0 +1,280 @@ +const assert = require('node:assert/strict'); + +const { + createPostgresDatabaseOpener, + PostgresCancellationDispatchRepository, +} = require('../packages/ql3-cluster-postgres/dist/entrypoints/runtime.js'); +const { + CancellationDispatchFenceRejectedError, + digestCancellationDispatchLeaseToken, +} = require('../packages/ql3-runtime-core/dist/run/cancellation-dispatch/cancellationDispatch.js'); + +const FIXTURE = Object.freeze({ + 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', +}); + +async function openRuntime(connectionString, applicationName) { + return createPostgresDatabaseOpener({ + role: 'runtime', + connection: { connectionString, tls: { mode: 'disable' } }, + pool: { maxConnections: 2, applicationName }, + onPoolError(error) { + throw error; + }, + })(); +} + +async function cancellationDispatchFacts(pool) { + const result = await pool.query( + `SELECT dispatch.status, dispatch.version, + dispatch.dispatch_count AS "dispatchCount", + dispatch.lease_token_digest AS "leaseTokenDigest", + dispatch.last_result AS "lastResult", + run.version AS "runVersion", + run.event_sequence AS "eventSequence", + count(event.id)::integer AS "eventCount" + FROM "ql3"."run_cancellation_dispatches" dispatch + JOIN "ql3"."runs" run ON run.id = dispatch.run_id + LEFT JOIN "ql3"."run_events" event ON event.run_id = run.id + WHERE dispatch.run_id = $1 + GROUP BY dispatch.run_id, run.id`, + [FIXTURE.runId], + ); + assert.equal(result.rows.length, 1); + return result.rows[0]; +} + +async function persistCancellationDispatchHaFixture(options) { + const { migrationPool, runtimeConnectionString } = options; + const beforeClock = await migrationPool.query( + `SELECT floor(extract(epoch FROM clock_timestamp()) * 1000)::bigint + AS "nowMs"`, + ); + 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' + )`, + [FIXTURE.runId, FIXTURE.requestedAtMs - 100, FIXTURE.requestedAtMs], + ); + await migrationPool.query( + `INSERT INTO "ql3"."run_attempts" ( + id, run_id, attempt, status, executor_type, callback_sequence, + created_at_ms + ) VALUES ($1, $2, 1, 'running', 'local_process', 0, $3)`, + [FIXTURE.attemptId, FIXTURE.runId, FIXTURE.requestedAtMs - 50], + ); + + const [firstDatabase, secondDatabase] = await Promise.all([ + openRuntime(runtimeConnectionString, 'ql3-ha-cancel-primary-a'), + openRuntime(runtimeConnectionString, 'ql3-ha-cancel-primary-b'), + ]); + try { + const first = new PostgresCancellationDispatchRepository( + firstDatabase.pool, + ); + const second = new PostgresCancellationDispatchRepository( + secondDatabase.pool, + ); + const candidate = { + runId: FIXTURE.runId, + attemptId: FIXTURE.attemptId, + requestedAtMs: FIXTURE.requestedAtMs, + leaseDurationMs: 30_000, + }; + const claims = await Promise.all([ + first.claim({ + ...candidate, + owner: 'ha-cancel-primary-a', + leaseToken: 'ha-cancel-lease-a', + }), + second.claim({ + ...candidate, + owner: 'ha-cancel-primary-b', + leaseToken: 'ha-cancel-lease-b', + }), + ]); + const claimed = claims.find((result) => result.status === 'claimed'); + const competing = claims.find((result) => result.status !== 'claimed'); + assert.equal(claimed?.status, 'claimed'); + assert.equal(competing?.status, 'leased'); + assert.equal( + claimed.dispatch.createdAtMs >= Number(beforeClock.rows[0].nowMs), + true, + ); + const storedLease = await migrationPool.query( + `SELECT lease_token_digest AS "leaseTokenDigest" + FROM "ql3"."run_cancellation_dispatches" WHERE run_id = $1`, + [FIXTURE.runId], + ); + assert.equal( + storedLease.rows[0].leaseTokenDigest, + digestCancellationDispatchLeaseToken(claimed.leaseToken), + ); + assert.notEqual(storedLease.rows[0].leaseTokenDigest, claimed.leaseToken); + + await migrationPool.query( + `UPDATE "ql3"."run_cancellation_dispatches" + SET lease_expires_at_ms = 0 WHERE run_id = $1`, + [FIXTURE.runId], + ); + const takeover = await second.claim({ + ...candidate, + owner: 'ha-cancel-takeover', + leaseToken: 'ha-cancel-takeover-token', + }); + assert.equal(takeover.status, 'claimed'); + assert.equal(takeover.dispatch.version, 2); + await assert.rejects( + first.recordResult({ + runId: FIXTURE.runId, + attemptId: FIXTURE.attemptId, + owner: claimed.dispatch.leaseOwner, + leaseToken: claimed.leaseToken, + expectedVersion: claimed.dispatch.version, + result: 'already_exited', + eventId: FIXTURE.terminalEventId, + }), + CancellationDispatchFenceRejectedError, + ); + + const retry = await second.recordResult({ + runId: FIXTURE.runId, + attemptId: FIXTURE.attemptId, + owner: 'ha-cancel-takeover', + leaseToken: 'ha-cancel-takeover-token', + expectedVersion: takeover.dispatch.version, + result: 'dispatch_error', + retryDelayMs: 60_000, + eventId: FIXTURE.retryEventId, + }); + assert.equal(retry.dispatch.status, 'retry_wait'); + assert.equal( + (await first.claim({ + ...candidate, + owner: 'ha-cancel-early', + leaseToken: 'ha-cancel-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 finalLease = await first.claim({ + ...candidate, + owner: 'ha-cancel-final', + leaseToken: 'ha-cancel-final-token', + }); + assert.equal(finalLease.status, 'claimed'); + assert.equal(finalLease.dispatch.dispatchCount, 3); + const terminal = await first.recordResult({ + runId: FIXTURE.runId, + attemptId: FIXTURE.attemptId, + owner: 'ha-cancel-final', + leaseToken: 'ha-cancel-final-token', + expectedVersion: finalLease.dispatch.version, + result: 'already_exited', + eventId: FIXTURE.terminalEventId, + }); + assert.equal(terminal.dispatch.status, 'dispatched'); + assert.equal(terminal.event.sequence, 2); + await migrationPool.query( + `WITH observed AS ( + SELECT floor(extract(epoch FROM transaction_timestamp()) * 1000)::bigint + AS at_ms + ), closed_attempt AS ( + UPDATE "ql3"."run_attempts" + SET status = 'cancelled', finished_at_ms = observed.at_ms + FROM observed + WHERE id = $2 AND run_id = $1 + RETURNING id + ), closed_run AS ( + UPDATE "ql3"."runs" + SET status = 'cancelled', version = version + 1, + event_sequence = event_sequence + 1, + finished_at_ms = observed.at_ms + FROM observed + WHERE id = $1 + AND EXISTS (SELECT 1 FROM closed_attempt) + RETURNING id, event_sequence, finished_at_ms + ) + INSERT INTO "ql3"."run_events" ( + id, run_id, sequence, type, dedupe_key, actor_type, actor_id, + attempt_id, payload, created_at_ms + ) + SELECT $3, id, event_sequence, 'run.cancelled', + 'ha-cancel-fixture-settled', 'system', 'ha-contract', $2, + '{"reason":"fixture_settled"}'::jsonb, finished_at_ms + FROM closed_run`, + [FIXTURE.runId, FIXTURE.attemptId, FIXTURE.settledEventId], + ); + const beforePromotion = await cancellationDispatchFacts(migrationPool); + assert.deepEqual(beforePromotion, { + status: 'dispatched', + version: 5, + dispatchCount: 3, + leaseTokenDigest: null, + lastResult: 'already_exited', + runVersion: 5, + eventSequence: 3, + eventCount: 3, + }); + return { + fixture: FIXTURE, + beforePromotion, + databaseTimed: true, + crossPoolClaimExactlyOnce: true, + rawLeaseTokenNeverStored: true, + expiredLeaseTakenOver: true, + staleLeaseFenced: true, + retryDeferredUntilDue: true, + replicatedBeforePromotion: false, + survivedPromotion: false, + }; + } finally { + await Promise.allSettled([ + firstDatabase.close(), + secondDatabase.close(), + ]); + } +} + +async function verifyPromotedCancellationDispatchHaFixture(options) { + const { promotedPool, runtimeConnectionString, evidence } = options; + const afterPromotion = await cancellationDispatchFacts(promotedPool); + assert.deepEqual(afterPromotion, evidence.beforePromotion); + const runtimeDatabase = await openRuntime( + runtimeConnectionString, + 'ql3-ha-cancel-promoted', + ); + try { + const dispatch = await new PostgresCancellationDispatchRepository( + runtimeDatabase.pool, + ).findByRunId(FIXTURE.runId); + assert.equal(dispatch?.status, 'dispatched'); + assert.equal(dispatch?.dispatchCount, 3); + assert.equal(dispatch?.leaseTokenDigest, undefined); + } finally { + await runtimeDatabase.close(); + } + evidence.afterPromotion = afterPromotion; + evidence.survivedPromotion = true; + return evidence; +} + +module.exports = { + cancellationDispatchFacts, + persistCancellationDispatchHaFixture, + verifyPromotedCancellationDispatchHaFixture, +}; diff --git a/scripts/ql3-postgres-ha-contract.cjs b/scripts/ql3-postgres-ha-contract.cjs index 13084286..24da821c 100644 --- a/scripts/ql3-postgres-ha-contract.cjs +++ b/scripts/ql3-postgres-ha-contract.cjs @@ -374,6 +374,11 @@ const { persistNonEmptyToolResultRetirement, verifyPromotedNonEmptyToolResult, } = require('./ql3-postgres-ha-tool-result-fixture.cjs'); +const { + cancellationDispatchFacts, + persistCancellationDispatchHaFixture, + verifyPromotedCancellationDispatchHaFixture, +} = require('./ql3-postgres-ha-cancellation-dispatch-fixture.cjs'); const { activateInstall, pluginPackageTaskReconciliationFixture, @@ -11399,6 +11404,7 @@ async function main(argv = process.argv.slice(2)) { let runAttemptLogRetentionEvidence; let runAttemptLogRetention; let manualRunRetry; + let cancellationDispatch; const startedAt = performance.now(); const timeline = []; let report; @@ -11772,6 +11778,18 @@ async function main(argv = process.argv.slice(2)) { state: 'durable_identity_keyset_ledger_verified', atMs: Number((performance.now() - startedAt).toFixed(3)), }); + cancellationDispatch = await persistCancellationDispatchHaFixture({ + migrationPool: primaryDatabase.pool, + runtimeConnectionString: databaseUrl( + RUNTIME_USER, + RUNTIME_PASSWORD, + primaryPort, + ), + }); + timeline.push({ + state: 'cancellation_dispatch_fenced_on_primary', + atMs: Number((performance.now() - startedAt).toFixed(3)), + }); await primaryDatabase.pool.query( `CREATE ROLE ${REPLICATION_USER} WITH REPLICATION LOGIN`, ); @@ -12409,6 +12427,14 @@ async function main(argv = process.argv.slice(2)) { ); return marker.rows[0]?.count === 1 ? marker.rows[0] : null; }, 'pre-promotion marker WAL replay'); + await waitFor(async () => { + const facts = await cancellationDispatchFacts(standbyDatabase.pool); + return JSON.stringify(facts) === + JSON.stringify(cancellationDispatch.beforePromotion) + ? facts + : null; + }, 'cancellation dispatch WAL replay'); + cancellationDispatch.replicatedBeforePromotion = true; const projectToolSnapshotProjectId = 'ha-tool-snapshot'; await primaryDatabase.pool.query( `INSERT INTO "ql3"."projects" ( @@ -12993,6 +13019,15 @@ async function main(argv = process.argv.slice(2)) { modelInvocationFeaturePromotion.afterPromotion = promotedModelInvocationFeature; modelInvocationFeaturePromotion.survivedPromotion = true; + await verifyPromotedCancellationDispatchHaFixture({ + promotedPool: promotedDatabase.pool, + runtimeConnectionString: databaseUrl( + RUNTIME_USER, + RUNTIME_PASSWORD, + standbyPort, + ), + evidence: cancellationDispatch, + }); const promotedCopilotFailureDiagnosisAdmission = await copilotFailureDiagnosisAdmissionFacts( promotedDatabase.pool, @@ -13788,8 +13823,10 @@ async function main(argv = process.argv.slice(2)) { ]); const sideEffects = await promotedDatabase.pool.query( `SELECT - (SELECT count(*)::integer FROM "ql3"."runs") AS runs, - (SELECT count(*)::integer FROM "ql3"."run_events") AS "runEvents", + (SELECT count(*)::integer FROM "ql3"."runs" + WHERE id <> $3) AS runs, + (SELECT count(*)::integer FROM "ql3"."run_events" + WHERE run_id <> $3) AS "runEvents", (SELECT count(*)::integer FROM "ql3"."runs" WHERE id IN ($1, $2)) AS "diagnosisReadRuns", @@ -13801,6 +13838,7 @@ async function main(argv = process.argv.slice(2)) { [ copilotFailureDiagnosisRead.sourceRunId, copilotFailureDiagnosisRead.runId, + cancellationDispatch.fixture.runId, ], ); assert.deepEqual(sideEffects.rows, [ @@ -13879,6 +13917,7 @@ async function main(argv = process.argv.slice(2)) { unexpectedDomainSideEffects: 0, }, manualRunRetry: manualRunRetry.report, + cancellationDispatch, transactionWindows: { ambiguousCommit: { clientObservedFailure: ambiguousCommitClientRejected, @@ -13934,6 +13973,16 @@ async function main(argv = process.argv.slice(2)) { runAttemptLogRetention, timeline, gates: { + cancellationDispatchUsesDatabaseTimeAndExactFences: + cancellationDispatch.databaseTimed && + cancellationDispatch.crossPoolClaimExactlyOnce && + cancellationDispatch.rawLeaseTokenNeverStored && + cancellationDispatch.expiredLeaseTakenOver && + cancellationDispatch.staleLeaseFenced && + cancellationDispatch.retryDeferredUntilDue, + cancellationDispatchReplicatesAndSurvivesPromotion: + cancellationDispatch.replicatedBeforePromotion && + cancellationDispatch.survivedPromotion, runAttemptLogRetentionLeaseTakeoverAndTombstoneConverge: runAttemptLogRetention.replicatedBeforePromotion && runAttemptLogRetention.stalePrimarySettlementFenced && diff --git a/test/back/cancellationDispatchRepository.test.cjs b/test/back/cancellationDispatchRepository.test.cjs index ab5ea1bd..ed69ae99 100644 --- a/test/back/cancellationDispatchRepository.test.cjs +++ b/test/back/cancellationDispatchRepository.test.cjs @@ -61,9 +61,15 @@ async function createRepository() { logger: { info() {} }, }); databases.push(database); + let nowMs = 1_750_000_000_100; return { database, - repository: new LegacySequelizeCancellationDispatchRepository(database), + repository: new LegacySequelizeCancellationDispatchRepository(database, { + clock: () => nowMs, + }), + setNow(value) { + nowMs = value; + }, }; } @@ -113,7 +119,6 @@ function claim(candidate, overrides = {}) { ...candidate, owner: 'worker-a', leaseToken: 'lease-a', - nowMs: candidate.requestedAtMs, leaseDurationMs: 50, ...overrides, }; @@ -124,7 +129,7 @@ afterEach(async () => { }); test('fences two workers and lets a second worker recover an expired lease', async () => { - const { database, repository } = await createRepository(); + const { database, repository, setNow } = await createRepository(); const candidate = await insertCandidate(database); const first = await repository.claim(claim(candidate)); @@ -133,21 +138,21 @@ test('fences two workers and lets a second worker recover an expired lease', asy assert.equal(first.dispatch.dispatchCount, 1); assert.equal(first.dispatch.leaseOwner, 'worker-a'); + setNow(candidate.requestedAtMs + 25); const competing = await repository.claim( claim(candidate, { owner: 'worker-b', leaseToken: 'lease-b', - nowMs: candidate.requestedAtMs + 25, }), ); assert.equal(competing.status, 'leased'); assert.equal(competing.dispatch.leaseOwner, 'worker-a'); + setNow(candidate.requestedAtMs + 50); const recovered = await repository.claim( claim(candidate, { owner: 'worker-b', leaseToken: 'lease-b', - nowMs: candidate.requestedAtMs + 50, }), ); assert.equal(recovered.status, 'claimed'); @@ -155,6 +160,7 @@ test('fences two workers and lets a second worker recover an expired lease', asy assert.equal(recovered.dispatch.dispatchCount, 2); assert.equal(recovered.dispatch.leaseOwner, 'worker-b'); + setNow(candidate.requestedAtMs + 51); await assert.rejects( repository.recordResult({ runId: candidate.runId, @@ -163,12 +169,12 @@ test('fences two workers and lets a second worker recover an expired lease', asy leaseToken: 'lease-a', expectedVersion: 1, result: 'termination_requested', - atMs: candidate.requestedAtMs + 51, eventId: nextId(), }), CancellationDispatchFenceRejectedError, ); + setNow(candidate.requestedAtMs + 52); const recorded = await repository.recordResult({ runId: candidate.runId, attemptId: candidate.attemptId, @@ -176,7 +182,6 @@ test('fences two workers and lets a second worker recover an expired lease', asy leaseToken: 'lease-b', expectedVersion: recovered.dispatch.version, result: 'termination_requested', - atMs: candidate.requestedAtMs + 52, eventId: nextId(), }); assert.equal(recorded.dispatch.status, 'dispatched'); @@ -188,22 +193,23 @@ test('fences two workers and lets a second worker recover an expired lease', asy result: 'termination_requested', }); + setNow(candidate.requestedAtMs + 100); const terminal = await repository.claim( claim(candidate, { owner: 'worker-c', leaseToken: 'lease-c', - nowMs: candidate.requestedAtMs + 100, }), ); assert.equal(terminal.status, 'dispatched'); }); test('persists retry backoff and only reclaims when it becomes due', async () => { - const { database, repository } = await createRepository(); + const { database, repository, setNow } = await createRepository(); const candidate = await insertCandidate(database); const leased = await repository.claim(claim(candidate)); const retryAtMs = candidate.requestedAtMs + 1_000; + setNow(candidate.requestedAtMs + 1); const failed = await repository.recordResult({ runId: candidate.runId, attemptId: candidate.attemptId, @@ -211,33 +217,33 @@ test('persists retry backoff and only reclaims when it becomes due', async () => leaseToken: 'lease-a', expectedVersion: leased.dispatch.version, result: 'dispatch_error', - atMs: candidate.requestedAtMs + 1, - nextAttemptAtMs: retryAtMs, + retryDelayMs: retryAtMs - (candidate.requestedAtMs + 1), eventId: nextId(), }); assert.equal(failed.dispatch.status, 'retry_wait'); assert.equal(failed.dispatch.nextAttemptAtMs, retryAtMs); assert.equal(failed.event.type, 'run.cancel_dispatch_failed'); + setNow(retryAtMs - 1); const early = await repository.claim( claim(candidate, { owner: 'worker-b', leaseToken: 'lease-b', - nowMs: retryAtMs - 1, }), ); assert.equal(early.status, 'not_due'); + setNow(retryAtMs); const retry = await repository.claim( claim(candidate, { owner: 'worker-b', leaseToken: 'lease-b', - nowMs: retryAtMs, }), ); assert.equal(retry.status, 'claimed'); assert.equal(retry.dispatch.dispatchCount, 2); + setNow(retryAtMs + 1); const missingController = await repository.recordResult({ runId: candidate.runId, attemptId: candidate.attemptId, @@ -245,8 +251,7 @@ test('persists retry backoff and only reclaims when it becomes due', async () => leaseToken: 'lease-b', expectedVersion: retry.dispatch.version, result: 'controller_missing', - atMs: retryAtMs + 1, - nextAttemptAtMs: retryAtMs + 2_000, + retryDelayMs: 1_999, eventId: nextId(), }); assert.equal(missingController.dispatch.status, 'retry_wait'); @@ -257,7 +262,7 @@ test('persists retry backoff and only reclaims when it becomes due', async () => }); test('fails closed for stale candidates and conflicting Attempt bindings', async () => { - const { database, repository } = await createRepository(); + const { database, repository, setNow } = await createRepository(); const stale = await insertCandidate(database, { runStatus: 'succeeded' }); assert.deepEqual(await repository.claim(claim(stale)), { status: 'not_eligible', @@ -278,6 +283,7 @@ test('fails closed for stale candidates and conflicting Attempt bindings', async created_at_ms: candidate.requestedAtMs + 1, }, ]); + setNow(candidate.requestedAtMs + 50); await assert.rejects( repository.claim( claim( @@ -285,7 +291,6 @@ test('fails closed for stale candidates and conflicting Attempt bindings', async { owner: 'worker-b', leaseToken: 'lease-b', - nowMs: candidate.requestedAtMs + 50, }, ), ), @@ -294,7 +299,7 @@ test('fails closed for stale candidates and conflicting Attempt bindings', async }); test('rolls back dispatch state and Run version when event append fails', async () => { - const { database, repository } = await createRepository(); + const { database, repository, setNow } = await createRepository(); const candidate = await insertCandidate(database); const leased = await repository.claim(claim(candidate)); const duplicateEventId = nextId(); @@ -311,6 +316,7 @@ test('rolls back dispatch state and Run version when event append fails', async }, ]); + setNow(candidate.requestedAtMs + 1); await assert.rejects( repository.recordResult({ runId: candidate.runId, @@ -319,7 +325,6 @@ test('rolls back dispatch state and Run version when event append fails', async leaseToken: 'lease-a', expectedVersion: leased.dispatch.version, result: 'already_exited', - atMs: candidate.requestedAtMs + 1, eventId: duplicateEventId, }), ); @@ -336,6 +341,7 @@ test('rolls back dispatch state and Run version when event append fails', async ); assert.deepEqual(run, { version: 2, event_sequence: 0 }); + setNow(candidate.requestedAtMs + 2); const recovered = await repository.recordResult({ runId: candidate.runId, attemptId: candidate.attemptId, @@ -343,7 +349,6 @@ test('rolls back dispatch state and Run version when event append fails', async leaseToken: 'lease-a', expectedVersion: leased.dispatch.version, result: 'already_exited', - atMs: candidate.requestedAtMs + 2, eventId: nextId(), }); assert.equal(recovered.dispatch.status, 'dispatched'); @@ -439,7 +444,6 @@ test('fails closed instead of reclaiming a corrupt persisted lease', async () => claim(candidate, { owner: 'worker-b', leaseToken: 'lease-b', - nowMs: candidate.requestedAtMs, }), ), CancellationDispatchRepositoryError, diff --git a/test/back/primaryCancellationDispatcher.test.cjs b/test/back/primaryCancellationDispatcher.test.cjs index 6168df84..3b0c5795 100644 --- a/test/back/primaryCancellationDispatcher.test.cjs +++ b/test/back/primaryCancellationDispatcher.test.cjs @@ -39,6 +39,7 @@ function fakeDispatchRepository(overrides = {}) { if (overrides.claim) return overrides.claim(command); return { status: 'claimed', + leaseToken: command.leaseToken, dispatch: { runId: command.runId, attemptId: command.attemptId, @@ -46,10 +47,10 @@ function fakeDispatchRepository(overrides = {}) { version: 1, dispatchCount: 1, leaseOwner: command.owner, - leaseToken: command.leaseToken, - leaseExpiresAtMs: command.nowMs + command.leaseDurationMs, - createdAtMs: command.nowMs, - updatedAtMs: command.nowMs, + leaseTokenDigest: 'a'.repeat(64), + leaseExpiresAtMs: NOW_MS + command.leaseDurationMs, + createdAtMs: NOW_MS, + updatedAtMs: NOW_MS, }, }; }, @@ -60,12 +61,12 @@ function fakeDispatchRepository(overrides = {}) { dispatch: { runId: command.runId, attemptId: command.attemptId, - status: command.nextAttemptAtMs ? 'retry_wait' : 'dispatched', + status: command.retryDelayMs ? 'retry_wait' : 'dispatched', version: command.expectedVersion + 1, dispatchCount: 1, lastResult: command.result, createdAtMs: NOW_MS, - updatedAtMs: command.atMs, + updatedAtMs: NOW_MS, }, event: { id: command.eventId, @@ -74,7 +75,7 @@ function fakeDispatchRepository(overrides = {}) { type: 'fixture', actorType: 'worker', payload: {}, - createdAtMs: command.atMs, + createdAtMs: NOW_MS, }, }; }, @@ -88,7 +89,6 @@ function dispatcherOptions(overrides = {}) { leaseDurationMs: 100, retryBaseMs: 1_000, retryMaxMs: 8_000, - clock: () => NOW_MS, createId: () => `019f71d0-0000-7000-8000-${String(++id).padStart(12, '0')}`, ...overrides, }; @@ -149,7 +149,7 @@ test('leases before signalling and durably classifies every controller result', version: 3, dispatchCount: 2, leaseOwner: 'worker-b', - leaseToken: 'other-lease', + leaseTokenDigest: 'b'.repeat(64), leaseExpiresAtMs: NOW_MS + 1_000, createdAtMs: NOW_MS - 5_000, updatedAtMs: NOW_MS - 100, @@ -158,6 +158,7 @@ test('leases before signalling and durably classifies every controller result', } return { status: 'claimed', + leaseToken: command.leaseToken, dispatch: { runId: command.runId, attemptId: command.attemptId, @@ -165,10 +166,10 @@ test('leases before signalling and durably classifies every controller result', version: 1, dispatchCount: 1, leaseOwner: command.owner, - leaseToken: command.leaseToken, - leaseExpiresAtMs: command.nowMs + command.leaseDurationMs, - createdAtMs: command.nowMs, - updatedAtMs: command.nowMs, + leaseTokenDigest: 'a'.repeat(64), + leaseExpiresAtMs: NOW_MS + command.leaseDurationMs, + createdAtMs: NOW_MS, + updatedAtMs: NOW_MS, }, }; }, @@ -243,8 +244,8 @@ test('leases before signalling and durably classifies every controller result', ); assert.equal( dispatches.results.find((result) => result.result === 'dispatch_error') - .nextAttemptAtMs, - NOW_MS + 1_000, + .retryDelayMs, + 1_000, ); }); diff --git a/test/back/ql3PackageBoundaryAudit.test.cjs b/test/back/ql3PackageBoundaryAudit.test.cjs index e890eb15..ee542dcc 100644 --- a/test/back/ql3PackageBoundaryAudit.test.cjs +++ b/test/back/ql3PackageBoundaryAudit.test.cjs @@ -299,10 +299,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( rootSourceFileRoles: runtimeCore.rootSourceFileRoles, }, { - sourceFiles: 168, + sourceFiles: 169, rootSourceFiles: 1, rootSourceLines: 160, - nestedSourceFiles: 167, + nestedSourceFiles: 168, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, ); @@ -421,10 +421,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( rootSourceFileRoles: clusterPostgres.rootSourceFileRoles, }, { - sourceFiles: 168, + sourceFiles: 170, rootSourceFiles: 1, rootSourceLines: 126, - nestedSourceFiles: 167, + nestedSourceFiles: 169, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, );