From 2bfa8ca279d511c1e944a3c6dcae8b43dd9e16dd Mon Sep 17 00:00:00 2001 From: whyour Date: Wed, 12 Aug 2026 02:57:43 +0800 Subject: [PATCH] feat(ql3): add local run log retention --- .github/workflows/ql3-ci.yml | 2 +- .../ql3-local-application/Dockerfile | 6 +- docs/QINGLONG_3_0_ARCHITECTURE_RFC.md | 17 + ...0026-local-artifact-quota-and-retention.md | 3 + ...-authorization-and-local-range-contract.md | 3 + ...un-attempt-log-retention-and-tombstones.md | 128 ++++ docs/ql3-package-boundaries.json | 2 +- .../src/run/runAttemptLogReadRoute.ts | 34 +- .../test/runAttemptLogReadRoute.test.cjs | 70 ++ .../adopted-profile/localAdoptedProfile.ts | 2 + .../src/run/runAttemptLogReadRoute.ts | 12 + .../test/runAttemptLogReadRoute.test.cjs | 25 + .../src/application-runtime/activation.ts | 30 + .../test/activation.test.cjs | 70 ++ .../src/artifact-read/index.ts | 1 + .../localRunAttemptLogRetirementStore.ts | 351 ++++++++++ .../src/control/lifecycle.ts | 14 + .../ql3-local-execution/test/control.test.cjs | 22 + ...localRunAttemptLogRetirementStore.test.cjs | 131 ++++ .../test/localDeployment.test.cjs | 45 +- .../test/localReadiness.test.cjs | 9 +- packages/ql3-local-sqlite/package.json | 5 + .../src/migration/migration.ts | 4 + .../src/migration/migrationManifest.ts | 10 + .../0087-run-attempt-log-retention.ts | 79 +++ .../src/migrations/0088-capability-v44.ts | 24 + .../src/profile/localProfile.ts | 2 + .../src/readiness/readiness.ts | 39 +- .../run/runAttemptLogRetentionRepository.ts | 428 +++++++++++++ .../src/runtime/runtimeDatabase.ts | 5 + .../ql3-local-sqlite/src/storage/schema.ts | 100 ++- .../ql3-local-sqlite/test/database.test.cjs | 14 +- ...ackageWorkflowAdmissionRepository.test.cjs | 8 +- ...lowTaskAttemptAdmissionRepository.test.cjs | 2 +- .../test/rolloutSafety.test.cjs | 6 +- .../runAttemptLogRetentionRepository.test.cjs | 194 ++++++ packages/ql3-runtime-core/package.json | 8 + .../src/run/log-read/runAttemptLogRead.ts | 55 +- .../log-retention/runAttemptLogRetention.ts | 600 ++++++++++++++++++ .../test/runAttemptLogRead.test.cjs | 88 ++- .../test/runAttemptLogRetention.test.cjs | 150 +++++ scripts/ql3-cluster-dependency-audit.cjs | 1 + ...-local-compose-preflight-live-contract.cjs | 2 +- ...l3-local-compose-rollout-live-contract.cjs | 4 +- scripts/ql3-local-image-audit.cjs | 6 +- scripts/ql3-physical-edge-compose-storage.cjs | 4 +- test/back/ql3LocalImageAudit.test.cjs | 4 +- test/back/ql3PackageBoundaryAudit.test.cjs | 14 +- .../ql3PhysicalEdgeComposeStorage.test.cjs | 2 +- test/back/ql3PhysicalEdgeEvidence.test.cjs | 2 +- 50 files changed, 2752 insertions(+), 85 deletions(-) create mode 100644 docs/adr/ADR-0378-local-run-attempt-log-retention-and-tombstones.md create mode 100644 packages/ql3-local-execution/src/artifact-read/localRunAttemptLogRetirementStore.ts create mode 100644 packages/ql3-local-execution/test/localRunAttemptLogRetirementStore.test.cjs create mode 100644 packages/ql3-local-sqlite/src/migrations/0087-run-attempt-log-retention.ts create mode 100644 packages/ql3-local-sqlite/src/migrations/0088-capability-v44.ts create mode 100644 packages/ql3-local-sqlite/src/run/runAttemptLogRetentionRepository.ts create mode 100644 packages/ql3-local-sqlite/test/runAttemptLogRetentionRepository.test.cjs create mode 100644 packages/ql3-runtime-core/src/run/log-retention/runAttemptLogRetention.ts create mode 100644 packages/ql3-runtime-core/test/runAttemptLogRetention.test.cjs diff --git a/.github/workflows/ql3-ci.yml b/.github/workflows/ql3-ci.yml index 943b3fb5..1f601611 100644 --- a/.github/workflows/ql3-ci.yml +++ b/.github/workflows/ql3-ci.yml @@ -371,7 +371,7 @@ jobs: - name: Verify non-root identity and architecture env: IMAGE: qinglong3-local-application:ci-${{ matrix.image_arch }} - EXPECTED: ${{ matrix.image_arch }} 65532:65532 2 37 37 37 1 + EXPECTED: ${{ matrix.image_arch }} 65532:65532 2 44 44 44 1 run: | set -euo pipefail actual="$(docker image inspect --format '{{.Architecture}} {{.Config.User}} {{index .Config.Labels "io.qinglong.local.application-config"}} {{index .Config.Labels "io.qinglong.local.sqlite-contract-min"}} {{index .Config.Labels "io.qinglong.local.sqlite-contract-max"}} {{index .Config.Labels "io.qinglong.local.sqlite-write-contract"}} {{index .Config.Labels "io.qinglong.local.compose-selection"}}' "${IMAGE}")" diff --git a/deploy/containers/ql3-local-application/Dockerfile b/deploy/containers/ql3-local-application/Dockerfile index 01955a8c..0d715904 100644 --- a/deploy/containers/ql3-local-application/Dockerfile +++ b/deploy/containers/ql3-local-application/Dockerfile @@ -143,9 +143,9 @@ LABEL org.opencontainers.image.title="QingLong 3.0 Local Application" \ io.qinglong.profile="edge,standalone" \ io.qinglong.ai="excluded" \ io.qinglong.local.application-config="2" \ - io.qinglong.local.sqlite-contract-min="43" \ - io.qinglong.local.sqlite-contract-max="43" \ - io.qinglong.local.sqlite-write-contract="43" \ + io.qinglong.local.sqlite-contract-min="44" \ + io.qinglong.local.sqlite-contract-max="44" \ + io.qinglong.local.sqlite-write-contract="44" \ io.qinglong.local.compose-selection="1" ENV NODE_ENV=production diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index 32d45bdf..bbac469c 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -11,6 +11,23 @@ 最新增量证据（2026-08-12）： +- D-290/ADR-0378（已接受） + Local Run Attempt 日志 retention 已形成真实纵向切片：Runtime Core 增加精确 identity、canonical SHA-256 的 immutable + retirement record、容量压力策略、有界 page/delete budget 与 durable cursor；日志读取在存储前检查 tombstone，并在 missing 后二次 + 检查，授权调用方稳定获得 `410 retired`，未授权/不存在继续在 metadata 前遮蔽为 404。SQLite 追加 0087/0088，将 Local contract + 升至 v44，候选严格排除 lost、legacy owner、非 Local executor、未终态 Run/Attempt、未完成 receipt 与既有 tombstone。私有文件端只 + 删除 0700 owner-only shard 中 0600、单 hard-link 的 canonical 日志及唯一 truncation fact，执行 unlink→helper unlink→directory + fsync→tombstone；unlink 后崩溃以下一轮 `already_absent` 收敛。Edge 为 7 天/压力 24 小时、64 MiB 阈值、page 4/delete 2； + Standalone 为 30 天/压力 24 小时、256 MiB、page 16/delete 8。sweep 复用现有 completion cleanup cadence，不新增 package、依赖、 + timer、listener、连接、watcher 或 cache。Runtime Core 定向 9/9、Local SQLite 220/220、文件与 control 定向 10/10、应用真实删除+ + tombstone+读取闭环通过。最终验收为 Runtime Core 498/498、Local SQLite 220/220、Local Execution 39/39、Local API + 45/45、Local Application 46 pass/4 skip、Local Admin 91/91、Local Owner CLI 157 pass/5 skip、Cluster Control + 216 pass/2 skip；完整 18-package 门退出 0，backend 1,163 pass/2 skip/0 fail。package/dependency/Edge import + 审计与 14 档 Profile artifact 全部 compatible，且没有 single-source/shallow package。真实 arm64 Local image 以固定 repo + digest 完成 Edge/Standalone preflight 和 rollout，均观测 SQLite contract v44；镜像、静态审计和 CI compatibility label + 已统一为 v44。PostgreSQL 18.4 arm64 HA 回归通过 112 gates、timeline 1→2，独立证据审计无 finding。Cluster 仅冻结可注入的 + 410 语义； + PostgreSQL 多副本 claim/backoff、ETag/version 条件 S3 删除与 HA/MinIO 门由 D-291 独立完成。 - D-289/ADR-0377（已接受） Local/Cluster 已增加同构 `GET /api/v3/projects/{projectId}/runs/{runId}/attempts/{attemptId}/log`、 `run.log.read`/`artifact.read`。请求只接受 Project/Run/Attempt identity 与 offset/length，不接受 Artifact ID、路径、URI、bucket diff --git a/docs/adr/ADR-0026-local-artifact-quota-and-retention.md b/docs/adr/ADR-0026-local-artifact-quota-and-retention.md index 20a6ee88..ca6f21ed 100644 --- a/docs/adr/ADR-0026-local-artifact-quota-and-retention.md +++ b/docs/adr/ADR-0026-local-artifact-quota-and-retention.md @@ -4,6 +4,9 @@ - 日期：2026-07-18 - 关联：QL-RFC-0001、ADR-0007、ADR-0021、ADR-0024、ADR-0025 +> 2026-08-12：Local Run Attempt 日志 retention、durable tombstone、压力档位与 lifecycle 接线已由 +> ADR-0378 接受；本 ADR 仍保持 Proposed，仅表示其余通用 Artifact quota/跨 Profile 扩展尚未整体关闭。 + ## 上下文 ADR-0024 已建立 Attempt-scoped opaque Artifact 和 direct-file durable output，但未限制单次运行能够写入的字节数，也没有终态清理证据。对小型路由设备，这意味着一个失控脚本可以填满系统盘；只按 timer 删除目录又可能误删仍在运行、等待 completion receipt 或无法证明进程退出的日志。standalone/cluster 虽然容量更高，同样需要可审计的配额和生命周期语义。 diff --git a/docs/adr/ADR-0027-artifact-read-authorization-and-local-range-contract.md b/docs/adr/ADR-0027-artifact-read-authorization-and-local-range-contract.md index f73a5f21..1bcb0cf6 100644 --- a/docs/adr/ADR-0027-artifact-read-authorization-and-local-range-contract.md +++ b/docs/adr/ADR-0027-artifact-read-authorization-and-local-range-contract.md @@ -4,6 +4,9 @@ - 日期：2026-07-19 - 关联：QL-RFC-0001、ADR-0024、ADR-0026 +> 2026-08-12：Local/Cluster 生产读取接线由 ADR-0377 接受，Local retention 二次检查与 durable 410 +> 由 ADR-0378 接受；Cluster retention 删除实现仍属于 D-291，本 ADR 的其余跨 adapter 扩展继续保持 Proposed。 + ## 上下文 2.x 日志 API 由全局登录或 Open API scope 保护，读取端仍接受 `path + file`，无法表达 Project 所有权、`artifact.read` 策略和 opaque Artifact identity。3.0 已把 LocalProcess 日志绑定到 `(projectId, runId, attemptId, logArtifactId)`，并用 retention tombstone 与 canonical truncation fact 保存清理和截断证据，但尚无稳定读取语义。 diff --git a/docs/adr/ADR-0378-local-run-attempt-log-retention-and-tombstones.md b/docs/adr/ADR-0378-local-run-attempt-log-retention-and-tombstones.md new file mode 100644 index 00000000..d5fa608b --- /dev/null +++ b/docs/adr/ADR-0378-local-run-attempt-log-retention-and-tombstones.md @@ -0,0 +1,128 @@ +# ADR-0378：Local Run Attempt 日志有界保留与 durable tombstone + +- 状态：Accepted +- 日期：2026-08-12 +- 关联 RFC：QL-RFC-0001 D-290 +- 前置决策：ADR-0026、ADR-0027、ADR-0377 + +## 上下文 + +ADR-0377 已让 Local/Cluster 能在 Project、Run、Attempt、executor 与 Artifact identity 全部可信后读取日志，但故意没有把缺失文件解释成 retention。若没有 durable tombstone，文件被清理、文件损坏、错误路径和未发布 Artifact 对读取方都是同一个 `missing`；若只在删除前写标记，又会把仍存在的数据错误暴露为已清理。 + +Local Profile 还必须覆盖两类差异很大的设备：128 MiB/64 PID 级路由设备不能承担目录全扫、每任务定时器或无界删除；Standalone 可以提高吞吐，但仍应复用单连接和既有 lifecycle。Cluster 的 PostgreSQL 多副本 claim、S3 条件删除和外部网络失败语义与本机 unlink 不同，不能为了表面复用而塞进同一个事务或 adapter。 + +## 决策 + +### 1. 共享不可变证据，不共享删除实现 + +`runtime-core` 在既有 Run 能力内增加 `qinglong/run-attempt-log-retirement@v1`，不创建新 package。每条记录精确绑定： + +- `projectId`、`runId`、`attemptId`、`logArtifactId`； +- `executorType`、`finishedAtMs`、`eligibleAtMs`、`retiredAtMs`； +- `deleted | already_absent`； +- 删除前观测到的 `byteLength` 与 `true | false | unknown` truncation； +- 覆盖全部语义字段的 canonical SHA-256 `recordDigest`。 + +读取 repository 必须重新计算摘要并验证精确身份；持久化行被原地修改时失败关闭。`already_absent` 的 byteLength 固定为 0，不伪造回收字节。 + +### 2. 读取与删除竞态 + +读取仍先完成 HTTP authentication、Policy、durable audit、credential confirmation 和 Run/Attempt metadata 验证。获得 canonical Artifact identity 后： + +1. 在访问文件/对象前检查 tombstone； +2. 已 retired 直接返回 `410 artifact retired`，不触碰存储； +3. active 时执行原有 Range read； +4. 存储返回 missing 后再次检查 tombstone； +5. 二次检查发现 retired 返回 410，否则保持 503 unavailable/pending 语义。 + +打开文件后发生 unlink 的并发读取可完成该已打开快照；这是以安全打开的 inode 为线性化点。删除完成但 tombstone 尚未提交的极短崩溃窗口由下次 `already_absent` sweep 收敛，不能预写“已删除”tombstone。 + +对外 `retired` 只包含 Project/Run/Attempt、retired time、原 byte length 与 truncation，不返回路径、Artifact ID、bucket 或 key。Policy deny、不存在和跨 Project 继续在此之前遮蔽为 404。 + +### 3. Local SQLite 44 号契约 + +追加 `0087-run-attempt-log-retention` 与 `0088-capability-v44`： + +- `QingLong3RunAttemptLogArtifactTombstones` 保存不可变精确证据； +- `QingLong3RunAttemptLogRetentionState` 保存唯一 Local maintenance cursor； +- `RunAttempts` 增加局部候选索引； +- runtime readiness 仍只读取冻结 manifest，不加载可执行 DDL。 + +候选必须同时满足： + +- Run `executionOwner=runtime`； +- Run 和 Attempt 都是 `succeeded | failed | cancelled | timed_out`，明确排除 `lost`； +- Attempt 为 `local_process` 且绑定 canonical `local-*` 日志； +- Run/Attempt 都有终态时间且早于当前 retention cutoff； +- 不存在任何 completion receipt journal； +- 不存在同 Attempt 或 Artifact 的 tombstone。 + +删除后写 tombstone 的事务会重新验证上述持久化事实。若 receipt 或状态在文件删除后改变，写入失败，后续 sweep 以 absent 重新收敛，不能绕过 completion 协议。 + +### 4. 私有文件删除顺序 + +Local store 只接受： + +- 绝对且非文件系统根的 Artifact root； +- 当前 owner、模式 0700、非 symlink 的 root/shard； +- 当前 owner、模式 0600、普通文件、单 hard link、最大 1 GiB 的主日志； +- 唯一允许的 `..log.truncated.json` helper，且其 identity 与主记录完全一致。 + +顺序固定为：验证目录和 fact → 打开并验证主日志 inode → 删除主日志 → 删除允许的 fact → fsync shard directory → SQLite 写 tombstone。禁止递归删除、扫描任意目录、跟随 symlink 或清理未知 helper。若主日志已不存在但 exact fact 仍在，只删除该 fact、fsync 并记录 `already_absent`。 + +### 5. Edge 与 Standalone 资源档位 + +每次 sweep 先用 `statfs` 采样可用空间，根目录尚未创建时只检查最近存在父目录，不主动创建路径。 + +| Profile | 正常保留 | 压力保留 | 压力阈值 | page | 最大删除 | +| --- | ---: | ---: | ---: | ---: | ---: | +| Edge | 7 天 | 24 小时 | 64 MiB | 4 | 2 | +| Standalone | 30 天 | 24 小时 | 256 MiB | 16 | 8 | + +cursor 持久化到 SQLite；失败候选会被本轮越过，并在游标回绕后重试，避免一个坏文件永久阻塞队首。容量证据以十进制字符串进入 lifecycle summary，避免 JSON 对 BigInt 失败。 + +### 6. 不新增常驻资源 + +Artifact sweep 复用 `LocalExecutionControlLifecycle` 已有 completion cleanup cadence：先清 completion receipt，再做一次有界 Artifact sweep。它不新增 timer、listener、SQLite connection、watcher、cache 或 background thread；stop drain 不额外执行 Artifact 删除，避免扩大关停超时。 + +### 7. Cluster 留给 D-291 + +Cluster 路由和共享读取服务已经能注入 retention state 并映射 410，但 D-290 不授予 S3 DeleteObject 权限、不新增 PostgreSQL tombstone/claim，也不宣称 Cluster retention 已完成。D-291 必须独立解决： + +- 多副本 durable claim、lease/backoff 与 bounded scheduler budget； +- validated HEAD 后基于 ETag/version 的条件删除； +- PostgreSQL 事务不得跨 S3 网络调用； +- delete 成功/对象已不存在后的 exact tombstone finalize； +- MinIO/S3 失败矩阵与 PostgreSQL HA failover。 + +## 被否决的替代方案 + +1. **删除前写 tombstone**：可能把仍存在的数据声明为已删除，拒绝。 +2. **把 missing 直接映射 410**：无法区分损坏、误配和 retention，拒绝。 +3. **递归扫描 Artifact root**：I/O 无界且目录内容成为隐式 authority，拒绝。 +4. **每个 Run/Attempt 启动 retention timer**：常驻资源随任务数增长，拒绝。 +5. **Local/Cluster 共用一个删除事务 adapter**：会把外部 S3 调用放入数据库事务并模糊多副本 ownership，拒绝。 +6. **新建 retention 微包**：现有 runtime-core、local-sqlite、local-execution ownership 已足够，拒绝。 + +## 验收 + +1. 共享 contract 覆盖 digest tamper、精确 identity、压力策略、删除预算和 durable cursor； +2. 读取覆盖 tombstone-before-storage 与 missing-after-storage 二次检查； +3. SQLite migration、typed schema、readiness、candidate、receipt fence、replay/tamper 全部通过； +4. 文件 adapter 覆盖权限、symlink/hard-link、fact drift、unlink-before-tombstone 与 statfs； +5. 完整 Local application 证明启动 sweep 删除真实文件、写 tombstone，并在产品读取前返回 retired； +6. Runtime Core、Local SQLite/Execution/API/Application/Admin/Owner CLI、Cluster Control 与完整 18-package/backend/boundary/Profile/Local image 门全部通过后才改为 Accepted。 + +2026-08-12 验收完成：Runtime Core 498/498、Local SQLite 220/220、Local Execution 39/39、Local API +45/45、Local Application 46 pass/4 skip、Local Admin 91/91、Local Owner CLI 157 pass/5 skip、Cluster +Control 216 pass/2 skip；完整 18-package 门退出 0，backend 1,163 pass/2 skip/0 fail。package boundary 保持 +18 个 workspace package，`singleSourcePackages=[]`、`shallowSourcePackages=[]`；dependency 与 Edge import 审计 +均 compatible。14 个 Local Profile artifact 全部在文件数、体积与 RSS 预算内，最小 Edge 为 2,450,378 bytes， +Edge Application 为 3,482,708 bytes，未引入 Cluster/PostgreSQL/AWS SDK 闭包。 + +真实 arm64 Local image 以 repo digest +`sha256:73a93094ebc53effbbe619e29ea866e59872211bf840db86509c011389ed10b8` 完成 Edge/Standalone +Compose preflight，均观测 SQLite contract v44；两档 rollout 的备份、恢复、响应丢失重放、证据收集与优雅清理 +全部 compatible。PostgreSQL 18.4 arm64 HA 回归通过 112 gates、timeline `1→2`，私有报告 SHA-256 为 +`6a205d8bc596097f91a900d7cdabef21c6ff3ad61152e8dfb31291cb8a356b12`，独立审计 +`compatible=true/findings=[]`。发布镜像、静态审计和 CI 的 SQLite compatibility label 已统一为 v44。 diff --git a/docs/ql3-package-boundaries.json b/docs/ql3-package-boundaries.json index da455ab0..6c37cee6 100644 --- a/docs/ql3-package-boundaries.json +++ b/docs/ql3-package-boundaries.json @@ -12,7 +12,7 @@ }, { "kind": "ordered_ledger", - "maxDirectSourceFiles": 87, + "maxDirectSourceFiles": 89, "path": "packages/ql3-local-sqlite/src/migrations", "rationale": "SQLite migrations are an append-only version ledger whose ordering and discoverability are safer in one reviewed directory." } diff --git a/packages/ql3-cluster-control/src/run/runAttemptLogReadRoute.ts b/packages/ql3-cluster-control/src/run/runAttemptLogReadRoute.ts index 0e98d0ec..ff3a0c44 100644 --- a/packages/ql3-cluster-control/src/run/runAttemptLogReadRoute.ts +++ b/packages/ql3-cluster-control/src/run/runAttemptLogReadRoute.ts @@ -6,6 +6,7 @@ import { type RunAttemptLogReadResult, } from '@qinglong/runtime-core/run-attempt-log-read'; import type { RunRepositoryReader } from '@qinglong/runtime-core/run-repository'; +import type { RunAttemptLogRetentionStateReader } from '@qinglong/runtime-core/run-attempt-log-retention'; import type { ClusterControlAdmissionResponse } from '../transport/httpSurface'; import type { @@ -99,12 +100,14 @@ function projection( export function createClusterControlRunAttemptLogReadRoute( runs: Pick, reader?: RunAttemptLogRangeReader, + retention?: RunAttemptLogRetentionStateReader, ): Readonly { if ( !runs || typeof runs.findRunById !== 'function' || typeof runs.findAttemptById !== 'function' || - (reader !== undefined && typeof reader.read !== 'function') + (reader !== undefined && typeof reader.read !== 'function') || + (retention !== undefined && typeof retention.inspect !== 'function') ) { throw new TypeError( 'Cluster-control Run Attempt log read dependencies are invalid', @@ -113,12 +116,17 @@ export function createClusterControlRunAttemptLogReadRoute( const service = reader === undefined ? undefined - : new RunAttemptLogReadService(runs, reader, { - executorType: 'remote_worker', - artifactIdPattern: /^wlog-[a-f0-9]{30}$/, - maximumReadBytes: MAXIMUM_READ_BYTES, - activeMissingIsPending: true, - }); + : new RunAttemptLogReadService( + runs, + reader, + { + executorType: 'remote_worker', + artifactIdPattern: /^wlog-[a-f0-9]{30}$/, + maximumReadBytes: MAXIMUM_READ_BYTES, + activeMissingIsPending: true, + }, + retention, + ); return Object.freeze({ ...CLUSTER_CONTROL_RUN_ATTEMPT_LOG_READ_ROUTE, validateQuery, @@ -161,6 +169,18 @@ export function createClusterControlRunAttemptLogReadRoute( if (result.status === 'missing') { return response(503, { code: 'artifact_unavailable' }); } + if (result.status === 'retired') { + return response(410, { + schema: 'qinglong/run-attempt-log-read-result@v1', + status: 'retired', + projectId: result.projectId, + runId: result.runId, + attemptId: result.attemptId, + retiredAtMs: result.retiredAtMs, + byteLength: result.byteLength, + truncation: result.truncation, + }); + } return response(200, projection(result)); } catch (error) { if (error instanceof InvalidRunAttemptLogReadError) { diff --git a/packages/ql3-cluster-control/test/runAttemptLogReadRoute.test.cjs b/packages/ql3-cluster-control/test/runAttemptLogReadRoute.test.cjs index 80e606ba..451a1a6b 100644 --- a/packages/ql3-cluster-control/test/runAttemptLogReadRoute.test.cjs +++ b/packages/ql3-cluster-control/test/runAttemptLogReadRoute.test.cjs @@ -249,3 +249,73 @@ test('returns pending during upload and fails closed without an object reader', body: { code: 'artifact_unavailable' }, }); }); + +test('maps an injected durable retention state to 410 without object access', async () => { + const { + createRunAttemptLogRetirementRecord, + } = require('@qinglong/runtime-core/run-attempt-log-retention'); + let reads = 0; + const route = createClusterControlRunAttemptLogReadRoute( + { + async findRunById() { + return run({ status: 'succeeded', finishedAtMs: 10 }); + }, + async findAttemptById() { + return attempt({ status: 'succeeded', finishedAtMs: 10 }); + }, + }, + { + async read() { + reads += 1; + return { status: 'missing' }; + }, + }, + { + async inspect() { + return { + status: 'retired', + record: createRunAttemptLogRetirementRecord({ + projectId: 'prj_default', + runId: 'run_123', + attemptId: 'attempt_123', + logArtifactId: `wlog-${'a'.repeat(30)}`, + executorType: 'remote_worker', + finishedAtMs: 10, + eligibleAtMs: 20, + retiredAtMs: 30, + disposition: 'deleted', + byteLength: 42, + truncation: { truncated: 'unknown' }, + }), + }; + }, + }, + ); + const prepared = await createClusterControlAdmissionPipeline({ + routes: createClusterControlRouteRegistry([route]), + authenticator: { authenticate: () => PRINCIPAL }, + policy: { + authorize: () => ({ + effect: 'allow', + reasons: ['role_grant'], + fence: { projectVersion: 1, bindingVersion: 1 }, + }), + }, + audit: { record() {} }, + now: () => 10_000, + }).prepare(metadata()); + assert.deepEqual(await prepared.handle(null), { + statusCode: 410, + body: { + schema: 'qinglong/run-attempt-log-read-result@v1', + status: 'retired', + projectId: 'prj_default', + runId: 'run_123', + attemptId: 'attempt_123', + retiredAtMs: 30, + byteLength: 42, + truncation: { truncated: 'unknown' }, + }, + }); + assert.equal(reads, 0); +}); diff --git a/packages/ql3-local-admin/src/adopted-profile/localAdoptedProfile.ts b/packages/ql3-local-admin/src/adopted-profile/localAdoptedProfile.ts index 1db7d76c..525aa0f0 100644 --- a/packages/ql3-local-admin/src/adopted-profile/localAdoptedProfile.ts +++ b/packages/ql3-local-admin/src/adopted-profile/localAdoptedProfile.ts @@ -87,6 +87,7 @@ export type LocalAdoptedProfileBootstrapResult = readonly dispatch: LocalDispatchStore; readonly executionControl: ReadyLocalStorage['executionControl']; readonly completionReceipts: ReadyLocalStorage['completionReceipts']; + readonly runAttemptLogRetention: ReadyLocalStorage['runAttemptLogRetention']; readonly localSecrets: LocalSecretEnvelopeRepository; readonly localSecretAdministration: LocalSecretAdministrationRepository; readonly projectPolicy: ProjectPolicyRepository; @@ -231,6 +232,7 @@ export async function bootstrapLocalAdoptedProfileStorage( dispatch: readyStorage.dispatch, executionControl: readyStorage.executionControl, completionReceipts: readyStorage.completionReceipts, + runAttemptLogRetention: readyStorage.runAttemptLogRetention, localSecrets: readyStorage.localSecrets, localSecretAdministration: readyStorage.localSecretAdministration, projectPolicy: readyStorage.projectPolicy, diff --git a/packages/ql3-local-api/src/run/runAttemptLogReadRoute.ts b/packages/ql3-local-api/src/run/runAttemptLogReadRoute.ts index 69eafca8..9c5aa843 100644 --- a/packages/ql3-local-api/src/run/runAttemptLogReadRoute.ts +++ b/packages/ql3-local-api/src/run/runAttemptLogReadRoute.ts @@ -96,6 +96,18 @@ export function createLocalApiRunAttemptLogReadRoute( if (result.status === 'missing') { return response(503, { code: 'artifact_unavailable' }); } + if (result.status === 'retired') { + return response(410, { + schema: 'qinglong/run-attempt-log-read-result@v1', + status: 'retired', + projectId: result.projectId, + runId: result.runId, + attemptId: result.attemptId, + retiredAtMs: result.retiredAtMs, + byteLength: result.byteLength, + truncation: result.truncation, + }); + } return response(200, projection(result)); } catch (error) { if (error instanceof InvalidRunAttemptLogReadError) { diff --git a/packages/ql3-local-api/test/runAttemptLogReadRoute.test.cjs b/packages/ql3-local-api/test/runAttemptLogReadRoute.test.cjs index 28923451..3fe04ab9 100644 --- a/packages/ql3-local-api/test/runAttemptLogReadRoute.test.cjs +++ b/packages/ql3-local-api/test/runAttemptLogReadRoute.test.cjs @@ -88,6 +88,31 @@ test('maps pending, masked absence, missing storage and unavailable evidence', a }, { statusCode: 503, body: { code: 'artifact_unavailable' } }, ], + [ + { + status: 'retired', + projectId: 'prj_default', + runId: 'run_123', + attemptId: 'attempt_123', + logArtifactId: `local-${'a'.repeat(30)}`, + retiredAtMs: 30, + byteLength: 42, + truncation: { truncated: 'unknown' }, + }, + { + statusCode: 410, + body: { + schema: 'qinglong/run-attempt-log-read-result@v1', + status: 'retired', + projectId: 'prj_default', + runId: 'run_123', + attemptId: 'attempt_123', + retiredAtMs: 30, + byteLength: 42, + truncation: { truncated: 'unknown' }, + }, + }, + ], ]; for (const [result, expected] of cases) { const route = createLocalApiRunAttemptLogReadRoute({ diff --git a/packages/ql3-local-application/src/application-runtime/activation.ts b/packages/ql3-local-application/src/application-runtime/activation.ts index a4e0157c..9bfca581 100644 --- a/packages/ql3-local-application/src/application-runtime/activation.ts +++ b/packages/ql3-local-application/src/application-runtime/activation.ts @@ -75,6 +75,11 @@ const EXECUTION_CONTROL_POLICIES = Object.freeze({ controlPageSize: 4, maxDrainPages: 2, retentionMs: 24 * 60 * 60_000, + artifactNormalRetentionMs: 7 * 24 * 60 * 60_000, + artifactPressureRetentionMs: 24 * 60 * 60_000, + artifactMinimumFreeBytes: 64 * 1024 * 1024, + artifactRetentionPageSize: 4, + artifactMaximumDeletions: 2, stopTimeoutMs: 5_000, }), standalone: Object.freeze({ @@ -84,6 +89,11 @@ const EXECUTION_CONTROL_POLICIES = Object.freeze({ controlPageSize: 32, maxDrainPages: 8, retentionMs: 60 * 60_000, + artifactNormalRetentionMs: 30 * 24 * 60 * 60_000, + artifactPressureRetentionMs: 24 * 60 * 60_000, + artifactMinimumFreeBytes: 256 * 1024 * 1024, + artifactRetentionPageSize: 16, + artifactMaximumDeletions: 8, stopTimeoutMs: 10_000, }), }); @@ -267,6 +277,24 @@ export async function bootstrapLocalApplication( }); const executionPolicy = EXECUTION_CONTROL_POLICIES[options.profile]; + const [artifactStorage, retentionCore] = await Promise.all([ + import('@qinglong/local-execution/artifact-read'), + import('@qinglong/runtime-core/run-attempt-log-retention'), + ]); + const artifactRetention = new retentionCore.RunAttemptLogRetentionService( + storage.runAttemptLogRetention, + new artifactStorage.LocalRunAttemptLogRetirementStore( + options.artifactRoot, + ), + new artifactStorage.LocalRunAttemptLogCapacityProbe(options.artifactRoot), + { + normalRetentionMs: executionPolicy.artifactNormalRetentionMs, + pressureRetentionMs: executionPolicy.artifactPressureRetentionMs, + minimumFreeBytes: executionPolicy.artifactMinimumFreeBytes, + pageSize: executionPolicy.artifactRetentionPageSize, + maximumDeletions: executionPolicy.artifactMaximumDeletions, + }, + ); const receipts = new CompletionReceiptFileStore(options.receiptRoot); const localProcessLauncher = new LocalProcessLauncher( storage.completionReceipts, @@ -316,6 +344,7 @@ export async function bootstrapLocalApplication( cleanupPageSize: executionPolicy.cleanupPageSize, stopTimeoutMs: executionPolicy.stopTimeoutMs, maxDrainPages: executionPolicy.maxDrainPages, + artifactRetention, onDiagnostic: async (error) => { if (error === undefined) return; await bestEffortAudit(options, { @@ -476,6 +505,7 @@ export async function bootstrapLocalApplication( artifactIdPattern: /^local-[a-f0-9]{30}$/, maximumReadBytes: 32 * 1024, }, + storage.runAttemptLogRetention, ); productSurfaceLifecycle = await options.productSurface.start( Object.freeze({ diff --git a/packages/ql3-local-application/test/activation.test.cjs b/packages/ql3-local-application/test/activation.test.cjs index 9eb69afb..47bb383b 100644 --- a/packages/ql3-local-application/test/activation.test.cjs +++ b/packages/ql3-local-application/test/activation.test.cjs @@ -1733,6 +1733,76 @@ test('starts an optional product surface after recovery and drains it before own ); }); +test('retires one eligible Local log before product reads and returns durable 410 state', async (t) => { + const value = await prepare(t, 'edge'); + const runId = 'retention-run-1'; + const attemptId = 'retention-attempt-1'; + const artifactId = `local-${'b'.repeat(30)}`; + const database = new DatabaseSync(value.targetPath); + database + .prepare( + `INSERT INTO "Runs" ( + id, project_id, task_id, task_revision, trigger_type, + execution_origin, execution_owner, status, version, event_sequence, + priority, created_at_ms, finished_at_ms + ) VALUES (?, 'default', 'task-retention', 'revision-1', 'manual', + 'manual', 'runtime', 'succeeded', 1, 1, 0, 1, 1)`, + ) + .run(runId); + database + .prepare( + `INSERT INTO "RunAttempts" ( + id, run_id, attempt, status, executor_type, log_artifact_id, + callback_sequence, created_at_ms, finished_at_ms + ) VALUES (?, ?, 1, 'succeeded', 'local_process', ?, 0, 1, 1)`, + ) + .run(attemptId, runId, artifactId); + database.close(); + + const artifactRoot = path.join(value.directory, 'artifacts'); + const shard = path.join(artifactRoot, 'bb'); + fs.mkdirSync(shard, { recursive: true, mode: 0o700 }); + fs.chmodSync(artifactRoot, 0o700); + fs.chmodSync(shard, 0o700); + const logPath = path.join(shard, `${artifactId}.log`); + fs.writeFileSync(logPath, 'expired', { mode: 0o600 }); + fs.chmodSync(logPath, 0o600); + + let readResult; + const result = await bootstrapLocalApplication( + options(value, { + productSurface: { + async start(authority) { + readResult = await authority.runAttemptLogRead.read({ + projectId: 'default', + runId, + attemptId, + range: { offset: 0, length: 16 }, + }); + return { stopAndDrain: async () => 'stopped' }; + }, + }, + }), + ); + assert.equal(readResult.status, 'retired'); + assert.equal(readResult.byteLength, 7); + assert.equal(readResult.truncation.truncated, 'unknown'); + assert.equal(fs.existsSync(logPath), false); + const evidence = new DatabaseSync(value.targetPath, { readonly: true }); + const tombstone = evidence + .prepare( + `SELECT disposition, byte_length AS "byteLength", record_digest AS "recordDigest" + FROM "QingLong3RunAttemptLogArtifactTombstones" + WHERE attempt_id = ?`, + ) + .get(attemptId); + evidence.close(); + assert.equal(tombstone.disposition, 'deleted'); + assert.equal(tombstone.byteLength, 7); + assert.match(tombstone.recordDigest, /^[a-f0-9]{64}$/); + assert.equal(await result.stop(), 'stopped'); +}); + test('executes one admitted Workflow through the single application cadence without duplicate Tasks', async (t) => { if (process.platform !== 'linux') { t.skip('durable local process identity requires Linux /proc'); diff --git a/packages/ql3-local-execution/src/artifact-read/index.ts b/packages/ql3-local-execution/src/artifact-read/index.ts index 6cb8526a..d4f0e120 100644 --- a/packages/ql3-local-execution/src/artifact-read/index.ts +++ b/packages/ql3-local-execution/src/artifact-read/index.ts @@ -1 +1,2 @@ export * from './localRunAttemptLogRangeReader'; +export * from './localRunAttemptLogRetirementStore'; diff --git a/packages/ql3-local-execution/src/artifact-read/localRunAttemptLogRetirementStore.ts b/packages/ql3-local-execution/src/artifact-read/localRunAttemptLogRetirementStore.ts new file mode 100644 index 00000000..d912f2d8 --- /dev/null +++ b/packages/ql3-local-execution/src/artifact-read/localRunAttemptLogRetirementStore.ts @@ -0,0 +1,351 @@ +import { constants, type Stats } from 'node:fs'; +import fs, { type FileHandle } from 'node:fs/promises'; +import path from 'node:path'; + +import { + normalizeRunAttemptLogRetentionCandidate, + type RunAttemptLogCapacitySource, + type RunAttemptLogRetentionCandidate, + type RunAttemptLogRetirementStore, + type RunAttemptLogRetirementStoreResult, +} from '@qinglong/runtime-core/run-attempt-log-retention'; +import type { + RunAttemptLogReadIdentity, + RunAttemptLogTruncationView, +} from '@qinglong/runtime-core/run-attempt-log-read'; + +const LOCAL_ARTIFACT_ID = /^local-[a-f0-9]{30}$/; +const MAXIMUM_ARTIFACT_BYTES = 1024 * 1024 * 1024; +const MAXIMUM_FACT_BYTES = 1024; + +export class LocalRunAttemptLogRetirementError extends Error { + constructor( + readonly reason: + | 'invalid_configuration' + | 'unsafe_path' + | 'integrity_mismatch', + options?: ErrorOptions, + ) { + super(`Local Run Attempt log retirement failed: ${reason}`, options); + this.name = 'LocalRunAttemptLogRetirementError'; + } +} + +function isCode(error: unknown, code: string): boolean { + return ( + !!error && + typeof error === 'object' && + 'code' in error && + (error as { code?: unknown }).code === code + ); +} + +function currentUid(): number | undefined { + return typeof process.getuid === 'function' ? process.getuid() : undefined; +} + +function artifactRoot(value: string): string { + if ( + typeof value !== 'string' || + !path.isAbsolute(value) || + path.parse(value).root === value || + value.includes('\0') || + Buffer.byteLength(value, 'utf8') > 4096 + ) { + throw new LocalRunAttemptLogRetirementError('invalid_configuration'); + } + return path.resolve(value); +} + +function assertOwnedDirectory(stat: Stats): void { + const uid = currentUid(); + if ( + !stat.isDirectory() || + stat.isSymbolicLink() || + (stat.mode & 0o777) !== 0o700 || + (uid !== undefined && stat.uid !== uid) + ) { + throw new LocalRunAttemptLogRetirementError('unsafe_path'); + } +} + +function assertOwnedFile(stat: Stats, maximumBytes: number): void { + const uid = currentUid(); + if ( + !stat.isFile() || + stat.isSymbolicLink() || + stat.nlink !== 1 || + (stat.mode & 0o777) !== 0o600 || + (uid !== undefined && stat.uid !== uid) || + !Number.isSafeInteger(stat.size) || + stat.size < 0 || + stat.size > maximumBytes + ) { + throw new LocalRunAttemptLogRetirementError('unsafe_path'); + } +} + +async function optionalOwnedDirectory(directory: string): Promise { + try { + assertOwnedDirectory(await fs.lstat(directory)); + return true; + } catch (error) { + if (isCode(error, 'ENOENT')) return false; + if (error instanceof LocalRunAttemptLogRetirementError) throw error; + throw new LocalRunAttemptLogRetirementError('unsafe_path', { + cause: error, + }); + } +} + +async function openPrivateFile( + filePath: string, +): Promise { + try { + return await fs.open( + filePath, + constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0), + ); + } catch (error) { + if (isCode(error, 'ENOENT')) return undefined; + throw new LocalRunAttemptLogRetirementError('unsafe_path', { + cause: error, + }); + } +} + +function sameFile(left: Stats, right: Stats): boolean { + return left.dev === right.dev && left.ino === right.ino; +} + +function exactFact( + value: unknown, + expected: Readonly, +): Readonly { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new LocalRunAttemptLogRetirementError('integrity_mismatch'); + } + const fact = value as Record; + if ( + Object.keys(fact).sort().join(',') !== + 'attemptId,logArtifactId,maximumBytes,observedAtMs,quotaReached,runId,schemaVersion' || + fact.schemaVersion !== 1 || + fact.runId !== expected.runId || + fact.attemptId !== expected.attemptId || + fact.logArtifactId !== expected.logArtifactId || + !Number.isSafeInteger(fact.maximumBytes) || + Number(fact.maximumBytes) < 64 * 1024 || + Number(fact.maximumBytes) > MAXIMUM_ARTIFACT_BYTES || + typeof fact.quotaReached !== 'boolean' || + !Number.isSafeInteger(fact.observedAtMs) || + Number(fact.observedAtMs) < 0 + ) { + throw new LocalRunAttemptLogRetirementError('integrity_mismatch'); + } + return Object.freeze({ + truncated: fact.quotaReached, + maximumBytes: fact.maximumBytes as number, + observedAtMs: fact.observedAtMs as number, + }); +} + +async function readFact( + factPath: string, + expected: Readonly, +): Promise< + Readonly<{ + truncation: Readonly; + stat?: Stats; + }> +> { + const handle = await openPrivateFile(factPath); + if (!handle) { + return Object.freeze({ + truncation: Object.freeze({ truncated: 'unknown' as const }), + }); + } + try { + const before = await handle.stat(); + assertOwnedFile(before, MAXIMUM_FACT_BYTES); + if (before.size < 2) { + throw new LocalRunAttemptLogRetirementError('integrity_mismatch'); + } + const content = Buffer.allocUnsafe(before.size); + try { + let read = 0; + while (read < content.byteLength) { + const result = await handle.read( + content, + read, + content.byteLength - read, + read, + ); + if (result.bytesRead < 1) { + throw new LocalRunAttemptLogRetirementError('integrity_mismatch'); + } + read += result.bytesRead; + } + const after = await handle.stat(); + assertOwnedFile(after, MAXIMUM_FACT_BYTES); + if (!sameFile(before, after) || before.size !== after.size) { + throw new LocalRunAttemptLogRetirementError('integrity_mismatch'); + } + const decoded = new TextDecoder('utf-8', { fatal: true }).decode(content); + return Object.freeze({ + truncation: exactFact(JSON.parse(decoded), expected), + stat: before, + }); + } catch (error) { + if (error instanceof LocalRunAttemptLogRetirementError) throw error; + throw new LocalRunAttemptLogRetirementError('integrity_mismatch', { + cause: error, + }); + } finally { + content.fill(0); + } + } finally { + await handle.close().catch(() => undefined); + } +} + +async function assertPathStillMatches( + filePath: string, + expected: Stats, + maximumBytes: number, +): Promise { + try { + const current = await fs.lstat(filePath); + assertOwnedFile(current, maximumBytes); + if (!sameFile(current, expected)) { + throw new LocalRunAttemptLogRetirementError('integrity_mismatch'); + } + } catch (error) { + if (error instanceof LocalRunAttemptLogRetirementError) throw error; + throw new LocalRunAttemptLogRetirementError('unsafe_path', { + cause: error, + }); + } +} + +async function syncDirectory(directory: string): Promise { + const handle = await fs.open(directory, constants.O_RDONLY); + try { + await handle.sync(); + } finally { + await handle.close().catch(() => undefined); + } +} + +export class LocalRunAttemptLogRetirementStore + implements RunAttemptLogRetirementStore +{ + private readonly root: string; + + constructor(artifactRootPath: string) { + this.root = artifactRoot(artifactRootPath); + } + + async retire( + raw: Readonly, + ): Promise> { + const candidate = normalizeRunAttemptLogRetentionCandidate(raw); + if ( + candidate.executorType !== 'local_process' || + !LOCAL_ARTIFACT_ID.test(candidate.logArtifactId) + ) { + throw new LocalRunAttemptLogRetirementError('integrity_mismatch'); + } + if (!(await optionalOwnedDirectory(this.root))) { + return Object.freeze({ + disposition: 'already_absent' as const, + byteLength: 0, + truncation: Object.freeze({ truncated: 'unknown' as const }), + }); + } + const directory = path.join( + this.root, + candidate.logArtifactId.slice('local-'.length, 'local-'.length + 2), + ); + if (!(await optionalOwnedDirectory(directory))) { + return Object.freeze({ + disposition: 'already_absent' as const, + byteLength: 0, + truncation: Object.freeze({ truncated: 'unknown' as const }), + }); + } + + const target = path.join(directory, `${candidate.logArtifactId}.log`); + const factPath = path.join( + directory, + `.${candidate.logArtifactId}.log.truncated.json`, + ); + const fact = await readFact(factPath, candidate); + const handle = await openPrivateFile(target); + if (!handle) { + if (fact.stat) { + await assertPathStillMatches(factPath, fact.stat, MAXIMUM_FACT_BYTES); + await fs.unlink(factPath); + await syncDirectory(directory); + } + return Object.freeze({ + disposition: 'already_absent' as const, + byteLength: 0, + truncation: fact.truncation, + }); + } + try { + const before = await handle.stat(); + assertOwnedFile(before, MAXIMUM_ARTIFACT_BYTES); + await assertPathStillMatches(target, before, MAXIMUM_ARTIFACT_BYTES); + if (fact.stat) { + await assertPathStillMatches(factPath, fact.stat, MAXIMUM_FACT_BYTES); + } + await fs.unlink(target); + if (fact.stat) await fs.unlink(factPath); + await syncDirectory(directory); + return Object.freeze({ + disposition: 'deleted' as const, + byteLength: before.size, + truncation: fact.truncation, + }); + } finally { + await handle.close().catch(() => undefined); + } + } +} + +export class LocalRunAttemptLogCapacityProbe + implements RunAttemptLogCapacitySource +{ + private readonly root: string; + + constructor(artifactRootPath: string) { + this.root = artifactRoot(artifactRootPath); + } + + async inspect() { + let current = this.root; + while (true) { + try { + const stat = await fs.statfs(current, { bigint: true }); + return Object.freeze({ + availableBytes: stat.bavail * stat.bsize, + totalBytes: stat.blocks * stat.bsize, + }); + } catch (error) { + if (!isCode(error, 'ENOENT')) { + throw new LocalRunAttemptLogRetirementError('unsafe_path', { + cause: error, + }); + } + const parent = path.dirname(current); + if (parent === current) { + throw new LocalRunAttemptLogRetirementError('unsafe_path', { + cause: error, + }); + } + current = parent; + } + } + } +} diff --git a/packages/ql3-local-execution/src/control/lifecycle.ts b/packages/ql3-local-execution/src/control/lifecycle.ts index a3167b84..1f77537d 100644 --- a/packages/ql3-local-execution/src/control/lifecycle.ts +++ b/packages/ql3-local-execution/src/control/lifecycle.ts @@ -1,5 +1,6 @@ import type { LocalCompletionReceiptJournalCursor } from '@qinglong/runtime-core/local-completion-receipt-journal'; import { assertLocalExecutionControlLimit } from '@qinglong/runtime-core/local-execution-control'; +import type { RunAttemptLogRetentionSweepSummary } from '@qinglong/runtime-core/run-attempt-log-retention'; import type { LocalCompletionReceiptCleanupScanner, LocalCompletionReceiptCleanupSummary, @@ -21,6 +22,9 @@ export interface LocalExecutionControlLifecycleOptions { readonly stopTimeoutMs: number; readonly maxDrainPages: number; readonly maxNotifications?: number; + readonly artifactRetention?: Readonly<{ + sweep(): Promise; + }>; readonly clock?: { now(): number }; readonly onDiagnostic?: ( error: unknown, @@ -33,6 +37,7 @@ export interface LocalExecutionControlCycleSummary { readonly completionFailures: number; readonly control: LocalExecutionControlScanSummary; readonly cleanup?: LocalCompletionReceiptCleanupSummary; + readonly artifactRetention?: RunAttemptLogRetentionSweepSummary; } export interface LocalExecutionControlStopSummary { @@ -111,6 +116,12 @@ export class LocalExecutionControlLifecycle { ) { throw new RangeError('Local completion notification budget is invalid'); } + if ( + options.artifactRetention !== undefined && + typeof options.artifactRetention.sweep !== 'function' + ) { + throw new TypeError('Local Artifact retention lifecycle is invalid'); + } this.clock = options.clock ?? { now: Date.now }; } @@ -234,6 +245,7 @@ export class LocalExecutionControlLifecycle { ); } let cleanup: LocalCompletionReceiptCleanupSummary | undefined; + let artifactRetention: RunAttemptLogRetentionSweepSummary | undefined; if ( forceCleanup || this.lastCleanupAtMs === undefined || @@ -246,6 +258,7 @@ export class LocalExecutionControlLifecycle { : { cursor: this.cleanupCursor }), }); this.cleanupCursor = cleanup.truncated ? cleanup.nextCursor : undefined; + artifactRetention = await this.options.artifactRetention?.sweep(); this.lastCleanupAtMs = now; } if (this.pending.size > 0) this.kick(); @@ -254,6 +267,7 @@ export class LocalExecutionControlLifecycle { completionFailures: completion.failed, control, ...(cleanup === undefined ? {} : { cleanup }), + ...(artifactRetention === undefined ? {} : { artifactRetention }), }); } diff --git a/packages/ql3-local-execution/test/control.test.cjs b/packages/ql3-local-execution/test/control.test.cjs index 8d9f59b5..851e5ebd 100644 --- a/packages/ql3-local-execution/test/control.test.cjs +++ b/packages/ql3-local-execution/test/control.test.cjs @@ -367,6 +367,7 @@ test('coalesces completion notifications and owns one idempotent shutdown drain' let scans = 0; let drains = 0; let cleanups = 0; + let retentionSweeps = 0; const lifecycle = new LocalExecutionControlLifecycle( { async process(attemptId) { @@ -420,6 +421,25 @@ test('coalesces completion notifications and owns one idempotent shutdown drain' stopTimeoutMs: 1_000, maxDrainPages: 1, clock: { now: () => 100 }, + artifactRetention: { + async sweep() { + retentionSweeps += 1; + return { + status: 'complete', + pressure: false, + observedAtMs: 100, + retentionMs: 60_000, + availableBytes: '100', + totalBytes: '100', + candidatesScanned: 0, + deletionsAttempted: 0, + recordsWritten: 0, + failedCandidates: 0, + bytesReclaimed: 0, + entries: [], + }; + }, + }, }, ); assert.equal(lifecycle.notifyCompletion(IDS.completionAttempt), true); @@ -428,10 +448,12 @@ test('coalesces completion notifications and owns one idempotent shutdown drain' assert.deepEqual(completions, [IDS.completionAttempt]); assert.equal(scans, 1); assert.equal(cleanups, 1); + assert.equal(retentionSweeps, 1); const first = lifecycle.stopAndDrain(); const second = lifecycle.stopAndDrain(); assert.equal(first, second); assert.equal((await first).status, 'stopped'); assert.equal(drains, 1); assert.equal(cleanups, 2); + assert.equal(retentionSweeps, 1); }); diff --git a/packages/ql3-local-execution/test/localRunAttemptLogRetirementStore.test.cjs b/packages/ql3-local-execution/test/localRunAttemptLogRetirementStore.test.cjs new file mode 100644 index 00000000..14af7c7f --- /dev/null +++ b/packages/ql3-local-execution/test/localRunAttemptLogRetirementStore.test.cjs @@ -0,0 +1,131 @@ +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const { test } = require('node:test'); + +const { + LocalRunAttemptLogCapacityProbe, + LocalRunAttemptLogRetirementError, + LocalRunAttemptLogRetirementStore, +} = require('../dist/artifact-read/localRunAttemptLogRetirementStore.js'); + +const ARTIFACT_ID = `local-${'a'.repeat(30)}`; + +function fixture(t) { + const parent = fs.mkdtempSync(path.join(os.tmpdir(), 'ql3-log-retire-')); + const root = path.join(parent, 'artifacts'); + fs.mkdirSync(root, { mode: 0o700 }); + fs.chmodSync(root, 0o700); + const directory = path.join(root, 'aa'); + fs.mkdirSync(directory, { mode: 0o700 }); + fs.chmodSync(directory, 0o700); + t.after(() => fs.rmSync(parent, { recursive: true, force: true })); + return { + parent, + root, + directory, + target: path.join(directory, `${ARTIFACT_ID}.log`), + fact: path.join(directory, `.${ARTIFACT_ID}.log.truncated.json`), + }; +} + +function candidate() { + return { + projectId: 'prj_default', + runId: 'run_1', + attemptId: 'attempt_1', + logArtifactId: ARTIFACT_ID, + executorType: 'local_process', + finishedAtMs: 1, + }; +} + +function privateFile(filePath, content) { + fs.writeFileSync(filePath, content, { mode: 0o600 }); + fs.chmodSync(filePath, 0o600); +} + +function fact() { + return JSON.stringify({ + schemaVersion: 1, + runId: 'run_1', + attemptId: 'attempt_1', + logArtifactId: ARTIFACT_ID, + maximumBytes: 64 * 1024, + quotaReached: true, + observedAtMs: 2, + }); +} + +test('deletes only the exact private log and truncation fact', async (t) => { + const value = fixture(t); + privateFile(value.target, 'hello'); + privateFile(value.fact, fact()); + privateFile(path.join(value.directory, 'unrelated'), 'keep'); + + const retired = await new LocalRunAttemptLogRetirementStore( + value.root, + ).retire(candidate()); + assert.deepEqual(retired, { + disposition: 'deleted', + byteLength: 5, + truncation: { + truncated: true, + maximumBytes: 64 * 1024, + observedAtMs: 2, + }, + }); + assert.equal(fs.existsSync(value.target), false); + assert.equal(fs.existsSync(value.fact), false); + assert.equal(fs.existsSync(path.join(value.directory, 'unrelated')), true); +}); + +test('converges an unlink-before-tombstone crash and removes its orphan fact', async (t) => { + const value = fixture(t); + privateFile(value.fact, fact()); + const retired = await new LocalRunAttemptLogRetirementStore( + value.root, + ).retire(candidate()); + assert.equal(retired.disposition, 'already_absent'); + assert.equal(retired.byteLength, 0); + assert.equal(retired.truncation.truncated, true); + assert.equal(fs.existsSync(value.fact), false); +}); + +test('fails closed for links, unsafe modes and fact identity drift', async (t) => { + const cases = []; + + const hardLink = fixture(t); + privateFile(hardLink.target, 'hello'); + fs.linkSync(hardLink.target, path.join(hardLink.directory, 'second-link')); + cases.push(hardLink); + + const unsafeMode = fixture(t); + privateFile(unsafeMode.target, 'hello'); + fs.chmodSync(unsafeMode.target, 0o644); + cases.push(unsafeMode); + + const drift = fixture(t); + privateFile(drift.target, 'hello'); + privateFile(drift.fact, fact().replace('"attempt_1"', '"attempt_other"')); + cases.push(drift); + + for (const value of cases) { + await assert.rejects( + new LocalRunAttemptLogRetirementStore(value.root).retire(candidate()), + LocalRunAttemptLogRetirementError, + ); + assert.equal(fs.existsSync(value.target), true); + } +}); + +test('capacity probe uses the nearest existing parent without creating roots', async (t) => { + const value = fixture(t); + const missing = path.join(value.parent, 'future', 'artifacts'); + const snapshot = await new LocalRunAttemptLogCapacityProbe(missing).inspect(); + assert.equal(snapshot.totalBytes > 0n, true); + assert.equal(snapshot.availableBytes >= 0n, true); + assert.equal(snapshot.availableBytes <= snapshot.totalBytes, true); + assert.equal(fs.existsSync(missing), false); +}); diff --git a/packages/ql3-local-owner-cli/test/localDeployment.test.cjs b/packages/ql3-local-owner-cli/test/localDeployment.test.cjs index e078a815..5fd61887 100644 --- a/packages/ql3-local-owner-cli/test/localDeployment.test.cjs +++ b/packages/ql3-local-owner-cli/test/localDeployment.test.cjs @@ -363,9 +363,9 @@ function composeDockerHarness( '/opt/qinglong/node_modules/@qinglong/local-application/dist/cli.js', ], Labels: { - 'io.qinglong.local.sqlite-contract-min': '43', - 'io.qinglong.local.sqlite-contract-max': '43', - 'io.qinglong.local.sqlite-write-contract': '43', + 'io.qinglong.local.sqlite-contract-min': '44', + 'io.qinglong.local.sqlite-contract-max': '44', + 'io.qinglong.local.sqlite-write-contract': '44', 'io.qinglong.local.application-config': '2', 'io.qinglong.local.compose-selection': '1', 'io.qinglong.ai': 'excluded', @@ -557,7 +557,8 @@ test('durably stops one exact legacy Docker owner before publishing commitment', validateSocket() {}, runDocker({ args }) { calls.push(args); - if (args[1] !== 'inspect') return `${command.request.expectedLegacyContainerId}\n`; + if (args[1] !== 'inspect') + return `${command.request.expectedLegacyContainerId}\n`; return JSON.stringify([ { Id: command.request.expectedLegacyContainerId, @@ -604,12 +605,24 @@ test('durably stops one exact legacy Docker owner before publishing commitment', ); const commitmentPath = path.join(journal, '0002-legacy-stopped.json'); assert.equal(mode(journal), 0o700); - assert.equal(mode(path.join(journal, '0001-legacy-stop-requested.json')), 0o600); + assert.equal( + mode(path.join(journal, '0001-legacy-stop-requested.json')), + 0o600, + ); assert.equal(mode(commitmentPath), 0o600); const commitment = JSON.parse(fs.readFileSync(commitmentPath, 'utf8')); - assert.equal(commitment.activationDigest, command.request.expectedActivationDigest); - assert.equal(commitment.controller.legacyContainerId, command.request.expectedLegacyContainerId); - assert.match(commitment.controller.legacySourceBindingDigest, /^[0-9a-f]{64}$/); + assert.equal( + commitment.activationDigest, + command.request.expectedActivationDigest, + ); + assert.equal( + commitment.controller.legacyContainerId, + command.request.expectedLegacyContainerId, + ); + assert.match( + commitment.controller.legacySourceBindingDigest, + /^[0-9a-f]{64}$/, + ); assert.equal(commitment.commitmentDigest, prepared.commitmentDigest); const replayCalls = []; @@ -962,9 +975,9 @@ test('preflights exact local image, Compose merge and SQLite capability', async '/opt/qinglong/node_modules/@qinglong/local-application/dist/cli.js', ], Labels: { - 'io.qinglong.local.sqlite-contract-min': '43', - 'io.qinglong.local.sqlite-contract-max': '43', - 'io.qinglong.local.sqlite-write-contract': '43', + 'io.qinglong.local.sqlite-contract-min': '44', + 'io.qinglong.local.sqlite-contract-max': '44', + 'io.qinglong.local.sqlite-write-contract': '44', 'io.qinglong.local.application-config': '2', 'io.qinglong.local.compose-selection': '1', 'io.qinglong.ai': 'excluded', @@ -1016,7 +1029,7 @@ test('preflights exact local image, Compose merge and SQLite capability', async assert.equal(result.status, 'ready'); assert.equal(result.generation, 1); assert.equal(result.profile, 'edge'); - assert.equal(result.sqlite.contractVersion, 43); + assert.equal(result.sqlite.contractVersion, 44); assert.equal(result.image.architecture, 'arm64'); assert.equal(calls.length, 2); assert.deepEqual(calls[0].slice(0, 2), ['image', 'inspect']); @@ -1116,8 +1129,8 @@ test('applies one Compose generation and exactly replays its health receipt', as assert.equal(mode(receiptPath), 0o600); const receipt = JSON.parse(fs.readFileSync(receiptPath, 'utf8')); assert.deepEqual(receipt.sqlite, { - contractVersion: 43, - writeContractVersion: 43, + contractVersion: 44, + writeContractVersion: 44, writeObservation: 'unchanged', backup: null, }); @@ -1414,8 +1427,8 @@ test('rolls a failed Compose candidate forward to a healthy prior digest', async `${command.request.rolloutId}.sqlite`, ); assert.equal(mode(backupPath), 0o600); - assert.equal(receipt.sqlite.contractVersion, 43); - assert.equal(receipt.sqlite.writeContractVersion, 43); + assert.equal(receipt.sqlite.contractVersion, 44); + assert.equal(receipt.sqlite.writeContractVersion, 44); assert.equal(receipt.sqlite.writeObservation, 'changed'); assert.match(receipt.sqlite.backup.sha256, /^[0-9a-f]{64}$/); assert.equal(receipt.sqlite.backup.bytes > 0, true); diff --git a/packages/ql3-local-owner-cli/test/localReadiness.test.cjs b/packages/ql3-local-owner-cli/test/localReadiness.test.cjs index b9769cde..ad39af01 100644 --- a/packages/ql3-local-owner-cli/test/localReadiness.test.cjs +++ b/packages/ql3-local-owner-cli/test/localReadiness.test.cjs @@ -34,18 +34,15 @@ test('inspects the exact fresh Profile schema without exposing its path', async assert.equal(result.status, 'ready'); assert.equal(result.profile, 'edge'); assert.equal(result.storage.contractName, 'local-control-core'); - assert.equal(result.storage.contractVersion, 43); - assert.equal(result.storage.migrationCount, 86); + assert.equal(result.storage.contractVersion, 44); + assert.equal(result.storage.migrationCount, 88); assert.equal(result.storage.journalMode, 'delete'); assert.equal(JSON.stringify(result).includes(state.directory), false); }); test('CLI is explicit, content-free and rejects a non-private database', async (t) => { const state = await fixture(t, 'standalone'); - const cli = path.resolve( - __dirname, - '../dist/lifecycle/localReadinessCli.js', - ); + const cli = path.resolve(__dirname, '../dist/lifecycle/localReadinessCli.js'); const args = [ cli, `--database=${state.databasePath}`, diff --git a/packages/ql3-local-sqlite/package.json b/packages/ql3-local-sqlite/package.json index 8ac1a366..c2270d38 100644 --- a/packages/ql3-local-sqlite/package.json +++ b/packages/ql3-local-sqlite/package.json @@ -245,6 +245,11 @@ "require": "./dist/task-start/taskStartRepository.js", "default": "./dist/task-start/taskStartRepository.js" }, + "./run-attempt-log-retention": { + "types": "./dist/run/runAttemptLogRetentionRepository.d.ts", + "require": "./dist/run/runAttemptLogRetentionRepository.js", + "default": "./dist/run/runAttemptLogRetentionRepository.js" + }, "./trigger-administration": { "types": "./dist/scheduling/triggerAdministration.d.ts", "require": "./dist/scheduling/triggerAdministration.js", diff --git a/packages/ql3-local-sqlite/src/migration/migration.ts b/packages/ql3-local-sqlite/src/migration/migration.ts index 0a2aabd7..27d56e64 100644 --- a/packages/ql3-local-sqlite/src/migration/migration.ts +++ b/packages/ql3-local-sqlite/src/migration/migration.ts @@ -96,6 +96,8 @@ import { local0083PluginPackageWorkflowTaskAttemptAdmissionsMigration } from '.. import { local0084CapabilityV42Migration } from '../migrations/0084-capability-v42'; import { local0085PluginPackageWorkflowRunListIndexMigration } from '../migrations/0085-plugin-package-workflow-run-list-index'; import { local0086CapabilityV43Migration } from '../migrations/0086-capability-v43'; +import { local0087RunAttemptLogRetentionMigration } from '../migrations/0087-run-attempt-log-retention'; +import { local0088CapabilityV44Migration } from '../migrations/0088-capability-v44'; import type { LocalSqliteMigrationContext } from '../migrations/sqlMigration'; import { LOCAL_SQLITE_MIGRATION_STREAM_ID, @@ -204,6 +206,8 @@ export const localSqliteMigrationDefinition: MigrationStreamDefinition= 0 AND + eligible_at_ms >= finished_at_ms AND + retired_at_ms >= eligible_at_ms + ), + CONSTRAINT ql3_run_log_tombstone_size_check CHECK ( + byte_length BETWEEN 0 AND 1073741824 AND + (disposition <> 'already_absent' OR byte_length = 0) + ), + CONSTRAINT ql3_run_log_tombstone_truncation_shape_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 ( + length(record_digest) = 64 AND record_digest NOT GLOB '*[^0-9a-f]*' + ), + CONSTRAINT ql3_run_log_tombstone_attempt_fk + FOREIGN KEY (attempt_id) REFERENCES "RunAttempts" (id) ON DELETE CASCADE, + CONSTRAINT ql3_run_log_tombstone_run_fk + FOREIGN KEY (run_id) REFERENCES "Runs" (id) ON DELETE CASCADE +) + `, + `CREATE INDEX ql3_run_log_tombstone_retired_idx ON "QingLong3RunAttemptLogArtifactTombstones" (retired_at_ms, attempt_id)`, + `CREATE UNIQUE INDEX ql3_run_log_tombstone_attempt_uidx ON "QingLong3RunAttemptLogArtifactTombstones" (attempt_id)`, + `CREATE INDEX ql3_run_log_retention_candidate_idx ON "RunAttempts" (executor_type, status, finished_at_ms, id) WHERE log_artifact_id IS NOT NULL`, + ` +CREATE TABLE "QingLong3RunAttemptLogRetentionState" ( + maintenance_id TEXT PRIMARY KEY + CONSTRAINT ql3_run_log_retention_state_id_check + CHECK (maintenance_id = 'local-run-attempt-log'), + cursor_finished_at_ms INTEGER, + cursor_attempt_id TEXT, + updated_at_ms INTEGER NOT NULL, + CONSTRAINT ql3_run_log_retention_state_cursor_check CHECK ( + (cursor_finished_at_ms IS NULL AND cursor_attempt_id IS NULL) OR + (cursor_finished_at_ms >= 0 AND length(cursor_attempt_id) BETWEEN 1 AND 128) + ), + CONSTRAINT ql3_run_log_retention_state_time_check CHECK (updated_at_ms >= 0) +) + `, + `INSERT INTO "QingLong3RunAttemptLogRetentionState" (maintenance_id, cursor_finished_at_ms, cursor_attempt_id, updated_at_ms) VALUES ('local-run-attempt-log', NULL, NULL, 0)`, + ], + }); diff --git a/packages/ql3-local-sqlite/src/migrations/0088-capability-v44.ts b/packages/ql3-local-sqlite/src/migrations/0088-capability-v44.ts new file mode 100644 index 00000000..a436cd8d --- /dev/null +++ b/packages/ql3-local-sqlite/src/migrations/0088-capability-v44.ts @@ -0,0 +1,24 @@ +import { CAPABILITIES_V43 } from './0086-capability-v43'; +import { defineLocalSqliteMigration } from './sqlMigration'; + +export const CAPABILITIES_V44 = CAPABILITIES_V43.replace( + '"plugin_package_workflow_run_list":1,', + '"plugin_package_workflow_run_list":1,"run_attempt_log_retention":1,', +); + +export const local0088CapabilityV44Migration = defineLocalSqliteMigration({ + id: '0088-capability-v44', + statements: [ + ` +UPDATE "QingLong3SchemaCapabilities" +SET contract_version = 44, + migration_id = '0087-run-attempt-log-retention', + capabilities = '${CAPABILITIES_V44}', + updated_at_ms = CAST(unixepoch('subsec') * 1000 AS INTEGER) +WHERE contract_name = 'local-control-core' + AND contract_version = 43 + AND migration_id = '0085-plugin-package-workflow-run-list-index' + AND capabilities = '${CAPABILITIES_V43}' + `, + ], +}); diff --git a/packages/ql3-local-sqlite/src/profile/localProfile.ts b/packages/ql3-local-sqlite/src/profile/localProfile.ts index 28c1e628..d35ffb25 100644 --- a/packages/ql3-local-sqlite/src/profile/localProfile.ts +++ b/packages/ql3-local-sqlite/src/profile/localProfile.ts @@ -46,6 +46,7 @@ export type LocalProfileStorageBootstrapResult = readonly dispatch: LocalSqliteRuntimeDatabase['localDispatch']; readonly executionControl: LocalSqliteRuntimeDatabase['executionControl']; readonly completionReceipts: LocalSqliteRuntimeDatabase['completionReceipts']; + readonly runAttemptLogRetention: LocalSqliteRuntimeDatabase['runAttemptLogRetention']; readonly localSecrets: LocalSqliteRuntimeDatabase['localSecrets']; readonly localSecretAdministration: LocalSqliteRuntimeDatabase['localSecretAdministration']; readonly projectPolicy: LocalSqliteRuntimeDatabase['projectPolicy']; @@ -138,6 +139,7 @@ export async function bootstrapLocalProfileStorage( dispatch: database.localDispatch, executionControl: database.executionControl, completionReceipts: database.completionReceipts, + runAttemptLogRetention: database.runAttemptLogRetention, localSecrets: database.localSecrets, localSecretAdministration: database.localSecretAdministration, projectPolicy: database.projectPolicy, diff --git a/packages/ql3-local-sqlite/src/readiness/readiness.ts b/packages/ql3-local-sqlite/src/readiness/readiness.ts index a9d6798b..9e1fdf4e 100644 --- a/packages/ql3-local-sqlite/src/readiness/readiness.ts +++ b/packages/ql3-local-sqlite/src/readiness/readiness.ts @@ -8,7 +8,7 @@ import { } from '../run/stepRunSchemaContract'; export const LOCAL_SQLITE_CONTRACT_NAME = 'local-control-core'; -export const LOCAL_SQLITE_CONTRACT_VERSION = 43; +export const LOCAL_SQLITE_CONTRACT_VERSION = 44; const OPTIONAL_FEATURE_TABLE_NAMES = new Set([ 'QingLong3AiSchemaMigrations', @@ -164,6 +164,7 @@ const REQUIRED_SCHEMA = Object.freeze({ 'ql3_local_attempts_run_status_idx', 'ql3_local_attempts_lease_idx', 'ql3_local_attempts_deadline_idx', + 'ql3_run_log_retention_candidate_idx', ]), }), RunEvents: Object.freeze({ @@ -514,6 +515,37 @@ const REQUIRED_SCHEMA = Object.freeze({ 'ql3_local_receipt_journal_purge_idx', ]), }), + QingLong3RunAttemptLogArtifactTombstones: Object.freeze({ + columns: Object.freeze([ + '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', + ]), + indexes: Object.freeze([ + 'ql3_run_log_tombstone_attempt_uidx', + 'ql3_run_log_tombstone_retired_idx', + ]), + }), + QingLong3RunAttemptLogRetentionState: Object.freeze({ + columns: Object.freeze([ + 'maintenance_id', + 'cursor_finished_at_ms', + 'cursor_attempt_id', + 'updated_at_ms', + ]), + indexes: Object.freeze([]), + }), QingLong3LocalExecutionContextRecipes: Object.freeze({ columns: Object.freeze([ 'context_ref', @@ -2464,11 +2496,10 @@ export async function auditLocalSqliteReadiness( !capability || capability.contract_name !== LOCAL_SQLITE_CONTRACT_NAME || capability.contract_version !== LOCAL_SQLITE_CONTRACT_VERSION || - capability.migration_id !== - '0085-plugin-package-workflow-run-list-index' || + capability.migration_id !== '0087-run-attempt-log-retention' || typeof capability.capabilities !== 'string' || capability.capabilities !== - '{"run_core":1,"run_retry_policy":1,"completion_receipt_journal":1,"local_dispatch_plan":1,"local_secret_envelope":1,"local_project_policy":1,"local_project_administration":1,"local_security_audit":1,"local_security_audit_compaction":1,"local_secret_authorized_mutation":1,"local_identity":1,"local_api_credential":1,"local_identity_provisioning":1,"local_identity_credential_administration":1,"local_owner_bootstrap":1,"local_owner_delivery_acknowledgement":1,"api_credential_pepper_binding":1,"local_owner_pepper_catalog":1,"local_owner_credential_recovery":1,"local_owner_pepper_reference_inspection":1,"local_owner_pepper_material_gc":1,"local_owner_delivery_acknowledgement_gc":1,"task_definition":1,"local_execution_revision_digest":1,"trigger_definition":1,"legacy_adoption_ledger":1,"local_scheduler_admission":1,"plugin_package_install":1,"approved_action":1,"plugin_package_admission":1,"approved_action_execution":1,"plugin_package_proposal":1,"plugin_package_materialized_revision":1,"plugin_package_task_reconciliation":1,"project_tool_definition_snapshot":1,"step_run":1,"tool_execution_evidence":1,"tool_execution_start_barrier":1,"tool_invocation_artifact":1,"tool_execution_artifact_binding":1,"tool_execution_completion":1,"tool_execution_failure_completion":1,"tool_result_key_catalog":1,"tool_result_rekey":1,"plugin_package_quarantine":1,"plugin_package_lifecycle":1,"plugin_package_automation_publication":1,"plugin_package_workflow_admission":1,"plugin_package_workflow_run_list":1,"plugin_package_workflow_task_attempt_admission":1}' || + '{"run_core":1,"run_retry_policy":1,"completion_receipt_journal":1,"local_dispatch_plan":1,"local_secret_envelope":1,"local_project_policy":1,"local_project_administration":1,"local_security_audit":1,"local_security_audit_compaction":1,"local_secret_authorized_mutation":1,"local_identity":1,"local_api_credential":1,"local_identity_provisioning":1,"local_identity_credential_administration":1,"local_owner_bootstrap":1,"local_owner_delivery_acknowledgement":1,"api_credential_pepper_binding":1,"local_owner_pepper_catalog":1,"local_owner_credential_recovery":1,"local_owner_pepper_reference_inspection":1,"local_owner_pepper_material_gc":1,"local_owner_delivery_acknowledgement_gc":1,"task_definition":1,"local_execution_revision_digest":1,"trigger_definition":1,"legacy_adoption_ledger":1,"local_scheduler_admission":1,"plugin_package_install":1,"approved_action":1,"plugin_package_admission":1,"approved_action_execution":1,"plugin_package_proposal":1,"plugin_package_materialized_revision":1,"plugin_package_task_reconciliation":1,"project_tool_definition_snapshot":1,"step_run":1,"tool_execution_evidence":1,"tool_execution_start_barrier":1,"tool_invocation_artifact":1,"tool_execution_artifact_binding":1,"tool_execution_completion":1,"tool_execution_failure_completion":1,"tool_result_key_catalog":1,"tool_result_rekey":1,"plugin_package_quarantine":1,"plugin_package_lifecycle":1,"plugin_package_automation_publication":1,"plugin_package_workflow_admission":1,"plugin_package_workflow_run_list":1,"run_attempt_log_retention":1,"plugin_package_workflow_task_attempt_admission":1}' || typeof capability.updated_at_ms !== 'number' || !Number.isSafeInteger(capability.updated_at_ms) || capability.updated_at_ms < 0 diff --git a/packages/ql3-local-sqlite/src/run/runAttemptLogRetentionRepository.ts b/packages/ql3-local-sqlite/src/run/runAttemptLogRetentionRepository.ts new file mode 100644 index 00000000..3d90f15d --- /dev/null +++ b/packages/ql3-local-sqlite/src/run/runAttemptLogRetentionRepository.ts @@ -0,0 +1,428 @@ +import { + InvalidRunAttemptLogRetentionError, + MAX_RUN_ATTEMPT_LOG_RETENTION_PAGE_SIZE, + RunAttemptLogRetentionUnavailableError, + normalizeRunAttemptLogRetentionCandidate, + normalizeRunAttemptLogRetentionCursor, + normalizeRunAttemptLogRetirementRecord, + type RunAttemptLogRetentionCursor, + type RunAttemptLogRetentionPage, + type RunAttemptLogRetentionRepository, + type RunAttemptLogRetirementRecord, +} from '@qinglong/runtime-core/run-attempt-log-retention'; +import type { RunAttemptLogReadIdentity } from '@qinglong/runtime-core/run-attempt-log-read'; + +import { LocalSqliteOperationAuthority } from '../authority/operationAuthority'; + +type Row = Record; + +const LOCAL_ARTIFACT_ID = /^local-[a-f0-9]{30}$/; +const ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; + +const TOMBSTONE_SELECT = ` + tombstone."log_artifact_id" AS "logArtifactId", + tombstone."project_id" AS "projectId", + tombstone."run_id" AS "runId", + tombstone."attempt_id" AS "attemptId", + tombstone."executor_type" AS "executorType", + tombstone."finished_at_ms" AS "finishedAtMs", + tombstone."eligible_at_ms" AS "eligibleAtMs", + tombstone."retired_at_ms" AS "retiredAtMs", + tombstone."disposition" AS "disposition", + tombstone."byte_length" AS "byteLength", + tombstone."truncated" AS "truncated", + tombstone."maximum_bytes" AS "maximumBytes", + tombstone."truncation_observed_at_ms" AS "truncationObservedAtMs", + tombstone."record_digest" AS "recordDigest" +`; + +function text(row: Row, key: string): string { + const value = row[key]; + if (typeof value !== 'string') { + throw new RunAttemptLogRetentionUnavailableError(); + } + return value; +} + +function integer(row: Row, key: string): number { + const value = row[key]; + if (!Number.isSafeInteger(value) || Number(value) < 0) { + throw new RunAttemptLogRetentionUnavailableError(); + } + return Number(value); +} + +function optionalInteger(row: Row, key: string): number | undefined { + return row[key] === null ? undefined : integer(row, key); +} + +function identity( + value: Readonly, +): Readonly { + if ( + !value || + typeof value !== 'object' || + Array.isArray(value) || + Object.keys(value).sort().join(',') !== + 'attemptId,logArtifactId,projectId,runId' || + !ID_PATTERN.test(value.projectId) || + !ID_PATTERN.test(value.runId) || + !ID_PATTERN.test(value.attemptId) || + !LOCAL_ARTIFACT_ID.test(value.logArtifactId) + ) { + throw new InvalidRunAttemptLogRetentionError('identity is invalid'); + } + return Object.freeze({ ...value }); +} + +function tombstone(row: Row): Readonly { + const truncated = text(row, 'truncated'); + return normalizeRunAttemptLogRetirementRecord({ + schema: 'qinglong/run-attempt-log-retirement@v1', + projectId: text(row, 'projectId'), + runId: text(row, 'runId'), + attemptId: text(row, 'attemptId'), + logArtifactId: text(row, 'logArtifactId'), + executorType: text(row, 'executorType') as 'local_process', + finishedAtMs: integer(row, 'finishedAtMs'), + eligibleAtMs: integer(row, 'eligibleAtMs'), + retiredAtMs: integer(row, 'retiredAtMs'), + disposition: text(row, 'disposition') as 'deleted' | 'already_absent', + byteLength: integer(row, 'byteLength'), + truncation: + truncated === 'unknown' + ? Object.freeze({ truncated: 'unknown' as const }) + : Object.freeze({ + truncated: truncated === 'true', + maximumBytes: optionalInteger(row, 'maximumBytes')!, + observedAtMs: optionalInteger(row, 'truncationObservedAtMs')!, + }), + recordDigest: text(row, 'recordDigest'), + }); +} + +function unavailable(error?: unknown): RunAttemptLogRetentionUnavailableError { + return new RunAttemptLogRetentionUnavailableError( + error === undefined ? undefined : { cause: error }, + ); +} + +export class LocalSqliteRunAttemptLogRetentionRepository + implements RunAttemptLogRetentionRepository +{ + constructor(private readonly authority: LocalSqliteOperationAuthority) { + if (!(authority instanceof LocalSqliteOperationAuthority)) { + throw new TypeError( + 'Local SQLite Run Attempt log retention authority is invalid', + ); + } + } + + inspect(rawIdentity: Readonly) { + const expected = identity(rawIdentity); + return this.authority.enqueue( + async () => { + try { + const row = this.authority.client + .prepare( + `SELECT ${TOMBSTONE_SELECT} + FROM "QingLong3RunAttemptLogArtifactTombstones" AS tombstone + WHERE tombstone."log_artifact_id" = ?`, + ) + .get(expected.logArtifactId) as Row | undefined; + if (!row) return Object.freeze({ status: 'active' as const }); + const record = tombstone(row); + if ( + record.projectId !== expected.projectId || + record.runId !== expected.runId || + record.attemptId !== expected.attemptId || + record.logArtifactId !== expected.logArtifactId + ) { + throw unavailable(); + } + return Object.freeze({ status: 'retired' as const, record }); + } catch (error) { + if (error instanceof RunAttemptLogRetentionUnavailableError) { + throw error; + } + throw unavailable(error); + } + }, + () => unavailable(), + ); + } + + loadCursor(): Promise | undefined> { + return this.authority.enqueue( + async () => { + try { + const row = this.authority.client + .prepare( + `SELECT cursor_finished_at_ms AS "finishedAtMs", + cursor_attempt_id AS "attemptId" + FROM "QingLong3RunAttemptLogRetentionState" + WHERE maintenance_id = 'local-run-attempt-log'`, + ) + .get() as Row | undefined; + if (!row) throw unavailable(); + if (row.finishedAtMs === null && row.attemptId === null) { + return undefined; + } + return normalizeRunAttemptLogRetentionCursor({ + finishedAtMs: integer(row, 'finishedAtMs'), + attemptId: text(row, 'attemptId'), + }); + } catch (error) { + if (error instanceof RunAttemptLogRetentionUnavailableError) { + throw error; + } + throw unavailable(error); + } + }, + () => unavailable(), + ); + } + + list(input: { + readonly cutoffMs: number; + readonly limit: number; + readonly cursor?: Readonly; + }): Promise { + if ( + !input || + typeof input !== 'object' || + Array.isArray(input) || + !Number.isSafeInteger(input.cutoffMs) || + input.cutoffMs < 0 || + !Number.isSafeInteger(input.limit) || + input.limit < 1 || + input.limit > MAX_RUN_ATTEMPT_LOG_RETENTION_PAGE_SIZE + ) { + throw new InvalidRunAttemptLogRetentionError('list input is invalid'); + } + const cursor = + input.cursor === undefined + ? undefined + : normalizeRunAttemptLogRetentionCursor(input.cursor); + return this.authority.enqueue( + async () => { + try { + const rows = this.authority.client + .prepare( + `SELECT run.project_id AS "projectId", + attempt.run_id AS "runId", + attempt.id AS "attemptId", + attempt.log_artifact_id AS "logArtifactId", + attempt.executor_type AS "executorType", + attempt.finished_at_ms AS "finishedAtMs" + FROM "RunAttempts" AS attempt + JOIN "Runs" AS run ON run.id = attempt.run_id + WHERE run.execution_owner = 'runtime' + AND run.status IN ('succeeded','failed','cancelled','timed_out') + AND attempt.status IN ('succeeded','failed','cancelled','timed_out') + AND attempt.executor_type = 'local_process' + AND attempt.finished_at_ms IS NOT NULL + AND run.finished_at_ms IS NOT NULL + AND attempt.finished_at_ms <= ? + AND run.finished_at_ms <= ? + AND length(attempt.log_artifact_id) = 36 + AND substr(attempt.log_artifact_id, 1, 6) = 'local-' + AND substr(attempt.log_artifact_id, 7) NOT GLOB '*[^0-9a-f]*' + AND NOT EXISTS ( + SELECT 1 FROM "LocalCompletionReceiptJournal" AS receipt + WHERE receipt.attempt_id = attempt.id + ) + AND NOT EXISTS ( + SELECT 1 + FROM "QingLong3RunAttemptLogArtifactTombstones" AS tombstone + WHERE tombstone.attempt_id = attempt.id + OR tombstone.log_artifact_id = attempt.log_artifact_id + ) + AND ( + ? IS NULL OR attempt.finished_at_ms > ? OR + (attempt.finished_at_ms = ? AND attempt.id > ?) + ) + ORDER BY attempt.finished_at_ms, attempt.id + LIMIT ?`, + ) + .all( + input.cutoffMs, + input.cutoffMs, + cursor?.finishedAtMs ?? null, + cursor?.finishedAtMs ?? null, + cursor?.finishedAtMs ?? null, + cursor?.attemptId ?? null, + input.limit + 1, + ) as Row[]; + const truncated = rows.length > input.limit; + const candidates = rows.slice(0, input.limit).map((row) => + normalizeRunAttemptLogRetentionCandidate({ + projectId: text(row, 'projectId'), + runId: text(row, 'runId'), + attemptId: text(row, 'attemptId'), + logArtifactId: text(row, 'logArtifactId'), + executorType: text(row, 'executorType') as 'local_process', + finishedAtMs: integer(row, 'finishedAtMs'), + }), + ); + const last = candidates.at(-1); + return Object.freeze({ + candidates: Object.freeze(candidates), + truncated, + ...(truncated && last + ? { + nextCursor: Object.freeze({ + finishedAtMs: last.finishedAtMs, + attemptId: last.attemptId, + }), + } + : {}), + }); + } catch (error) { + throw unavailable(error); + } + }, + () => unavailable(), + ); + } + + record(raw: Readonly) { + const record = normalizeRunAttemptLogRetirementRecord(raw); + if ( + record.executorType !== 'local_process' || + !LOCAL_ARTIFACT_ID.test(record.logArtifactId) + ) { + throw new InvalidRunAttemptLogRetentionError( + 'Local retirement record is invalid', + ); + } + return this.authority.enqueue( + async () => { + const client = this.authority.client; + client.exec('BEGIN IMMEDIATE'); + try { + const replay = client + .prepare( + `SELECT ${TOMBSTONE_SELECT} + FROM "QingLong3RunAttemptLogArtifactTombstones" AS tombstone + WHERE tombstone.log_artifact_id = ? OR tombstone.attempt_id = ?`, + ) + .get(record.logArtifactId, record.attemptId) as Row | undefined; + if (replay) { + const existing = tombstone(replay); + if (existing.recordDigest !== record.recordDigest) { + throw unavailable(); + } + client.exec('COMMIT'); + return 'existing' as const; + } + const eligible = client + .prepare( + `SELECT 1 AS eligible + FROM "RunAttempts" AS attempt + JOIN "Runs" AS run ON run.id = attempt.run_id + WHERE attempt.id = ? + AND attempt.run_id = ? + AND attempt.log_artifact_id = ? + AND attempt.executor_type = 'local_process' + AND attempt.finished_at_ms = ? + AND attempt.status IN ('succeeded','failed','cancelled','timed_out') + AND run.project_id = ? + AND run.execution_owner = 'runtime' + AND run.status IN ('succeeded','failed','cancelled','timed_out') + AND run.finished_at_ms IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM "LocalCompletionReceiptJournal" AS receipt + WHERE receipt.attempt_id = attempt.id + )`, + ) + .get( + record.attemptId, + record.runId, + record.logArtifactId, + record.finishedAtMs, + record.projectId, + ); + if (!eligible) throw unavailable(); + client + .prepare( + `INSERT INTO "QingLong3RunAttemptLogArtifactTombstones" ( + 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + record.logArtifactId, + record.projectId, + record.runId, + record.attemptId, + record.executorType, + record.finishedAtMs, + record.eligibleAtMs, + record.retiredAtMs, + record.disposition, + record.byteLength, + String(record.truncation.truncated), + record.truncation.maximumBytes ?? null, + record.truncation.observedAtMs ?? null, + record.recordDigest, + ); + client.exec('COMMIT'); + return 'recorded' as const; + } catch (error) { + try { + client.exec('ROLLBACK'); + } catch { + // Preserve the retention failure. + } + if (error instanceof RunAttemptLogRetentionUnavailableError) { + throw error; + } + throw unavailable(error); + } + }, + () => unavailable(), + ); + } + + saveCursor( + rawCursor: Readonly | undefined, + updatedAtMs: number, + ): Promise { + const cursor = + rawCursor === undefined + ? undefined + : normalizeRunAttemptLogRetentionCursor(rawCursor); + if (!Number.isSafeInteger(updatedAtMs) || updatedAtMs < 0) { + throw new InvalidRunAttemptLogRetentionError( + 'cursor update time is invalid', + ); + } + return this.authority.enqueue( + async () => { + try { + const result = this.authority.client + .prepare( + `UPDATE "QingLong3RunAttemptLogRetentionState" + SET cursor_finished_at_ms = ?, cursor_attempt_id = ?, updated_at_ms = ? + WHERE maintenance_id = 'local-run-attempt-log'`, + ) + .run( + cursor?.finishedAtMs ?? null, + cursor?.attemptId ?? null, + updatedAtMs, + ); + if (result.changes !== 1) throw unavailable(); + } catch (error) { + if (error instanceof RunAttemptLogRetentionUnavailableError) { + throw error; + } + throw unavailable(error); + } + }, + () => unavailable(), + ); + } +} diff --git a/packages/ql3-local-sqlite/src/runtime/runtimeDatabase.ts b/packages/ql3-local-sqlite/src/runtime/runtimeDatabase.ts index b18b6d81..9b169cbb 100644 --- a/packages/ql3-local-sqlite/src/runtime/runtimeDatabase.ts +++ b/packages/ql3-local-sqlite/src/runtime/runtimeDatabase.ts @@ -62,6 +62,7 @@ import type { LocalSqliteWorkflowTaskExecutionRepository } from '../plugin-packa import type { LocalSqlitePluginPackageWorkflowFrontierRepository } from '../plugin-package/workflow/pluginPackageWorkflowFrontierRepository'; import type { LocalSqlitePluginPackageWorkflowTaskAttemptAdmissionRepository } from '../plugin-package/workflow/pluginPackageWorkflowTaskAttemptAdmissionRepository'; import type { LocalSqlitePluginPackageWorkflowCancellationConvergenceRepository } from '../plugin-package/workflow/pluginPackageWorkflowCancellationConvergenceRepository'; +import { LocalSqliteRunAttemptLogRetentionRepository } from '../run/runAttemptLogRetentionRepository'; export interface LocalSqliteRuntimeDependencies { readonly taskSpecSemanticRegistry?: TaskSpecSemanticRegistry; @@ -97,6 +98,7 @@ export interface LocalSqliteRuntimeDatabase { readonly localDispatch: LocalDispatchStore; readonly executionControl: LocalExecutionControlSource; readonly completionReceipts: LocalCompletionReceiptJournal; + readonly runAttemptLogRetention: LocalSqliteRunAttemptLogRetentionRepository; readonly localSecrets: LocalSecretEnvelopeRepository; readonly localSecretAdministration: LocalSecretAdministrationRepository; readonly projectPolicy: ProjectPolicyRepository; @@ -179,6 +181,8 @@ export async function openLocalSqliteRuntimeDatabase( const schedules = new LocalSqliteScheduleRepository(authority); const apiCredentials = new LocalSqliteApiCredentialRepository(authority); const ownerPepper = new LocalSqliteOwnerPepperRepository(authority); + const runAttemptLogRetention = + new LocalSqliteRunAttemptLogRetentionRepository(authority); let pluginPackageInstallsPromise: | Promise | undefined; @@ -232,6 +236,7 @@ export async function openLocalSqliteRuntimeDatabase( localDispatch: runRuntimeCapabilities.dispatch, executionControl: runRuntimeCapabilities.executionControl, completionReceipts: runRuntimeCapabilities.completionReceipts, + runAttemptLogRetention, localSecrets: securityAuthority, localSecretAdministration: securityAuthority, projectPolicy, diff --git a/packages/ql3-local-sqlite/src/storage/schema.ts b/packages/ql3-local-sqlite/src/storage/schema.ts index cc2af158..42ba7400 100644 --- a/packages/ql3-local-sqlite/src/storage/schema.ts +++ b/packages/ql3-local-sqlite/src/storage/schema.ts @@ -331,6 +331,9 @@ export const runAttempts = sqliteTable( table.deadlineAtMs, table.id, ), + index('ql3_run_log_retention_candidate_idx') + .on(table.executorType, table.status, table.finishedAtMs, table.id) + .where(sql`${table.logArtifactId} is not null`), ], ); @@ -511,6 +514,93 @@ export const localCompletionReceiptJournal = sqliteTable( ], ); +export const runAttemptLogArtifactTombstones = sqliteTable( + 'QingLong3RunAttemptLogArtifactTombstones', + { + logArtifactId: text('log_artifact_id').primaryKey(), + projectId: text('project_id').notNull(), + runId: text('run_id') + .notNull() + .references(() => runs.id, { onDelete: 'cascade' }), + attemptId: text('attempt_id') + .notNull() + .references(() => runAttempts.id, { onDelete: 'cascade' }), + executorType: text('executor_type').notNull(), + finishedAtMs: integer('finished_at_ms').notNull(), + eligibleAtMs: integer('eligible_at_ms').notNull(), + retiredAtMs: integer('retired_at_ms').notNull(), + disposition: text('disposition').notNull(), + byteLength: integer('byte_length').notNull(), + truncated: text('truncated').notNull(), + maximumBytes: integer('maximum_bytes'), + truncationObservedAtMs: integer('truncation_observed_at_ms'), + recordDigest: text('record_digest').notNull(), + }, + (table) => [ + uniqueIndex('ql3_run_log_tombstone_attempt_uidx').on(table.attemptId), + index('ql3_run_log_tombstone_retired_idx').on( + table.retiredAtMs, + table.attemptId, + ), + check( + 'ql3_run_log_tombstone_executor_check', + sql`${table.executorType} = 'local_process'`, + ), + check( + 'ql3_run_log_tombstone_disposition_check', + sql`${table.disposition} in ('deleted','already_absent')`, + ), + check( + 'ql3_run_log_tombstone_truncated_check', + sql`${table.truncated} in ('true','false','unknown')`, + ), + check( + 'ql3_run_log_tombstone_identity_check', + sql`length(${table.projectId}) between 1 and 128 and length(${table.runId}) between 1 and 128 and length(${table.attemptId}) between 1 and 128 and length(${table.logArtifactId}) = 36 and substr(${table.logArtifactId}, 1, 6) = 'local-' and substr(${table.logArtifactId}, 7) not glob '*[^0-9a-f]*'`, + ), + 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_size_check', + sql`${table.byteLength} between 0 and 1073741824 and (${table.disposition} <> 'already_absent' or ${table.byteLength} = 0)`, + ), + check( + 'ql3_run_log_tombstone_truncation_shape_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`length(${table.recordDigest}) = 64 and ${table.recordDigest} not glob '*[^0-9a-f]*'`, + ), + ], +); + +export const runAttemptLogRetentionState = sqliteTable( + 'QingLong3RunAttemptLogRetentionState', + { + maintenanceId: text('maintenance_id').primaryKey(), + cursorFinishedAtMs: integer('cursor_finished_at_ms'), + cursorAttemptId: text('cursor_attempt_id'), + updatedAtMs: integer('updated_at_ms').notNull(), + }, + (table) => [ + check( + 'ql3_run_log_retention_state_id_check', + sql`${table.maintenanceId} = 'local-run-attempt-log'`, + ), + check( + 'ql3_run_log_retention_state_cursor_check', + sql`(${table.cursorFinishedAtMs} is null and ${table.cursorAttemptId} is null) or (${table.cursorFinishedAtMs} >= 0 and length(${table.cursorAttemptId}) between 1 and 128)`, + ), + check( + 'ql3_run_log_retention_state_time_check', + sql`${table.updatedAtMs} >= 0`, + ), + ], +); + export const localExecutionContextRecipes = sqliteTable( 'QingLong3LocalExecutionContextRecipes', { @@ -4702,16 +4792,12 @@ export const pluginPackageWorkflowTaskAttemptAdmissions = sqliteTable( .onDelete('restrict') .onUpdate('restrict'), foreignKey({ - columns: [ - table.generationDigest, - table.taskReconciliationReceiptDigest, - ], + columns: [table.generationDigest, table.taskReconciliationReceiptDigest], foreignColumns: [ pluginPackageTaskReconciliations.generationDigest, pluginPackageTaskReconciliations.receiptDigest, ], - name: - 'ql3_plugin_package_workflow_task_attempt_admission_reconciliation_fk', + name: 'ql3_plugin_package_workflow_task_attempt_admission_reconciliation_fk', }) .onDelete('restrict') .onUpdate('restrict'), @@ -4770,6 +4856,8 @@ export const localSqliteSchema = Object.freeze({ stepRunMutations, runRetryPolicies, localCompletionReceiptJournal, + runAttemptLogArtifactTombstones, + runAttemptLogRetentionState, localExecutionContextRecipes, localTaskExecutionRevisions, localSecretEnvelopes, diff --git a/packages/ql3-local-sqlite/test/database.test.cjs b/packages/ql3-local-sqlite/test/database.test.cjs index 8c04ca10..5f86d712 100644 --- a/packages/ql3-local-sqlite/test/database.test.cjs +++ b/packages/ql3-local-sqlite/test/database.test.cjs @@ -136,9 +136,11 @@ test('creates a reviewed edge database and opens runtime only after readiness', '0084-capability-v42', '0085-plugin-package-workflow-run-list-index', '0086-capability-v43', + '0087-run-attempt-log-retention', + '0088-capability-v44', ]); assert.equal(migrated.readiness.contractName, 'local-control-core'); - assert.equal(migrated.readiness.contractVersion, 43); + assert.equal(migrated.readiness.contractVersion, 44); assert.equal(migrated.readiness.journalMode, 'delete'); assert.equal(fs.statSync(databasePath).mode & 0o777, 0o600); @@ -501,8 +503,8 @@ test('backfills v14 execution revisions with a verified independent digest', asy .get(), }, { - contract_version: 43, - migration_id: '0085-plugin-package-workflow-run-list-index', + contract_version: 44, + migration_id: '0087-run-attempt-log-retention', }, ); } finally { @@ -689,19 +691,19 @@ test('excludes reviewed optional feature tables while preserving unknown table d const options = { databasePath, profile: 'edge' }; await migrateLocalSqlitePath(options); const client = new DatabaseSync(databasePath); - assert.equal((await auditLocalSqlitePath(options)).tableCount, 76); + assert.equal((await auditLocalSqlitePath(options)).tableCount, 78); client.exec( 'CREATE TABLE "ModelInvocationFeatureHead" (feature_id TEXT PRIMARY KEY)', ); client.close(); - assert.equal((await auditLocalSqlitePath(options)).tableCount, 76); + assert.equal((await auditLocalSqlitePath(options)).tableCount, 78); const unknownClient = new DatabaseSync(databasePath); unknownClient.exec('CREATE TABLE "UserExtensionData" (id TEXT PRIMARY KEY)'); unknownClient.close(); - assert.equal((await auditLocalSqlitePath(options)).tableCount, 77); + assert.equal((await auditLocalSqlitePath(options)).tableCount, 79); const triggerClient = new DatabaseSync(databasePath); triggerClient.exec(` diff --git a/packages/ql3-local-sqlite/test/pluginPackageWorkflowAdmissionRepository.test.cjs b/packages/ql3-local-sqlite/test/pluginPackageWorkflowAdmissionRepository.test.cjs index 18e006f8..ff97e0c4 100644 --- a/packages/ql3-local-sqlite/test/pluginPackageWorkflowAdmissionRepository.test.cjs +++ b/packages/ql3-local-sqlite/test/pluginPackageWorkflowAdmissionRepository.test.cjs @@ -34,7 +34,9 @@ const { const { LocalSqlitePluginPackageWorkflowAdmissionRepository, } = require('../dist/plugin-package/workflow/pluginPackageWorkflowAdmissionRepository'); -const { LocalSqliteStepRunRepository } = require('../dist/run/stepRunRepository'); +const { + LocalSqliteStepRunRepository, +} = require('../dist/run/stepRunRepository'); const { auditLocalSqliteReadiness } = require('../dist/readiness/readiness'); function fixture(namespace) { @@ -154,7 +156,7 @@ test('atomically admits one generation-bound Workflow Run and exactly replays it }, { runs: 1, steps: 2, events: 3, mutations: 2, admissions: 1 }, ); - assert.equal((await auditLocalSqliteReadiness(client)).contractVersion, 43); + assert.equal((await auditLocalSqliteReadiness(client)).contractVersion, 44); }); test('runs an optional authorization guard inside new and replay transactions', async (t) => { @@ -286,7 +288,7 @@ test('exactly replays immutable admission after the Workflow StepRun advances', }, { status: 'running', version: 5, eventSequence: 5 }, ); - assert.equal((await auditLocalSqliteReadiness(client)).contractVersion, 43); + assert.equal((await auditLocalSqliteReadiness(client)).contractVersion, 44); }); test('fails closed before writing when the exact installation is not active', async (t) => { diff --git a/packages/ql3-local-sqlite/test/pluginPackageWorkflowTaskAttemptAdmissionRepository.test.cjs b/packages/ql3-local-sqlite/test/pluginPackageWorkflowTaskAttemptAdmissionRepository.test.cjs index 0eb0a5bb..d6a676c5 100644 --- a/packages/ql3-local-sqlite/test/pluginPackageWorkflowTaskAttemptAdmissionRepository.test.cjs +++ b/packages/ql3-local-sqlite/test/pluginPackageWorkflowTaskAttemptAdmissionRepository.test.cjs @@ -231,7 +231,7 @@ test('atomically admits the exact reconciled local Task revision and replays it' stepAttemptCount: 0, }, ); - assert.equal((await auditLocalSqliteReadiness(client)).contractVersion, 43); + assert.equal((await auditLocalSqliteReadiness(client)).contractVersion, 44); }); test('bounds candidate paging before SQL and fences cancellation', async (t) => { diff --git a/packages/ql3-local-sqlite/test/rolloutSafety.test.cjs b/packages/ql3-local-sqlite/test/rolloutSafety.test.cjs index 577e1b12..248c3f6c 100644 --- a/packages/ql3-local-sqlite/test/rolloutSafety.test.cjs +++ b/packages/ql3-local-sqlite/test/rolloutSafety.test.cjs @@ -40,9 +40,9 @@ test('creates and exactly replays a reviewed rollout backup', async (t) => { await migrateLocalSqlitePath(state); const prepared = await createLocalSqliteRolloutBackup(state); assert.equal(prepared.status, 'prepared'); - assert.equal(prepared.contractVersion, 43); - assert.equal(prepared.writeContractVersion, 43); - assert.equal(LOCAL_SQLITE_WRITE_CONTRACT_VERSION, 43); + assert.equal(prepared.contractVersion, 44); + assert.equal(prepared.writeContractVersion, 44); + assert.equal(LOCAL_SQLITE_WRITE_CONTRACT_VERSION, 44); assert.match(prepared.sha256, /^[0-9a-f]{64}$/); assert.equal(prepared.bytes > 0, true); assert.equal(prepared.pageCount > 0, true); diff --git a/packages/ql3-local-sqlite/test/runAttemptLogRetentionRepository.test.cjs b/packages/ql3-local-sqlite/test/runAttemptLogRetentionRepository.test.cjs new file mode 100644 index 00000000..914b285f --- /dev/null +++ b/packages/ql3-local-sqlite/test/runAttemptLogRetentionRepository.test.cjs @@ -0,0 +1,194 @@ +const assert = require('node:assert/strict'); +const { DatabaseSync } = require('node:sqlite'); +const { test } = require('node:test'); + +const { + createRunAttemptLogRetirementRecord, + RunAttemptLogRetentionUnavailableError, +} = require('@qinglong/runtime-core/run-attempt-log-retention'); +const { + LocalSqliteOperationAuthority, +} = require('../dist/authority/operationAuthority.js'); +const { + migrateLocalSqliteDatabase, +} = require('../dist/migration/migration.js'); +const { + LocalSqliteRunAttemptLogRetentionRepository, +} = require('../dist/run/runAttemptLogRetentionRepository.js'); + +async function fixture() { + const client = new DatabaseSync(':memory:'); + client.exec('PRAGMA foreign_keys = ON'); + await migrateLocalSqliteDatabase(client); + const authority = new LocalSqliteOperationAuthority(client); + return { + client, + authority, + repository: new LocalSqliteRunAttemptLogRetentionRepository(authority), + }; +} + +function seed(client, index, overrides = {}) { + const runId = `run_${index}`; + const attemptId = `attempt_${index}`; + const artifactId = `local-${index.toString(16).padStart(30, '0')}`; + client + .prepare( + `INSERT INTO "Runs" ( + id, project_id, task_id, task_revision, trigger_type, + execution_origin, execution_owner, status, version, event_sequence, + priority, created_at_ms, finished_at_ms + ) VALUES (?, 'prj_default', 'task_1', 'revision_1', 'task_start', + 'manual', 'runtime', ?, 1, 1, 0, 1, ?)`, + ) + .run( + runId, + overrides.runStatus ?? 'succeeded', + overrides.finishedAtMs ?? index, + ); + client + .prepare( + `INSERT INTO "RunAttempts" ( + id, run_id, attempt, status, executor_type, log_artifact_id, + callback_sequence, created_at_ms, finished_at_ms + ) VALUES (?, ?, 1, ?, ?, ?, 0, 1, ?)`, + ) + .run( + attemptId, + runId, + overrides.attemptStatus ?? 'succeeded', + overrides.executorType ?? 'local_process', + artifactId, + overrides.finishedAtMs ?? index, + ); + return { runId, attemptId, artifactId }; +} + +test('lists only safe terminal Local candidates with a durable cursor', async () => { + const { client, authority, repository } = await fixture(); + try { + const one = seed(client, 1); + const two = seed(client, 2, { attemptStatus: 'lost' }); + const three = seed(client, 3); + client + .prepare( + `INSERT INTO "LocalCompletionReceiptJournal" ( + attempt_id, run_id, state, registered_at_ms, updated_at_ms + ) VALUES (?, ?, 'pending', 1, 1)`, + ) + .run(three.attemptId, three.runId); + + const page = await repository.list({ cutoffMs: 100, limit: 1 }); + assert.deepEqual(page.candidates, [ + { + projectId: 'prj_default', + runId: one.runId, + attemptId: one.attemptId, + logArtifactId: one.artifactId, + executorType: 'local_process', + finishedAtMs: 1, + }, + ]); + assert.equal(page.truncated, false); + assert.equal(two.attemptId, 'attempt_2'); + + await repository.saveCursor( + { finishedAtMs: 1, attemptId: one.attemptId }, + 101, + ); + assert.deepEqual(await repository.loadCursor(), { + finishedAtMs: 1, + attemptId: one.attemptId, + }); + await repository.saveCursor(undefined, 102); + assert.equal(await repository.loadCursor(), undefined); + } finally { + await authority.close(); + } +}); + +test('records exact tombstones idempotently and exposes retired state', async () => { + const { client, authority, repository } = await fixture(); + try { + const value = seed(client, 1); + const record = createRunAttemptLogRetirementRecord({ + projectId: 'prj_default', + runId: value.runId, + attemptId: value.attemptId, + logArtifactId: value.artifactId, + executorType: 'local_process', + finishedAtMs: 1, + eligibleAtMs: 2, + retiredAtMs: 3, + disposition: 'deleted', + byteLength: 7, + truncation: { truncated: false, maximumBytes: 1024, observedAtMs: 1 }, + }); + assert.equal(await repository.record(record), 'recorded'); + assert.equal(await repository.record(record), 'existing'); + assert.deepEqual( + await repository.inspect({ + projectId: 'prj_default', + runId: value.runId, + attemptId: value.attemptId, + logArtifactId: value.artifactId, + }), + { status: 'retired', record }, + ); + assert.deepEqual(await repository.list({ cutoffMs: 100, limit: 2 }), { + candidates: [], + truncated: false, + }); + + client + .prepare( + `UPDATE "QingLong3RunAttemptLogArtifactTombstones" + SET record_digest = ? WHERE attempt_id = ?`, + ) + .run('0'.repeat(64), value.attemptId); + await assert.rejects( + repository.inspect({ + projectId: 'prj_default', + runId: value.runId, + attemptId: value.attemptId, + logArtifactId: value.artifactId, + }), + RunAttemptLogRetentionUnavailableError, + ); + } finally { + await authority.close(); + } +}); + +test('refuses to tombstone an attempt while a completion receipt exists', async () => { + const { client, authority, repository } = await fixture(); + try { + const value = seed(client, 1); + client + .prepare( + `INSERT INTO "LocalCompletionReceiptJournal" ( + attempt_id, run_id, state, registered_at_ms, updated_at_ms + ) VALUES (?, ?, 'pending', 1, 1)`, + ) + .run(value.attemptId, value.runId); + const record = createRunAttemptLogRetirementRecord({ + projectId: 'prj_default', + runId: value.runId, + attemptId: value.attemptId, + logArtifactId: value.artifactId, + executorType: 'local_process', + finishedAtMs: 1, + eligibleAtMs: 2, + retiredAtMs: 3, + disposition: 'already_absent', + byteLength: 0, + truncation: { truncated: 'unknown' }, + }); + await assert.rejects( + repository.record(record), + RunAttemptLogRetentionUnavailableError, + ); + } finally { + await authority.close(); + } +}); diff --git a/packages/ql3-runtime-core/package.json b/packages/ql3-runtime-core/package.json index f5cf5772..31dd522d 100644 --- a/packages/ql3-runtime-core/package.json +++ b/packages/ql3-runtime-core/package.json @@ -235,6 +235,9 @@ ], "run-attempt-log-read": [ "dist/run/log-read/runAttemptLogRead.d.ts" + ], + "run-attempt-log-retention": [ + "dist/run/log-retention/runAttemptLogRetention.d.ts" ] } }, @@ -289,6 +292,11 @@ "require": "./dist/run/log-read/runAttemptLogRead.js", "default": "./dist/run/log-read/runAttemptLogRead.js" }, + "./run-attempt-log-retention": { + "types": "./dist/run/log-retention/runAttemptLogRetention.d.ts", + "require": "./dist/run/log-retention/runAttemptLogRetention.js", + "default": "./dist/run/log-retention/runAttemptLogRetention.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-read/runAttemptLogRead.ts b/packages/ql3-runtime-core/src/run/log-read/runAttemptLogRead.ts index 11bef44c..c873044f 100644 --- a/packages/ql3-runtime-core/src/run/log-read/runAttemptLogRead.ts +++ b/packages/ql3-runtime-core/src/run/log-read/runAttemptLogRead.ts @@ -1,5 +1,10 @@ import { RUN_ATTEMPT_STATUSES, type RunAttemptStatus } from '../run'; import type { RunRepositoryReader } from '../runRepository'; +import { + normalizeRunAttemptLogRetirementRecord, + type RunAttemptLogRetentionStateReader, + type RunAttemptLogRetirementRecord, +} from '../log-retention/runAttemptLogRetention'; export const MAX_RUN_ATTEMPT_LOG_READ_BYTES = 256 * 1024; @@ -69,6 +74,13 @@ export type RunAttemptLogReadResult = }> | (Readonly & Readonly<{ readonly status: 'missing' }>) + | (Readonly & + Readonly<{ + readonly status: 'retired'; + readonly retiredAtMs: number; + readonly byteLength: number; + readonly truncation: Readonly; + }>) | (Readonly & Extract); @@ -244,13 +256,15 @@ export class RunAttemptLogReadService { >, private readonly reader: RunAttemptLogRangeReader, options: RunAttemptLogReadServiceOptions, + private readonly retention?: RunAttemptLogRetentionStateReader, ) { if ( !runs || typeof runs.findRunById !== 'function' || typeof runs.findAttemptById !== 'function' || !reader || - typeof reader.read !== 'function' + typeof reader.read !== 'function' || + (retention !== undefined && typeof retention.inspect !== 'function') ) { throw new InvalidRunAttemptLogReadError('dependencies are invalid'); } @@ -319,8 +333,12 @@ export class RunAttemptLogReadService { attemptId, logArtifactId: attempt.logArtifactId, }); + const retiredBeforeRead = await this.retired(identity); + if (retiredBeforeRead) return retiredBeforeRead; const result = await this.reader.read(identity, range, request.signal); if (result.status === 'missing') { + const retiredAfterMissing = await this.retired(identity); + if (retiredAfterMissing) return retiredAfterMissing; if ( this.options.activeMissingIsPending === true && !TERMINAL_ATTEMPT_STATUSES.has(attempt.status) @@ -340,4 +358,39 @@ export class RunAttemptLogReadService { throw new RunAttemptLogReadUnavailableError({ cause: error }); } } + + private async retired(identity: Readonly): Promise< + | (Readonly & + Readonly<{ + readonly status: 'retired'; + readonly retiredAtMs: number; + readonly byteLength: number; + readonly truncation: Readonly; + }>) + | undefined + > { + if (!this.retention) return undefined; + const state = await this.retention.inspect(identity); + if (!state || (state.status !== 'active' && state.status !== 'retired')) { + throw new RunAttemptLogReadUnavailableError(); + } + if (state.status === 'active') return undefined; + const record: Readonly = + normalizeRunAttemptLogRetirementRecord(state.record); + if ( + record.projectId !== identity.projectId || + record.runId !== identity.runId || + record.attemptId !== identity.attemptId || + record.logArtifactId !== identity.logArtifactId + ) { + throw new RunAttemptLogReadUnavailableError(); + } + return Object.freeze({ + status: 'retired' as const, + ...identity, + retiredAtMs: record.retiredAtMs, + byteLength: record.byteLength, + truncation: record.truncation, + }); + } } diff --git a/packages/ql3-runtime-core/src/run/log-retention/runAttemptLogRetention.ts b/packages/ql3-runtime-core/src/run/log-retention/runAttemptLogRetention.ts new file mode 100644 index 00000000..f566f16a --- /dev/null +++ b/packages/ql3-runtime-core/src/run/log-retention/runAttemptLogRetention.ts @@ -0,0 +1,600 @@ +import { createHash } from 'node:crypto'; + +import type { + RunAttemptLogReadIdentity, + RunAttemptLogTruncationView, +} from '../log-read/runAttemptLogRead'; + +export const MAX_RUN_ATTEMPT_LOG_RETENTION_PAGE_SIZE = 64; +export const MAX_RUN_ATTEMPT_LOG_RETENTION_DELETIONS = 16; +export const MIN_RUN_ATTEMPT_LOG_RETENTION_MS = 60_000; +export const MAX_RUN_ATTEMPT_LOG_RETENTION_MS = 365 * 24 * 60 * 60_000; + +const ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; +const DIGEST_PATTERN = /^[a-f0-9]{64}$/; + +export type RunAttemptLogRetirementDisposition = 'deleted' | 'already_absent'; + +export interface RunAttemptLogRetentionCursor { + readonly finishedAtMs: number; + readonly attemptId: string; +} + +export interface RunAttemptLogRetentionCandidate + extends RunAttemptLogReadIdentity, + RunAttemptLogRetentionCursor { + readonly executorType: 'local_process' | 'remote_worker'; +} + +export interface RunAttemptLogRetirementRecord + extends RunAttemptLogRetentionCandidate { + readonly schema: 'qinglong/run-attempt-log-retirement@v1'; + readonly eligibleAtMs: number; + readonly retiredAtMs: number; + readonly disposition: RunAttemptLogRetirementDisposition; + readonly byteLength: number; + readonly truncation: Readonly; + readonly recordDigest: string; +} + +export type RunAttemptLogRetentionState = + | Readonly<{ readonly status: 'active' }> + | Readonly<{ + readonly status: 'retired'; + readonly record: Readonly; + }>; + +export interface RunAttemptLogRetentionStateReader { + inspect( + identity: Readonly, + ): Promise; +} + +export interface RunAttemptLogRetentionPage { + readonly candidates: readonly RunAttemptLogRetentionCandidate[]; + readonly truncated: boolean; + readonly nextCursor?: Readonly; +} + +export interface RunAttemptLogRetentionRepository + extends RunAttemptLogRetentionStateReader { + loadCursor(): Promise | undefined>; + list(input: { + readonly cutoffMs: number; + readonly limit: number; + readonly cursor?: Readonly; + }): Promise; + record( + record: Readonly, + ): Promise<'recorded' | 'existing'>; + saveCursor( + cursor: Readonly | undefined, + updatedAtMs: number, + ): Promise; +} + +export interface RunAttemptLogRetirementStoreResult { + readonly disposition: RunAttemptLogRetirementDisposition; + readonly byteLength: number; + readonly truncation: Readonly; +} + +export interface RunAttemptLogRetirementStore { + retire( + candidate: Readonly, + ): Promise>; +} + +export interface RunAttemptLogCapacitySnapshot { + readonly availableBytes: bigint; + readonly totalBytes: bigint; +} + +export interface RunAttemptLogCapacitySource { + inspect(): Promise>; +} + +export interface RunAttemptLogRetentionServiceOptions { + readonly normalRetentionMs: number; + readonly pressureRetentionMs: number; + readonly minimumFreeBytes: number; + readonly pageSize: number; + readonly maximumDeletions: number; + readonly clock?: { now(): number }; +} + +export interface RunAttemptLogRetentionEntry { + readonly attemptId: string; + readonly logArtifactId: string; + readonly outcome: + | RunAttemptLogRetirementDisposition + | 'file_failed' + | 'record_failed'; + readonly byteLength: number; +} + +export interface RunAttemptLogRetentionSweepSummary { + readonly status: 'complete' | 'page_complete' | 'deletion_budget_exhausted'; + readonly pressure: boolean; + readonly observedAtMs: number; + readonly retentionMs: number; + readonly availableBytes: string; + readonly totalBytes: string; + readonly candidatesScanned: number; + readonly deletionsAttempted: number; + readonly recordsWritten: number; + readonly failedCandidates: number; + readonly bytesReclaimed: number; + readonly entries: readonly RunAttemptLogRetentionEntry[]; + readonly nextCursor?: Readonly; +} + +export class InvalidRunAttemptLogRetentionError extends TypeError { + constructor(message: string) { + super(`Run Attempt log retention is invalid: ${message}`); + this.name = 'InvalidRunAttemptLogRetentionError'; + } +} + +export class RunAttemptLogRetentionUnavailableError extends Error { + constructor(options?: ErrorOptions) { + super('Run Attempt log retention is unavailable', options); + this.name = 'RunAttemptLogRetentionUnavailableError'; + } +} + +function exactKeys( + value: object, + required: readonly string[], + optional: readonly string[], + name: string, +): void { + const keys = Object.keys(value); + const allowed = new Set([...required, ...optional]); + if ( + required.some((key) => !Object.hasOwn(value, key)) || + keys.some((key) => !allowed.has(key)) + ) { + throw new InvalidRunAttemptLogRetentionError(`${name} shape is invalid`); + } +} + +function id(name: string, value: unknown): string { + if (typeof value !== 'string' || !ID_PATTERN.test(value)) { + throw new InvalidRunAttemptLogRetentionError(`${name} is invalid`); + } + return value; +} + +function timestamp(name: string, value: unknown): number { + if (!Number.isSafeInteger(value) || Number(value) < 0) { + throw new InvalidRunAttemptLogRetentionError(`${name} is invalid`); + } + return Number(value); +} + +function nonNegativeInteger(name: string, value: unknown): number { + return timestamp(name, value); +} + +function normalizedTruncation( + value: Readonly, +): Readonly { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new InvalidRunAttemptLogRetentionError('truncation is invalid'); + } + exactKeys( + value, + ['truncated'], + ['maximumBytes', 'observedAtMs'], + 'truncation', + ); + if ( + (value.truncated !== true && + value.truncated !== false && + value.truncated !== 'unknown') || + (value.maximumBytes !== undefined && + (!Number.isSafeInteger(value.maximumBytes) || value.maximumBytes < 1)) || + (value.observedAtMs !== undefined && + (!Number.isSafeInteger(value.observedAtMs) || value.observedAtMs < 0)) || + (value.truncated === 'unknown' && + (value.maximumBytes !== undefined || value.observedAtMs !== undefined)) + ) { + throw new InvalidRunAttemptLogRetentionError('truncation is invalid'); + } + return Object.freeze({ ...value }); +} + +export function normalizeRunAttemptLogRetentionCursor( + value: Readonly, +): Readonly { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new InvalidRunAttemptLogRetentionError('cursor is invalid'); + } + exactKeys(value, ['attemptId', 'finishedAtMs'], [], 'cursor'); + return Object.freeze({ + finishedAtMs: timestamp('finishedAtMs', value.finishedAtMs), + attemptId: id('attemptId', value.attemptId), + }); +} + +export function normalizeRunAttemptLogRetentionCandidate( + value: Readonly, +): Readonly { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new InvalidRunAttemptLogRetentionError('candidate is invalid'); + } + exactKeys( + value, + [ + 'attemptId', + 'executorType', + 'finishedAtMs', + 'logArtifactId', + 'projectId', + 'runId', + ], + [], + 'candidate', + ); + if ( + value.executorType !== 'local_process' && + value.executorType !== 'remote_worker' + ) { + throw new InvalidRunAttemptLogRetentionError('executorType is invalid'); + } + return Object.freeze({ + projectId: id('projectId', value.projectId), + runId: id('runId', value.runId), + attemptId: id('attemptId', value.attemptId), + logArtifactId: id('logArtifactId', value.logArtifactId), + executorType: value.executorType, + finishedAtMs: timestamp('finishedAtMs', value.finishedAtMs), + }); +} + +function recordPayload( + value: Omit, +): string { + return JSON.stringify([ + value.schema, + value.projectId, + value.runId, + value.attemptId, + value.logArtifactId, + value.executorType, + value.finishedAtMs, + value.eligibleAtMs, + value.retiredAtMs, + value.disposition, + value.byteLength, + value.truncation.truncated, + value.truncation.maximumBytes ?? null, + value.truncation.observedAtMs ?? null, + ]); +} + +export function digestRunAttemptLogRetirementRecord( + value: Omit, +): string { + return createHash('sha256') + .update('qinglong/run-attempt-log-retirement@v1\0', 'utf8') + .update(recordPayload(value), 'utf8') + .digest('hex'); +} + +export function createRunAttemptLogRetirementRecord( + input: Omit, +): Readonly { + const value = { + schema: 'qinglong/run-attempt-log-retirement@v1' as const, + ...input, + }; + return normalizeRunAttemptLogRetirementRecord({ + ...value, + recordDigest: digestRunAttemptLogRetirementRecord(value), + }); +} + +export function normalizeRunAttemptLogRetirementRecord( + value: Readonly, +): Readonly { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new InvalidRunAttemptLogRetentionError('record is invalid'); + } + exactKeys( + value, + [ + 'attemptId', + 'byteLength', + 'disposition', + 'eligibleAtMs', + 'executorType', + 'finishedAtMs', + 'logArtifactId', + 'projectId', + 'recordDigest', + 'retiredAtMs', + 'runId', + 'schema', + 'truncation', + ], + [], + 'record', + ); + const candidate = normalizeRunAttemptLogRetentionCandidate({ + projectId: value.projectId, + runId: value.runId, + attemptId: value.attemptId, + logArtifactId: value.logArtifactId, + executorType: value.executorType, + finishedAtMs: value.finishedAtMs, + }); + const record = Object.freeze({ + schema: value.schema, + ...candidate, + eligibleAtMs: timestamp('eligibleAtMs', value.eligibleAtMs), + retiredAtMs: timestamp('retiredAtMs', value.retiredAtMs), + disposition: value.disposition, + byteLength: nonNegativeInteger('byteLength', value.byteLength), + truncation: normalizedTruncation(value.truncation), + recordDigest: value.recordDigest, + }); + if ( + record.schema !== 'qinglong/run-attempt-log-retirement@v1' || + (record.disposition !== 'deleted' && + record.disposition !== 'already_absent') || + record.eligibleAtMs < record.finishedAtMs || + record.retiredAtMs < record.eligibleAtMs || + (record.disposition === 'already_absent' && record.byteLength !== 0) || + typeof record.recordDigest !== 'string' || + !DIGEST_PATTERN.test(record.recordDigest) || + digestRunAttemptLogRetirementRecord(record) !== record.recordDigest + ) { + throw new InvalidRunAttemptLogRetentionError('record evidence is invalid'); + } + return record; +} + +function assertOptions(options: RunAttemptLogRetentionServiceOptions): void { + if (!options || typeof options !== 'object' || Array.isArray(options)) { + throw new InvalidRunAttemptLogRetentionError('options are invalid'); + } + const integer = (value: number, minimum: number, maximum: number) => + Number.isSafeInteger(value) && value >= minimum && value <= maximum; + if ( + !integer( + options.normalRetentionMs, + MIN_RUN_ATTEMPT_LOG_RETENTION_MS, + MAX_RUN_ATTEMPT_LOG_RETENTION_MS, + ) || + !integer( + options.pressureRetentionMs, + MIN_RUN_ATTEMPT_LOG_RETENTION_MS, + options.normalRetentionMs, + ) || + !integer(options.minimumFreeBytes, 0, Number.MAX_SAFE_INTEGER) || + !integer(options.pageSize, 1, MAX_RUN_ATTEMPT_LOG_RETENTION_PAGE_SIZE) || + !integer( + options.maximumDeletions, + 1, + Math.min(options.pageSize, MAX_RUN_ATTEMPT_LOG_RETENTION_DELETIONS), + ) || + (options.clock !== undefined && typeof options.clock.now !== 'function') + ) { + throw new InvalidRunAttemptLogRetentionError('options are invalid'); + } +} + +export class RunAttemptLogRetentionService { + private readonly clock: { now(): number }; + + constructor( + private readonly repository: RunAttemptLogRetentionRepository, + private readonly store: RunAttemptLogRetirementStore, + private readonly capacity: RunAttemptLogCapacitySource, + private readonly options: RunAttemptLogRetentionServiceOptions, + ) { + if ( + !repository || + typeof repository.inspect !== 'function' || + typeof repository.loadCursor !== 'function' || + typeof repository.list !== 'function' || + typeof repository.record !== 'function' || + typeof repository.saveCursor !== 'function' || + !store || + typeof store.retire !== 'function' || + !capacity || + typeof capacity.inspect !== 'function' + ) { + throw new InvalidRunAttemptLogRetentionError('dependencies are invalid'); + } + assertOptions(options); + this.clock = options.clock ?? { now: Date.now }; + } + + async sweep(): Promise { + const observedAtMs = timestamp('clock', this.clock.now()); + const snapshot = await this.capacity.inspect(); + if ( + typeof snapshot?.availableBytes !== 'bigint' || + typeof snapshot.totalBytes !== 'bigint' || + snapshot.availableBytes < 0n || + snapshot.totalBytes < 1n || + snapshot.availableBytes > snapshot.totalBytes + ) { + throw new RunAttemptLogRetentionUnavailableError(); + } + const pressure = + snapshot.availableBytes < BigInt(this.options.minimumFreeBytes); + const retentionMs = pressure + ? this.options.pressureRetentionMs + : this.options.normalRetentionMs; + const cutoffMs = Math.max(0, observedAtMs - retentionMs); + const cursor = await this.repository.loadCursor(); + const page = await this.repository.list({ + cutoffMs, + limit: this.options.pageSize, + ...(cursor === undefined ? {} : { cursor }), + }); + this.assertPage(page, cursor); + + const entries: RunAttemptLogRetentionEntry[] = []; + let deletionsAttempted = 0; + let recordsWritten = 0; + let failedCandidates = 0; + let bytesReclaimed = 0; + let lastProcessed = cursor; + + for (const rawCandidate of page.candidates) { + if (deletionsAttempted >= this.options.maximumDeletions) break; + const candidate = normalizeRunAttemptLogRetentionCandidate(rawCandidate); + deletionsAttempted += 1; + lastProcessed = Object.freeze({ + finishedAtMs: candidate.finishedAtMs, + attemptId: candidate.attemptId, + }); + let retired: Readonly; + try { + retired = this.normalizeRetirement(await this.store.retire(candidate)); + } catch { + failedCandidates += 1; + entries.push( + Object.freeze({ + attemptId: candidate.attemptId, + logArtifactId: candidate.logArtifactId, + outcome: 'file_failed' as const, + byteLength: 0, + }), + ); + continue; + } + try { + const result = await this.repository.record( + createRunAttemptLogRetirementRecord({ + ...candidate, + eligibleAtMs: candidate.finishedAtMs + retentionMs, + retiredAtMs: observedAtMs, + ...retired, + }), + ); + if (result !== 'recorded' && result !== 'existing') { + throw new RunAttemptLogRetentionUnavailableError(); + } + recordsWritten += result === 'recorded' ? 1 : 0; + bytesReclaimed += retired.byteLength; + entries.push( + Object.freeze({ + attemptId: candidate.attemptId, + logArtifactId: candidate.logArtifactId, + outcome: retired.disposition, + byteLength: retired.byteLength, + }), + ); + } catch { + failedCandidates += 1; + entries.push( + Object.freeze({ + attemptId: candidate.attemptId, + logArtifactId: candidate.logArtifactId, + outcome: 'record_failed' as const, + byteLength: retired.byteLength, + }), + ); + } + } + + const budgetExhausted = + deletionsAttempted < page.candidates.length && + deletionsAttempted >= this.options.maximumDeletions; + const nextCursor = budgetExhausted + ? lastProcessed + : page.truncated + ? page.nextCursor + : undefined; + await this.repository.saveCursor(nextCursor, observedAtMs); + return Object.freeze({ + status: budgetExhausted + ? ('deletion_budget_exhausted' as const) + : page.truncated + ? ('page_complete' as const) + : ('complete' as const), + pressure, + observedAtMs, + retentionMs, + availableBytes: snapshot.availableBytes.toString(10), + totalBytes: snapshot.totalBytes.toString(10), + candidatesScanned: deletionsAttempted, + deletionsAttempted, + recordsWritten, + failedCandidates, + bytesReclaimed, + entries: Object.freeze(entries), + ...(nextCursor === undefined ? {} : { nextCursor }), + }); + } + + private normalizeRetirement( + value: Readonly, + ): Readonly { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new RunAttemptLogRetentionUnavailableError(); + } + exactKeys( + value, + ['byteLength', 'disposition', 'truncation'], + [], + 'retirement result', + ); + const byteLength = nonNegativeInteger('byteLength', value.byteLength); + if ( + (value.disposition !== 'deleted' && + value.disposition !== 'already_absent') || + (value.disposition === 'already_absent' && byteLength !== 0) + ) { + throw new RunAttemptLogRetentionUnavailableError(); + } + return Object.freeze({ + disposition: value.disposition, + byteLength, + truncation: normalizedTruncation(value.truncation), + }); + } + + private assertPage( + page: RunAttemptLogRetentionPage, + cursor: Readonly | undefined, + ): void { + if ( + !page || + !Array.isArray(page.candidates) || + page.candidates.length > this.options.pageSize || + typeof page.truncated !== 'boolean' + ) { + throw new RunAttemptLogRetentionUnavailableError(); + } + let previous = cursor; + for (const raw of page.candidates) { + const candidate = normalizeRunAttemptLogRetentionCandidate(raw); + if ( + previous && + (candidate.finishedAtMs < previous.finishedAtMs || + (candidate.finishedAtMs === previous.finishedAtMs && + candidate.attemptId <= previous.attemptId)) + ) { + throw new RunAttemptLogRetentionUnavailableError(); + } + previous = candidate; + } + const last = page.candidates.at(-1); + if ( + page.truncated !== (page.nextCursor !== undefined) || + (page.nextCursor !== undefined && + (!last || + page.nextCursor.finishedAtMs !== last.finishedAtMs || + page.nextCursor.attemptId !== last.attemptId)) + ) { + throw new RunAttemptLogRetentionUnavailableError(); + } + } +} diff --git a/packages/ql3-runtime-core/test/runAttemptLogRead.test.cjs b/packages/ql3-runtime-core/test/runAttemptLogRead.test.cjs index 01d30f25..1d4c876a 100644 --- a/packages/ql3-runtime-core/test/runAttemptLogRead.test.cjs +++ b/packages/ql3-runtime-core/test/runAttemptLogRead.test.cjs @@ -67,14 +67,20 @@ function service(overrides = {}) { }; return { calls, - value: new RunAttemptLogReadService(runs, reader, { - executorType: overrides.executorType ?? 'local_process', - artifactIdPattern: overrides.artifactIdPattern ?? /^local-[a-f0-9]{30}$/, - maximumReadBytes: overrides.maximumReadBytes ?? 32 * 1024, - ...(overrides.activeMissingIsPending === undefined - ? {} - : { activeMissingIsPending: overrides.activeMissingIsPending }), - }), + value: new RunAttemptLogReadService( + runs, + reader, + { + executorType: overrides.executorType ?? 'local_process', + artifactIdPattern: + overrides.artifactIdPattern ?? /^local-[a-f0-9]{30}$/, + maximumReadBytes: overrides.maximumReadBytes ?? 32 * 1024, + ...(overrides.activeMissingIsPending === undefined + ? {} + : { activeMissingIsPending: overrides.activeMissingIsPending }), + }, + overrides.retention, + ), }; } @@ -224,6 +230,72 @@ test('returns a validated bounded snapshot without copying storage bytes', async assert.equal(calls.length, 0); }); +test('returns a durable retirement before storage and rechecks after a missing read', async () => { + const { + createRunAttemptLogRetirementRecord, + } = require('../dist/run/log-retention/runAttemptLogRetention.js'); + const tombstone = createRunAttemptLogRetirementRecord({ + projectId: 'prj_default', + runId: 'run_123', + attemptId: 'attempt_123', + logArtifactId: `local-${'a'.repeat(30)}`, + executorType: 'local_process', + finishedAtMs: 10, + eligibleAtMs: 20, + retiredAtMs: 30, + disposition: 'deleted', + byteLength: 42, + truncation: { truncated: false, maximumBytes: 1024, observedAtMs: 9 }, + }); + let inspections = 0; + let reads = 0; + const before = service({ + retention: { + async inspect() { + inspections += 1; + return { status: 'retired', record: tombstone }; + }, + }, + reader: { + async read() { + reads += 1; + return { status: 'missing' }; + }, + }, + }); + assert.deepEqual(await before.value.read(request()), { + status: 'retired', + projectId: 'prj_default', + runId: 'run_123', + attemptId: 'attempt_123', + logArtifactId: `local-${'a'.repeat(30)}`, + retiredAtMs: 30, + byteLength: 42, + truncation: { truncated: false, maximumBytes: 1024, observedAtMs: 9 }, + }); + assert.equal(inspections, 1); + assert.equal(reads, 0); + + inspections = 0; + const after = service({ + retention: { + async inspect() { + inspections += 1; + return inspections === 1 + ? { status: 'active' } + : { status: 'retired', record: tombstone }; + }, + }, + reader: { + async read() { + return { status: 'missing' }; + }, + }, + }); + assert.equal((await after.value.read(request())).status, 'retired'); + assert.equal(inspections, 2); +}); + test('fails closed on malformed storage results and dependency failures', async () => { const malformed = service({ reader: { diff --git a/packages/ql3-runtime-core/test/runAttemptLogRetention.test.cjs b/packages/ql3-runtime-core/test/runAttemptLogRetention.test.cjs new file mode 100644 index 00000000..f52af580 --- /dev/null +++ b/packages/ql3-runtime-core/test/runAttemptLogRetention.test.cjs @@ -0,0 +1,150 @@ +const assert = require('node:assert/strict'); +const { test } = require('node:test'); + +const { + InvalidRunAttemptLogRetentionError, + RunAttemptLogRetentionService, + createRunAttemptLogRetirementRecord, + normalizeRunAttemptLogRetirementRecord, +} = require('../dist/run/log-retention/runAttemptLogRetention.js'); + +function candidate(index = 1) { + return { + projectId: 'prj_default', + runId: `run_${index}`, + attemptId: `attempt_${index}`, + logArtifactId: `local-${String(index).padStart(30, 'a')}`, + executorType: 'local_process', + finishedAtMs: index, + }; +} + +test('creates tamper-evident exact retirement records', () => { + const record = createRunAttemptLogRetirementRecord({ + ...candidate(), + eligibleAtMs: 2, + retiredAtMs: 3, + disposition: 'deleted', + byteLength: 10, + truncation: { truncated: true, maximumBytes: 64, observedAtMs: 1 }, + }); + assert.match(record.recordDigest, /^[a-f0-9]{64}$/); + assert.deepEqual(normalizeRunAttemptLogRetirementRecord(record), record); + assert.throws( + () => + normalizeRunAttemptLogRetirementRecord({ + ...record, + byteLength: 11, + }), + InvalidRunAttemptLogRetentionError, + ); +}); + +test('uses pressure policy and persists a bounded resume cursor', async () => { + let saved; + const recorded = []; + const values = [candidate(1), candidate(2), candidate(3)]; + const service = new RunAttemptLogRetentionService( + { + async inspect() { + return { status: 'active' }; + }, + async loadCursor() { + return undefined; + }, + async list(input) { + assert.equal(input.cutoffMs, 9 * 60_000); + assert.equal(input.limit, 3); + return { + candidates: values, + truncated: false, + }; + }, + async record(record) { + recorded.push(record); + return 'recorded'; + }, + async saveCursor(cursor) { + saved = cursor; + }, + }, + { + async retire(value) { + return { + disposition: 'deleted', + byteLength: value.finishedAtMs, + truncation: { truncated: 'unknown' }, + }; + }, + }, + { + async inspect() { + return { availableBytes: 9n, totalBytes: 100n }; + }, + }, + { + normalRetentionMs: 5 * 60_000, + pressureRetentionMs: 60_000, + minimumFreeBytes: 10, + pageSize: 3, + maximumDeletions: 2, + clock: { now: () => 10 * 60_000 }, + }, + ); + const result = await service.sweep(); + assert.equal(result.status, 'deletion_budget_exhausted'); + assert.equal(result.pressure, true); + assert.equal(result.deletionsAttempted, 2); + assert.equal(result.bytesReclaimed, 3); + assert.equal(recorded.length, 2); + assert.deepEqual(saved, { finishedAtMs: 2, attemptId: 'attempt_2' }); +}); + +test('advances past failures and clears the cursor after a complete page', async () => { + const saved = []; + const service = new RunAttemptLogRetentionService( + { + async inspect() { + return { status: 'active' }; + }, + async loadCursor() { + return { finishedAtMs: 1, attemptId: 'attempt_1' }; + }, + async list() { + return { candidates: [candidate(2)], truncated: false }; + }, + async record() { + throw new Error('database unavailable'); + }, + async saveCursor(cursor) { + saved.push(cursor); + }, + }, + { + async retire() { + return { + disposition: 'already_absent', + byteLength: 0, + truncation: { truncated: 'unknown' }, + }; + }, + }, + { + async inspect() { + return { availableBytes: 100n, totalBytes: 100n }; + }, + }, + { + normalRetentionMs: 60_000, + pressureRetentionMs: 60_000, + minimumFreeBytes: 0, + pageSize: 2, + maximumDeletions: 2, + clock: { now: () => 10 * 60_000 }, + }, + ); + const result = await service.sweep(); + assert.equal(result.failedCandidates, 1); + assert.equal(result.entries[0].outcome, 'record_failed'); + assert.deepEqual(saved, [undefined]); +}); diff --git a/scripts/ql3-cluster-dependency-audit.cjs b/scripts/ql3-cluster-dependency-audit.cjs index 14ba8c13..0568ea15 100644 --- a/scripts/ql3-cluster-dependency-audit.cjs +++ b/scripts/ql3-cluster-dependency-audit.cjs @@ -2420,6 +2420,7 @@ function auditSourceImports(root, packagePath, findings) { '@qinglong/runtime-core/plugin-package-task-publication', '@qinglong/runtime-core/project-tool-definition-snapshot', '@qinglong/runtime-core/run-attempt-log-read', + '@qinglong/runtime-core/run-attempt-log-retention', '@qinglong/runtime-core/task-spec-semantic', ].includes(specifier) && !( diff --git a/scripts/ql3-local-compose-preflight-live-contract.cjs b/scripts/ql3-local-compose-preflight-live-contract.cjs index 26e40bf0..a98cc522 100644 --- a/scripts/ql3-local-compose-preflight-live-contract.cjs +++ b/scripts/ql3-local-compose-preflight-live-contract.cjs @@ -148,7 +148,7 @@ async function main() { report.status !== 'ready' || report.generation !== 1 || report.profile !== input.profile || - report.sqlite?.contractVersion !== 43 || + report.sqlite?.contractVersion !== 44 || (report.image?.architecture !== 'amd64' && report.image?.architecture !== 'arm64') || report.service?.kind !== 'compose' diff --git a/scripts/ql3-local-compose-rollout-live-contract.cjs b/scripts/ql3-local-compose-rollout-live-contract.cjs index dfb92e07..b750abf3 100644 --- a/scripts/ql3-local-compose-rollout-live-contract.cjs +++ b/scripts/ql3-local-compose-rollout-live-contract.cjs @@ -361,8 +361,8 @@ async function main() { (fs.statSync(path.join(rolloutRoot, name)).mode & 0o777) !== 0o600, ) || (fs.statSync(backupPath).mode & 0o777) !== 0o600 || - receipt.sqlite?.contractVersion !== 43 || - receipt.sqlite?.writeContractVersion !== 43 || + receipt.sqlite?.contractVersion !== 44 || + receipt.sqlite?.writeContractVersion !== 44 || (receipt.sqlite?.writeObservation !== 'unchanged' && receipt.sqlite?.writeObservation !== 'changed') || receipt.sqlite?.backup?.sha256 !== backup.sha256 || diff --git a/scripts/ql3-local-image-audit.cjs b/scripts/ql3-local-image-audit.cjs index b2b7a4df..e9b62e08 100644 --- a/scripts/ql3-local-image-audit.cjs +++ b/scripts/ql3-local-image-audit.cjs @@ -245,9 +245,9 @@ function auditDockerfile(contents, findings) { !contents.includes('io.qinglong.ai="excluded"') || !contents.includes('io.qinglong.profile="edge,standalone"') || !contents.includes('io.qinglong.local.application-config="2"') || - !contents.includes('io.qinglong.local.sqlite-contract-min="43"') || - !contents.includes('io.qinglong.local.sqlite-contract-max="43"') || - !contents.includes('io.qinglong.local.sqlite-write-contract="43"') || + !contents.includes('io.qinglong.local.sqlite-contract-min="44"') || + !contents.includes('io.qinglong.local.sqlite-contract-max="44"') || + !contents.includes('io.qinglong.local.sqlite-write-contract="44"') || !contents.includes('io.qinglong.local.compose-selection="1"') ) { addFinding(findings, 'RUNTIME_IDENTITY_OR_LABEL_DRIFT'); diff --git a/scripts/ql3-physical-edge-compose-storage.cjs b/scripts/ql3-physical-edge-compose-storage.cjs index 8e5b4801..0ab4eea6 100644 --- a/scripts/ql3-physical-edge-compose-storage.cjs +++ b/scripts/ql3-physical-edge-compose-storage.cjs @@ -535,7 +535,7 @@ function snapshotReceipt(evidence) { function isSnapshot(value) { return ( hasExactKeys(value, SNAPSHOT_KEYS) && - value.contractVersion === 43 && + value.contractVersion === 44 && typeof value.sha256 === 'string' && SHA256_PATTERN.test(value.sha256) && Number.isSafeInteger(value.bytes) && @@ -1337,7 +1337,7 @@ async function resumePhase(options, manifest) { const outcomes = Object.freeze({ commitStatus: committed.status, replayStatus: replay.status, - sqliteIntegrity: databaseEvidence.contractVersion === 43 ? 'ok' : 'invalid', + sqliteIntegrity: databaseEvidence.contractVersion === 44 ? 'ok' : 'invalid', stageRemoved: !fs.existsSync(session.collection.stagePath), tombstonePresent: fs.existsSync(tombstonePath), retainedSnapshots, diff --git a/test/back/ql3LocalImageAudit.test.cjs b/test/back/ql3LocalImageAudit.test.cjs index f3f5cd6d..f52bf9d9 100644 --- a/test/back/ql3LocalImageAudit.test.cjs +++ b/test/back/ql3LocalImageAudit.test.cjs @@ -128,7 +128,7 @@ test('rejects retaining npm bin links, debug maps or declarations in the product const dockerfile = fs .readFileSync(dockerfilePath, 'utf8') .replace( - 'RUN rm -rf node_modules/.bin \\\n' + + 'RUN rm -rf node_modules/.bin \\\n' + ' && node /tmp/ql3-prune-runtime-artifact.cjs node_modules/@qinglong \\\n' + ' @qinglong/local-application \\\n' + ' @qinglong/local-application/process \\\n' + @@ -156,7 +156,7 @@ test('rejects removal of the SQLite rollout compatibility labels', () => { const dockerfilePath = path.join(current.target, 'Dockerfile'); const dockerfile = fs .readFileSync(dockerfilePath, 'utf8') - .replace(' io.qinglong.local.sqlite-write-contract="43" \\\n', ''); + .replace(' io.qinglong.local.sqlite-write-contract="44" \\\n', ''); fs.writeFileSync(dockerfilePath, dockerfile); const report = auditLocalImageContract(current.root); assert.equal(report.compatible, false); diff --git a/test/back/ql3PackageBoundaryAudit.test.cjs b/test/back/ql3PackageBoundaryAudit.test.cjs index 239e72ca..9c0b79aa 100644 --- a/test/back/ql3PackageBoundaryAudit.test.cjs +++ b/test/back/ql3PackageBoundaryAudit.test.cjs @@ -79,7 +79,7 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( }, { directory: 'packages/ql3-local-sqlite/src/migrations', - directSourceFiles: 87, + directSourceFiles: 89, reviewKind: 'ordered_ledger', }, ], @@ -146,10 +146,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( { path: 'packages/ql3-local-execution', name: '@qinglong/local-execution', - sourceFiles: 21, + sourceFiles: 22, rootSourceFiles: 0, rootSourceLines: 0, - nestedSourceFiles: 21, + nestedSourceFiles: 22, rootSourceFileHardCap: 0, rootSourceLineHardCap: 0, rootSourceFileRoles: {}, @@ -299,10 +299,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( rootSourceFileRoles: runtimeCore.rootSourceFileRoles, }, { - sourceFiles: 149, + sourceFiles: 150, rootSourceFiles: 1, rootSourceLines: 160, - nestedSourceFiles: 148, + nestedSourceFiles: 149, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, ); @@ -540,10 +540,10 @@ test('current QL3 workspace has exactly eighteen reviewed package boundaries', ( rootSourceFileRoles: localSqlite.rootSourceFileRoles, }, { - sourceFiles: 173, + sourceFiles: 176, rootSourceFiles: 1, rootSourceLines: 31, - nestedSourceFiles: 172, + nestedSourceFiles: 175, rootSourceFileRoles: { 'index.ts': 'public_export' }, }, ); diff --git a/test/back/ql3PhysicalEdgeComposeStorage.test.cjs b/test/back/ql3PhysicalEdgeComposeStorage.test.cjs index 1729ecff..5fab2e92 100644 --- a/test/back/ql3PhysicalEdgeComposeStorage.test.cjs +++ b/test/back/ql3PhysicalEdgeComposeStorage.test.cjs @@ -107,7 +107,7 @@ function sessionFixture(overrides = {}) { target: { rolloutId, snapshot: { - contractVersion: 43, + contractVersion: 44, sha256: 'a'.repeat(64), bytes: 4_000_000, pageCount: 1000, diff --git a/test/back/ql3PhysicalEdgeEvidence.test.cjs b/test/back/ql3PhysicalEdgeEvidence.test.cjs index 4120f73b..cee082e9 100644 --- a/test/back/ql3PhysicalEdgeEvidence.test.cjs +++ b/test/back/ql3PhysicalEdgeEvidence.test.cjs @@ -552,7 +552,7 @@ test('imports a different-boot Compose collection storage candidate without wide maximumResumeWriteAmplificationPermille: 50_000, }); const targetSnapshot = { - contractVersion: 43, + contractVersion: 44, sha256: 'a'.repeat(64), bytes: 4_000_000, pageCount: 1000,