diff --git a/cmd/orbit-admin/main.go b/cmd/orbit-admin/main.go index 2a5ed3b..8de07b3 100644 --- a/cmd/orbit-admin/main.go +++ b/cmd/orbit-admin/main.go @@ -23,17 +23,19 @@ func main() { } func run() error { if len(os.Args) < 2 { - return errors.New("usage: orbit-admin bootstrap|verify|credential status|rotate|revoke|apply") + return errors.New("usage: orbit-admin bootstrap|verify|retention inventory|credential status|rotate|revoke|apply") } switch os.Args[1] { case "bootstrap": return runBootstrap(os.Args[2:]) case "verify": return runVerify(os.Args[2:]) + case "retention": + return runRetention(os.Args[2:]) case "credential": return runCredential(os.Args[2:]) } - return errors.New("usage: orbit-admin bootstrap|verify|credential status|rotate|revoke|apply") + return errors.New("usage: orbit-admin bootstrap|verify|retention inventory|credential status|rotate|revoke|apply") } func runVerify(args []string) error { @@ -67,6 +69,43 @@ func runVerify(args []string) error { return nil } +func runRetention(args []string) error { + if len(args) == 0 || args[0] != "inventory" { + return errors.New("usage: orbit-admin retention inventory --before RFC3339") + } + flags := flag.NewFlagSet("retention inventory", flag.ContinueOnError) + var before string + flags.StringVar(&before, "before", "", "UTC cutoff; only records strictly before it are counted") + if err := flags.Parse(args[1:]); err != nil { + return err + } + if flags.NArg() != 0 || before == "" { + return errors.New("--before RFC3339 is required") + } + cutoff, err := time.Parse(time.RFC3339Nano, before) + if err != nil { + return errors.New("--before must be RFC3339") + } + cfg, err := config.Load() + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + db, err := mysql.Open(ctx, cfg.MySQLDSN, 2, 1) + if err != nil { + return errors.New("database unavailable") + } + defer db.Close() + inventory, err := mysql.ReadRetentionInventory(ctx, db, cutoff) + encoded, encodeErr := json.Marshal(inventory) + if encodeErr != nil { + return encodeErr + } + fmt.Println(string(encoded)) + return err +} + func runBootstrap(args []string) error { flags := flag.NewFlagSet("bootstrap", flag.ContinueOnError) var in mysql.BootstrapInput diff --git a/docs/04-reliability-and-data.md b/docs/04-reliability-and-data.md index 5dc2b7e..749e0e8 100644 --- a/docs/04-reliability-and-data.md +++ b/docs/04-reliability-and-data.md @@ -159,4 +159,6 @@ Orbit 数据库事务不能自动包含 接入方的业务事务。SDK 的事务 - 恢复旧备份后，可能已有外部业务效果而平台记录回退。必须停止派发、使恢复前的会话与租约失效，并完成必要对账后再开放执行。 - 数据库故障切换与备份恢复是不同故障模型，分别定义验证和可接受损失范围。 +当前可用的[`orbit-admin retention inventory`](api/retention.md)只提供按 UTC 截止时间的数量盘点，不执行归档或删除；保留周期、归档格式和受保护记录核对完成前，不得把盘点结果当作清理授权。 + MySQL 持久化建议基线为 `innodb_flush_log_at_trx_commit=1`，启用 binlog 时采用 `sync_binlog=1`。副本故障切换的确认语义、RPO / RTO 仍要由具体高可用方案决定。[MySQL 持久化配置](https://dev.mysql.com/doc/refman/8.4/en/innodb-parameters.html#sysvar_innodb_flush_log_at_trx_commit) diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 8bffd6e..93b6f4f 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -61,7 +61,7 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( 1. 已补[首批可复现故障工具](reports/phase-6-fault-boundaries.md)：A04“派发已提交但消息尚未送达”、W01/W13事务提交结果不明，以及W16最终回执/旧PENDING结果重放/下一轮竞争。每个切点必须有实际数据库或进程证据，不能用响应错误推断事务是否已提交，也不能用普通SIGNAL竞争替代最终查询场景。 2. 已实现[基础运行观测](api/observability.md)及[有界业务状态与事件采样](reports/phase-6-business-observability.md)：HTTP/WS、数据库连接池、后台批次、固定状态计数、最近事件类别、终态耗时样本、最早年龄、截断/新鲜度指标和可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签；发布前只读结构核验由[`orbit-admin verify`](api/operations.md)提供。仍需受理/授权阶段端到端延迟与全局持久吞吐、流程Inbox推进延迟、全局发送队列、多节点聚合及生产采集/通知验证。 3. 已补齐受保护管理环境中的固定凭据操作文件、轮换、撤销、幂等回执和脱敏管理审计；轮换保留授权绑定并要求新令牌重新建立会话，撤销允许在停用范围内执行且不改写历史记录。仍需补齐提交不明故障切点、多节点安全重绑演练、凭据最小权限配置核对和生产密钥托管/审计演练。 -4. 实现保留/归档与受保护记录核对，再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证，不缩减为本机多进程检查。 +4. 已补[`orbit-admin retention inventory`](api/retention.md)只读盘点和[阶段报告](reports/phase-6-retention-inventory.md)，能区分过期幂等、确定终态、未知效果、未终结流程、已确认/未确认交付、已消费/未消费Inbox、事件和审计规模；它不执行归档或删除。后续仍需评审各类保留周期、归档格式、受保护记录核对，再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证，不缩减为本机多进程检查。 ### P3 功能范围核对 diff --git a/docs/api/README.md b/docs/api/README.md index 6d777cd..216ffbc 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -21,6 +21,7 @@ | 任务 `priority` 与节点领取池 | P3 已实现有界优先级加速、应用/队列轮转 | [优先级与公平派发](priority-and-fairness.md) | | 凭据校验、初始化、固定操作文件轮换/撤销与回执重放 CLI | P1/P6 已实现；不提供公网管理路由，测试专用提交不明切点已回归 | [鉴权](authentication.md)、[凭据报告](../reports/phase-6-credentials.md) | | `orbit-admin verify` 发布前结构核验 | 已实现只读迁移、零外键和 000026–000028 固定索引检查 | [运维核验](operations.md) | +| `orbit-admin retention inventory` 保留清单 | 已实现带 UTC 截止时间的只读盘点；不执行归档、删除或权限变更 | [保留清单](retention.md) | | `/v1/worker/connect` | P2 已实现并通过阶段验收 | [Worker 契约](worker.md)、[Schema](worker-schema.md) | | 任务相对延迟 `delay_ms` | 已实现数据库计时、原键恢复与实际执行例子 | [任务接口](tasks.md)、[延迟示例](../examples/delayed-task.md) | | 任务冻结策略与 Worker `retryable` | P3 安全重试已实现 | [安全重试](retries.md) | diff --git a/docs/api/retention.md b/docs/api/retention.md new file mode 100644 index 0000000..e572730 --- /dev/null +++ b/docs/api/retention.md @@ -0,0 +1,19 @@ +# 保留清单（只读） + +`orbit-admin retention inventory` 用于在制定保留窗口、归档和备份恢复方案前盘点历史数据。它只执行带截止时间的统计查询，不删除、搬迁、改写数据，也不会自动判断某条记录可以安全归档。 + +```sh +ORBIT_MYSQL_DSN='user:password@tcp(127.0.0.1:3306)/orbit' \ + go run ./cmd/orbit-admin retention inventory \ + --before 2026-09-18T00:00:00Z +``` + +示例输出： + +```json +{"cutoff":"2026-09-18T00:00:00Z","expired_idempotency":0,"terminal_tasks":0,"uncertain_tasks":0,"active_tasks":0,"terminal_workflows":0,"active_workflows":0,"acknowledged_deliveries":0,"unconfirmed_deliveries":0,"processed_inbox":0,"unprocessed_inbox":0,"events":0,"audit_events":0} +``` + +所有时间按 UTC 解析，严格早于 `--before` 的记录才计入。截止时间不能晚于数据库当前时间，避免把未来记录误认为历史。`terminal_tasks` 只统计确定终态且 `effect_uncertain=false` 的任务；`uncertain_tasks`、`active_tasks`、`active_workflows`、`unconfirmed_deliveries` 和 `unprocessed_inbox` 是需要先处理或保留的阻塞项。 + +结果是盘点快照，不是归档授权。真正的清理或归档还必须完成保留周期评审、未确认副作用核对、事件投递和流程推进核对、备份恢复演练、权限隔离及审计；当前命令没有 `apply` 或删除路径。 diff --git a/docs/reports/phase-6-retention-inventory.md b/docs/reports/phase-6-retention-inventory.md new file mode 100644 index 0000000..b935856 --- /dev/null +++ b/docs/reports/phase-6-retention-inventory.md @@ -0,0 +1,25 @@ +# P6：保留清单与归档前置核对 + +日期：2026-09-19。本批先交付只读盘点，避免在保留周期、备份恢复和外部效果尚未评审时提供危险的自动删除入口。 + +## 已实现 + +- [`ReadRetentionInventory`](../../internal/storage/mysql/retention_inventory.go) 接受明确的 UTC 截止时间，只统计严格早于截止时间的记录。 +- `orbit-admin retention inventory --before RFC3339` 通过独立管理命令输出 JSON；数据库时间校验会拒绝未来截止时间，数据库连接、查询失败不会泄露 DSN 或 SQL。 +- 盘点固定覆盖幂等、确定终态/未知效果/未终结任务、确定终态/未终结流程、已确认/未确认交付、已消费/未消费流程 Inbox、追加事件和管理审计。 +- 任何查询只读，不执行 DDL、删除、搬迁、权限修改或幂等重置；CLI 没有 `apply` 参数。未确认交付、未消费 Inbox、未知效果和未终结实体被单独列为阻塞数量。 + +## 回归证据 + +- `go test ./cmd/orbit-admin ./internal/storage/mysql` 通过；存储集成用例使用真实 MySQL 验证：明确放到截止时间之前的确定终态任务计入 `terminal_tasks`，未知效果为零，未来截止被拒绝。 +- 标准 `DOCKER=... GO=... sh scripts/test-mysql.sh` 在 Go 1.27.1、Docker MySQL 8.4.10 环境退出码为 0。脚本实际输出： + + ```text + {"cutoff":"2000-01-01T00:00:00Z","expired_idempotency":0,"terminal_tasks":0,"uncertain_tasks":0,"active_tasks":0,"terminal_workflows":0,"active_workflows":0,"acknowledged_deliveries":0,"unconfirmed_deliveries":0,"processed_inbox":0,"unprocessed_inbox":0,"events":0,"audit_events":0} + ``` + + 同轮已有迁移、零外键、凭据、Pulse/Weave 真实进程强杀恢复、SDK 和两个独立业务示例均通过。 + +## 边界与后续 + +盘点数量不是归档授权，也不证明某类记录可独立删除。正式归档前仍需确定每类保留周期和归档格式，核对事件交付、流程推进、有效幂等身份、未知外部效果、审计保留与备份恢复；还需要隔离环境恢复演练、权限隔离、容量和生产拓扑验证。当前没有自动清理路径，防止把测试环境的历史数量误当成生产策略。 diff --git a/internal/storage/mysql/operational_metrics_integration_test.go b/internal/storage/mysql/operational_metrics_integration_test.go index 48ce9e3..89445f4 100644 --- a/internal/storage/mysql/operational_metrics_integration_test.go +++ b/internal/storage/mysql/operational_metrics_integration_test.go @@ -98,4 +98,15 @@ func TestIntegrationOperationalSnapshot(t *testing.T) { if _, err := ReadOperationalSnapshot(cancelled, db); err == nil { t.Fatal("cancelled context unexpectedly completed snapshot") } + f.exec("UPDATE pulse_tasks SET completed_at=DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 2 HOUR), updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?", f.resources.NamespaceID, task.ID) + inventory, err := ReadRetentionInventory(ctx, db, time.Now().UTC().Add(-time.Hour)) + if err != nil { + t.Fatal(err) + } + if inventory.Cutoff.IsZero() || inventory.TerminalTasks == 0 || inventory.UncertainTasks != 0 { + t.Fatalf("retention inventory=%+v", inventory) + } + if _, err := ReadRetentionInventory(ctx, db, time.Now().UTC().Add(time.Hour)); err == nil { + t.Fatal("future retention cutoff unexpectedly accepted") + } } diff --git a/internal/storage/mysql/retention_inventory.go b/internal/storage/mysql/retention_inventory.go new file mode 100644 index 0000000..20d5ade --- /dev/null +++ b/internal/storage/mysql/retention_inventory.go @@ -0,0 +1,83 @@ +package mysql + +import ( + "context" + "database/sql" + "errors" + "time" +) + +// RetentionInventory 是指定截止时间前的只读保留清单。 +// 它只提供核对数量，不代表记录已经满足归档条件，也不执行删除或搬迁。 +type RetentionInventory struct { + // Cutoff 是清单使用的 UTC 截止时间，严格早于该时间的记录才会计入。 + Cutoff time.Time `json:"cutoff"` + // ExpiredIdempotency 是已过幂等保留期的请求记录数量。 + ExpiredIdempotency uint64 `json:"expired_idempotency"` + // TerminalTasks 是已进入确定终态且已完成的任务数量。 + TerminalTasks uint64 `json:"terminal_tasks"` + // UncertainTasks 是仍带未知效果的任务数量，必须人工核对后才能处理。 + UncertainTasks uint64 `json:"uncertain_tasks"` + // ActiveTasks 是尚未终结的任务数量，不能进入普通历史归档。 + ActiveTasks uint64 `json:"active_tasks"` + // TerminalWorkflows 是已进入确定终态且已完成的流程数量。 + TerminalWorkflows uint64 `json:"terminal_workflows"` + // ActiveWorkflows 是尚未终结的流程数量，不能进入普通历史归档。 + ActiveWorkflows uint64 `json:"active_workflows"` + // AcknowledgedDeliveries 是已经确认的结果交付数量。 + AcknowledgedDeliveries uint64 `json:"acknowledged_deliveries"` + // UnconfirmedDeliveries 是仍未确认的结果交付数量，不能无声清理。 + UnconfirmedDeliveries uint64 `json:"unconfirmed_deliveries"` + // ProcessedInbox 是已经完成内部消费的流程事件数量。 + ProcessedInbox uint64 `json:"processed_inbox"` + // UnprocessedInbox 是尚未完成内部消费的流程事件数量，不能清理。 + UnprocessedInbox uint64 `json:"unprocessed_inbox"` + // Events 是截止时间前追加的不可变事件数量。 + Events uint64 `json:"events"` + // AuditEvents 是截止时间前追加的管理审计数量，默认应单独保留。 + AuditEvents uint64 `json:"audit_events"` +} + +// ReadRetentionInventory performs bounded-by-time, read-only count queries. +// It deliberately does not decide policy or mutate any table; callers must +// review retention windows, backups, outstanding effects and audit rules first. +func ReadRetentionInventory(ctx context.Context, db *sql.DB, cutoff time.Time) (RetentionInventory, error) { + if db == nil { + return RetentionInventory{}, errors.New("database is required") + } + if cutoff.IsZero() { + return RetentionInventory{}, errors.New("cutoff is required") + } + cutoff = cutoff.UTC() + var now time.Time + if err := db.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&now); err != nil { + return RetentionInventory{}, errors.New("database time unavailable") + } + if cutoff.After(now.UTC()) { + return RetentionInventory{}, errors.New("cutoff cannot be in the future") + } + result := RetentionInventory{Cutoff: cutoff} + queries := []struct { + destination *uint64 + query string + }{ + {&result.ExpiredIdempotency, "SELECT COUNT(*) FROM orbit_idempotency_keys WHERE expires_at'ACKED' AND created_at'PROCESSED' AND created_at "$TASK_TEST_DIR/server.log" 2>&1 & TASK_SERVER_PID=$!