diff --git a/.env.example b/.env.example index 4d76b3a..08777be 100644 --- a/.env.example +++ b/.env.example @@ -30,3 +30,7 @@ ORBIT_RETENTION_AUDIT_DAYS=2555 # Empty disables metrics. To enable a separate private listener, use e.g. # ORBIT_METRICS_ADDR=127.0.0.1:9090. Never expose it on the public API ingress. ORBIT_METRICS_ADDR= + +# 仅供 orbit-admin retention archive / verify-archive 使用;每个作业单独的绝对目录。 +# 不会启动后台定时归档,完整参数见 docs/api/retention-jobs.md。 +# ORBIT_ARCHIVE_DIRECTORY=/srv/orbit-archives/tasks-20260923 diff --git a/README.md b/README.md index 01b7785..ba37944 100644 --- a/README.md +++ b/README.md @@ -85,6 +85,8 @@ make docs-check 受理层负载采样工具见 [有界受理负载工具](docs/api/load.md),只允许在隔离命名空间或一次性数据库中运行;它不自动重试、不启动 Worker,也不提供生产容量承诺。 +历史数据可通过控制台配置保留窗口,也可使用 [`orbit-admin retention archive`](docs/api/retention-jobs.md) 自动分批落盘、断点续跑和离线核验。当前保留源数据,周期归档调度和删除执行器另行交付。 + 真实Worker与任务完成路径测量使用[`orbit-exec-load`](docs/api/execution-load.md):短/混合/中长模拟任务、持久结果和单次尝试核验、阶段延迟分布;不修改队列额度、不调用外部业务系统。定时积压复现可运行 `make schedule-backlog`,它只使用一次性 MySQL,并保留有界截止负例的实际分母。Weave 并行流程链路使用[`orbit-workflow-load`](docs/api/workflow-load.md),核对固定版本、步骤、来源和最终输出。定时集中触发使用[`orbit-schedule-load`](docs/api/schedule-load.md),逐项核对触发、任务、唯一尝试与来源,并保留超时分母和暂停收尾证据。 SDK 独立发布门禁使用 `make sdk-check`,在 `GOWORK=off` 下检查完整依赖图,并运行 SDK 的竞态测试、vet 和全包构建;规则和接入边界见 [SDK 发布门禁](docs/sdk/release.md)。 diff --git a/cmd/orbit-admin/main.go b/cmd/orbit-admin/main.go index 54b8bb2..8ac5d2a 100644 --- a/cmd/orbit-admin/main.go +++ b/cmd/orbit-admin/main.go @@ -27,7 +27,7 @@ func main() { } func run() error { if len(os.Args) < 2 { - return errors.New("usage: orbit-admin release manifest|bootstrap|console-user create|verify|observe snapshot|retention inventory|credential status|rotate|revoke|apply") + return errors.New("usage: orbit-admin release manifest|bootstrap|console-user create|verify|observe snapshot|retention inventory|archive|verify-archive|credential status|rotate|revoke|apply") } switch os.Args[1] { case "release": @@ -45,7 +45,7 @@ func run() error { case "credential": return runCredential(os.Args[2:]) } - return errors.New("usage: orbit-admin release manifest|bootstrap|console-user create|verify|observe snapshot|retention inventory|credential status|rotate|revoke|apply") + return errors.New("usage: orbit-admin release manifest|bootstrap|console-user create|verify|observe snapshot|retention inventory|archive|verify-archive|credential status|rotate|revoke|apply") } // releaseManifest 是发布物自身的静态身份清单,不读取数据库、不包含 DSN、凭据或业务数据。 @@ -208,8 +208,16 @@ func runObserve(args []string) error { } func runRetention(args []string) error { + if len(args) > 0 { + switch args[0] { + case "archive": + return runRetentionArchive(args[1:]) + case "verify-archive": + return runRetentionVerifyArchive(args[1:]) + } + } if len(args) == 0 || args[0] != "inventory" { - return errors.New("usage: orbit-admin retention inventory --before RFC3339") + return errors.New("usage: orbit-admin retention inventory --before RFC3339 | archive | verify-archive") } flags := flag.NewFlagSet("retention inventory", flag.ContinueOnError) var before string diff --git a/cmd/orbit-admin/retention.go b/cmd/orbit-admin/retention.go new file mode 100644 index 0000000..4fd850b --- /dev/null +++ b/cmd/orbit-admin/retention.go @@ -0,0 +1,108 @@ +package main + +import ( + "context" + "encoding/hex" + "encoding/json" + "errors" + "flag" + "fmt" + "os" + "os/signal" + "path/filepath" + "syscall" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/config" + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/retention" + "git.yuntuo.hk/k2safe/orbit/internal/storage/mysql" + core "git.yuntuo.hk/k2safe/orbit/model" +) + +func runRetentionArchive(args []string) error { + flags := flag.NewFlagSet("retention archive", flag.ContinueOnError) + var spec retention.JobSpec + var directory, namespace string + var options retention.JobOptions + var timeout time.Duration + flags.StringVar(&directory, "directory", os.Getenv("ORBIT_ARCHIVE_DIRECTORY"), "private per-job output directory; same directory resumes saved batches") + flags.StringVar(&namespace, "namespace", "", "namespace ID to archive") + flags.StringVar(&spec.Kind, "kind", "", "tasks|workflows|events|audit|deliveries|inbox|idempotency") + flags.Uint64Var(&spec.PolicyRevision, "revision", 0, "confirmed database retention policy revision") + flags.IntVar(&spec.BatchSize, "batch-size", 20, "roots per batch, 1-100; immutable for this job") + flags.IntVar(&options.MaxBatches, "max-batches", 10, "maximum new batches per invocation, 1-1000") + flags.Int64Var(&options.MaxBytes, "max-bytes", 256<<20, "total saved manifest and batch bytes, 1 MiB-100 GiB") + flags.DurationVar(&timeout, "timeout", 5*time.Minute, "invocation deadline, 1s-1h") + if err := flags.Parse(args); err != nil { + return err + } + spec.NamespaceID = core.ID(namespace) + if flags.NArg() != 0 || directory == "" || !filepath.IsAbs(directory) || timeout < time.Second || timeout > time.Hour || options.MaxBatches < 1 || options.MaxBatches > 1000 || options.MaxBytes < 1<<20 || options.MaxBytes > 100<<30 { + return errors.New("invalid archive arguments; use --help for limits and an absolute --directory") + } + if err := spec.Validate(); err != nil { + return err + } + cfg, err := config.Load() + if err != nil { + return err + } + signalCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + ctx, cancel := context.WithTimeout(signalCtx, timeout) + defer cancel() + db, err := mysql.Open(ctx, cfg.MySQLDSN, 2, 1) + if err != nil { + return errors.New("database unavailable") + } + defer db.Close() + if err := mysql.CheckSchema(ctx, db); err != nil { + return errors.New("database schema is not ready") + } + store := mysql.NewRetentionPolicyStore(db, cfg.RetentionDefaults) + status, runErr := retention.RunArchive(ctx, directory, spec, options, func(ctx context.Context, job core.ID, input retention.ArchiveInput) (retention.ArchiveSnapshot, error) { + snapshot, err := store.ExportRetentionForJob(ctx, spec.NamespaceID, job, input) + if err != nil { + // 数据库原始错误可能含 SQL 参数,只输出固定错误码,业务正文不进入控制台日志。 + if ctx.Err() != nil { + return retention.ArchiveSnapshot{}, ctx.Err() + } + return retention.ArchiveSnapshot{}, fmt.Errorf("archive database export failed (%s); keep original directory and verify policy before retrying", fault.Code(err)) + } + return snapshot, nil + }) + if err := json.NewEncoder(os.Stdout).Encode(status); err != nil { + return err + } + return runErr +} + +func runRetentionVerifyArchive(args []string) error { + flags := flag.NewFlagSet("retention verify-archive", flag.ContinueOnError) + var directory, expectedSHA string + flags.StringVar(&directory, "directory", os.Getenv("ORBIT_ARCHIVE_DIRECTORY"), "saved archive job directory; no database connection required") + flags.StringVar(&expectedSHA, "expected-sha256", "", "optional independently saved last_sha256 to pin the entire chain") + if err := flags.Parse(args); err != nil { + return err + } + if flags.NArg() != 0 || directory == "" { + return errors.New("--directory is required") + } + if expectedSHA != "" { + raw, err := hex.DecodeString(expectedSHA) + if err != nil || len(raw) != 32 || hex.EncodeToString(raw) != expectedSHA { + return errors.New("--expected-sha256 must be 64 lowercase hexadecimal characters") + } + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + status, err := retention.VerifyArchive(ctx, directory) + if encodeErr := json.NewEncoder(os.Stdout).Encode(status); encodeErr != nil { + return encodeErr + } + if err == nil && expectedSHA != "" && status.LastSHA256 != expectedSHA { + return errors.New("archive differs from independently saved digest") + } + return err +} diff --git a/cmd/orbit-admin/retention_test.go b/cmd/orbit-admin/retention_test.go new file mode 100644 index 0000000..9633c16 --- /dev/null +++ b/cmd/orbit-admin/retention_test.go @@ -0,0 +1,50 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "path/filepath" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/retention" + core "git.yuntuo.hk/k2safe/orbit/model" +) + +func TestRetentionArchiveValidationBeforeDatabase(t *testing.T) { + t.Setenv("ORBIT_MYSQL_DSN", "") + for _, args := range [][]string{ + {}, {"--directory", "relative"}, {"--directory", t.TempDir(), "--namespace", "invalid"}, + {"--directory", t.TempDir(), "--namespace", "11111111111111111111111111111111", "--kind", "tasks", "--revision", "1", "--max-batches", "0"}, + } { + if err := runRetentionArchive(args); err == nil { + t.Fatalf("invalid args accepted: %v", args) + } + } +} + +func TestVerifyArchiveOfflinePinsDigest(t *testing.T) { + t.Setenv("ORBIT_MYSQL_DSN", "") + dir := filepath.Join(t.TempDir(), "job") + ns := core.ID("11111111111111111111111111111111") + spec := retention.JobSpec{NamespaceID: ns, Kind: "tasks", PolicyRevision: 1, BatchSize: 1} + status, err := retention.RunArchive(context.Background(), dir, spec, retention.JobOptions{MaxBatches: 1, MaxBytes: 1 << 20}, func(ctx context.Context, id core.ID, input retention.ArchiveInput) (retention.ArchiveSnapshot, error) { + policy := retention.Defaults(ns) + policy.Enabled = true + policy.Revision = 1 + raw, err := json.Marshal(map[string]any{"policy": policy, "kind": "tasks", "cutoff": time.Now().UTC(), "root_count": 0, "source_retained": true, "next_cursor": ""}) + digest := sha256.Sum256(raw) + return retention.ArchiveSnapshot{FormatVersion: 1, SHA256: hex.EncodeToString(digest[:]), Content: string(raw)}, err + }) + if err != nil { + t.Fatal(err) + } + if err := runRetentionVerifyArchive([]string{"--directory", dir, "--expected-sha256", status.LastSHA256}); err != nil { + t.Fatal(err) + } + if err := runRetentionVerifyArchive([]string{"--directory", dir, "--expected-sha256", "0000000000000000000000000000000000000000000000000000000000000000"}); err == nil { + t.Fatal("trusted digest mismatch accepted") + } +} diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 6935895..203194b 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -262,3 +262,7 @@ Event.WorkflowID为只读STORED生成列,从原不可变事件派生流程归 000029增加 `orbit_console_users` 和 `orbit_console_sessions`,只保存控制台账号的 bcrypt 哈希、绑定凭据及 HttpOnly 会话摘要;不保存密码或令牌原文。控制台继续单管理员模型,服务层每次修改前核对会话、命名空间和绑定凭据权限。 000030增加 `orbit_retention_policies`,按命名空间保存归档开关、幂等/任务/流程/事件/交付/Inbox/审计窗口、修订号及时间。所有字段都有中文 COMMENT,策略不含外键或级联。数据库策略优先于部署默认值;默认关闭归档,配置变化不触发删除。服务层用同一事务完成版本核对、幂等回执和审计快照,并在归档预览/导出时保护未确认交付、未知效果、未结束流程和有效幂等身份。具体窗口、候选目录和控制台接口见[数据保留与归档策略](api/retention.md)。 + +## 文件归档作业模型 + +`internal/retention` 中的 `JobSpec`、`JobManifest`、`JobBatch` 和 `JobStatus` 定义固定作业范围、不可变清单、连续批次及汇总状态;各持久字段提供中文注释。它们只存入私有归档目录,不新增数据库表。连续文件与摘要链就是恢复检查点,作业 `JobID` 进入现有管理审计的执行者身份。协议与失败边界见[文件归档作业](api/retention-jobs.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 5a0b883..a289899 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -109,7 +109,7 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( 1. 已补[首批可复现故障工具](reports/phase-6-fault-boundaries.md):A04“派发已提交但消息尚未送达”、W01/W13事务提交结果不明,以及W16最终回执/旧PENDING结果重放/下一轮竞争。迁移器同时提供当前版本无外键新装基线;每个切点必须有实际数据库或进程证据,不能用响应错误推断事务是否已提交,也不能用普通SIGNAL竞争替代最终查询场景。数据库事务封装另对提交前 MySQL 1205/1213 做最多两次短退避重试,Commit 回执不明保持原身份核对语义,真实锁环回归见[P7当前回归状态](reports/phase-7-regression-status.md)。 2. 已实现[基础运行观测](api/observability.md)及[有界业务状态与事件采样](reports/phase-6-business-observability.md):HTTP/WS、数据库连接池、后台批次、固定任务/交付/流程/Inbox状态计数、最近事件类别、终态耗时样本、固定事件阶段耗时、最早年龄、截断/新鲜度指标、提交前 MySQL 事务重试/耗尽计数和可离线验证的告警规则;独立采集监听默认关闭,固定指标目录拒绝动态标签;发布前只读结构核验由[`orbit-admin verify`](api/operations.md)提供,运维人员也可用 `orbit-admin observe snapshot` 获取同一有界单节点快照;后台循环间隔、批量上限、Worker/Event连接上限、Hub领取协程和有界内存排队均支持有界环境配置,并在启动日志和固定指标中记录。跨节点重连风暴、慢消费者、真实锁等待/死锁、磁盘增长已经加入定向回归入口;目标拓扑的全局持久吞吐、锁等待/死锁告警阈值和生产采集/通知验证仍需环境验收。 3. 已补齐受保护管理环境中的固定凭据操作文件、轮换、撤销、提交不明故障切点、幂等回执和脱敏管理审计;轮换保留授权绑定并要求新令牌重新建立会话,旧 Worker 会话被排空且立即过期,撤销允许在停用范围内执行且不改写历史记录。已用一次性 MySQL 定向验证旧令牌失效、新令牌重绑和零外键;仍需多节点安全重绑演练、凭据最小权限配置核对和生产密钥托管/审计演练。 -4. 已补[数据库保留策略、预览和只读归档导出](api/retention.md)以及[`orbit-admin retention inventory`](api/retention.md)只读盘点。策略按命名空间存入 MySQL,控制台单管理员可修改,版本、原因和前后快照进入审计;默认关闭归档,源数据不会因配置变化自动删除。固定候选目录保护有效幂等身份、未确认交付、未知效果、未终结流程和未处理 Inbox;导出包带冻结策略版本和 SHA-256;控制台支持连续批次、累计数量、摘要校验和本批重下载,真实 MySQL 回归核对同时间戳分页、保护对象、版本及权限边界。仍需在目标环境评审最终归档介质、备份恢复、删除执行器和保留周期,再执行兼容升级和容量负载。生产HA及RPO/RTO继续按实际拓扑验证,不缩减为本机多进程检查。 +4. 已补[数据库保留策略、预览和只读归档导出](api/retention.md)以及[`orbit-admin retention inventory`](api/retention.md)只读盘点。策略按命名空间存入 MySQL,控制台单管理员可修改,版本、原因和前后快照进入审计;默认关闭归档,源数据不会因配置变化自动删除。固定候选目录保护有效幂等身份、未确认交付、未知效果、未终结流程和未处理 Inbox;导出包带冻结策略版本和 SHA-256;控制台支持连续批次、累计数量、摘要校验和本批重下载,真实 MySQL 回归核对同时间戳分页、保护对象、版本及权限边界。已补[持久文件归档作业](api/retention-jobs.md):自动有限分批、原目录续跑、原子发布、容量限制、摘要链、离线核验及真实强杀恢复;尚未接入周期调度、控制台作业列表或跨节点自动接管。仍需在目标环境评审最终归档介质、备份恢复、删除执行器和保留周期,再执行兼容升级和容量负载。生产HA及RPO/RTO继续按实际拓扑验证,不缩减为本机多进程检查。 5. 已补[`orbit-load`](api/load.md)有界受理负载工具及24次/1000次隔离样本,见[受理报告](reports/phase-6-load.md)。新增独立[`orbit-exec-load`](api/execution-load.md),通过真实Worker、任务结果和尝试历史核对,提供短/混合/中长任务的受理、排队、授权执行及完成耗时分布;具体环境、原始JSON和失败分母见[执行测量报告](reports/phase-6-execution-load.md)。新增[`orbit-workflow-load`](api/workflow-load.md),用本次发布版本的双分支并行流程,逐个核对不同任务、唯一尝试、来源、精确输出与本地调用;24/100实例及24实例截止负例已重新构建回归并绑定源码/二进制哈希,见[流程测量报告](reports/phase-6-workflow-load.md)。新增[`orbit-schedule-load`](api/schedule-load.md),已完成24/128次定时集中触发与24次计划量的截止负例,保存实际分母、唯一任务/尝试、来源与暂停收尾证据,见[定时测量报告](reports/phase-6-schedule-load.md)。Go 1.27.1 模块级门禁、MySQL 存储集成、关键 SDK/Orbit 进程故障组、24个可运行 SDK 示例真实服务串跑和 result-inbox/task-outbox 真实业务集成已复跑通过。已补[长期积压与资源曲线样本](reports/phase-6-backlog.md):32×8 共256个触发点,在1个Worker/容量1和180秒截止下保留173个成功、83个未核验任务,并以独立数据库读取记录等待积压曲线;该单节点样本不替代目标拓扑容量。SDK受控重连风暴及[真实慢消费者背压](reports/phase-6-slow-consumer.md)已补边界回归,积压脚本已加入 CPU/RSS、MySQL 资源、连接池等待、业务快照和事务重试计数采样;已补跨节点重连、慢消费者与有界磁盘增长样本(见 [P1 定向报告](reports/phase-6-stability-p1.md));长期磁盘曲线与目标数据库拓扑容量继续按实际负载验收。 6. 已补[隔离备份恢复演练](api/backup-restore.md):源库和恢复库均为一次性 MySQL 容器,Go 1.27.1 构建的当前二进制在逻辑备份恢复后核对迁移账本、核心记录数量、固定索引和零外键;证据见[恢复报告](reports/phase-6-backup-restore.md)。该演练不替代生产备份介质、复制切换、跨版本恢复或 RPO/RTO 验收。 7. 已补[A18 数据库短暂不可用恢复演练](api/db-recovery.md):Go 1.27.1 构建的当前二进制在数据库停止期间返回 readiness 503,同卷恢复后返回 200,并用原权限范围继续受理任务;证据见[数据库恢复报告](reports/phase-6-db-recovery.md)。该演练不替代主从切换、复制确认、跨故障域或 RPO/RTO 验收。 @@ -236,6 +236,7 @@ A04派发已提交而消息尚未到达Worker的精确进程故障已通过[专 - [P2 自动 Worker 运行器记录](reports/phase-2-worker-runtime.md) - [P2 执行历史与大载荷记录](reports/phase-2-execution-history.md) - [P1 稳定性与归档定向回归](reports/phase-6-stability-p1.md) +- [持久文件归档作业回归](reports/phase-6-retention-jobs.md) ### 状态事件时间线子项 diff --git a/docs/README.md b/docs/README.md index 69627e2..4cab33a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -39,6 +39,8 @@ | [接口手册](api/README.md) | 实际接口状态、调用说明与后续接口规划 | | [SDK 与场景](sdk/README.md) | 可运行示例、接入方法及分阶段 SDK 交付 | | [Orbit Console](console.md) | 内置控制台、本地打开方式与权限边界 | +| [保留策略](api/retention.md) | 管理端窗口配置、候选保护和只读快照 | +| [文件归档作业](api/retention-jobs.md) | 自动有限分批、断点续跑、容量限制和离线核验 | ## 已确认的决策 diff --git a/docs/api/operations.md b/docs/api/operations.md index bcd0c34..1fe32b7 100644 --- a/docs/api/operations.md +++ b/docs/api/operations.md @@ -26,8 +26,8 @@ make release-manifest > orbit-release-manifest.json "workflow_schema_version": 1 }, "schema": { - "current_version": 28, - "migration_count": 28, + "current_version": 30, + "migration_count": 30, "migrations": [ {"version": 1, "name": "000001_initial.up.sql", "checksum": ""} ] @@ -47,7 +47,7 @@ ORBIT_MYSQL_DSN='user:password@tcp(127.0.0.1:3306)/orbit' \ 成功时输出类似: ```json -{"schema_version":28,"migration_count":28,"foreign_key_count":0,"required_index_count":6,"complete":true} +{"schema_version":30,"migration_count":30,"foreign_key_count":0,"required_index_count":6,"complete":true} ``` `complete=true` 表示当前二进制的迁移账本、Orbit 表零外键,以及 000026–000028 的六个有界观测索引均满足要求。需要同时核对发布物身份时,传入保存的清单: @@ -79,3 +79,7 @@ ORBIT_MYSQL_DSN='user:password@tcp(127.0.0.1:3306)/orbit' \ ``` 它与 `/metrics` 共享同一有界读取实现,但不会改变指标注册器或业务状态;遇到数据库超时会返回非零状态,调用方应保留上一次成功快照并排查数据库。 + +## 文件归档与离线核验 + +[`orbit-admin retention archive / verify-archive`](retention-jobs.md) 提供持久文件作业、原目录续跑及摘要链核验。读取命名空间需显式指定已保存策略版本,每个新批次记录独立作业审计,源记录不删除;离线核验无需数据库。 diff --git a/docs/api/retention-jobs.md b/docs/api/retention-jobs.md new file mode 100644 index 0000000..6902211 --- /dev/null +++ b/docs/api/retention-jobs.md @@ -0,0 +1,113 @@ +# 可恢复的文件归档作业 + +`orbit-admin retention archive` 自动按已保存的保留策略分批导出,并把每个批次可靠写入本地持久目录。它面向持有数据库运维权限的操作者,复用控制台导出的候选保护和版本核对,记录 `archive:` 审计身份;不需要控制台 Cookie,也不向业务 SDK 增加管理权限。数据库业务记录始终保留。 + +本次提供的是可重复运行的有限批次作业执行器。周期触发、控制台作业列表、对象存储和清理执行器尚未接入;不能把它描述为已部署的定时归档服务。 + +## 配置与启动 + +先在控制台“保留策略”保存所需窗口并允许导出,记下命名空间 ID 和策略版本。为每次扫描选择独立目录,父目录需提前创建,数据目录应位于受备份保护的持久卷中;进程会创建权限为 `0700` 的作业子目录和 `0600` 文件。 + +```sh +# 按实际数据库注入 ORBIT_MYSQL_DSN,不要把密码放入版本库。 +export ORBIT_MYSQL_DSN='operator:password@tcp(127.0.0.1:3306)/orbit' +mkdir -p /srv/orbit-archives +chmod 700 /srv/orbit-archives + +./bin/orbit-admin retention archive \ + --namespace 11111111111111111111111111111111 \ + --kind tasks \ + --revision 1 \ + --directory /srv/orbit-archives/tasks-20260923 \ + --batch-size 20 \ + --max-batches 10 \ + --max-bytes 268435456 \ + --timeout 5m +``` + +ID、日期、版本均为示例,使用当前环境的值。目录也可通过 `ORBIT_ARCHIVE_DIRECTORY` 配置,显式 `--directory` 优先;该环境变量只供归档 CLI 使用,不会使 Orbit 服务后台自动开始归档。 + +| 参数 | 默认/约束 | +| --- | --- | +| `--namespace` | 必填,32 位小写十六进制命名空间 ID | +| `--kind` | 必填,`tasks/workflows/events/audit/deliveries/inbox/idempotency` | +| `--revision` | 必填,已保存的非零策略版本 | +| `--directory` | 必填绝对路径,或设置 `ORBIT_ARCHIVE_DIRECTORY`;每个作业独立私有目录 | +| `--batch-size` | 默认 20,1–100;与范围一同固定在清单,恢复时不可更改 | +| `--max-batches` | 默认 10,1–1000;本次调用最多新增的批次数 | +| `--max-bytes` | 默认 256 MiB,1 MiB–100 GiB;本作业清单及已发布批次文件的总上限 | +| `--timeout` | 默认 5 分钟,1 秒–1 小时;单批数据库读取仍受 10 秒截止及原有行数/正文限制 | + +所有调用仍核对数据库 Schema;每个新批次在事务内核对应用、命名空间启用状态和当前策略版本。不能使用已关闭或改版的策略继续生成新批次。 + +输出为摘要,不包含正文、凭据或游标: + +```json +{"job_id":"<固定作业ID>","completed":false,"batches":10,"roots":200,"bytes":1234567,"last_sha256":"<清单及批次链的末端摘要>"} +``` + +命令达到 `--max-batches` 后正常返回;`completed=false` 明确表示尚未导出完。使用**相同目录、命名空间、类型、版本和批量**再次运行,自动校验已有批次后接续。可提高本次批数/容量/时间上限,不会改写已发布文件。已完成作业再次运行只验证,不再从头导出。 + +作业执行中遇到数据库错误、策略变更、文件损坏、权限问题或容量不足时输出最后已确认进度并以非零状态退出。参数或数据库连接等启动前校验失败只报错,不输出成功状态;此时可用离线核验命令读取已保存进度。调用方必须同时检查退出码和 `completed`,不能把“进程结束”当成“作业完成”。 + +## 文件结构与恢复 + +```text +/srv/orbit-archives/tasks-20260923/ +├── .lock # 非阻塞进程文件锁;进程退出后自动释放 +├── manifest.json # 不可变范围、随机作业ID及清单版本 +├── batch-000001.json # 输入游标、服务端原始快照、上一文件摘要 +├── batch-000002.json +└── ... +``` + +写入顺序是:数据库读取并记录导出审计 → 私有临时文件 → 文件同步 → 原子发布且拒绝覆盖 → 同步目录 → 原字节回读。没有独立的可变游标文件;连续且校验通过的批次本身就是检查点。 + +- 文件发布前退出:残留 `.write.tmp` 不算成功,恢复时清理该临时文件并重新导出同一输入游标。 +- 文件发布后、进程报告成功前退出:新进程读取已发布批次,从其下一游标恢复。 +- 缺少批次、摘要不一致、混入其他作业、符号链接、非私有文件或未知文件:立即停止,不跳过、不覆盖原档案。 +- 数据库导出已记审计、随后磁盘写入失败:可能多出一次导出审计;审计不表示文件持久化成功。源数据不变,恢复后从最后文件继续。 +- 相同目录的并发进程:后进入者立即失败,避免两个进程同时推进。 + +首批绑定清单原字节摘要,后续每批绑定前一文件的摘要,快照正文另按服务端 SHA-256 校验。保留窗口截止时间跨批固定;每批使用独立数据库快照,整个作业不承诺全库同一时间点。期间解除保护的旧记录可能需要下一次独立扫描。 + +## 离线核验 + +不需要数据库连接或网络: + +```sh +./bin/orbit-admin retention verify-archive \ + --directory /srv/orbit-archives/tasks-20260923 + +# 推荐另外保存成功输出中的 last_sha256,再以可信摘要核对完整链。 +./bin/orbit-admin retention verify-archive \ + --directory /srv/orbit-archives/tasks-20260923 \ + --expected-sha256 '<另行保存的64位小写摘要>' +``` + +核验输出与执行输出结构相同;部分作业也可以核验,`completed=false` 时仍未完成。`--expected-sha256` 可发现文件与摘要整体重写或尾部截断;前提是期望摘要保存在档案目录之外的可信位置。未提供期望摘要时只能验证文件之间的内部一致性,不能证明来源真实性。 + +单批快照在 `batch-*.json` 的 `snapshot` 字段中,仍遵循[归档接口](retention.md)的 `format_version/sha256/content` 协议。 + +## Docker 运行 + +把持久卷挂载到容器目录,并确保镜像内 UID 65532 有权限创建 `0700` 子目录。数据库和路径按目标环境配置;使用当前服务相同的不可变镜像版本: + +```sh +docker run --rm --network orbit-network \ + --env-file /secure/orbit-archive.env \ + --mount type=bind,src=/srv/orbit-archives,dst=/archives \ + --entrypoint /usr/local/bin/orbit-admin orbit:local retention archive \ + --namespace 11111111111111111111111111111111 --kind tasks --revision 1 \ + --directory /archives/tasks-20260923 + +# 离线核验也需要文件锁,挂载目录需允许打开锁文件;不连接数据库。 +docker run --rm --network none \ + --mount type=bind,src=/srv/orbit-archives,dst=/archives \ + --entrypoint /usr/local/bin/orbit-admin orbit:local retention verify-archive \ + --directory /archives/tasks-20260923 +``` + +当前文件执行器支持 Linux/macOS 本地 POSIX 文件锁、硬链接原子发布及文件/目录同步语义;其他系统显式拒绝。不同节点独立磁盘不共享进度;网络共享文件系统必须验证上述语义后才能使用,当前没有跨节点自动接管承诺。 + +容量上限覆盖一个作业的清单和已发布批次,不是整个卷的配额;还需预留一个临时批次空间。已有档案不会自动删除腾空间,不会自动生成新目录重试。每个作业最多 10000 批。该归档不是数据库备份,也不构成源数据删除授权。 diff --git a/docs/api/retention.md b/docs/api/retention.md index 378e845..bae7305 100644 --- a/docs/api/retention.md +++ b/docs/api/retention.md @@ -109,6 +109,10 @@ if page["next_cursor"]: 摘要针对 `content` 的原始 UTF-8 字节,不对解析后重新格式化的 JSON 计算。SHA-256 用于发现内容损坏;摘要与正文一起被篡改时无法证明真实性,需要把审计摘要或外部签名另行保存在可信介质中。 +## 自动分批与持久归档 + +需要在页面之外连续处理并断点续跑时,使用 [`orbit-admin retention archive`](retention-jobs.md)。目录、批量、总字节及执行时限可配置,已发布文件就是恢复检查点,支持离线核验;源业务数据仍保留。周期触发和清理执行器尚未接入。 + ## 只读盘点 `orbit-admin retention inventory` 用于在制定保留窗口、归档和备份恢复方案前盘点历史数据。它只执行带截止时间的统计查询,不删除、搬迁、改写数据,也不会自动判断某条记录可以安全归档。 diff --git a/docs/reports/phase-6-retention-jobs.md b/docs/reports/phase-6-retention-jobs.md new file mode 100644 index 0000000..cda4828 --- /dev/null +++ b/docs/reports/phase-6-retention-jobs.md @@ -0,0 +1,43 @@ +# 持久文件归档作业回归(2026-09-23) + +本批在 `2d8467e` 的只读归档导出之上增加本地持久作业执行器及运维 CLI。无数据库迁移、外键或源数据清理操作;控制台导出的授权方式不变。使用说明与参数见[文件归档作业](../api/retention-jobs.md)。 + +环境:macOS arm64、Go 1.27.1、本地 POSIX 文件系统;Docker MySQL 8.4 与 Linux 容器。证据仅覆盖隔离环境中的文件恢复和数据库行为,不代表共享存储或跨节点归档高可用已经验收。 + +## 文件与进程恢复 + +`go test -race ./internal/retention ./cmd/orbit-admin` 覆盖: + +| 场景 | 核对内容 | +| --- | --- | +| 有限批次与续跑 | 首次仅一批,保留原作业 ID;下次仅导出第二批;完成后重复调用不重复导出 | +| 崩溃残留 | 部分 `.write.tmp` 不算检查点,恢复会重新导出未发布批次 | +| 真实进程强杀 | 子进程第一批落盘后发出就绪标记,父进程 Kill 并 Wait 后,新进程取得锁并只继续第二批 | +| 文件链损坏 | 正文改动、缺批、互换、符号链接、未知文件全部阻止核验与续跑 | +| 容量上限 | 第一批已保存,第二批超过 1 MiB 总额时停止;显式提高预算后继续,第一批原字节不变 | +| 写入失败 | 在非 root 进程中把目录改为只读,第二批写入失败但原检查点仍可核验;恢复写权限后续跑 | +| 取消与并发 | 批次返回后上下文取消不发布文件;相同目录第二个写者被非阻塞文件锁拒绝 | +| 独立可信摘要 | 无 DSN 也能校验;与另行保存的末端摘要不匹配时非零退出 | + +每个批次先同步正文,再原子发布、同步目录并回读。文件权限为 0600,目录为 0700。SHA-256 和前序摘要用于完整性校验;没有独立可信末端摘要时,不声称能抵抗整套档案与摘要一起被重写。 + +## Docker 首次部署闭环 + +`make console-e2e` 从空库迁移、管理员初始化和登录开始,保留原任务/重试/计划/退出后拒绝访问验证,增加: + +- 使用刚保存的保留策略生成一个合法空作业,`completed=true`、`batches=1`、`roots=0`。 +- 在 `--network none` 的独立容器运行 `retention verify-archive`,不注入 DSN;结果与运行输出逐字段一致。 + +数据库提交后的导出审计与文件持久化不是同一事务。文件失败可能导致重试产生新的导出审计;不会改写源记录,不能以审计存在推断文件已经保存。 + +## 边界 + +本批交付可重复执行的有限批次执行器,不新增周期调度服务或控制台作业列表。本地同目录文件锁不能证明跨节点共享文件系统可靠;不同节点独立磁盘也不能互相恢复。新一轮扫描需要独立目录,归档介质备份、定时触发、长期容量、跨节点接管及清理执行器仍需后续实现和验收。 + +## 真实 MySQL 作业结果 + +标准集成入口中的 `TestIntegrationRetentionDiskGrowthAndPreview` 使用 256 条样本,其中 254 条终态且完成时间相同,另 2 条为未知效果/运行租约保护对象。首次文件作业只导出 37 条,下一次从原目录接续,最终 254 条根记录、7 个批次、7 条 `archive:` 审计;离线核验返回相同摘要和计数。停用命名空间返回 `PERMISSION_DENIED`,不能继续导出。控制台原有分页仍为 8 批、254 个唯一根记录,源表保持 256 条。 + +有界磁盘采样观察到表/索引统计 1,835,008 → 16,564,224 字节,增量 14,729,216;预览仍为 100 条样本且有后续候选。该统计只证明本次样本增长,不用于推算生产容量。 + +最终门禁:`make test`(四个模块 race)、`make vet`、`make build`、`make docs-check`、`make console-e2e` 和修正断言后的完整 `make integration` 均通过。全量 MySQL 存储组用时约 286 秒,真实 SDK Worker/Event 进程故障组约 130 秒,最后的双节点流程暂停恢复组约 13 秒;包含独立 Inbox/Outbox 业务示例、Schema 30 及零外键验证。首轮集成中停用命名空间的预期错误码写成了 `FORBIDDEN`,实际服务端正确返回 `PERMISSION_DENIED`;修正测试后重跑全链路通过,没有放宽权限判断。 diff --git a/internal/retention/job.go b/internal/retention/job.go new file mode 100644 index 0000000..6807d9a --- /dev/null +++ b/internal/retention/job.go @@ -0,0 +1,413 @@ +package retention + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" +) + +const maxJobBatches = 10000 +const maxBatchFileBytes = 64 << 20 + +// JobSpec 是首次落盘后不可变的作业范围,不包含凭据、DSN 或任意 SQL。 +type JobSpec struct { + NamespaceID core.ID `json:"namespace_id"` // NamespaceID 唯一命名空间。 + Kind string `json:"kind"` // Kind 固定的历史类别。 + PolicyRevision uint64 `json:"policy_revision,string"` // PolicyRevision 已在控制台确认的策略版本,必须非零。 + BatchSize int `json:"batch_size"` // BatchSize 每批根记录上限,1–100。 +} + +// JobManifest 为作业初始化清单;后续进度由连续且校验通过的批次文件重建。 +type JobManifest struct { + FormatVersion int `json:"format_version"` // FormatVersion 当前为1。 + JobID core.ID `json:"job_id"` // JobID 随机且持久化的作业身份,也是数据库导出审计的执行者标识。 + CreatedAt time.Time `json:"created_at"` // CreatedAt 清单创建时间,UTC,不作为数据库截止时间。 + Spec JobSpec `json:"spec"` // Spec 不可变范围,恢复时必须完全一致。 +} + +// JobBatch 是一个原子发布的归档批次,正文和摘要保持服务端原样。 +type JobBatch struct { + FormatVersion int `json:"format_version"` // FormatVersion 当前为1。 + JobID core.ID `json:"job_id"` // JobID 与清单一致。 + Sequence int `json:"sequence"` // Sequence 从1开始的连续编号,缺批拒绝恢复。 + PreviousSHA256 string `json:"previous_sha256"` // PreviousSHA256 首批绑定清单文件摘要,后续绑定前一批文件摘要。 + InputCursor string `json:"input_cursor"` // InputCursor 此次导出使用的游标,首批为空。 + Snapshot ArchiveSnapshot `json:"snapshot"` // Snapshot 包含原始正文与摘要。 +} + +// JobOptions 控制本次进程的工作量,重试时可显式提高上限,不改变作业身份。 +type JobOptions struct { + MaxBatches int // MaxBatches 本次最多新增批次数,1–1000。 + MaxBytes int64 // MaxBytes 作业清单和全部批次文件总字节上限,1 MiB–100 GiB;不自动删除旧文件腾空间。 +} + +// JobStatus 是可输出到日志的摘要,不包含业务正文或游标。 +type JobStatus struct { + JobID core.ID `json:"job_id"` // JobID 原作业标识,恢复不会生成新标识。 + Completed bool `json:"completed"` // Completed 仅在末批成功持久化且游标为空时为真。 + Batches int `json:"batches"` // Batches 已校验的连续文件数。 + Roots int64 `json:"roots"` // Roots 已归档根记录总数,不包含关联历史行。 + Bytes int64 `json:"bytes"` // Bytes 清单及批次文件实际总字节,不包含临时文件和锁。 + LastSHA256 string `json:"last_sha256"` // LastSHA256 最新批次或初始清单的摘要,用于独立保存核对。 +} + +// ExportBatch 由运维调用方提供,数据库读取和授权仍由存储层完成。 +type ExportBatch func(context.Context, core.ID, ArchiveInput) (ArchiveSnapshot, error) + +func (s JobSpec) Validate() error { + if s.NamespaceID.Validate() != nil || s.PolicyRevision == 0 || s.BatchSize < 1 || s.BatchSize > 100 { + return errors.New("invalid archive namespace, revision or batch size") + } + switch s.Kind { + case "tasks", "workflows", "events", "audit", "deliveries", "inbox", "idempotency": + return nil + } + return errors.New("unsupported archive kind") +} + +// RunArchive 自动处理有限批次,先验证全部已有文件再恢复。每个已发布文件就是检查点, +// 不存在“文件成功但内存游标尚未落盘”的跨资源提交窗口;崩溃留下的临时文件不计为成功。 +func RunArchive(ctx context.Context, directory string, spec JobSpec, options JobOptions, export ExportBatch) (JobStatus, error) { + if err := spec.Validate(); err != nil { + return JobStatus{}, err + } + 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 err != nil { + return JobStatus{}, err + } + defer root.Close() + defer lock.Close() + if err := discardStaging(root); err != nil { + return JobStatus{}, err + } + if _, err := root.Lstat("manifest.json"); errors.Is(err, os.ErrNotExist) { + // 不在未知目录或缺少清单的残留批次上悄悄创建新身份。 + 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 + } + manifest := JobManifest{1, id, time.Now().UTC(), spec} + raw, err := json.Marshal(manifest) + if err != nil { + return JobStatus{}, err + } + if err := publishFile(root, "manifest.json", raw); err != nil { + return JobStatus{}, err + } + } else if err != nil { + return JobStatus{}, err + } + manifest, status, cursor, cutoff, err := readJob(ctx, root, options.MaxBytes) + if err != nil { + return status, err + } + if manifest.Spec != spec { + return status, errors.New("archive scope differs from saved manifest; resume with original arguments") + } + if err := syncJobDirectory(root); err != nil { + return status, err + } + for i := 0; i < options.MaxBatches && !status.Completed; i++ { + if err := ctx.Err(); err != nil { + return status, err + } + if status.Batches >= maxJobBatches { + return status, errors.New("archive batch count limit reached") + } + snapshot, err := export(ctx, manifest.JobID, ArchiveInput{Kind: spec.Kind, Limit: spec.BatchSize, Cursor: cursor, ExpectedRevision: spec.PolicyRevision}) + if err != nil { + return status, err + } + packet := JobBatch{1, manifest.JobID, status.Batches + 1, status.LastSHA256, cursor, snapshot} + next, err := verifyBatch(manifest, packet, status, cursor, cutoff) + if err != nil { + return status, err + } + raw, err := json.Marshal(packet) + if err != nil { + 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") + } + if err := ctx.Err(); err != nil { + return status, err + } + if err := publishFile(root, batchName(packet.Sequence), raw); err != nil { + return status, err + } + // 重新读取原始字节核验,磁盘错误不能推进内存检查点或输出成功。 + saved, err := readJobFile(root, batchName(packet.Sequence), maxBatchFileBytes) + if err != nil || !bytes.Equal(saved, raw) { + return status, errors.New("archive file readback failed; verify job before resuming") + } + advanceJob(&status, &cursor, &cutoff, next, raw) + } + return status, nil +} + +// VerifyArchive 离线逐批校验整个作业,无需数据库或控制台会话;未到末批会返回 Completed=false。 +func VerifyArchive(ctx context.Context, directory string) (JobStatus, error) { + root, lock, err := openJob(directory, false) + if err != nil { + return JobStatus{}, err + } + defer root.Close() + defer lock.Close() + _, status, _, _, err := readJob(ctx, root, 100<<30) + return status, err +} + +func openJob(directory string, create bool) (*os.Root, *os.File, error) { + if !filepath.IsAbs(directory) { + return nil, nil, errors.New("archive directory must be an absolute path") + } + if create { + if err := os.Mkdir(directory, 0700); err == nil { + parent, err := os.Open(filepath.Dir(directory)) + if err != nil { + return nil, nil, err + } + err = parent.Sync() + parent.Close() + if err != nil { + return nil, nil, err + } + } else if !errors.Is(err, os.ErrExist) { + return nil, nil, err + } + } + info, err := os.Lstat(directory) + if err != nil { + return nil, nil, err + } + if !info.IsDir() || info.Mode().Perm()&0077 != 0 { + return nil, nil, errors.New("archive directory must be a private directory (0700), not a symlink") + } + root, err := os.OpenRoot(directory) + if err != nil { + return nil, nil, err + } + lock, err := lockJob(root) + if err != nil { + root.Close() + return nil, nil, err + } + return root, lock, nil +} + +func jobEntries(root *os.Root) ([]os.DirEntry, error) { + dir, err := root.Open(".") + if err != nil { + return nil, err + } + defer dir.Close() + entries, err := dir.ReadDir(maxJobBatches + 4) + if err != nil && !errors.Is(err, io.EOF) { + return nil, err + } + if len(entries) >= maxJobBatches+4 { + return nil, errors.New("archive directory contains too many files") + } + return entries, nil +} + +func readJob(ctx context.Context, root *os.Root, maxBytes int64) (JobManifest, JobStatus, string, time.Time, error) { + var manifest JobManifest + var status JobStatus + var cursor string + var cutoff time.Time + fail := func(err error) (JobManifest, JobStatus, string, time.Time, error) { + return manifest, status, cursor, cutoff, err + } + raw, err := readJobFile(root, "manifest.json", 65536) + if err != nil { + return fail(err) + } + if decodeJobJSON(raw, &manifest) != nil || manifest.FormatVersion != 1 || manifest.JobID.Validate() != nil || manifest.CreatedAt.IsZero() || manifest.Spec.Validate() != nil { + return fail(errors.New("invalid archive manifest")) + } + status.JobID, status.Bytes, status.LastSHA256 = manifest.JobID, int64(len(raw)), digest(raw) + entries, err := jobEntries(root) + if err != nil { + return fail(err) + } + files := make(map[string]bool, len(entries)) + for _, e := range entries { + files[e.Name()] = true + } + delete(files, ".lock") + delete(files, "manifest.json") + // 临时文件不属于已发布链,离线校验不会删除它;恢复运行时再清理。 + if files[".write.tmp"] { + if _, err := readJobFile(root, ".write.tmp", maxBatchFileBytes); err != nil { + return fail(err) + } + delete(files, ".write.tmp") + } + seen := map[string]bool{} + for files[batchName(status.Batches+1)] { + if err := ctx.Err(); err != nil { + return fail(err) + } + name := batchName(status.Batches + 1) + raw, err := readJobFile(root, name, maxBatchFileBytes) + if err != nil { + return fail(err) + } + if int64(len(raw)) > maxBytes-status.Bytes { + return fail(errors.New("archive byte limit reached")) + } + var packet JobBatch + if decodeJobJSON(raw, &packet) != nil { + return fail(errors.New("invalid archive batch JSON")) + } + next, err := verifyBatch(manifest, packet, status, cursor, cutoff) + if err != nil { + return fail(err) + } + if next.NextCursor != "" && seen[next.NextCursor] { + return fail(errors.New("archive cursor cycle detected")) + } + seen[next.NextCursor] = true + advanceJob(&status, &cursor, &cutoff, next, raw) + delete(files, name) + } + if len(files) != 0 { + return fail(errors.New("archive has missing batches or unexpected files")) + } + return manifest, status, cursor, cutoff, nil +} + +type archivePage struct { + Policy Policy `json:"policy"` // Policy 本批数据库策略。 + Kind string `json:"kind"` // Kind 本批根类别。 + Cutoff time.Time `json:"cutoff"` // Cutoff 所有批次冻结的UTC边界。 + RootCount int `json:"root_count"` // RootCount 本批根数量。 + SourceRetained bool `json:"source_retained"` // SourceRetained 必须为真。 + NextCursor string `json:"next_cursor"` // NextCursor 空表示末批。 +} + +func verifyBatch(m JobManifest, b JobBatch, status JobStatus, cursor string, cutoff time.Time) (archivePage, error) { + var page archivePage + if status.Completed || b.FormatVersion != 1 || b.JobID != m.JobID || b.Sequence != status.Batches+1 || b.PreviousSHA256 != status.LastSHA256 || b.InputCursor != cursor || b.Snapshot.FormatVersion != 1 || len(b.Snapshot.Content) > 8<<20 || digest([]byte(b.Snapshot.Content)) != b.Snapshot.SHA256 { + return page, errors.New("archive batch identity or digest mismatch") + } + if json.Unmarshal([]byte(b.Snapshot.Content), &page) != nil || page.Policy.NamespaceID != m.Spec.NamespaceID || page.Policy.Revision != m.Spec.PolicyRevision || !page.Policy.Enabled || page.Kind != m.Spec.Kind || !page.SourceRetained || page.Cutoff.IsZero() || (!cutoff.IsZero() && !page.Cutoff.Equal(cutoff)) || page.RootCount < 0 || page.RootCount > m.Spec.BatchSize || len(page.NextCursor) > 2048 || (page.NextCursor != "" && (page.NextCursor == cursor || page.RootCount == 0)) { + return page, errors.New("archive batch scope or continuation mismatch") + } + return page, nil +} + +func advanceJob(status *JobStatus, cursor *string, cutoff *time.Time, page archivePage, raw []byte) { + status.Batches++ + status.Roots += int64(page.RootCount) + status.Bytes += int64(len(raw)) + status.LastSHA256 = digest(raw) + status.Completed = page.NextCursor == "" + *cursor, *cutoff = page.NextCursor, page.Cutoff +} + +func digest(raw []byte) string { sum := sha256.Sum256(raw); return hex.EncodeToString(sum[:]) } +func batchName(n int) string { return fmt.Sprintf("batch-%06d.json", n) } +func decodeJobJSON(raw []byte, value any) error { + d := json.NewDecoder(bytes.NewReader(raw)) + d.DisallowUnknownFields() + if err := d.Decode(value); err != nil { + return err + } + if err := d.Decode(new(any)); err != io.EOF { + return errors.New("trailing archive JSON") + } + return nil +} + +func readJobFile(root *os.Root, name string, limit int64) ([]byte, error) { + info, err := root.Lstat(name) + if err != nil { + return nil, err + } + if !info.Mode().IsRegular() || info.Mode().Perm()&0077 != 0 || info.Size() > limit { + return nil, errors.New("archive file must be bounded, regular and private (0600)") + } + f, err := root.Open(name) + if err != nil { + return nil, err + } + defer f.Close() + raw, err := io.ReadAll(io.LimitReader(f, limit+1)) + if err != nil { + return nil, err + } + if int64(len(raw)) > limit { + return nil, errors.New("archive file too large") + } + return raw, nil +} + +func discardStaging(root *os.Root) error { + if _, err := root.Lstat(".write.tmp"); errors.Is(err, os.ErrNotExist) { + return nil + } else if err != nil { + return err + } + if _, err := readJobFile(root, ".write.tmp", maxBatchFileBytes); err != nil { + return err + } + return root.Remove(".write.tmp") +} + +// publishFile 先写私有临时文件并 fsync,再以硬链接原子发布(拒绝覆盖),最后同步目录。 +// 仅依赖本地 POSIX 文件系统语义;网络文件系统和多节点共享盘须独立验证。 +func publishFile(root *os.Root, name string, raw []byte) error { + f, err := root.OpenFile(".write.tmp", os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) + if err != nil { + return err + } + defer root.Remove(".write.tmp") + if _, err := f.Write(raw); err != nil { + f.Close() + return err + } + err = f.Sync() + closeErr := f.Close() + if err != nil { + return err + } + if closeErr != nil { + return closeErr + } + if err := root.Link(".write.tmp", name); err != nil { + return err + } + if err := root.Remove(".write.tmp"); err != nil { + return err + } + return syncJobDirectory(root) +} + +func syncJobDirectory(root *os.Root) error { + dir, err := root.Open(".") + if err != nil { + return err + } + defer dir.Close() + return dir.Sync() +} diff --git a/internal/retention/job_lock_other.go b/internal/retention/job_lock_other.go new file mode 100644 index 0000000..ca3f245 --- /dev/null +++ b/internal/retention/job_lock_other.go @@ -0,0 +1,12 @@ +//go:build !linux && !darwin + +package retention + +import ( + "errors" + "os" +) + +func lockJob(root *os.Root) (*os.File, error) { + return nil, errors.New("archive file jobs require Linux or macOS POSIX locking") +} diff --git a/internal/retention/job_lock_unix.go b/internal/retention/job_lock_unix.go new file mode 100644 index 0000000..c08c69e --- /dev/null +++ b/internal/retention/job_lock_unix.go @@ -0,0 +1,27 @@ +//go:build linux || darwin + +package retention + +import ( + "errors" + "os" + "syscall" +) + +// lockJob 的非阻塞文件锁随进程退出自动释放,不使用永不过期的“锁目录”。 +func lockJob(root *os.Root) (*os.File, error) { + f, err := root.OpenFile(".lock", os.O_CREATE|os.O_RDWR|syscall.O_NOFOLLOW, 0600) + if err != nil { + return nil, err + } + info, err := f.Stat() + if err != nil || !info.Mode().IsRegular() || info.Mode().Perm()&0077 != 0 { + f.Close() + return nil, errors.New("invalid archive lock file") + } + 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") + } + return f, nil +} diff --git a/internal/retention/job_test.go b/internal/retention/job_test.go new file mode 100644 index 0000000..2fb650e --- /dev/null +++ b/internal/retention/job_test.go @@ -0,0 +1,261 @@ +package retention + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" +) + +func testJobSpec() JobSpec { + return JobSpec{NamespaceID: "11111111111111111111111111111111", Kind: "tasks", PolicyRevision: 1, BatchSize: 1} +} +func testJobOptions() JobOptions { return JobOptions{MaxBatches: 10, MaxBytes: 16 << 20} } +func testJobSnapshot(t *testing.T, cursor string, payload int) ArchiveSnapshot { + t.Helper() + p := Defaults(testJobSpec().NamespaceID) + p.Enabled = true + p.Revision = 1 + next := "next-page" + if cursor != "" { + next = "" + } + raw, err := json.Marshal(map[string]any{"policy": p, "kind": "tasks", "cutoff": time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC), "root_count": 1, "source_retained": true, "next_cursor": next, "payload": strings.Repeat("x", payload)}) + if err != nil { + t.Fatal(err) + } + return ArchiveSnapshot{1, digest(raw), string(raw)} +} +func testJobExporter(t *testing.T, calls *int) ExportBatch { + return func(ctx context.Context, job core.ID, in ArchiveInput) (ArchiveSnapshot, error) { + *calls++ + if job.Validate() != nil || in.ExpectedRevision != 1 || in.Kind != "tasks" || in.Limit != 1 { + t.Fatal("export request changed identity") + } + return testJobSnapshot(t, in.Cursor, 0), nil + } +} + +func TestArchiveJobResumeAndOfflineVerification(t *testing.T) { + dir := filepath.Join(t.TempDir(), "job") + calls := 0 + options := testJobOptions() + options.MaxBatches = 1 + first, err := RunArchive(context.Background(), dir, testJobSpec(), options, testJobExporter(t, &calls)) + if err != nil || first.Completed || first.Roots != 1 || first.Batches != 1 { + t.Fatalf("first=%+v err=%v", first, err) + } + // 模拟强杀前只写了一部分临时文件;它不是有效检查点。 + if err := os.WriteFile(filepath.Join(dir, ".write.tmp"), []byte(`{"unfinished":`), 0600); err != nil { + t.Fatal(err) + } + second, err := RunArchive(context.Background(), dir, testJobSpec(), options, testJobExporter(t, &calls)) + if err != nil || !second.Completed || second.Roots != 2 || second.Batches != 2 || second.JobID != first.JobID || calls != 2 { + t.Fatalf("resume=%+v calls=%d err=%v", second, calls, err) + } + verified, err := VerifyArchive(context.Background(), dir) + if err != nil || verified != second { + t.Fatalf("verify=%+v err=%v", verified, err) + } + repeated, err := RunArchive(context.Background(), dir, testJobSpec(), options, testJobExporter(t, &calls)) + if err != nil || repeated != second || calls != 2 { + t.Fatalf("completed job re-exported: %+v err=%v", repeated, err) + } + spec := testJobSpec() + spec.PolicyRevision++ + if _, err := RunArchive(context.Background(), dir, spec, options, testJobExporter(t, &calls)); err == nil || calls != 2 { + t.Fatal("changed scope accepted") + } + for _, name := range []string{"manifest.json", "batch-000001.json", "batch-000002.json"} { + info, err := os.Stat(filepath.Join(dir, name)) + if err != nil || info.Mode().Perm() != 0600 { + t.Fatalf("file permissions %s: %v", name, err) + } + } +} + +func TestArchiveJobRejectsDamagedChain(t *testing.T) { + for _, kind := range []string{"body", "missing", "swapped", "symlink", "unexpected"} { + t.Run(kind, func(t *testing.T) { + dir := filepath.Join(t.TempDir(), "job") + calls := 0 + if _, err := RunArchive(context.Background(), dir, testJobSpec(), testJobOptions(), testJobExporter(t, &calls)); err != nil { + t.Fatal(err) + } + first := filepath.Join(dir, "batch-000001.json") + second := filepath.Join(dir, "batch-000002.json") + var err error + switch kind { + case "body": + raw, _ := os.ReadFile(first) + var packet JobBatch + if err = json.Unmarshal(raw, &packet); err != nil { + t.Fatal(err) + } + packet.Snapshot.Content += " " + raw, _ = json.Marshal(packet) + err = os.WriteFile(first, raw, 0600) + case "missing": + err = os.Remove(first) + case "swapped": + raw, _ := os.ReadFile(second) + err = os.WriteFile(first, raw, 0600) + case "symlink": + if err = os.Remove(first); err == nil { + err = os.Symlink(second, first) + } + case "unexpected": + err = os.WriteFile(filepath.Join(dir, "unrelated.json"), []byte("{}"), 0600) + } + if err != nil { + t.Fatal(err) + } + if _, err := VerifyArchive(context.Background(), dir); err == nil { + t.Fatal("damaged archive accepted") + } + if _, err := RunArchive(context.Background(), dir, testJobSpec(), testJobOptions(), testJobExporter(t, &calls)); err == nil || calls != 2 { + t.Fatalf("resume skipped corrupt history: calls=%d err=%v", calls, err) + } + }) + } +} + +func TestArchiveJobQuotaFailureAndCancellationPreserveProgress(t *testing.T) { + dir := filepath.Join(t.TempDir(), "quota") + options := testJobOptions() + options.MaxBytes = 1 << 20 + calls := 0 + export := func(ctx context.Context, id core.ID, in ArchiveInput) (ArchiveSnapshot, error) { + calls++ + return testJobSnapshot(t, in.Cursor, 700<<10), nil + } + status, err := RunArchive(context.Background(), dir, testJobSpec(), options, export) + if err == nil || status.Completed || status.Batches != 1 { + t.Fatalf("quota=%+v err=%v", status, err) + } + first, _ := os.ReadFile(filepath.Join(dir, "batch-000001.json")) + options.MaxBytes = 4 << 20 + final, err := RunArchive(context.Background(), dir, testJobSpec(), options, export) + if err != nil || !final.Completed || final.Batches != 2 || calls != 3 { + t.Fatalf("quota resume=%+v calls=%d err=%v", final, calls, err) + } + after, _ := os.ReadFile(filepath.Join(dir, "batch-000001.json")) + if string(first) != string(after) { + t.Fatal("prior file overwritten") + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + status, err = RunArchive(ctx, filepath.Join(t.TempDir(), "cancel"), testJobSpec(), testJobOptions(), func(ctx context.Context, id core.ID, in ArchiveInput) (ArchiveSnapshot, error) { + cancel() + return testJobSnapshot(t, "", 0), nil + }) + if !errors.Is(err, context.Canceled) || status.Batches != 0 { + t.Fatalf("cancellation advanced progress: %+v %v", status, err) + } +} + +func TestArchiveJobExclusiveLock(t *testing.T) { + dir := filepath.Join(t.TempDir(), "job") + root, lock, err := openJob(dir, true) + if err != nil { + t.Fatal(err) + } + defer root.Close() + defer lock.Close() + calls := 0 + if _, err := RunArchive(context.Background(), dir, testJobSpec(), testJobOptions(), testJobExporter(t, &calls)); err == nil || calls != 0 { + t.Fatal("parallel writer accepted") + } +} + +func TestArchiveJobProcessKillAndResume(t *testing.T) { + if dir := os.Getenv("ORBIT_ARCHIVE_KILL_TEST_DIR"); dir != "" { + _, err := RunArchive(context.Background(), dir, testJobSpec(), testJobOptions(), func(ctx context.Context, id core.ID, in ArchiveInput) (ArchiveSnapshot, error) { + if in.Cursor != "" { + os.Stdout.WriteString("first-batch-durable\n") + select {} + } + return testJobSnapshot(t, "", 0), nil + }) + if err != nil { + t.Fatal(err) + } + return + } + dir := filepath.Join(t.TempDir(), "job") + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestArchiveJobProcessKillAndResume$") + cmd.Env = append(os.Environ(), "ORBIT_ARCHIVE_KILL_TEST_DIR="+dir) + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + defer func() { + if cmd.ProcessState == nil { + cmd.Process.Kill() + cmd.Wait() + } + }() + scanner := bufio.NewScanner(stdout) + if !scanner.Scan() || scanner.Text() != "first-batch-durable" { + t.Fatal("child did not publish first batch") + } + if err := cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + if err := cmd.Wait(); err == nil { + t.Fatal("child was not killed") + } + // 等待旧进程真正退出后,新进程重新取得锁,只导出第二批。 + calls := 0 + status, err := RunArchive(context.Background(), dir, testJobSpec(), testJobOptions(), testJobExporter(t, &calls)) + if err != nil || !status.Completed || status.Roots != 2 || status.Batches != 2 || calls != 1 { + t.Fatalf("killed-process recovery=%+v calls=%d err=%v", status, calls, err) + } +} + +func TestArchiveJobWriteFailureDoesNotAdvance(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("permission failure requires an unprivileged process") + } + dir := filepath.Join(t.TempDir(), "job") + calls := 0 + options := testJobOptions() + options.MaxBatches = 1 + first, err := RunArchive(context.Background(), dir, testJobSpec(), options, testJobExporter(t, &calls)) + if err != nil { + t.Fatal(err) + } + if err := os.Chmod(dir, 0500); err != nil { + t.Fatal(err) + } + defer os.Chmod(dir, 0700) + after, err := RunArchive(context.Background(), dir, testJobSpec(), options, testJobExporter(t, &calls)) + if err == nil || after != first { + t.Fatalf("failed write advanced checkpoint: %+v err=%v", after, err) + } + verified, err := VerifyArchive(context.Background(), dir) + if err != nil || verified != first { + t.Fatalf("saved archive changed: %+v err=%v", verified, err) + } + if err := os.Chmod(dir, 0700); err != nil { + t.Fatal(err) + } + final, err := RunArchive(context.Background(), dir, testJobSpec(), options, testJobExporter(t, &calls)) + if err != nil || !final.Completed || final.Roots != 2 { + t.Fatalf("write recovery=%+v err=%v", final, err) + } +} diff --git a/internal/storage/mysql/retention_archive.go b/internal/storage/mysql/retention_archive.go index e54cdd6..5e0ddfa 100644 --- a/internal/storage/mysql/retention_archive.go +++ b/internal/storage/mysql/retention_archive.go @@ -120,6 +120,32 @@ type archiveData struct { } func (s *RetentionPolicyStore) ExportRetention(ctx context.Context, t auth.Token, in retention.ArchiveInput) (retention.ArchiveSnapshot, error) { + return s.exportRetention(ctx, t.NamespaceID, in, func(ctx context.Context, tx *sql.Tx) (string, error) { + actor, err := retentionActor(ctx, tx, t) + return "console:" + string(actor), err + }) +} + +// ExportRetentionForJob 仅供持有数据库管理权限的归档 CLI 使用,不注册为 HTTP 路由。 +// 每批核验应用/命名空间启用状态和数据库策略版本;作业标识进入审计,不能冒充控制台用户。 +func (s *RetentionPolicyStore) ExportRetentionForJob(ctx context.Context, namespace, job core.ID, in retention.ArchiveInput) (retention.ArchiveSnapshot, error) { + if namespace.Validate() != nil || job.Validate() != nil { + return retention.ArchiveSnapshot{}, fault.Invalid("invalid archive job identity") + } + return s.exportRetention(ctx, namespace, in, func(ctx context.Context, tx *sql.Tx) (string, error) { + var enabled bool + 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`, namespace).Scan(&enabled) + if err != nil { + return "", err + } + if !enabled { + return "", fault.Forbidden + } + return "archive:" + string(job), nil + }) +} + +func (s *RetentionPolicyStore) exportRetention(ctx context.Context, namespace core.ID, in retention.ArchiveInput, authorize func(context.Context, *sql.Tx) (string, error)) (retention.ArchiveSnapshot, error) { ctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() if in.Limit == 0 { @@ -134,11 +160,11 @@ func (s *RetentionPolicyStore) ExportRetention(ctx context.Context, t auth.Token return retention.ArchiveSnapshot{}, err } defer tx.Rollback() - actor, err := retentionActor(ctx, tx, t) + actor, err := authorize(ctx, tx) if err != nil { return retention.ArchiveSnapshot{}, err } - p, err := s.readPolicy(ctx, tx, t.NamespaceID, true) + p, err := s.readPolicy(ctx, tx, namespace, true) if err != nil { return retention.ArchiveSnapshot{}, err } @@ -166,13 +192,13 @@ func (s *RetentionPolicyStore) ExportRetention(ctx context.Context, t auth.Token } b, e := base64.RawURLEncoding.DecodeString(in.Cursor) var c archiveCursor - if e != nil || json.Unmarshal(b, &c) != nil || c.Namespace != t.NamespaceID || c.Kind != in.Kind || c.Revision != p.Revision || c.ID.Validate() != nil || c.Cutoff.After(cutoff) || c.Time.IsZero() || !c.Time.Before(c.Cutoff) { + if e != nil || json.Unmarshal(b, &c) != nil || c.Namespace != namespace || c.Kind != in.Kind || c.Revision != p.Revision || c.ID.Validate() != nil || c.Cutoff.After(cutoff) || c.Time.IsZero() || !c.Time.Before(c.Cutoff) { return retention.ArchiveSnapshot{}, fault.Invalid("invalid archive cursor") } cursor = &c cutoff = c.Cutoff } - roots, err := archiveCandidates(ctx, tx, t.NamespaceID, source, cutoff, in.Limit+1, cursor) + roots, err := archiveCandidates(ctx, tx, namespace, source, cutoff, in.Limit+1, cursor) if err != nil { return retention.ArchiveSnapshot{}, err } @@ -180,7 +206,7 @@ func (s *RetentionPolicyStore) ExportRetention(ctx context.Context, t auth.Token if len(roots) > in.Limit { roots = roots[:in.Limit] last := roots[len(roots)-1] - raw, _ := json.Marshal(archiveCursor{t.NamespaceID, in.Kind, p.Revision, cutoff, last.at, last.id}) + raw, _ := json.Marshal(archiveCursor{namespace, in.Kind, p.Revision, cutoff, last.at, last.id}) data.NextCursor = base64.RawURLEncoding.EncodeToString(raw) } data.RootCount = len(roots) @@ -206,10 +232,10 @@ func (s *RetentionPolicyStore) ExportRetention(ctx context.Context, t auth.Token } } // 流程固定版本与全部历史关联任务也一起导出,不能只保存实例头。 - if err := archiveRows(ctx, tx, &data, &budget, "weave_definition_versions", "id IN (SELECT definition_version_id FROM weave_instances WHERE namespace_id=? AND id=?)", t.NamespaceID, root.id); err != nil { + if err := archiveRows(ctx, tx, &data, &budget, "weave_definition_versions", "id IN (SELECT definition_version_id FROM weave_instances WHERE namespace_id=? AND id=?)", namespace, root.id); err != nil { return retention.ArchiveSnapshot{}, err } - taskRows, err := tx.QueryContext(ctx, `SELECT id FROM pulse_tasks WHERE namespace_id=? AND gate_id=(SELECT gate_id FROM weave_instances WHERE namespace_id=? AND id=?) LIMIT 10001`, t.NamespaceID, t.NamespaceID, root.id) + taskRows, err := tx.QueryContext(ctx, `SELECT id FROM pulse_tasks WHERE namespace_id=? AND gate_id=(SELECT gate_id FROM weave_instances WHERE namespace_id=? AND id=?) LIMIT 10001`, namespace, namespace, root.id) if err != nil { return retention.ArchiveSnapshot{}, err } @@ -265,7 +291,7 @@ func (s *RetentionPolicyStore) ExportRetention(ctx context.Context, t auth.Token return retention.ArchiveSnapshot{}, err } info, _ := json.Marshal(map[string]any{"kind": in.Kind, "policy_revision": p.Revision, "root_count": len(roots), "rows": budget.rows, "sha256": digest, "source_retained": true}) - _, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,`after`) VALUES (?,?,?,?,'retention.archive.export','retention_policy',?,'export retained history snapshot',?)", t.NamespaceID, audit, op, "console:"+string(actor), t.NamespaceID, string(info)) + _, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,`after`) VALUES (?,?,?,?,'retention.archive.export','retention_policy',?,'export retained history snapshot',?)", namespace, audit, op, actor, namespace, string(info)) if err != nil { return retention.ArchiveSnapshot{}, err } diff --git a/internal/storage/mysql/retention_policy_integration_test.go b/internal/storage/mysql/retention_policy_integration_test.go index 06958db..0c188b4 100644 --- a/internal/storage/mysql/retention_policy_integration_test.go +++ b/internal/storage/mysql/retention_policy_integration_test.go @@ -8,12 +8,14 @@ import ( "encoding/hex" "encoding/json" "os" + "path/filepath" "strings" "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" pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" "git.yuntuo.hk/k2safe/orbit/pulse/service" @@ -300,6 +302,40 @@ func TestIntegrationRetentionDiskGrowthAndPreview(t *testing.T) { if _, err := store.ExportRetention(ctx, apiToken, retention.ArchiveInput{Kind: "tasks", ExpectedRevision: 1}); err == nil { t.Fatal("business bearer token accepted as console admin") } + + jobSpec := retention.JobSpec{NamespaceID: boot.NamespaceID, Kind: "tasks", PolicyRevision: 1, BatchSize: 37} + jobDir := filepath.Join(t.TempDir(), "retention-job") + jobOptions := retention.JobOptions{MaxBatches: 1, MaxBytes: 64 << 20} + exportJob := func(ctx context.Context, job core.ID, in retention.ArchiveInput) (retention.ArchiveSnapshot, error) { + return store.ExportRetentionForJob(ctx, boot.NamespaceID, job, in) + } + firstJob, err := retention.RunArchive(ctx, jobDir, jobSpec, jobOptions, exportJob) + if err != nil || firstJob.Completed || firstJob.Roots != 37 { + t.Fatalf("archive job first batch=%+v err=%v", firstJob, err) + } + jobOptions.MaxBatches = 10 + finishedJob, err := retention.RunArchive(ctx, jobDir, jobSpec, jobOptions, exportJob) + if err != nil || !finishedJob.Completed || finishedJob.Roots != 254 || finishedJob.Batches != 7 || finishedJob.JobID != firstJob.JobID { + t.Fatalf("archive job resumed=%+v err=%v", finishedJob, err) + } + verifiedJob, err := retention.VerifyArchive(ctx, jobDir) + if err != nil || verifiedJob != finishedJob { + t.Fatalf("offline job verification=%+v err=%v", verifiedJob, err) + } + var jobAudits int + if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND actor_id=? AND action='retention.archive.export'", boot.NamespaceID, "archive:"+string(finishedJob.JobID)).Scan(&jobAudits); err != nil || jobAudits != 7 { + t.Fatalf("job audit count=%d err=%v", jobAudits, err) + } + if _, err := db.ExecContext(ctx, "UPDATE orbit_namespaces SET enabled=FALSE WHERE id=?", boot.NamespaceID); err != nil { + t.Fatal(err) + } + if _, err := store.ExportRetentionForJob(ctx, boot.NamespaceID, finishedJob.JobID, retention.ArchiveInput{Kind: "tasks", ExpectedRevision: 1}); faultCode(err) != "PERMISSION_DENIED" { + t.Fatalf("disabled namespace exported: %v", err) + } + if _, err := db.ExecContext(ctx, "UPDATE orbit_namespaces SET enabled=TRUE WHERE id=?", boot.NamespaceID); err != nil { + t.Fatal(err) + } + t.Logf("durable archive job: roots=%d batches=%d audit_rows=%d offline_verified=true", finishedJob.Roots, finishedJob.Batches, jobAudits) policyChange := retention.UpdateInput{ExpectedRevision: 1, Enabled: true, IdempotencyDays: 90, TaskHistoryDays: 2, WorkflowHistoryDays: 30, EventLogDays: 30, DeliveryHistoryDays: 30, InboxHistoryDays: 30, AuditLogDays: 365, Reason: "verify frozen cursor boundary"} if _, err := store.UpdateRetentionPolicy(ctx, consoleToken, "growth-retention-revise", policyChange); err != nil { t.Fatal(err) diff --git a/scripts/test-console-e2e.sh b/scripts/test-console-e2e.sh index bf78c41..c88fc43 100755 --- a/scripts/test-console-e2e.sh +++ b/scripts/test-console-e2e.sh @@ -15,6 +15,12 @@ E2E_CREDENTIALS_FILE="$E2E_DIR/credentials.json" cleanup() { "$E2E_DOCKER" stop "$E2E_ORBIT" "$E2E_MYSQL" >/dev/null 2>&1 || true "$E2E_DOCKER" container rm "$E2E_ORBIT" "$E2E_MYSQL" >/dev/null 2>&1 || true + if [ -d "$E2E_DIR/archive-job" ]; then + for e2e_artifact in "$E2E_DIR/archive-job"/.lock "$E2E_DIR/archive-job"/manifest.json "$E2E_DIR/archive-job"/batch-000001.json; do + if [ -f "$e2e_artifact" ]; then unlink "$e2e_artifact"; fi + done + rmdir "$E2E_DIR/archive-job" >/dev/null 2>&1 || true + fi for e2e_artifact in "$E2E_DIR"/*; do if [ -f "$e2e_artifact" ]; then unlink "$e2e_artifact" @@ -250,4 +256,25 @@ assert status_retention == 401, status_retention 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 +printf '%s\n' "[archive] durable job and offline verification" +E2E_NAMESPACE=$(python3 - "$E2E_CREDENTIALS_FILE" <<'PYCODE' +import json, sys +print(json.load(open(sys.argv[1], encoding="utf-8"))["namespace_id"]) +PYCODE +) +"$E2E_DOCKER" run --rm --network bridge --link "$E2E_MYSQL:mysql" \ + --env ORBIT_MYSQL_DSN="$E2E_DSN" --volume "$E2E_DIR:/run/orbit" \ + --entrypoint /usr/local/bin/orbit-admin "$E2E_IMAGE" retention archive \ + --namespace "$E2E_NAMESPACE" --kind tasks --revision 1 --directory /run/orbit/archive-job > "$E2E_DIR/archive-result.json" +"$E2E_DOCKER" run --rm --network none --volume "$E2E_DIR:/run/orbit" \ + --entrypoint /usr/local/bin/orbit-admin "$E2E_IMAGE" retention verify-archive \ + --directory /run/orbit/archive-job > "$E2E_DIR/archive-verified.json" +python3 - "$E2E_DIR/archive-result.json" "$E2E_DIR/archive-verified.json" <<'PYCODE' +import json, sys +result, verified = [json.load(open(p, encoding="utf-8")) for p in sys.argv[1:]] +assert result == verified +assert result["completed"] is True and result["batches"] == 1 and result["roots"] == 0 +print("durable_archive=ok offline_verification=ok roots=0 completed=true") +PYCODE + printf '%s\n' "console-e2e=passed image=$E2E_IMAGE url=$E2E_URL"