From 40628c843d1ffad0cdd548ba320d5b4383bb4f5a Mon Sep 17 00:00:00 2001 From: whyour Date: Wed, 12 Aug 2026 03:38:27 +0800 Subject: [PATCH] feat(ql3): add cluster log retention authority --- docs/QINGLONG_3_0_ARCHITECTURE_RFC.md | 11 + ...-0379-cluster-run-attempt-log-retention.md | 74 ++ docs/adr/README.md | 4 + docs/ql3-package-boundaries.json | 2 +- .../test/application.test.cjs | 2 + .../test/bootstrap.test.cjs | 2 + .../src/entrypoints/runtime.ts | 1 + .../src/migration/migrationManifest.ts | 5 + .../src/migrations/index.ts | 2 + .../pg-0055-run-attempt-log-retention.ts | 189 +++++ .../runAttemptLogRetentionClaimRepository.ts | 746 ++++++++++++++++++ .../ql3-cluster-postgres/src/schema/schema.ts | 158 ++++ .../src/schema/schemaContract.ts | 74 +- .../src/schema/schemaReadiness.ts | 24 + .../postgresqlMigrationDefinitions.test.cjs | 29 + .../test/postgresqlSchemaReadiness.test.cjs | 23 +- ...ttemptLogRetentionClaimRepository.test.cjs | 330 ++++++++ packages/ql3-runtime-core/package.json | 8 + .../clusterRunAttemptLogRetention.ts | 62 ++ test/back/ql3PackageBoundaryAudit.test.cjs | 10 +- 20 files changed, 1737 insertions(+), 19 deletions(-) create mode 100644 docs/adr/ADR-0379-cluster-run-attempt-log-retention.md create mode 100644 packages/ql3-cluster-postgres/src/migrations/pg-0055-run-attempt-log-retention.ts create mode 100644 packages/ql3-cluster-postgres/src/run/runAttemptLogRetentionClaimRepository.ts create mode 100644 packages/ql3-cluster-postgres/test/runAttemptLogRetentionClaimRepository.test.cjs create mode 100644 packages/ql3-runtime-core/src/run/log-retention/clusterRunAttemptLogRetention.ts diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index bbac469c..8748491f 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -11,6 +11,17 @@ 最新增量证据（2026-08-12）： +- D-291/ADR-0379（进行中，PostgreSQL authority 阶段完成） + Cluster Run Attempt 日志 retention 第一阶段已冻结并实现多副本权威边界，且没有新增 package：共享 claim contract 位于既有 + Runtime Core Run log-retention 目录，PostgreSQL v54 增加 durable control 与 immutable tombstone、terminal remote Worker + candidate index 和 `run_attempt_log_retention` capability。候选只允许 runtime-owned、Run/Attempt 双终态、非 lost、canonical + `remote_worker/wlog-*` 且超过 cutoff；每批最多 16 条，通过短 `READ COMMITTED` 事务和 `FOR UPDATE ... SKIP LOCKED` 获取 + owner/token/version/expiry fence。S3 网络调用明确位于数据库事务之外；finalize 在第二个短事务中重验完整 fence 与 durable + Run/Attempt identity，写 exact retirement record 后删除 control。retry/manual 持久化失败分类并提供最长 24 小时 backoff；lease + 过期可由其他副本安全接管。`ql3_runtime` 可完整维护 control，但 tombstone 只有 SELECT/INSERT，其他角色保持零权限；repository + 已提供 digest/identity 失败关闭的 retention state read，为 Cluster 410 wiring 提供权威来源。migration/schema/readiness/repository + 定向 63/63 通过。ADR 暂保持 Proposed；下一阶段继续完成 validated S3 HEAD、ETag/VersionId 条件删除、bounded lifecycle、MinIO + failure matrix 与 PostgreSQL HA failover，全部完整门通过后才接受。 - D-290/ADR-0378（已接受） Local Run Attempt 日志 retention 已形成真实纵向切片：Runtime Core 增加精确 identity、canonical SHA-256 的 immutable retirement record、容量压力策略、有界 page/delete budget 与 durable cursor；日志读取在存储前检查 tombstone，并在 missing 后二次 diff --git a/docs/adr/ADR-0379-cluster-run-attempt-log-retention.md b/docs/adr/ADR-0379-cluster-run-attempt-log-retention.md new file mode 100644 index 00000000..877191b8 --- /dev/null +++ b/docs/adr/ADR-0379-cluster-run-attempt-log-retention.md @@ -0,0 +1,74 @@ +# ADR-0379：Cluster Run Attempt 日志多副本保留与条件删除 + +- 状态：Proposed（PostgreSQL authority 已实现，S3/lifecycle/HA 验收待完成） +- 日期：2026-08-12 +- 关联 RFC：QL-RFC-0001 D-291 +- 前置决策：ADR-0026、ADR-0027、ADR-0377、ADR-0378 + +## 上下文 + +Local retention 已证明 durable tombstone、删除后收敛和读取 410 的通用语义，但 Cluster 不能复用 Local cursor + unlink 模型。多个控制面副本可能同时发现同一 remote Worker 日志；对象删除需要跨 S3 网络，PostgreSQL 事务不能覆盖该网络调用；删除响应也可能丢失。若没有 durable ownership fence，副本会重复删除或把另一个副本的结果写成自己的证据；若删除前写 tombstone，则可能把仍存在的对象声明为 retired。 + +Cluster 还必须保持与低配 Local 部署的物理隔离：本能力不得让 Edge/Standalone closure 引入 PostgreSQL、AWS SDK、额外连接或后台任务。 + +## 决策 + +### 1. 不新增 package + +共享多副本 claim contract 放在 `runtime-core` 的既有 Run log-retention 目录；PostgreSQL authority 放在 `cluster-postgres`；S3 条件删除和调度 lifecycle 放在 `cluster-control`。不创建单文件 retention 微包，也不把 Cluster 依赖加入 Local closure。 + +### 2. PostgreSQL v54 是唯一 ownership authority + +`pg-0055-run-attempt-log-retention` 将 control-core 升至 v54，并声明 `run_attempt_log_retention` capability： + +- `run_attempt_log_retention_controls` 保存精确 Project/Run/Attempt/Artifact identity、eligibility、claim owner/token/version/expiry、retry time、failure count 与最后失败分类； +- `run_attempt_log_artifact_tombstones` 保存 immutable `qinglong/run-attempt-log-retirement@v1` 标量证据与 canonical digest； +- terminal remote Worker Attempt 增加局部候选索引； +- `ql3_runtime` 对 control 拥有 `SELECT/INSERT/UPDATE/DELETE`，对 tombstone 仅拥有 `SELECT/INSERT`，其他运行角色继续零权限。 + +候选必须同时满足 runtime-owned Run、Run/Attempt 均为非 `lost` 终态、`remote_worker` executor、canonical `wlog-*` identity、Run/Attempt 均超过 retention cutoff 且不存在 tombstone。claim 在短 `READ COMMITTED` 事务中以 `FOR UPDATE ... SKIP LOCKED` 获取；冲突更新再次校验 immutable identity、retry due/lease expiry 与单调 version。 + +### 3. S3 调用不进入数据库事务 + +每个副本先取得有界 durable claim，提交并释放 PostgreSQL client 后才执行 S3： + +1. validated HEAD 校验 content type、metadata identity、checksum、byte length，并取得 ETag 与可用的 VersionId； +2. 对 versioned object 使用精确 VersionId，对未版本化对象使用 ETag `If-Match` 条件删除； +3. 412/对象身份变化失败关闭并进入 retry/manual，不得删除新对象； +4. 删除成功或 HEAD 已不存在后，开启第二个短 PostgreSQL 事务； +5. 事务重验 owner/token/version/expiry、terminal Run/Attempt 与 immutable identity，插入 exact tombstone 后删除 control； +6. 删除响应丢失时，lease 过期后的新 claim 以 HEAD absent 写入 `already_absent`，最终收敛。 + +### 4. 有界调度与退避 + +每轮 claim 不超过 16 条，lease 范围 5 秒至 5 分钟；retry delay 最大 24 小时。副本不得持有跨 sweep cursor，也不得为每个 Attempt 建 timer。调度复用 Cluster control application 既有 lifecycle，并受每轮 claim 数、删除数和 wall-clock budget 共同限制。`artifact_unavailable`、`artifact_integrity_mismatch`、`retirement_record_unavailable` 是持久化失败分类；达到策略阈值后转 manual，避免坏对象形成热循环。 + +### 5. 读取收敛 + +PostgreSQL claim repository 同时实现 retention state reader。Cluster 日志读取在 S3 前检查 tombstone，并在 S3 missing 后二次检查：已 retired 返回 410；没有 tombstone 的 missing 继续保持 503，而不是伪造 retention。tombstone identity 或 digest 漂移失败关闭。 + +## 阶段验收 + +第一阶段已完成 PostgreSQL authority：共享 claim contract、v54 migration、typed Drizzle/schema/readiness、最小权限、短事务 claim、完整 lease fence、retry/manual settlement、exact tombstone finalize/replay、tombstone state read。定向 63 项 migration/schema/readiness/repository 门通过；`runtime-core` 498 项全通过，`cluster-postgres` 302 项通过、1 项条件跳过，`cluster-control` 216 项通过、2 项条件跳过。 + +阶段收口还通过 18 个 QL3 workspace package 的完整 build/test 门，以及后端兼容回归 1163 项通过、2 项条件跳过、0 失败/取消。v54 readiness 引入的两张表已同步进入 `cluster-control` 测试数据库的最小权限 fixture，避免旧 v53 fixture 把正常启动误判为 `runtime_role_invalid`。 + +结构门保持 18 个 workspace package，`singleSourcePackages=[]`、`shallowSourcePackages=[]`。新 repository 只从明确的 Cluster runtime entrypoint 导出，不扩大 package root；PostgreSQL migration append-only `ordered_ledger` 的 reviewed hard cap 随 pg-0055 由 57 精确推进到 58，不把版本账本伪拆为子目录。 + +ADR 保持 Proposed，只有以下剩余项全部完成后才转 Accepted： + +1. S3 validated HEAD + ETag/VersionId 条件删除和失败矩阵； +2. Cluster service 的 bounded claim/delete/backoff/manual 策略； +3. production composition、lifecycle drain 与读取 410； +4. MinIO versioned/unversioned 集成、响应丢失重放； +5. PostgreSQL 18 HA failover 中 lease takeover/tombstone 收敛； +6. 完整 package/backend/boundary/Profile/image gates，证明 Local closure 无 PostgreSQL/AWS SDK 回归。 + +## 被否决的替代方案 + +1. **复用 Local durable cursor**：无法形成多副本 ownership，拒绝。 +2. **在 PostgreSQL 事务中调用 S3**：把网络停顿扩大为数据库锁和故障域，拒绝。 +3. **删除前写 tombstone**：可能把仍存在的对象声明为 retired，拒绝。 +4. **无条件 DeleteObject**：无法防止对象身份变化或运维侧替换，拒绝。 +5. **授予 tombstone UPDATE/DELETE**：破坏 immutable evidence，拒绝。 +6. **新增 Cluster retention 微包**：现有聚合边界足够，且会继续放大 package 碎片化，拒绝。 diff --git a/docs/adr/README.md b/docs/adr/README.md index baea031d..0a8e3d40 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -379,6 +379,10 @@ | [ADR-0373](./ADR-0373-profile-reachable-runtime-javascript-projection.md) | Profile 可达的 Runtime JavaScript 投影 | Accepted | | [ADR-0374](./ADR-0374-shared-bounded-task-discovery-http-api.md) | 共享、有界的 Task Discovery HTTP API | Accepted | | [ADR-0375](./ADR-0375-shared-current-task-point-read-api.md) | 共享的 current Task point-read API | Proposed（设计冻结，实现中） | +| [ADR-0376](./ADR-0376-policy-and-digest-fenced-task-start.md) | Policy 与 digest fenced 的 Task Start | Accepted | +| [ADR-0377](./ADR-0377-profile-aware-run-attempt-log-range-read.md) | Profile-aware Run Attempt 日志 Range 读取 | Accepted | +| [ADR-0378](./ADR-0378-local-run-attempt-log-retention-and-tombstones.md) | Local Run Attempt 日志有界保留与 durable tombstone | Accepted | +| [ADR-0379](./ADR-0379-cluster-run-attempt-log-retention.md) | Cluster Run Attempt 日志多副本保留与条件删除 | Proposed（PostgreSQL authority 已完成） | ## 规则 diff --git a/docs/ql3-package-boundaries.json b/docs/ql3-package-boundaries.json index 6c37cee6..98962fbe 100644 --- a/docs/ql3-package-boundaries.json +++ b/docs/ql3-package-boundaries.json @@ -6,7 +6,7 @@ "reviewedDenseDirectories": [ { "kind": "ordered_ledger", - "maxDirectSourceFiles": 57, + "maxDirectSourceFiles": 58, "path": "packages/ql3-cluster-postgres/src/migrations", "rationale": "PostgreSQL migrations are an append-only version ledger whose ordering and discoverability are safer in one reviewed directory." }, diff --git a/packages/ql3-cluster-control/test/application.test.cjs b/packages/ql3-cluster-control/test/application.test.cjs index 01f99926..d5638263 100644 --- a/packages/ql3-cluster-control/test/application.test.cjs +++ b/packages/ql3-cluster-control/test/application.test.cjs @@ -137,6 +137,8 @@ function runtimePrivileges() { tool_invocation_input_artifacts: [true, true, false, false], tool_invocation_preview_artifacts: [true, true, false, false], run_attempts: [true, true, true, false], + run_attempt_log_retention_controls: [true, true, true, true], + run_attempt_log_artifact_tombstones: [true, true, false, false], run_recovery_controls: [true, true, true, false], worker_sessions: [true, true, true, false], run_dispatch_leases: [true, true, true, false], diff --git a/packages/ql3-cluster-control/test/bootstrap.test.cjs b/packages/ql3-cluster-control/test/bootstrap.test.cjs index 20c95410..a34fbf4e 100644 --- a/packages/ql3-cluster-control/test/bootstrap.test.cjs +++ b/packages/ql3-cluster-control/test/bootstrap.test.cjs @@ -51,6 +51,8 @@ function runtimePrivileges() { tool_invocation_input_artifacts: [true, true, false, false], tool_invocation_preview_artifacts: [true, true, false, false], run_attempts: [true, true, true, false], + run_attempt_log_retention_controls: [true, true, true, true], + run_attempt_log_artifact_tombstones: [true, true, false, false], run_recovery_controls: [true, true, true, false], worker_sessions: [true, true, true, false], run_dispatch_leases: [true, true, true, false], diff --git a/packages/ql3-cluster-postgres/src/entrypoints/runtime.ts b/packages/ql3-cluster-postgres/src/entrypoints/runtime.ts index b1468681..0b6d158f 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/runAttemptLogRetentionClaimRepository'; export * from '../security/projectPolicyRepository'; export * from '../security/apiCredentialRepository'; export * from '../security/securityAuditRepository'; diff --git a/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts b/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts index 0dc26ccd..ad2a2d1f 100644 --- a/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts +++ b/packages/ql3-cluster-postgres/src/migration/migrationManifest.ts @@ -278,5 +278,10 @@ export const postgresqlMainMigrationManifest: MigrationStreamManifest = checksum: '5e3e6b222269f095e0d7a985fdeb0ea154510e59dfe15873192af8c8d603fca3', }), + Object.freeze({ + id: 'pg-0055-run-attempt-log-retention', + checksum: + 'c775c65ec03ae3a1606f899064d2d38fa63fd136ce52cbd1b1172c3a51e6bf30', + }), ]), }); diff --git a/packages/ql3-cluster-postgres/src/migrations/index.ts b/packages/ql3-cluster-postgres/src/migrations/index.ts index 9f790596..73e19b76 100644 --- a/packages/ql3-cluster-postgres/src/migrations/index.ts +++ b/packages/ql3-cluster-postgres/src/migrations/index.ts @@ -57,6 +57,7 @@ import { pg0051AutomationManagementBoundaryMigration } from './pg-0051-automatio import { pg0052AutomationManagementIdentityKeysetLedgerMigration } from './pg-0052-automation-management-identity-keyset-ledger'; import { pg0053PluginPackageWorkflowRunListIndexMigration } from './pg-0053-plugin-package-workflow-run-list-index'; import { pg0054ApprovalManagementBoundaryMigration } from './pg-0054-approval-management-boundary'; +import { pg0055RunAttemptLogRetentionMigration } from './pg-0055-run-attempt-log-retention'; export const postgresqlMainMigrationStream: MigrationStreamDefinition = Object.freeze({ @@ -119,5 +120,6 @@ export const postgresqlMainMigrationStream: MigrationStreamDefinition= 0 + AND eligible_at_ms >= finished_at_ms + AND created_at_ms >= 0 + AND updated_at_ms >= created_at_ms + ), + CONSTRAINT ql3_run_log_retention_control_state_check + CHECK (state IN ('claimed', 'retry', 'manual')), + CONSTRAINT ql3_run_log_retention_control_claim_owner_check CHECK ( + claim_owner IS NULL + OR claim_owner ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$' + ), + CONSTRAINT ql3_run_log_retention_control_claim_token_check CHECK ( + claim_token IS NULL + OR claim_token ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{15,63}$' + ), + CONSTRAINT ql3_run_log_retention_control_claim_version_check + CHECK (claim_version BETWEEN 1 AND 2147483647), + CONSTRAINT ql3_run_log_retention_control_claim_expiry_check + CHECK (claim_expires_at_ms IS NULL OR claim_expires_at_ms >= 0), + CONSTRAINT ql3_run_log_retention_control_next_claim_check + CHECK (next_claim_at_ms IS NULL OR next_claim_at_ms >= 0), + CONSTRAINT ql3_run_log_retention_control_failure_count_check + CHECK (failure_count BETWEEN 0 AND 2147483647), + CONSTRAINT ql3_run_log_retention_control_failure_code_check CHECK ( + last_failure_code IS NULL + OR last_failure_code IN ( + 'artifact_unavailable', + 'artifact_integrity_mismatch', + 'retirement_record_unavailable' + ) + ), + CONSTRAINT ql3_run_log_retention_control_state_shape_check CHECK ( + ( + state = 'claimed' + AND claim_owner IS NOT NULL + AND claim_token IS NOT NULL + AND claim_expires_at_ms IS NOT NULL + AND next_claim_at_ms IS NULL + ) + OR ( + state = 'retry' + AND claim_owner IS NULL + AND claim_token IS NULL + AND claim_expires_at_ms IS NULL + AND next_claim_at_ms IS NOT NULL + AND last_failure_code IS NOT NULL + ) + OR ( + state = 'manual' + AND claim_owner IS NULL + AND claim_token IS NULL + AND claim_expires_at_ms IS NULL + AND next_claim_at_ms IS NULL + AND last_failure_code IS NOT NULL + ) + ), + CONSTRAINT ql3_run_log_retention_control_attempt_fk + FOREIGN KEY (attempt_id) REFERENCES "ql3"."run_attempts" (id) + ON DELETE CASCADE, + CONSTRAINT ql3_run_log_retention_control_run_fk + FOREIGN KEY (run_id) REFERENCES "ql3"."runs" (id) + ON DELETE CASCADE +) + `.trim(), + ` +CREATE TABLE "ql3"."run_attempt_log_artifact_tombstones" ( + log_artifact_id varchar(36) PRIMARY KEY, + project_id varchar(128) NOT NULL, + run_id varchar(36) NOT NULL, + attempt_id varchar(36) NOT NULL, + executor_type varchar(32) NOT NULL, + finished_at_ms bigint NOT NULL, + eligible_at_ms bigint NOT NULL, + retired_at_ms bigint NOT NULL, + disposition varchar(16) NOT NULL, + byte_length bigint NOT NULL, + truncated varchar(16) NOT NULL, + maximum_bytes bigint, + truncation_observed_at_ms bigint, + record_digest char(64) NOT NULL, + CONSTRAINT ql3_run_log_tombstone_attempt_key UNIQUE (attempt_id), + CONSTRAINT ql3_run_log_tombstone_identity_check CHECK ( + char_length(project_id) BETWEEN 1 AND 128 + AND char_length(run_id) BETWEEN 1 AND 36 + AND char_length(attempt_id) BETWEEN 1 AND 36 + AND log_artifact_id ~ '^wlog-[a-f0-9]{30}$' + AND executor_type = 'remote_worker' + ), + CONSTRAINT ql3_run_log_tombstone_time_check CHECK ( + finished_at_ms >= 0 + AND eligible_at_ms >= finished_at_ms + AND retired_at_ms >= eligible_at_ms + ), + CONSTRAINT ql3_run_log_tombstone_disposition_check CHECK ( + disposition IN ('deleted', 'already_absent') + AND (disposition <> 'already_absent' OR byte_length = 0) + ), + CONSTRAINT ql3_run_log_tombstone_size_check + CHECK (byte_length BETWEEN 0 AND 1073741824), + CONSTRAINT ql3_run_log_tombstone_truncation_check CHECK ( + ( + truncated = 'unknown' + AND maximum_bytes IS NULL + AND truncation_observed_at_ms IS NULL + ) + OR ( + truncated IN ('true', 'false') + AND maximum_bytes >= 1 + AND truncation_observed_at_ms >= 0 + ) + ), + CONSTRAINT ql3_run_log_tombstone_digest_check + CHECK (record_digest ~ '^[a-f0-9]{64}$'), + CONSTRAINT ql3_run_log_tombstone_attempt_fk + FOREIGN KEY (attempt_id) REFERENCES "ql3"."run_attempts" (id) + ON DELETE CASCADE, + CONSTRAINT ql3_run_log_tombstone_run_fk + FOREIGN KEY (run_id) REFERENCES "ql3"."runs" (id) + ON DELETE CASCADE +) + `.trim(), + `CREATE INDEX ql3_run_log_retention_retry_idx ON "ql3"."run_attempt_log_retention_controls" (next_claim_at_ms, finished_at_ms, attempt_id) WHERE state = 'retry'`, + `CREATE INDEX ql3_run_log_retention_claim_expiry_idx ON "ql3"."run_attempt_log_retention_controls" (claim_expires_at_ms, finished_at_ms, attempt_id) WHERE state = 'claimed'`, + `CREATE INDEX ql3_run_log_tombstone_retired_idx ON "ql3"."run_attempt_log_artifact_tombstones" (retired_at_ms, attempt_id)`, + `CREATE INDEX ql3_run_log_retention_candidate_idx ON "ql3"."run_attempts" (finished_at_ms, id) WHERE executor_type = 'remote_worker' AND log_artifact_id IS NOT NULL AND status IN ('succeeded', 'failed', 'cancelled', 'timed_out')`, + `GRANT SELECT, INSERT, UPDATE, DELETE ON "ql3"."run_attempt_log_retention_controls" TO ql3_runtime`, + `GRANT SELECT, INSERT ON "ql3"."run_attempt_log_artifact_tombstones" TO ql3_runtime`, + ` +DO $ql3$ +BEGIN + UPDATE "ql3"."schema_capabilities" + SET contract_version = 54, + migration_id = 'pg-0055-run-attempt-log-retention', + capabilities = '${CAPABILITIES_V54}'::jsonb, + updated_at_ms = floor(extract(epoch FROM transaction_timestamp()) * 1000)::bigint + WHERE contract_name = 'control-core' + AND contract_version = 53 + AND migration_id = 'pg-0054-approval-management-boundary' + AND capabilities = '${CAPABILITIES_V53}'::jsonb; + IF NOT FOUND THEN + RAISE EXCEPTION 'control-core capability is not at version 53' + USING ERRCODE = 'check_violation'; + END IF; +END +$ql3$ + `.trim(), + ], + }); diff --git a/packages/ql3-cluster-postgres/src/run/runAttemptLogRetentionClaimRepository.ts b/packages/ql3-cluster-postgres/src/run/runAttemptLogRetentionClaimRepository.ts new file mode 100644 index 00000000..4d5a1721 --- /dev/null +++ b/packages/ql3-cluster-postgres/src/run/runAttemptLogRetentionClaimRepository.ts @@ -0,0 +1,746 @@ +import { randomUUID } from 'node:crypto'; + +import { + MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS, + MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS, + MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS, + MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS, + type ClusterRunAttemptLogRetentionClaim, + type ClusterRunAttemptLogRetentionClaimPage, + type ClusterRunAttemptLogRetentionClaimRepository, + type ClusterRunAttemptLogRetentionFailureCode, + type ClusterRunAttemptLogRetentionSettlement, +} from '@qinglong/runtime-core/cluster-run-attempt-log-retention'; +import { + MAX_RUN_ATTEMPT_LOG_RETENTION_MS, + MIN_RUN_ATTEMPT_LOG_RETENTION_MS, + RunAttemptLogRetentionUnavailableError, + normalizeRunAttemptLogRetentionCandidate, + normalizeRunAttemptLogRetirementRecord, + type RunAttemptLogRetentionState, + type RunAttemptLogRetirementRecord, +} from '@qinglong/runtime-core/run-attempt-log-retention'; +import type { RunAttemptLogReadIdentity } from '@qinglong/runtime-core/run-attempt-log-read'; +import type { + PostgresPool, + PostgresQueryable, +} from '@qinglong/runtime-core'; + +type Row = Record; + +const CLAIM_SQL = ` +WITH observation AS ( + SELECT floor(extract(epoch FROM statement_timestamp()) * 1000)::bigint + AS observed_at_ms +), eligible AS ( + SELECT attempt.id, + run.project_id, + attempt.run_id, + attempt.log_artifact_id, + attempt.executor_type, + attempt.finished_at_ms, + observation.observed_at_ms + FROM "ql3"."run_attempts" AS attempt + JOIN "ql3"."runs" AS run ON run.id = attempt.run_id + CROSS JOIN observation + LEFT JOIN "ql3"."run_attempt_log_retention_controls" AS control + ON control.attempt_id = attempt.id + LEFT JOIN "ql3"."run_attempt_log_artifact_tombstones" AS tombstone + ON tombstone.attempt_id = attempt.id + OR tombstone.log_artifact_id = attempt.log_artifact_id + WHERE run.execution_owner = 'runtime' + AND run.status IN ('succeeded', 'failed', 'cancelled', 'timed_out') + AND run.finished_at_ms IS NOT NULL + AND attempt.status IN ('succeeded', 'failed', 'cancelled', 'timed_out') + AND attempt.executor_type = 'remote_worker' + AND attempt.finished_at_ms IS NOT NULL + AND attempt.log_artifact_id ~ '^wlog-[a-f0-9]{30}$' + AND attempt.finished_at_ms <= observation.observed_at_ms - $1::bigint + AND run.finished_at_ms <= observation.observed_at_ms - $1::bigint + AND tombstone.attempt_id IS NULL + AND ( + control.attempt_id IS NULL + OR ( + control.claim_version < 2147483647 + AND ( + (control.state = 'retry' + AND control.next_claim_at_ms <= observation.observed_at_ms) + OR (control.state = 'claimed' + AND control.claim_expires_at_ms <= observation.observed_at_ms) + ) + ) + ) + ORDER BY attempt.finished_at_ms, attempt.id + FOR UPDATE OF attempt SKIP LOCKED + LIMIT $2 +) +INSERT INTO "ql3"."run_attempt_log_retention_controls" ( + attempt_id, project_id, run_id, log_artifact_id, executor_type, + finished_at_ms, eligible_at_ms, state, claim_owner, claim_token, + claim_version, claim_expires_at_ms, failure_count, last_failure_code, + created_at_ms, updated_at_ms +) +SELECT id, project_id, run_id, log_artifact_id, executor_type, + finished_at_ms, finished_at_ms + $1::bigint, 'claimed', $3, $4, + 1, observed_at_ms + $5::bigint, 0, NULL, + observed_at_ms, observed_at_ms +FROM eligible +ON CONFLICT (attempt_id) DO UPDATE +SET state = 'claimed', + claim_owner = EXCLUDED.claim_owner, + claim_token = EXCLUDED.claim_token, + claim_version = "ql3"."run_attempt_log_retention_controls".claim_version + 1, + claim_expires_at_ms = EXCLUDED.claim_expires_at_ms, + next_claim_at_ms = NULL, + updated_at_ms = GREATEST( + "ql3"."run_attempt_log_retention_controls".updated_at_ms, + EXCLUDED.updated_at_ms + ) +WHERE "ql3"."run_attempt_log_retention_controls".project_id = EXCLUDED.project_id + AND "ql3"."run_attempt_log_retention_controls".run_id = EXCLUDED.run_id + AND "ql3"."run_attempt_log_retention_controls".log_artifact_id = EXCLUDED.log_artifact_id + AND "ql3"."run_attempt_log_retention_controls".executor_type = EXCLUDED.executor_type + AND "ql3"."run_attempt_log_retention_controls".finished_at_ms = EXCLUDED.finished_at_ms + AND "ql3"."run_attempt_log_retention_controls".eligible_at_ms = EXCLUDED.eligible_at_ms + AND "ql3"."run_attempt_log_retention_controls".claim_version < 2147483647 + AND ( + ("ql3"."run_attempt_log_retention_controls".state = 'retry' + AND "ql3"."run_attempt_log_retention_controls".next_claim_at_ms <= EXCLUDED.updated_at_ms) + OR ("ql3"."run_attempt_log_retention_controls".state = 'claimed' + AND "ql3"."run_attempt_log_retention_controls".claim_expires_at_ms <= EXCLUDED.updated_at_ms) + ) +RETURNING project_id AS "projectId", run_id AS "runId", + attempt_id AS "attemptId", log_artifact_id AS "logArtifactId", + executor_type AS "executorType", finished_at_ms AS "finishedAtMs", + eligible_at_ms AS "eligibleAtMs", updated_at_ms AS "observedAtMs", + claim_owner AS "claimOwner", claim_token AS "claimToken", + claim_version AS "claimVersion", claim_expires_at_ms AS "claimExpiresAtMs", + failure_count AS "failureCount" +`.trim(); + +const LOCK_CLAIM_SQL = ` +WITH observation AS ( + SELECT floor(extract(epoch FROM statement_timestamp()) * 1000)::bigint + AS observed_at_ms +) +SELECT control.project_id AS "projectId", control.run_id AS "runId", + control.attempt_id AS "attemptId", control.log_artifact_id AS "logArtifactId", + control.executor_type AS "executorType", control.finished_at_ms AS "finishedAtMs", + control.eligible_at_ms AS "eligibleAtMs", + control.claim_owner AS "claimOwner", control.claim_token AS "claimToken", + control.claim_version AS "claimVersion", + control.claim_expires_at_ms AS "claimExpiresAtMs", + control.failure_count AS "failureCount", + observation.observed_at_ms AS "observedAtMs" +FROM "ql3"."run_attempt_log_retention_controls" AS control +JOIN "ql3"."run_attempts" AS attempt ON attempt.id = control.attempt_id +JOIN "ql3"."runs" AS run ON run.id = control.run_id +CROSS JOIN observation +WHERE control.attempt_id = $1 + AND control.state = 'claimed' + AND control.claim_owner = $2 + AND control.claim_token = $3 + AND control.claim_version = $4 + AND control.claim_expires_at_ms = $5::bigint + AND control.claim_expires_at_ms > observation.observed_at_ms + AND attempt.run_id = control.run_id + AND attempt.log_artifact_id = control.log_artifact_id + AND attempt.executor_type = 'remote_worker' + AND attempt.finished_at_ms = control.finished_at_ms + AND attempt.status IN ('succeeded', 'failed', 'cancelled', 'timed_out') + AND run.project_id = control.project_id + AND run.execution_owner = 'runtime' + AND run.status IN ('succeeded', 'failed', 'cancelled', 'timed_out') +FOR UPDATE OF control +`.trim(); + +const INSERT_TOMBSTONE_SQL = ` +INSERT INTO "ql3"."run_attempt_log_artifact_tombstones" ( + log_artifact_id, project_id, run_id, attempt_id, executor_type, + finished_at_ms, eligible_at_ms, retired_at_ms, disposition, + byte_length, truncated, maximum_bytes, truncation_observed_at_ms, + record_digest +) VALUES ($1, $2, $3, $4, $5, $6::bigint, $7::bigint, $8::bigint, + $9, $10::bigint, $11, $12::bigint, $13::bigint, $14) +ON CONFLICT DO NOTHING +RETURNING record_digest AS "recordDigest" +`.trim(); + +const READ_TOMBSTONE_SQL = ` +SELECT record_digest AS "recordDigest" +FROM "ql3"."run_attempt_log_artifact_tombstones" +WHERE attempt_id = $1 OR log_artifact_id = $2 +`.trim(); + +const DELETE_CONTROL_SQL = ` +DELETE FROM "ql3"."run_attempt_log_retention_controls" +WHERE attempt_id = $1 AND state = 'claimed' + AND claim_owner = $2 AND claim_token = $3 AND claim_version = $4 + AND claim_expires_at_ms = $5::bigint +`.trim(); + +const SET_FAILURE_SQL = ` +WITH observation AS ( + SELECT floor(extract(epoch FROM statement_timestamp()) * 1000)::bigint + AS observed_at_ms +) +UPDATE "ql3"."run_attempt_log_retention_controls" AS control +SET state = $6, + claim_owner = NULL, + claim_token = NULL, + claim_expires_at_ms = NULL, + next_claim_at_ms = CASE WHEN $6 = 'retry' + THEN observation.observed_at_ms + $7::bigint ELSE NULL END, + failure_count = LEAST(control.failure_count + 1, 2147483647), + last_failure_code = $8, + updated_at_ms = GREATEST(control.updated_at_ms, observation.observed_at_ms) +FROM observation +WHERE control.attempt_id = $1 + AND control.state = 'claimed' + AND control.claim_owner = $2 + AND control.claim_token = $3 + AND control.claim_version = $4 + AND control.claim_expires_at_ms = $5::bigint + AND control.claim_expires_at_ms > observation.observed_at_ms +RETURNING control.attempt_id AS "attemptId" +`.trim(); + +const READ_TOMBSTONE_STATE_SQL = ` +SELECT project_id AS "projectId", run_id AS "runId", + attempt_id AS "attemptId", log_artifact_id AS "logArtifactId", + executor_type AS "executorType", finished_at_ms AS "finishedAtMs", + eligible_at_ms AS "eligibleAtMs", retired_at_ms AS "retiredAtMs", + disposition, byte_length AS "byteLength", truncated, + maximum_bytes AS "maximumBytes", + truncation_observed_at_ms AS "truncationObservedAtMs", + record_digest AS "recordDigest" +FROM "ql3"."run_attempt_log_artifact_tombstones" +WHERE log_artifact_id = $1 +`.trim(); + +const FAILURE_CODES = new Set([ + 'artifact_unavailable', + 'artifact_integrity_mismatch', + 'retirement_record_unavailable', +]); + +function integer( + name: string, + value: unknown, + minimum: number, + maximum = Number.MAX_SAFE_INTEGER, +): number { + const converted = + typeof value === 'string' && /^\d+$/.test(value) ? Number(value) : value; + if ( + typeof converted !== 'number' || + !Number.isSafeInteger(converted) || + converted < minimum || + converted > maximum + ) { + throw new TypeError(`${name} is invalid`); + } + return converted; +} + +function identifier(name: string, value: unknown, maximum = 128): string { + if ( + typeof value !== 'string' || + value.length > maximum || + !/^[A-Za-z0-9][A-Za-z0-9._:-]*$/.test(value) + ) { + throw new TypeError(`${name} is invalid`); + } + return value; +} + +function optionalInteger( + name: string, + value: unknown, + minimum: number, +): number | undefined { + return value === null ? undefined : integer(name, value, minimum); +} + +function readIdentity( + value: Readonly, +): Readonly { + const normalized = Object.freeze({ + projectId: identifier('Cluster log retention projectId', value?.projectId), + runId: identifier('Cluster log retention runId', value?.runId), + attemptId: identifier('Cluster log retention attemptId', value?.attemptId), + logArtifactId: identifier( + 'Cluster log retention logArtifactId', + value?.logArtifactId, + 36, + ), + }); + if ( + !value || + typeof value !== 'object' || + Array.isArray(value) || + Object.keys(value).sort().join(',') !== + 'attemptId,logArtifactId,projectId,runId' || + !/^wlog-[a-f0-9]{30}$/.test(normalized.logArtifactId) + ) { + throw new TypeError('Cluster Run Attempt log retention identity is invalid'); + } + return normalized; +} + +function tombstoneFromRow( + row: Row, +): Readonly { + const truncated = row.truncated; + if ( + truncated !== 'true' && + truncated !== 'false' && + truncated !== 'unknown' + ) { + throw new TypeError('PostgreSQL retention tombstone is invalid'); + } + return normalizeRunAttemptLogRetirementRecord({ + schema: 'qinglong/run-attempt-log-retirement@v1', + projectId: identifier('PostgreSQL tombstone projectId', row.projectId), + runId: identifier('PostgreSQL tombstone runId', row.runId), + attemptId: identifier('PostgreSQL tombstone attemptId', row.attemptId), + logArtifactId: identifier( + 'PostgreSQL tombstone logArtifactId', + row.logArtifactId, + 36, + ), + executorType: row.executorType as 'remote_worker', + finishedAtMs: integer( + 'PostgreSQL tombstone finishedAtMs', + row.finishedAtMs, + 0, + ), + eligibleAtMs: integer( + 'PostgreSQL tombstone eligibleAtMs', + row.eligibleAtMs, + 0, + ), + retiredAtMs: integer( + 'PostgreSQL tombstone retiredAtMs', + row.retiredAtMs, + 0, + ), + disposition: row.disposition as 'deleted' | 'already_absent', + byteLength: integer( + 'PostgreSQL tombstone byteLength', + row.byteLength, + 0, + ), + truncation: + truncated === 'unknown' + ? Object.freeze({ truncated: 'unknown' as const }) + : Object.freeze({ + truncated: truncated === 'true', + maximumBytes: optionalInteger( + 'PostgreSQL tombstone maximumBytes', + row.maximumBytes, + 1, + )!, + observedAtMs: optionalInteger( + 'PostgreSQL tombstone truncationObservedAtMs', + row.truncationObservedAtMs, + 0, + )!, + }), + recordDigest: identifier( + 'PostgreSQL tombstone recordDigest', + row.recordDigest, + 64, + ), + }); +} + +function failureCode( + value: unknown, +): ClusterRunAttemptLogRetentionFailureCode { + if (!FAILURE_CODES.has(value as ClusterRunAttemptLogRetentionFailureCode)) { + throw new TypeError('Cluster Run Attempt log retention failure code is invalid'); + } + return value as ClusterRunAttemptLogRetentionFailureCode; +} + +function claimFromRow(row: Row): Readonly { + const candidate = normalizeRunAttemptLogRetentionCandidate({ + projectId: identifier('PostgreSQL retention projectId', row.projectId), + runId: identifier('PostgreSQL retention runId', row.runId), + attemptId: identifier('PostgreSQL retention attemptId', row.attemptId), + logArtifactId: identifier( + 'PostgreSQL retention logArtifactId', + row.logArtifactId, + ), + executorType: row.executorType as 'remote_worker', + finishedAtMs: integer( + 'PostgreSQL retention finishedAtMs', + row.finishedAtMs, + 0, + ), + }); + if ( + candidate.executorType !== 'remote_worker' || + !/^wlog-[a-f0-9]{30}$/.test(candidate.logArtifactId) + ) { + throw new TypeError('PostgreSQL retention candidate is invalid'); + } + const observedAtMs = integer( + 'PostgreSQL retention observedAtMs', + row.observedAtMs, + 0, + ); + const eligibleAtMs = integer( + 'PostgreSQL retention eligibleAtMs', + row.eligibleAtMs, + candidate.finishedAtMs, + ); + return Object.freeze({ + candidate, + eligibleAtMs, + observedAtMs, + ownerId: identifier('PostgreSQL retention claimOwner', row.claimOwner), + token: identifier('PostgreSQL retention claimToken', row.claimToken, 64), + version: integer( + 'PostgreSQL retention claimVersion', + row.claimVersion, + 1, + 2147483647, + ), + expiresAtMs: integer( + 'PostgreSQL retention claimExpiresAtMs', + row.claimExpiresAtMs, + observedAtMs + 1, + ), + failureCount: integer( + 'PostgreSQL retention failureCount', + row.failureCount, + 0, + 2147483647, + ), + }); +} + +function assertClaim( + value: Readonly, +): Readonly { + const normalized = claimFromRow({ + projectId: value?.candidate?.projectId, + runId: value?.candidate?.runId, + attemptId: value?.candidate?.attemptId, + logArtifactId: value?.candidate?.logArtifactId, + executorType: value?.candidate?.executorType, + finishedAtMs: value?.candidate?.finishedAtMs, + eligibleAtMs: value?.eligibleAtMs, + observedAtMs: value?.observedAtMs, + claimOwner: value?.ownerId, + claimToken: value?.token, + claimVersion: value?.version, + claimExpiresAtMs: value?.expiresAtMs, + failureCount: value?.failureCount, + }); + if (normalized.expiresAtMs !== value.expiresAtMs) { + throw new TypeError('Cluster Run Attempt log retention claim is invalid'); + } + return normalized; +} + +function sameClaim( + actual: Readonly, + expected: Readonly, +): boolean { + return ( + actual.candidate.projectId === expected.candidate.projectId && + actual.candidate.runId === expected.candidate.runId && + actual.candidate.attemptId === expected.candidate.attemptId && + actual.candidate.logArtifactId === expected.candidate.logArtifactId && + actual.candidate.executorType === expected.candidate.executorType && + actual.candidate.finishedAtMs === expected.candidate.finishedAtMs && + actual.eligibleAtMs === expected.eligibleAtMs && + actual.ownerId === expected.ownerId && + actual.token === expected.token && + actual.version === expected.version && + actual.expiresAtMs === expected.expiresAtMs + ); +} + +function sameRecordClaim( + record: Readonly, + claim: Readonly, +): boolean { + const candidate = claim.candidate; + return ( + record.projectId === candidate.projectId && + record.runId === candidate.runId && + record.attemptId === candidate.attemptId && + record.logArtifactId === candidate.logArtifactId && + record.executorType === candidate.executorType && + record.finishedAtMs === candidate.finishedAtMs && + record.eligibleAtMs === claim.eligibleAtMs + ); +} + +function unavailable(error: unknown): RunAttemptLogRetentionUnavailableError { + return error instanceof RunAttemptLogRetentionUnavailableError + ? error + : new RunAttemptLogRetentionUnavailableError({ cause: error }); +} + +async function rollback(queryable: PostgresQueryable): Promise { + try { + await queryable.query('ROLLBACK'); + } catch { + // Preserve the originating authority failure. + } +} + +export class PostgresRunAttemptLogRetentionClaimRepository + implements ClusterRunAttemptLogRetentionClaimRepository +{ + constructor( + private readonly pool: PostgresPool, + private readonly createToken: () => string = randomUUID, + ) { + if ( + !pool || + typeof pool.query !== 'function' || + typeof pool.connect !== 'function' || + typeof createToken !== 'function' + ) { + throw new TypeError( + 'PostgreSQL Run Attempt log retention repository is invalid', + ); + } + } + + async inspect( + rawIdentity: Readonly, + ): Promise { + const expected = readIdentity(rawIdentity); + try { + const result = await this.pool.query(READ_TOMBSTONE_STATE_SQL, [ + expected.logArtifactId, + ]); + if (result.rows.length === 0) { + return Object.freeze({ status: 'active' as const }); + } + if (result.rows.length !== 1) { + throw new TypeError('PostgreSQL retention tombstone is not unique'); + } + const record = tombstoneFromRow(result.rows[0]!); + if ( + record.projectId !== expected.projectId || + record.runId !== expected.runId || + record.attemptId !== expected.attemptId || + record.logArtifactId !== expected.logArtifactId + ) { + throw new TypeError('PostgreSQL retention tombstone identity changed'); + } + return Object.freeze({ status: 'retired' as const, record }); + } catch (error) { + throw unavailable(error); + } + } + + async claim(options: Readonly<{ + ownerId: string; + retentionMs: number; + limit: number; + leaseMs: number; + }>): Promise> { + const ownerId = identifier( + 'Cluster Run Attempt log retention ownerId', + options?.ownerId, + ); + const retentionMs = integer( + 'Cluster Run Attempt log retention duration', + options?.retentionMs, + MIN_RUN_ATTEMPT_LOG_RETENTION_MS, + MAX_RUN_ATTEMPT_LOG_RETENTION_MS, + ); + const limit = integer( + 'Cluster Run Attempt log retention claim limit', + options?.limit, + 1, + MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS, + ); + const leaseMs = integer( + 'Cluster Run Attempt log retention lease', + options?.leaseMs, + MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS, + MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS, + ); + const token = identifier( + 'Cluster Run Attempt log retention generated token', + this.createToken(), + 64, + ); + if (token.length < 16) { + throw new TypeError( + 'Cluster Run Attempt log retention generated token is invalid', + ); + } + const client = await this.pool.connect().catch((error: unknown) => { + throw unavailable(error); + }); + let transactionOpen = false; + try { + await client.query('BEGIN ISOLATION LEVEL READ COMMITTED'); + transactionOpen = true; + await client.query("SET LOCAL statement_timeout = '5000ms'"); + await client.query("SET LOCAL lock_timeout = '1000ms'"); + const result = await client.query(CLAIM_SQL, [ + retentionMs, + limit, + ownerId, + token, + leaseMs, + ]); + const claims = result.rows.map(claimFromRow); + if (claims.length > limit) { + throw new TypeError('PostgreSQL retention claim bound was violated'); + } + await client.query('COMMIT'); + transactionOpen = false; + return Object.freeze({ + claims: Object.freeze(claims), + hasMore: claims.length === limit, + }); + } catch (error) { + if (transactionOpen) await rollback(client); + throw unavailable(error); + } finally { + client.release(); + } + } + + async settle( + rawClaim: Readonly, + settlement: Readonly, + ): Promise<'settled' | 'fenced'> { + const claim = assertClaim(rawClaim); + if (!settlement || typeof settlement !== 'object') { + throw new TypeError( + 'Cluster Run Attempt log retention settlement is invalid', + ); + } + if (settlement.status === 'retired') { + return this.recordRetirement(claim, settlement.record); + } + if (settlement.status !== 'retry' && settlement.status !== 'manual') { + throw new TypeError( + 'Cluster Run Attempt log retention settlement is invalid', + ); + } + const delayMs = + settlement.status === 'retry' + ? integer( + 'Cluster Run Attempt log retention retry delay', + settlement.delayMs, + 0, + MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS, + ) + : 0; + const code = failureCode(settlement.failureCode); + try { + const result = await this.pool.query(SET_FAILURE_SQL, [ + claim.candidate.attemptId, + claim.ownerId, + claim.token, + claim.version, + claim.expiresAtMs, + settlement.status, + delayMs, + code, + ]); + return result.rows.length === 1 ? 'settled' : 'fenced'; + } catch (error) { + throw unavailable(error); + } + } + + private async recordRetirement( + claim: Readonly, + rawRecord: Readonly, + ): Promise<'settled' | 'fenced'> { + const record = normalizeRunAttemptLogRetirementRecord(rawRecord); + if (!sameRecordClaim(record, claim)) { + throw new TypeError( + 'Cluster Run Attempt log retirement record does not match its claim', + ); + } + const client = await this.pool.connect().catch((error: unknown) => { + throw unavailable(error); + }); + let transactionOpen = false; + try { + await client.query('BEGIN ISOLATION LEVEL READ COMMITTED'); + transactionOpen = true; + await client.query("SET LOCAL statement_timeout = '5000ms'"); + await client.query("SET LOCAL lock_timeout = '1000ms'"); + const locked = await client.query(LOCK_CLAIM_SQL, [ + claim.candidate.attemptId, + claim.ownerId, + claim.token, + claim.version, + claim.expiresAtMs, + ]); + if (locked.rows.length !== 1) { + await client.query('COMMIT'); + transactionOpen = false; + return 'fenced'; + } + if (!sameClaim(claimFromRow(locked.rows[0]!), claim)) { + throw new TypeError('PostgreSQL retention claim authority changed'); + } + const truncation = record.truncation; + const inserted = await client.query(INSERT_TOMBSTONE_SQL, [ + record.logArtifactId, + record.projectId, + record.runId, + record.attemptId, + record.executorType, + record.finishedAtMs, + record.eligibleAtMs, + record.retiredAtMs, + record.disposition, + record.byteLength, + String(truncation.truncated), + truncation.maximumBytes ?? null, + truncation.observedAtMs ?? null, + record.recordDigest, + ]); + if (inserted.rows.length === 0) { + const existing = await client.query(READ_TOMBSTONE_SQL, [ + record.attemptId, + record.logArtifactId, + ]); + if ( + existing.rows.length !== 1 || + existing.rows[0]?.recordDigest !== record.recordDigest + ) { + throw new TypeError('PostgreSQL retention tombstone conflicts'); + } + } + const removed = await client.query(DELETE_CONTROL_SQL, [ + claim.candidate.attemptId, + claim.ownerId, + claim.token, + claim.version, + claim.expiresAtMs, + ]); + if (removed.rowCount !== 1) { + throw new TypeError('PostgreSQL retention claim fence was lost'); + } + await client.query('COMMIT'); + transactionOpen = false; + return 'settled'; + } catch (error) { + if (transactionOpen) await rollback(client); + throw unavailable(error); + } finally { + client.release(); + } + } +} diff --git a/packages/ql3-cluster-postgres/src/schema/schema.ts b/packages/ql3-cluster-postgres/src/schema/schema.ts index 4db156c0..616d556d 100644 --- a/packages/ql3-cluster-postgres/src/schema/schema.ts +++ b/packages/ql3-cluster-postgres/src/schema/schema.ts @@ -4474,6 +4474,162 @@ export const runAttempts = ql3Schema.table( index('ql3_run_attempts_lease_idx') .on(table.leaseExpiresAtMs, table.id) .where(sql`${table.leaseExpiresAtMs} is not null`), + index('ql3_run_log_retention_candidate_idx') + .on(table.finishedAtMs, table.id) + .where( + sql`${table.executorType} = 'remote_worker' and ${table.logArtifactId} is not null and ${table.status} in ('succeeded', 'failed', 'cancelled', 'timed_out')`, + ), + ], +); + +export const runAttemptLogRetentionControls = ql3Schema.table( + 'run_attempt_log_retention_controls', + { + attemptId: varchar('attempt_id', { length: 36 }).primaryKey(), + projectId: varchar('project_id', { length: 128 }).notNull(), + runId: varchar('run_id', { length: 36 }).notNull(), + logArtifactId: varchar('log_artifact_id', { length: 36 }).notNull(), + executorType: varchar('executor_type', { length: 32 }).notNull(), + finishedAtMs: bigint('finished_at_ms', { mode: 'number' }).notNull(), + eligibleAtMs: bigint('eligible_at_ms', { mode: 'number' }).notNull(), + state: varchar('state', { length: 16 }).notNull(), + claimOwner: varchar('claim_owner', { length: 128 }), + claimToken: varchar('claim_token', { length: 64 }), + claimVersion: integer('claim_version').default(1).notNull(), + claimExpiresAtMs: bigint('claim_expires_at_ms', { mode: 'number' }), + nextClaimAtMs: bigint('next_claim_at_ms', { mode: 'number' }), + failureCount: integer('failure_count').default(0).notNull(), + lastFailureCode: varchar('last_failure_code', { length: 64 }), + createdAtMs: bigint('created_at_ms', { mode: 'number' }).notNull(), + updatedAtMs: bigint('updated_at_ms', { mode: 'number' }).notNull(), + }, + (table) => [ + uniqueIndex('ql3_run_log_retention_control_artifact_key').on( + table.logArtifactId, + ), + check( + 'ql3_run_log_retention_control_identity_check', + sql`char_length(${table.projectId}) between 1 and 128 and char_length(${table.runId}) between 1 and 36 and char_length(${table.attemptId}) between 1 and 36 and ${table.logArtifactId} ~ '^wlog-[a-f0-9]{30}$' and ${table.executorType} = 'remote_worker'`, + ), + check( + 'ql3_run_log_retention_control_time_check', + sql`${table.finishedAtMs} >= 0 and ${table.eligibleAtMs} >= ${table.finishedAtMs} and ${table.createdAtMs} >= 0 and ${table.updatedAtMs} >= ${table.createdAtMs}`, + ), + check( + 'ql3_run_log_retention_control_state_check', + sql`${table.state} in ('claimed', 'retry', 'manual')`, + ), + check( + 'ql3_run_log_retention_control_claim_owner_check', + sql`${table.claimOwner} is null or ${table.claimOwner} ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$'`, + ), + check( + 'ql3_run_log_retention_control_claim_token_check', + sql`${table.claimToken} is null or ${table.claimToken} ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{15,63}$'`, + ), + check( + 'ql3_run_log_retention_control_claim_version_check', + sql`${table.claimVersion} between 1 and 2147483647`, + ), + check( + 'ql3_run_log_retention_control_claim_expiry_check', + sql`${table.claimExpiresAtMs} is null or ${table.claimExpiresAtMs} >= 0`, + ), + check( + 'ql3_run_log_retention_control_next_claim_check', + sql`${table.nextClaimAtMs} is null or ${table.nextClaimAtMs} >= 0`, + ), + check( + 'ql3_run_log_retention_control_failure_count_check', + sql`${table.failureCount} between 0 and 2147483647`, + ), + check( + 'ql3_run_log_retention_control_failure_code_check', + sql`${table.lastFailureCode} is null or ${table.lastFailureCode} in ('artifact_unavailable', 'artifact_integrity_mismatch', 'retirement_record_unavailable')`, + ), + check( + 'ql3_run_log_retention_control_state_shape_check', + sql`(${table.state} = 'claimed' and ${table.claimOwner} is not null and ${table.claimToken} is not null and ${table.claimExpiresAtMs} is not null and ${table.nextClaimAtMs} is null) or (${table.state} = 'retry' and ${table.claimOwner} is null and ${table.claimToken} is null and ${table.claimExpiresAtMs} is null and ${table.nextClaimAtMs} is not null and ${table.lastFailureCode} is not null) or (${table.state} = 'manual' and ${table.claimOwner} is null and ${table.claimToken} is null and ${table.claimExpiresAtMs} is null and ${table.nextClaimAtMs} is null and ${table.lastFailureCode} is not null)`, + ), + foreignKey({ + name: 'ql3_run_log_retention_control_attempt_fk', + columns: [table.attemptId], + foreignColumns: [runAttempts.id], + }).onDelete('cascade'), + foreignKey({ + name: 'ql3_run_log_retention_control_run_fk', + columns: [table.runId], + foreignColumns: [runs.id], + }).onDelete('cascade'), + index('ql3_run_log_retention_retry_idx') + .on(table.nextClaimAtMs, table.finishedAtMs, table.attemptId) + .where(sql`${table.state} = 'retry'`), + index('ql3_run_log_retention_claim_expiry_idx') + .on(table.claimExpiresAtMs, table.finishedAtMs, table.attemptId) + .where(sql`${table.state} = 'claimed'`), + ], +); + +export const runAttemptLogArtifactTombstones = ql3Schema.table( + 'run_attempt_log_artifact_tombstones', + { + logArtifactId: varchar('log_artifact_id', { length: 36 }).primaryKey(), + projectId: varchar('project_id', { length: 128 }).notNull(), + runId: varchar('run_id', { length: 36 }).notNull(), + attemptId: varchar('attempt_id', { length: 36 }).notNull(), + executorType: varchar('executor_type', { length: 32 }).notNull(), + finishedAtMs: bigint('finished_at_ms', { mode: 'number' }).notNull(), + eligibleAtMs: bigint('eligible_at_ms', { mode: 'number' }).notNull(), + retiredAtMs: bigint('retired_at_ms', { mode: 'number' }).notNull(), + disposition: varchar('disposition', { length: 16 }).notNull(), + byteLength: bigint('byte_length', { mode: 'number' }).notNull(), + truncated: varchar('truncated', { length: 16 }).notNull(), + maximumBytes: bigint('maximum_bytes', { mode: 'number' }), + truncationObservedAtMs: bigint('truncation_observed_at_ms', { + mode: 'number', + }), + recordDigest: char('record_digest', { length: 64 }).notNull(), + }, + (table) => [ + uniqueIndex('ql3_run_log_tombstone_attempt_key').on(table.attemptId), + check( + 'ql3_run_log_tombstone_identity_check', + sql`char_length(${table.projectId}) between 1 and 128 and char_length(${table.runId}) between 1 and 36 and char_length(${table.attemptId}) between 1 and 36 and ${table.logArtifactId} ~ '^wlog-[a-f0-9]{30}$' and ${table.executorType} = 'remote_worker'`, + ), + check( + 'ql3_run_log_tombstone_time_check', + sql`${table.finishedAtMs} >= 0 and ${table.eligibleAtMs} >= ${table.finishedAtMs} and ${table.retiredAtMs} >= ${table.eligibleAtMs}`, + ), + check( + 'ql3_run_log_tombstone_disposition_check', + sql`${table.disposition} in ('deleted', 'already_absent') and (${table.disposition} <> 'already_absent' or ${table.byteLength} = 0)`, + ), + check( + 'ql3_run_log_tombstone_size_check', + sql`${table.byteLength} between 0 and 1073741824`, + ), + check( + 'ql3_run_log_tombstone_truncation_check', + sql`(${table.truncated} = 'unknown' and ${table.maximumBytes} is null and ${table.truncationObservedAtMs} is null) or (${table.truncated} in ('true', 'false') and ${table.maximumBytes} >= 1 and ${table.truncationObservedAtMs} >= 0)`, + ), + check( + 'ql3_run_log_tombstone_digest_check', + sql`${table.recordDigest} ~ '^[a-f0-9]{64}$'`, + ), + foreignKey({ + name: 'ql3_run_log_tombstone_attempt_fk', + columns: [table.attemptId], + foreignColumns: [runAttempts.id], + }).onDelete('cascade'), + foreignKey({ + name: 'ql3_run_log_tombstone_run_fk', + columns: [table.runId], + foreignColumns: [runs.id], + }).onDelete('cascade'), + index('ql3_run_log_tombstone_retired_idx').on( + table.retiredAtMs, + table.attemptId, + ), ], ); @@ -5700,6 +5856,8 @@ export const ql3PostgresTables = [ toolExecutionResultRekeyHeads, toolResultKeyRetirementReceipts, runAttempts, + runAttemptLogRetentionControls, + runAttemptLogArtifactTombstones, workerSessions, runDispatchLeases, workerCredentials, diff --git a/packages/ql3-cluster-postgres/src/schema/schemaContract.ts b/packages/ql3-cluster-postgres/src/schema/schemaContract.ts index 247de2d2..915967a2 100644 --- a/packages/ql3-cluster-postgres/src/schema/schemaContract.ts +++ b/packages/ql3-cluster-postgres/src/schema/schemaContract.ts @@ -15,12 +15,13 @@ export interface PostgresSchemaContractFunction { export interface PostgresSchemaContract { readonly schema: 'ql3'; readonly contractName: 'control-core'; - readonly contractVersion: 53; - readonly migrationId: 'pg-0054-approval-management-boundary'; + readonly contractVersion: 54; + readonly migrationId: 'pg-0055-run-attempt-log-retention'; readonly minimumServerMajor: 16; readonly maximumServerMajor: 18; readonly capabilities: Readonly<{ run_core: 1; + run_attempt_log_retention: 1; run_dispatch_lease: 1; run_retry_policy: 1; project_policy: 1; @@ -100,8 +101,8 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = Object.freeze({ schema: 'ql3', contractName: 'control-core', - contractVersion: 53, - migrationId: 'pg-0054-approval-management-boundary', + contractVersion: 54, + migrationId: 'pg-0055-run-attempt-log-retention', minimumServerMajor: 16, maximumServerMajor: 18, capabilities: Object.freeze({ @@ -141,6 +142,7 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = project_policy: 1, project_tool_definition_snapshot: 1, run_core: 1, + run_attempt_log_retention: 1, run_dispatch_lease: 1, run_retry_policy: 1, security_audit: 1, @@ -1166,6 +1168,41 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = 'error_code', 'error_summary', ]), + table('run_attempt_log_retention_controls', [ + 'attempt_id', + 'project_id', + 'run_id', + 'log_artifact_id', + 'executor_type', + 'finished_at_ms', + 'eligible_at_ms', + 'state', + 'claim_owner', + 'claim_token', + 'claim_version', + 'claim_expires_at_ms', + 'next_claim_at_ms', + 'failure_count', + 'last_failure_code', + 'created_at_ms', + 'updated_at_ms', + ]), + table('run_attempt_log_artifact_tombstones', [ + 'log_artifact_id', + 'project_id', + 'run_id', + 'attempt_id', + 'executor_type', + 'finished_at_ms', + 'eligible_at_ms', + 'retired_at_ms', + 'disposition', + 'byte_length', + 'truncated', + 'maximum_bytes', + 'truncation_observed_at_ms', + 'record_digest', + ]), table('worker_sessions', [ 'worker_id', 'session_id', @@ -1594,6 +1631,14 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = 'ql3_run_attempts_dispatch_candidates_idx', 'ql3_run_attempts_recovery_idx', 'ql3_run_attempts_lease_idx', + 'run_attempt_log_retention_controls_pkey', + 'ql3_run_log_retention_control_artifact_key', + 'ql3_run_log_retention_retry_idx', + 'ql3_run_log_retention_claim_expiry_idx', + 'run_attempt_log_artifact_tombstones_pkey', + 'ql3_run_log_tombstone_attempt_key', + 'ql3_run_log_tombstone_retired_idx', + 'ql3_run_log_retention_candidate_idx', 'worker_sessions_pkey', 'ql3_worker_sessions_available_idx', 'run_dispatch_leases_pkey', @@ -1913,6 +1958,23 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = 'ql3_run_attempts_created_at_check', 'ql3_run_attempts_started_at_check', 'ql3_run_attempts_finished_at_check', + 'ql3_run_log_retention_control_identity_check', + 'ql3_run_log_retention_control_time_check', + 'ql3_run_log_retention_control_state_check', + 'ql3_run_log_retention_control_claim_owner_check', + 'ql3_run_log_retention_control_claim_token_check', + 'ql3_run_log_retention_control_claim_version_check', + 'ql3_run_log_retention_control_claim_expiry_check', + 'ql3_run_log_retention_control_next_claim_check', + 'ql3_run_log_retention_control_failure_count_check', + 'ql3_run_log_retention_control_failure_code_check', + 'ql3_run_log_retention_control_state_shape_check', + 'ql3_run_log_tombstone_identity_check', + 'ql3_run_log_tombstone_time_check', + 'ql3_run_log_tombstone_disposition_check', + 'ql3_run_log_tombstone_size_check', + 'ql3_run_log_tombstone_truncation_check', + 'ql3_run_log_tombstone_digest_check', 'ql3_worker_sessions_worker_id_check', 'ql3_worker_sessions_session_id_check', 'ql3_worker_sessions_generation_check', @@ -2211,6 +2273,10 @@ export const postgresqlControlSchemaContract: PostgresSchemaContract = 'ql3_result_retirement_catalog_fk', 'ql3_run_attempts_run_fk', 'ql3_run_attempts_step_run_fk', + 'ql3_run_log_retention_control_attempt_fk', + 'ql3_run_log_retention_control_run_fk', + 'ql3_run_log_tombstone_attempt_fk', + 'ql3_run_log_tombstone_run_fk', 'ql3_run_dispatch_leases_attempt_fk', 'ql3_run_dispatch_leases_run_fk', 'ql3_run_dispatch_leases_worker_fk', diff --git a/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts b/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts index 415710f7..78302937 100644 --- a/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts +++ b/packages/ql3-cluster-postgres/src/schema/schemaReadiness.ts @@ -539,6 +539,18 @@ const REQUIRED_RUNTIME_PRIVILEGES = Object.freeze({ update: true, delete: false, }), + run_attempt_log_retention_controls: Object.freeze({ + select: true, + insert: true, + update: true, + delete: true, + }), + run_attempt_log_artifact_tombstones: Object.freeze({ + select: true, + insert: true, + update: false, + delete: false, + }), worker_sessions: Object.freeze({ select: true, insert: true, @@ -1034,6 +1046,18 @@ const REQUIRED_ADMIN_PRIVILEGES = Object.freeze({ update: false, delete: false, }), + run_attempt_log_retention_controls: Object.freeze({ + select: false, + insert: false, + update: false, + delete: false, + }), + run_attempt_log_artifact_tombstones: Object.freeze({ + select: false, + insert: false, + update: false, + delete: false, + }), worker_sessions: Object.freeze({ select: false, insert: false, diff --git a/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs b/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs index c07eb06b..ec78b5e2 100644 --- a/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgresqlMigrationDefinitions.test.cjs @@ -103,6 +103,7 @@ test('defines the immutable PostgreSQL capability and Run core stream', async () 'pg-0052-automation-management-identity-keyset-ledger', 'pg-0053-plugin-package-workflow-run-list-index', 'pg-0054-approval-management-boundary', + 'pg-0055-run-attempt-log-retention', ], ); for (const migration of postgresqlMainMigrationStream.migrations) { @@ -508,6 +509,11 @@ test('freezes every published PostgreSQL migration checksum', () => { checksum: '5e3e6b222269f095e0d7a985fdeb0ea154510e59dfe15873192af8c8d603fca3', }, + { + id: 'pg-0055-run-attempt-log-retention', + checksum: + 'c775c65ec03ae3a1606f899064d2d38fa63fd136ce52cbd1b1172c3a51e6bf30', + }, ]; assert.deepEqual( postgresqlMainMigrationStream.migrations.map(({ id, checksum }) => ({ @@ -1875,3 +1881,26 @@ test('advances capability v53 with isolated human Approval management authority' /migration_id = 'pg-0053-plugin-package-workflow-run-list-index'/, ); }); + +test('advances capability v54 with durable Cluster log retention authority', async () => { + const migration = migrationById('pg-0055-run-attempt-log-retention'); + const statements = []; + await migration.up({ + async query(statement) { + statements.push(statement); + return { rows: [] }; + }, + }); + const sql = statements.join('\n'); + assert.match(sql, /run_attempt_log_retention_controls/); + assert.match(sql, /run_attempt_log_artifact_tombstones/); + assert.match(sql, /FOR UPDATE|SKIP LOCKED|claim_expires_at_ms/); + assert.match(sql, /TO ql3_runtime/); + assert.match(sql, /contract_version = 54/); + assert.match(sql, /"run_attempt_log_retention":1/); + assert.match(sql, /contract_version = 53/); + assert.match( + sql, + /migration_id = 'pg-0054-approval-management-boundary'/, + ); +}); diff --git a/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs b/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs index 183266a2..8736c1fb 100644 --- a/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs +++ b/packages/ql3-cluster-postgres/test/postgresqlSchemaReadiness.test.cjs @@ -58,6 +58,8 @@ 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_attempt_log_retention_controls: [true, true, true, true], + run_attempt_log_artifact_tombstones: [true, true, false, false], worker_sessions: [true, true, true, false], run_dispatch_leases: [true, true, true, false], worker_credentials: [false, false, false, false], @@ -178,6 +180,8 @@ 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_attempt_log_retention_controls: [false, false, false, false], + run_attempt_log_artifact_tombstones: [false, false, false, false], worker_sessions: [false, false, false, false], run_dispatch_leases: [false, false, false, false], worker_credentials: [true, true, true, false], @@ -692,7 +696,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro serverMajor: 16, currentUser: 'ql3_runtime', contractName: 'control-core', - contractVersion: 53, + contractVersion: 54, migrationIds: [ 'pg-0001-schema-capability', 'pg-0002-run-core', @@ -748,6 +752,7 @@ test('accepts the exact PostgreSQL control schema and least-privilege runtime ro 'pg-0052-automation-management-identity-keyset-ledger', 'pg-0053-plugin-package-workflow-run-list-index', 'pg-0054-approval-management-boundary', + 'pg-0055-run-attempt-log-retention', ], }); }); @@ -778,10 +783,10 @@ test('accepts the exact schema and isolated least-privilege admin role', async ( }), ); assert.equal(report.currentUser, 'ql3_admin'); - assert.equal(report.contractVersion, 53); + assert.equal(report.contractVersion, 54); assert.equal( report.migrationIds.at(-1), - 'pg-0054-approval-management-boundary', + 'pg-0055-run-attempt-log-retention', ); }); @@ -794,10 +799,10 @@ test('accepts the isolated least-privilege automation manager role', async () => }), ); assert.equal(report.currentUser, 'ql3_automation_manager'); - assert.equal(report.contractVersion, 53); + assert.equal(report.contractVersion, 54); assert.equal( report.migrationIds.at(-1), - 'pg-0054-approval-management-boundary', + 'pg-0055-run-attempt-log-retention', ); const widened = automationManagerPrivileges(); @@ -826,10 +831,10 @@ test('accepts the isolated least-privilege human Approval manager role', async ( }), ); assert.equal(report.currentUser, 'ql3_approval_manager'); - assert.equal(report.contractVersion, 53); + assert.equal(report.contractVersion, 54); assert.equal( report.migrationIds.at(-1), - 'pg-0054-approval-management-boundary', + 'pg-0055-run-attempt-log-retention', ); const widened = approvalManagerPrivileges(); @@ -936,10 +941,10 @@ test('accepts the exact schema and isolated Worker ingress role', async () => { }), ); assert.equal(report.currentUser, 'ql3_worker_ingress'); - assert.equal(report.contractVersion, 53); + assert.equal(report.contractVersion, 54); assert.equal( report.migrationIds.at(-1), - 'pg-0054-approval-management-boundary', + 'pg-0055-run-attempt-log-retention', ); }); diff --git a/packages/ql3-cluster-postgres/test/runAttemptLogRetentionClaimRepository.test.cjs b/packages/ql3-cluster-postgres/test/runAttemptLogRetentionClaimRepository.test.cjs new file mode 100644 index 00000000..c69849bd --- /dev/null +++ b/packages/ql3-cluster-postgres/test/runAttemptLogRetentionClaimRepository.test.cjs @@ -0,0 +1,330 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { + createRunAttemptLogRetirementRecord, + RunAttemptLogRetentionUnavailableError, +} = require('@qinglong/runtime-core/run-attempt-log-retention'); +const { + PostgresRunAttemptLogRetentionClaimRepository, +} = require('../dist/entrypoints/runtime'); + +const TOKEN = '00000000-0000-4000-8000-000000000055'; +const ARTIFACT_ID = `wlog-${'a'.repeat(30)}`; + +function claimRow(overrides = {}) { + return { + projectId: 'project-1', + runId: 'run-1', + attemptId: 'attempt-1', + logArtifactId: ARTIFACT_ID, + executorType: 'remote_worker', + finishedAtMs: '1000', + eligibleAtMs: '61000', + observedAtMs: '70000', + claimOwner: 'replica-a', + claimToken: TOKEN, + claimVersion: 1, + claimExpiresAtMs: '100000', + failureCount: 0, + ...overrides, + }; +} + +function claim() { + return Object.freeze({ + candidate: Object.freeze({ + projectId: 'project-1', + runId: 'run-1', + attemptId: 'attempt-1', + logArtifactId: ARTIFACT_ID, + executorType: 'remote_worker', + finishedAtMs: 1000, + }), + eligibleAtMs: 61000, + observedAtMs: 70000, + ownerId: 'replica-a', + token: TOKEN, + version: 1, + expiresAtMs: 100000, + failureCount: 0, + }); +} + +test('claims one bounded remote log page under a short database lease', async () => { + const calls = []; + let released = false; + const repository = new PostgresRunAttemptLogRetentionClaimRepository( + { + async connect() { + return { + async query(text, values = []) { + calls.push({ text, values }); + if (text.includes('FOR UPDATE OF attempt SKIP LOCKED')) { + return { rows: [claimRow()], rowCount: 1 }; + } + return { rows: [], rowCount: 0 }; + }, + release() { + released = true; + }, + }; + }, + async query() { + throw new Error('pool query not expected'); + }, + }, + () => TOKEN, + ); + + const page = await repository.claim({ + ownerId: 'replica-a', + retentionMs: 60000, + limit: 4, + leaseMs: 30000, + }); + + assert.deepEqual(page, { claims: [claim()], hasMore: false }); + assert.deepEqual( + calls.map(({ text }) => text.split('\n', 1)[0]), + [ + 'BEGIN ISOLATION LEVEL READ COMMITTED', + "SET LOCAL statement_timeout = '5000ms'", + "SET LOCAL lock_timeout = '1000ms'", + 'WITH observation AS (', + 'COMMIT', + ], + ); + assert.deepEqual(calls[3].values, [ + 60000, + 4, + 'replica-a', + TOKEN, + 30000, + ]); + assert.match(calls[3].text, /ON CONFLICT \(attempt_id\) DO UPDATE/); + assert.match(calls[3].text, /claim_expires_at_ms <= EXCLUDED\.updated_at_ms/); + assert.equal(released, true); +}); + +test('fences retry settlement by owner token version and database expiry', async () => { + const calls = []; + const repository = new PostgresRunAttemptLogRetentionClaimRepository({ + async connect() { + throw new Error('not expected'); + }, + async query(text, values) { + calls.push({ text, values }); + return { rows: [{ attemptId: 'attempt-1' }], rowCount: 1 }; + }, + }); + + assert.equal( + await repository.settle(claim(), { + status: 'retry', + delayMs: 2500, + failureCode: 'artifact_unavailable', + }), + 'settled', + ); + assert.deepEqual(calls[0].values, [ + 'attempt-1', + 'replica-a', + TOKEN, + 1, + 100000, + 'retry', + 2500, + 'artifact_unavailable', + ]); + assert.match(calls[0].text, /claim_expires_at_ms > observation\.observed_at_ms/); +}); + +test('reads an exact durable tombstone for the profile-aware log route', async () => { + const record = createRunAttemptLogRetirementRecord({ + ...claim().candidate, + eligibleAtMs: 61000, + retiredAtMs: 80000, + disposition: 'already_absent', + byteLength: 0, + truncation: { truncated: 'unknown' }, + }); + const rows = [ + { + ...record, + finishedAtMs: String(record.finishedAtMs), + eligibleAtMs: String(record.eligibleAtMs), + retiredAtMs: String(record.retiredAtMs), + byteLength: String(record.byteLength), + truncated: 'unknown', + maximumBytes: null, + truncationObservedAtMs: null, + }, + ]; + const repository = new PostgresRunAttemptLogRetentionClaimRepository({ + async connect() { + throw new Error('not expected'); + }, + async query(text, values) { + assert.match(text, /artifact_tombstones/); + assert.deepEqual(values, [ARTIFACT_ID]); + return { rows: rows.splice(0), rowCount: 1 }; + }, + }); + const identity = { + projectId: 'project-1', + runId: 'run-1', + attemptId: 'attempt-1', + logArtifactId: ARTIFACT_ID, + }; + + assert.deepEqual(await repository.inspect(identity), { + status: 'retired', + record, + }); + assert.deepEqual(await repository.inspect(identity), { status: 'active' }); +}); + +test('records the exact tombstone and removes its claim atomically', async () => { + const record = createRunAttemptLogRetirementRecord({ + ...claim().candidate, + eligibleAtMs: 61000, + retiredAtMs: 80000, + disposition: 'deleted', + byteLength: 11, + truncation: { + truncated: false, + maximumBytes: 1048576, + observedAtMs: 80000, + }, + }); + const calls = []; + let released = false; + const repository = new PostgresRunAttemptLogRetentionClaimRepository({ + async connect() { + return { + async query(text, values = []) { + calls.push({ text, values }); + if (text.includes('FOR UPDATE OF control')) { + return { + rows: [claimRow({ observedAtMs: '80000' })], + rowCount: 1, + }; + } + if (text.startsWith('INSERT INTO "ql3"."run_attempt_log_artifact_tombstones"')) { + return { + rows: [{ recordDigest: record.recordDigest }], + rowCount: 1, + }; + } + if (text.startsWith('DELETE FROM "ql3"."run_attempt_log_retention_controls"')) { + return { rows: [], rowCount: 1 }; + } + return { rows: [], rowCount: 0 }; + }, + release() { + released = true; + }, + }; + }, + async query() { + throw new Error('not expected'); + }, + }); + + assert.equal( + await repository.settle(claim(), { status: 'retired', record }), + 'settled', + ); + assert.deepEqual( + calls.map(({ text }) => text.split('\n', 1)[0]), + [ + 'BEGIN ISOLATION LEVEL READ COMMITTED', + "SET LOCAL statement_timeout = '5000ms'", + "SET LOCAL lock_timeout = '1000ms'", + 'WITH observation AS (', + 'INSERT INTO "ql3"."run_attempt_log_artifact_tombstones" (', + 'DELETE FROM "ql3"."run_attempt_log_retention_controls"', + 'COMMIT', + ], + ); + assert.equal(calls[4].values.at(-1), record.recordDigest); + assert.equal(released, true); +}); + +test('returns fenced without writing when the durable lease changed', async () => { + const calls = []; + const repository = new PostgresRunAttemptLogRetentionClaimRepository({ + async connect() { + return { + async query(text) { + calls.push(text); + if (text.includes('FOR UPDATE OF control')) { + return { rows: [], rowCount: 0 }; + } + return { rows: [], rowCount: 0 }; + }, + release() {}, + }; + }, + async query() { + throw new Error('not expected'); + }, + }); + const record = createRunAttemptLogRetirementRecord({ + ...claim().candidate, + eligibleAtMs: 61000, + retiredAtMs: 80000, + disposition: 'already_absent', + byteLength: 0, + truncation: { truncated: 'unknown' }, + }); + + assert.equal( + await repository.settle(claim(), { status: 'retired', record }), + 'fenced', + ); + assert.equal( + calls.some((text) => text.includes('artifact_tombstones" (')), + false, + ); +}); + +test('rolls back and wraps claim failures without leaking the client', async () => { + const calls = []; + let released = false; + const repository = new PostgresRunAttemptLogRetentionClaimRepository( + { + async connect() { + return { + async query(text) { + calls.push(text); + if (text.includes('FOR UPDATE OF attempt')) throw new Error('offline'); + return { rows: [], rowCount: 0 }; + }, + release() { + released = true; + }, + }; + }, + async query() { + throw new Error('not expected'); + }, + }, + () => TOKEN, + ); + + await assert.rejects( + repository.claim({ + ownerId: 'replica-a', + retentionMs: 60000, + limit: 1, + leaseMs: 5000, + }), + RunAttemptLogRetentionUnavailableError, + ); + assert.equal(calls.at(-1), 'ROLLBACK'); + assert.equal(released, true); +}); diff --git a/packages/ql3-runtime-core/package.json b/packages/ql3-runtime-core/package.json index 31dd522d..89b2b528 100644 --- a/packages/ql3-runtime-core/package.json +++ b/packages/ql3-runtime-core/package.json @@ -238,6 +238,9 @@ ], "run-attempt-log-retention": [ "dist/run/log-retention/runAttemptLogRetention.d.ts" + ], + "cluster-run-attempt-log-retention": [ + "dist/run/log-retention/clusterRunAttemptLogRetention.d.ts" ] } }, @@ -297,6 +300,11 @@ "require": "./dist/run/log-retention/runAttemptLogRetention.js", "default": "./dist/run/log-retention/runAttemptLogRetention.js" }, + "./cluster-run-attempt-log-retention": { + "types": "./dist/run/log-retention/clusterRunAttemptLogRetention.d.ts", + "require": "./dist/run/log-retention/clusterRunAttemptLogRetention.js", + "default": "./dist/run/log-retention/clusterRunAttemptLogRetention.js" + }, "./task-definition": { "types": "./dist/task-definition/taskDefinition.d.ts", "require": "./dist/task-definition/taskDefinition.js", diff --git a/packages/ql3-runtime-core/src/run/log-retention/clusterRunAttemptLogRetention.ts b/packages/ql3-runtime-core/src/run/log-retention/clusterRunAttemptLogRetention.ts new file mode 100644 index 00000000..ffeaa519 --- /dev/null +++ b/packages/ql3-runtime-core/src/run/log-retention/clusterRunAttemptLogRetention.ts @@ -0,0 +1,62 @@ +import type { + RunAttemptLogRetentionCandidate, + RunAttemptLogRetentionStateReader, + RunAttemptLogRetirementRecord, +} from './runAttemptLogRetention'; + +export const MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS = 16; +export const MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS = 5_000; +export const MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS = 5 * 60_000; +export const MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS = + 24 * 60 * 60_000; + +export type ClusterRunAttemptLogRetentionFailureCode = + | 'artifact_unavailable' + | 'artifact_integrity_mismatch' + | 'retirement_record_unavailable'; + +export interface ClusterRunAttemptLogRetentionClaim { + readonly candidate: Readonly; + readonly eligibleAtMs: number; + readonly observedAtMs: number; + readonly ownerId: string; + readonly token: string; + readonly version: number; + readonly expiresAtMs: number; + readonly failureCount: number; +} + +export interface ClusterRunAttemptLogRetentionClaimPage { + readonly claims: readonly Readonly[]; + readonly hasMore: boolean; +} + +export type ClusterRunAttemptLogRetentionSettlement = + | Readonly<{ + readonly status: 'retired'; + readonly record: Readonly; + }> + | Readonly<{ + readonly status: 'retry'; + readonly delayMs: number; + readonly failureCode: ClusterRunAttemptLogRetentionFailureCode; + }> + | Readonly<{ + readonly status: 'manual'; + readonly failureCode: ClusterRunAttemptLogRetentionFailureCode; + }>; + +export interface ClusterRunAttemptLogRetentionClaimRepository + extends RunAttemptLogRetentionStateReader { + claim(options: Readonly<{ + readonly ownerId: string; + readonly retentionMs: number; + readonly limit: number; + readonly leaseMs: number; + }>): Promise>; + + settle( + claim: Readonly, + settlement: Readonly, + ): Promise<'settled' | 'fenced'>; +} diff --git a/test/back/ql3PackageBoundaryAudit.test.cjs b/test/back/ql3PackageBoundaryAudit.test.cjs index 9c0b79aa..65bcf5d8 100644 --- a/test/back/ql3PackageBoundaryAudit.test.cjs +++ b/test/back/ql3PackageBoundaryAudit.test.cjs @@ -74,7 +74,7 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( [ { directory: 'packages/ql3-cluster-postgres/src/migrations', - directSourceFiles: 57, + directSourceFiles: 58, reviewKind: 'ordered_ledger', }, { @@ -299,10 +299,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( rootSourceFileRoles: runtimeCore.rootSourceFileRoles, }, { - sourceFiles: 150, + sourceFiles: 151, rootSourceFiles: 1, rootSourceLines: 160, - nestedSourceFiles: 149, + nestedSourceFiles: 150, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, ); @@ -421,10 +421,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( rootSourceFileRoles: clusterPostgres.rootSourceFileRoles, }, { - sourceFiles: 147, + sourceFiles: 149, rootSourceFiles: 1, rootSourceLines: 125, - nestedSourceFiles: 146, + nestedSourceFiles: 148, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, );