diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index 0c37df6d..ac96144f 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -11,6 +11,20 @@ 最新增量证据(2026-08-12): +- D-294/ADR-0382(已接受) + Cluster 已补齐共享 `qinglong/run-manual-retry@v1` 的 PostgreSQL 原子 authority,但在可信强认证 transport 完成前保持产品 route + 关闭。adapter 只接受五分钟内的 `multi_factor|hardware` User,在单个 `SERIALIZABLE` 事务中取得 Project 行锁,重验 + Project/RoleBinding、源 Run、current Task 与 immutable remote Worker execution revision,随后原子创建新 queued Run、claimed + Attempt、双 Event 与 allowed audit;精确 replay 返回原 identity,源 Run 不变且新 Run 不继承自动 retry policy。Project 行锁同时 + 序列化每 Project/User 64 次/分钟的既有 Run-ledger quota,因此不新增 package、dependency、migration、table/index、role、timer、 + listener、cache 或 sidecar。实现刻意不使用会隐含要求 UPDATE privilege 的 RoleBinding/Task `FOR SHARE`,避免为了读取证明扩权。 + PostgreSQL package 307 pass/1 conditional skip;真实 PostgreSQL 18.4 arm64 physical HA 在两个独立 Runtime Pool 下通过 exact + concurrent replay、最后 quota slot 竞争、同步 WAL 与 promotion 后核验,共 119 gates、timeline `1→2`,报告 SHA-256 为 + `ca5d33a30f2768072223fb22346d962866948b0c0c970b62a6338d25a3ac9dda`。完整 18-package clean build/test 退出 + 0,backend 1163 pass/2 skip/0 fail;workspace 为 18 package/1061 source/1043 nested,package/dependency/Edge import + 边界零 finding,14 种 Local Profile artifact 与 Local image static audit 全部 compatible;最小 Edge 保持 53 个 loaded module 且 + 不包含 Cluster/PostgreSQL/pg。现有 Cluster bearer 仍只有 + `single_factor`,不得连接此 mutation;purpose-bound OIDC MFA/hardware HTTP/UI 与 Kubernetes 多节点组合证据仍待后续完成。 - D-293/ADR-0381(已接受) Local Edge/Standalone 已增加强认证的人工 `run.retry` 纵向切片。共享 `qinglong/run-manual-retry@v1` 契约位于 Runtime Core; 手工 retry 只接受 `failed|cancelled|timed_out` 的顶层 runtime-owned 非 Workflow Run,源 Run 与历史保持终态不变,并创建以 @@ -8508,7 +8522,7 @@ flowchart LR > ADR-0058 至 ADR-0092 的以下段落是历史快照,其 PostgreSQL 数字与“下一切片”已由 ADR-0104/0105 及后续 Accepted ADR 取代;资源、Package 和物理证据边界仍保留作为演进记录:PostgreSQL 当时 baseline 为 11 条 reviewed migration、capability v10、19 张表和 migration/runtime/admin/worker-ingress 四角色;本机 SQLite 为二十八条 reviewed migration、capability v14、二十六张 owned table,并新增正式 TaskDefinition head/immutable revision Repository。当前有二十七个受审 3.0 importer:Owner maintenance 已合并为两个权限隔离 subpath,execution/control/recovery/dispatch 已合并为 `@qinglong/local-execution` 四个单向 subpath,bootstrap/credential-recovery 已合并为 `@qinglong/local-owner-ceremony` 两个互不依赖 subpath;三个 package 都不提供聚合根入口,hard cap 已同步降为 27。一次拓扑 build/test 会先清理已登记 QL3 package 的 stale dist;27 包全量测试、dependency/source boundary、联网 vulnerability audit 与六种 Profile 制品门禁均通过。ADR-0090 让 production packlist 只发布 JS、`.d.ts` 和受审 assets/drizzle,不再携带开发 map;ADR-0092 后当前最大 application 为 1,691,009 bytes、267 files、61 loaded modules,最大抽样 RSS delta 11,780,096 bytes,低于 4 MiB/512 files/16 MiB。所有 package build 已变为 self-only,全量从约 198 次编译降至精确 27 次,clean 状态单包测试仍按依赖闭包工作。资源门禁已拆为原生 Linux x64/arm64 的 128 MiB router stress、256 MiB Edge release guard 与 512 MiB Cluster control guard,并从容器内部验证 cgroup v2、零 swap/OOM、非 root、只读挂载、seccomp 和 `NoNewPrivs`;本轮本地原生 arm64 三档已通过,x64 等待远端 CI,所有档位均明确不是物理设备或生产容量承诺。物理 Edge candidate recorder 已绑定设备 manifest、实际 Linux/存储环境、Edge/SQLite 基准与 no-replace SHA-256 报告,并会拒绝容器/VM;同设备同 boot 的 idle sampler、不主动填盘的专用文件系统 fault probe,以及通过正式 Repository 写入/扫描 100/1000/10000 个 `qinglong/command@v1` TaskDefinition 的规模记录协议均已具备契约门禁,但尚未取得固定实机报告或 signature。ADR-0091 已冻结 1–32 个 exact descriptor 的不可变 TaskSpec semantic registry、内建 command v1 和本机写前门禁;历史 revision 在 provider 缺失时仍可读。ADR-0092 已实现绑定 source revision/content digest 的 Profile-neutral command plan 与确定性本机 context/execution 映射,且保持 subpath-only。 -> 下表 PR-1 的旧“未完成”累计文字中,`completion/cancellation/timeout` 已由 ADR-0072 取代并闭环,Artifact range read 与 Local/Cluster retention 已由 ADR-0377/0378/0379 闭环,Local admission-safe lost retry lifecycle 已由 ADR-0380 闭环,强认证的 Local 手动 retry CLI、审计与 rate limit 已由 ADR-0381 闭环;当前本机剩余项是可信 HTTP MFA/hardware adapter、UI、Workflow recovery、部署 controller 与固定路由设备实机门。远端 Worker completion、Workflow cancellation 和 Cluster expiry/retry 的核心数据库链、HA 重放与启动装配也已闭环;仍缺 Cluster 手动 retry/PostgreSQL 全局 quota 与真实 Kubernetes 多节点故障证据,不能与本机结论混用。 +> 下表 PR-1 的旧“未完成”累计文字中,`completion/cancellation/timeout` 已由 ADR-0072 取代并闭环,Artifact range read 与 Local/Cluster retention 已由 ADR-0377/0378/0379 闭环,Local admission-safe lost retry lifecycle 已由 ADR-0380 闭环,强认证的 Local 手动 retry CLI、审计与 rate limit 已由 ADR-0381 闭环,Cluster 手动 retry/PostgreSQL 全局 quota 与 physical HA authority 已由 ADR-0382 闭环;当前剩余项是可信 HTTP MFA/hardware adapter、UI、Workflow recovery、部署 controller、固定路由设备实机门与真实 Kubernetes 多节点故障证据。远端 Worker completion、Workflow cancellation 和 Cluster expiry/retry 的核心数据库链、HA 重放与启动装配也已闭环;数据库 authority 完成不代表单因子 Cluster bearer 已获授权,不能与产品 transport 结论混用。 > > 下表 PR-0 的累积长文本仍含“四角色、21 条 migration、capability v20、21 项 HA”历史短语;当前权威基线应读取为六角色、23 条 migration、capability v22/36 表和 23 项 physical HA gate。ADR-0145/0146 又增加默认关闭的 manager-only TLS 1.3 management process、可选双副本 operation 与 durable distributed quota;它仍缺全副本重启 keyset anti-rollback 和真实 IdP/live ingress,所以“受认证管理入口”保持“已孵化但生产失败关闭”。 diff --git a/docs/adr/ADR-0382-postgresql-cluster-manual-run-retry-authority.md b/docs/adr/ADR-0382-postgresql-cluster-manual-run-retry-authority.md new file mode 100644 index 00000000..fd5ad6f1 --- /dev/null +++ b/docs/adr/ADR-0382-postgresql-cluster-manual-run-retry-authority.md @@ -0,0 +1,75 @@ +# ADR-0382:PostgreSQL Cluster 手动 Run retry 原子 Authority + +- 状态:Accepted +- 日期:2026-08-12 +- 关联 RFC:QL-RFC-0001 D-294 +- 前置决策:ADR-0039、ADR-0047、ADR-0119、ADR-0361、ADR-0381 + +## 上下文 + +ADR-0381 已冻结手动 retry 的共享语义并完成 Local SQLite/CLI 纵向切片,但 Cluster 尚缺少可在多副本下工作的 PostgreSQL authority。不能把 Local 的进程内状态或单连接假设复制到 Cluster,也不能让每个 `cluster-control` 副本各自维护限流 bucket。 + +现有 Cluster Control bearer 只建立 `single_factor` User。即使数据库事务能够安全创建新 Run,也不能因此把该 bearer 宣称为强认证产品入口。存储 authority、可信认证 transport 与 HTTP route 必须分层交付。 + +## 决策 + +### 1. PostgreSQL adapter 复用共享领域契约 + +`@qinglong/cluster-postgres/run-manual-retry` 实现 ADR-0381 的 `RunManualRetryRepository`,不定义第二套 Cluster 请求或结果。它只接受五分钟内且未过期的 `multi_factor|hardware` User,并在单个 `SERIALIZABLE` 事务内: + +1. 取得 Project 行锁并重验 Project/RoleBinding fence; +2. 精确检查 mutation replay; +3. 锁定源 Run,重验终态、版本、顶层/runtime-owned/非 Workflow 与 remote Worker execution revision; +4. 重验 current Task enabled 与 immutable execution digest; +5. 消耗持久全局配额; +6. 原子插入新 queued Run、claimed remote Worker Attempt、两个 Event 与 allowed security audit。 + +源 Run 保持终态;新 Run 以 `retry_of_run_id` 关联源 Run,不继承 `run_retry_policies`。相同 mutation 的精确重放返回原 Run/Attempt identity;字段漂移返回 `mutation_conflict`。 + +### 2. Project 行锁是跨副本序列化点 + +同一 Project 的授权管理 mutation 已使用 Project 行锁。手动 retry 复用该锁来序列化 Policy fence、Task current head 和 append-only RoleBinding 观察;RoleBinding/Task 查询不使用 `FOR SHARE`,因为 PostgreSQL 行锁语法会额外要求 UPDATE table privilege,不能为了读取证明扩大 runtime role。 + +事务使用既有 `ql3_runtime` 最小权限:读取 Project/RoleBinding/Task/execution,读写 Run aggregate,并只 INSERT security audit。不会新增数据库角色、迁移、表、索引或函数。 + +### 3. 全局限流复用 Run ledger + +Cluster 固定为同 Project、同 User 每分钟最多 64 个新 `run_manual_retry` Run。事务取得 Project 锁后,使用既有 `(project_id, created_at_ms, id)` Run ledger/index 查询窗口;因此多个 Pod、进程重启和主备切换共享同一额度。精确 replay 不消耗额度,也不新增 quota table、cache、timer、listener 或 sidecar。 + +### 4. 存储可用不等于产品入口可用 + +本 ADR 只接受 PostgreSQL authority。现有 Cluster Control bearer 仍是 `single_factor`,不得连接该 mutation。公开 HTTP/UI 必须等待 purpose-bound OIDC MFA 或 hardware-backed User transport,并在 route admission 与事务提交前分别完成认证/Policy/audit fence;MCP、Worker、AI Tool 不取得该 authority。 + +## 验收 + +- SQL 契约测试覆盖原子五类写入、精确 replay、认证/授权 fence、持久配额和 serialization retry; +- 真实 PostgreSQL 必须使用两个独立 Runtime Pool 并发竞争同一 mutation 与最后一个 quota slot; +- 同步 standby 必须在 promotion 前观察 64 个 retry Run/Attempt、128 个 Event、64 条 allowed audit、零继承 policy; +- promotion 后必须保留相同事实,源 Run 仍为 failed; +- 完整 package、backend、dependency/package/Edge import 与 Profile artifact 门通过后才允许阶段性提交。 + +### 验收证据(2026-08-12) + +- Cluster PostgreSQL package:308 项中 307 通过、1 项按外部数据库条件跳过;新增 5 项专项测试全部通过; +- PostgreSQL 18.4 arm64 physical HA:119 gates,timeline `1→2`,两个独立 Runtime Pool 完成 exact concurrent replay 与全局 quota 竞争;报告 SHA-256 `ca5d33a30f2768072223fb22346d962866948b0c0c970b62a6338d25a3ac9dda`; +- promotion 前后均为 64 retry Run、64 claimed remote Attempt、128 Event、64 allowed audit、0 retry policy,源 Run 保持 failed; +- 18-package clean build/test 完整退出 0;backend 1,165 项中 1,163 通过、2 项环境条件跳过、0 失败; +- package/dependency/Edge import audit 零 finding;workspace 为 18 package、1,061 source、1,043 nested,`singleSourcePackages=[]`、`shallowSourcePackages=[]`; +- 14 种 Local Profile artifact 与 Local image static audit 全部 `compatible:true`,最小 Edge 仍只有 Local SQLite/Runtime Core/SemVer、53 个 loaded module,不包含 Cluster/PostgreSQL/pg; +- 本批没有新增 workspace package、第三方依赖、migration、数据库对象或常驻资源。 + +## 被否决的替代方案 + +1. **每副本内存 token bucket**:重启可绕过且总额度随副本数增长,拒绝。 +2. **新增 quota 表或定时清理器**:既有 Run ledger 足以裁决,会增加迁移与低配/集群共同维护成本,拒绝。 +3. **为 `FOR SHARE` 授予 RoleBinding/Task UPDATE**:只是满足 SQL 语法,不是业务写权限,会破坏最小权限,拒绝。 +4. **使用 superuser 或 migration Pool 执行 retry**:掩盖生产角色缺口并扩大事故半径,拒绝。 +5. **直接接入 Cluster bearer HTTP**:当前只有单因子认证,不满足恢复执行的强认证边界,拒绝。 +6. **为 adapter 新建 workspace package**:能力属于既有 PostgreSQL driver domain,不具备独立制品或依赖隔离收益,拒绝。 + +## 后续工作 + +- 为 Cluster 管理面接入 purpose-bound OIDC MFA/hardware transport,再装配 `run.retry` route; +- 增加受审 UI 的 source/new Run linkage、terminal reason 与 retry preview; +- 为 Workflow/StepRun 独立冻结 recovery 语义,不复用顶层 Run retry; +- 在真实 Kubernetes 多节点门中补充管理 transport、Pod 重调度与数据库 failover 的组合证据。 diff --git a/docs/adr/README.md b/docs/adr/README.md index c2106ce4..5c335d64 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -385,6 +385,7 @@ | [ADR-0379](./ADR-0379-cluster-run-attempt-log-retention.md) | Cluster Run Attempt 日志多副本保留与条件删除 | Accepted | | [ADR-0380](./ADR-0380-local-lost-run-retry-single-control-cadence.md) | Local lost Run retry 复用单一 execution-control cadence | Accepted | | [ADR-0381](./ADR-0381-strong-local-manual-run-retry.md) | 强认证的 Local 手动 Run retry | Accepted | +| [ADR-0382](./ADR-0382-postgresql-cluster-manual-run-retry-authority.md) | PostgreSQL Cluster 手动 Run retry 原子 Authority | Accepted | ## 规则 diff --git a/packages/ql3-cluster-postgres/package.json b/packages/ql3-cluster-postgres/package.json index 2ea7d5e6..a67aa73c 100644 --- a/packages/ql3-cluster-postgres/package.json +++ b/packages/ql3-cluster-postgres/package.json @@ -70,6 +70,11 @@ "require": "./dist/task-start/taskStartRepository.js", "default": "./dist/task-start/taskStartRepository.js" }, + "./run-manual-retry": { + "types": "./dist/run-management/runManualRetryRepository.d.ts", + "require": "./dist/run-management/runManualRetryRepository.js", + "default": "./dist/run-management/runManualRetryRepository.js" + }, "./approval-manager": { "types": "./dist/approval-management/index.d.ts", "require": "./dist/approval-management/index.js", diff --git a/packages/ql3-cluster-postgres/src/run-management/runManualRetryRepository.ts b/packages/ql3-cluster-postgres/src/run-management/runManualRetryRepository.ts new file mode 100644 index 00000000..08f35c11 --- /dev/null +++ b/packages/ql3-cluster-postgres/src/run-management/runManualRetryRepository.ts @@ -0,0 +1,660 @@ +import type { PostgresClient, PostgresPool } from '@qinglong/runtime-core'; +import { + InvalidRunManualRetryError, + MAX_RUN_MANUAL_RETRY_AUTHENTICATION_AGE_MS, + RUN_MANUAL_RETRY_SOURCE_STATUSES, + RunManualRetryFenceRejectedError, + RunManualRetryNotFoundError, + RunManualRetryRateLimitedError, + RunManualRetryUnavailableError, + normalizeRunManualRetryCommand, + normalizeRunManualRetryResult, + type RunManualRetryAllowedRole, + type RunManualRetryCommand, + type RunManualRetryRepository, + type RunManualRetryResult, + type RunManualRetrySourceStatus, +} from '@qinglong/runtime-core/run-manual-retry'; + +import { + POSTGRES_DEFINITION_RETRYABLE_SQL_STATES, + POSTGRES_DEFINITION_TRANSACTION_ATTEMPTS, + configurePostgresDefinitionTransaction, + postgresRequiredBoolean, + postgresRequiredInteger, + postgresRequiredJsonObject, + postgresRequiredString, + postgresSqlState, + rollbackPostgresDefinitionTransaction, +} from '../repository/definitionRepositorySupport'; + +type Row = Record; + +export const CLUSTER_RUN_MANUAL_RETRY_RATE_WINDOW_MS = 60_000; +export const CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT = 64; + +const ALLOWED_ROLES = new Set([ + 'owner', + 'admin', + 'operator', +]); +const CLUSTER_STRONG_ASSURANCES = new Set(['multi_factor', 'hardware']); +const TASK_REVISION_PATTERN = /^qltd:v1:([1-9]\d*):([0-9a-f]{64})$/; + +interface SourceRun { + readonly taskId: string; + readonly taskRevision: string; + readonly taskName?: string; + readonly taskSnapshotRef: string; + readonly inputRef?: string; + readonly priority: number; +} + +interface ExecutionRevision { + readonly contentDigest: string; + readonly sourceContentDigest: string; +} + +function unavailable(options?: ErrorOptions): RunManualRetryUnavailableError { + return new RunManualRetryUnavailableError(options); +} + +function text(row: Row, key: string): string { + return postgresRequiredString(row[key], unavailable); +} + +function integer(row: Row, key: string): number { + return postgresRequiredInteger(row[key], unavailable); +} + +function optionalText(row: Row, key: string): string | undefined { + if (row[key] === null) return undefined; + return text(row, key); +} + +function json(row: Row, key: string): Record { + return postgresRequiredJsonObject(row[key], unavailable); +} + +function sourceStatus(value: string): RunManualRetrySourceStatus { + if ( + !RUN_MANUAL_RETRY_SOURCE_STATUSES.includes( + value as RunManualRetrySourceStatus, + ) + ) { + throw new RunManualRetryFenceRejectedError('source_not_terminal'); + } + return value as RunManualRetrySourceStatus; +} + +function sameFencePayload( + value: unknown, + command: Readonly, +): boolean { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const fence = value as Record; + return ( + fence.project_version === command.policyFence.projectVersion && + fence.binding_version === command.policyFence.bindingVersion + ); +} + +async function databaseNow(client: PostgresClient): Promise { + const result = await client.query(` + SELECT floor(extract(epoch FROM statement_timestamp()) * 1000)::bigint + AS "nowMs" + `); + if (result.rows.length !== 1) throw unavailable(); + return integer(result.rows[0]!, 'nowMs'); +} + +function confirmStrongAuthentication( + command: Readonly, + observedAtMs: number, +): void { + if ( + command.principal.subject.type !== 'user' || + !CLUSTER_STRONG_ASSURANCES.has(command.principal.assurance) || + command.principal.authenticatedAtMs > observedAtMs || + command.principal.expiresAtMs <= observedAtMs || + observedAtMs - command.principal.authenticatedAtMs > + MAX_RUN_MANUAL_RETRY_AUTHENTICATION_AGE_MS + ) { + throw new RunManualRetryFenceRejectedError('authentication_changed'); + } +} + +async function confirmAuthorization( + client: PostgresClient, + command: Readonly, +): Promise { + const project = await client.query( + ` + SELECT status AS "projectStatus", version AS "projectVersion" + FROM "ql3"."projects" WHERE id = $1 FOR UPDATE + `, + [command.projectId], + ); + if (project.rows.length === 0) throw new RunManualRetryNotFoundError(); + if (project.rows.length !== 1) throw unavailable(); + // Authorized management mutations take the same Project lock. Keeping this + // append-only RoleBinding read lock-free avoids granting UPDATE authority to + // the runtime role merely to use PostgreSQL row-lock syntax. + const binding = await client.query( + ` + SELECT version AS "bindingVersion", state AS "bindingState", + role AS "bindingRole" + FROM "ql3"."project_role_bindings" + WHERE project_id = $1 AND subject_type = $2 AND subject_id = $3 + ORDER BY version DESC LIMIT 1 + `, + [ + command.projectId, + command.principal.subject.type, + command.principal.subject.id, + ], + ); + const currentProject = project.rows[0]!; + const currentBinding = binding.rows[0]; + if ( + text(currentProject, 'projectStatus') !== 'active' || + integer(currentProject, 'projectVersion') !== + command.policyFence.projectVersion || + !currentBinding || + integer(currentBinding, 'bindingVersion') !== + command.policyFence.bindingVersion || + text(currentBinding, 'bindingState') !== 'active' || + !ALLOWED_ROLES.has( + text(currentBinding, 'bindingRole') as RunManualRetryAllowedRole, + ) + ) { + throw new RunManualRetryFenceRejectedError('authorization_changed'); + } +} + +async function findReplay( + client: PostgresClient, + command: Readonly, +): Promise { + const result = await client.query( + ` + SELECT run.id AS "runId", run.project_id AS "projectId", + run.retry_of_run_id AS "retryOfRunId", + run.task_id AS "taskId", run.task_revision AS "taskRevision", + run.trigger_type AS "triggerType", + run.execution_origin AS "executionOrigin", + run.execution_owner AS "executionOwner", + run.triggered_by AS "triggeredBy", run.request_id AS "requestId", + run.status AS "runStatus", run.version AS "runVersion", + run.event_sequence AS "eventSequence", + run.created_at_ms AS "createdAtMs", + attempt.id AS "attemptId", attempt.executor_type AS "executorType", + created.actor_type AS "createdActorType", + created.actor_id AS "createdActorId", + created.payload AS "createdPayload", + queued.actor_type AS "queuedActorType", + queued.actor_id AS "queuedActorId", + queued.payload AS "queuedPayload" + FROM "ql3"."runs" AS run + JOIN "ql3"."run_attempts" AS attempt + ON attempt.run_id = run.id AND attempt.attempt = 1 + JOIN "ql3"."run_events" AS created + ON created.run_id = run.id AND created.sequence = 1 + AND created.type = 'run.created' + JOIN "ql3"."run_events" AS queued + ON queued.run_id = run.id AND queued.sequence = 2 + AND queued.type = 'run.queued' + WHERE run.project_id = $1 AND run.idempotency_key = $2 + FOR UPDATE OF run + `, + [command.projectId, `ql3:run-manual-retry:v1:${command.mutationId}`], + ); + if (result.rows.length > 1) throw unavailable(); + return result.rows[0]; +} + +function replayResult( + command: Readonly, + row: Row, +): Readonly { + const created = json(row, 'createdPayload'); + const queued = json(row, 'queuedPayload'); + if ( + text(row, 'projectId') !== command.projectId || + text(row, 'retryOfRunId') !== command.sourceRunId || + text(row, 'triggerType') !== 'run_manual_retry' || + text(row, 'executionOrigin') !== 'manual' || + text(row, 'executionOwner') !== 'runtime' || + text(row, 'triggeredBy') !== command.principal.subject.id || + text(row, 'requestId') !== command.mutationId || + text(row, 'runStatus') !== 'queued' || + integer(row, 'runVersion') !== 2 || + integer(row, 'eventSequence') !== 2 || + text(row, 'executorType') !== 'remote_worker' || + text(row, 'createdActorType') !== command.principal.subject.type || + text(row, 'createdActorId') !== command.principal.subject.id || + text(row, 'queuedActorType') !== command.principal.subject.type || + text(row, 'queuedActorId') !== command.principal.subject.id || + created.mutation_id !== command.mutationId || + created.retry_of_run_id !== command.sourceRunId || + created.source_run_status !== command.expectedRunStatus || + created.source_run_version !== command.expectedRunVersion || + created.inherit_retry_policy !== false || + created.authentication_id !== command.principal.authenticationId || + created.audit_event_id !== command.auditEventId || + typeof created.execution_revision_digest !== 'string' || + !/^[0-9a-f]{64}$/.test(created.execution_revision_digest) || + !sameFencePayload(created.policy_fence, command) || + queued.from_status !== 'created' || + queued.to_status !== 'queued' || + queued.version !== 2 + ) { + throw new RunManualRetryFenceRejectedError('mutation_conflict'); + } + return normalizeRunManualRetryResult({ + status: 'existing', + projectId: command.projectId, + sourceRunId: command.sourceRunId, + sourceRunStatus: command.expectedRunStatus, + sourceRunVersion: command.expectedRunVersion, + runId: text(row, 'runId'), + retryOfRunId: text(row, 'retryOfRunId'), + taskId: text(row, 'taskId'), + taskRevision: text(row, 'taskRevision'), + attemptId: text(row, 'attemptId'), + runStatus: 'queued', + runVersion: 2, + eventSequence: 2, + executorType: 'remote_worker', + executionRevisionDigest: created.execution_revision_digest, + createdAtMs: integer(row, 'createdAtMs'), + }); +} + +async function findSource( + client: PostgresClient, + command: Readonly, +): Promise { + const result = await client.query( + ` + SELECT run.project_id AS "projectId", run.task_id AS "taskId", + run.task_revision AS "taskRevision", run.task_name AS "taskName", + run.task_snapshot_ref AS "taskSnapshotRef", + run.parent_run_id AS "parentRunId", + run.trigger_type AS "triggerType", + run.execution_owner AS "executionOwner", + run.input_ref AS "inputRef", run.priority, + run.status AS "runStatus", run.version AS "runVersion", + attempt.executor_type AS "attemptExecutorType" + FROM "ql3"."runs" AS run + LEFT JOIN LATERAL ( + SELECT executor_type + FROM "ql3"."run_attempts" + WHERE run_id = run.id + ORDER BY attempt DESC + LIMIT 1 + ) AS attempt ON true + WHERE run.id = $1 + FOR UPDATE OF run + `, + [command.sourceRunId], + ); + if ( + result.rows.length === 0 || + result.rows[0]?.projectId !== command.projectId + ) { + throw new RunManualRetryNotFoundError(); + } + if (result.rows.length !== 1) throw unavailable(); + const row = result.rows[0]!; + const status = sourceStatus(text(row, 'runStatus')); + if ( + status !== command.expectedRunStatus || + integer(row, 'runVersion') !== command.expectedRunVersion + ) { + throw new RunManualRetryFenceRejectedError('source_changed'); + } + const taskRevision = text(row, 'taskRevision'); + const taskSnapshotRef = optionalText(row, 'taskSnapshotRef'); + if ( + text(row, 'executionOwner') !== 'runtime' || + optionalText(row, 'parentRunId') !== undefined || + text(row, 'triggerType') === 'plugin_package_workflow' || + taskSnapshotRef === undefined || + taskSnapshotRef !== taskRevision || + optionalText(row, 'attemptExecutorType') !== 'remote_worker' || + !TASK_REVISION_PATTERN.test(taskRevision) + ) { + throw new RunManualRetryFenceRejectedError('source_not_retryable'); + } + const taskName = optionalText(row, 'taskName'); + const inputRef = optionalText(row, 'inputRef'); + return Object.freeze({ + taskId: text(row, 'taskId'), + taskRevision, + ...(taskName === undefined ? {} : { taskName }), + taskSnapshotRef, + ...(inputRef === undefined ? {} : { inputRef }), + priority: integer(row, 'priority'), + }); +} + +async function confirmTaskAndExecution( + client: PostgresClient, + projectId: string, + source: Readonly, +): Promise { + const task = await client.query( + ` + SELECT revision.enabled + FROM "ql3"."task_definitions" AS head + JOIN "ql3"."task_definition_revisions" AS revision + ON revision.project_id = head.project_id + AND revision.task_id = head.task_id + AND revision.revision = head.current_revision + WHERE head.project_id = $1 AND head.task_id = $2 + `, + [projectId, source.taskId], + ); + if ( + task.rows.length !== 1 || + !postgresRequiredBoolean(task.rows[0]!.enabled, unavailable) + ) { + throw new RunManualRetryFenceRejectedError('task_disabled'); + } + const execution = await client.query( + ` + SELECT source_content_digest AS "sourceContentDigest", + content_digest AS "contentDigest" + FROM "ql3"."task_execution_revisions" + WHERE project_id = $1 AND task_id = $2 AND task_revision = $3 + AND executor_type = 'remote_worker' + LIMIT 2 + `, + [projectId, source.taskId, source.taskRevision], + ); + if (execution.rows.length !== 1) { + throw new RunManualRetryFenceRejectedError('source_not_retryable'); + } + const match = TASK_REVISION_PATTERN.exec(source.taskRevision); + const row = execution.rows[0]!; + const sourceContentDigest = text(row, 'sourceContentDigest'); + const contentDigest = text(row, 'contentDigest'); + if ( + !match || + match[2] !== sourceContentDigest || + !/^[0-9a-f]{64}$/.test(contentDigest) + ) { + throw new RunManualRetryFenceRejectedError('source_not_retryable'); + } + return Object.freeze({ contentDigest, sourceContentDigest }); +} + +async function consumeRateLimit( + client: PostgresClient, + command: Readonly, + observedAtMs: number, +): Promise { + const threshold = Math.max( + 0, + observedAtMs - CLUSTER_RUN_MANUAL_RETRY_RATE_WINDOW_MS, + ); + const result = await client.query( + ` + SELECT created_at_ms AS "createdAtMs" + FROM "ql3"."runs" + WHERE project_id = $1 AND trigger_type = 'run_manual_retry' + AND execution_origin = 'manual' AND triggered_by = $2 + AND created_at_ms > $3 + ORDER BY created_at_ms DESC, id DESC + LIMIT $4 + `, + [ + command.projectId, + command.principal.subject.id, + threshold, + CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + ], + ); + if (result.rows.length < CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT) return; + const earliestAtMs = integer( + result.rows[result.rows.length - 1]!, + 'createdAtMs', + ); + throw new RunManualRetryRateLimitedError( + Math.max( + 1, + earliestAtMs + CLUSTER_RUN_MANUAL_RETRY_RATE_WINDOW_MS - observedAtMs, + ), + ); +} + +async function insertRetry( + client: PostgresClient, + command: Readonly, + source: Readonly, + execution: Readonly, + observedAtMs: number, +): Promise { + await client.query( + ` + INSERT INTO "ql3"."runs" ( + id, project_id, task_id, task_revision, task_name, + task_snapshot_ref, retry_of_run_id, trigger_type, execution_origin, + execution_owner, triggered_by, request_id, status, version, + event_sequence, priority, idempotency_key, input_ref, + created_at_ms, queued_at_ms + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7, 'run_manual_retry', 'manual', + 'runtime', $8, $9, 'queued', 2, 2, $10, $11, $12, $13, $13 + ) + `, + [ + command.runId, + command.projectId, + source.taskId, + source.taskRevision, + source.taskName ?? null, + source.taskSnapshotRef, + command.sourceRunId, + command.principal.subject.id, + command.mutationId, + source.priority, + `ql3:run-manual-retry:v1:${command.mutationId}`, + source.inputRef ?? null, + observedAtMs, + ], + ); + await client.query( + ` + INSERT INTO "ql3"."run_attempts" ( + id, run_id, attempt, status, executor_type, + callback_sequence, created_at_ms + ) VALUES ($1, $2, 1, 'claimed', 'remote_worker', 0, $3) + `, + [command.attemptId, command.runId, observedAtMs], + ); + await client.query( + ` + INSERT INTO "ql3"."run_events" ( + id, run_id, sequence, type, dedupe_key, actor_type, actor_id, + attempt_id, payload, created_at_ms + ) VALUES ($1, $2, 1, 'run.created', $3, $4, $5, $6, $7::jsonb, $8) + `, + [ + command.createdEventId, + command.runId, + `run-manual-retry-created:${command.mutationId}`, + command.principal.subject.type, + command.principal.subject.id, + command.attemptId, + JSON.stringify({ + status: 'created', + version: 1, + execution_owner: 'runtime', + executor_type: 'remote_worker', + execution_revision_digest: execution.contentDigest, + source_content_digest: execution.sourceContentDigest, + retry_of_run_id: command.sourceRunId, + source_run_status: command.expectedRunStatus, + source_run_version: command.expectedRunVersion, + inherit_retry_policy: false, + mutation_id: command.mutationId, + authentication_id: command.principal.authenticationId, + audit_event_id: command.auditEventId, + policy_fence: { + project_version: command.policyFence.projectVersion, + binding_version: command.policyFence.bindingVersion, + }, + }), + observedAtMs, + ], + ); + await client.query( + ` + INSERT INTO "ql3"."run_events" ( + id, run_id, sequence, type, dedupe_key, actor_type, actor_id, + attempt_id, payload, created_at_ms + ) VALUES ($1, $2, 2, 'run.queued', $3, $4, $5, $6, $7::jsonb, $8) + `, + [ + command.queuedEventId, + command.runId, + `run-manual-retry-queued:${command.mutationId}`, + command.principal.subject.type, + command.principal.subject.id, + command.attemptId, + JSON.stringify({ + from_status: 'created', + to_status: 'queued', + version: 2, + }), + observedAtMs, + ], + ); + await client.query( + ` + INSERT INTO "ql3"."security_audit_events" ( + event_id, request_id, operation_id, project_id, + subject_type, subject_id, authentication_id, outcome, reasons, + project_version, binding_version, occurred_at_ms + ) VALUES ( + $1, $2, 'run.retry', $3, $4, $5, $6, 'allowed', $7::jsonb, + $8, $9, $10 + ) + `, + [ + command.auditEventId, + command.requestId, + command.projectId, + command.principal.subject.type, + command.principal.subject.id, + command.principal.authenticationId, + JSON.stringify(['role_grant', 'strong_authentication']), + command.policyFence.projectVersion, + command.policyFence.bindingVersion, + observedAtMs, + ], + ); +} + +/** + * PostgreSQL authority for one strongly authenticated manual retry. The + * Project row lock serializes the existing Run-ledger quota across replicas. + */ +export class PostgresRunManualRetryRepository + implements RunManualRetryRepository +{ + constructor(private readonly pool: PostgresPool) { + if (!pool || typeof pool.connect !== 'function') { + throw new TypeError('PostgreSQL Run manual retry pool is invalid'); + } + } + + async retryRun( + value: Readonly, + ): Promise> { + const command = normalizeRunManualRetryCommand(value); + for ( + let transactionAttempt = 0; + transactionAttempt < POSTGRES_DEFINITION_TRANSACTION_ATTEMPTS; + transactionAttempt += 1 + ) { + let client: PostgresClient; + try { + client = await this.pool.connect(); + } catch (error) { + throw unavailable({ cause: error }); + } + let began = false; + try { + await configurePostgresDefinitionTransaction(client); + began = true; + const observedAtMs = await databaseNow(client); + confirmStrongAuthentication(command, observedAtMs); + await confirmAuthorization(client, command); + + const replay = await findReplay(client, command); + if (replay) { + const result = replayResult(command, replay); + await client.query('COMMIT'); + began = false; + return result; + } + + const source = await findSource(client, command); + const execution = await confirmTaskAndExecution( + client, + command.projectId, + source, + ); + await consumeRateLimit(client, command, observedAtMs); + await insertRetry(client, command, source, execution, observedAtMs); + const result = normalizeRunManualRetryResult({ + status: 'accepted', + projectId: command.projectId, + sourceRunId: command.sourceRunId, + sourceRunStatus: command.expectedRunStatus, + sourceRunVersion: command.expectedRunVersion, + runId: command.runId, + retryOfRunId: command.sourceRunId, + taskId: source.taskId, + taskRevision: source.taskRevision, + attemptId: command.attemptId, + runStatus: 'queued', + runVersion: 2, + eventSequence: 2, + executorType: 'remote_worker', + executionRevisionDigest: execution.contentDigest, + createdAtMs: observedAtMs, + }); + await client.query('COMMIT'); + began = false; + return result; + } catch (error) { + if (began) await rollbackPostgresDefinitionTransaction(client); + const state = postgresSqlState(error); + if ( + (state === '23505' || + POSTGRES_DEFINITION_RETRYABLE_SQL_STATES.has(state ?? '')) && + transactionAttempt + 1 < POSTGRES_DEFINITION_TRANSACTION_ATTEMPTS + ) { + continue; + } + if ( + error instanceof InvalidRunManualRetryError || + error instanceof RunManualRetryNotFoundError || + error instanceof RunManualRetryFenceRejectedError || + error instanceof RunManualRetryRateLimitedError || + error instanceof RunManualRetryUnavailableError + ) { + throw error; + } + throw unavailable({ cause: error }); + } finally { + client.release(); + } + } + throw unavailable(); + } +} diff --git a/packages/ql3-cluster-postgres/test/runManualRetryRepository.test.cjs b/packages/ql3-cluster-postgres/test/runManualRetryRepository.test.cjs new file mode 100644 index 00000000..b72806c5 --- /dev/null +++ b/packages/ql3-cluster-postgres/test/runManualRetryRepository.test.cjs @@ -0,0 +1,334 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { + RunManualRetryFenceRejectedError, + RunManualRetryRateLimitedError, +} = require('@qinglong/runtime-core/run-manual-retry'); +const { + CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + PostgresRunManualRetryRepository, +} = require('@qinglong/cluster-postgres/run-manual-retry'); + +const SOURCE_DIGEST = 'a'.repeat(64); +const EXECUTION_DIGEST = 'b'.repeat(64); +const TASK_REVISION = `qltd:v1:7:${SOURCE_DIGEST}`; +const IDS = Object.freeze({ + mutationId: '019f9200-0000-4000-8000-000000000001', + runId: '019f9200-0000-4000-8000-000000000002', + attemptId: '019f9200-0000-4000-8000-000000000003', + createdEventId: '019f9200-0000-4000-8000-000000000004', + queuedEventId: '019f9200-0000-4000-8000-000000000005', + auditEventId: '019f9200-0000-4000-8000-000000000006', +}); + +function command(overrides = {}) { + return { + projectId: 'project-1', + sourceRunId: 'source-run-1', + mutationId: IDS.mutationId, + expectedRunVersion: 7, + expectedRunStatus: 'failed', + runId: IDS.runId, + attemptId: IDS.attemptId, + createdEventId: IDS.createdEventId, + queuedEventId: IDS.queuedEventId, + auditEventId: IDS.auditEventId, + requestId: 'cluster-run-retry-1', + principal: { + subject: { type: 'user', id: 'operator-1' }, + authenticationId: 'oidc:mfa-session-1', + authenticatedAtMs: 900_000, + expiresAtMs: 1_100_000, + assurance: 'multi_factor', + }, + policyFence: { projectVersion: 2, bindingVersion: 3 }, + ...overrides, + }; +} + +function sourceRow(overrides = {}) { + return { + projectId: 'project-1', + taskId: 'task-1', + taskRevision: TASK_REVISION, + taskName: 'Task 1', + taskSnapshotRef: TASK_REVISION, + parentRunId: null, + triggerType: 'task_start', + executionOwner: 'runtime', + inputRef: 'artifact:input-1', + priority: 3, + runStatus: 'failed', + runVersion: 7, + attemptExecutorType: 'remote_worker', + ...overrides, + }; +} + +function replayRow(overrides = {}) { + return { + runId: IDS.runId, + projectId: 'project-1', + retryOfRunId: 'source-run-1', + taskId: 'task-1', + taskRevision: TASK_REVISION, + triggerType: 'run_manual_retry', + executionOrigin: 'manual', + executionOwner: 'runtime', + triggeredBy: 'operator-1', + requestId: IDS.mutationId, + runStatus: 'queued', + runVersion: 2, + eventSequence: 2, + createdAtMs: 1_000_000, + attemptId: IDS.attemptId, + executorType: 'remote_worker', + createdActorType: 'user', + createdActorId: 'operator-1', + createdPayload: { + mutation_id: IDS.mutationId, + retry_of_run_id: 'source-run-1', + source_run_status: 'failed', + source_run_version: 7, + inherit_retry_policy: false, + authentication_id: 'oidc:mfa-session-1', + audit_event_id: IDS.auditEventId, + execution_revision_digest: EXECUTION_DIGEST, + policy_fence: { project_version: 2, binding_version: 3 }, + }, + queuedActorType: 'user', + queuedActorId: 'operator-1', + queuedPayload: { + from_status: 'created', + to_status: 'queued', + version: 2, + }, + ...overrides, + }; +} + +function fixture(options = {}) { + const calls = []; + let connections = 0; + const pool = { + async connect() { + connections += 1; + const connection = connections; + return { + async query(sql, params = []) { + const normalized = sql.replace(/\s+/g, ' ').trim(); + calls.push({ connection, sql: normalized, params }); + if (normalized.startsWith('BEGIN')) { + if (options.failFirstBegin && connection === 1) { + const error = new Error('serialization retry'); + error.code = '40001'; + throw error; + } + return { rows: [], rowCount: 0 }; + } + if ( + normalized === 'COMMIT' || + normalized === 'ROLLBACK' || + normalized.startsWith('SELECT set_config') + ) + return { rows: [], rowCount: 0 }; + if (normalized.includes('statement_timestamp()')) { + return { rows: [{ nowMs: 1_000_000 }], rowCount: 1 }; + } + if (normalized.includes('FROM "ql3"."projects"')) { + const rows = options.projectRows ?? [ + { projectStatus: 'active', projectVersion: 2 }, + ]; + return { rows, rowCount: rows.length }; + } + if (normalized.includes('project_role_bindings')) { + const rows = options.bindingRows ?? [ + { + bindingVersion: 3, + bindingState: 'active', + bindingRole: 'operator', + }, + ]; + return { rows, rowCount: rows.length }; + } + if (normalized.includes('idempotency_key = $2')) { + const rows = options.replayRows ?? []; + return { rows, rowCount: rows.length }; + } + if (normalized.includes('WHERE run.id = $1')) { + const rows = options.sourceRows ?? [sourceRow()]; + return { rows, rowCount: rows.length }; + } + if (normalized.includes('FROM "ql3"."task_definitions"')) { + const rows = options.taskRows ?? [{ enabled: true }]; + return { rows, rowCount: rows.length }; + } + if (normalized.includes('task_execution_revisions')) { + const rows = options.executionRows ?? [ + { + sourceContentDigest: SOURCE_DIGEST, + contentDigest: EXECUTION_DIGEST, + }, + ]; + return { rows, rowCount: rows.length }; + } + if ( + normalized.includes("trigger_type = 'run_manual_retry'") && + normalized.startsWith('SELECT') + ) { + const rows = options.rateRows ?? []; + return { rows, rowCount: rows.length }; + } + if (normalized.startsWith('INSERT INTO')) { + return { rows: [], rowCount: 1 }; + } + throw new Error(`Unexpected SQL: ${normalized}`); + }, + release() { + calls.push({ connection, sql: 'RELEASE', params: [] }); + }, + }; + }, + }; + return { + calls, + repository: new PostgresRunManualRetryRepository(pool), + }; +} + +test('atomically appends a linked queued Run, remote Attempt, events and allowed audit', async () => { + const { calls, repository } = fixture(); + assert.deepEqual(await repository.retryRun(command()), { + status: 'accepted', + projectId: 'project-1', + sourceRunId: 'source-run-1', + sourceRunStatus: 'failed', + sourceRunVersion: 7, + runId: IDS.runId, + retryOfRunId: 'source-run-1', + taskId: 'task-1', + taskRevision: TASK_REVISION, + attemptId: IDS.attemptId, + runStatus: 'queued', + runVersion: 2, + eventSequence: 2, + executorType: 'remote_worker', + executionRevisionDigest: EXECUTION_DIGEST, + createdAtMs: 1_000_000, + }); + const inserts = calls.filter(({ sql }) => sql.startsWith('INSERT INTO')); + assert.equal(inserts.length, 5); + assert.match(inserts.at(-1).sql, /security_audit_events/); + assert.deepEqual(JSON.parse(inserts[2].params[6]), { + status: 'created', + version: 1, + execution_owner: 'runtime', + executor_type: 'remote_worker', + execution_revision_digest: EXECUTION_DIGEST, + source_content_digest: SOURCE_DIGEST, + retry_of_run_id: 'source-run-1', + source_run_status: 'failed', + source_run_version: 7, + inherit_retry_policy: false, + mutation_id: IDS.mutationId, + authentication_id: 'oidc:mfa-session-1', + audit_event_id: IDS.auditEventId, + policy_fence: { project_version: 2, binding_version: 3 }, + }); + assert.equal( + calls.some(({ sql }) => sql === 'COMMIT'), + true, + ); +}); + +test('returns durable identities for an exact replay without appending again', async () => { + const { calls, repository } = fixture({ replayRows: [replayRow()] }); + const result = await repository.retryRun( + command({ + runId: '019f9200-0000-4000-8000-000000000102', + attemptId: '019f9200-0000-4000-8000-000000000103', + createdEventId: '019f9200-0000-4000-8000-000000000104', + queuedEventId: '019f9200-0000-4000-8000-000000000105', + }), + ); + assert.equal(result.status, 'existing'); + assert.equal(result.runId, IDS.runId); + assert.equal(result.attemptId, IDS.attemptId); + assert.equal( + calls.some(({ sql }) => sql.startsWith('INSERT INTO')), + false, + ); +}); + +test('rejects stale authentication and changed authorization inside the transaction', async () => { + const stale = fixture(); + await assert.rejects( + stale.repository.retryRun( + command({ + principal: { + ...command().principal, + authenticatedAtMs: 600_000, + }, + }), + ), + (error) => + error instanceof RunManualRetryFenceRejectedError && + error.reason === 'authentication_changed', + ); + assert.equal( + stale.calls.some(({ sql }) => sql === 'ROLLBACK'), + true, + ); + assert.equal( + stale.calls.some(({ sql }) => sql.includes('FROM "ql3"."projects"')), + false, + ); + + const changed = fixture({ + bindingRows: [ + { bindingVersion: 4, bindingState: 'active', bindingRole: 'operator' }, + ], + }); + await assert.rejects( + changed.repository.retryRun(command()), + (error) => + error instanceof RunManualRetryFenceRejectedError && + error.reason === 'authorization_changed', + ); +}); + +test('uses the durable Run ledger for the per-project User quota', async () => { + const rateRows = Array.from( + { length: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT }, + (_, index) => ({ createdAtMs: 999_000 - index }), + ); + const { calls, repository } = fixture({ rateRows }); + await assert.rejects( + repository.retryRun(command()), + (error) => + error instanceof RunManualRetryRateLimitedError && + error.retryAfterMs === 58_937, + ); + assert.equal( + calls.some(({ sql }) => sql.startsWith('INSERT INTO')), + false, + ); + assert.equal( + calls.some(({ sql }) => sql === 'ROLLBACK'), + true, + ); +}); + +test('retries one serializable conflict with a fresh connection', async () => { + const { calls, repository } = fixture({ failFirstBegin: true }); + const result = await repository.retryRun(command()); + assert.equal(result.status, 'accepted'); + assert.deepEqual( + [...new Set(calls.map(({ connection }) => connection))], + [1, 2], + ); + assert.equal(calls.filter(({ sql }) => sql === 'RELEASE').length, 2); +}); diff --git a/scripts/ql3-postgres-ha-contract.cjs b/scripts/ql3-postgres-ha-contract.cjs index 8d0f0c23..d18ae233 100644 --- a/scripts/ql3-postgres-ha-contract.cjs +++ b/scripts/ql3-postgres-ha-contract.cjs @@ -168,6 +168,16 @@ const { PostgresWorkerCredentialAdministrationRepository, assertPostgresAdminSchemaReady, } = require('../packages/ql3-cluster-postgres/dist/entrypoints/admin.js'); +const { + PostgresTaskStartRepository, +} = require('../packages/ql3-cluster-postgres/dist/task-start/taskStartRepository.js'); +const { + CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + PostgresRunManualRetryRepository, +} = require('../packages/ql3-cluster-postgres/dist/run-management/runManualRetryRepository.js'); +const { + RunManualRetryRateLimitedError, +} = require('../packages/ql3-runtime-core/dist/run/manual-retry/runManualRetry.js'); const { assertPostgresPackageManagerSchemaReady, PostgresPluginPackageIdentityKeysetLedgerConflictError, @@ -9347,6 +9357,288 @@ async function verifyRunAttemptLogRetentionAfterPromotion({ } } +async function manualRunRetryFacts(pool, fixture) { + const result = await pool.query( + `SELECT + (SELECT status FROM "ql3"."runs" WHERE id = $1) AS "sourceStatus", + (SELECT count(*)::integer FROM "ql3"."runs" + WHERE project_id = $2 AND trigger_type = 'run_manual_retry' + AND retry_of_run_id = $1) AS "retryRunCount", + (SELECT count(*)::integer FROM "ql3"."run_attempts" AS attempt + JOIN "ql3"."runs" AS run ON run.id = attempt.run_id + WHERE run.project_id = $2 + AND run.trigger_type = 'run_manual_retry' + AND attempt.executor_type = 'remote_worker') AS "attemptCount", + (SELECT count(*)::integer FROM "ql3"."run_events" AS event + JOIN "ql3"."runs" AS run ON run.id = event.run_id + WHERE run.project_id = $2 + AND run.trigger_type = 'run_manual_retry') AS "eventCount", + (SELECT count(*)::integer FROM "ql3"."security_audit_events" + WHERE project_id = $2 AND operation_id = 'run.retry' + AND outcome = 'allowed') AS "allowedAuditCount", + (SELECT count(*)::integer FROM "ql3"."run_retry_policies" AS policy + JOIN "ql3"."runs" AS run ON run.id = policy.run_id + WHERE run.project_id = $2 + AND run.trigger_type = 'run_manual_retry') AS "retryPolicyCount"`, + [fixture.sourceRunId, fixture.projectId], + ); + assert.equal(result.rowCount, 1); + return result.rows[0]; +} + +async function runManualRunRetryHaEvidence(options) { + const { primaryPort, primaryDatabase, standbyDatabase } = options; + const suffix = `${process.pid}-${randomBytes(3).toString('hex')}`; + const fixture = Object.freeze({ + projectId: `ha-manual-retry-${suffix}`, + actorId: `ha-manual-retry-operator-${suffix}`, + taskId: `ha-manual-retry-task-${suffix}`, + sourceRunId: randomUUID(), + sourceAttemptId: randomUUID(), + }); + const clock = await primaryDatabase.pool.query( + `SELECT floor(extract(epoch FROM statement_timestamp()) * 1000)::bigint + AS "nowMs"`, + ); + const nowMs = Number(clock.rows[0].nowMs); + await primaryDatabase.pool.query( + `INSERT INTO "ql3"."projects" ( + id, name, slug, status, version, created_at_ms, updated_at_ms + ) VALUES ($1, 'HA manual Run retry', $1, 'active', 1, $2, $2)`, + [fixture.projectId, nowMs], + ); + await primaryDatabase.pool.query( + `INSERT INTO "ql3"."project_role_bindings" ( + project_id, subject_type, subject_id, version, state, role, + mutation_id, changed_by_type, changed_by_id, created_at_ms + ) VALUES ($1, 'user', $2, 1, 'active', 'operator', $3, + 'system', 'ha-contract', $4)`, + [ + fixture.projectId, + fixture.actorId, + `ha-manual-retry-binding-${suffix}`, + nowMs, + ], + ); + const task = ( + await new PostgresTaskDefinitionRepository( + primaryDatabase.pool, + ).appendTaskDefinitionRevision({ + projectId: fixture.projectId, + taskId: fixture.taskId, + expectedRevision: null, + mutationId: randomUUID(), + name: 'HA manual Run retry source', + kind: 'command', + spec: { + schema: 'qinglong/command@v1', + config: { + command: { + kind: 'argv', + file: '/bin/echo', + args: ['manual-retry-ha'], + }, + }, + }, + labels: {}, + enabled: true, + occurredAtMs: nowMs, + }) + ).definition; + const firstRuntime = await databaseOpener( + 'runtime', + databaseUrl(RUNTIME_USER, RUNTIME_PASSWORD, primaryPort), + 'ql3-ha-manual-run-retry-a', + )(); + const secondRuntime = await databaseOpener( + 'runtime', + databaseUrl(RUNTIME_USER, RUNTIME_PASSWORD, primaryPort), + 'ql3-ha-manual-run-retry-b', + )(); + try { + const source = await new PostgresTaskStartRepository( + primaryDatabase.pool, + ).startTask({ + projectId: fixture.projectId, + taskId: fixture.taskId, + mutationId: randomUUID(), + expectedRevision: task.revision, + expectedContentDigest: task.contentDigest, + runId: fixture.sourceRunId, + attemptId: fixture.sourceAttemptId, + createdEventId: randomUUID(), + queuedEventId: randomUUID(), + subject: { type: 'user', id: fixture.actorId }, + policyFence: { projectVersion: 1, bindingVersion: 1 }, + }); + await primaryDatabase.pool.query('BEGIN'); + try { + await primaryDatabase.pool.query( + `UPDATE "ql3"."runs" + SET status = 'failed', version = 3, event_sequence = 3, + finished_at_ms = $2, error_code = 'HA_SOURCE_FAILURE', + error_summary = 'terminal source for manual retry' + WHERE id = $1`, + [fixture.sourceRunId, nowMs + 1], + ); + await primaryDatabase.pool.query( + `UPDATE "ql3"."run_attempts" + SET status = 'failed', finished_at_ms = $2, + error_code = 'HA_SOURCE_FAILURE', + error_summary = 'terminal source for manual retry' + WHERE id = $1`, + [fixture.sourceAttemptId, nowMs + 1], + ); + await primaryDatabase.pool.query( + `INSERT INTO "ql3"."run_events" ( + id, run_id, sequence, type, dedupe_key, actor_type, actor_id, + attempt_id, payload, created_at_ms + ) VALUES ($1, $2, 3, 'run.failed', $3, 'executor', 'ha-contract', + $4, $5::jsonb, $6)`, + [ + randomUUID(), + fixture.sourceRunId, + `ha-manual-retry-source-failed-${suffix}`, + fixture.sourceAttemptId, + JSON.stringify({ + from_status: 'queued', + to_status: 'failed', + version: 3, + error_code: 'HA_SOURCE_FAILURE', + }), + nowMs + 1, + ], + ); + await primaryDatabase.pool.query('COMMIT'); + } catch (error) { + await primaryDatabase.pool.query('ROLLBACK'); + throw error; + } + + const authentication = Object.freeze({ + subject: { type: 'user', id: fixture.actorId }, + authenticationId: `oidc:mfa-${suffix}`, + authenticatedAtMs: nowMs, + expiresAtMs: nowMs + 60 * 60_000, + assurance: 'multi_factor', + }); + let commandIndex = 0; + const retryCommand = () => { + commandIndex += 1; + return { + projectId: fixture.projectId, + sourceRunId: fixture.sourceRunId, + mutationId: randomUUID(), + expectedRunVersion: 3, + expectedRunStatus: 'failed', + runId: randomUUID(), + attemptId: randomUUID(), + createdEventId: randomUUID(), + queuedEventId: randomUUID(), + auditEventId: randomUUID(), + requestId: `ha-manual-retry-${suffix}-${commandIndex}`, + principal: authentication, + policyFence: { projectVersion: 1, bindingVersion: 1 }, + }; + }; + const firstRepository = new PostgresRunManualRetryRepository( + firstRuntime.pool, + ); + const secondRepository = new PostgresRunManualRetryRepository( + secondRuntime.pool, + ); + const replayCommand = retryCommand(); + const exactRace = await Promise.all([ + firstRepository.retryRun(replayCommand), + secondRepository.retryRun(replayCommand), + ]); + assert.deepEqual(exactRace.map(({ status }) => status).sort(), [ + 'accepted', + 'existing', + ]); + assert.equal(exactRace[0].runId, exactRace[1].runId); + assert.equal(exactRace[0].attemptId, exactRace[1].attemptId); + + for ( + let index = 1; + index < CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT - 1; + index += 1 + ) { + const repository = index % 2 === 0 ? firstRepository : secondRepository; + assert.equal( + (await repository.retryRun(retryCommand())).status, + 'accepted', + ); + } + const quotaRace = await Promise.allSettled([ + firstRepository.retryRun(retryCommand()), + secondRepository.retryRun(retryCommand()), + ]); + const accepted = quotaRace.filter(({ status }) => status === 'fulfilled'); + const rejected = quotaRace.filter(({ status }) => status === 'rejected'); + assert.equal(accepted.length, 1); + assert.equal(rejected.length, 1); + assert.equal(accepted[0].value.status, 'accepted'); + assert.equal( + rejected[0].reason instanceof RunManualRetryRateLimitedError, + true, + ); + assert.ok(rejected[0].reason.retryAfterMs > 0); + + const expectedFacts = { + sourceStatus: 'failed', + retryRunCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + attemptCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + eventCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT * 2, + allowedAuditCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + retryPolicyCount: 0, + }; + assert.deepEqual( + await manualRunRetryFacts(primaryDatabase.pool, fixture), + expectedFacts, + ); + await waitFor(async () => { + const facts = await manualRunRetryFacts(standbyDatabase.pool, fixture); + return facts.retryRunCount === CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT + ? facts + : null; + }, 'manual Run retry WAL replay'); + assert.deepEqual( + await manualRunRetryFacts(standbyDatabase.pool, fixture), + expectedFacts, + ); + return { + fixture, + report: { + sourceRunId: source.runId, + exactConcurrentReplay: true, + crossReplicaQuotaSerialized: true, + acceptedRetryRuns: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + rejectedOverQuota: 1, + inheritedRetryPolicies: 0, + allowedAuditEvents: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + replicatedBeforePromotion: true, + survivedPromotion: false, + }, + }; + } finally { + await Promise.all([firstRuntime.close(), secondRuntime.close()]); + } +} + +async function verifyManualRunRetryAfterPromotion(options) { + const { promotedPool, evidence } = options; + assert.deepEqual(await manualRunRetryFacts(promotedPool, evidence.fixture), { + sourceStatus: 'failed', + retryRunCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + attemptCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + eventCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT * 2, + allowedAuditCount: CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + retryPolicyCount: 0, + }); + evidence.report.survivedPromotion = true; +} + async function main(argv = process.argv.slice(2)) { const reportFile = privateReportPath(argv); const nodeMajor = Number(process.versions.node.split('.')[0]); @@ -9432,6 +9724,7 @@ async function main(argv = process.argv.slice(2)) { let modelProviderCredentialTestConnection; let runAttemptLogRetentionEvidence; let runAttemptLogRetention; + let manualRunRetry; const startedAt = performance.now(); const timeline = []; let report; @@ -9867,6 +10160,15 @@ async function main(argv = process.argv.slice(2)) { state: 'synchronous_remote_apply_ready', atMs: Number((performance.now() - startedAt).toFixed(3)), }); + manualRunRetry = await runManualRunRetryHaEvidence({ + primaryPort, + primaryDatabase, + standbyDatabase, + }); + timeline.push({ + state: 'manual_run_retry_replicated', + atMs: Number((performance.now() - startedAt).toFixed(3)), + }); modelProviderCredentialTestConnection = await runModelProviderCredentialTestConnectionHaEvidence({ primaryPort, @@ -10828,6 +11130,14 @@ async function main(argv = process.argv.slice(2)) { promotedPool: promotedDatabase.pool, report: modelProviderCredentialTestConnection, }); + await verifyManualRunRetryAfterPromotion({ + promotedPool: promotedDatabase.pool, + evidence: manualRunRetry, + }); + timeline.push({ + state: 'manual_run_retry_survived_promotion', + atMs: Number((performance.now() - startedAt).toFixed(3)), + }); timeline.push({ state: 'optional_ai_feature_schema_survived_promotion', atMs: Number((performance.now() - startedAt).toFixed(3)), @@ -11560,7 +11870,7 @@ async function main(argv = process.argv.slice(2)) { FROM "ql3"."worker_credential_deliveries") AS "credentialDeliveries"`, ); assert.deepEqual(sideEffects.rows, [ - { runs: 11, runEvents: 39, credentialDeliveries: 4 }, + { runs: 76, runEvents: 170, credentialDeliveries: 4 }, ]); timeline.push({ state: 'two_fresh_control_replicas_ready', @@ -11628,6 +11938,7 @@ async function main(argv = process.argv.slice(2)) { ...sideEffects.rows[0], unexpectedDomainSideEffects: 0, }, + manualRunRetry: manualRunRetry.report, transactionWindows: { ambiguousCommit: { clientObservedFailure: ambiguousCommitClientRejected, @@ -11981,6 +12292,19 @@ async function main(argv = process.argv.slice(2)) { toolResultCatalogCommitResponseLossConvergesExactlyOnce: true, toolResultCompletionCommitResponseLossRecoversWithoutReexecution: true, toolResultRekeyCommitResponseLossConvergesExactlyOnce: true, + manualRunRetryConcurrentReplayIsExact: + manualRunRetry.report.exactConcurrentReplay, + manualRunRetryQuotaIsSerializedAcrossReplicas: + manualRunRetry.report.crossReplicaQuotaSerialized, + manualRunRetryDoesNotInheritAutomaticPolicy: + manualRunRetry.report.inheritedRetryPolicies === 0, + manualRunRetryAllowedAuditIsDurable: + manualRunRetry.report.allowedAuditEvents === + CLUSTER_RUN_MANUAL_RETRY_RATE_LIMIT, + manualRunRetryReplicatesBeforePromotion: + manualRunRetry.report.replicatedBeforePromotion, + manualRunRetrySurvivesPromotion: + manualRunRetry.report.survivedPromotion, physicalStreaming: true, oldPrimaryFencedBeforePromotion: true, bothOldReplicasNotReady: true, diff --git a/test/back/ql3PackageBoundaryAudit.test.cjs b/test/back/ql3PackageBoundaryAudit.test.cjs index 00c7e74c..e94c6624 100644 --- a/test/back/ql3PackageBoundaryAudit.test.cjs +++ b/test/back/ql3PackageBoundaryAudit.test.cjs @@ -421,10 +421,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( rootSourceFileRoles: clusterPostgres.rootSourceFileRoles, }, { - sourceFiles: 149, + sourceFiles: 150, rootSourceFiles: 1, rootSourceLines: 125, - nestedSourceFiles: 148, + nestedSourceFiles: 149, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, );