diff --git a/.env.example b/.env.example index 81263ea..4d76b3a 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,15 @@ ORBIT_EVENT_CONNECTION_LIMIT=128 # 每个 WebSocket Hub 的数据库领取并发和有界内存排队，范围分别为1–256、1–65536。 ORBIT_DISPATCH_WORKERS=4 ORBIT_DISPATCH_CAPACITY=128 +# 归档策略首次启动时的数据库默认值；保存到管理端后，以数据库版本为准。 +# Enabled 没有环境变量，默认关闭；窗口仅允许1–3650天，幂等记录不得少于90天。 +ORBIT_RETENTION_IDEMPOTENCY_DAYS=90 +ORBIT_RETENTION_TASK_DAYS=365 +ORBIT_RETENTION_WORKFLOW_DAYS=365 +ORBIT_RETENTION_EVENT_DAYS=365 +ORBIT_RETENTION_DELIVERY_DAYS=90 +ORBIT_RETENTION_INBOX_DAYS=90 +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= diff --git a/Makefile b/Makefile index 12272e4..f0abb76 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ GO ?= go -.PHONY: fmt test vet build image integration console-e2e compose-check mysql-replication mysql-compatibility sdk-check protocol-check docs-check schedule-backlog release-manifest release-local-check +.PHONY: fmt test vet build image integration console-e2e compose-check mysql-replication mysql-compatibility sdk-check protocol-check docs-check stability-check schedule-backlog release-manifest release-local-check fmt: $(GO) fmt ./... cd sdk/go && $(GO) fmt ./... @@ -35,6 +35,9 @@ integration: console-e2e: ./scripts/test-console-e2e.sh +stability-check: + ./scripts/test-stability.sh + compose-check: @test -n "$${ORBIT_DB_USER:-}" || (echo 'ORBIT_DB_USER is required' >&2; exit 1) @test -n "$${ORBIT_DB_PASSWORD:-}" || (echo 'ORBIT_DB_PASSWORD is required' >&2; exit 1) diff --git a/cmd/orbit/main.go b/cmd/orbit/main.go index 08aabe1..5815a3a 100644 --- a/cmd/orbit/main.go +++ b/cmd/orbit/main.go @@ -85,6 +85,7 @@ func run() error { metrics.RecordTransactionStats(observability.TransactionStats{Retries: txStats.Retries, Exhausted: txStats.Exhausted}) } consoleAuth := mysql.NewConsoleAuthStore(db) + retentionStore := mysql.NewRetentionPolicyStore(db, cfg.RetentionDefaults) server := httpserver.NewWithConsoleAuth(cfg.HTTPAddr, readiness.Check, consoleAuth, service.New(mysql.NewTaskStore(db))) httpserver.RegisterApplicationResources(server, resources.New(mysql.NewTaskStore(db))) node, err := core.NewID() @@ -121,6 +122,7 @@ func run() error { httpserver.RegisterSchedules(server, schedule.New(schedules)) 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.WrapConsoleAuth(server, consoleAuth) httpserver.WrapMetrics(server, metrics) workflows, err := mysql.NewWorkflowRuntime(db, node) diff --git a/deploy/docker-compose.ha.yml b/deploy/docker-compose.ha.yml index ff04144..169203e 100644 --- a/deploy/docker-compose.ha.yml +++ b/deploy/docker-compose.ha.yml @@ -16,6 +16,13 @@ x-orbit-environment: &orbit-environment ORBIT_EVENT_CONNECTION_LIMIT: ${ORBIT_EVENT_CONNECTION_LIMIT:-128} ORBIT_DISPATCH_WORKERS: ${ORBIT_DISPATCH_WORKERS:-4} ORBIT_DISPATCH_CAPACITY: ${ORBIT_DISPATCH_CAPACITY:-128} + ORBIT_RETENTION_IDEMPOTENCY_DAYS: ${ORBIT_RETENTION_IDEMPOTENCY_DAYS:-90} + ORBIT_RETENTION_TASK_DAYS: ${ORBIT_RETENTION_TASK_DAYS:-365} + ORBIT_RETENTION_WORKFLOW_DAYS: ${ORBIT_RETENTION_WORKFLOW_DAYS:-365} + ORBIT_RETENTION_EVENT_DAYS: ${ORBIT_RETENTION_EVENT_DAYS:-365} + ORBIT_RETENTION_DELIVERY_DAYS: ${ORBIT_RETENTION_DELIVERY_DAYS:-90} + ORBIT_RETENTION_INBOX_DAYS: ${ORBIT_RETENTION_INBOX_DAYS:-90} + ORBIT_RETENTION_AUDIT_DAYS: ${ORBIT_RETENTION_AUDIT_DAYS:-2555} ORBIT_CONSOLE_QUEUE_ID: ${ORBIT_CONSOLE_QUEUE_ID:-} ORBIT_METRICS_ADDR: ${ORBIT_METRICS_ADDR:-} diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 1af82a9..c8a91c0 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -85,6 +85,13 @@ services: ORBIT_EVENT_CONNECTION_LIMIT: ${ORBIT_EVENT_CONNECTION_LIMIT:-128} ORBIT_DISPATCH_WORKERS: ${ORBIT_DISPATCH_WORKERS:-4} ORBIT_DISPATCH_CAPACITY: ${ORBIT_DISPATCH_CAPACITY:-128} + ORBIT_RETENTION_IDEMPOTENCY_DAYS: ${ORBIT_RETENTION_IDEMPOTENCY_DAYS:-90} + ORBIT_RETENTION_TASK_DAYS: ${ORBIT_RETENTION_TASK_DAYS:-365} + ORBIT_RETENTION_WORKFLOW_DAYS: ${ORBIT_RETENTION_WORKFLOW_DAYS:-365} + ORBIT_RETENTION_EVENT_DAYS: ${ORBIT_RETENTION_EVENT_DAYS:-365} + ORBIT_RETENTION_DELIVERY_DAYS: ${ORBIT_RETENTION_DELIVERY_DAYS:-90} + ORBIT_RETENTION_INBOX_DAYS: ${ORBIT_RETENTION_INBOX_DAYS:-90} + ORBIT_RETENTION_AUDIT_DAYS: ${ORBIT_RETENTION_AUDIT_DAYS:-2555} ORBIT_CONSOLE_QUEUE_ID: ${ORBIT_CONSOLE_QUEUE_ID:-} ORBIT_METRICS_ADDR: ${ORBIT_METRICS_ADDR:-} ports: diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 4659827..c68de87 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -1,6 +1,6 @@ # 数据模型与实现基线 -更新：2026-09-18。数据库统一禁止外键。当前迁移版本为28；000026–000028只增加固定业务观测读取所需索引，不增加实体或字段。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现，独立 Inbox 场景已实现，流程编排继续开发。 +更新：2026-09-22。数据库统一禁止外键。当前迁移版本为30；000026–000028只增加固定业务观测读取所需索引，000029增加控制台账号/会话，000030增加命名空间级保留策略。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现，独立 Inbox 场景已实现，流程编排继续开发。 ## 1. 模块与依赖 @@ -35,7 +35,7 @@ ## 3. 实体清单 -当前共 36 个实体：初始迁移包含 30 个，000003 增加进程会话头和迟到证据，000010 增加应用共享额度，000013 增加结果流与不可变结果快照。000024 增加固定流程版本的查询范围。迁移自身另有 `orbit_schema_migrations` 记录。 +当前共 39 个业务实体：原运行模型为36个，000029增加控制台账号与会话，000030增加保留策略。迁移自身另有 `orbit_schema_migrations` 记录；控制台和策略表同样遵循零外键、中文 COMMENT、应用层归属校验。 | 模块 | Go 模型 | 数据表 | | --- | --- | --- | @@ -50,6 +50,9 @@ | platform | `Subscription` | `orbit_subscriptions` | | platform | `Delivery` | `orbit_deliveries` | | platform | `AuditEvent` | `orbit_audit_events` | +| platform | `ConsoleUser` | `orbit_console_users` | +| platform | `ConsoleSession` | `orbit_console_sessions` | +| platform | `RetentionPolicy` | `orbit_retention_policies` | | pulse | `Queue` | `pulse_queues` | | pulse | `ExecutionGate` | `pulse_execution_gates` | | pulse | `WorkerSession` | `pulse_worker_sessions` | @@ -126,7 +129,7 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 000007_task_cancellation 允许尝试证据记录 CANCELLED，不新增实体；取消意图、结果和历史不确定效果的规则见 [取消契约](api/cancellation.md)。 -000008_no_foreign_keys 移除历史 70 个外键，保留索引、唯一约束、CHECK、触发器和数据。旧迁移已在回归库应用，按不可变校验和约定保留原文件；已有库继续按历史链前向升级。新装库由迁移器使用当前版本的无外键基线直接建表，再登记 1–28 的不可变校验和，因此初始化过程也不会先创建历史外键；基线版本变化时必须随新迁移重新生成并回归。后续迁移禁止再引入外键。升级旧库时先停用旧服务，应用完整迁移再启动新版；迁移需要 DDL 锁，应在目标部署环境演练维护窗口。 +000008_no_foreign_keys 移除历史 70 个外键，保留索引、唯一约束、CHECK、触发器和数据。旧迁移已在回归库应用，按不可变校验和约定保留原文件；已有库继续按历史链前向升级。新装库由迁移器使用当前版本的无外键基线直接建表，再登记 1–30 的不可变校验和，因此初始化过程也不会先创建历史外键；基线版本变化时必须随新迁移重新生成并回归。后续迁移禁止再引入外键。升级旧库时先停用旧服务，应用完整迁移再启动新版；迁移需要 DDL 锁，应在目标部署环境演练维护窗口。 000009_queue_resources 增加 Queue.MaxUnfinishedTasks、UnfinishedTasks、StartWindowUntil、StartsInWindow，Go 字段与 SQL COMMENT 均有中文语义，实体仍为 32。未终结任务按 WAITING/LEASED/RUNNING/NEEDS_REVIEW 回填；窗口初始为空、计数为零。运行计数与任务/开始授权同事务维护，不改变队列配置 revision，配置变更不能重置它们。降额允许已有使用量暂时超过新上限，不删除任务或撤销已有许可；详见 [队列契约](api/queues.md) 和 [资源回归](reports/phase-3-queue-resources.md)。 @@ -253,3 +256,9 @@ Event.WorkflowID为只读STORED生成列，从原不可变事件派生流程归 000027为`orbit_events(type,occurred_at,id)`增加固定事件窗口索引。采样器只读取代码列出的任务和流程事件类别、最近5分钟、每类最多10000条；未知事件类型不写入标签，事件指标只说明本节点有限趋势，不能代替审计查询、端到端阶段延迟或跨节点吞吐。索引不改变事件原文、不可变触发器、外键策略或事务边界。 000028为`pulse_tasks(state,completed_at,id)`和`weave_instances(state,completed_at,id)`增加终态耗时样本索引。采样器只读取最近5分钟内成功、失败或取消记录，最多10000条，计算创建到终态的平均/最大秒数并在截断时复用统一标记；不把该样本解释为端到端外部效果或全局SLO。 + +## 控制台身份与历史保留策略（000029–000030） + +000029增加 `orbit_console_users` 和 `orbit_console_sessions`，只保存控制台账号的 bcrypt 哈希、绑定凭据及 HttpOnly 会话摘要；不保存密码或令牌原文。控制台继续单管理员模型，服务层每次修改前核对会话、命名空间和绑定凭据权限。 + +000030增加 `orbit_retention_policies`，按命名空间保存归档开关、幂等/任务/流程/事件/交付/Inbox/审计窗口、修订号及时间。所有字段都有中文 COMMENT，策略不含外键或级联。数据库策略优先于部署默认值；默认关闭归档，配置变化不触发删除。服务层用同一事务完成版本核对、幂等回执和审计快照，并在归档预览/导出时保护未确认交付、未知效果、未结束流程和有效幂等身份。具体窗口、候选目录和控制台接口见[数据保留与归档策略](api/retention.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 5731a94..17a84b1 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -42,9 +42,9 @@ 定时计划详情已接入当前版本、规则、下一次触发和 occurrence 历史；暂停/恢复沿用 `schedules:manage`、`expected_revision`、原因和幂等键。计划发布、跨命名空间管理身份和更高风险的交付控制仍按独立权限模型推进。 -控制台五个核心模块已拆为独立 hash 视图，导航状态和页面标题随视图同步，刷新或直接打开对应 hash 后仍保持正确模块。数据请求继续由统一 Bearer Token 和当前队列范围驱动。 +控制台六个核心模块已拆为独立 hash 视图，导航状态和页面标题随视图同步，刷新或直接打开对应 hash 后仍保持正确模块。数据请求继续由统一会话和当前命名空间范围驱动。 -已完成一次真实链路回归：标准 `scripts/test-mysql.sh` 在一次性 MySQL 和真实 Orbit 进程中跑通任务、Worker、结果订阅、Weave 流程、进程恢复及独立业务示例；控制台五个入口、健康检查和详情所依赖的读取 API 已用真实结果核对，证据见 [Console 真实链路回归](reports/phase-7-console-e2e.md)。这关闭了本地闭环验证项，生产拓扑和发布门槛仍保持未验收。 +已完成一次真实链路回归：标准 `scripts/test-mysql.sh` 在一次性 MySQL 和真实 Orbit 进程中跑通任务、Worker、结果订阅、Weave 流程、进程恢复及独立业务示例；控制台六个入口、健康检查和详情所依赖的读取 API 已用真实结果核对，保留策略页面另有真实 MySQL 策略回归，证据见 [Console 真实链路回归](reports/phase-7-console-e2e.md)。这关闭了本地闭环验证项，生产拓扑和发布门槛仍保持未验收。 新增 `make console-e2e` 作为首次部署闭环门禁：在一次性 MySQL 和 Docker Orbit 中从空库迁移、初始化控制台管理员，完成登录、受保护页面、WS Worker 成功执行、明确失败重试、定时任务、控制台任务读取和登出后的 401 校验，证据见[首次部署与控制台任务闭环](reports/phase-8-console-e2e.md)。该门禁证明功能链路可重复，不替代目标生产拓扑、容量和 RPO/RTO 验收。 @@ -107,9 +107,9 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( ### P6 首批实施顺序 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领取协程和有界内存排队均支持有界环境配置，并在启动日志和固定指标中记录。仍需端到端外部效果延迟、全局持久吞吐、Inbox从事件产生到推进完成的精确阶段延迟、全局发送队列、多节点聚合、锁等待/死锁告警阈值和生产采集/通知验证；SDK受控重连风暴和真实慢消费者背压已补[边界回归](reports/phase-6-reconnect-storm.md)与[慢消费者报告](reports/phase-6-slow-consumer.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. 已补[`orbit-admin retention inventory`](api/retention.md)只读盘点和[阶段报告](reports/phase-6-retention-inventory.md)，能区分过期幂等、确定终态、未知效果、未终结流程、已确认/未确认交付、已消费/未消费Inbox、事件和审计规模；它不执行归档或删除。已补[容量模型](19-capacity-model.md)，把企业数转换为任务、流程、事件、连接、写放大和保留期输入；仍需评审各类保留周期、归档格式、受保护记录核对，再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证，不缩减为本机多进程检查。 +4. 已补[数据库保留策略、预览和只读归档导出](api/retention.md)以及[`orbit-admin retention inventory`](api/retention.md)只读盘点。策略按命名空间存入 MySQL，控制台单管理员可修改，版本、原因和前后快照进入审计；默认关闭归档，源数据不会因配置变化自动删除。固定候选目录保护有效幂等身份、未确认交付、未知效果、未终结流程和未处理 Inbox；导出包带冻结策略版本和 SHA-256。仍需在目标环境评审最终归档介质、备份恢复、删除执行器和保留周期，再执行兼容升级和容量负载。生产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 资源、连接池等待、业务快照和事务重试计数采样；下一步补长期存储增长、跨节点重连与目标数据库拓扑容量。 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 验收。 @@ -235,6 +235,7 @@ A04派发已提交而消息尚未到达Worker的精确进程故障已通过[专 - [P2 SDK 协议阶段记录](reports/phase-2-sdk-protocol.md) - [P2 自动 Worker 运行器记录](reports/phase-2-worker-runtime.md) - [P2 执行历史与大载荷记录](reports/phase-2-execution-history.md) +- [P1 稳定性与归档定向回归](reports/phase-6-stability-p1.md) ### 状态事件时间线子项 diff --git a/docs/api/retention.md b/docs/api/retention.md index e572730..21f05d1 100644 --- a/docs/api/retention.md +++ b/docs/api/retention.md @@ -1,4 +1,38 @@ -# 保留清单（只读） +# 数据保留与归档策略 + +Orbit 将保留策略作为命名空间级数据库配置保存。控制台保持单管理员模型，登录后打开“保留策略”即可查看和修改窗口；每次修改都要求上一版本号、幂等键和变更原因，前后值写入 `orbit_audit_events`。多个 Orbit 节点读取同一个 MySQL 版本，因此不会因为节点环境变量不同而产生不同清理边界。 + +首次安装和数据库策略尚未保存时使用部署默认值：幂等记录 90 天、任务历史 365 天、流程历史 365 天、事件日志 365 天、已确认交付 90 天、已处理 Inbox 90 天、管理审计 2555 天。默认 `enabled=false`，保存策略也不会立即删除数据。 + +环境变量只负责首次展示/未保存策略的默认值： + +```text +ORBIT_RETENTION_IDEMPOTENCY_DAYS=90 +ORBIT_RETENTION_TASK_DAYS=365 +ORBIT_RETENTION_WORKFLOW_DAYS=365 +ORBIT_RETENTION_EVENT_DAYS=365 +ORBIT_RETENTION_DELIVERY_DAYS=90 +ORBIT_RETENTION_INBOX_DAYS=90 +ORBIT_RETENTION_AUDIT_DAYS=2555 +``` + +窗口范围为 1–3650 天，幂等记录不得短于 90 天；所有节点应使用同一组默认值。数据库中已有策略后，环境变量不覆盖它。管理端的“启用归档”只是允许归档执行器读取固定版本，不能绕过版本核对。 + +归档分两步：先用“预览”或 `GET /console/retention/preview` 查看各类候选，再按固定类别调用 `POST /console/retention/export` 分批生成只读 JSON 快照。快照包含冻结的策略版本、数据库时间、截止时间、根记录及关联历史，并返回 SHA-256；源表仍然保留，导出成功只记录摘要和数量审计，不代表可以删除。后续删除执行器必须在独立发布和备份恢复验收后接入，不能把导出接口当成删除接口。 + +候选筛选由服务端固定目录实现，浏览器不能提交表名或 SQL。终态任务必须没有未知效果、租约和流程所有权；流程必须终态且没有未结任务或未处理 Inbox；事件必须没有未确认交付或未处理 Inbox；交付只允许已确认记录；Inbox 只允许已处理且流程已终态记录；幂等记录需同时超过过期时间和窗口。审计单独使用审计窗口，普通清理不处理审计记录。未确认交付、未处理 Inbox、未知效果、进行中的任务/流程和有效幂等身份始终受保护。 + +控制台接口示例： + +```sh +# 已登录控制台会话；浏览器通常直接使用页面，不需要手工传令牌 +curl -b orbit-console.cookie http://127.0.0.1:8080/console/retention +curl -b orbit-console.cookie -X PUT http://127.0.0.1:8080/console/retention \ + -H 'Content-Type: application/json' -H 'Idempotency-Key: retention-2026-q3' \ + --data '{"expected_revision":"0","enabled":false,"idempotency_days":90,"task_history_days":365,"workflow_history_days":365,"event_log_days":365,"delivery_history_days":90,"inbox_history_days":90,"audit_log_days":2555,"reason":"初始生产保留评审"}' +``` + +## 只读盘点 `orbit-admin retention inventory` 用于在制定保留窗口、归档和备份恢复方案前盘点历史数据。它只执行带截止时间的统计查询，不删除、搬迁、改写数据，也不会自动判断某条记录可以安全归档。 diff --git a/docs/console.md b/docs/console.md index f2d3d8c..40b5eb2 100644 --- a/docs/console.md +++ b/docs/console.md @@ -20,7 +20,8 @@ http://127.0.0.1:8080/console/ - 在具备 `workflows:control` 权限时，流程详情可以暂停、恢复或取消；操作原因、`control_revision` 和幂等键交给服务端做最终校验。 - `NEEDS_REVIEW` 流程会显示按步骤类型划分的人工恢复入口：任务核查、步骤结论、继续、步骤重试、PENDING 接续和等待重建；每次提交都要求当前修订、证据引用和显式确认，服务端恢复 API 负责最终门控。 - 在具备 `events:manage` 权限时，订阅详情可以启停订阅，并对 `EXPIRED` 交付追加次数预算后重试；原交付身份、已用次数和事件不会被页面重置。 -- 左侧五个模块是独立的控制台视图，使用 `/console/#/overview`、`#/tasks`、`#/schedules`、`#/workflows` 和 `#/subscriptions` 定位；切换视图不会把所有模块堆在一张长页面上。 +- 左侧六个模块是独立的控制台视图，使用 `/console/#/overview`、`#/tasks`、`#/schedules`、`#/workflows`、`#/subscriptions` 和 `#/retention` 定位；切换视图不会把所有模块堆在一张长页面上。 +- “保留策略”只允许单管理员查看和修改命名空间级窗口；修改需要版本号、幂等键和原因并留下审计。预览和只读导出不会删除源数据，未确认交付、未知效果和未结束流程不会成为候选。 - 任务状态按 10 秒自动刷新；页面只调用服务端授权接口，不把账号密码或业务凭据上传到独立服务，也不读取数据库。 - 页面使用当前请求方的业务权限；没有权限的模块显示为空或等待权限，不绕过服务端授权。 diff --git a/docs/release-checklist.md b/docs/release-checklist.md index 3818e1a..5152f34 100644 --- a/docs/release-checklist.md +++ b/docs/release-checklist.md @@ -45,7 +45,7 @@ | 受理、派发、流程推进和恢复吞吐 | 固定硬件、数据量、任务/流程混合模型，p50/p95/p99和资源曲线 | 已有受理样本、[真实Worker短/混合/中长执行样本](reports/phase-6-execution-load.md)、[Weave双分支流程样本](reports/phase-6-workflow-load.md)、[定时集中触发/积压曲线样本](reports/phase-6-backlog.md)及本轮[四项闭环验收](reports/phase-7-acceptance-20260921.md)；目标硬件资源曲线、生产容量和HA仍未验收 | | 100万+企业规模推导 | [容量模型](19-capacity-model.md)：租户、队列、连接、事件、索引和保存周期的参数化推导 | 已建立计算方法和演算样例；目标业务数据、硬件、拓扑和生产容量仍未验收，不能由企业数直接承诺 | | MySQL 复制、切换、RPO/RTO | [隔离 GTID 复制与手工提升演练](reports/phase-6-mysql-replication.md)；目标拓扑上的实际切换、复制确认、恢复点和恢复时间 | 隔离源库/副本复制、手工提升和原幂等键重放已通过；自动切换、目标拓扑、RPO/RTO仍未验收 | -| 慢消费者、重连风暴、磁盘和锁等待 | 受控故障负载及积压清空证据 | SDK 受控连续断线和真实 HTTP/WebSocket 容量1慢消费者已回归（见[重连报告](reports/phase-6-reconnect-storm.md)、[慢消费者报告](reports/phase-6-slow-consumer.md)）；跨节点风暴、磁盘和锁等待未验收 | +| 慢消费者、重连风暴、磁盘和锁等待 | `make stability-check`，受控故障负载及积压清空证据 | SDK 受控连续断线、跨节点重连风暴、真实 HTTP/WebSocket 容量1慢消费者、真实 MySQL 锁等待/死锁和保留期磁盘增长已加入定向回归；目标生产拓扑的规模和告警阈值仍待验收 | | 跨版本 SDK/服务端 | 至少一个正式旧标签与候选版本的双向矩阵 | 历史提交 `a50d5fa` 的稳定任务 API 双向 smoke 已通过（见[报告](reports/phase-7-historical-compatibility.md)）；正式标签和全协议矩阵仍未验收 | 发布候选只有在所有“待目标环境/未验收”项得到对应环境证据后，才能进入生产审批。当前代码和阶段回归不等同于生产高可用、容量、RPO/RTO或百万企业承诺。 diff --git a/docs/reports/phase-6-stability-p1.md b/docs/reports/phase-6-stability-p1.md new file mode 100644 index 0000000..199cbbb --- /dev/null +++ b/docs/reports/phase-6-stability-p1.md @@ -0,0 +1,21 @@ +# P1 稳定性与历史归档定向回归 + +这份报告定义当前代码的可复现入口，命令成功才算目标环境完成验证。它不把本机测试结果当作生产容量承诺。 + +```sh +ORBIT_TEST_MYSQL_DSN='root:...@tcp(127.0.0.1:3306)/orbit_test' \ +ORBIT_TEST_SERVER_BINARY="$PWD/bin/orbit" \ +ORBIT_TEST_WORKER_URL='http://127.0.0.1:8080' \ +ORBIT_TEST_WORKER_CREDENTIALS_FILE="$PWD/.local/worker.credentials.json" \ +make stability-check +``` + +入口覆盖： + +- `TestIntegrationEventProcessCrossNodeReconnectStorm`：在两个独立 Orbit 进程之间反复切换 WebSocket 上游，验证重连风暴期间一次处理、一次确认和一次持久交付尝试。 +- `TestIntegrationEventProcessSlowConsumerPeerProgress`：节点 A 的容量 1 处理器阻塞时，节点 B 仍可推进自己的交付，确认慢消费者隔离。 +- `TestIntegrationWithTxRetriesPreCommitDeadlock`：两个真实 MySQL 事务反向加锁，使用数据库锁竞争和事务重试指标确认提交前死锁被有界恢复。 +- `TestIntegrationFairDispatchApplicationIsolation` 与 `TestIntegrationWorkflowVersionAndRaces`：真实锁等待和多节点竞争路径，不使用固定 sleep 推断锁结果。 +- `TestIntegrationRetentionDiskGrowthAndPreview`：写入带终态时间的有界任务样本，读取 `information_schema.TABLES` 的数据/索引字节增长，并核对保留策略预览只返回安全候选。 + +保留策略由数据库按命名空间保存，默认关闭归档。控制台单管理员通过“保留策略”页面修改窗口，要求版本、幂等键和原因；审计记录保留前后快照。导出是固定表目录的只读 JSON 快照，带 SHA-256 和策略版本，源数据仍保留。未确认交付、未知效果、未结束任务/流程、未处理 Inbox 和有效幂等记录不会成为普通归档候选。最终删除执行器、归档介质和目标拓扑 RPO/RTO 需要单独验收。 diff --git a/internal/config/config.go b/internal/config/config.go index 3ec0c62..f677a7d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -7,9 +7,12 @@ import ( "os" "strconv" "time" + + "git.yuntuo.hk/k2safe/orbit/internal/retention" ) type Config struct { + RetentionDefaults retention.Policy // RetentionDefaults 未保存数据库策略时使用的部署默认值。 HTTPAddr string MySQLDSN string MaxOpen int @@ -120,5 +123,10 @@ func FromEnv(get func(string) string) (Config, error) { return Config{}, errors.New("invalid ORBIT_METRICS_ADDR") } } + var err error + c.RetentionDefaults, err = retention.FromEnv(get) + if err != nil { + return Config{}, err + } return c, nil } diff --git a/internal/retention/archive.go b/internal/retention/archive.go new file mode 100644 index 0000000..2ab000c --- /dev/null +++ b/internal/retention/archive.go @@ -0,0 +1,43 @@ +package retention + +import ( + "context" + "time" + + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +// ArchiveInput 选择一类历史快照；每批最多100条根记录，游标绑定类别和策略版本。 +type ArchiveInput struct { + Kind string `json:"kind" required:"true"` // Kind tasks、workflows、events、audit、deliveries、inbox 或 idempotency。 + Limit int `json:"limit"` // Limit 根记录数量，默认20，最大100。 + Cursor string `json:"cursor"` // Cursor 上一批返回的继续位置，空表示从最早记录开始。 + ExpectedRevision uint64 `json:"expected_revision,string" required:"true"` // ExpectedRevision 防止使用页面未确认的保留窗口导出。 +} + +// ArchiveSnapshot 是一个只读归档批次；SourceRetained始终为true，禁止把导出当作数据库清理。 +type ArchiveSnapshot struct { + FormatVersion int `json:"format_version"` // FormatVersion 归档信封格式版本。 + SHA256 string `json:"sha256"` // SHA256 对Content原始UTF-8字节的摘要，校验传输和文件完整性。 + Content string `json:"content"` // Content 是原始 JSON 文本；下载时按 UTF-8 原样保存再核验 SHA256，不重新序列化。 +} + +// CandidatePage 描述按策略筛选的一批根记录，不宣称它们可以删除。 +type CandidatePage struct { + Kind string `json:"kind"` // Kind 历史类型。 + Cutoff time.Time `json:"cutoff"` // Cutoff 当前策略的UTC排除边界。 + Count int `json:"count"` // Count 本批候选数量，不是全库数量。 + More bool `json:"more"` // More 后面还有候选根记录。 +} + +// Preview 返回按各类保留窗口读取的有限候选数量，不加载业务正文。 +type Preview struct { + Policy Policy `json:"policy"` // Policy 实际读取的版本和窗口。 + SampleLimit int `json:"sample_limit"` // SampleLimit 每类候选的取样上限。 + Items []CandidatePage `json:"items"` // Items 固定类别的候选样本。 +} + +type ArchiveStore interface { + PreviewRetention(context.Context, auth.Token) (Preview, error) + ExportRetention(context.Context, auth.Token, ArchiveInput) (ArchiveSnapshot, error) +} diff --git a/internal/retention/policy.go b/internal/retention/policy.go new file mode 100644 index 0000000..349f786 --- /dev/null +++ b/internal/retention/policy.go @@ -0,0 +1,192 @@ +// Package retention 定义 Orbit 历史数据保留策略的通用模型。 +// +// 策略只描述候选数据的最短保留窗口，不直接删除任务、事件或审计记录。 +// 归档执行器必须在独立流程中先完成核查、备份和归档证明，再按策略处理。 +package retention + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "time" + "unicode/utf8" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +const ( + // DefaultIdempotencyDays 是请求幂等记录的默认保留天数。 + DefaultIdempotencyDays uint32 = 90 + // DefaultTaskHistoryDays 是终态任务历史的默认保留天数。 + DefaultTaskHistoryDays uint32 = 365 + // DefaultWorkflowHistoryDays 是终态流程历史的默认保留天数。 + DefaultWorkflowHistoryDays uint32 = 365 + // DefaultEventLogDays 是状态事件和结果事件的默认保留天数。 + DefaultEventLogDays uint32 = 365 + // DefaultDeliveryHistoryDays 是已确认交付历史的默认保留天数。 + DefaultDeliveryHistoryDays uint32 = 90 + // DefaultInboxHistoryDays 是已处理流程 Inbox 历史的默认保留天数。 + DefaultInboxHistoryDays uint32 = 90 + // DefaultAuditLogDays 是管理审计的默认保留天数，约七年。 + DefaultAuditLogDays uint32 = 2555 + + minRetentionDays uint32 = 1 + maxRetentionDays uint32 = 3650 +) + +// Policy 是按命名空间保存的历史数据保留策略。 +// Enabled 默认关闭；即使打开，也允许只读归档导出使用满足窗口且已核验的数据。 +// 当前版本不会因为配置变更自动删除任何数据。 +type Policy struct { + // NamespaceID 是策略所属命名空间，不能由控制台请求替换。 + NamespaceID core.ID `json:"namespace_id"` + // Revision 是乐观并发版本；默认策略未落库时为0，首次保存后从1开始。 + Revision uint64 `json:"revision,string"` + // Enabled 表示是否允许归档执行器使用该策略；默认关闭以避免误清理。 + Enabled bool `json:"enabled"` + // IdempotencyDays 是请求幂等记录至少保留的天数。 + IdempotencyDays uint32 `json:"idempotency_days"` + // TaskHistoryDays 是终态任务及执行证据至少保留的天数。 + TaskHistoryDays uint32 `json:"task_history_days"` + // WorkflowHistoryDays 是终态流程、步骤、等待和流程事件至少保留的天数。 + WorkflowHistoryDays uint32 `json:"workflow_history_days"` + // EventLogDays 是状态事件和结果事件至少保留的天数。 + EventLogDays uint32 `json:"event_log_days"` + // DeliveryHistoryDays 是已确认结果交付至少保留的天数；未确认交付不得按窗口清理。 + DeliveryHistoryDays uint32 `json:"delivery_history_days"` + // InboxHistoryDays 是已处理流程 Inbox 事件至少保留的天数；未处理记录不得清理。 + InboxHistoryDays uint32 `json:"inbox_history_days"` + // AuditLogDays 是管理审计至少保留的天数；审计记录不参与普通清理。 + AuditLogDays uint32 `json:"audit_log_days"` + // UpdatedAt 是策略最后修改时间，统一使用 UTC。 + UpdatedAt time.Time `json:"updated_at"` + // Source 表示策略来自默认值还是数据库配置，便于控制台明确当前状态。 + Source string `json:"source"` +} + +// UpdateInput 是控制台保存策略所需的完整替换输入。 +// 所有窗口必须显式提交，避免遗漏字段被解释成缩短保留期。 +type UpdateInput struct { + // ExpectedRevision 是页面读取到的版本；首次保存必须为0。 + ExpectedRevision uint64 `json:"expected_revision,string" required:"true"` + // Enabled 是否允许归档执行器读取本策略。 + Enabled bool `json:"enabled" required:"true"` + // IdempotencyDays 请求幂等记录保留天数。 + IdempotencyDays uint32 `json:"idempotency_days" required:"true"` + // TaskHistoryDays 终态任务历史保留天数。 + TaskHistoryDays uint32 `json:"task_history_days" required:"true"` + // WorkflowHistoryDays 终态流程历史保留天数。 + WorkflowHistoryDays uint32 `json:"workflow_history_days" required:"true"` + // EventLogDays 状态和结果事件保留天数。 + EventLogDays uint32 `json:"event_log_days" required:"true"` + // DeliveryHistoryDays 已确认交付历史保留天数。 + DeliveryHistoryDays uint32 `json:"delivery_history_days" required:"true"` + // InboxHistoryDays 已处理 Inbox 历史保留天数。 + InboxHistoryDays uint32 `json:"inbox_history_days" required:"true"` + // AuditLogDays 管理审计保留天数。 + AuditLogDays uint32 `json:"audit_log_days" required:"true"` + // Reason 是本次策略变更原因，会进入管理审计。 + Reason string `json:"reason" required:"true"` +} + +// Store 是控制台和归档执行器共享的最小策略存储边界。 +// 实现必须保证 Update 的版本核对、幂等和审计与策略写入处于同一事务。 +type Store interface { + GetRetentionPolicy(context.Context, core.ID) (Policy, error) + UpdateRetentionPolicy(context.Context, auth.Token, string, UpdateInput) (Policy, error) +} + +// Defaults 返回不会触发任何归档执行的安全默认策略。 +func Defaults(namespace core.ID) Policy { + return Policy{ + NamespaceID: namespace, + Enabled: false, + IdempotencyDays: DefaultIdempotencyDays, + TaskHistoryDays: DefaultTaskHistoryDays, + WorkflowHistoryDays: DefaultWorkflowHistoryDays, + EventLogDays: DefaultEventLogDays, + DeliveryHistoryDays: DefaultDeliveryHistoryDays, + InboxHistoryDays: DefaultInboxHistoryDays, + AuditLogDays: DefaultAuditLogDays, + Source: "default", + } +} + +// Validate 检查策略身份、窗口和审计原因。 +func (p Policy) Validate() error { + if p.NamespaceID.Validate() != nil { + return errors.New("retention namespace is invalid") + } + for name, days := range map[string]uint32{ + "idempotency_days": p.IdempotencyDays, + "task_history_days": p.TaskHistoryDays, + "workflow_history_days": p.WorkflowHistoryDays, + "event_log_days": p.EventLogDays, + "delivery_history_days": p.DeliveryHistoryDays, + "inbox_history_days": p.InboxHistoryDays, + "audit_log_days": p.AuditLogDays, + } { + if days < minRetentionDays || days > maxRetentionDays { + return fmt.Errorf("%s must be between %d and %d days", name, minRetentionDays, maxRetentionDays) + } + } + if p.IdempotencyDays < DefaultIdempotencyDays { + return errors.New("idempotency_days cannot be less than 90 days") + } + return nil +} + +// ToPolicy 将完整输入转为待保存策略，版本由事务内分配。 +func (in UpdateInput) ToPolicy(namespace core.ID) (Policy, error) { + if strings.TrimSpace(in.Reason) == "" || !utf8.ValidString(in.Reason) || len(in.Reason) > 1024 || strings.ContainsAny(in.Reason, "\r\n") { + return Policy{}, errors.New("retention reason must be 1-1024 UTF-8 bytes without newlines") + } + p := Policy{ + NamespaceID: namespace, + Enabled: in.Enabled, + IdempotencyDays: in.IdempotencyDays, + TaskHistoryDays: in.TaskHistoryDays, + WorkflowHistoryDays: in.WorkflowHistoryDays, + EventLogDays: in.EventLogDays, + DeliveryHistoryDays: in.DeliveryHistoryDays, + InboxHistoryDays: in.InboxHistoryDays, + AuditLogDays: in.AuditLogDays, + } + if err := p.Validate(); err != nil { + return Policy{}, err + } + return p, nil +} + +// FromEnv 读取部署默认值；数据库策略保存后优先于部署默认值。 +// 所有节点应注入同一组默认值，保证尚未保存策略的展示一致。 +func FromEnv(get func(string) string) (Policy, error) { + p := Defaults("") + for _, f := range []struct { + name string + dst *uint32 + }{ + {"ORBIT_RETENTION_IDEMPOTENCY_DAYS", &p.IdempotencyDays}, + {"ORBIT_RETENTION_TASK_DAYS", &p.TaskHistoryDays}, + {"ORBIT_RETENTION_WORKFLOW_DAYS", &p.WorkflowHistoryDays}, + {"ORBIT_RETENTION_EVENT_DAYS", &p.EventLogDays}, + {"ORBIT_RETENTION_DELIVERY_DAYS", &p.DeliveryHistoryDays}, + {"ORBIT_RETENTION_INBOX_DAYS", &p.InboxHistoryDays}, + {"ORBIT_RETENTION_AUDIT_DAYS", &p.AuditLogDays}, + } { + if raw := get(f.name); raw != "" { + n, err := strconv.ParseUint(raw, 10, 32) + if err != nil || n < 1 || n > 3650 { + return Policy{}, fmt.Errorf("%s must be 1-3650 days", f.name) + } + *f.dst = uint32(n) + } + } + if p.IdempotencyDays < 90 { + return Policy{}, errors.New("ORBIT_RETENTION_IDEMPOTENCY_DAYS cannot be less than 90") + } + return p, nil +} diff --git a/internal/retention/policy_test.go b/internal/retention/policy_test.go new file mode 100644 index 0000000..7485e82 --- /dev/null +++ b/internal/retention/policy_test.go @@ -0,0 +1,65 @@ +package retention + +import ( + "testing" + + core "git.yuntuo.hk/k2safe/orbit/model" +) + +func TestDefaultsAndValidation(t *testing.T) { + ns, err := core.NewID() + if err != nil { + t.Fatal(err) + } + p := Defaults(ns) + if p.Enabled || p.Source != "default" || p.IdempotencyDays != 90 || p.TaskHistoryDays != 365 || p.AuditLogDays != 2555 { + t.Fatalf("unexpected defaults: %+v", p) + } + if err := p.Validate(); err != nil { + t.Fatal(err) + } + p.IdempotencyDays = 89 + if err := p.Validate(); err == nil { + t.Fatal("accepted an idempotency window shorter than the safety minimum") + } +} + +func TestUpdateInputRequiresReasonAndKeepsSafetyMinimum(t *testing.T) { + ns, _ := core.NewID() + in := UpdateInput{ + IdempotencyDays: 90, TaskHistoryDays: 30, WorkflowHistoryDays: 30, + EventLogDays: 30, DeliveryHistoryDays: 30, InboxHistoryDays: 30, + AuditLogDays: 365, Reason: "quarterly retention review", + } + if _, err := in.ToPolicy(ns); err != nil { + t.Fatal(err) + } + in.Reason = "" + if _, err := in.ToPolicy(ns); err == nil { + t.Fatal("accepted an empty audit reason") + } + in.Reason = "shorten" + in.IdempotencyDays = 89 + if _, err := in.ToPolicy(ns); err == nil { + t.Fatal("accepted an unsafe idempotency window") + } +} + +func TestFromEnvOverridesOnlyDeploymentDefaults(t *testing.T) { + values := map[string]string{ + "ORBIT_RETENTION_IDEMPOTENCY_DAYS": "120", + "ORBIT_RETENTION_TASK_DAYS": "180", + "ORBIT_RETENTION_AUDIT_DAYS": "730", + } + p, err := FromEnv(func(k string) string { return values[k] }) + if err != nil { + t.Fatal(err) + } + if p.IdempotencyDays != 120 || p.TaskHistoryDays != 180 || p.AuditLogDays != 730 || p.EventLogDays != DefaultEventLogDays { + t.Fatalf("environment defaults not applied: %+v", p) + } + values["ORBIT_RETENTION_IDEMPOTENCY_DAYS"] = "89" + if _, err := FromEnv(func(k string) string { return values[k] }); err == nil { + t.Fatal("accepted unsafe environment idempotency window") + } +} diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index c83a622..f2e3b69 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -92,6 +92,9 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_concurrency_key, DROP COLUMN concurrency_key, DROP COLUMN concurrency_key_digest`) f.exec(`ALTER TABLE pulse_attempts DROP COLUMN concurrency_token`) 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(`DELETE FROM orbit_schema_migrations WHERE version>=10`) if err := Migrate(ctx, db); err != nil { t.Fatal(err) diff --git a/internal/storage/mysql/baseline.sql b/internal/storage/mysql/baseline.sql index 893dbc2..ce9a49d 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 28. +-- Orbit fresh-install baseline, generated from migration version 30. -- This file intentionally contains no database-level relation constraints. CREATE TABLE `orbit_application_resources` ( `application_id` binary(16) NOT NULL COMMENT '所属应用标识及主键，跨命名空间共享额度', @@ -852,3 +852,29 @@ CREATE TABLE orbit_console_sessions ( KEY ix_console_session_expiry (expires_at, revoked_at), CONSTRAINT ck_console_sessions_0 CHECK (expires_at > created_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='控制台 HttpOnly 登录会话，只保存令牌摘要'; + +-- 命名空间级历史保留策略；只保存窗口和开关，不执行自动删除。 +-- 归档执行器必须在独立事务中核对状态、备份和归档证明后再使用这些窗口。 +CREATE TABLE orbit_retention_policies ( + namespace_id BINARY(16) NOT NULL COMMENT '策略所属命名空间，服务端核对登录会话后使用', + revision BIGINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '策略乐观并发版本，从1开始；首次默认策略未落库时版本为0', + enabled BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否允许归档执行器使用策略；默认关闭，不触发删除', + idempotency_days INT UNSIGNED NOT NULL DEFAULT 90 COMMENT '请求幂等记录至少保留天数，范围90至3650', + task_history_days INT UNSIGNED NOT NULL DEFAULT 365 COMMENT '终态任务和执行证据至少保留天数，范围1至3650', + workflow_history_days INT UNSIGNED NOT NULL DEFAULT 365 COMMENT '终态流程、步骤、等待和流程事件至少保留天数，范围1至3650', + event_log_days INT UNSIGNED NOT NULL DEFAULT 365 COMMENT '状态事件和结果事件至少保留天数，范围1至3650', + delivery_history_days INT UNSIGNED NOT NULL DEFAULT 90 COMMENT '已确认结果交付至少保留天数，未确认交付不得按窗口清理', + inbox_history_days INT UNSIGNED NOT NULL DEFAULT 90 COMMENT '已处理流程Inbox事件至少保留天数，未处理记录不得清理', + audit_log_days INT UNSIGNED NOT NULL DEFAULT 2555 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), + CONSTRAINT ck_retention_policy_revision CHECK (revision > 0), + CONSTRAINT ck_retention_policy_idempotency CHECK (idempotency_days BETWEEN 90 AND 3650), + CONSTRAINT ck_retention_policy_task CHECK (task_history_days BETWEEN 1 AND 3650), + CONSTRAINT ck_retention_policy_workflow CHECK (workflow_history_days BETWEEN 1 AND 3650), + CONSTRAINT ck_retention_policy_event CHECK (event_log_days BETWEEN 1 AND 3650), + CONSTRAINT ck_retention_policy_delivery CHECK (delivery_history_days BETWEEN 1 AND 3650), + 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='命名空间历史数据保留策略，只配置窗口和开关，不自动删除'; diff --git a/internal/storage/mysql/migrate.go b/internal/storage/mysql/migrate.go index d815674..8a20a54 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 = 29 +const freshInstallBaselineVersion uint64 = 30 type Migration struct { Version uint64 diff --git a/internal/storage/mysql/migrations/000030_retention_policies.up.sql b/internal/storage/mysql/migrations/000030_retention_policies.up.sql new file mode 100644 index 0000000..9395c77 --- /dev/null +++ b/internal/storage/mysql/migrations/000030_retention_policies.up.sql @@ -0,0 +1,25 @@ +-- 命名空间级历史保留策略；只保存窗口和开关，不执行自动删除。 +-- 归档执行器必须在独立事务中核对状态、备份和归档证明后再使用这些窗口。 +CREATE TABLE orbit_retention_policies ( + namespace_id BINARY(16) NOT NULL COMMENT '策略所属命名空间，服务端核对登录会话后使用', + revision BIGINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '策略乐观并发版本，从1开始；首次默认策略未落库时版本为0', + enabled BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否允许归档执行器使用策略；默认关闭，不触发删除', + idempotency_days INT UNSIGNED NOT NULL DEFAULT 90 COMMENT '请求幂等记录至少保留天数，范围90至3650', + task_history_days INT UNSIGNED NOT NULL DEFAULT 365 COMMENT '终态任务和执行证据至少保留天数，范围1至3650', + workflow_history_days INT UNSIGNED NOT NULL DEFAULT 365 COMMENT '终态流程、步骤、等待和流程事件至少保留天数，范围1至3650', + event_log_days INT UNSIGNED NOT NULL DEFAULT 365 COMMENT '状态事件和结果事件至少保留天数，范围1至3650', + delivery_history_days INT UNSIGNED NOT NULL DEFAULT 90 COMMENT '已确认结果交付至少保留天数，未确认交付不得按窗口清理', + inbox_history_days INT UNSIGNED NOT NULL DEFAULT 90 COMMENT '已处理流程Inbox事件至少保留天数，未处理记录不得清理', + audit_log_days INT UNSIGNED NOT NULL DEFAULT 2555 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), + CONSTRAINT ck_retention_policy_revision CHECK (revision > 0), + CONSTRAINT ck_retention_policy_idempotency CHECK (idempotency_days BETWEEN 90 AND 3650), + CONSTRAINT ck_retention_policy_task CHECK (task_history_days BETWEEN 1 AND 3650), + CONSTRAINT ck_retention_policy_workflow CHECK (workflow_history_days BETWEEN 1 AND 3650), + CONSTRAINT ck_retention_policy_event CHECK (event_log_days BETWEEN 1 AND 3650), + CONSTRAINT ck_retention_policy_delivery CHECK (delivery_history_days BETWEEN 1 AND 3650), + 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='命名空间历史数据保留策略，只配置窗口和开关，不自动删除'; diff --git a/internal/storage/mysql/no_foreign_keys_integration_test.go b/internal/storage/mysql/no_foreign_keys_integration_test.go index 31523ab..1c6fe35 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','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','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/storage/mysql/retention_archive.go b/internal/storage/mysql/retention_archive.go new file mode 100644 index 0000000..e54cdd6 --- /dev/null +++ b/internal/storage/mysql/retention_archive.go @@ -0,0 +1,377 @@ +package mysql + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" + "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 archiveMaxBytes = 8 << 20 +const archiveMaxRows = 10000 + +// 查询定义只来自固定目录；浏览器不能提交表名、列名或SQL条件。 +type archiveSource struct { + kind, table, clock, predicate string + days uint32 +} + +func archiveSources(p retention.Policy) []archiveSource { + return []archiveSource{ + {"tasks", "pulse_tasks", "completed_at", `state IN ('SUCCEEDED','FAILED','CANCELLED') AND effect_uncertain=FALSE AND lease_until IS NULL AND owner_kind IS NULL`, p.TaskHistoryDays}, + {"workflows", "weave_instances", "completed_at", `state IN ('SUCCEEDED','FAILED','CANCELLED') AND NOT EXISTS (SELECT 1 FROM pulse_tasks t WHERE t.namespace_id=r.namespace_id AND t.gate_id=r.gate_id AND (t.state NOT IN ('SUCCEEDED','FAILED','CANCELLED') OR t.effect_uncertain=TRUE)) AND NOT EXISTS (SELECT 1 FROM weave_inbox i WHERE i.namespace_id=r.namespace_id AND i.workflow_id=r.id AND i.state<>'PROCESSED')`, p.WorkflowHistoryDays}, + {"events", "orbit_events", "occurred_at", `NOT EXISTS (SELECT 1 FROM orbit_deliveries d WHERE d.namespace_id=r.namespace_id AND d.event_id=r.id AND d.state<>'ACKED') AND NOT EXISTS (SELECT 1 FROM weave_inbox i WHERE i.namespace_id=r.namespace_id AND i.event_id=r.id AND i.state<>'PROCESSED')`, p.EventLogDays}, + {"audit", "orbit_audit_events", "created_at", "TRUE", p.AuditLogDays}, + {"deliveries", "orbit_deliveries", "acknowledged_at", "state='ACKED'", p.DeliveryHistoryDays}, + {"inbox", "weave_inbox", "processed_at", `state='PROCESSED' AND EXISTS (SELECT 1 FROM weave_instances w WHERE w.namespace_id=r.namespace_id AND w.id=r.workflow_id AND w.state IN ('SUCCEEDED','FAILED','CANCELLED'))`, p.InboxHistoryDays}, + {"idempotency", "orbit_idempotency_keys", "created_at", "expires_at? OR (r.` + source.clock + `=? AND r.id>?))` + args = append(args, cursor.Time, cursor.Time, cursor.ID) + } + q += ` ORDER BY r.` + source.clock + `,r.id LIMIT ?` + args = append(args, limit) + rows, err := tx.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []archiveRoot + for rows.Next() { + var r archiveRoot + if err := rows.Scan(&r.id, &r.at); err != nil { + return nil, err + } + items = append(items, r) + } + return items, rows.Err() +} + +func (s *RetentionPolicyStore) PreviewRetention(ctx context.Context, t auth.Token) (retention.Preview, error) { + ctx, cancel := context.WithTimeout(ctx, 8*time.Second) + defer cancel() + out := retention.Preview{SampleLimit: 100, Items: []retention.CandidatePage{}} + err := WithTx(ctx, s.db, func(tx *sql.Tx) error { + if _, err := retentionActor(ctx, tx, t); err != nil { + return err + } + p, err := s.readPolicy(ctx, tx, t.NamespaceID, false) + if err != nil { + return err + } + out.Items = nil // WithTx 发生提交前重试时丢弃前次样本。 + out.Policy = p + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + for _, source := range archiveSources(p) { + cutoff := now.Add(-time.Duration(source.days) * 24 * time.Hour) + roots, err := archiveCandidates(ctx, tx, t.NamespaceID, source, cutoff, 101, nil) + if err != nil { + return err + } + out.Items = append(out.Items, retention.CandidatePage{Kind: source.kind, Cutoff: cutoff, Count: min(len(roots), 100), More: len(roots) > 100}) + } + return nil + }) + return out, err +} + +// archiveData 是独立历史快照，不是迁移/灾难恢复备份，也不是源数据可删除的证明。 +type archiveData struct { + Policy retention.Policy `json:"policy"` // Policy 冻结的保留策略。 + Kind string `json:"kind"` // Kind 当前批次根资源类别。 + SnapshotAt time.Time `json:"snapshot_at"` // SnapshotAt 读取数据库时间。 + Cutoff time.Time `json:"cutoff"` // Cutoff 当前批次截止时间。 + SourceRetained bool `json:"source_retained"` // SourceRetained 表示源数据库全部记录仍然保留。 + RootCount int `json:"root_count"` // RootCount 本批根记录数量。 + NextCursor string `json:"next_cursor"` // NextCursor 后续批次位置，非提交水位。 + Tables map[string][]map[string]any `json:"tables"` // Tables 固定表目录下的历史快照；二进制为十六进制，JSON及整数为文本以保留精度。 +} + +func (s *RetentionPolicyStore) ExportRetention(ctx context.Context, t auth.Token, in retention.ArchiveInput) (retention.ArchiveSnapshot, error) { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if in.Limit == 0 { + in.Limit = 20 + } + if in.Limit < 1 || in.Limit > 100 { + return retention.ArchiveSnapshot{}, fault.Invalid("limit must be 1-100") + } + // 导出不自动重放事务；失败没有成功包，源数据始终保留。 + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead}) + if err != nil { + return retention.ArchiveSnapshot{}, err + } + defer tx.Rollback() + actor, err := retentionActor(ctx, tx, t) + if err != nil { + return retention.ArchiveSnapshot{}, err + } + p, err := s.readPolicy(ctx, tx, t.NamespaceID, true) + if err != nil { + return retention.ArchiveSnapshot{}, err + } + if !p.Enabled || p.Revision != in.ExpectedRevision { + return retention.ArchiveSnapshot{}, fault.Conflict + } + var source archiveSource + for _, v := range archiveSources(p) { + if v.kind == in.Kind { + source = v + } + } + if source.kind == "" { + return retention.ArchiveSnapshot{}, fault.Invalid("unsupported archive kind") + } + now, err := dbNow(ctx, tx) + if err != nil { + return retention.ArchiveSnapshot{}, err + } + cutoff := now.Add(-time.Duration(source.days) * 24 * time.Hour) + var cursor *archiveCursor + if in.Cursor != "" { + if len(in.Cursor) > 2048 { + return retention.ArchiveSnapshot{}, fault.Invalid("invalid archive cursor") + } + 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) { + 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) + if err != nil { + return retention.ArchiveSnapshot{}, err + } + data := archiveData{Policy: p, Kind: in.Kind, SnapshotAt: now, Cutoff: cutoff, SourceRetained: true, Tables: map[string][]map[string]any{}} + 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}) + data.NextCursor = base64.RawURLEncoding.EncodeToString(raw) + } + data.RootCount = len(roots) + budget := archiveBudget{} + for _, root := range roots { + if err := archiveRows(ctx, tx, &data, &budget, source.table, "id=?", root.id); err != nil { + return retention.ArchiveSnapshot{}, err + } + if source.kind == "tasks" { + if err := archiveTaskHistory(ctx, tx, &data, &budget, root.id); err != nil { + return retention.ArchiveSnapshot{}, err + } + } + if source.kind == "events" { + if err := archiveRows(ctx, tx, &data, &budget, "orbit_result_events", "id=?", root.id); err != nil { + return retention.ArchiveSnapshot{}, err + } + } + if source.kind == "workflows" { + for _, table := range []string{"weave_steps", "weave_activations", "weave_waits", "weave_signals", "weave_poll_states", "weave_poll_rounds", "weave_inbox"} { + if err := archiveRows(ctx, tx, &data, &budget, table, "workflow_id=?", root.id); err != nil { + return retention.ArchiveSnapshot{}, err + } + } + // 流程固定版本与全部历史关联任务也一起导出，不能只保存实例头。 + 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 { + 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) + if err != nil { + return retention.ArchiveSnapshot{}, err + } + var taskIDs []core.ID + for taskRows.Next() { + var id core.ID + if err := taskRows.Scan(&id); err != nil { + taskRows.Close() + return retention.ArchiveSnapshot{}, err + } + taskIDs = append(taskIDs, id) + } + err = taskRows.Err() + taskRows.Close() + if err != nil { + return retention.ArchiveSnapshot{}, err + } + if len(taskIDs) > archiveMaxRows { + return retention.ArchiveSnapshot{}, fault.Invalid("archive too large; use a smaller batch") + } + for _, id := range taskIDs { + if err := archiveRows(ctx, tx, &data, &budget, "pulse_tasks", "id=?", id); err != nil { + return retention.ArchiveSnapshot{}, err + } + if err := archiveTaskHistory(ctx, tx, &data, &budget, id); err != nil { + return retention.ArchiveSnapshot{}, err + } + } + if err := archiveRows(ctx, tx, &data, &budget, "orbit_events", "workflow_id=?", root.id); err != nil { + return retention.ArchiveSnapshot{}, err + } + if err := archiveRows(ctx, tx, &data, &budget, "orbit_audit_events", "resource_kind='workflow' AND resource_id=?", root.id); err != nil { + return retention.ArchiveSnapshot{}, err + } + } + } + content, err := json.Marshal(data) + if err != nil { + return retention.ArchiveSnapshot{}, err + } + if len(content) > archiveMaxBytes { + return retention.ArchiveSnapshot{}, fault.Invalid("archive too large; use a smaller batch") + } + sum := sha256.Sum256(content) + digest := hex.EncodeToString(sum[:]) + // 只记录包摘要和数量，不将归档正文或秘密写到审计/日志。 + op, err := core.NewID() + if err != nil { + return retention.ArchiveSnapshot{}, err + } + audit, err := core.NewID() + if err != nil { + 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)) + if err != nil { + return retention.ArchiveSnapshot{}, err + } + if err := tx.Commit(); err != nil { + return retention.ArchiveSnapshot{}, err + } + return retention.ArchiveSnapshot{FormatVersion: 1, SHA256: digest, Content: string(content)}, nil +} + +type archiveBudget struct{ rows, bytes int } + +func archiveTaskHistory(ctx context.Context, tx *sql.Tx, d *archiveData, b *archiveBudget, id core.ID) error { + if err := archiveRows(ctx, tx, d, b, "pulse_attempts", "task_id=?", id); err != nil { + return err + } + rows, err := tx.QueryContext(ctx, "SELECT id FROM pulse_attempts WHERE namespace_id=? AND task_id=? ORDER BY id LIMIT ?", d.Policy.NamespaceID, id, archiveMaxRows+1) + if err != nil { + return err + } + defer rows.Close() + var attemptIDs []core.ID + for rows.Next() { + var attemptID core.ID + if err := rows.Scan(&attemptID); err != nil { + return err + } + attemptIDs = append(attemptIDs, attemptID) + } + if err := rows.Err(); err != nil { + return err + } + if len(attemptIDs) > archiveMaxRows { + return fault.Invalid("archive too large; use a smaller batch") + } + for _, attemptID := range attemptIDs { + for _, table := range []string{"pulse_attempt_receipts", "pulse_attempt_evidence"} { + if err := archiveRows(ctx, tx, d, b, table, "attempt_id=?", attemptID); err != nil { + return err + } + } + } + if err := archiveRows(ctx, tx, d, b, "orbit_result_events", "id IN (SELECT id FROM orbit_events WHERE namespace_id=? AND aggregate_kind='task' AND aggregate_id=?)", d.Policy.NamespaceID, id); err != nil { + return err + } + if err := archiveRows(ctx, tx, d, b, "orbit_events", "aggregate_kind='task' AND aggregate_id=?", id); err != nil { + return err + } + return archiveRows(ctx, tx, d, b, "orbit_audit_events", "resource_kind='task' AND resource_id=?", id) +} + +// archiveRows 仅由上方固定目录调用。密钥摘要不导出，业务JSON文本及大整数不经过浮点转换。 +func archiveRows(ctx context.Context, tx *sql.Tx, d *archiveData, b *archiveBudget, table, where string, args ...any) error { + q := "SELECT /*+ MAX_EXECUTION_TIME(1500) */ * FROM " + table + " WHERE namespace_id=? AND " + where + " ORDER BY id LIMIT ?" + params := append([]any{d.Policy.NamespaceID}, args...) + params = append(params, archiveMaxRows-b.rows+1) + rows, err := tx.QueryContext(ctx, q, params...) + if err != nil { + return err + } + defer rows.Close() + cols, err := rows.ColumnTypes() + if err != nil { + return err + } + for rows.Next() { + values := make([]any, len(cols)) + dest := make([]any, len(cols)) + for i := range values { + dest[i] = &values[i] + } + if err := rows.Scan(dest...); err != nil { + return err + } + record := map[string]any{} + for i, col := range cols { + name := col.Name() + if strings.Contains(name, "secret") || name == "token_digest" { + continue + } + v := values[i] + switch x := v.(type) { + case []byte: + if col.DatabaseTypeName() == "BINARY" || col.DatabaseTypeName() == "VARBINARY" { + v = hex.EncodeToString(x) + } else { + v = string(x) + } + case time.Time: + v = x.UTC().Format(time.RFC3339Nano) + case int64, uint64, float64: + v = fmt.Sprint(x) + } + record[name] = v + } + raw, err := json.Marshal(record) + if err != nil { + return err + } + b.rows++ + b.bytes += len(raw) + if b.rows > archiveMaxRows || b.bytes > archiveMaxBytes { + return fault.Invalid("archive too large; use a smaller batch") + } + d.Tables[table] = append(d.Tables[table], record) + } + return rows.Err() +} + +var _ retention.ArchiveStore = (*RetentionPolicyStore)(nil) diff --git a/internal/storage/mysql/retention_policy.go b/internal/storage/mysql/retention_policy.go new file mode 100644 index 0000000..d36d0a6 --- /dev/null +++ b/internal/storage/mysql/retention_policy.go @@ -0,0 +1,158 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "strings" + + "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" + driver "github.com/go-sql-driver/mysql" +) + +const retentionPolicyOperation = "retention.policy.update" +const retentionPolicyColumns = `revision,enabled,idempotency_days,task_history_days,workflow_history_days,event_log_days,delivery_history_days,inbox_history_days,audit_log_days,updated_at` + +// RetentionPolicyStore 保存命名空间级策略，数据库配置优先于部署默认值。 +// 不缓存配置；节点重启和跨节点读取保持数据库中同一版本。 +type RetentionPolicyStore struct { + db *sql.DB + defaults retention.Policy +} + +func NewRetentionPolicyStore(db *sql.DB, defaults ...retention.Policy) *RetentionPolicyStore { + p := retention.Defaults("") + if len(defaults) > 0 { + p = defaults[0] + } + return &RetentionPolicyStore{db: db, defaults: p} +} + +var _ retention.Store = (*RetentionPolicyStore)(nil) + +func (s *RetentionPolicyStore) defaultPolicy(namespace core.ID) retention.Policy { + p := s.defaults + p.NamespaceID = namespace + p.Source = "default" + return p +} + +type retentionQuery interface { + QueryRowContext(context.Context, string, ...any) *sql.Row +} + +func (s *RetentionPolicyStore) readPolicy(ctx context.Context, db retentionQuery, namespace core.ID, lock bool) (retention.Policy, error) { + p := s.defaultPolicy(namespace) + q := `SELECT ` + retentionPolicyColumns + ` FROM orbit_retention_policies WHERE namespace_id=?` + if lock { + q += " FOR UPDATE" + } + err := db.QueryRowContext(ctx, q, namespace).Scan(&p.Revision, &p.Enabled, &p.IdempotencyDays, &p.TaskHistoryDays, &p.WorkflowHistoryDays, &p.EventLogDays, &p.DeliveryHistoryDays, &p.InboxHistoryDays, &p.AuditLogDays, &p.UpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return p, p.Validate() + } + if err != nil { + return retention.Policy{}, err + } + p.Source = "database" + return p, p.Validate() +} + +// GetRetentionPolicy 供经过会话认证的控制台和数据库管理工具读取配置。 +func (s *RetentionPolicyStore) GetRetentionPolicy(ctx context.Context, namespace core.ID) (retention.Policy, error) { + if namespace.Validate() != nil { + return retention.Policy{}, fault.Invalid("invalid namespace") + } + return s.readPolicy(ctx, s.db, namespace, false) +} + +// retentionActor 在策略事务内验证控制台会话。普通 SDK 凭据不能修改保留策略。 +// 管理员必须具备当前命名空间全部队列的管理范围；不添加角色体系或跨命名空间权限。 +func retentionActor(ctx context.Context, tx *sql.Tx, token auth.Token) (core.ID, error) { + p, err := authenticate(ctx, tx, token) + if err != nil { + return "", err + } + all := false + for _, q := range p.Queues { + if q == "*" { + all = true + } + } + if !all || !p.Allows("queues:manage") { + return "", fault.Forbidden + } + var user core.ID + err = tx.QueryRowContext(ctx, `SELECT u.id FROM orbit_console_sessions s + JOIN orbit_console_users u ON u.id=s.user_id AND u.namespace_id=s.namespace_id AND u.credential_id=s.credential_id + WHERE s.namespace_id=? AND s.credential_id=? AND s.token_digest=? AND s.revoked_at IS NULL + AND s.expires_at>UTC_TIMESTAMP(6) AND u.enabled=TRUE FOR SHARE`, token.NamespaceID, token.CredentialID, token.Digest).Scan(&user) + if errors.Is(err, sql.ErrNoRows) { + return "", fault.Unauthorized + } + return user, err +} + +// UpdateRetentionPolicy 在同一事务中核验版本、保存策略、前后快照、账号身份和幂等回执。 +func (s *RetentionPolicyStore) UpdateRetentionPolicy(ctx context.Context, token auth.Token, key string, in retention.UpdateInput) (retention.Policy, error) { + if strings.TrimSpace(key) == "" || len(key) > 128 { + return retention.Policy{}, fault.Invalid("idempotency key must be 1-128 bytes") + } + want, err := in.ToPolicy(token.NamespaceID) + if err != nil { + return retention.Policy{}, fault.Invalid(err.Error()) + } + var out retention.Policy + err = WithTx(ctx, s.db, func(tx *sql.Tx) error { + user, err := retentionActor(ctx, tx, token) + if err != nil { + return err + } + identity, cached, err := reserveIdentity(ctx, tx, token.NamespaceID, retentionPolicyOperation, "", key, "retention_policy", token.NamespaceID, in) + if err != nil { + return err + } + if cached != nil { + return json.Unmarshal(cached, &out) + } + current, err := s.readPolicy(ctx, tx, token.NamespaceID, true) + if err != nil { + return err + } + if current.Revision != in.ExpectedRevision { + return fault.Conflict + } + if current.Revision == 0 { + _, err = tx.ExecContext(ctx, `INSERT INTO orbit_retention_policies(namespace_id,enabled,idempotency_days,task_history_days,workflow_history_days,event_log_days,delivery_history_days,inbox_history_days,audit_log_days) VALUES (?,?,?,?,?,?,?,?,?)`, token.NamespaceID, want.Enabled, want.IdempotencyDays, want.TaskHistoryDays, want.WorkflowHistoryDays, want.EventLogDays, want.DeliveryHistoryDays, want.InboxHistoryDays, want.AuditLogDays) + var me *driver.MySQLError + if errors.As(err, &me) && me.Number == 1062 { + return fault.Conflict + } + } else { + _, err = tx.ExecContext(ctx, `UPDATE orbit_retention_policies SET revision=revision+1,enabled=?,idempotency_days=?,task_history_days=?,workflow_history_days=?,event_log_days=?,delivery_history_days=?,inbox_history_days=?,audit_log_days=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND revision=?`, want.Enabled, want.IdempotencyDays, want.TaskHistoryDays, want.WorkflowHistoryDays, want.EventLogDays, want.DeliveryHistoryDays, want.InboxHistoryDays, want.AuditLogDays, token.NamespaceID, in.ExpectedRevision) + } + if err != nil { + return err + } + out, err = s.readPolicy(ctx, tx, token.NamespaceID, false) + if err != nil { + return err + } + auditID, err := core.NewID() + if err != nil { + return err + } + old, _ := json.Marshal(current) + updated, _ := json.Marshal(out) + _, 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 (?,?,?,?,?,'retention_policy',?,?,?,?)", token.NamespaceID, auditID, identity, "console:"+string(user), retentionPolicyOperation, token.NamespaceID, in.Reason, string(old), string(updated)) + if err != nil { + return err + } + return saveIdentity(ctx, tx, token.NamespaceID, identity, out) + }) + return out, err +} diff --git a/internal/storage/mysql/retention_policy_integration_test.go b/internal/storage/mysql/retention_policy_integration_test.go new file mode 100644 index 0000000..30853cc --- /dev/null +++ b/internal/storage/mysql/retention_policy_integration_test.go @@ -0,0 +1,228 @@ +package mysql + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "os" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/retention" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +func TestIntegrationRetentionPolicyIsSharedAuditedAndIdempotent(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("set ORBIT_TEST_MYSQL_DSN to a disposable database, or run make integration") + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 8, 2) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + + var boot BootstrapResult + app := testID(t) + if err := Bootstrap(ctx, db, BootstrapInput{ + Application: string(app), Namespace: "default", Environment: "test", Queue: "admin", + ConsoleUsername: "admin-" + string(app), ConsolePassword: "a-strong-password-123", + }, func(v BootstrapResult) error { + boot = v + return nil + }); err != nil { + t.Fatal(err) + } + + login, err := NewConsoleAuthStore(db).Login(ctx, "admin-"+string(app), "a-strong-password-123", "retention-test", "127.0.0.1") + if err != nil { + t.Fatal(err) + } + token, err := auth.Parse(login.Token) + if err != nil { + t.Fatal(err) + } + defaults := retention.Defaults(boot.NamespaceID) + store := NewRetentionPolicyStore(db, defaults) + got, err := store.GetRetentionPolicy(ctx, boot.NamespaceID) + if err != nil || got.Source != "default" || got.Revision != 0 || got.TaskHistoryDays != 365 { + t.Fatalf("default policy: %+v %v", got, err) + } + + in := retention.UpdateInput{ + ExpectedRevision: 0, Enabled: true, IdempotencyDays: 120, + TaskHistoryDays: 730, WorkflowHistoryDays: 730, EventLogDays: 365, + DeliveryHistoryDays: 120, InboxHistoryDays: 120, AuditLogDays: 2555, + Reason: "production retention policy review", + } + updated, err := store.UpdateRetentionPolicy(ctx, token, "retention-change-1", in) + if err != nil { + t.Fatal(err) + } + if updated.Source != "database" || updated.Revision != 1 || !updated.Enabled || updated.TaskHistoryDays != 730 { + t.Fatalf("updated policy: %+v", updated) + } + // A second store represents another Orbit node; it must read the same MySQL policy. + other := NewRetentionPolicyStore(db, retention.Defaults(boot.NamespaceID)) + shared, err := other.GetRetentionPolicy(ctx, boot.NamespaceID) + if err != nil || shared.Revision != 1 || shared.TaskHistoryDays != 730 || !shared.Enabled { + t.Fatalf("cross-node policy read: %+v %v", shared, err) + } + replayed, err := store.UpdateRetentionPolicy(ctx, token, "retention-change-1", in) + if err != nil || replayed.Revision != updated.Revision || replayed.TaskHistoryDays != updated.TaskHistoryDays { + t.Fatalf("idempotent replay: %+v %v", replayed, err) + } + conflict := in + conflict.ExpectedRevision = 0 + conflict.TaskHistoryDays = 900 + if _, err := store.UpdateRetentionPolicy(ctx, token, "retention-change-2", conflict); faultCode(err) != "STATE_CONFLICT" { + t.Fatalf("stale revision accepted: %v", err) + } + var audits int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND action='retention.policy.update' AND resource_kind='retention_policy'`, boot.NamespaceID).Scan(&audits); err != nil { + t.Fatal(err) + } + if audits != 1 { + t.Fatalf("audit rows=%d want 1 after idempotent replay", audits) + } + var actor, reason string + if err := db.QueryRowContext(ctx, "SELECT actor_id,reason FROM orbit_audit_events WHERE namespace_id=? AND action='retention.policy.update'", boot.NamespaceID).Scan(&actor, &reason); err != nil { + t.Fatal(err) + } + if actor == "" || reason != in.Reason || actor == "mysql:database" { + t.Fatalf("unexpected retention audit identity: actor=%q reason=%q", actor, reason) + } + t.Logf("retention policy revision %d is shared across stores, audited once, and safe on replay", updated.Revision) +} + +func faultCode(err error) string { + if err == nil { + return "" + } + return fault.Code(err) +} + +func TestIntegrationRetentionDiskGrowthAndPreview(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("set ORBIT_TEST_MYSQL_DSN to a disposable database, or run make integration") + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 12, 4) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + var boot BootstrapResult + app := testID(t) + if err := Bootstrap(ctx, db, BootstrapInput{ + Application: string(app), Namespace: "growth", Environment: "test", Queue: "growth", + ConsoleUsername: "growth-admin-" + string(app), ConsolePassword: "a-strong-password-123", + }, func(v BootstrapResult) error { boot = v; return nil }); err != nil { + t.Fatal(err) + } + apiToken, err := auth.Parse(boot.Token) + if err != nil { + t.Fatal(err) + } + login, err := NewConsoleAuthStore(db).Login(ctx, "growth-admin-"+string(app), "a-strong-password-123", "retention-growth-test", "127.0.0.1") + if err != nil { + t.Fatal(err) + } + consoleToken, err := auth.Parse(login.Token) + if err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(ctx, "ANALYZE TABLE pulse_tasks, orbit_events, orbit_audit_events"); err != nil { + t.Fatal(err) + } + before := retentionTableBytes(t, ctx, db) + serviceStore := service.New(NewTaskStore(db)) + // 256 rows × 32 KiB is deliberately larger than one InnoDB allocation page, + // so the information_schema sample must observe physical table growth. + payload := json.RawMessage(`{"blob":"` + strings.Repeat("x", 32768) + `"}`) + ids := make([]string, 0, 256) + for i := 0; i < 256; i++ { + out, err := serviceStore.Submit(ctx, apiToken, "growth-"+string(app)+"-"+string(rune('a'+i%26))+"-"+string(testID(t)), service.SubmitInput{ + QueueID: boot.QueueID, Handler: "retention.growth", HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, + TaskTimeoutMS: 60000, RecoverySafety: pulse.RecoveryReconcileOnly, Retry: pulse.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 30000}, + }) + if err != nil { + t.Fatal(err) + } + ids = append(ids, string(out.Task.ID)) + } + for _, id := range ids { + if _, err := db.ExecContext(ctx, `UPDATE pulse_tasks SET state='FAILED',completed_at=UTC_TIMESTAMP(6)-INTERVAL 2 DAY,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=UNHEX(?)`, boot.NamespaceID, id); err != nil { + t.Fatal(err) + } + } + if _, err := db.ExecContext(ctx, "ANALYZE TABLE pulse_tasks, orbit_events, orbit_audit_events"); err != nil { + t.Fatal(err) + } + after := retentionTableBytes(t, ctx, db) + if after <= before { + t.Fatalf("retention table bytes did not grow during bounded growth sample: before=%d after=%d", before, after) + } + store := NewRetentionPolicyStore(db, retention.Defaults(boot.NamespaceID)) + _, err = store.UpdateRetentionPolicy(ctx, consoleToken, "growth-retention-policy", retention.UpdateInput{ + ExpectedRevision: 0, Enabled: true, IdempotencyDays: 90, TaskHistoryDays: 1, + WorkflowHistoryDays: 30, EventLogDays: 30, DeliveryHistoryDays: 30, InboxHistoryDays: 30, + AuditLogDays: 365, Reason: "bounded disk growth rehearsal", + }) + if err != nil { + t.Fatal(err) + } + preview, err := store.PreviewRetention(ctx, consoleToken) + if err != nil { + t.Fatal(err) + } + var taskSample retention.CandidatePage + for _, item := range preview.Items { + if item.Kind == "tasks" { + taskSample = item + } + } + if taskSample.Count == 0 { + t.Fatalf("backdated terminal tasks were not visible as retention candidates: %+v", preview.Items) + } + snapshot, err := store.ExportRetention(ctx, consoleToken, retention.ArchiveInput{Kind: "tasks", Limit: 1, ExpectedRevision: 1}) + if err != nil { + t.Fatal(err) + } + if snapshot.FormatVersion != 1 || snapshot.Content == "" { + t.Fatalf("invalid retention snapshot envelope: %+v", snapshot) + } + sum := sha256.Sum256([]byte(snapshot.Content)) + if snapshot.SHA256 != hex.EncodeToString(sum[:]) || !json.Valid([]byte(snapshot.Content)) || !strings.Contains(snapshot.Content, `"source_retained":true`) { + t.Fatalf("retention snapshot integrity/source marker failed: sha=%s content=%s", snapshot.SHA256, snapshot.Content[:min(len(snapshot.Content), 200)]) + } + t.Logf("bounded retention growth sample inserted %d terminal tasks; bytes before=%d after=%d delta=%d; preview tasks=%d more=%v", len(ids), before, after, after-before, taskSample.Count, taskSample.More) +} + +func retentionTableBytes(t *testing.T, ctx context.Context, db interface { + QueryRowContext(context.Context, string, ...any) *sql.Row +}) int64 { + t.Helper() + var bytes int64 + if err := db.QueryRowContext(ctx, `SELECT COALESCE(SUM(data_length+index_length),0) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME IN ('pulse_tasks','orbit_events','orbit_audit_events')`).Scan(&bytes); err != nil { + t.Fatal(err) + } + return bytes +} diff --git a/internal/transport/httpserver/console/app.js b/internal/transport/httpserver/console/app.js index 31249ac..0709dcd 100644 --- a/internal/transport/httpserver/console/app.js +++ b/internal/transport/httpserver/console/app.js @@ -1,8 +1,8 @@ (() => { - const state = { token: '', authenticated: false, queue: '', base: location.origin, consoleConfig: null, timer: null, connected: false, loading: false, selectedTask: null, actionMode: '', selectedSchedule: null, selectedWorkflow: null, workflowActionMode: '', workflowRecoveryMode: '', workflowRecoveryContext: null, selectedSubscription: null, subscriptionActionMode: '', selectedDelivery: null }; + 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: '结果订阅' }; + const pageNames = { overview: '总览', tasks: '任务执行', schedules: '定时计划', workflows: '流程实例', subscriptions: '结果订阅', retention: '保留策略' }; function showToast(message) { const toast = $('toast'); toast.textContent = message; toast.classList.add('show'); @@ -28,6 +28,7 @@ text('page-title', pageNames[next]); if (updateHash && location.hash !== `#/${next}`) location.hash = `/${next}`; window.scrollTo({ top: 0, behavior: 'smooth' }); + if (next === 'retention' && state.authenticated) loadRetention(); } function redirectToLogin() { const returnTo = `${location.pathname}${location.search}${location.hash}`; @@ -62,6 +63,80 @@ // 配置只是已登录会话的默认队列，失败时业务 API 会返回明确错误。 } } + function renderRetention(policy) { + if (!policy) return; + state.retention = policy; + $('retention-enabled').checked = Boolean(policy.enabled); + $('retention-idempotency').value = policy.idempotency_days ?? ''; + $('retention-task').value = policy.task_history_days ?? ''; + $('retention-workflow').value = policy.workflow_history_days ?? ''; + $('retention-event').value = policy.event_log_days ?? ''; + $('retention-delivery').value = policy.delivery_history_days ?? ''; + $('retention-inbox').value = policy.inbox_history_days ?? ''; + $('retention-audit').value = policy.audit_log_days ?? ''; + text('retention-source', policy.source === 'database' ? '已保存配置' : '默认策略'); + text('retention-revision', policy.revision ? String(policy.revision) : '默认'); + text('retention-updated', policy.updated_at ? formatTime(policy.updated_at) : '尚未保存'); + } + async function loadRetention() { + try { + const response = await request('/console/retention'); + renderRetention(extractData(response)); + } catch (error) { + $('retention-error').textContent = error.message || '保留策略读取失败'; + } + } + function retentionNumber(id) { + const value = Number($(id).value); + return Number.isSafeInteger(value) && value >= 1 && value <= 3650 ? value : 0; + } + async function submitRetention(event) { + event.preventDefault(); + const fields = { idempotency: 'retention-idempotency', task: 'retention-task', workflow: 'retention-workflow', event: 'retention-event', delivery: 'retention-delivery', inbox: 'retention-inbox', audit: 'retention-audit' }; + const values = Object.fromEntries(Object.entries(fields).map(([key, id]) => [key, retentionNumber(id)])); + if (Object.values(values).some(value => !value)) { $('retention-error').textContent = '所有保留天数必须是 1 到 3650 的整数。'; return; } + const reason = $('retention-reason').value.trim(); + if (!reason) { $('retention-error').textContent = '请填写变更原因。'; return; } + const button = $('retention-submit'); button.disabled = true; $('retention-error').textContent = ''; + const current = state.retention || {}; + try { + const response = await request('/console/retention', { method: 'PUT', headers: { 'Content-Type': 'application/json', 'Idempotency-Key': operationKey('retention-policy') }, body: JSON.stringify({ expected_revision: String(current.revision || 0), enabled: $('retention-enabled').checked, idempotency_days: values.idempotency, task_history_days: values.task, workflow_history_days: values.workflow, event_log_days: values.event, delivery_history_days: values.delivery, inbox_history_days: values.inbox, audit_log_days: values.audit, reason }) }); + renderRetention(extractData(response)); $('retention-reason').value = ''; showToast('保留策略已保存，已写入管理审计'); + } catch (error) { $('retention-error').textContent = error.message || '保留策略保存失败'; } + finally { button.disabled = false; } + } + function renderRetentionPreview(preview) { + const el = $('retention-preview'); + const items = Array.isArray(preview?.items) ? preview.items : []; + if (!items.length) { el.innerHTML = '当前策略没有可预览的候选数据。'; return; } + const labels = { tasks: '任务历史', workflows: '流程历史', events: '事件日志', audit: '管理审计', deliveries: '交付历史', inbox: 'Inbox 历史', idempotency: '幂等记录' }; + el.innerHTML = `
候选预览仅展示样本，不会删除数据
${items.map(item => `
${escapeHTML(labels[item.kind] || item.kind)}${escapeHTML(String(item.count || 0))} 条截止 ${escapeHTML(formatTime(item.cutoff))}${item.more ? ' · 还有更多' : ''}
`).join('')}`; + } + async function previewRetention() { + const button = $('retention-preview-button'); button.disabled = true; $('retention-error').textContent = ''; + try { + const response = await request('/console/retention/preview'); + renderRetentionPreview(extractData(response)); + } catch (error) { $('retention-error').textContent = error.message || '候选预览失败'; } + finally { button.disabled = false; } + } + async function exportRetention() { + const button = $('retention-export-button'); button.disabled = true; $('retention-error').textContent = ''; + try { + const kind = $('retention-kind').value; + const current = state.retention || {}; + const response = await request('/console/retention/export', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ kind, limit: 20, expected_revision: String(current.revision || 0) }) }); + const snapshot = extractData(response); + const content = String(snapshot.content || ''); + if (!content) throw new Error('服务端返回了空的归档快照'); + const blob = new Blob([content], { type: 'application/json;charset=utf-8' }); + const url = URL.createObjectURL(blob); const anchor = document.createElement('a'); + anchor.href = url; anchor.download = `orbit-retention-${kind}-${new Date().toISOString().replace(/[:.]/g, '-')}.json`; anchor.click(); + setTimeout(() => URL.revokeObjectURL(url), 1000); + showToast(`快照已下载 · SHA256 ${String(snapshot.sha256 || '').slice(0, 12)}…`); + } catch (error) { $('retention-error').textContent = error.message || '归档快照导出失败'; } + finally { button.disabled = false; } + } function renderConnectionSummary() { text('summary-api', state.base || location.origin); text('summary-api-hint', state.connected ? '接口响应正常，按 10 秒周期刷新' : 'API 地址自动跟随当前页面'); @@ -473,13 +548,13 @@ finally { state.loading = false; $('refresh-button').classList.remove('spinning'); } } async function logout() { await fetch('/console/auth/logout', { method: 'POST', credentials: 'same-origin' }); redirectToLogin(); } - $('logout-hero').addEventListener('click', logout); $('logout-button').addEventListener('click', logout); $('logout-inline').addEventListener('click', logout); $('refresh-button').addEventListener('click', refresh); + $('logout-hero').addEventListener('click', logout); $('logout-button').addEventListener('click', logout); $('logout-inline').addEventListener('click', logout); $('refresh-button').addEventListener('click', refresh); $('retention-form').addEventListener('submit', submitRetention); $('retention-preview-button').addEventListener('click', previewRetention); $('retention-export-button').addEventListener('click', exportRetention); $('close-task-drawer').addEventListener('click', closeTaskDrawer); $('task-drawer-backdrop').addEventListener('click', event => { if (event.target.id === 'task-drawer-backdrop') closeTaskDrawer(); }); $('action-cancel').addEventListener('click', closeActionForm); $('task-action-form').addEventListener('submit', submitAction); $('task-cancel-button').addEventListener('click', () => openActionForm('cancel')); $('task-retry-button').addEventListener('click', () => openActionForm('retry')); $('close-schedule-drawer').addEventListener('click', closeScheduleDrawer); $('schedule-drawer-backdrop').addEventListener('click', event => { if (event.target.id === 'schedule-drawer-backdrop') closeScheduleDrawer(); }); $('schedule-control-button').addEventListener('click', openScheduleAction); $('schedule-action-cancel').addEventListener('click', closeScheduleAction); $('schedule-action-form').addEventListener('submit', submitScheduleAction); $('close-workflow-drawer').addEventListener('click', closeWorkflowDrawer); $('workflow-drawer-backdrop').addEventListener('click', event => { if (event.target.id === 'workflow-drawer-backdrop') closeWorkflowDrawer(); }); $('workflow-control-button').addEventListener('click', () => openWorkflowAction(state.selectedWorkflow?.paused ? 'RESUME' : 'PAUSE')); $('workflow-cancel-button').addEventListener('click', () => openWorkflowAction('CANCEL')); $('workflow-action-cancel').addEventListener('click', closeWorkflowAction); $('workflow-action-form').addEventListener('submit', submitWorkflowAction); $('close-subscription-drawer').addEventListener('click', closeSubscriptionDrawer); $('subscription-drawer-backdrop').addEventListener('click', event => { if (event.target.id === 'subscription-drawer-backdrop') closeSubscriptionDrawer(); }); $('subscription-control-button').addEventListener('click', openSubscriptionAction); $('subscription-action-cancel').addEventListener('click', closeSubscriptionAction); $('subscription-action-form').addEventListener('submit', submitSubscriptionAction); $('subscription-retry-cancel').addEventListener('click', closeSubscriptionRetry); $('subscription-retry-form').addEventListener('submit', submitSubscriptionRetry); document.addEventListener('keydown', event => { if (event.key === 'Escape') { if (!$('task-action-form').hidden) closeActionForm(); else if (!$('schedule-action-form').hidden) closeScheduleAction(); else if (!$('workflow-action-form').hidden) closeWorkflowAction(); else if (!$('subscription-action-form').hidden) closeSubscriptionAction(); else if (!$('subscription-retry-form').hidden) closeSubscriptionRetry(); else if (!$('task-drawer-backdrop').hidden) closeTaskDrawer(); else if (!$('schedule-drawer-backdrop').hidden) closeScheduleDrawer(); else if (!$('workflow-drawer-backdrop').hidden) closeWorkflowDrawer(); else if (!$('subscription-drawer-backdrop').hidden) closeSubscriptionDrawer(); } }); $('workflow-recovery-actions').addEventListener('click', event => { const button = event.target.closest('[data-recovery-action]'); if (!button) return; let context = {}; try { context = JSON.parse(button.dataset.recoveryContext || '{}'); } catch { showToast('恢复按钮上下文无效，请重新读取流程'); return; } openWorkflowRecovery(button.dataset.recoveryAction, context); }); $('workflow-recovery-cancel').addEventListener('click', closeWorkflowRecovery); $('workflow-recovery-form').addEventListener('submit', submitWorkflowRecovery); document.addEventListener('keydown', event => { if (event.key === 'Escape' && !$('workflow-recovery-form').hidden) closeWorkflowRecovery(); }); 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(() => refresh()); }).catch(() => redirectToLogin()); + ensureSession().then(ok => { if (ok) return loadConsoleConfig().then(() => Promise.all([loadRetention(), refresh()])); }).catch(() => redirectToLogin()); state.timer = setInterval(() => { if (state.authenticated) refresh(); }, 10000); })(); diff --git a/internal/transport/httpserver/console/index.html b/internal/transport/httpserver/console/index.html index 5a63968..3d24def 100644 --- a/internal/transport/httpserver/console/index.html +++ b/internal/transport/httpserver/console/index.html @@ -23,6 +23,7 @@ +
@@ -129,6 +130,37 @@
+
+
+
+
DATA / RETENTION

历史数据保留策略

默认策略
+

这里配置各类历史记录的最短保留时间。保存策略只记录配置，不会立即删除数据；归档执行器还会核对终态、未确认结果、备份和审计要求。

+
+
允许归档执行器使用策略默认关闭。打开后仍需独立归档任务执行，未确认任务、交付和 Inbox 不会按窗口清理。
+
+ + + + + + + +
+
当前版本 默认最后修改 —
+ + +
+
+ +
+
+
保存策略后可预览当前窗口下的候选数据；预览不会删除任何记录。
+ +
+
+
+
+
diff --git a/internal/transport/httpserver/console/styles.css b/internal/transport/httpserver/console/styles.css index 0973845..ab5fbed 100644 --- a/internal/transport/httpserver/console/styles.css +++ b/internal/transport/httpserver/console/styles.css @@ -6,3 +6,5 @@ .detail-row-action{border:1px solid rgba(255,179,107,.28);border-radius:8px;padding:6px 9px;color:var(--orange);background:rgba(255,179,107,.07);font:10px inherit;cursor:pointer;white-space:nowrap}.detail-row-action:hover{border-color:rgba(255,179,107,.65);color:#ffe0bb;background:rgba(255,179,107,.13)} .recovery-block{border-color:rgba(255,179,107,.2);background:linear-gradient(135deg,rgba(255,179,107,.045),rgba(98,230,212,.025))}.recovery-intro{margin:0 0 12px;color:#8399b0;font-size:10px;line-height:1.6}.recovery-action-grid{display:grid;grid-template-columns:repeat(2,minmax(0,1fr));gap:9px}.recovery-action-button{display:flex;flex-direction:column;align-items:flex-start;gap:4px;border:1px solid rgba(255,179,107,.2);border-radius:10px;padding:10px 11px;color:#e6f0ff;background:rgba(4,12,23,.4);text-align:left;cursor:pointer;transition:border-color .2s,background .2s,transform .2s}.recovery-action-button:hover,.recovery-action-button:focus{outline:0;border-color:rgba(98,230,212,.6);background:rgba(98,230,212,.08);transform:translateY(-1px)}.recovery-action-button b{font-size:10px;font-weight:650}.recovery-action-button small{color:#7890a8;font-size:9px;line-height:1.4}.recovery-form{border-color:rgba(255,179,107,.22)}.recovery-snapshot{margin:0 0 14px;padding:8px 10px;border-radius:8px;background:rgba(98,230,212,.06);color:#8db4b6;font:10px ui-monospace,SFMono-Regular,Menlo,monospace}.recovery-inline-fields{display:grid;grid-template-columns:150px minmax(0,1fr);gap:10px}.recovery-inline-fields select{width:100%;border:1px solid rgba(156,183,218,.15);outline:0;border-radius:9px;background:rgba(4,12,23,.45);color:#dceafb;font:inherit;font-size:10px;padding:10px}.recovery-inline-fields select:focus{border-color:rgba(98,230,212,.58);box-shadow:0 0 0 3px rgba(98,230,212,.07)}@media(max-width:780px){.recovery-action-grid{grid-template-columns:1fr}.recovery-inline-fields{grid-template-columns:1fr}} .overview-grid{grid-template-columns:minmax(0,1.45fr) minmax(300px,.7fr)}.stats-panel,.connection-summary-panel{min-height:326px}.stat-grid{display:grid;grid-template-columns:repeat(4,1fr);gap:10px}.stat-tile{padding:14px 13px;border:1px solid rgba(156,183,218,.1);border-radius:13px;background:rgba(4,13,24,.28)}.stat-tile span{display:block;color:#7188a3;font-size:9px}.stat-tile strong{display:block;margin-top:10px;color:#e4f1fc;font-size:23px;letter-spacing:-.04em}.stat-tile small{display:block;margin-top:5px;color:#607791;font-size:9px;white-space:nowrap;overflow:hidden;text-overflow:ellipsis}.distribution-block{margin-top:22px;padding-top:18px;border-top:1px solid rgba(156,183,218,.1)}.distribution-heading{display:flex;align-items:center;justify-content:space-between;margin-bottom:14px;color:#afc2d6;font-size:11px}.distribution-heading small{color:#687f99;font-size:9px}.distribution-list{display:grid;gap:10px}.distribution-row{display:grid;grid-template-columns:42px minmax(0,1fr) 28px;gap:10px;align-items:center;color:#7187a1;font-size:9px}.distribution-row b{text-align:right;color:#c1d2e2;font-size:10px;font-weight:500}.status-track{height:5px;border-radius:5px;background:#1c3048;overflow:hidden}.status-track i{display:block;width:0;height:100%;border-radius:5px;transition:width .45s ease}.distribution-row:nth-child(1) .status-track i{background:var(--cyan)}.distribution-row:nth-child(2) .status-track i{background:var(--blue)}.distribution-row:nth-child(3) .status-track i{background:var(--purple)}.distribution-row:nth-child(4) .status-track i{background:var(--orange)}.connection-summary-copy{display:flex;align-items:center;gap:12px;padding:14px;border:1px solid rgba(98,230,212,.12);border-radius:14px;background:rgba(4,15,26,.3)}.connection-summary-icon{display:grid;place-items:center;width:33px;height:33px;border-radius:11px;background:rgba(98,230,212,.1);color:var(--cyan);font-size:17px}.connection-summary-copy b{display:block;color:#d5e5f4;font-size:12px;max-width:230px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap}.connection-summary-copy small{display:block;margin-top:5px;color:#68809a;font-size:9px}.connection-summary-grid{display:grid;grid-template-columns:repeat(2,1fr);gap:1px;margin-top:12px;border:1px solid rgba(156,183,218,.1);border-radius:13px;overflow:hidden}.connection-summary-grid>div{padding:12px;background:rgba(4,13,24,.25)}.connection-summary-grid span{display:block;color:#647b96;font-size:9px;margin-bottom:5px}.connection-summary-grid b{display:block;color:#b9cde0;font-size:10px;white-space:nowrap;overflow:hidden;text-overflow:ellipsis}.connection-summary-note{min-height:31px;margin:15px 0 14px;color:#6f86a0;font-size:10px;line-height:1.6}.connection-summary-panel>.quiet-button{width:100%;justify-content:center}.connection-settings-intro{margin:0 0 18px}.connection-drawer{width:min(510px,100%)}@media(max-width:1200px){.stat-grid{grid-template-columns:repeat(2,1fr)}}@media(max-width:780px){.overview-grid{grid-template-columns:1fr}.stat-grid{grid-template-columns:repeat(2,1fr)}.connection-drawer{width:100%}} +.retention-panel{max-width:980px}.retention-panel .panel-description{max-width:760px}.retention-form{max-width:820px}.retention-switch{display:flex;align-items:center;justify-content:space-between;gap:20px;padding:16px 17px;border:1px solid rgba(98,230,212,.16);border-radius:14px;background:rgba(4,15,26,.34)}.retention-switch b{display:block;color:#dceafb;font-size:11px;font-weight:550}.retention-switch small{display:block;margin-top:6px;color:#7086a0;font-size:9px;line-height:1.5}.switch{position:relative;display:block;width:43px;height:24px;flex:0 0 43px}.switch input{position:absolute;opacity:0;width:0;height:0}.switch span{position:absolute;inset:0;border-radius:999px;background:#243852;cursor:pointer;transition:.2s}.switch span:before{content:"";position:absolute;width:18px;height:18px;left:3px;top:3px;border-radius:50%;background:#8ca3ba;transition:.2s}.switch input:checked+span{background:rgba(98,230,212,.55);box-shadow:0 0 18px rgba(98,230,212,.18)}.switch input:checked+span:before{transform:translateX(19px);background:#ecfffb}.retention-grid{display:grid;grid-template-columns:repeat(3,minmax(0,1fr));gap:11px;margin-top:15px}.retention-field{display:block;padding:14px;border:1px solid rgba(156,183,218,.1);border-radius:13px;background:rgba(4,13,24,.28)}.retention-field span{color:#afc2d6;font-size:10px}.retention-field small{float:right;color:#637b96;font-size:9px}.retention-field input{display:block;width:100%;margin-top:12px;border:1px solid rgba(156,183,218,.15);outline:0;border-radius:9px;background:rgba(4,12,23,.45);color:#dceafb;font:inherit;font-size:13px;padding:10px}.retention-field input:focus{border-color:rgba(98,230,212,.58);box-shadow:0 0 0 3px rgba(98,230,212,.07)}.retention-field-wide{grid-column:span 3}.retention-meta{display:flex;gap:25px;margin-top:16px;color:#637b96;font-size:9px}.retention-meta b{margin-left:5px;color:#afc2d6;font-weight:500}.retention-form>.field-label{margin-top:22px}.retention-form>textarea{width:100%;border:1px solid rgba(156,183,218,.15);outline:0;border-radius:9px;background:rgba(4,12,23,.45);color:#dceafb;font:inherit;font-size:10px;padding:10px;resize:vertical}.retention-form>textarea:focus{border-color:rgba(98,230,212,.58);box-shadow:0 0 0 3px rgba(98,230,212,.07)}.retention-form .action-form-buttons{justify-content:flex-start}.retention-form .action-error{min-height:18px}@media(max-width:780px){.retention-grid{grid-template-columns:1fr 1fr}.retention-field-wide{grid-column:span 2}.retention-meta{flex-direction:column;gap:7px}} +.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}} diff --git a/internal/transport/httpserver/retention.go b/internal/transport/httpserver/retention.go new file mode 100644 index 0000000..fda6937 --- /dev/null +++ b/internal/transport/httpserver/retention.go @@ -0,0 +1,106 @@ +package httpserver + +import ( + "context" + "net/http" + "strings" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/retention" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +// RegisterConsoleRetention 注册保留策略配置、候选预览和只读归档导出。 +// 所有接口只接受控制台 HttpOnly 会话，不接受业务 Bearer token。 +func RegisterConsoleRetention(server *http.Server, consoleAuth RetentionConsoleAuth, store retention.Store, archive retention.ArchiveStore) { + if server == nil || consoleAuth == nil || store == nil { + return + } + mux, ok := server.Handler.(*http.ServeMux) + if !ok { + return + } + 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 + } + mux.HandleFunc("GET /console/retention", func(w http.ResponseWriter, r *http.Request) { + t, ok := check(w, r) + if !ok { + return + } + p, err := store.GetRetentionPolicy(r.Context(), t.NamespaceID) + if err != nil { + writeError(w, "", err) + return + } + writeConsoleJSON(w, 200, map[string]any{"data": p}) + }) + mux.HandleFunc("PUT /console/retention", func(w http.ResponseWriter, r *http.Request) { + t, ok := check(w, r) + if !ok { + return + } + var in retention.UpdateInput + if err := decodeBody(r, &in); err != nil { + writeError(w, "", err) + return + } + keys := r.Header.Values("Idempotency-Key") + if len(keys) != 1 { + writeError(w, "", fault.Invalid("exactly one Idempotency-Key is required")) + return + } + p, err := store.UpdateRetentionPolicy(r.Context(), t, keys[0], in) + if err != nil { + writeError(w, "", err) + return + } + writeConsoleJSON(w, 200, map[string]any{"data": p}) + }) + if archive == nil { + return + } + mux.HandleFunc("GET /console/retention/preview", func(w http.ResponseWriter, r *http.Request) { + t, ok := check(w, r) + if !ok { + return + } + p, err := archive.PreviewRetention(r.Context(), t) + if err != nil { + writeError(w, "", err) + return + } + writeConsoleJSON(w, 200, map[string]any{"data": p}) + }) + mux.HandleFunc("POST /console/retention/export", func(w http.ResponseWriter, r *http.Request) { + t, ok := check(w, r) + if !ok { + return + } + var in retention.ArchiveInput + if err := decodeBody(r, &in); err != nil { + writeError(w, "", err) + return + } + if strings.TrimSpace(in.Kind) == "" { + writeError(w, "", fault.Invalid("archive kind is required")) + return + } + snapshot, err := archive.ExportRetention(r.Context(), t, in) + if err != nil { + writeError(w, "", err) + return + } + writeConsoleJSON(w, 200, map[string]any{"data": snapshot}) + }) +} + +// RetentionConsoleAuth keeps route registration independent from the concrete DB auth type. +type RetentionConsoleAuth interface { + ValidateSession(context.Context, auth.Token) error +} diff --git a/internal/transport/httpserver/retention_test.go b/internal/transport/httpserver/retention_test.go new file mode 100644 index 0000000..2e831ee --- /dev/null +++ b/internal/transport/httpserver/retention_test.go @@ -0,0 +1,78 @@ +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 fakeRetentionStore struct { + policy retention.Policy + gotKey string + updates int +} + +func (f *fakeRetentionStore) GetRetentionPolicy(_ context.Context, namespace core.ID) (retention.Policy, error) { + p := f.policy + p.NamespaceID = namespace + return p, nil +} + +func (f *fakeRetentionStore) UpdateRetentionPolicy(_ context.Context, token auth.Token, key string, in retention.UpdateInput) (retention.Policy, error) { + f.gotKey = key + f.updates++ + p, err := in.ToPolicy(token.NamespaceID) + if err != nil { + return retention.Policy{}, err + } + p.Revision = 1 + p.Source = "database" + f.policy = p + return p, nil +} + +func TestConsoleRetentionRequiresSessionAndIdempotencyKey(t *testing.T) { + ns, _ := core.NewID() + cred, _ := core.NewID() + raw, _, err := auth.Issue(ns, cred) + if err != nil { + t.Fatal(err) + } + authStore := &fakeConsoleAuth{valid: true} + retentionStore := &fakeRetentionStore{policy: retention.Defaults(ns)} + server := NewWithConsoleAuth("127.0.0.1:0", nil, authStore) + RegisterConsoleRetention(server, authStore, retentionStore, nil) + + unauthenticated := httptest.NewRequest(http.MethodGet, "/console/retention", nil) + unauthenticatedRes := httptest.NewRecorder() + server.Handler.ServeHTTP(unauthenticatedRes, unauthenticated) + if unauthenticatedRes.Code != http.StatusUnauthorized { + t.Fatalf("unauthenticated status=%d", unauthenticatedRes.Code) + } + + body := `{"expected_revision":"0","enabled":true,"idempotency_days":90,"task_history_days":365,"workflow_history_days":365,"event_log_days":365,"delivery_history_days":90,"inbox_history_days":90,"audit_log_days":2555,"reason":"test"}` + missingKey := httptest.NewRequest(http.MethodPut, "/console/retention", strings.NewReader(body)) + missingKey.Header.Set("Content-Type", "application/json") + missingKey.AddCookie(&http.Cookie{Name: consoleSessionCookie, Value: raw}) + missingKeyRes := httptest.NewRecorder() + server.Handler.ServeHTTP(missingKeyRes, missingKey) + if missingKeyRes.Code != http.StatusBadRequest { + t.Fatalf("missing key status=%d", missingKeyRes.Code) + } + + req := httptest.NewRequest(http.MethodPut, "/console/retention", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Idempotency-Key", "retention-http-1") + req.AddCookie(&http.Cookie{Name: consoleSessionCookie, Value: raw}) + res := httptest.NewRecorder() + server.Handler.ServeHTTP(res, req) + if res.Code != http.StatusOK || retentionStore.updates != 1 || retentionStore.gotKey != "retention-http-1" { + t.Fatalf("update status=%d updates=%d key=%q body=%s", res.Code, retentionStore.updates, retentionStore.gotKey, res.Body.String()) + } +} diff --git a/scripts/test-stability.sh b/scripts/test-stability.sh new file mode 100755 index 0000000..539942d --- /dev/null +++ b/scripts/test-stability.sh @@ -0,0 +1,19 @@ +#!/bin/sh +# P1 生产稳定性定向回归：真实 MySQL 锁等待/死锁、保留期磁盘增长， +# 以及两个独立 Orbit 进程之间的重连风暴和慢消费者。 +set -eu + +TASK_GO=${GO:-go} +if [ -z "${ORBIT_TEST_MYSQL_DSN:-}" ]; then + printf '%s\n' 'ORBIT_TEST_MYSQL_DSN must point to a disposable MySQL 8.4 database.' >&2 + exit 2 +fi + +"$TASK_GO" test -race -count=1 -run 'TestIntegration(WithTxRetriesPreCommitDeadlock|RetentionPolicyIsSharedAuditedAndIdempotent|RetentionDiskGrowthAndPreview|FairDispatchApplicationIsolation|WorkflowVersionAndRaces)$' ./internal/storage/mysql + +if [ -z "${ORBIT_TEST_SERVER_BINARY:-}" ] || [ -z "${ORBIT_TEST_WORKER_URL:-}" ] || [ -z "${ORBIT_TEST_WORKER_CREDENTIALS_FILE:-}" ]; then + printf '%s\n' 'ORBIT_TEST_SERVER_BINARY, ORBIT_TEST_WORKER_URL and ORBIT_TEST_WORKER_CREDENTIALS_FILE are required for the cross-node WebSocket checks.' >&2 + exit 2 +fi + +(cd sdk/go && "$TASK_GO" test -race -count=1 -run 'TestIntegrationEventProcess(SlowConsumerPeerProgress|CrossNodeReconnectStorm)$' -v .) diff --git a/sdk/go/event_process_integration_test.go b/sdk/go/event_process_integration_test.go index ec3d7ca..e50e92f 100644 --- a/sdk/go/event_process_integration_test.go +++ b/sdk/go/event_process_integration_test.go @@ -193,3 +193,87 @@ func TestIntegrationEventProcessSlowConsumerPeerProgress(t *testing.T) { } t.Log("node B acknowledged a second delivery while node A's capacity-one handler was blocked; both nodes preserved one attempt per delivery") } + +// TestIntegrationEventProcessCrossNodeReconnectStorm exercises the failure mode +// where a consumer loses several WebSocket sessions while the gateway alternates +// between two Orbit nodes. The durable delivery must be acknowledged once and +// the handler must never run again during the reconnect storm. +func TestIntegrationEventProcessCrossNodeReconnectStorm(t *testing.T) { + a := startOrbitProcess(t) + _, _, token := integrationWorkerClient(t) + apiA := processClient(t, a, token) + b := startOrbitProcess(t) + apiB := processClient(t, b, token) + ctx, stop := context.WithTimeout(context.Background(), 45*time.Second) + defer stop() + sub := integrationEventSubscription(t, ctx, apiA, 3) + _ = integrationCancelledEvent(t, ctx, apiA, sub) + gateway := newProtocolFaultGateway(t, a.url, "/v1/events/connect", eventSubprotocol) + client, err := NewClient(gateway.server.URL, Options{Token: token, HTTPClient: gateway.server.Client()}) + if err != nil { + t.Fatal(err) + } + var calls atomic.Int64 + started := make(chan struct{}) + release := make(chan struct{}) + consumer, err := NewEventConsumer(client, EventConsumerOptions{ + SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, + ReconnectMinDelay: 10 * time.Millisecond, ReconnectMaxDelay: 60 * time.Millisecond, + Handle: func(work context.Context, _ ResultEvent) error { + if calls.Add(1) == 1 { + close(started) + select { + case <-release: + case <-work.Done(): + return work.Err() + } + } + return nil + }, + }) + if err != nil { + t.Fatal(err) + } + run, cancel := context.WithCancel(ctx) + defer cancel() + done := make(chan error, 1) + go func() { done <- consumer.Run(run) }() + awaitConsumerEvent(t, consumer, "connected") + awaitConsumerEvent(t, consumer, "handler_started") + + // Close whichever node currently owns the socket, then move the gateway to + // the other node. Repeat enough times to cover a reconnect burst, while the + // event itself remains one durable delivery. + for i := 0; i < 4; i++ { + gateway.disconnectActive() + if i%2 == 0 { + gateway.target.Store(b.url) + } else { + gateway.target.Store(a.url) + } + awaitConsumerEvent(t, consumer, "connected") + } + close(release) + for consumer.Stats().Acknowledged < 1 { + select { + case <-ctx.Done(): + t.Fatal("cross-node consumer did not acknowledge delivery") + case err := <-done: + t.Fatalf("consumer stopped before acknowledgement: %v", err) + case <-time.After(20 * time.Millisecond): + } + } + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + stats := consumer.Stats() + if stats.Reconnects < 4 || stats.Acknowledged != 1 || calls.Load() != 1 { + t.Fatalf("cross-node reconnect storm duplicated work: stats=%+v calls=%d", stats, calls.Load()) + } + page, err := apiB.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked}) + if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 1 { + t.Fatalf("durable delivery attempt changed during storm: err=%v deliveries=%+v", err, page.Items) + } + t.Logf("cross-node reconnect storm completed with %d reconnects, one handler call, and one durable delivery attempt", stats.Reconnects) +}