diff --git a/.env.example b/.env.example index 08777be..48cf436 100644 --- a/.env.example +++ b/.env.example @@ -34,3 +34,10 @@ ORBIT_METRICS_ADDR= # 仅供 orbit-admin retention archive / verify-archive 使用；每个作业单独的绝对目录。 # 不会启动后台定时归档，完整参数见 docs/api/retention-jobs.md。 # ORBIT_ARCHIVE_DIRECTORY=/srv/orbit-archives/tasks-20260923 + +# 周期归档：持久卷目录和固定卷ID必须一起配置，默认关闭。 +# 镜像用户65532须有目录权限；共享卷节点必须使用同一物理卷。 +ORBIT_ARCHIVE_ROOT= +ORBIT_ARCHIVE_STORAGE_ID= +ORBIT_ARCHIVE_MAX_BYTES=10737418240 +ORBIT_ARCHIVE_TICK_INTERVAL=10s diff --git a/README.md b/README.md index ba37944..a5660ec 100644 --- a/README.md +++ b/README.md @@ -114,3 +114,5 @@ go get git.yuntuo.hk/k2safe/orbit/sdk/go@v0.1.0 无需本地 `replace`；导入方式、完整调用与版本边界见 [SDK 接入](docs/sdk/README.md#5-外部项目通过远端接入)。 [流程实例筛选分页](docs/api/workflow-queries.md)已提供独立SDK和[完整示例](docs/examples/workflow-query.md)，按固定版本全部队列鉴权，数据库保持零外键。 + +- [周期归档与作业管理](docs/api/archive-scheduler.md) diff --git a/cmd/orbit/main.go b/cmd/orbit/main.go index 5815a3a..7488e71 100644 --- a/cmd/orbit/main.go +++ b/cmd/orbit/main.go @@ -13,6 +13,7 @@ import ( "git.yuntuo.hk/k2safe/orbit/internal/config" "git.yuntuo.hk/k2safe/orbit/internal/observability" + "git.yuntuo.hk/k2safe/orbit/internal/retention" "git.yuntuo.hk/k2safe/orbit/internal/storage/mysql" "git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver" core "git.yuntuo.hk/k2safe/orbit/model" @@ -123,8 +124,37 @@ func run() error { httpserver.RegisterWorkflowDefinitions(server, definition.New(mysql.NewWorkflowDefinitionStore(db))) httpserver.RegisterWorkflows(server, instance.New(mysql.NewWorkflowInstanceStore(db)), control.New(mysql.NewWorkflowControlStore(db)), weavesignal.New(mysql.NewWorkflowSignalStore(db)), recovery.New(mysql.NewWorkflowRecoveryStore(db))) httpserver.RegisterConsoleRetention(server, consoleAuth, retentionStore, retentionStore) + httpserver.RegisterConsoleArchiveJobs(server, consoleAuth, retentionStore, cfg.ArchiveStorageID) httpserver.WrapConsoleAuth(server, consoleAuth) httpserver.WrapMetrics(server, metrics) + + if cfg.ArchiveRoot != "" { + archives := &retention.ArchiveRuntime{Directory: cfg.ArchiveRoot, StorageID: cfg.ArchiveStorageID, MaxBytes: cfg.ArchiveMaxBytes, Store: retentionStore} + if err := archives.Init(); err != nil { + return errors.New("archive volume initialization failed; check path, ownership and storage identity") + } + archiveCtx, stopArchives := context.WithCancel(root) + archiveDone := make(chan struct{}) + go func() { + defer close(archiveDone) + ticker := time.NewTicker(cfg.ArchiveTickInterval) + defer ticker.Stop() + for { + select { + case <-archiveCtx.Done(): + return + case <-ticker.C: + call, cancel := context.WithTimeout(archiveCtx, 45*time.Second) + err := archives.Tick(call) + cancel() + if err != nil && archiveCtx.Err() == nil { + slog.Error("archive batch could not be confirmed", "node_id", node) + } + } + } + }() + defer func() { stopArchives(); <-archiveDone }() + } workflows, err := mysql.NewWorkflowRuntime(db, node) if err != nil { return err diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 203194b..f634562 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -266,3 +266,11 @@ Event.WorkflowID为只读STORED生成列，从原不可变事件派生流程归 ## 文件归档作业模型 `internal/retention` 中的 `JobSpec`、`JobManifest`、`JobBatch` 和 `JobStatus` 定义固定作业范围、不可变清单、连续批次及汇总状态；各持久字段提供中文注释。它们只存入私有归档目录，不新增数据库表。连续文件与摘要链就是恢复检查点，作业 `JobID` 进入现有管理审计的执行者身份。协议与失败边界见[文件归档作业](api/retention-jobs.md)。 + +### 周期归档（000031） + +`orbit_archive_plans` 以命名空间和类别唯一，保存配置版本、保留策略版本、卷身份、周期秒数、批大小、字节预算、下次触发时间和未结清作业占位。失败也占位，关闭计划只停止新作业。 + +`orbit_archive_jobs` 固定范围和卷，保存 PENDING/RUNNING/SUCCEEDED/FAILED/ABANDONED 状态、版本、领取次数、已确认的批数/根数/字节/链摘要，以及随机租约代次和截止。租约确认必须匹配代次与版本且未过期；人工操作需要原版本、幂等键与审计原因。两个表均无外键，关联在同一事务按命名空间核验。 + +数据库与文件不做跨资源事务。持久文件是恢复依据，数据库链前缀是可信核验锚点；文件发布后确认丢失，由原作业重新读文件校验后收敛，缺失文件不自动重建。见[接口与部署说明](api/archive-scheduler.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index a289899..508984f 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -245,3 +245,10 @@ A04派发已提交而消息尚未到达Worker的精确进程故障已通过[专 ### P4/P5 阶段结论与下一批 [逐项证据与生产边界](reports/phase-4-5.md)已落表，P4/P5平台功能阶段验收通过。W06/W09/W14/W15真实竞争和全停、A04未送达、W01/W13提交断线及W16旧结果重放竞争均有对应回归，不重复列为待开发。运行指标、凭据轮换、保留策略及只读归档、隔离容量样本和备份恢复均已按上方 P6 留有验证记录；后续重点是归档介质与删除执行器、长期磁盘增长、目标生产拓扑的容量/数据库故障及 RPO/RTO 验收。P7 负责剩余场景和版本发布。完整首版目标保持进行中。 + +### 周期归档与控制台作业（本批） + +- 已实现：000031 无外键计划/作业模型及中文注释；按类别配置周期与预算；数据库时间合并错过周期；固定版本/卷身份；有界领取、过期接管与迟到隔离；连续文件链和数据库前缀核验；单作业及全卷预算；单管理员页面、分页、失败重试/关闭、幂等与审计。 +- 已验证：真实 MySQL 多节点竞争、落盘后确认丢失、文件损坏、策略变化、预算提高与卷预算、分页；标准测试、容器端到端、页面实际操作与全量集成回归均已通过。 +- 不计入本批完成项：源数据删除、跨作业增量去重、对象存储、任意网络文件系统语义及目标生产跨故障域验收。 +- [接口/部署/恢复示例](api/archive-scheduler.md)与[验证记录](reports/phase-6-archive-scheduler.md)。 diff --git a/docs/README.md b/docs/README.md index 4cab33a..f577778 100644 --- a/docs/README.md +++ b/docs/README.md @@ -108,3 +108,7 @@ Orbit 不内置特定行业规则；Pulse 管理执行，Weave 管理步骤依 持续开发目标已建立。当前实现和验收进度以 [开发计划](08-development-plan.md)、[P0 报告](reports/phase-0.md)、[P1 报告](reports/phase-1.md) 与 [P2 阶段验收](reports/phase-2.md) 、[P3 安全重试报告](reports/phase-3-safe-retry.md) 、[P3 计划调度报告](reports/phase-3-schedules.md) 和 [P3 完整范围验收](reports/phase-3.md) 为准；早期设计稿中的“未实现”描述保留其编写阶段语境。 [实例查询接口与SDK](api/workflow-queries.md)已接入，固定筛选沿游标分页，覆盖未执行分支的全部队列权限。 + +- [周期归档与作业管理](api/archive-scheduler.md) + +- [周期归档与作业管理验证记录](reports/phase-6-archive-scheduler.md) diff --git a/docs/api/README.md b/docs/api/README.md index 1db68f2..f5310b3 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -64,3 +64,5 @@ P1 业务接口已遵守下述约定，后续接口继续沿用： 实现时必须补齐：用途与权限、HTTP 方法和路径、请求头、字段类型/必填/默认值/限制、完整请求与响应、错误码、幂等与并发规则、超时/重试/取消语义、SDK 对应方法、可运行示例、兼容版本和已覆盖的回归场景。 [流程实例筛选分页与权限](workflow-queries.md)已接入，固定筛选沿游标分页，覆盖未执行分支的全部队列权限。 + +- [周期归档与作业管理](archive-scheduler.md) diff --git a/docs/api/archive-scheduler.md b/docs/api/archive-scheduler.md new file mode 100644 index 0000000..0631609 --- /dev/null +++ b/docs/api/archive-scheduler.md @@ -0,0 +1,146 @@ +# 周期归档与作业管理 + +Orbit 控制台的「归档作业」页面提供周期配置、作业列表、失败重试及关闭。沿用单管理员账号密码登录，不引入角色体系。业务 SDK 不需要也不能领取归档作业。 + +## 1. 部署执行器 + +需要 Schema **31**。停止旧节点后执行 `orbit-migrate up`，再启动新版本节点。000031 只增加 `orbit_archive_plans`、`orbit_archive_jobs`，不包含外键或源数据清理。 + +```dotenv +# 必须挂载持久卷，并让 Orbit 运行用户拥有目录，权限 0700。 +ORBIT_ARCHIVE_ROOT=/var/lib/orbit/archive +# 初次部署生成并保存一个固定的 32 位小写十六进制 ID，例如 openssl rand -hex 16。 +ORBIT_ARCHIVE_STORAGE_ID=11111111111111111111111111111111 +# 全卷逻辑文件字节预算，默认 10 GiB；范围 1 MiB–1 TiB。 +ORBIT_ARCHIVE_MAX_BYTES=10737418240 +# 默认 10s，范围 1s–1m。 +ORBIT_ARCHIVE_TICK_INTERVAL=10s +``` + +`ROOT` 和 `STORAGE_ID` 均为空时关闭此节点后台执行器，控制台仍可查询历史。两者必须一起设置。路径只来自部署配置，页面不能指定任意服务器目录。镜像用户为 `65532:65532`，目录须可由该用户访问。 + +运行示意（`orbit.env` 同时包含已有数据库等配置，不要提交到 Git）： + +```sh +mkdir -p ./archive-data +chmod 700 ./archive-data +# Linux 主机上按镜像实际 UID/GID 配置归档目录归属。 +sudo chown 65532:65532 ./archive-data +docker run --name orbit --env-file orbit.env \ + --mount type=bind,src="$(pwd)/archive-data",dst=/var/lib/orbit/archive \ + -p 127.0.0.1:8080:8080 orbit:local +``` + +同一 `STORAGE_ID` 的所有执行节点**必须挂载同一物理卷**，支持 POSIX 文件锁、原子硬链接和目录 fsync；同名路径、复制目录或相同 ID 不等于共享存储。卷根目录持久化 `.storage-id` 身份标记，恢复已有作业时要求原清单存在。独立本地盘使用不同卷 ID；其归档作业不能由别的独立盘接管。没有验证任意 NFS/对象存储语义，不据此宣称跨故障域归档高可用。 + +同卷执行串行，后台每轮最多调度 16 个到期类别、执行一个文件批次，工作截止 45 秒，数据库租约 5 分钟。正常继续在下一轮领取；进程异常退出且未确认时，等待租约到期后恢复。关机取消不写成文件损坏。 + +## 2. 使用顺序 + +1. 登录控制台，在「保留策略」确认各类保留天数，开启只读归档并保存。 +2. 打开「归档作业」，选择类别，配置间隔、批大小、单作业容量和变更原因，开启周期并保存。 +3. 首次启用立即到期，后台生成固定身份的作业；页面每 10 秒刷新，显示批次、根记录数、文件字节和链尾 SHA-256。 +4. 失败时查看分类，修复原因后填写处理原因，从原作业重试；不能恢复的旧版本作业可以关闭，原文件与审计保留。 + +任务、流程、事件、审计、交付、Inbox、幂等记录各有独立计划。间隔范围 **1–720 小时**，HTTP 支持 3600–2592000 整数秒；每批 **1–100** 条根记录，单作业 **1 MiB–1 GiB**。关联历史行不计入根记录数，但全部计入文件容量。 + +同一类别最多一个未结清作业，失败也占位，避免周期不断制造重复失败。错过多个周期合并触发一次，`next_run_at` 按本次数据库时间加周期计算，不补齐每个错过的时间槽。关闭计划仅停止新作业，已有作业继续结清。 + +每次周期是**新的有限全量候选快照**：会再次读取仍满足保留窗口的源记录，不是跨作业增量清理。每个作业内的分页截止、游标、类别、策略版本及批大小固定。文件落盘后才确认数据库进度；已落盘而未确认的文件可在恢复时重新核验并收敛。源数据库内容不删除。 + +保留策略修改后，计划仍绑定旧版本，因此不会再生成新作业，须重新保存计划确认新版本。正在运行的旧作业在下一次导出时以 `POLICY_CHANGED` 停止；若已完整落盘，则可核验并确认已完成的原快照。旧作业重试不能更换策略版本或重置截止。 + +## 3. HTTP 接口 + +以下接口只接受控制台 HttpOnly 会话 Cookie，普通 Bearer 凭据不可访问。命名空间取自会话，写请求拒绝跨来源浏览器调用。保存与人工操作必须提供一个 `Idempotency-Key`；网络或服务器错误后保持原路径、原请求体和原键重试。明确的 4xx 拒绝可核对并修正后使用新键。 + +### 查询周期配置 + +`GET /console/archive/plans` + +```json +{"data":{"executor_configured":true,"storage_id":"11111111111111111111111111111111","items":[{"namespace_id":"22222222222222222222222222222222","kind":"tasks","revision":"1","enabled":true,"policy_revision":"3","storage_id":"11111111111111111111111111111111","interval_seconds":86400,"batch_size":20,"max_bytes":268435456,"next_run_at":"2026-09-24T02:00:00Z","active_job_id":null,"updated_at":"2026-09-23T02:00:00Z"}]}} +``` + +实际返回全部七类。未保存的类别 `revision="0"`、`enabled=false`，没有卷绑定，时间为空。`executor_configured` 只表示当前 HTTP 节点已配置执行器，不表示实时健康探针或整个集群状态。 + +### 保存周期配置 + +`PUT /console/archive/plans/{kind}` + +```http +Content-Type: application/json +Idempotency-Key: archive-plan-change-001 +``` + +```json +{ + "expected_revision":"0", + "policy_revision":"3", + "enabled":true, + "interval_seconds":86400, + "batch_size":20, + "max_bytes":268435456, + "reason":"每日归档已超过保留窗口的任务历史" +} +``` + +全部字段必填，原因是 1–1024 个 UTF-8 字节的非空单行。响应 `data` 为保存后的计划。计划版本或保留策略版本不符返回 409；开启时保留策略必须已启用，当前节点必须配置卷。关闭已有计划时允许当前节点未配置执行器。存在未结清作业时不允许更换卷绑定。 + +### 查询作业列表 + +`GET /console/archive/jobs`，下一页 `GET /console/archive/jobs?before={next_before}`。 + +每页最多 20 条，按创建时间和 ID 倒序；位置必须是当前命名空间内的作业 ID。不是跨页冻结快照。返回： + +```json +{"data":{"items":[{"id":"33333333333333333333333333333333","namespace_id":"22222222222222222222222222222222","kind":"tasks","plan_revision":"1","policy_revision":"3","storage_id":"11111111111111111111111111111111","batch_size":20,"max_bytes":268435456,"state":"SUCCEEDED","revision":"5","attempts":2,"batches":2,"roots":32,"bytes":86124,"last_sha256":"64位小写十六进制摘要","error_code":"","lease_until":null,"created_at":"2026-09-23T02:00:00Z","updated_at":"2026-09-23T02:00:20Z"}],"next_before":""}} +``` + +`attempts` 是批次领取次数，包含分批继续和崩溃重领，不是业务任务重试次数。`batches/roots/bytes/last_sha256` 只展示已经入库确认的进度，落盘后确认失败时页面可能暂时落后。列表不返回文件正文、分页游标或内部租约代次。 + +| 状态 | 含义 | +| --- | --- | +| PENDING | 等待首批或下一批执行 | +| RUNNING | 已领取，租约内执行；到期可接管 | +| SUCCEEDED | 全部批次落盘、校验并确认完成 | +| FAILED | 等待管理员核查；阻止本类别创建后续作业 | +| ABANDONED | 管理员关闭失败作业，文件、数据库历史与审计保留 | + +### 重试或关闭失败作业 + +`POST /console/archive/jobs/{id}/actions` + +```http +Content-Type: application/json +Idempotency-Key: archive-job-retry-001 +``` + +```json +{"expected_revision":"8","action":"retry","max_bytes":536870912,"reason":"确认原归档卷完整，并将单作业预算提高到512MiB"} +``` + +响应 `data` 为操作后的作业。仅 FAILED 可操作；重试保持原 ID、目录、范围、截止与已确认链前缀，预算只可保持或提高。`action="abandon"` 关闭作业，`max_bytes` 必须等于原值；关闭后不删除文件，开启的计划可按到期时间生成新作业。原键重放仅返回原操作回执，不代表作业当前仍处于回执中的状态；随后查询列表确认实时状态。 + +## 4. 故障处理和文件核验 + +| 错误分类 | 处理方式 | +| --- | --- | +| STORAGE_LIMIT | 检查单作业预算和全卷预算；按需提高，或在运维确认备份后转移已经结清的文件。不会自动删除旧文件腾空间 | +| FILE_CHECK_FAILED | 检查原卷、目录、权限和文件完整性；恢复原字节后重试，禁止伪造摘要或跳过缺失批次 | +| STORAGE_BUSY | 同目录有其他 CLI/进程操作；核实旧执行者退出后重试 | +| POLICY_CHANGED | 原策略已变化或关闭；通常关闭旧作业，再按新版本保存计划 | +| PERMISSION_DENIED | 应用或命名空间停用；核查后按原状态重试或关闭 | +| EXPORT_FAILED | 数据库读取未完成；确认数据库恢复后用原作业重试 | + +如果在首份清单落盘前进程退出、卷不可写或全卷预算耗尽，接管不会擅自创建一个新的空目录。此时可核查后关闭失败作业，让计划生成新作业。已有清单的容量故障可提高预算后重试原作业。 + +文件位于 `ORBIT_ARCHIVE_ROOT/{namespace_id}-{job_id}/`，清单及批次格式与 [文件归档 CLI](retention-jobs.md) 一致。数据库确认的链尾摘要可独立保存。服务端停止操作该目录后，可离线核验： + +```sh +orbit-admin retention verify-archive \ + --directory /var/lib/orbit/archive/22222222222222222222222222222222-33333333333333333333333333333333 \ + --expected-sha256 "$TRUSTED_SHA256" +``` + +全卷预算按普通文件的逻辑大小累计，单次扫描最多 100000 个目录项，包含未完成、失败与已关闭作业；超过边界停止。并发同卷执行受全卷文件锁约束。它不等价于文件系统物理空间配额，不包含块分配/元数据开销，也不能约束绕过平台直接写目录的外部进程。生产使用独立持久卷，并监测实际磁盘余量。暂不提供文件删除、自动迁移介质、下载正文或对象存储接口。 diff --git a/docs/api/operations.md b/docs/api/operations.md index 1fe32b7..ac98592 100644 --- a/docs/api/operations.md +++ b/docs/api/operations.md @@ -47,7 +47,7 @@ ORBIT_MYSQL_DSN='user:password@tcp(127.0.0.1:3306)/orbit' \ 成功时输出类似： ```json -{"schema_version":30,"migration_count":30,"foreign_key_count":0,"required_index_count":6,"complete":true} +{"schema_version":31,"migration_count":31,"foreign_key_count":0,"required_index_count":6,"complete":true} ``` `complete=true` 表示当前二进制的迁移账本、Orbit 表零外键，以及 000026–000028 的六个有界观测索引均满足要求。需要同时核对发布物身份时，传入保存的清单： @@ -83,3 +83,5 @@ ORBIT_MYSQL_DSN='user:password@tcp(127.0.0.1:3306)/orbit' \ ## 文件归档与离线核验 [`orbit-admin retention archive / verify-archive`](retention-jobs.md) 提供持久文件作业、原目录续跑及摘要链核验。读取命名空间需显式指定已保存策略版本，每个新批次记录独立作业审计，源记录不删除；离线核验无需数据库。 + +周期归档部署、持久卷及故障处理见[周期归档管理](archive-scheduler.md)；需 Schema 31，默认关闭后台归档执行器。 diff --git a/docs/api/retention-jobs.md b/docs/api/retention-jobs.md index 6902211..6deade5 100644 --- a/docs/api/retention-jobs.md +++ b/docs/api/retention-jobs.md @@ -2,7 +2,7 @@ `orbit-admin retention archive` 自动按已保存的保留策略分批导出，并把每个批次可靠写入本地持久目录。它面向持有数据库运维权限的操作者，复用控制台导出的候选保护和版本核对，记录 `archive:` 审计身份；不需要控制台 Cookie，也不向业务 SDK 增加管理权限。数据库业务记录始终保留。 -本次提供的是可重复运行的有限批次作业执行器。周期触发、控制台作业列表、对象存储和清理执行器尚未接入；不能把它描述为已部署的定时归档服务。 +本页介绍可独立运行的有限批次 CLI 执行器。周期触发、控制台作业列表与人工重试见[周期归档管理](archive-scheduler.md)；对象存储和源数据清理尚未接入。 ## 配置与启动 @@ -111,3 +111,5 @@ docker run --rm --network none \ 当前文件执行器支持 Linux/macOS 本地 POSIX 文件锁、硬链接原子发布及文件/目录同步语义；其他系统显式拒绝。不同节点独立磁盘不共享进度；网络共享文件系统必须验证上述语义后才能使用，当前没有跨节点自动接管承诺。 容量上限覆盖一个作业的清单和已发布批次，不是整个卷的配额；还需预留一个临时批次空间。已有档案不会自动删除腾空间，不会自动生成新目录重试。每个作业最多 10000 批。该归档不是数据库备份，也不构成源数据删除授权。 + +周期触发、控制台作业列表与人工重试已由 Schema 31 增加，见[周期归档管理](archive-scheduler.md)。本页 CLI 仍可独立使用；独立 CLI 作业不会自动注册到管理列表。 diff --git a/docs/reports/phase-6-archive-scheduler.md b/docs/reports/phase-6-archive-scheduler.md new file mode 100644 index 0000000..b9168ac --- /dev/null +++ b/docs/reports/phase-6-archive-scheduler.md @@ -0,0 +1,49 @@ +# P6：周期归档、持久作业与管理页面 + +日期：2026-09-23。变更基于 `6eb429e`，新增 Schema 31。范围为通用周期归档，不包含源数据删除。 + +## 交付内容 + +- 每个命名空间七类独立周期计划，保存策略版本、卷身份、周期、批大小、容量、触发时间和未结清作业占位。 +- 作业状态与随机租约持久化；周期错过合并一次；失败占位，管理员可原作业重试或关闭。变更、人工操作、作业受理和终态审计同事务。 +- 卷身份文件、全卷锁与作业锁；每轮一个文件批次；固定作业身份，验证文件链与数据库已确认前缀后继续。文件先发布后确认，提交不明时保留原状态恢复。 +- 单作业和全卷逻辑文件容量上限，目录按 256 个条目分块扫描，最多 100000 项；不删除文件腾空间。 +- 独立「归档作业」页面：周期表单、七类启停概况、本页待处理与根记录数、作业分页、校验摘要、失败解释、原作业恢复/关闭。沿用单管理员登录。 +- [部署、接口与恢复操作示例](../api/archive-scheduler.md)；文件格式与独立 CLI 兼容。 + +## 验证证据 + +| 场景 | 实际验证 | +| --- | --- | +| 并发触发 | 真实 MySQL，8 个独立存储调用竞争同一到期计划，仅生成 1 个作业 | +| 并发领取 | 8 个竞争者仅一个有效租约；过期后换代领取，原代次确认返回状态冲突 | +| 文件/数据库确认间隙 | 首批 2 条审计已持久化，尚未确认数据库；控制测试将租约设为过期，新代次确认文件前缀，再完成末批，最终 2 批/3 条、同一 ID，离线摘要一致 | +| 文件损坏 | 损坏首批后停止，数据库保留原确认的批数与摘要；恢复原字节后以原 ID 重试，重复操作只留一条重试审计 | +| 策略变化 | 修改保留窗口后旧作业以 POLICY_CHANGED 停止；旧计划不再触发，关闭旧作业并确认新策略版本后才创建后续作业 | +| 暂停周期 | 已在执行的作业可正常结清，停止新周期触发 | +| 容量与分页 | 3 条各含 600000 字节正文的真实审计；1 MiB 预算停止，提高到 4 MiB 后原作业完成；独立全卷预算阻止新输出；26 个作业分 20+6 两页，无重复 | +| 文件与卷约束 | 单测覆盖数据库链前缀不符、原目录缺失、卷身份不符、初始化临时文件恢复、另一节点持锁时启动、忙卷不调度、目录分块计数与符号链接拒绝 | +| 身份与来源限制 | 未登录、只有 Bearer、跨命名空间和跨来源写操作拒绝；普通列表不返回租约代次 | +| 容器端到端 | 隔离 MySQL + Orbit 镜像，管理员登录、策略保存、周期保存/原键重放、后台生成归档、SUCCEEDED 和摘要、登出后 401；原 Worker/重试/计划及独立离线归档核验仍通过 | +| 浏览器实际操作 | 隔离测试账号从页面选择审计类别、保存周期，自动完成 2 批/3 条；制造 1 MiB 容量失败，页面填写原因提高到 4 MiB，原作业完成 3 批/5 条；失败计数从1回到0，旧周期结果保留 | +| 数据模型 | 当前基线与前向迁移无外键；历史升级保留原数据和既有索引；归档后源审计仍完整 | + +本批覆盖的是数据库竞争、受控过期与文件确认间隙，不把修改测试租约截止等同于生产等待五分钟或真实跨机房切换。独立 CLI 的实际 SIGKILL 恢复证据见[文件归档报告](phase-6-retention-jobs.md)。 + +## 回归入口 + +- `make test vet build docs-check`：通过。 +- `make console-e2e`：通过；首次运行发现 HTTP 路由范围冲突，修复后重跑通过，已加入路由注册/认证/跨来源回归。 +- `go test -race -count=1 -run '^TestIntegrationArchive' -v ./internal/storage/mysql`（隔离数据库）：3 个专项全部通过；最终目录分块扫描实现再次通过此回归。 +- 最终文件执行器、配置和 HTTP 包的 `go test -race`：通过。 +- `make integration`：通过。存储全量 305.578 秒，最终 SDK 进程故障组 132.167 秒、流程暂停全停组 12.906 秒；旧 Worker、订阅、调度、流程和两个独立业务模块回归通过。 + +新增容量夹具最初使用 UPDATE 填充旧审计，被已有 append-only 触发器拒绝；已修正为插入时写入大正文，未放宽审计不可改写约束。 + +## 当前边界 + +- 每次周期是新的候选快照，跨作业可能重复归档同一源记录。只有作业内分页固定截止和位置。 +- 保留策略调整不会自动改变旧作业；关闭失败作业保留全部文件和历史。 +- 首份清单写入前中断时，恢复不会重建缺失目录，需核查后关闭旧作业再生成新作业。 +- 根目录物理空间、备份介质、对象存储与源库清理不在本批交付内；容量上限按逻辑文件字节计，不替代文件系统配额。 +- 同卷接管要求所有节点访问同一支持所需 POSIX 语义的物理卷。隔离测试和本机 Docker 结果不代表任意网络文件系统或目标生产跨故障域验收。 diff --git a/internal/config/config.go b/internal/config/config.go index f677a7d..76a1113 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -3,8 +3,10 @@ package config import ( "errors" "fmt" + core "git.yuntuo.hk/k2safe/orbit/model" "net" "os" + "path/filepath" "strconv" "time" @@ -12,6 +14,10 @@ import ( ) type Config struct { + ArchiveRoot string // ArchiveRoot 私有持久归档卷绝对路径，空表示关闭后台执行器。 + ArchiveStorageID core.ID // ArchiveStorageID 32位小写十六进制卷身份，各节点须挂载同一物理卷。 + ArchiveMaxBytes int64 // ArchiveMaxBytes 全卷文件预算，默认10GiB。 + ArchiveTickInterval time.Duration // ArchiveTickInterval 后台扫描间隔，默认10秒。 RetentionDefaults retention.Policy // RetentionDefaults 未保存数据库策略时使用的部署默认值。 HTTPAddr string MySQLDSN string @@ -32,6 +38,7 @@ type Config struct { func Load() (Config, error) { return FromEnv(os.Getenv) } func FromEnv(get func(string) string) (Config, error) { c := Config{ + ArchiveRoot: get("ORBIT_ARCHIVE_ROOT"), ArchiveStorageID: core.ID(get("ORBIT_ARCHIVE_STORAGE_ID")), ArchiveMaxBytes: 10 << 30, ArchiveTickInterval: 10 * time.Second, HTTPAddr: get("ORBIT_HTTP_ADDR"), MySQLDSN: get("ORBIT_MYSQL_DSN"), MaxOpen: 20, @@ -123,6 +130,26 @@ func FromEnv(get func(string) string) (Config, error) { return Config{}, errors.New("invalid ORBIT_METRICS_ADDR") } } + + if c.ArchiveRoot != "" || c.ArchiveStorageID != "" { + if !filepath.IsAbs(c.ArchiveRoot) || c.ArchiveStorageID.Validate() != nil { + return Config{}, errors.New("ORBIT_ARCHIVE_ROOT and ORBIT_ARCHIVE_STORAGE_ID must be configured together") + } + } + if raw := get("ORBIT_ARCHIVE_MAX_BYTES"); raw != "" { + n, err := strconv.ParseInt(raw, 10, 64) + if err != nil || n < 1<<20 || n > 1<<40 { + return Config{}, errors.New("ORBIT_ARCHIVE_MAX_BYTES must be 1MiB to 1TiB") + } + c.ArchiveMaxBytes = n + } + if raw := get("ORBIT_ARCHIVE_TICK_INTERVAL"); raw != "" { + d, err := time.ParseDuration(raw) + if err != nil || d < time.Second || d > time.Minute { + return Config{}, errors.New("ORBIT_ARCHIVE_TICK_INTERVAL must be 1s to 1m") + } + c.ArchiveTickInterval = d + } var err error c.RetentionDefaults, err = retention.FromEnv(get) if err != nil { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index fe6583f..2c2b834 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -65,3 +65,28 @@ func TestMetricsAddressIsExplicitAndValidated(t *testing.T) { } } } + +func TestArchiveRuntimeRequiresBoundedPersistentVolumeConfig(t *testing.T) { + base := map[string]string{"ORBIT_MYSQL_DSN": "test"} + get := func(k string) string { return base[k] } + cfg, err := FromEnv(get) + if err != nil || cfg.ArchiveRoot != "" || cfg.ArchiveMaxBytes != 10<<30 { + t.Fatal(cfg, err) + } + base["ORBIT_ARCHIVE_ROOT"] = "/private/archive" + if _, err = FromEnv(get); err == nil { + t.Fatal("missing volume identity accepted") + } + base["ORBIT_ARCHIVE_STORAGE_ID"] = "11111111111111111111111111111111" + if _, err = FromEnv(get); err != nil { + t.Fatal(err) + } + for k, v := range map[string]string{"ORBIT_ARCHIVE_MAX_BYTES": "0", "ORBIT_ARCHIVE_TICK_INTERVAL": "100ms", "ORBIT_ARCHIVE_ROOT": "relative", "ORBIT_ARCHIVE_STORAGE_ID": "not-an-id"} { + old := base[k] + base[k] = v + if _, err = FromEnv(get); err == nil { + t.Fatal("bad archive config", k) + } + base[k] = old + } +} diff --git a/internal/retention/job.go b/internal/retention/job.go index 6807d9a..fbc9985 100644 --- a/internal/retention/job.go +++ b/internal/retention/job.go @@ -16,6 +16,12 @@ import ( core "git.yuntuo.hk/k2safe/orbit/model" ) +// ErrArchiveLimit 表示达到显式容量边界，源数据与已发布文件均保留。 +var ErrArchiveLimit = errors.New("archive byte limit reached; source and saved progress retained") + +// ErrArchiveBusy 表示原作业被另一个进程持锁，不允许并发写入。 +var ErrArchiveBusy = errors.New("archive job is locked") + const maxJobBatches = 10000 const maxBatchFileBytes = 64 << 20 @@ -47,8 +53,11 @@ type JobBatch struct { // JobOptions 控制本次进程的工作量，重试时可显式提高上限，不改变作业身份。 type JobOptions struct { - MaxBatches int // MaxBatches 本次最多新增批次数，1–1000。 - MaxBytes int64 // MaxBytes 作业清单和全部批次文件总字节上限，1 MiB–100 GiB；不自动删除旧文件腾空间。 + JobID core.ID // JobID 周期作业预分配身份；空值保持CLI自动生成行为。 + RequireExisting bool // RequireExisting 接管时必须存在原清单，禁止卷丢失后重建。 + Checkpoint *JobStatus // Checkpoint 数据库已确认的链前缀，校验后才允许继续导出。 + MaxBatches int // MaxBatches 本次最多新增批次数，1–1000。 + MaxBytes int64 // MaxBytes 作业清单和全部批次文件总字节上限，1 MiB–100 GiB；不自动删除旧文件腾空间。 } // JobStatus 是可输出到日志的摘要，不包含业务正文或游标。 @@ -84,7 +93,10 @@ func RunArchive(ctx context.Context, directory string, spec JobSpec, options Job if options.MaxBatches < 1 || options.MaxBatches > 1000 || options.MaxBytes < 1<<20 || options.MaxBytes > 100<<30 || export == nil { return JobStatus{}, errors.New("invalid archive execution limits") } - root, lock, err := openJob(directory, true) + if options.JobID != "" && options.JobID.Validate() != nil { + return JobStatus{}, errors.New("invalid archive job id") + } + root, lock, err := openJob(directory, !options.RequireExisting) if err != nil { return JobStatus{}, err } @@ -94,14 +106,20 @@ func RunArchive(ctx context.Context, directory string, spec JobSpec, options Job return JobStatus{}, err } if _, err := root.Lstat("manifest.json"); errors.Is(err, os.ErrNotExist) { + if options.RequireExisting { + return JobStatus{}, errors.New("saved archive manifest is missing") + } // 不在未知目录或缺少清单的残留批次上悄悄创建新身份。 entries, err := jobEntries(root) if err != nil || len(entries) != 1 || entries[0].Name() != ".lock" { return JobStatus{}, errors.New("archive directory is not empty or manifest is missing") } - id, err := core.NewID() - if err != nil { - return JobStatus{}, err + id := options.JobID + if id == "" { + id, err = core.NewID() + if err != nil { + return JobStatus{}, err + } } manifest := JobManifest{1, id, time.Now().UTC(), spec} raw, err := json.Marshal(manifest) @@ -118,9 +136,23 @@ func RunArchive(ctx context.Context, directory string, spec JobSpec, options Job if err != nil { return status, err } - if manifest.Spec != spec { + if manifest.Spec != spec || (options.JobID != "" && manifest.JobID != options.JobID) { return status, errors.New("archive scope differs from saved manifest; resume with original arguments") } + + if saved := options.Checkpoint; saved != nil && saved.LastSHA256 != "" { + if saved.JobID != status.JobID || saved.Batches > status.Batches || saved.Roots > status.Roots || saved.Bytes > status.Bytes { + return status, errors.New("archive progress is behind database checkpoint") + } + name := "manifest.json" + if saved.Batches > 0 { + name = batchName(saved.Batches) + } + raw, err := readJobFile(root, name, maxBatchFileBytes) + if err != nil || digest(raw) != saved.LastSHA256 { + return status, errors.New("archive prefix differs from database checkpoint") + } + } if err := syncJobDirectory(root); err != nil { return status, err } @@ -145,7 +177,7 @@ func RunArchive(ctx context.Context, directory string, spec JobSpec, options Job return status, err } if len(raw) > maxBatchFileBytes || int64(len(raw)) > options.MaxBytes-status.Bytes { - return status, errors.New("archive byte limit reached; source and saved progress retained") + return status, ErrArchiveLimit } if err := ctx.Err(); err != nil { return status, err @@ -273,7 +305,7 @@ func readJob(ctx context.Context, root *os.Root, maxBytes int64) (JobManifest, J return fail(err) } if int64(len(raw)) > maxBytes-status.Bytes { - return fail(errors.New("archive byte limit reached")) + return fail(ErrArchiveLimit) } var packet JobBatch if decodeJobJSON(raw, &packet) != nil { diff --git a/internal/retention/job_lock_unix.go b/internal/retention/job_lock_unix.go index c08c69e..92e9129 100644 --- a/internal/retention/job_lock_unix.go +++ b/internal/retention/job_lock_unix.go @@ -21,7 +21,10 @@ func lockJob(root *os.Root) (*os.File, error) { } if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { f.Close() - return nil, errors.New("archive job is locked or filesystem does not support locking") + if errors.Is(err, syscall.EWOULDBLOCK) || errors.Is(err, syscall.EAGAIN) { + return nil, ErrArchiveBusy + } + return nil, errors.New("archive filesystem does not support locking") } return f, nil } diff --git a/internal/retention/job_test.go b/internal/retention/job_test.go index 2fb650e..b4728f6 100644 --- a/internal/retention/job_test.go +++ b/internal/retention/job_test.go @@ -259,3 +259,33 @@ func TestArchiveJobWriteFailureDoesNotAdvance(t *testing.T) { t.Fatalf("write recovery=%+v err=%v", final, err) } } + +func TestManagedArchiveChecksDatabasePrefixAndMissingVolume(t *testing.T) { + ctx := context.Background() + dir := filepath.Join(t.TempDir(), "job") + id, _ := core.NewID() + calls := 0 + options := testJobOptions() + options.JobID = id + options.MaxBatches = 1 + first, err := RunArchive(ctx, dir, testJobSpec(), options, testJobExporter(t, &calls)) + if err != nil || first.JobID != id { + t.Fatal(first, err) + } + options.Checkpoint = &first + options.RequireExisting = true + forged := first + forged.LastSHA256 = strings.Repeat("0", 64) + options.Checkpoint = &forged + if _, err = RunArchive(ctx, dir, testJobSpec(), options, testJobExporter(t, &calls)); err == nil || calls != 1 { + t.Fatal("untrusted prefix accepted") + } + options.Checkpoint = &first + if _, err = RunArchive(ctx, filepath.Join(t.TempDir(), "missing"), testJobSpec(), options, testJobExporter(t, &calls)); err == nil || calls != 1 { + t.Fatal("missing volume silently recreated") + } + second, err := RunArchive(ctx, dir, testJobSpec(), options, testJobExporter(t, &calls)) + if err != nil || !second.Completed || second.JobID != id { + t.Fatal(second, err) + } +} diff --git a/internal/retention/runtime.go b/internal/retention/runtime.go new file mode 100644 index 0000000..5180e91 --- /dev/null +++ b/internal/retention/runtime.go @@ -0,0 +1,209 @@ +package retention + +import ( + "context" + "errors" + "io" + "io/fs" + "os" + "path/filepath" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" +) + +// RuntimeStore 只供受信任的归档执行器使用，控制台不能直接领取或确认租约。 +type RuntimeStore interface { + ScheduleArchive(context.Context, core.ID) error + ClaimArchive(context.Context, core.ID) (*ManagedJob, error) + FinishArchive(context.Context, ManagedJob, JobStatus, string) error + ExportRetentionForJob(context.Context, core.ID, core.ID, ArchiveInput) (ArchiveSnapshot, error) +} + +// ArchiveRuntime 固定一个部署卷，每次最多运行一批；所有共享该卷的节点串行占用文件锁。 +type ArchiveRuntime struct { + Directory string // Directory 管理员配置的绝对路径，0700权限，不可使用容器临时层。 + StorageID core.ID // StorageID 卷内标记必须相同；不是共享存储的替代品。 + MaxBytes int64 // MaxBytes 整个卷内归档文件总字节上限，包含未完成作业。 + Store RuntimeStore // Store 持久化计划、领取和检查点。 +} + +// Init 仅在空私有目录建立卷标记，已有文件缺少标记时拒绝猜测身份。 +func (r *ArchiveRuntime) Init() error { + if r.StorageID.Validate() != nil || !filepath.IsAbs(r.Directory) || r.MaxBytes < 1<<20 || r.Store == nil { + return errors.New("invalid archive runtime configuration") + } + + root, lock, err := openJob(r.Directory, true) + for attempt := 0; errors.Is(err, ErrArchiveBusy) && attempt < 20; attempt++ { + // 已有卷身份文件不可变；其他节点正在执行时，启动节点可以只读核对， + // 不必等待整个文件批次释放卷锁。首次发布尚未结束时短暂等待。 + reader, readErr := os.OpenRoot(r.Directory) + if readErr == nil { + marker, e := readJobFile(reader, ".storage-id", 64) + reader.Close() + if e == nil { + if string(marker) != string(r.StorageID) { + return errors.New("archive volume identity mismatch") + } + return nil + } + } + time.Sleep(100 * time.Millisecond) + root, lock, err = openJob(r.Directory, true) + } + if err != nil { + return err + } + defer root.Close() + defer lock.Close() + if err := discardStaging(root); err != nil { + return err + } + data, err := readJobFile(root, ".storage-id", 64) + if errors.Is(err, fs.ErrNotExist) { + entries, e := jobEntries(root) + if e != nil { + return e + } + if len(entries) != 1 || entries[0].Name() != ".lock" { + return errors.New("archive volume identity is missing") + } + return publishFile(root, ".storage-id", []byte(r.StorageID)) + } + if err != nil { + return err + } + if string(data) != string(r.StorageID) { + return errors.New("archive volume identity mismatch") + } + return nil +} + +// Tick 不在数据库事务中进行文件I/O；发布后确认失败时，由同一作业验证文件再收敛。 +func (r *ArchiveRuntime) Tick(ctx context.Context) error { + root, lock, err := openJob(r.Directory, false) + if errors.Is(err, ErrArchiveBusy) { + return nil + } + if err != nil { + return err + } + defer root.Close() + defer lock.Close() + marker, err := readJobFile(root, ".storage-id", 64) + if err != nil || string(marker) != string(r.StorageID) { + return errors.New("archive volume identity mismatch") + } + if err = r.Store.ScheduleArchive(ctx, r.StorageID); err != nil { + return err + } + job, err := r.Store.ClaimArchive(ctx, r.StorageID) + if err != nil || job == nil { + return err + } + name := string(job.NamespaceID) + "-" + string(job.ID) + files := 0 + total, current, err := archiveVolumeUsage(ctx, root, ".", name, &files) + + code := "" + var status JobStatus + if err != nil { + code = "FILE_CHECK_FAILED" + if errors.Is(err, ErrArchiveLimit) { + code = "STORAGE_LIMIT" + } + } else { + budget := min(job.MaxBytes, r.MaxBytes-total+current) + if budget < 1<<20 { + code = "STORAGE_LIMIT" + } else { + checkpoint := JobStatus{JobID: job.ID, Batches: job.Batches, Roots: job.Roots, Bytes: job.Bytes, LastSHA256: job.LastSHA256} + options := JobOptions{JobID: job.ID, RequireExisting: job.Attempts > 1, Checkpoint: &checkpoint, MaxBatches: 1, MaxBytes: budget} + status, err = RunArchive(ctx, filepath.Join(r.Directory, name), JobSpec{job.NamespaceID, job.Kind, job.PolicyRevision, job.BatchSize}, options, func(ctx context.Context, id core.ID, in ArchiveInput) (ArchiveSnapshot, error) { + out, err := r.Store.ExportRetentionForJob(ctx, job.NamespaceID, id, in) + if err != nil { + code = "EXPORT_FAILED" + switch fault.Code(err) { + case "STATE_CONFLICT": + code = "POLICY_CHANGED" + case "PERMISSION_DENIED": + code = "PERMISSION_DENIED" + } + } + return out, err + }) + if err != nil && code == "" { + code = "FILE_CHECK_FAILED" + if errors.Is(err, ErrArchiveLimit) { + code = "STORAGE_LIMIT" + } + if errors.Is(err, ErrArchiveBusy) { + code = "STORAGE_BUSY" + } + } + } + } + // 关闭/超时不写成文件损坏；保留租约，下一节点在到期后从同一文件链接管。 + if ctx.Err() != nil { + return ctx.Err() + } + finish, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + return r.Store.FinishArchive(finish, *job, status, code) +} + +// archiveVolumeUsage 分块读取目录项，避免 WalkDir 在大目录中先一次载入全部文件名。 +// 平台卷只有根目录和单层作业目录，意外嵌套或链接一律拒绝。 +func archiveVolumeUsage(ctx context.Context, root *os.Root, directory, currentJob string, count *int) (total, current int64, err error) { + dir, err := root.Open(directory) + if err != nil { + return 0, 0, err + } + defer dir.Close() + for { + if err := ctx.Err(); err != nil { + return 0, 0, err + } + entries, readErr := dir.ReadDir(256) + if readErr != nil && !errors.Is(readErr, io.EOF) { + return 0, 0, readErr + } + for _, entry := range entries { + *count++ + if *count > 100000 { + return 0, 0, ErrArchiveLimit + } + if entry.Type()&fs.ModeSymlink != 0 { + return 0, 0, errors.New("archive volume contains a symlink") + } + if entry.IsDir() { + if directory != "." { + return 0, 0, errors.New("archive volume contains nested directories") + } + n, j, err := archiveVolumeUsage(ctx, root, entry.Name(), currentJob, count) + if err != nil { + return 0, 0, err + } + total += n + current += j + continue + } + info, err := entry.Info() + if err != nil { + return 0, 0, err + } + if !info.Mode().IsRegular() { + return 0, 0, errors.New("archive volume contains a special file") + } + total += info.Size() + if directory == currentJob { + current += info.Size() + } + } + if errors.Is(readErr, io.EOF) { + return total, current, nil + } + } +} diff --git a/internal/retention/runtime_test.go b/internal/retention/runtime_test.go new file mode 100644 index 0000000..fe62190 --- /dev/null +++ b/internal/retention/runtime_test.go @@ -0,0 +1,96 @@ +package retention + +import ( + "context" + "os" + "path/filepath" + "testing" + + core "git.yuntuo.hk/k2safe/orbit/model" +) + +type idleArchiveStore struct { + RuntimeStore + scheduled int +} + +func (s *idleArchiveStore) ScheduleArchive(context.Context, core.ID) error { s.scheduled++; return nil } +func (s *idleArchiveStore) ClaimArchive(context.Context, core.ID) (*ManagedJob, error) { + return nil, nil +} + +func TestArchiveVolumeIdentityLockAndInitializationRecovery(t *testing.T) { + store := &idleArchiveStore{} + id, _ := core.NewID() + runtime := &ArchiveRuntime{Directory: filepath.Join(t.TempDir(), "volume"), StorageID: id, MaxBytes: 8 << 20, Store: store} + if err := runtime.Init(); err != nil { + t.Fatal(err) + } + root, lock, err := openJob(runtime.Directory, false) + if err != nil { + t.Fatal(err) + } + // 其他节点正在执行时，新节点仍可读取不可变卷身份并正常启动。 + if err = runtime.Init(); err != nil { + t.Fatal("another node could not initialize", err) + } + other := *runtime + other.StorageID, _ = core.NewID() + if err = other.Init(); err == nil { + t.Fatal("wrong volume accepted while busy") + } + if err = runtime.Tick(context.Background()); err != nil || store.scheduled != 0 { + t.Fatal("busy volume scheduled work", err) + } + lock.Close() + root.Close() + if err = runtime.Tick(context.Background()); err != nil || store.scheduled != 1 { + t.Fatal("unlocked volume did not schedule", err) + } + if err = other.Init(); err == nil { + t.Fatal("wrong volume accepted") + } + // 首次身份文件原子发布前崩溃，只允许丢弃本工具的安全临时文件。 + fresh := *runtime + fresh.Directory = filepath.Join(t.TempDir(), "interrupted") + if err = os.Mkdir(fresh.Directory, 0700); err != nil { + t.Fatal(err) + } + if err = os.WriteFile(filepath.Join(fresh.Directory, ".write.tmp"), []byte(id), 0600); err != nil { + t.Fatal(err) + } + if err = fresh.Init(); err != nil { + t.Fatal("volume init did not recover", err) + } +} + +func TestArchiveVolumeScanCountsAllJobsAndRejectsLinks(t *testing.T) { + dir := t.TempDir() + job := "job" + if err := os.Mkdir(filepath.Join(dir, job), 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "other"), []byte("volume"), 0600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, job, "batch"), []byte("job"), 0600); err != nil { + t.Fatal(err) + } + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer root.Close() + count := 0 + total, current, err := archiveVolumeUsage(context.Background(), root, ".", job, &count) + if err != nil || total != 9 || current != 3 || count != 3 { + t.Fatal(total, current, count, err) + } + if err = os.Symlink("other", filepath.Join(dir, "unexpected")); err != nil { + t.Fatal(err) + } + count = 0 + if _, _, err = archiveVolumeUsage(context.Background(), root, ".", job, &count); err == nil { + t.Fatal("volume symlink accepted") + } +} diff --git a/internal/retention/scheduler.go b/internal/retention/scheduler.go new file mode 100644 index 0000000..803ca38 --- /dev/null +++ b/internal/retention/scheduler.go @@ -0,0 +1,101 @@ +package retention + +import ( + "context" + "errors" + "strings" + "time" + "unicode/utf8" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +// ArchiveKinds 是周期归档支持的固定类别，不接受任意表名或 SQL。 +var ArchiveKinds = [...]string{"tasks", "workflows", "events", "audit", "deliveries", "inbox", "idempotency"} + +// ArchivePlan 是命名空间内单一类别的周期计划；失败作业阻止本类别再生成新作业。 +type ArchivePlan struct { + NamespaceID core.ID `json:"namespace_id"` // NamespaceID 由登录会话确定。 + Kind string `json:"kind"` // Kind 同一命名空间内唯一的归档类别。 + Revision uint64 `json:"revision,string"` // Revision 配置版本；尚未保存时为0，后台触发不改变配置版本。 + Enabled bool `json:"enabled"` // Enabled 仅控制新作业触发，关闭不会中断已有作业。 + PolicyRevision uint64 `json:"policy_revision,string"` // PolicyRevision 管理员确认的保留策略版本，变更策略后须重新保存计划。 + StorageID core.ID `json:"storage_id"` // StorageID 固定归档卷身份，由部署配置提供，不接受浏览器任意路径。 + IntervalSeconds uint32 `json:"interval_seconds"` // IntervalSeconds 周期秒数，3600至2592000；错过多次只合并触发一次。 + BatchSize int `json:"batch_size"` // BatchSize 单批根记录数，1至100。 + MaxBytes int64 `json:"max_bytes"` // MaxBytes 每个作业的文件字节上限，1 MiB至1 GiB。 + NextRunAt *time.Time `json:"next_run_at"` // NextRunAt 下次触发UTC时间，未保存时为空；首次启用立即到期。 + ActiveJobID *core.ID `json:"active_job_id"` // ActiveJobID 未完成或待处理失败作业；成功或人工关闭后清空。 + UpdatedAt *time.Time `json:"updated_at"` // UpdatedAt 最后配置修改UTC时间，未保存为空。 +} + +// PlanInput 完整替换计划配置；保存与管理审计、幂等回执同事务。 +type PlanInput struct { + ExpectedRevision uint64 `json:"expected_revision,string" required:"true"` // ExpectedRevision 已读取的计划版本。 + PolicyRevision uint64 `json:"policy_revision,string" required:"true"` // PolicyRevision 已确认的保留策略版本。 + Enabled bool `json:"enabled" required:"true"` // Enabled 是否开始周期触发。 + IntervalSeconds uint32 `json:"interval_seconds" required:"true"` // IntervalSeconds 周期秒数。 + BatchSize int `json:"batch_size" required:"true"` // BatchSize 每批根记录数。 + MaxBytes int64 `json:"max_bytes" required:"true"` // MaxBytes 单作业总文件预算，字节。 + Reason string `json:"reason" required:"true"` // Reason 单行操作原因，最多1024个UTF-8字节。 +} + +// ManagedJob 保存周期作业权威状态；文件内容和游标不通过管理列表返回。 +type ManagedJob struct { + ID core.ID `json:"id"` // ID 与目录名及文件清单作业ID一致。 + NamespaceID core.ID `json:"namespace_id"` // NamespaceID 所属命名空间。 + Kind string `json:"kind"` // Kind 固定归档类别。 + PlanRevision uint64 `json:"plan_revision,string"` // PlanRevision 创建时固定的计划配置版本。 + PolicyRevision uint64 `json:"policy_revision,string"` // PolicyRevision 作业期间不可替换的保留策略版本。 + StorageID core.ID `json:"storage_id"` // StorageID 固定归档卷身份，接管节点必须挂载同一物理卷。 + BatchSize int `json:"batch_size"` // BatchSize 创建时固定的单批根记录上限。 + MaxBytes int64 `json:"max_bytes"` // MaxBytes 累计字节预算，仅失败重试时允许提高。 + State string `json:"state"` // State 为PENDING/RUNNING/SUCCEEDED/FAILED/ABANDONED。 + Revision uint64 `json:"revision,string"` // Revision 每次领取、确认、人工操作递增，用于防止迟到覆盖。 + Attempts uint32 `json:"attempts"` // Attempts 累计批次领取次数，包含崩溃接管，不等于业务任务重试次数。 + Batches int `json:"batches"` // Batches 已验证并入库的连续批次数。 + Roots int64 `json:"roots"` // Roots 已验证根记录总数。 + Bytes int64 `json:"bytes"` // Bytes 已验证清单和批次总字节数。 + LastSHA256 string `json:"last_sha256"` // LastSHA256 数据库确认的文件链尾摘要；恢复必须验证此前缀。 + ErrorCode string `json:"error_code"` // ErrorCode 固定错误分类，不保存原始路径、业务正文或数据库错误。 + LeaseID core.ID `json:"-"` // LeaseID 当前领取随机代次，不向浏览器返回。 + LeaseUntil *time.Time `json:"lease_until"` // LeaseUntil UTC租约截止，非执行中为空；过期可接管。 + CreatedAt time.Time `json:"created_at"` // CreatedAt 作业受理UTC时间。 + UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最近状态变更UTC时间。 +} + +// JobActionInput 是失败作业的人工恢复或关闭输入，必须留有原因。 +type JobActionInput struct { + ExpectedRevision uint64 `json:"expected_revision,string" required:"true"` // ExpectedRevision 已读取的作业版本。 + Action string `json:"action" required:"true"` // Action 仅retry或abandon；后者保留已有文件与审计。 + MaxBytes int64 `json:"max_bytes" required:"true"` // MaxBytes 重试可提高预算，关闭时保持原值。 + Reason string `json:"reason" required:"true"` // Reason 人工操作原因。 +} + +// JobPage 采用已返回作业ID作为翻页位置；位置必须属于当前命名空间。 +type JobPage struct { + Items []ManagedJob `json:"items"` // Items 本页最多20个作业。 + NextBefore string `json:"next_before"` // NextBefore 下一页位置；空表示当前已到末页。 +} + +// SchedulerStore 将控制台管理与后台执行的持久事务统一到存储层。 +type SchedulerStore interface { + ListArchivePlans(context.Context, auth.Token) ([]ArchivePlan, error) + SaveArchivePlan(context.Context, auth.Token, string, string, core.ID, PlanInput) (ArchivePlan, error) + ListArchiveJobs(context.Context, auth.Token, string) (JobPage, error) + ActArchiveJob(context.Context, auth.Token, core.ID, string, JobActionInput) (ManagedJob, error) +} + +func ValidReason(s string) bool { + return strings.TrimSpace(s) != "" && utf8.ValidString(s) && len(s) <= 1024 && !strings.ContainsAny(s, "\r\n") +} +func (in PlanInput) Validate(namespace core.ID, kind string) error { + if err := (JobSpec{namespace, kind, in.PolicyRevision, in.BatchSize}).Validate(); err != nil { + return err + } + if in.IntervalSeconds < 3600 || in.IntervalSeconds > 2592000 || in.MaxBytes < 1<<20 || in.MaxBytes > 1<<30 || !ValidReason(in.Reason) { + return errors.New("invalid archive interval, byte budget or reason") + } + return nil +} diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index f2e3b69..339201c 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -94,7 +94,7 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { f.exec(`ALTER TABLE pulse_worker_sessions DROP COLUMN concurrency_keys`) // 新增的控制台和保留策略迁移也要还原到历史 v9，避免升级测试把 // 已经存在的 v29/v30 表误当成迁移重复执行。 - f.exec(`DROP TABLE orbit_retention_policies,orbit_console_sessions,orbit_console_users`) + f.exec(`DROP TABLE orbit_archive_jobs,orbit_archive_plans,orbit_retention_policies,orbit_console_sessions,orbit_console_users`) f.exec(`DELETE FROM orbit_schema_migrations WHERE version>=10`) if err := Migrate(ctx, db); err != nil { t.Fatal(err) diff --git a/internal/storage/mysql/archive_scheduler.go b/internal/storage/mysql/archive_scheduler.go new file mode 100644 index 0000000..a2d6ef3 --- /dev/null +++ b/internal/storage/mysql/archive_scheduler.go @@ -0,0 +1,441 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "strings" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/retention" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +const archivePlanColumns = `namespace_id,kind,revision,enabled,policy_revision,storage_id,interval_seconds,batch_size,max_bytes,next_run_at,active_job_id,updated_at` +const archiveJobColumns = `namespace_id,id,kind,plan_revision,policy_revision,storage_id,batch_size,max_bytes,state,revision,attempts,batches,roots,bytes,last_sha256,error_code,lease_id,lease_until,created_at,updated_at` + +type archiveScanner interface{ Scan(...any) error } + +func scanArchivePlan(row archiveScanner) (p retention.ArchivePlan, err error) { + err = row.Scan(&p.NamespaceID, &p.Kind, &p.Revision, &p.Enabled, &p.PolicyRevision, &p.StorageID, &p.IntervalSeconds, &p.BatchSize, &p.MaxBytes, &p.NextRunAt, &p.ActiveJobID, &p.UpdatedAt) + return +} +func scanArchiveJob(row archiveScanner) (j retention.ManagedJob, err error) { + var lease *core.ID + err = row.Scan(&j.NamespaceID, &j.ID, &j.Kind, &j.PlanRevision, &j.PolicyRevision, &j.StorageID, &j.BatchSize, &j.MaxBytes, &j.State, &j.Revision, &j.Attempts, &j.Batches, &j.Roots, &j.Bytes, &j.LastSHA256, &j.ErrorCode, &lease, &j.LeaseUntil, &j.CreatedAt, &j.UpdatedAt) + if lease != nil { + j.LeaseID = *lease + } + return +} +func readArchivePlan(ctx context.Context, tx *sql.Tx, ns core.ID, kind string) (retention.ArchivePlan, error) { + p, err := scanArchivePlan(tx.QueryRowContext(ctx, `SELECT `+archivePlanColumns+` FROM orbit_archive_plans WHERE namespace_id=? AND kind=? FOR UPDATE`, ns, kind)) + if errors.Is(err, sql.ErrNoRows) { + return retention.ArchivePlan{NamespaceID: ns, Kind: kind, IntervalSeconds: 86400, BatchSize: 20, MaxBytes: 256 << 20}, nil + } + return p, err +} +func readArchiveJob(ctx context.Context, tx *sql.Tx, ns, id core.ID) (retention.ManagedJob, error) { + j, err := scanArchiveJob(tx.QueryRowContext(ctx, `SELECT `+archiveJobColumns+` FROM orbit_archive_jobs WHERE namespace_id=? AND id=? FOR UPDATE`, ns, id)) + if errors.Is(err, sql.ErrNoRows) { + err = fault.NotFound + } + return j, err +} +func archiveAudit(ctx context.Context, tx *sql.Tx, ns, resource, op core.ID, actor, action, kind, reason string, before, after any) error { + id, err := core.NewID() + if err != nil { + return err + } + b, err := json.Marshal(before) + if err != nil { + return err + } + a, err := json.Marshal(after) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,`before`,`after`) VALUES (?,?,?,?,?,?,?,?,?,?)", ns, id, op, actor, action, kind, resource, reason, string(b), string(a)) + return err +} +func archiveKey(key string) error { + if strings.TrimSpace(key) == "" || len(key) > 128 { + return fault.Invalid("idempotency key must be 1-128 bytes") + } + return nil +} + +func (s *RetentionPolicyStore) ListArchivePlans(ctx context.Context, t auth.Token) (out []retention.ArchivePlan, err error) { + err = WithTx(ctx, s.db, func(tx *sql.Tx) error { + if _, err := retentionActor(ctx, tx, t); err != nil { + return err + } + rows, err := tx.QueryContext(ctx, `SELECT `+archivePlanColumns+` FROM orbit_archive_plans WHERE namespace_id=? ORDER BY kind`, t.NamespaceID) + if err != nil { + return err + } + defer rows.Close() + found := map[string]retention.ArchivePlan{} + for rows.Next() { + p, err := scanArchivePlan(rows) + if err != nil { + return err + } + found[p.Kind] = p + } + if err := rows.Err(); err != nil { + return err + } + out = []retention.ArchivePlan{} + for _, kind := range retention.ArchiveKinds { + p, ok := found[kind] + if !ok { + p = retention.ArchivePlan{NamespaceID: t.NamespaceID, Kind: kind, IntervalSeconds: 86400, BatchSize: 20, MaxBytes: 256 << 20} + } + out = append(out, p) + } + return nil + }) + return +} +func (s *RetentionPolicyStore) SaveArchivePlan(ctx context.Context, t auth.Token, key, kind string, storage core.ID, in retention.PlanInput) (out retention.ArchivePlan, err error) { + if err = in.Validate(t.NamespaceID, kind); err != nil { + return out, fault.Invalid(err.Error()) + } + if err = archiveKey(key); err != nil { + return + } + // 卷身份属于幂等请求摘要；跨不同卷的节点不能把同一保存解释成不同目标。 + input := struct { + Kind string + Storage core.ID + Input retention.PlanInput + }{kind, storage, in} + err = WithTx(ctx, s.db, func(tx *sql.Tx) error { + user, err := retentionActor(ctx, tx, t) + if err != nil { + return err + } + identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "retention.plan.update", kind, key, "archive_plan", t.NamespaceID, input) + if err != nil { + return err + } + if cached != nil { + return json.Unmarshal(cached, &out) + } + policy, err := s.readPolicy(ctx, tx, t.NamespaceID, true) + if err != nil { + return err + } + if policy.Revision != in.PolicyRevision || (in.Enabled && !policy.Enabled) { + return fault.Conflict + } + before, err := readArchivePlan(ctx, tx, t.NamespaceID, kind) + if err != nil { + return err + } + if before.Revision != in.ExpectedRevision { + return fault.Conflict + } + target := storage + if target == "" && !in.Enabled { + target = before.StorageID + } + if target.Validate() != nil { + return fault.Invalid("archive storage is not configured") + } + if before.ActiveJobID != nil && before.StorageID != target { + return fault.Conflict + } + if before.Revision == 0 { + _, err = tx.ExecContext(ctx, `INSERT INTO orbit_archive_plans(namespace_id,kind,enabled,policy_revision,storage_id,interval_seconds,batch_size,max_bytes,next_run_at) VALUES(?,?,?,?,?,?,?,?,UTC_TIMESTAMP(6))`, t.NamespaceID, kind, in.Enabled, in.PolicyRevision, target, in.IntervalSeconds, in.BatchSize, in.MaxBytes) + } else { + _, err = tx.ExecContext(ctx, `UPDATE orbit_archive_plans SET revision=revision+1,next_run_at=IF(enabled=FALSE AND ?=TRUE,UTC_TIMESTAMP(6),next_run_at),enabled=?,policy_revision=?,storage_id=?,interval_seconds=?,batch_size=?,max_bytes=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND kind=?`, in.Enabled, in.Enabled, in.PolicyRevision, target, in.IntervalSeconds, in.BatchSize, in.MaxBytes, t.NamespaceID, kind) + } + if err != nil { + return err + } + out, err = readArchivePlan(ctx, tx, t.NamespaceID, kind) + if err != nil { + return err + } + if err = archiveAudit(ctx, tx, t.NamespaceID, t.NamespaceID, identity, "console:"+string(user), "retention.plan.update", "archive_plan", in.Reason, before, out); err != nil { + return err + } + return saveIdentity(ctx, tx, t.NamespaceID, identity, out) + }) + return +} +func (s *RetentionPolicyStore) ListArchiveJobs(ctx context.Context, t auth.Token, before string) (out retention.JobPage, err error) { + if before != "" && core.ID(before).Validate() != nil { + return out, fault.Invalid("invalid archive page position") + } + err = WithTx(ctx, s.db, func(tx *sql.Tx) error { + if _, err := retentionActor(ctx, tx, t); err != nil { + return err + } + q := `SELECT ` + archiveJobColumns + ` FROM orbit_archive_jobs WHERE namespace_id=?` + args := []any{t.NamespaceID} + if before != "" { + var at time.Time + if err := tx.QueryRowContext(ctx, `SELECT created_at FROM orbit_archive_jobs WHERE namespace_id=? AND id=?`, t.NamespaceID, core.ID(before)).Scan(&at); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } + return err + } + q += ` AND (created_at 20 { + out.Items = out.Items[:20] + out.NextBefore = string(out.Items[19].ID) + } + return nil + }) + return +} +func (s *RetentionPolicyStore) ActArchiveJob(ctx context.Context, t auth.Token, id core.ID, key string, in retention.JobActionInput) (out retention.ManagedJob, err error) { + if id.Validate() != nil || (in.Action != "retry" && in.Action != "abandon") || !retention.ValidReason(in.Reason) || in.MaxBytes < 1<<20 || in.MaxBytes > 1<<30 { + return out, fault.Invalid("invalid archive action, budget or reason") + } + if err = archiveKey(key); err != nil { + return + } + err = WithTx(ctx, s.db, func(tx *sql.Tx) error { + user, err := retentionActor(ctx, tx, t) + if err != nil { + return err + } + identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "retention.job.action", string(id), key, "archive_job", id, in) + if err != nil { + return err + } + if cached != nil { + return json.Unmarshal(cached, &out) + } + var kind string + if err := tx.QueryRowContext(ctx, `SELECT kind FROM orbit_archive_jobs WHERE namespace_id=? AND id=?`, t.NamespaceID, id).Scan(&kind); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } + return err + } + p, err := readArchivePlan(ctx, tx, t.NamespaceID, kind) + if err != nil { + return err + } + j, err := readArchiveJob(ctx, tx, t.NamespaceID, id) + if err != nil { + return err + } + if p.ActiveJobID == nil || *p.ActiveJobID != id || j.State != "FAILED" || j.Revision != in.ExpectedRevision || in.MaxBytes < j.MaxBytes { + return fault.Conflict + } + next := "PENDING" + if in.Action == "abandon" { + next = "ABANDONED" + if in.MaxBytes != j.MaxBytes { + return fault.Conflict + } + } + _, err = tx.ExecContext(ctx, `UPDATE orbit_archive_jobs SET state=?,revision=revision+1,max_bytes=?,error_code='',updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, next, in.MaxBytes, t.NamespaceID, id) + if err != nil { + return err + } + if next == "ABANDONED" { + if _, err = tx.ExecContext(ctx, `UPDATE orbit_archive_plans SET active_job_id=NULL WHERE namespace_id=? AND kind=? AND active_job_id=?`, t.NamespaceID, kind, id); err != nil { + return err + } + } + out, err = readArchiveJob(ctx, tx, t.NamespaceID, id) + if err != nil { + return err + } + if err = archiveAudit(ctx, tx, t.NamespaceID, id, identity, "console:"+string(user), "retention.job."+in.Action, "archive_job", in.Reason, j, out); err != nil { + return err + } + return saveIdentity(ctx, tx, t.NamespaceID, identity, out) + }) + return +} + +// ScheduleArchive 在数据库时钟下合并错过的时间槽；策略、计划和新作业同事务核验。 +func (s *RetentionPolicyStore) ScheduleArchive(ctx context.Context, storage core.ID) error { + if storage.Validate() != nil { + return fault.Invalid("invalid storage id") + } + rows, err := s.db.QueryContext(ctx, `SELECT p.namespace_id,p.kind FROM orbit_archive_plans p JOIN orbit_retention_policies r ON r.namespace_id=p.namespace_id AND r.revision=p.policy_revision AND r.enabled=TRUE JOIN orbit_namespaces n ON n.id=p.namespace_id AND n.enabled=TRUE JOIN orbit_applications a ON a.id=n.application_id AND a.enabled=TRUE WHERE p.storage_id=? AND p.enabled=TRUE AND p.active_job_id IS NULL AND p.next_run_at<=UTC_TIMESTAMP(6) ORDER BY p.next_run_at LIMIT 16`, storage) + if err != nil { + return err + } + type candidate struct { + ns core.ID + kind string + } + var candidates []candidate + for rows.Next() { + var c candidate + if err = rows.Scan(&c.ns, &c.kind); err != nil { + rows.Close() + return err + } + candidates = append(candidates, c) + } + err = rows.Err() + rows.Close() + if err != nil { + return err + } + for _, c := range candidates { + err = WithTx(ctx, s.db, func(tx *sql.Tx) error { + var enabled bool + if err := tx.QueryRowContext(ctx, `SELECT a.enabled AND n.enabled FROM orbit_namespaces n JOIN orbit_applications a ON a.id=n.application_id WHERE n.id=? FOR SHARE`, c.ns).Scan(&enabled); err != nil { + return err + } + if !enabled { + return nil + } + policy, err := s.readPolicy(ctx, tx, c.ns, true) + if err != nil { + return err + } + p, err := readArchivePlan(ctx, tx, c.ns, c.kind) + if err != nil { + return err + } + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + if !p.Enabled || p.ActiveJobID != nil || p.StorageID != storage || p.NextRunAt == nil || p.NextRunAt.After(now) || !policy.Enabled || policy.Revision != p.PolicyRevision { + return nil + } + id, err := core.NewID() + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO orbit_archive_jobs(namespace_id,id,kind,plan_revision,policy_revision,storage_id,batch_size,max_bytes) VALUES(?,?,?,?,?,?,?,?)`, c.ns, id, c.kind, p.Revision, p.PolicyRevision, storage, p.BatchSize, p.MaxBytes) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `UPDATE orbit_archive_plans SET active_job_id=?,next_run_at=? WHERE namespace_id=? AND kind=?`, id, now.Add(time.Duration(p.IntervalSeconds)*time.Second), c.ns, c.kind) + if err != nil { + return err + } + j, err := readArchiveJob(ctx, tx, c.ns, id) + if err != nil { + return err + } + return archiveAudit(ctx, tx, c.ns, id, id, "archive:scheduler", "retention.job.created", "archive_job", "周期归档到期", nil, j) + }) + if err != nil { + return err + } + } + return nil +} + +// ClaimArchive 领取一批工作；五分钟租约超时后重领，以随机代次隔离迟到确认。 +func (s *RetentionPolicyStore) ClaimArchive(ctx context.Context, storage core.ID) (job *retention.ManagedJob, err error) { + err = WithTx(ctx, s.db, func(tx *sql.Tx) error { + job = nil + j, err := scanArchiveJob(tx.QueryRowContext(ctx, `SELECT `+archiveJobColumns+` FROM orbit_archive_jobs WHERE storage_id=? AND (state='PENDING' OR (state='RUNNING' AND lease_until<=UTC_TIMESTAMP(6))) ORDER BY updated_at,id LIMIT 1 FOR UPDATE SKIP LOCKED`, storage)) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return err + } + lease, err := core.NewID() + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `UPDATE orbit_archive_jobs SET state='RUNNING',revision=revision+1,attempts=attempts+1,lease_id=?,lease_until=TIMESTAMPADD(SECOND,300,UTC_TIMESTAMP(6)),updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, lease, j.NamespaceID, j.ID) + if err != nil { + return err + } + claimed, err := readArchiveJob(ctx, tx, j.NamespaceID, j.ID) + if err != nil { + return err + } + job = &claimed + return nil + }) + return +} + +// FinishArchive 仅确认当前未过期领取；失败不释放计划占位，避免不断生成重复失败作业。 +func (s *RetentionPolicyStore) FinishArchive(ctx context.Context, claim retention.ManagedJob, status retention.JobStatus, code string) error { + switch code { + case "", "FILE_CHECK_FAILED", "STORAGE_LIMIT", "STORAGE_BUSY", "POLICY_CHANGED", "PERMISSION_DENIED", "EXPORT_FAILED", "INTERRUPTED": + default: + return fault.Invalid("invalid archive result code") + } + return WithTx(ctx, s.db, func(tx *sql.Tx) error { + p, err := readArchivePlan(ctx, tx, claim.NamespaceID, claim.Kind) + if err != nil { + return err + } + j, err := readArchiveJob(ctx, tx, claim.NamespaceID, claim.ID) + if err != nil { + return err + } + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + if p.ActiveJobID == nil || *p.ActiveJobID != j.ID || j.State != "RUNNING" || j.LeaseID != claim.LeaseID || j.Revision != claim.Revision || j.LeaseUntil == nil || !j.LeaseUntil.After(now) { + return fault.Conflict + } + next := "PENDING" + if code != "" { + next = "FAILED" + } else if status.Completed { + next = "SUCCEEDED" + } + if code == "" && (status.JobID != j.ID || status.Batches < j.Batches || status.Roots < j.Roots || status.Bytes < j.Bytes || len(status.LastSHA256) != 64) { + return fault.Conflict + } + // 校验失败的部分文件不能覆盖最后一次数据库确认的可信检查点。 + if code != "" { + status = retention.JobStatus{JobID: j.ID, Batches: j.Batches, Roots: j.Roots, Bytes: j.Bytes, LastSHA256: j.LastSHA256} + } + _, err = tx.ExecContext(ctx, `UPDATE orbit_archive_jobs SET state=?,revision=revision+1,batches=?,roots=?,bytes=?,last_sha256=?,error_code=?,lease_id=NULL,lease_until=NULL,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, next, status.Batches, status.Roots, status.Bytes, status.LastSHA256, code, j.NamespaceID, j.ID) + if err != nil { + return err + } + if next == "SUCCEEDED" { + if _, err = tx.ExecContext(ctx, `UPDATE orbit_archive_plans SET active_job_id=NULL WHERE namespace_id=? AND kind=? AND active_job_id=?`, j.NamespaceID, j.Kind, j.ID); err != nil { + return err + } + } + if next == "PENDING" { + return nil + } + out, err := readArchiveJob(ctx, tx, j.NamespaceID, j.ID) + if err != nil { + return err + } + return archiveAudit(ctx, tx, j.NamespaceID, j.ID, j.LeaseID, "archive:"+string(j.ID), "retention.job."+strings.ToLower(next), "archive_job", "归档执行结果", j, out) + }) +} diff --git a/internal/storage/mysql/archive_scheduler_integration_test.go b/internal/storage/mysql/archive_scheduler_integration_test.go new file mode 100644 index 0000000..1b5a4a4 --- /dev/null +++ b/internal/storage/mysql/archive_scheduler_integration_test.go @@ -0,0 +1,380 @@ +package mysql + +import ( + "context" + "database/sql" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/retention" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +func archiveFixture(t *testing.T, payloadBytes ...int) (context.Context, *sql.DB, *RetentionPolicyStore, auth.Token, core.ID) { + t.Helper() + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + t.Cleanup(cancel) + db, err := Open(ctx, dsn, 12, 4) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + app := testID(t) + username := "archive-" + string(app) + var boot BootstrapResult + if err = Bootstrap(ctx, db, BootstrapInput{Application: string(app), Namespace: "archive", Environment: "test", Queue: "archive", ConsoleUsername: username, ConsolePassword: "test-only-strong-password-123"}, func(b BootstrapResult) error { boot = b; return nil }); err != nil { + t.Fatal(err) + } + login, err := NewConsoleAuthStore(db).Login(ctx, username, "test-only-strong-password-123", "archive-test", "127.0.0.1") + if err != nil { + t.Fatal(err) + } + token, err := auth.Parse(login.Token) + if err != nil { + t.Fatal(err) + } + store := NewRetentionPolicyStore(db) + _, err = store.UpdateRetentionPolicy(ctx, token, "initial", retention.UpdateInput{Enabled: true, IdempotencyDays: 90, TaskHistoryDays: 1, WorkflowHistoryDays: 1, EventLogDays: 1, DeliveryHistoryDays: 1, InboxHistoryDays: 1, AuditLogDays: 1, Reason: "isolated archive test"}) + if err != nil { + t.Fatal(err) + } + // 三条过期审计作为真实归档数据，分两批读取，避免凭空伪造任务生命周期。 + size := 0 + if len(payloadBytes) > 0 { + size = payloadBytes[0] + } + for i := 0; i < 3; i++ { + _, err = db.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,`before`,created_at) VALUES(?,?,?,'test','fixture','archive_job',?,'source retained',JSON_OBJECT('fixture',REPEAT('x',?)),TIMESTAMPADD(DAY,-3,UTC_TIMESTAMP(6)))", boot.NamespaceID, testID(t), testID(t), testID(t), size) + if err != nil { + t.Fatal(err) + } + } + return ctx, db, store, token, testID(t) +} +func archivePlanInput() retention.PlanInput { + return retention.PlanInput{PolicyRevision: 1, Enabled: true, IntervalSeconds: 3600, BatchSize: 2, MaxBytes: 4 << 20, Reason: "enable isolated schedule"} +} +func mustArchiveJobs(t *testing.T, ctx context.Context, s *RetentionPolicyStore, token auth.Token) []retention.ManagedJob { + t.Helper() + p, err := s.ListArchiveJobs(ctx, token, "") + if err != nil { + t.Fatal(err) + } + return p.Items +} + +func TestIntegrationArchiveSchedulerRaceResumeAndFencing(t *testing.T) { + ctx, db, s, token, volume := archiveFixture(t) + in := archivePlanInput() + p, err := s.SaveArchivePlan(ctx, token, "plan", "audit", volume, in) + if err != nil { + t.Fatal(err) + } + replay, err := s.SaveArchivePlan(ctx, token, "plan", "audit", volume, in) + if err != nil || replay.Revision != p.Revision { + t.Fatal("plan replay", err) + } + if _, err = s.SaveArchivePlan(ctx, token, "stale", "audit", volume, in); fault.Code(err) != "STATE_CONFLICT" { + t.Fatal("stale plan accepted", err) + } + var wg sync.WaitGroup + errs := make(chan error, 8) + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { defer wg.Done(); errs <- NewRetentionPolicyStore(db).ScheduleArchive(ctx, volume) }() + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + jobs := mustArchiveJobs(t, ctx, s, token) + if len(jobs) != 1 || jobs[0].PlanRevision != p.Revision { + t.Fatalf("duplicate schedules: %+v", jobs) + } + claims := make(chan *retention.ManagedJob, 8) + errs = make(chan error, 8) + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { defer wg.Done(); j, e := s.ClaimArchive(ctx, volume); claims <- j; errs <- e }() + } + wg.Wait() + close(claims) + close(errs) + for e := range errs { + if e != nil { + t.Fatal(e) + } + } + var old *retention.ManagedJob + for j := range claims { + if j != nil { + if old != nil { + t.Fatal("duplicate claim") + } + old = j + } + } + if old == nil { + t.Fatal("no claim") + } + root := filepath.Join(t.TempDir(), "volume") + runtime := &retention.ArchiveRuntime{Directory: root, StorageID: volume, MaxBytes: 16 << 20, Store: s} + if err = runtime.Init(); err != nil { + t.Fatal(err) + } + dir := filepath.Join(root, string(token.NamespaceID)+"-"+string(old.ID)) + spec := retention.JobSpec{NamespaceID: token.NamespaceID, Kind: "audit", PolicyRevision: 1, BatchSize: 2} + first, err := retention.RunArchive(ctx, dir, spec, retention.JobOptions{JobID: old.ID, MaxBatches: 1, MaxBytes: 4 << 20}, func(ctx context.Context, id core.ID, in retention.ArchiveInput) (retention.ArchiveSnapshot, error) { + return s.ExportRetentionForJob(ctx, token.NamespaceID, id, in) + }) + if err != nil || first.Batches != 1 || first.Roots != 2 || first.Completed { + t.Fatal("first batch", first, err) + } + // 模拟文件已落盘但数据库确认尚未提交；租约到期后另一节点恢复同一作业。 + if _, err = db.ExecContext(ctx, `UPDATE orbit_archive_jobs SET lease_until=TIMESTAMPADD(SECOND,-1,UTC_TIMESTAMP(6)) WHERE namespace_id=? AND id=?`, token.NamespaceID, old.ID); err != nil { + t.Fatal(err) + } + newer, err := s.ClaimArchive(ctx, volume) + if err != nil || newer == nil || newer.LeaseID == old.LeaseID { + t.Fatal("reclaim", err) + } + if err = s.FinishArchive(ctx, *old, first, ""); fault.Code(err) != "STATE_CONFLICT" { + t.Fatal("stale claimant advanced", err) + } + if err = s.FinishArchive(ctx, *newer, first, ""); err != nil { + t.Fatal(err) + } + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + jobs = mustArchiveJobs(t, ctx, s, token) + if len(jobs) != 1 || jobs[0].ID != old.ID || jobs[0].State != "SUCCEEDED" || jobs[0].Roots != 3 || jobs[0].Batches != 2 { + t.Fatalf("resume result: %+v", jobs) + } + verified, err := retention.VerifyArchive(ctx, dir) + if err != nil || verified.LastSHA256 != jobs[0].LastSHA256 || !verified.Completed { + t.Fatal("offline mismatch", err) + } + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + if len(mustArchiveJobs(t, ctx, s, token)) != 1 { + t.Fatal("replayed due slot") + } + var sources, fks int + if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND action='fixture'`, token.NamespaceID).Scan(&sources); err != nil || sources != 3 { + t.Fatal("source deleted", err) + } + if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&fks); err != nil || fks != 0 { + t.Fatal("foreign keys introduced", err) + } + t.Log("8 schedulers -> 1 job; 8 claimants -> 1 lease; durable file prefix survives missing DB confirmation; old lease rejected; 3 source rows retained") +} + +func TestIntegrationArchiveFailureRetryPolicyAndScope(t *testing.T) { + ctx, db, s, token, volume := archiveFixture(t) + in := archivePlanInput() + if _, err := s.SaveArchivePlan(ctx, token, "plan", "audit", volume, in); err != nil { + t.Fatal(err) + } + runtime := &retention.ArchiveRuntime{Directory: filepath.Join(t.TempDir(), "volume"), StorageID: volume, MaxBytes: 16 << 20, Store: s} + if err := runtime.Init(); err != nil { + t.Fatal(err) + } + if err := runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + j := mustArchiveJobs(t, ctx, s, token)[0] + if j.State != "PENDING" || j.Batches != 1 { + t.Fatalf("first chunk %+v", j) + } + dir := filepath.Join(runtime.Directory, string(token.NamespaceID)+"-"+string(j.ID)) + file := filepath.Join(dir, "batch-000001.json") + saved, err := os.ReadFile(file) + if err != nil { + t.Fatal(err) + } + if err = os.WriteFile(file, []byte("broken"), 0600); err != nil { + t.Fatal(err) + } + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + failed := mustArchiveJobs(t, ctx, s, token)[0] + if failed.State != "FAILED" || failed.ErrorCode != "FILE_CHECK_FAILED" || failed.LastSHA256 != j.LastSHA256 || failed.Batches != 1 { + t.Fatalf("failed checkpoint %+v", failed) + } + // 失败占位阻止到期计划另建作业。 + if _, err = db.ExecContext(ctx, `UPDATE orbit_archive_plans SET next_run_at=TIMESTAMPADD(DAY,-2,UTC_TIMESTAMP(6)) WHERE namespace_id=?`, token.NamespaceID); err != nil { + t.Fatal(err) + } + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + if len(mustArchiveJobs(t, ctx, s, token)) != 1 { + t.Fatal("failure spawned another job") + } + action := retention.JobActionInput{ExpectedRevision: failed.Revision, Action: "retry", MaxBytes: failed.MaxBytes, Reason: "restored exact original archive file"} + if err = os.WriteFile(file, saved, 0600); err != nil { + t.Fatal(err) + } + retried, err := s.ActArchiveJob(ctx, token, j.ID, "retry-original", action) + if err != nil || retried.ID != j.ID || retried.State != "PENDING" { + t.Fatal("retry", err) + } + if _, err = s.ActArchiveJob(ctx, token, j.ID, "retry-original", action); err != nil { + t.Fatal("retry replay", err) + } + if _, err = s.ActArchiveJob(ctx, token, j.ID, "retry-stale", action); fault.Code(err) != "STATE_CONFLICT" { + t.Fatal("stale retry", err) + } + // 改策略不允许旧作业静默采用新保留窗口。 + _, err = s.UpdateRetentionPolicy(ctx, token, "changed-policy", retention.UpdateInput{ExpectedRevision: 1, Enabled: true, IdempotencyDays: 90, TaskHistoryDays: 2, WorkflowHistoryDays: 2, EventLogDays: 2, DeliveryHistoryDays: 2, InboxHistoryDays: 2, AuditLogDays: 2, Reason: "new reviewed retention policy"}) + if err != nil { + t.Fatal(err) + } + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + failed = mustArchiveJobs(t, ctx, s, token)[0] + if failed.State != "FAILED" || failed.ErrorCode != "POLICY_CHANGED" { + t.Fatalf("changed policy %+v", failed) + } + action = retention.JobActionInput{ExpectedRevision: failed.Revision, Action: "abandon", MaxBytes: failed.MaxBytes, Reason: "close old version; preserve files"} + closed, err := s.ActArchiveJob(ctx, token, j.ID, "close", action) + if err != nil || closed.State != "ABANDONED" { + t.Fatal("close", err) + } + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + if len(mustArchiveJobs(t, ctx, s, token)) != 1 { + t.Fatal("stale plan triggered") + } + // 无会话和异命名空间不能读取或处理作业。 + other := token + other.NamespaceID = testID(t) + if _, err = s.ListArchiveJobs(ctx, other, ""); err == nil { + t.Fatal("cross namespace read") + } + if _, err = s.ActArchiveJob(ctx, other, j.ID, "cross", action); err == nil { + t.Fatal("cross namespace action") + } + // 更新计划策略后，过期的多次周期只生成一个新作业。 + in.ExpectedRevision = 1 + in.PolicyRevision = 2 + if _, err = s.SaveArchivePlan(ctx, token, "plan-new-policy", "audit", volume, in); err != nil { + t.Fatal(err) + } + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + jobs := mustArchiveJobs(t, ctx, s, token) + if len(jobs) != 2 || jobs[0].PolicyRevision != 2 { + t.Fatalf("new policy job %+v", jobs) + } + // 暂停计划不影响已有作业结清，但不再生成后续作业。 + in.ExpectedRevision = 2 + in.Enabled = false + if _, err = s.SaveArchivePlan(ctx, token, "pause-plan", "audit", volume, in); err != nil { + t.Fatal(err) + } + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + jobs = mustArchiveJobs(t, ctx, s, token) + if len(jobs) != 2 || jobs[0].State != "SUCCEEDED" { + t.Fatalf("pause blocked in-flight %+v", jobs) + } + var retries int + if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND action='retention.job.retry'`, token.NamespaceID).Scan(&retries); err != nil || retries != 1 { + t.Fatal("retry audit duplicated", err) + } + t.Log("corrupt file fails closed; restored file retries same identity exactly once; policy change blocks old job; audited close releases slot; disabled plan lets current job settle") +} + +func TestIntegrationArchiveBudgetsResumeAndPagination(t *testing.T) { + ctx, db, s, token, volume := archiveFixture(t, 600000) + in := archivePlanInput() + in.MaxBytes = 1 << 20 + if _, err := s.SaveArchivePlan(ctx, token, "plan", "audit", volume, in); err != nil { + t.Fatal(err) + } + runtime := &retention.ArchiveRuntime{Directory: filepath.Join(t.TempDir(), "volume"), StorageID: volume, MaxBytes: 8 << 20, Store: s} + if err := runtime.Init(); err != nil { + t.Fatal(err) + } + if err := runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + failed := mustArchiveJobs(t, ctx, s, token)[0] + if failed.State != "FAILED" || failed.ErrorCode != "STORAGE_LIMIT" { + t.Fatalf("budget was ignored %+v", failed) + } + _, err := s.ActArchiveJob(ctx, token, failed.ID, "raise-budget", retention.JobActionInput{ExpectedRevision: failed.Revision, Action: "retry", MaxBytes: 4 << 20, Reason: "reviewed file budget"}) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + } + complete := mustArchiveJobs(t, ctx, s, token)[0] + if complete.ID != failed.ID || complete.State != "SUCCEEDED" || complete.Roots != 3 || complete.Bytes <= 1<<20 || complete.Bytes > 4<<20 { + t.Fatalf("budget retry %+v", complete) + } + // 全卷预算独立于单作业预算：即使单作业允许4MiB，也不能突破剩余卷容量。 + if _, err = db.ExecContext(ctx, `UPDATE orbit_archive_plans SET next_run_at=TIMESTAMPADD(DAY,-1,UTC_TIMESTAMP(6)) WHERE namespace_id=?`, token.NamespaceID); err != nil { + t.Fatal(err) + } + runtime.MaxBytes = complete.Bytes + 100 + if err = runtime.Tick(ctx); err != nil { + t.Fatal(err) + } + jobs := mustArchiveJobs(t, ctx, s, token) + if len(jobs) != 2 || jobs[0].State != "FAILED" || jobs[0].ErrorCode != "STORAGE_LIMIT" || jobs[0].Bytes != 0 { + t.Fatalf("volume budget %+v", jobs) + } + // 分页使用真实时间+ID顺序，数量限制由SQL完成；无效/异命名空间位置拒绝。 + for i := 0; i < 24; i++ { + _, err = db.ExecContext(ctx, `INSERT INTO orbit_archive_jobs(namespace_id,id,kind,plan_revision,policy_revision,storage_id,batch_size,max_bytes,state,created_at) VALUES(?,?,'audit',1,1,?,2,1048576,'ABANDONED',TIMESTAMPADD(DAY,-2,UTC_TIMESTAMP(6)))`, token.NamespaceID, testID(t), volume) + if err != nil { + t.Fatal(err) + } + } + first, err := s.ListArchiveJobs(ctx, token, "") + if err != nil || len(first.Items) != 20 || first.NextBefore == "" { + t.Fatal("first page", err) + } + second, err := s.ListArchiveJobs(ctx, token, first.NextBefore) + if err != nil || len(second.Items) != 6 || second.NextBefore != "" { + t.Fatal("second page", err) + } + seen := map[core.ID]bool{} + for _, j := range append(first.Items, second.Items...) { + if seen[j.ID] { + t.Fatal("duplicate page item") + } + seen[j.ID] = true + } + if _, err = s.ListArchiveJobs(ctx, token, string(testID(t))); fault.Code(err) != "NOT_FOUND" { + t.Fatal("unknown page position", err) + } + t.Log("single-job byte budget stops and resumes same ID; volume budget blocks extra output; 26 jobs paginate 20+6 without duplicates") +} diff --git a/internal/storage/mysql/baseline.sql b/internal/storage/mysql/baseline.sql index ce9a49d..661ebde 100644 --- a/internal/storage/mysql/baseline.sql +++ b/internal/storage/mysql/baseline.sql @@ -1,4 +1,4 @@ --- Orbit fresh-install baseline, generated from migration version 30. +-- Orbit fresh-install baseline, generated from migration version 31. -- This file intentionally contains no database-level relation constraints. CREATE TABLE `orbit_application_resources` ( `application_id` binary(16) NOT NULL COMMENT '所属应用标识及主键，跨命名空间共享额度', @@ -878,3 +878,52 @@ CREATE TABLE orbit_retention_policies ( CONSTRAINT ck_retention_policy_inbox CHECK (inbox_history_days BETWEEN 1 AND 3650), CONSTRAINT ck_retention_policy_audit CHECK (audit_log_days BETWEEN 1 AND 3650) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='命名空间历史数据保留策略，只配置窗口和开关，不自动删除'; + +-- 周期计划、固定范围作业和领取租约；不使用数据库外键，不删除历史源记录。 +CREATE TABLE orbit_archive_plans ( + namespace_id BINARY(16) NOT NULL COMMENT '所属命名空间，由事务核验', + kind VARCHAR(16) NOT NULL COMMENT '固定历史类别，同一命名空间每类一个计划', + revision BIGINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '配置版本，后台触发不修改此版本', + enabled BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否触发新作业，不中断已有作业', + policy_revision BIGINT UNSIGNED NOT NULL COMMENT '管理员确认的保留策略版本', + storage_id BINARY(16) NOT NULL COMMENT '固定归档卷身份，接管节点必须挂载同一物理卷', + interval_seconds INT UNSIGNED NOT NULL COMMENT '周期秒数，3600至2592000，错过多次合并一次', + batch_size INT UNSIGNED NOT NULL COMMENT '每批根记录上限，1至100', + max_bytes BIGINT UNSIGNED NOT NULL COMMENT '新作业累计文件字节预算，1MiB至1GiB', + next_run_at DATETIME(6) NOT NULL COMMENT '下次触发UTC时间，首次启用立即到期', + active_job_id BINARY(16) NULL COMMENT '未完成或失败待处理作业，成功或关闭后为空', + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '首次保存UTC时间', + updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '配置最后修改UTC时间', + PRIMARY KEY(namespace_id,kind), + KEY ix_archive_plan_due(storage_id,enabled,next_run_at), + CONSTRAINT ck_archive_plan_kind CHECK(kind IN ('tasks','workflows','events','audit','deliveries','inbox','idempotency')), + CONSTRAINT ck_archive_plan_config CHECK(revision>0 AND policy_revision>0 AND interval_seconds BETWEEN 3600 AND 2592000 AND batch_size BETWEEN 1 AND 100 AND max_bytes BETWEEN 1048576 AND 1073741824) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='按命名空间和类别配置的周期归档计划，无外键'; +CREATE TABLE orbit_archive_jobs ( + namespace_id BINARY(16) NOT NULL COMMENT '所属命名空间，由事务核验', + id BINARY(16) NOT NULL COMMENT '作业及目录及文件清单的固定标识', + kind VARCHAR(16) NOT NULL COMMENT '创建时固定的历史类别', + plan_revision BIGINT UNSIGNED NOT NULL COMMENT '创建时固定的计划版本', + policy_revision BIGINT UNSIGNED NOT NULL COMMENT '创建时固定的保留策略版本', + storage_id BINARY(16) NOT NULL COMMENT '固定归档卷身份，不允许跨独立磁盘接管', + batch_size INT UNSIGNED NOT NULL COMMENT '创建时固定的单批根记录上限', + max_bytes BIGINT UNSIGNED NOT NULL COMMENT '累计文件字节预算，人工重试仅可提高', + state VARCHAR(16) NOT NULL DEFAULT 'PENDING' COMMENT 'PENDING待领取RUNNING执行中SUCCEEDED完成FAILED待处理ABANDONED已关闭', + revision BIGINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '状态并发版本，领取确认和人工操作递增', + attempts INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '累计领取次数，包含分批继续和崩溃接管', + batches INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已验证并入库的连续文件批次数', + roots BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已确认根记录总数', + bytes BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已确认清单及批次总字节数', + last_sha256 CHAR(64) NOT NULL DEFAULT '' COMMENT '已确认文件链尾摘要，恢复必须验证此前缀', + error_code VARCHAR(32) NOT NULL DEFAULT '' COMMENT '固定失败分类，不记录原始异常或业务内容', + lease_id BINARY(16) NULL COMMENT '领取随机代次，非执行中为空', + lease_until DATETIME(6) NULL COMMENT 'UTC租约截止，过期可接管，非执行中为空', + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '作业创建UTC时间', + updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '状态最后变更UTC时间', + PRIMARY KEY(namespace_id,id), + KEY ix_archive_job_claim(storage_id,state,lease_until,updated_at), + KEY ix_archive_job_list(namespace_id,created_at,id), + CONSTRAINT ck_archive_job_state CHECK(state IN ('PENDING','RUNNING','SUCCEEDED','FAILED','ABANDONED')), + CONSTRAINT ck_archive_job_config CHECK(revision>0 AND plan_revision>0 AND policy_revision>0 AND batch_size BETWEEN 1 AND 100 AND max_bytes BETWEEN 1048576 AND 1073741824), + CONSTRAINT ck_archive_job_lease CHECK((state='RUNNING' AND lease_id IS NOT NULL AND lease_until IS NOT NULL) OR (state<>'RUNNING' AND lease_id IS NULL AND lease_until IS NULL)) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='持久归档作业及租约，只读源数据，无外键'; diff --git a/internal/storage/mysql/migrate.go b/internal/storage/mysql/migrate.go index 8a20a54..be1111e 100644 --- a/internal/storage/mysql/migrate.go +++ b/internal/storage/mysql/migrate.go @@ -26,7 +26,7 @@ var migrationFiles embed.FS //go:embed baseline.sql var freshInstallBaseline string -const freshInstallBaselineVersion uint64 = 30 +const freshInstallBaselineVersion uint64 = 31 type Migration struct { Version uint64 diff --git a/internal/storage/mysql/migrations/000031_archive_scheduler.up.sql b/internal/storage/mysql/migrations/000031_archive_scheduler.up.sql new file mode 100644 index 0000000..5b4b751 --- /dev/null +++ b/internal/storage/mysql/migrations/000031_archive_scheduler.up.sql @@ -0,0 +1,48 @@ +-- 周期计划、固定范围作业和领取租约；不使用数据库外键，不删除历史源记录。 +CREATE TABLE orbit_archive_plans ( + namespace_id BINARY(16) NOT NULL COMMENT '所属命名空间，由事务核验', + kind VARCHAR(16) NOT NULL COMMENT '固定历史类别，同一命名空间每类一个计划', + revision BIGINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '配置版本，后台触发不修改此版本', + enabled BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否触发新作业，不中断已有作业', + policy_revision BIGINT UNSIGNED NOT NULL COMMENT '管理员确认的保留策略版本', + storage_id BINARY(16) NOT NULL COMMENT '固定归档卷身份，接管节点必须挂载同一物理卷', + interval_seconds INT UNSIGNED NOT NULL COMMENT '周期秒数，3600至2592000，错过多次合并一次', + batch_size INT UNSIGNED NOT NULL COMMENT '每批根记录上限，1至100', + max_bytes BIGINT UNSIGNED NOT NULL COMMENT '新作业累计文件字节预算，1MiB至1GiB', + next_run_at DATETIME(6) NOT NULL COMMENT '下次触发UTC时间，首次启用立即到期', + active_job_id BINARY(16) NULL COMMENT '未完成或失败待处理作业，成功或关闭后为空', + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '首次保存UTC时间', + updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '配置最后修改UTC时间', + PRIMARY KEY(namespace_id,kind), + KEY ix_archive_plan_due(storage_id,enabled,next_run_at), + CONSTRAINT ck_archive_plan_kind CHECK(kind IN ('tasks','workflows','events','audit','deliveries','inbox','idempotency')), + CONSTRAINT ck_archive_plan_config CHECK(revision>0 AND policy_revision>0 AND interval_seconds BETWEEN 3600 AND 2592000 AND batch_size BETWEEN 1 AND 100 AND max_bytes BETWEEN 1048576 AND 1073741824) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='按命名空间和类别配置的周期归档计划，无外键'; +CREATE TABLE orbit_archive_jobs ( + namespace_id BINARY(16) NOT NULL COMMENT '所属命名空间，由事务核验', + id BINARY(16) NOT NULL COMMENT '作业及目录及文件清单的固定标识', + kind VARCHAR(16) NOT NULL COMMENT '创建时固定的历史类别', + plan_revision BIGINT UNSIGNED NOT NULL COMMENT '创建时固定的计划版本', + policy_revision BIGINT UNSIGNED NOT NULL COMMENT '创建时固定的保留策略版本', + storage_id BINARY(16) NOT NULL COMMENT '固定归档卷身份，不允许跨独立磁盘接管', + batch_size INT UNSIGNED NOT NULL COMMENT '创建时固定的单批根记录上限', + max_bytes BIGINT UNSIGNED NOT NULL COMMENT '累计文件字节预算，人工重试仅可提高', + state VARCHAR(16) NOT NULL DEFAULT 'PENDING' COMMENT 'PENDING待领取RUNNING执行中SUCCEEDED完成FAILED待处理ABANDONED已关闭', + revision BIGINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '状态并发版本，领取确认和人工操作递增', + attempts INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '累计领取次数，包含分批继续和崩溃接管', + batches INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已验证并入库的连续文件批次数', + roots BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已确认根记录总数', + bytes BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已确认清单及批次总字节数', + last_sha256 CHAR(64) NOT NULL DEFAULT '' COMMENT '已确认文件链尾摘要，恢复必须验证此前缀', + error_code VARCHAR(32) NOT NULL DEFAULT '' COMMENT '固定失败分类，不记录原始异常或业务内容', + lease_id BINARY(16) NULL COMMENT '领取随机代次，非执行中为空', + lease_until DATETIME(6) NULL COMMENT 'UTC租约截止，过期可接管，非执行中为空', + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '作业创建UTC时间', + updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '状态最后变更UTC时间', + PRIMARY KEY(namespace_id,id), + KEY ix_archive_job_claim(storage_id,state,lease_until,updated_at), + KEY ix_archive_job_list(namespace_id,created_at,id), + CONSTRAINT ck_archive_job_state CHECK(state IN ('PENDING','RUNNING','SUCCEEDED','FAILED','ABANDONED')), + CONSTRAINT ck_archive_job_config CHECK(revision>0 AND plan_revision>0 AND policy_revision>0 AND batch_size BETWEEN 1 AND 100 AND max_bytes BETWEEN 1048576 AND 1073741824), + CONSTRAINT ck_archive_job_lease CHECK((state='RUNNING' AND lease_id IS NOT NULL AND lease_until IS NOT NULL) OR (state<>'RUNNING' AND lease_id IS NULL AND lease_until IS NULL)) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='持久归档作业及租约，只读源数据，无外键'; diff --git a/internal/storage/mysql/no_foreign_keys_integration_test.go b/internal/storage/mysql/no_foreign_keys_integration_test.go index 1c6fe35..1310073 100644 --- a/internal/storage/mysql/no_foreign_keys_integration_test.go +++ b/internal/storage/mysql/no_foreign_keys_integration_test.go @@ -147,7 +147,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { indexes := func() []string { t.Helper() // 排除后续新增/替换索引；原有项在此逐项比较，v24/v25新增项由对应Query/Timeline升级测试核对。 - rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events','orbit_console_users','orbit_console_sessions','orbit_retention_policies','weave_version_scopes') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created','uq_weave_task_result','ix_signal_expiry','ix_signal_activation','uq_wait_winner_signal','ix_task_workflow_owner','ix_signal_history','ix_event_workflow_history','ix_workflow_created','ix_workflow_state_created','ix_workflow_business_state_created','ix_workflow_definition_created','ix_workflow_definition_state_created','ix_workflow_definition_business_created','ix_workflow_definition_business_state_created','ix_task_observe_state_available','ix_delivery_observe_state_available','ix_workflow_observe_state_updated','ix_event_observe_type_occurred','ix_task_observe_terminal_completed','ix_workflow_observe_terminal_completed') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) + rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events','orbit_console_users','orbit_console_sessions','orbit_retention_policies','orbit_archive_jobs','orbit_archive_plans','weave_version_scopes') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created','uq_weave_task_result','ix_signal_expiry','ix_signal_activation','uq_wait_winner_signal','ix_task_workflow_owner','ix_signal_history','ix_event_workflow_history','ix_workflow_created','ix_workflow_state_created','ix_workflow_business_state_created','ix_workflow_definition_created','ix_workflow_definition_state_created','ix_workflow_definition_business_created','ix_workflow_definition_business_state_created','ix_task_observe_state_available','ix_delivery_observe_state_available','ix_workflow_observe_state_updated','ix_event_observe_type_occurred','ix_task_observe_terminal_completed','ix_workflow_observe_terminal_completed') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) if err != nil { t.Fatal(err) } diff --git a/internal/transport/httpserver/archive_jobs.go b/internal/transport/httpserver/archive_jobs.go new file mode 100644 index 0000000..d519c01 --- /dev/null +++ b/internal/transport/httpserver/archive_jobs.go @@ -0,0 +1,108 @@ +package httpserver + +import ( + "net/http" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/retention" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +// RegisterConsoleArchiveJobs 注册周期归档管理；只接受管理员会话，目标目录由部署配置固定。 +func RegisterConsoleArchiveJobs(server *http.Server, consoleAuth RetentionConsoleAuth, store retention.SchedulerStore, storage core.ID) { + if server == nil || consoleAuth == nil || store == nil { + return + } + mux, ok := server.Handler.(*http.ServeMux) + if !ok { + return + } + parent := mux + mux = http.NewServeMux() + for _, method := range []string{"GET", "PUT", "POST"} { + parent.Handle(method+" /console/archive/", http.NewCrossOriginProtection().Handler(mux)) + } + check := func(w http.ResponseWriter, r *http.Request) (auth.Token, bool) { + t, ok := consoleCookieToken(r) + if !ok || consoleAuth.ValidateSession(r.Context(), t) != nil { + writeError(w, "", fault.Unauthorized) + return auth.Token{}, false + } + return t, true + } + key := func(w http.ResponseWriter, r *http.Request) (string, bool) { + keys := r.Header.Values("Idempotency-Key") + if len(keys) != 1 { + writeError(w, "", fault.Invalid("exactly one Idempotency-Key is required")) + return "", false + } + return keys[0], true + } + mux.HandleFunc("GET /console/archive/plans", func(w http.ResponseWriter, r *http.Request) { + t, ok := check(w, r) + if !ok { + return + } + items, err := store.ListArchivePlans(r.Context(), t) + if err != nil { + writeError(w, "", err) + return + } + writeConsoleJSON(w, 200, map[string]any{"data": map[string]any{"items": items, "executor_configured": storage != "", "storage_id": string(storage)}}) + }) + mux.HandleFunc("PUT /console/archive/plans/{kind}", func(w http.ResponseWriter, r *http.Request) { + t, ok := check(w, r) + if !ok { + return + } + k, ok := key(w, r) + if !ok { + return + } + var in retention.PlanInput + if err := decodeBody(r, &in); err != nil { + writeError(w, "", err) + return + } + out, err := store.SaveArchivePlan(r.Context(), t, k, r.PathValue("kind"), storage, in) + if err != nil { + writeError(w, "", err) + return + } + writeConsoleJSON(w, 200, map[string]any{"data": out}) + }) + mux.HandleFunc("GET /console/archive/jobs", func(w http.ResponseWriter, r *http.Request) { + t, ok := check(w, r) + if !ok { + return + } + out, err := store.ListArchiveJobs(r.Context(), t, r.URL.Query().Get("before")) + if err != nil { + writeError(w, "", err) + return + } + writeConsoleJSON(w, 200, map[string]any{"data": out}) + }) + mux.HandleFunc("POST /console/archive/jobs/{id}/actions", func(w http.ResponseWriter, r *http.Request) { + t, ok := check(w, r) + if !ok { + return + } + k, ok := key(w, r) + if !ok { + return + } + var in retention.JobActionInput + if err := decodeBody(r, &in); err != nil { + writeError(w, "", err) + return + } + out, err := store.ActArchiveJob(r.Context(), t, core.ID(r.PathValue("id")), k, in) + if err != nil { + writeError(w, "", err) + return + } + writeConsoleJSON(w, 200, map[string]any{"data": out}) + }) +} diff --git a/internal/transport/httpserver/archive_jobs_test.go b/internal/transport/httpserver/archive_jobs_test.go new file mode 100644 index 0000000..71ac769 --- /dev/null +++ b/internal/transport/httpserver/archive_jobs_test.go @@ -0,0 +1,69 @@ +package httpserver + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "git.yuntuo.hk/k2safe/orbit/internal/retention" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +type fakeArchiveManager struct { + retention.SchedulerStore + calls int +} + +func (f *fakeArchiveManager) ListArchivePlans(context.Context, auth.Token) ([]retention.ArchivePlan, error) { + f.calls++ + return []retention.ArchivePlan{}, nil +} +func TestArchiveRoutesRequireCookieAndRejectCrossOrigin(t *testing.T) { + ns, _ := core.NewID() + cred, _ := core.NewID() + raw, _, _ := auth.Issue(ns, cred) + volume, _ := core.NewID() + authStore := &fakeConsoleAuth{valid: true} + store := &fakeArchiveManager{} + server := NewWithConsoleAuth("127.0.0.1:0", nil, authStore) + RegisterConsoleArchiveJobs(server, authStore, store, volume) + WrapConsoleAuth(server, authStore) + tests := []struct { + name, method, path, origin string + cookie, bearer bool + status int + }{ + {"no session", "GET", "/console/archive/plans", "", false, false, 401}, + {"bearer alone", "GET", "/console/archive/plans", "", false, true, 401}, + {"cookie read", "GET", "/console/archive/plans", "", true, false, 200}, + {"missing key", "PUT", "/console/archive/plans/tasks", "", true, false, 400}, + {"cross origin", "PUT", "/console/archive/plans/tasks", "https://foreign.invalid", true, false, 403}, + {"cross origin retry", "POST", "/console/archive/jobs/11111111111111111111111111111111/actions", "https://foreign.invalid", true, false, 403}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := httptest.NewRequest(tt.method, tt.path, strings.NewReader(`{}`)) + r.Header.Set("Content-Type", "application/json") + if tt.cookie { + r.AddCookie(&http.Cookie{Name: consoleSessionCookie, Value: raw}) + } + if tt.bearer { + r.Header.Set("Authorization", "Bearer "+raw) + } + if tt.origin != "" { + r.Header.Set("Origin", tt.origin) + } + w := httptest.NewRecorder() + server.Handler.ServeHTTP(w, r) + if w.Code != tt.status { + t.Fatalf("status=%d body=%s", w.Code, w.Body.String()) + } + }) + } + if store.calls != 1 { + t.Fatal("unexpected store access", store.calls) + } +} diff --git a/internal/transport/httpserver/console/app.js b/internal/transport/httpserver/console/app.js index fde6710..64da899 100644 --- a/internal/transport/httpserver/console/app.js +++ b/internal/transport/httpserver/console/app.js @@ -2,7 +2,7 @@ const state = { token: '', authenticated: false, queue: '', base: location.origin, consoleConfig: null, retention: null, timer: null, connected: false, loading: false, selectedTask: null, actionMode: '', selectedSchedule: null, selectedWorkflow: null, workflowActionMode: '', workflowRecoveryMode: '', workflowRecoveryContext: null, selectedSubscription: null, subscriptionActionMode: '', selectedDelivery: null }; const $ = (id) => document.getElementById(id); const text = (id, value) => { const el = $(id); if (el) el.textContent = value; }; - const pageNames = { overview: '总览', tasks: '任务执行', schedules: '定时计划', workflows: '流程实例', subscriptions: '结果订阅', retention: '保留策略' }; + const pageNames = { overview: '总览', tasks: '任务执行', schedules: '定时计划', workflows: '流程实例', subscriptions: '结果订阅', retention: '保留策略', archives: '归档作业' }; function showToast(message) { const toast = $('toast'); toast.textContent = message; toast.classList.add('show'); @@ -29,6 +29,7 @@ if (updateHash && location.hash !== `#/${next}`) location.hash = `/${next}`; window.scrollTo({ top: 0, behavior: 'smooth' }); if (next === 'retention' && state.authenticated) loadRetention(); + if (next === 'archives' && state.authenticated) loadArchives(); } function redirectToLogin() { const returnTo = `${location.pathname}${location.search}${location.hash}`; @@ -46,7 +47,7 @@ const raw = await response.text(); let body = {}; try { body = raw ? JSON.parse(raw) : {}; } catch { body = { message: raw }; } if (response.status === 401) { state.authenticated = false; redirectToLogin(); throw new Error('登录已过期，请重新登录'); } - if (!response.ok) throw new Error(body?.error?.message || body?.message || `HTTP ${response.status}`); + if (!response.ok) { const error = new Error(body?.error?.message || body?.message || `HTTP ${response.status}`); error.status = response.status; throw error; } return body; } async function loadConsoleConfig() { @@ -186,6 +187,96 @@ } catch (error) { text('retention-error', error.message || '归档快照导出失败'); } finally { retentionUI.busy = false; syncRetentionControls(); } } + const archiveUI = { busy: false, loading: false, dirty: false, plans: [], jobs: [], configured: false, policy: null, editing: null, editPolicy: '0', before: '', next: '', pending: null, selected: null }; + const archiveLabels = { tasks: '任务历史', workflows: '流程历史', events: '事件日志', audit: '管理审计', deliveries: '交付历史', inbox: 'Inbox 历史', idempotency: '幂等记录' }; + const archiveStates = { PENDING: '等待执行', RUNNING: '执行中', SUCCEEDED: '已完成', FAILED: '待处理', ABANDONED: '已关闭' }; + const archiveErrors = { FILE_CHECK_FAILED: '文件检查未通过，请核查持久卷、原目录与文件完整性。', STORAGE_LIMIT: '归档容量达到上限，请核查单作业与归档卷容量。', STORAGE_BUSY: '文件正被其他执行器使用，确认旧进程退出后可重试。', POLICY_CHANGED: '保留策略已变更或关闭。原作业不能替换策略版本，可关闭后重新配置周期。', PERMISSION_DENIED: '所属应用或命名空间已停用。', EXPORT_FAILED: '数据库导出未完成，确认数据库恢复后可重试。', INTERRUPTED: '执行中断，核查后可从原作业续跑。' }; + function syncArchiveControls() { + document.querySelectorAll('#archive-plan-form input, #archive-plan-form select, #archive-plan-form button, #archive-action-form input, #archive-action-form button, [data-archive-action]').forEach(el => { el.disabled = archiveUI.busy; }); + $('archive-plan-save').disabled = archiveUI.busy || !archiveUI.editing || archiveUI.editPolicy === '0'; + $('archive-action-budget').disabled = archiveUI.busy || archiveUI.selected?.action === 'abandon'; + $('archive-next').disabled = archiveUI.busy || archiveUI.loading || !archiveUI.next; + $('archive-first').disabled = archiveUI.busy || archiveUI.loading || !archiveUI.before; + } + function renderArchiveEditor() { + const p = archiveUI.plans.find(p => p.kind === $('archive-kind').value); if (!p) return; + archiveUI.editing = p; archiveUI.editPolicy = String(archiveUI.policy?.revision || '0'); archiveUI.dirty = false; + $('archive-enabled').checked = p.enabled; $('archive-interval').value = p.interval_seconds / 3600; + $('archive-batch').value = p.batch_size; $('archive-budget').value = p.max_bytes / 1048576; + text('archive-plan-revision', `计划版本 ${p.revision} · 保留策略 ${archiveUI.editPolicy}`); + text('archive-next-run', p.active_job_id ? `当前作业 ${shortID(p.active_job_id)} · 完成或关闭后继续周期` : `下次执行 ${p.enabled ? formatTime(p.next_run_at) : '尚未开启'}`); + const stale = p.enabled && String(p.policy_revision) !== archiveUI.editPolicy; + text('archive-policy-note', !archiveUI.policy?.enabled ? '请先在“保留策略”中允许只读归档。' : stale ? '保留策略已变更，重新保存本计划后才会触发新作业。' : '每个作业固定创建时的策略版本。'); + text('archive-plan-hint', archiveUI.configured ? '关闭周期只停止新作业，已有作业继续结清。失败作业需先处理，后续周期才会继续。' : '当前节点尚未配置归档持久卷，可查看历史或关闭已保存的周期。'); + syncArchiveControls(); + } + function archiveSize(bytes) { return bytes < 1048576 ? `${(bytes / 1024).toFixed(1)} KiB` : `${(bytes / 1048576).toFixed(2)} MiB`; } + function renderArchiveJobs() { + const jobs = archiveUI.jobs; + text('archive-job-count', `${jobs.length} 项`); + text('archive-enabled-count', String(archiveUI.plans.filter(p => p.enabled).length)); + text('archive-failed-count', String(jobs.filter(j => j.state === 'FAILED').length)); + text('archive-root-count', jobs.reduce((sum, j) => sum + j.roots, 0).toLocaleString('zh-CN')); + text('archive-executor', archiveUI.configured ? '已配置持久卷' : '未配置持久卷'); + $('archive-executor').classList.toggle('ready', archiveUI.configured); + $('archive-plan-chips').innerHTML = archiveUI.plans.map(p => `${archiveLabels[p.kind]} · ${p.enabled ? '开启' : '关闭'}`).join(''); + $('archive-job-list').innerHTML = jobs.length ? jobs.map(j => `
${escapeHTML(archiveLabels[j.kind] || j.kind)}${escapeHTML(archiveStates[j.state] || j.state)}
${escapeHTML(j.id)}
${j.batches} 批已确认${j.roots.toLocaleString('zh-CN')} 条根记录${archiveSize(j.bytes)} / ${j.max_bytes / 1048576} MiB策略 v${escapeHTML(j.policy_revision)}
${j.last_sha256 ? `
查看文件校验摘要${escapeHTML(j.last_sha256)}
` : ''}${j.error_code ? `

${escapeHTML(archiveErrors[j.error_code] || j.error_code)}

` : ''}${j.state === 'RUNNING' ? `

领取次数 ${j.attempts} · 租约至 ${escapeHTML(formatTime(j.lease_until))}，中断后可接管。

` : ''}${j.state === 'FAILED' ? `
` : ''}
`).join('') : '
▤还没有归档作业保存并开启周期后，系统会按保留窗口生成作业。
'; + syncArchiveControls(); + } + async function loadArchives() { + if (archiveUI.loading || archiveUI.busy) return; archiveUI.loading = true; + try { + const [plans, policy, jobs] = await Promise.all([request('/console/archive/plans'), request('/console/retention'), request(`/console/archive/jobs${archiveUI.before ? `?before=${encodeURIComponent(archiveUI.before)}` : ''}`)]); + const config = extractData(plans), page = extractData(jobs); + archiveUI.plans = config.items || []; archiveUI.configured = config.executor_configured; + archiveUI.policy = extractData(policy); archiveUI.jobs = page.items || []; archiveUI.next = page.next_before || ''; + if (!archiveUI.dirty) renderArchiveEditor(); renderArchiveJobs(); + } catch (error) { text('archive-error', error.message || '归档作业读取失败'); } + finally { archiveUI.loading = false; syncArchiveControls(); } + } + async function archiveMutation(path, method, input) { + const body = JSON.stringify(input), signature = `${method}:${path}:${body}`; + if (archiveUI.pending && archiveUI.pending.signature !== signature) throw new Error('上次操作尚未确认，请保持原内容重试；确认结果后再修改。'); + if (!archiveUI.pending) archiveUI.pending = { signature, key: crypto.randomUUID() }; + try { + const response = await request(path, { method, headers: { 'Content-Type': 'application/json', 'Idempotency-Key': archiveUI.pending.key }, body }); + archiveUI.pending = null; return response; + } catch (error) { + // 明确的客户端拒绝允许核对后更正；网络故障和服务端错误保留原幂等键。 + if (error.status >= 400 && error.status < 500) archiveUI.pending = null; + throw error; + } + } + async function saveArchivePlan(event) { + event.preventDefault(); if (archiveUI.busy || !archiveUI.editing) return; + archiveUI.busy = true; syncArchiveControls(); text('archive-error', ''); + try { + await archiveMutation(`/console/archive/plans/${encodeURIComponent(archiveUI.editing.kind)}`, 'PUT', { expected_revision: String(archiveUI.editing.revision), policy_revision: archiveUI.editPolicy, enabled: $('archive-enabled').checked, interval_seconds: Number($('archive-interval').value) * 3600, batch_size: Number($('archive-batch').value), max_bytes: Number($('archive-budget').value) * 1048576, reason: $('archive-reason').value }); + archiveUI.dirty = false; $('archive-reason').value = ''; showToast('周期配置已保存，操作已留存审计'); + } catch (error) { text('archive-error', `${error.message}。网络故障请保持原内容重试；版本冲突请刷新核对。`); } + finally { archiveUI.busy = false; syncArchiveControls(); } + if (!archiveUI.dirty) await loadArchives(); + } + function openArchiveAction(id, action) { + if (archiveUI.busy) return; const job = archiveUI.jobs.find(j => j.id === id); if (!job) return; + archiveUI.selected = { job, action }; $('archive-action-form').hidden = false; + text('archive-action-title', action === 'retry' ? '从原作业继续' : '关闭失败作业'); + text('archive-action-note', action === 'retry' ? `${shortID(id)} · 校验原文件后续跑，保留原策略与作业身份。容量不足时可提高上限。` : `${shortID(id)} · 保留已有文件和审计，不再重试本作业。已开启的计划可继续生成后续作业。`); + $('archive-action-budget').value = job.max_bytes / 1048576; $('archive-action-budget').min = job.max_bytes / 1048576; + $('archive-action-reason').value = ''; syncArchiveControls(); $('archive-action-reason').focus(); + } + async function submitArchiveAction(event) { + event.preventDefault(); if (archiveUI.busy || !archiveUI.selected) return; + archiveUI.busy = true; syncArchiveControls(); text('archive-error', ''); + try { + const { job, action } = archiveUI.selected; + await archiveMutation(`/console/archive/jobs/${encodeURIComponent(job.id)}/actions`, 'POST', { expected_revision: String(job.revision), action, max_bytes: Number($('archive-action-budget').value) * 1048576, reason: $('archive-action-reason').value }); + $('archive-action-form').hidden = true; archiveUI.selected = null; showToast('作业处理已保存，操作已留存审计'); + } catch (error) { text('archive-error', `${error.message}。网络故障请保持原内容重试。`); } + finally { archiveUI.busy = false; syncArchiveControls(); } + await loadArchives(); + } + function renderConnectionSummary() { text('summary-api', state.base || location.origin); text('summary-api-hint', state.connected ? '接口响应正常，按 10 秒周期刷新' : 'API 地址自动跟随当前页面'); @@ -597,6 +688,15 @@ finally { state.loading = false; $('refresh-button').classList.remove('spinning'); } } async function logout() { await fetch('/console/auth/logout', { method: 'POST', credentials: 'same-origin' }); redirectToLogin(); } + $('archive-plan-form').addEventListener('submit', saveArchivePlan); + $('archive-plan-form').addEventListener('input', () => { archiveUI.dirty = true; }); + $('archive-kind').addEventListener('change', () => { if (archiveUI.pending) { $('archive-kind').value = archiveUI.editing.kind; showToast('请先保持原内容重试，确认上次操作'); return; } renderArchiveEditor(); $('archive-reason').value = ''; }); + $('archive-refresh').addEventListener('click', () => { if (!archiveUI.busy && !archiveUI.pending) { archiveUI.dirty = false; text('archive-error', ''); loadArchives(); } }); + $('archive-next').addEventListener('click', () => { if (archiveUI.loading || archiveUI.busy) return; archiveUI.before = archiveUI.next; loadArchives(); }); + $('archive-first').addEventListener('click', () => { if (archiveUI.loading || archiveUI.busy) return; archiveUI.before = ''; loadArchives(); }); + $('archive-job-list').addEventListener('click', event => { const button = event.target.closest('[data-archive-action]'); if (button) openArchiveAction(button.dataset.id, button.dataset.archiveAction); }); + $('archive-action-form').addEventListener('submit', submitArchiveAction); + $('archive-action-cancel').addEventListener('click', () => { if (archiveUI.pending) { showToast('请先保持原内容重试，确认上次操作'); return; } $('archive-action-form').hidden = true; archiveUI.selected = null; }); $('retention-kind').addEventListener('change', resetRetentionExport); $('retention-restart-button').addEventListener('click', resetRetentionExport); $('retention-form').addEventListener('input', event => { if (event.target.matches('input, textarea')) { retentionUI.dirty = true; syncRetentionControls(); } }); @@ -607,6 +707,6 @@ document.querySelectorAll('[data-view]').forEach(button => button.addEventListener('click', event => { event.preventDefault(); setView(button.dataset.view); })); window.addEventListener('hashchange', () => setView(currentView(), false)); setView(currentView(), false); - ensureSession().then(ok => { if (ok) return loadConsoleConfig().then(() => Promise.all([loadRetention(), refresh()])); }).catch(() => redirectToLogin()); - state.timer = setInterval(() => { if (state.authenticated) refresh(); }, 10000); + ensureSession().then(ok => { if (ok) return loadConsoleConfig().then(() => Promise.all([loadRetention(), refresh(), currentView() === 'archives' ? loadArchives() : Promise.resolve()])); }).catch(() => redirectToLogin()); + state.timer = setInterval(() => { if (state.authenticated) { refresh(); if (currentView() === 'archives') loadArchives(); } }, 10000); })(); diff --git a/internal/transport/httpserver/console/index.html b/internal/transport/httpserver/console/index.html index a225d64..35cf49b 100644 --- a/internal/transport/httpserver/console/index.html +++ b/internal/transport/httpserver/console/index.html @@ -24,6 +24,7 @@ +
@@ -130,6 +131,41 @@
+
+
DATA / ARCHIVE JOBS

让历史数据，有序归档。

配置周期，自动分批保存历史；每次执行都有进度与校验记录。

+
周期计划—已启用 / 7 个类别
本页待处理—失败作业保留进度，等待处理
本页归档记录—数据库已确认的根记录数
+ +
+
+
SCHEDULE

周期配置

正在读取
+
+ +
开启周期归档首次启用立即运行，之后按周期执行。
+
+ +
配置尚未保存下次执行 —
+ + +

归档使用已保存的保留策略。关闭周期只停止新作业，已有作业继续结清。

+
+
+ +
+
+
EXECUTION HISTORY

归档作业

0 项
+

源数据库记录始终保留。文件写入服务端持久卷，页面展示已确认的批次和 SHA-256 摘要。

+
正在读取归档作业…
+
+ +
+
+
+
diff --git a/internal/transport/httpserver/console/styles.css b/internal/transport/httpserver/console/styles.css index e1fbe52..48a1fa3 100644 --- a/internal/transport/httpserver/console/styles.css +++ b/internal/transport/httpserver/console/styles.css @@ -10,3 +10,5 @@ .retention-tools{display:flex;align-items:flex-end;justify-content:space-between;gap:16px;margin-top:22px;padding-top:18px;border-top:1px solid rgba(156,183,218,.1)}.retention-select{display:flex;align-items:center;gap:10px;color:#9eb4cb;font-size:10px}.retention-select select{min-width:150px;border:1px solid rgba(156,183,218,.15);border-radius:9px;background:#0a1a2e;color:#dceafb;font:inherit;padding:9px 11px;outline:0}.retention-select select:focus{border-color:rgba(98,230,212,.58)}.retention-tool-buttons{display:flex;gap:9px}.secondary-button.compact{min-height:37px;padding:0 14px}.retention-preview{margin-top:13px;padding:13px 14px;border:1px solid rgba(156,183,218,.1);border-radius:11px;background:rgba(4,13,24,.28);color:#7189a3;font-size:10px;line-height:1.5}.retention-preview-head{display:flex;justify-content:space-between;gap:10px;padding-bottom:8px;color:#c9d8e8}.retention-preview-head small{color:#7189a3}.retention-preview-row{display:grid;grid-template-columns:1fr auto 1.6fr;gap:12px;padding:7px 0;border-top:1px solid rgba(156,183,218,.08)}.retention-preview-row b{color:#9beade;font-weight:600}.retention-preview-row small{color:#7189a3;text-align:right}@media(max-width:780px){.retention-tools{align-items:stretch;flex-direction:column}.retention-tool-buttons{justify-content:flex-start}.retention-preview-row{grid-template-columns:1fr auto}.retention-preview-row small{grid-column:span 2;text-align:left}} .retention-tool-buttons{flex-wrap:wrap}.secondary-button{display:inline-flex;align-items:center;justify-content:center;gap:8px;border:1px solid rgba(156,183,218,.18);border-radius:9px;background:rgba(23,46,65,.6);color:#b9cddd;font:inherit;font-size:10px;cursor:pointer}.secondary-button:hover{border-color:#62e6d4;color:#9beade}.retention-form button:disabled{opacity:.45;cursor:not-allowed}.retention-export-result{display:flex;flex-wrap:wrap;gap:10px;color:#9eb4cb;font-size:10px;line-height:1.6}.retention-export-result a{color:#62e6d4}.retention-export-result code{width:100%;font-size:9px;overflow-wrap:anywhere;color:#7189a3}.retention-export-result [hidden]{display:none}.retention-tools{flex-wrap:wrap} + +.archive-heading{display:flex;justify-content:space-between;align-items:center;gap:20px;margin:22px 0}.archive-heading h1{font-size:28px;letter-spacing:-.7px;margin:12px 0;color:#e4effb}.archive-heading p{color:#849bb3;font-size:12px;line-height:1.7}.archive-stats{display:grid;grid-template-columns:repeat(3,minmax(0,1fr));gap:15px;margin-bottom:20px}.archive-stats article{padding:22px 24px;border:1px solid #20354a;border-radius:18px;background:linear-gradient(125deg,#112439,#0a1728)}.archive-stats article:first-child{background:linear-gradient(125deg,#13373e,#0b202d);border-color:#24535a}.archive-stats span,.archive-stats small{display:block;color:#8ba2b8;font-size:11px}.archive-stats strong{display:block;font-size:30px;margin:10px 0;color:#e5f9f5}.archive-grid{display:grid;grid-template-columns:minmax(290px,360px) minmax(0,1fr);gap:20px;align-items:start}.archive-config,.archive-history{padding:24px;min-width:0}.archive-config .panel-heading,.archive-history .panel-heading{padding:0;margin-bottom:20px}.archive-config input:not([type=checkbox]),.archive-config select,#archive-action-form input{width:100%;border:1px solid #294157;border-radius:9px;padding:11px 12px;background:#09192a;color:#dbe8f5;font:inherit;font-size:12px;outline:0}.archive-config input:focus,.archive-config select:focus,#archive-action-form input:focus{border-color:#61dfd0;box-shadow:0 0 0 3px #61dfd010}.archive-config .field-label,#archive-action-form .field-label{display:block;margin:18px 0 9px;font-size:11px}.archive-config .retention-switch{margin-top:19px;padding:13px 12px}.archive-input-grid{display:grid;grid-template-columns:1fr 1fr;gap:12px;margin-top:20px;color:#94adc4;font-size:11px}.archive-input-grid input{margin-top:9px}.archive-plan-meta{display:grid;gap:9px;margin:17px 0;color:#7893ae;font-size:10px;line-height:1.6}.archive-config .primary-button{width:100%;margin:17px 0 6px}.archive-config .panel-description{line-height:1.8;font-size:10px}.archive-plan-chips{display:flex;flex-wrap:wrap;gap:7px;margin:16px 0}.archive-plan-chips span{padding:6px 8px;border:1px solid #24374a;border-radius:7px;font-size:9px;color:#7590aa}.archive-plan-chips .enabled{color:#85e5d8;border-color:#2d6564;background:#184b4b44}.archive-history>.panel-description{margin-bottom:18px;line-height:1.8}.archive-job{border:1px solid #243a50;border-radius:13px;padding:18px;margin:12px 0;background:#09172766}.archive-job-head{display:flex;align-items:center;justify-content:space-between;gap:12px}.archive-job-head>div{display:flex;align-items:center;gap:12px}.archive-job-head b{font-size:13px;color:#d2e4f1}.archive-job-head time{font-size:10px;color:#7c98b1}.archive-job-id{display:block;margin:10px 0;color:#65849f;font-size:10px;overflow-wrap:anywhere}.archive-progress{display:flex;flex-wrap:wrap;gap:14px;margin:14px 0;color:#8eaac1;font-size:10px}.archive-progress b{color:#ccdee9;font-size:12px;font-weight:550}.archive-job details{font-size:10px;color:#7297b5;line-height:1.7}.archive-job summary{cursor:pointer}.archive-digest{display:block;overflow-wrap:anywhere;margin-top:8px;color:#9cbad0}.archive-job-error{padding:10px 12px;background:#59412338;border-radius:8px;color:#d9b484;font-size:11px;line-height:1.8}.archive-job-note{font-size:10px;color:#7698b2}.archive-job-actions,.archive-pager{display:flex;gap:14px;align-items:center;margin-top:16px}.archive-pager{justify-content:space-between}.archive-empty{padding:60px 20px}.archive-empty b,.archive-empty small{display:block;margin-top:15px}.archive-grid button:disabled{opacity:.4;cursor:not-allowed}.archive-history [hidden]{display:none}#archive-action-form{border:1px solid #36736f;border-radius:13px;padding:20px;margin-top:20px;background:#102c3544}#archive-action-form h3{font-size:16px;color:#cdeae6}.state-pill.abandoned{color:#9bb0c2;background:#52637933}@media(max-width:1150px){.archive-grid{grid-template-columns:1fr}.archive-stats{gap:10px}.archive-heading h1{font-size:24px}}@media(max-width:600px){.archive-stats{grid-template-columns:1fr}.archive-heading{align-items:flex-start;flex-direction:column}.archive-config,.archive-history{padding:18px}.archive-job-head{align-items:flex-start;flex-direction:column}} diff --git a/scripts/test-console-e2e.sh b/scripts/test-console-e2e.sh index c88fc43..ba1d2ea 100755 --- a/scripts/test-console-e2e.sh +++ b/scripts/test-console-e2e.sh @@ -21,6 +21,15 @@ cleanup() { done rmdir "$E2E_DIR/archive-job" >/dev/null 2>&1 || true fi + if [ -d "$E2E_DIR/archive-volume" ]; then + python3 - "$E2E_DIR/archive-volume" <<'PY_CLEAN' +import pathlib, shutil, sys +path = pathlib.Path(sys.argv[1]) +# 仅清理由该测试 mktemp 目录下创建的独立归档卷。 +if path.name == "archive-volume" and path.parent.name.startswith("orbit-console-e2e."): + shutil.rmtree(path) +PY_CLEAN + fi for e2e_artifact in "$E2E_DIR"/*; do if [ -f "$e2e_artifact" ]; then unlink "$e2e_artifact" @@ -90,11 +99,17 @@ print(queue) PY ) +mkdir -m 700 "$E2E_DIR/archive-volume" +E2E_STORAGE_ID=$(openssl rand -hex 16) printf '%s\n' "[4/8] start Orbit container" "$E2E_DOCKER" run -d --name "$E2E_ORBIT" --network bridge --link "$E2E_MYSQL:mysql" \ --env ORBIT_MYSQL_DSN="$E2E_DSN" \ --env ORBIT_HTTP_ADDR=0.0.0.0:8080 \ --env ORBIT_CONSOLE_QUEUE_ID="$E2E_QUEUE" \ + --env ORBIT_ARCHIVE_ROOT=/var/lib/orbit/archive \ + --env ORBIT_ARCHIVE_STORAGE_ID="$E2E_STORAGE_ID" \ + --env ORBIT_ARCHIVE_TICK_INTERVAL=1s \ + --volume "$E2E_DIR/archive-volume:/var/lib/orbit/archive" \ --publish 127.0.0.1::8080 "$E2E_IMAGE" >/dev/null E2E_PORT= for e2e_attempt in $(seq 1 30); do @@ -247,12 +262,39 @@ assert content["source_retained"] is True and content["root_count"] == 0 status, _ = request("/console/retention/export", "POST", {"kind": "tasks", "expected_revision": "0"}) assert status == 409, status print("retention_policy=ok replay=ok stale_revision=409 preview_categories=7 empty_export_digest=ok") +status, body = request("/console/archive/plans") +assert status == 200, (status, body) +plans = json.loads(body)["data"] +assert len(plans["items"]) == 7 and plans["executor_configured"] is True +plan = dict(expected_revision="0", policy_revision="1", enabled=True, interval_seconds=3600, batch_size=20, max_bytes=1048576, reason="console scheduler e2e") +plan_headers = {"Idempotency-Key": "archive-plan-e2e"} +status, body = request("/console/archive/plans/tasks", "PUT", plan, {**plan_headers, "Origin": "https://different-origin.invalid"}) +assert status == 403, (status, body) +status, body = request("/console/archive/plans/tasks", "PUT", plan, plan_headers) +assert status == 200 and json.loads(body)["data"]["revision"] == "1", (status, body) +status, body = request("/console/archive/plans/tasks", "PUT", plan, plan_headers) +assert status == 200 and json.loads(body)["data"]["revision"] == "1", (status, body) +import time +for attempt in range(30): + status, body = request("/console/archive/jobs") + assert status == 200, (status, body) + jobs = json.loads(body)["data"]["items"] + if jobs and jobs[0]["state"] in ("SUCCEEDED", "FAILED"): + break + time.sleep(1) +assert len(jobs) == 1 and jobs[0]["state"] == "SUCCEEDED", jobs +assert jobs[0]["batches"] == 1 and jobs[0]["roots"] == 0 and len(jobs[0]["last_sha256"]) == 64 +assert "lease_id" not in jobs[0] +print("archive_scheduler=ok plan_replay=ok cross_origin=403 durable_job=SUCCEEDED") + status, _ = request("/console/auth/logout", "POST", {}) assert status == 200, status status, _ = request("/v1/tasks?" + urllib.parse.urlencode({"queue_id": queue, "limit": 1})) assert status == 401, status status_retention, _ = request("/console/retention") assert status_retention == 401, status_retention +status_archive, _ = request("/console/archive/jobs") +assert status_archive == 401, status_archive print("console_task_data=ok succeeded_tasks=%d logout=ok unauthorized_after_logout=%d" % (sum(item.get("state") == "SUCCEEDED" for item in items), status)) PY