diff --git a/.env.example b/.env.example index 0c17b70..4a9e305 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,12 @@ ORBIT_HTTP_ADDR=127.0.0.1:8080 ORBIT_MYSQL_DSN= ORBIT_DB_MAX_OPEN=20 ORBIT_DB_MAX_IDLE=5 +# 后台控制循环的有界调度间隔；默认 workflow/schedule=1s，recovery=2s。 +ORBIT_WORKFLOW_TICK_INTERVAL=1s +ORBIT_SCHEDULE_TICK_INTERVAL=1s +ORBIT_RECOVERY_TICK_INTERVAL=2s +# 每个后台批次最多处理的持久化候选，范围1–100。 +ORBIT_RUNTIME_BATCH_LIMIT=100 # 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/cmd/orbit/main.go b/cmd/orbit/main.go index 24d570c..d8cbd8a 100644 --- a/cmd/orbit/main.go +++ b/cmd/orbit/main.go @@ -108,7 +108,7 @@ func run() error { workflowDone := make(chan struct{}) go func() { defer close(workflowDone) - ticker := time.NewTicker(time.Second) + ticker := time.NewTicker(cfg.WorkflowTickInterval) defer ticker.Stop() for { select { @@ -117,7 +117,7 @@ func run() error { case <-ticker.C: started := time.Now() call, cancel := context.WithTimeout(workflowCtx, 5*time.Second) - _, err := workflows.Tick(call, 100) + _, err := workflows.Tick(call, cfg.RuntimeBatchLimit) cancel() if metrics != nil { metrics.RecordBatch("workflow", started, err) @@ -133,7 +133,7 @@ func run() error { scheduleDone := make(chan struct{}) go func() { defer close(scheduleDone) - ticker := time.NewTicker(time.Second) + ticker := time.NewTicker(cfg.ScheduleTickInterval) defer ticker.Stop() for { select { @@ -142,7 +142,7 @@ func run() error { case <-ticker.C: started := time.Now() call, cancel := context.WithTimeout(scheduleCtx, 5*time.Second) - _, err := schedules.Tick(call, 100) + _, err := schedules.Tick(call, cfg.RuntimeBatchLimit) cancel() if metrics != nil { metrics.RecordBatch("schedule", started, err) @@ -163,7 +163,7 @@ func run() error { recoveryCtx, stopRecovery := context.WithCancel(root) go func() { defer close(recoveryDone) - ticker := time.NewTicker(2 * time.Second) + ticker := time.NewTicker(cfg.RecoveryTickInterval) defer ticker.Stop() for { select { @@ -172,7 +172,7 @@ func run() error { case <-ticker.C: started := time.Now() call, cancel := context.WithTimeout(recoveryCtx, 5*time.Second) - _, err := executions.Recover(call, 100) + _, err := executions.Recover(call, cfg.RuntimeBatchLimit) cancel() if metrics != nil { metrics.RecordBatch("execution_recovery", started, err) @@ -243,7 +243,7 @@ func run() error { slog.Info("Orbit metrics server starting", "metrics_address", metricsListener.Addr().String()) } go func() { done <- server.Serve(listener) }() - slog.Info("Orbit HTTP server starting", "address", listener.Addr().String(), "scope", "health, task, queue, schedule, subscription, workflow and Worker APIs", "node_id", node) + slog.Info("Orbit HTTP server starting", "address", listener.Addr().String(), "scope", "health, task, queue, schedule, subscription, workflow and Worker APIs", "node_id", node, "workflow_tick_interval", cfg.WorkflowTickInterval.String(), "schedule_tick_interval", cfg.ScheduleTickInterval.String(), "recovery_tick_interval", cfg.RecoveryTickInterval.String(), "runtime_batch_limit", cfg.RuntimeBatchLimit) select { case err := <-done: if errors.Is(err, http.ErrServerClosed) { diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 70f0c11..7157b28 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -59,7 +59,7 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( ### P6 首批实施顺序 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、数据库连接池、后台批次、固定任务/交付/流程/Inbox状态计数、最近事件类别、终态耗时样本、固定事件阶段耗时、最早年龄、截断/新鲜度指标和可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签；发布前只读结构核验由[`orbit-admin verify`](api/operations.md)提供，运维人员也可用 `orbit-admin observe snapshot` 获取同一有界单节点快照。仍需端到端外部效果延迟、全局持久吞吐、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状态计数、最近事件类别、终态耗时样本、固定事件阶段耗时、最早年龄、截断/新鲜度指标和可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签；发布前只读结构核验由[`orbit-admin verify`](api/operations.md)提供，运维人员也可用 `orbit-admin observe snapshot` 获取同一有界单节点快照；后台循环间隔和批量上限已支持有界环境配置并在启动日志中记录。仍需端到端外部效果延迟、全局持久吞吐、Inbox从事件产生到推进完成的精确阶段延迟、全局发送队列、多节点聚合及生产采集/通知验证；SDK受控重连风暴和真实慢消费者背压已补[边界回归](reports/phase-6-reconnect-storm.md)与[慢消费者报告](reports/phase-6-slow-consumer.md)，跨节点重连仍待目标拓扑。 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继续按实际拓扑验证，不缩减为本机多进程检查。 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)已补边界回归；下一步补指标采集对照、长期存储增长、跨节点重连与目标数据库拓扑容量。 diff --git a/docs/20-operations-runbook.md b/docs/20-operations-runbook.md index 576cb32..5f445ce 100644 --- a/docs/20-operations-runbook.md +++ b/docs/20-operations-runbook.md @@ -37,6 +37,8 @@ curl -fsS http://127.0.0.1:8080/health/ready `live` 只表示进程仍在运行；`ready` 同时核对 MySQL、迁移账本和零外键要求。收到 `503` 时停止向该节点发送新的受理、执行授权和流程推进请求，保留原幂等键并等待核对，不把网络错误当作业务失败。 +后台调度默认每秒检查 Weave 和计划、每两秒执行一次执行恢复，每批最多处理100条持久化候选。容量验证后可以通过 `ORBIT_WORKFLOW_TICK_INTERVAL`、`ORBIT_SCHEDULE_TICK_INTERVAL`、`ORBIT_RECOVERY_TICK_INTERVAL` 和 `ORBIT_RUNTIME_BATCH_LIMIT` 调整；间隔只允许 `100ms` 到 `1m`，批量只允许 `1` 到 `100`。调优必须和数据库锁等待、连接池、积压及恢复窗口一起记录，不能只提高批量或缩短间隔来掩盖容量不足。 + ## 3. 滚动停止与替换 Orbit 接收到 `SIGTERM` 后停止新的 Worker/Event 连接，取消后台循环，关闭 HTTP/WS 监听并在有限时间内等待已有连接退出。发布脚本应按以下顺序操作： diff --git a/docs/21-container-deployment.md b/docs/21-container-deployment.md index 0c847eb..5033d98 100644 --- a/docs/21-container-deployment.md +++ b/docs/21-container-deployment.md @@ -49,6 +49,10 @@ docker run --rm \ docker run --rm \ --env ORBIT_MYSQL_DSN \ --env ORBIT_HTTP_ADDR=0.0.0.0:8080 \ + --env ORBIT_RUNTIME_BATCH_LIMIT=100 \ + --env ORBIT_WORKFLOW_TICK_INTERVAL=1s \ + --env ORBIT_SCHEDULE_TICK_INTERVAL=1s \ + --env ORBIT_RECOVERY_TICK_INTERVAL=2s \ --env ORBIT_METRICS_ADDR=127.0.0.1:9090 \ --publish 8080:8080 \ registry.example/orbit:dev diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index fbee229..4338f81 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -7,6 +7,7 @@ | 检查 | 环境/命令 | 结果 | | --- | --- | --- | | 服务端单元与竞态、vet、构建 | Docker `golang:1.27.1`，仓库声明版本；`GOWORK=off go test -race ./... && go vet ./... && go build ./...` | PASS | +| 后台运行参数边界 | `internal/config.TestRuntimeDefaultsAndBoundedOverrides`；默认值与四个 `ORBIT_*` 覆盖项 | PASS；循环间隔限定100ms–1m，批量限定1–100，服务启动日志记录生效值 | | SDK 与独立示例竞态、vet、构建 | Docker `golang:1.27.1`；SDK 全 module、全部示例、`orbit-load`、`orbit-exec-load`、`orbit-schedule-load`、`orbit-workflow-load`，以及 result-inbox/task-outbox | PASS；健康/版本 SDK 契约包含 `Client.Version(ctx)`，返回协议身份且不暴露敏感响应 | | 容器发布物构建 | `make image`；宿主平台 linux/arm64；显式 `docker build --platform linux/amd64`；带 `VERSION`、`REVISION`、`BUILD_TIME` 构建参数的版本化镜像；镜像内 `GOWORK=off` | PASS；native 与 amd64 镜像均构建成功，包含 `orbit`、`orbit-migrate`、`orbit-admin`，运行用户为非 root；版本化 amd64 二进制和 OCI 标签均包含注入的构建身份，`GET /version` 契约单测通过；无 DSN 时服务按预期拒绝启动 | | 当前协议门禁 | `make protocol-check` | PASS，当前 v1 契约同步 | diff --git a/internal/config/config.go b/internal/config/config.go index c346f1d..2ed38c7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,19 +6,34 @@ import ( "net" "os" "strconv" + "time" ) type Config struct { - HTTPAddr string - MySQLDSN string - MaxOpen int - MaxIdle int - MetricsAddr string + HTTPAddr string + MySQLDSN string + MaxOpen int + MaxIdle int + MetricsAddr string + WorkflowTickInterval time.Duration // WorkflowTickInterval Weave 持久化推进扫描间隔，范围100ms–1m。 + ScheduleTickInterval time.Duration // ScheduleTickInterval 计划扫描间隔，范围100ms–1m。 + RecoveryTickInterval time.Duration // RecoveryTickInterval 执行租约恢复扫描间隔，范围100ms–1m。 + RuntimeBatchLimit int // RuntimeBatchLimit 每个后台批次的候选上限，范围1–100。 } func Load() (Config, error) { return FromEnv(os.Getenv) } func FromEnv(get func(string) string) (Config, error) { - c := Config{HTTPAddr: get("ORBIT_HTTP_ADDR"), MySQLDSN: get("ORBIT_MYSQL_DSN"), MaxOpen: 20, MaxIdle: 5, MetricsAddr: get("ORBIT_METRICS_ADDR")} + c := Config{ + HTTPAddr: get("ORBIT_HTTP_ADDR"), + MySQLDSN: get("ORBIT_MYSQL_DSN"), + MaxOpen: 20, + MaxIdle: 5, + MetricsAddr: get("ORBIT_METRICS_ADDR"), + WorkflowTickInterval: time.Second, + ScheduleTickInterval: time.Second, + RecoveryTickInterval: 2 * time.Second, + RuntimeBatchLimit: 100, + } if c.HTTPAddr == "" { c.HTTPAddr = "127.0.0.1:8080" } @@ -43,6 +58,25 @@ func FromEnv(get func(string) string) (Config, error) { if c.MaxOpen <= 0 || c.MaxIdle > c.MaxOpen { return Config{}, errors.New("database pool requires 0 <= max_idle <= max_open and max_open > 0") } + for _, field := range []struct { + name string + dst *time.Duration + }{{"ORBIT_WORKFLOW_TICK_INTERVAL", &c.WorkflowTickInterval}, {"ORBIT_SCHEDULE_TICK_INTERVAL", &c.ScheduleTickInterval}, {"ORBIT_RECOVERY_TICK_INTERVAL", &c.RecoveryTickInterval}} { + if raw := get(field.name); raw != "" { + d, err := time.ParseDuration(raw) + if err != nil || d < 100*time.Millisecond || d > time.Minute { + return Config{}, fmt.Errorf("%s must be a duration from 100ms to 1m", field.name) + } + *field.dst = d + } + } + if raw := get("ORBIT_RUNTIME_BATCH_LIMIT"); raw != "" { + n, err := strconv.Atoi(raw) + if err != nil || n < 1 || n > 100 { + return Config{}, errors.New("ORBIT_RUNTIME_BATCH_LIMIT must be an integer from 1 to 100") + } + c.RuntimeBatchLimit = n + } if c.MetricsAddr != "" { _, port, err := net.SplitHostPort(c.MetricsAddr) n, parseErr := strconv.Atoi(port) diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 41f8598..f8e3863 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1,6 +1,38 @@ package config -import "testing" +import ( + "testing" + "time" +) + +func TestRuntimeDefaultsAndBoundedOverrides(t *testing.T) { + base := map[string]string{"ORBIT_MYSQL_DSN": "private-dsn"} + get := func(k string) string { return base[k] } + c, err := FromEnv(get) + if err != nil || c.WorkflowTickInterval != time.Second || c.ScheduleTickInterval != time.Second || c.RecoveryTickInterval != 2*time.Second || c.RuntimeBatchLimit != 100 { + t.Fatalf("unexpected runtime defaults: %+v err=%v", c, err) + } + base["ORBIT_WORKFLOW_TICK_INTERVAL"] = "250ms" + base["ORBIT_SCHEDULE_TICK_INTERVAL"] = "1500ms" + base["ORBIT_RECOVERY_TICK_INTERVAL"] = "1m" + base["ORBIT_RUNTIME_BATCH_LIMIT"] = "37" + c, err = FromEnv(get) + if err != nil || c.WorkflowTickInterval != 250*time.Millisecond || c.ScheduleTickInterval != 1500*time.Millisecond || c.RecoveryTickInterval != time.Minute || c.RuntimeBatchLimit != 37 { + t.Fatalf("valid runtime override rejected: %+v err=%v", c, err) + } + for key, value := range map[string]string{ + "ORBIT_WORKFLOW_TICK_INTERVAL": "99ms", + "ORBIT_SCHEDULE_TICK_INTERVAL": "2m", + "ORBIT_RECOVERY_TICK_INTERVAL": "not-a-duration", + "ORBIT_RUNTIME_BATCH_LIMIT": "101", + } { + base[key] = value + if _, err := FromEnv(get); err == nil { + t.Fatalf("invalid runtime override accepted: %s=%s", key, value) + } + delete(base, key) + } +} func TestMetricsAddressIsExplicitAndValidated(t *testing.T) { base := map[string]string{"ORBIT_MYSQL_DSN": "private-dsn"}