diff --git a/AGENTS.md b/AGENTS.md index e653e8e..af3be1e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -85,3 +85,4 @@ - HTTP成功升级在Hijack完成时记录握手，WS处理器存续计入独立连接指标；保持1xx临时响应、Flush、Hijack与原panic行为。DB池取本地Stats，抓取不查询业务库；后台批次success不是业务任务成功量。观测丢样或失败不得改变权威事务及恢复决定。 - 进程指标、固定状态采样、格式验证、告警规则和标准回归证据见docs/reports/phase-6-observability.md及docs/reports/phase-6-business-observability.md。仍须补受理/授权/完成与流程推进延迟、持久吞吐/积压的全局聚合、重试/核查/恢复量、发送队列、实际采集告警及负载开销，不能把进程观测通过写成P6已完成。普通清理仍须保护有效幂等、未确认事件、未知结果和未终结流程。 - 凭据轮换保留credential_id及订阅/Worker授权绑定，在同一事务替换摘要、固定到期时间、幂等回执和管理审计；新令牌必须重新建立会话，认证路径每次重新核对凭据状态。撤销必须让认证失效并保留原任务、订阅、交付、会话和审计证据，允许在应用或命名空间停用后执行；重复应用同一固定操作幂等返回。管理CLI先把固定操作文件写成0600，再提交；提交不明时保留文件并用原文件重放，不能重新生成秘密。日志、审计摘要、指标和业务表不得保存秘密原文或摘要。不要把轮换同一凭据ID解释成历史令牌仍有效，也不得级联删除订阅、任务或事件。 +- `sdk/go/cmd/orbit-load` 只用于隔离环境受理测量：每项使用稳定独立幂等键，固定请求形状和并发上限，按请求超时，不自动重试，不启动Worker或执行外部副作用；输出延迟百分位与有限错误分类，失败必须以非零状态返回。其结果不能替代真实Worker、流程推进、数据库拓扑或生产容量验收。 diff --git a/README.md b/README.md index 303d9cb..cf5aa13 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,8 @@ make integration `make integration` 要求 Docker、OpenSSL 等标准命令，启动独立的临时 MySQL 容器，使用随机凭据和回环随机端口；完成后删除该测试容器。不会连接已有业务数据库。可通过 `GO`、`DOCKER` 和 `ORBIT_TEST_MYSQL_IMAGE` 指定工具/测试镜像。 +受理层负载采样工具见 [有界受理负载工具](docs/api/load.md)，只允许在隔离命名空间或一次性数据库中运行；它不自动重试、不启动 Worker，也不提供生产容量承诺。 + 修改已应用迁移时会触发校验和拒绝；新增结构使用新的前向迁移。初始结构尚未部署生产。 ## 文档入口 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index f5707f9..5e1536a 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -62,6 +62,7 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( 2. 已实现[基础运行观测](api/observability.md)及[有界业务状态与事件采样](reports/phase-6-business-observability.md)：HTTP/WS、数据库连接池、后台批次、固定任务/交付/流程/Inbox状态计数、最近事件类别、终态耗时样本、固定事件阶段耗时、最早年龄、截断/新鲜度指标和可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签；发布前只读结构核验由[`orbit-admin verify`](api/operations.md)提供。仍需端到端外部效果延迟、全局持久吞吐、Inbox从事件产生到推进完成的精确阶段延迟、全局发送队列、多节点聚合及生产采集/通知验证。 3. 已补齐受保护管理环境中的固定凭据操作文件、轮换、撤销、幂等回执和脱敏管理审计；轮换保留授权绑定并要求新令牌重新建立会话，撤销允许在停用范围内执行且不改写历史记录。仍需补齐提交不明故障切点、多节点安全重绑演练、凭据最小权限配置核对和生产密钥托管/审计演练。 4. 已补[`orbit-admin retention inventory`](api/retention.md)只读盘点和[阶段报告](reports/phase-6-retention-inventory.md)，能区分过期幂等、确定终态、未知效果、未终结流程、已确认/未确认交付、已消费/未消费Inbox、事件和审计规模；它不执行归档或删除。后续仍需评审各类保留周期、归档格式、受保护记录核对，再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证，不缩减为本机多进程检查。 +5. 已补[`orbit-load`](api/load.md)有界受理负载工具：固定请求形状、并发上限、每请求超时、无自动重试、延迟百分位和有限错误分类；已完成一次隔离 MySQL + HTTP 小样本实测，证据见[负载报告](reports/phase-6-load.md)。标准 Go 1.27.1 隔离回归仍需在工具链恢复后复跑；该工具只提供受理层测量，真实 Worker/流程/数据库拓扑容量仍需独立报告。 ### P3 功能范围核对 diff --git a/docs/api/README.md b/docs/api/README.md index 216ffbc..eae2a12 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -22,6 +22,7 @@ | 凭据校验、初始化、固定操作文件轮换/撤销与回执重放 CLI | P1/P6 已实现；不提供公网管理路由，测试专用提交不明切点已回归 | [鉴权](authentication.md)、[凭据报告](../reports/phase-6-credentials.md) | | `orbit-admin verify` 发布前结构核验 | 已实现只读迁移、零外键和 000026–000028 固定索引检查 | [运维核验](operations.md) | | `orbit-admin retention inventory` 保留清单 | 已实现带 UTC 截止时间的只读盘点；不执行归档、删除或权限变更 | [保留清单](retention.md) | +| `sdk/go/cmd/orbit-load` 有界受理负载工具 | 已实现固定并发、无自动重试、延迟百分位和错误分类；只用于隔离环境 | [负载工具](load.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/load.md b/docs/api/load.md new file mode 100644 index 0000000..119f824 --- /dev/null +++ b/docs/api/load.md @@ -0,0 +1,35 @@ +# 有界受理负载工具 + +`orbit-load` 是独立 Go SDK module 中的受理负载工具，用于在隔离环境测量 HTTP 受理请求的延迟分布。它只提交带独立幂等键的任务，不启动 Worker、不执行业务副作用、不自动重试，也不把测量结果解释为生产吞吐或租户容量保证。 + +## 运行 + +先为专用测试命名空间准备凭据，再运行： + +```sh +cd sdk/go +ORBIT_URL='http://127.0.0.1:8080' \ +ORBIT_CREDENTIALS_FILE='/private/orbit-load.credentials.json' \ +go run ./cmd/orbit-load \ + -requests 1000 \ + -concurrency 32 \ + -timeout 10s +``` + +也可以直接传 `-url`、`-token` 和 `-queue`。命令拒绝非 32 位小写十六进制队列 ID，默认上限是 100000 次请求和 1024 个并发请求；并发数超过请求数时自动降到请求数。每次请求的超时独立计算，超时或 HTTP 错误不会再次提交。 + +每个任务使用形如 `orbit-load-{run_id}-{sequence}` 的幂等键，输入只包含本次运行标识和序号，处理器默认为 `load.noop`，任务尝试预算固定为 1。命令不会删除生成的任务；测试必须使用一次性数据库或专用命名空间，结束后按保留策略处理，不能直接对生产命名空间运行。 + +## 输出 + +标准输出是一行 JSON： + +```json +{"started_at":"2026-09-19T00:00:00Z","finished_at":"2026-09-19T00:00:02Z","requests":100,"concurrency":8,"success":100,"errors":0,"p50_ms":11.2,"p95_ms":24.7,"p99_ms":31.4,"max_ms":35.1} +``` + +`p50_ms`、`p95_ms`、`p99_ms` 和 `max_ms` 统计所有请求（包括失败请求）的端到端 HTTP 交换耗时，百分位采用最近秩；`error_codes` 只保留有限错误分类，例如 `TIMEOUT`、`TRANSPORT`、`HTTP_409` 或服务端稳定错误码，不输出令牌、请求体或内部错误文本。存在失败请求时仍输出报告，并以非零状态退出。 + +## 解释边界 + +这项工具固定的是客户端请求数、并发和消息形状，不能替代以下验证：真实 Worker 执行吞吐、定时触发、流程推进、数据库锁等待、复制延迟、磁盘与网络 I/O、故障恢复、跨主机拓扑或长时间积压。正式容量报告必须同时记录硬件、MySQL 拓扑、连接池、数据量、任务时长、结果大小、故障注入和积压清空时间，并将本工具结果与其他阶段指标分开。 diff --git a/docs/reports/phase-6-load.md b/docs/reports/phase-6-load.md new file mode 100644 index 0000000..d5fad8e --- /dev/null +++ b/docs/reports/phase-6-load.md @@ -0,0 +1,28 @@ +# P6：受理层有界负载工具 + +验证日期：2026-09-19。本批新增 `sdk/go/cmd/orbit-load`，用于隔离环境测量任务受理 HTTP 交换耗时。工具固定请求并发上限和每请求超时，使用独立幂等键，不自动重试、不启动 Worker、不执行外部副作用；失败仍输出报告并以非零状态退出。它不构成生产容量、可用性或百万租户承诺。 + +## 实现边界 + +- 请求数限制为 1–100000，并发限制为 1–1024；每次请求最多一个 HTTP 提交。 +- 默认处理器为 `load.noop`，输入只包含随机运行标识和序号，任务尝试预算为 1，避免把测试逻辑误当业务处理器。 +- 输出成功数、错误分类、p50/p95/p99/最大耗时；百分位采用最近秩，错误分类不包含令牌、请求体或内部错误文本。 +- 工具不清理生成任务，运行前必须使用一次性数据库或专用命名空间；正式容量测试还需配套 Worker、流程、数据库锁等待、积压恢复和拓扑故障测量。 + +## 隔离实测 + +环境：macOS arm64 主机，Docker MySQL 8.4，已有 Orbit 服务二进制，负载命令使用 Go 1.25 兼容 modfile 交叉编译为 darwin/arm64；数据库和任务容器在测试结束后销毁。该次只验证工具与受理链路，不能替代项目要求的 Go 1.27.1 标准全量回归。 + +命令参数：`requests=24`、`concurrency=4`、`timeout=5s`，独立命名空间和队列，未启动 Worker。 + +实际输出： + +```json +{"started_at":"2026-09-19T15:39:43.724491Z","finished_at":"2026-09-19T15:39:43.953625Z","requests":24,"concurrency":4,"success":24,"errors":0,"p50_ms":32.179459,"p95_ms":61.578333,"p99_ms":70.748416,"max_ms":70.748416} +``` + +结果证明 24 次独立 HTTP 受理均成功且没有错误分类；延迟数值只属于本次主机、连接、数据量和请求形状，不能外推到生产。Go 1.27.1 工具链恢复后，标准 `scripts/test-mysql.sh` 会再次执行相同负载样本并把输出纳入全量日志。 + +## 未完成的生产验收 + +仍需在固定硬件和目标拓扑中补充短/中/长任务混合、Worker 连接与重连、定时集中触发、流程推进、慢消费者、长期积压、数据库锁等待、复制延迟、故障切换、备份恢复及容量余量测试；这些结果必须与本工具的受理层样本分开记录。 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 98d19a6..6450f0c 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -43,10 +43,12 @@ ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" "$TASK_GO" test -race -count=1 - ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" "$TASK_GO" test -race -count=1 -run '^TestIntegrationWorkflowRuntimeProcess$' -v ./internal/storage/mysql "$TASK_GO" build -o "$TASK_TEST_DIR/orbit-migrate" ./cmd/orbit-migrate "$TASK_GO" build -o "$TASK_TEST_DIR/orbit-admin" ./cmd/orbit-admin +(cd sdk/go && GOWORK=off "$TASK_GO" build -o "$TASK_TEST_DIR/orbit-load" ./cmd/orbit-load) ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-migrate" status ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" verify ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" retention inventory --before 2000-01-01T00:00:00Z ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap --application sdk-example --namespace default --environment test --queue files --output "$TASK_TEST_DIR/sdk.credentials.json" +ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap --application sdk-load --namespace default --environment test --queue load --output "$TASK_TEST_DIR/load.credentials.json" ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" ORBIT_HTTP_ADDR=127.0.0.1:0 "$TASK_TEST_DIR/orbit" > "$TASK_TEST_DIR/server.log" 2>&1 & TASK_SERVER_PID=$! TASK_HTTP_PORT= @@ -58,6 +60,7 @@ for task_attempt in $(seq 1 20); do done if [ -z "$TASK_HTTP_PORT" ]; then printf '%s\n' 'Bootstrap HTTP server did not start.' >&2; exit 1; fi (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" "$TASK_GO" run ./examples/health) +(cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/load.credentials.json" "$TASK_TEST_DIR/orbit-load" -requests 24 -concurrency 4 -timeout 5s) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/sdk.credentials.json" "$TASK_GO" run ./examples/tasks) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/sdk.credentials.json" "$TASK_GO" run ./examples/task-query) ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap --application sdk-worker --namespace default --environment test --queue worker-examples --worker-access --schedule-access --event-access --output "$TASK_TEST_DIR/worker.credentials.json" diff --git a/sdk/go/cmd/orbit-load/main.go b/sdk/go/cmd/orbit-load/main.go new file mode 100644 index 0000000..845e9a6 --- /dev/null +++ b/sdk/go/cmd/orbit-load/main.go @@ -0,0 +1,300 @@ +// Command orbit-load runs a bounded, non-retrying task admission sample. +// It is a measurement tool for an isolated environment, not a production +// scheduler and not a capacity promise. +package main + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "flag" + "fmt" + "math" + "net/http" + "os" + "sort" + "strings" + "sync" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +type credentials struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` +} + +type sampleConfig struct { + URL string + Token string + QueueID string + Requests int + Concurrency int + Timeout time.Duration + Handler string + HandlerVersion string + PayloadVersion string +} + +type sampleResult struct { + duration time.Duration + code string +} + +type loadReport struct { + StartedAt time.Time `json:"started_at"` + FinishedAt time.Time `json:"finished_at"` + Requests int `json:"requests"` + Concurrency int `json:"concurrency"` + Success int `json:"success"` + Errors int `json:"errors"` + P50MS float64 `json:"p50_ms"` + P95MS float64 `json:"p95_ms"` + P99MS float64 `json:"p99_ms"` + MaxMS float64 `json:"max_ms"` + ErrorCodes map[string]int `json:"error_codes,omitempty"` +} + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run() error { + cfg, err := parseConfig(os.Args[1:]) + if err != nil { + return err + } + client, err := orbit.NewClient(cfg.URL, orbit.Options{ + HTTPClient: &http.Client{Timeout: cfg.Timeout}, + Token: cfg.Token, + }) + if err != nil { + return err + } + runID, err := newRunID() + if err != nil { + return err + } + started := time.Now().UTC() + results := runSamples(client, cfg, runID) + finished := time.Now().UTC() + report := makeReport(cfg, started, finished, results) + if err := json.NewEncoder(os.Stdout).Encode(report); err != nil { + return err + } + if report.Errors != 0 { + return errors.New("load sample had failed requests; inspect error_codes and rerun in an isolated environment") + } + return nil +} + +func parseConfig(args []string) (sampleConfig, error) { + flags := flag.NewFlagSet("orbit-load", flag.ContinueOnError) + flags.SetOutput(os.Stderr) + cfg := sampleConfig{ + URL: os.Getenv("ORBIT_URL"), + Token: os.Getenv("ORBIT_TOKEN"), + QueueID: os.Getenv("ORBIT_QUEUE_ID"), + Requests: 100, + Concurrency: 8, + Timeout: 10 * time.Second, + Handler: "load.noop", + HandlerVersion: "v1", + PayloadVersion: "v1", + } + credentialsPath := os.Getenv("ORBIT_CREDENTIALS_FILE") + flags.StringVar(&cfg.URL, "url", cfg.URL, "Orbit HTTP URL") + flags.StringVar(&cfg.Token, "token", cfg.Token, "Bearer token; prefer ORBIT_CREDENTIALS_FILE") + flags.StringVar(&cfg.QueueID, "queue", cfg.QueueID, "target queue ID") + flags.StringVar(&credentialsPath, "credentials", credentialsPath, "private bootstrap credentials JSON file") + flags.IntVar(&cfg.Requests, "requests", cfg.Requests, "number of independent admissions, 1..100000") + flags.IntVar(&cfg.Concurrency, "concurrency", cfg.Concurrency, "maximum concurrent HTTP requests, 1..1024") + flags.DurationVar(&cfg.Timeout, "timeout", cfg.Timeout, "per-request HTTP timeout") + flags.StringVar(&cfg.Handler, "handler", cfg.Handler, "handler name frozen in each task") + flags.StringVar(&cfg.HandlerVersion, "handler-version", cfg.HandlerVersion, "handler version frozen in each task") + flags.StringVar(&cfg.PayloadVersion, "payload-version", cfg.PayloadVersion, "payload contract version frozen in each task") + if err := flags.Parse(args); err != nil { + return sampleConfig{}, err + } + if flags.NArg() != 0 { + return sampleConfig{}, errors.New("orbit-load does not accept positional arguments") + } + if credentialsPath != "" { + file, err := os.ReadFile(credentialsPath) + if err != nil { + return sampleConfig{}, fmt.Errorf("read credentials: %w", err) + } + if len(file) > 16<<10 { + return sampleConfig{}, errors.New("credentials file exceeds 16 KiB") + } + var saved credentials + if err := json.Unmarshal(file, &saved); err != nil { + return sampleConfig{}, errors.New("invalid credentials file") + } + if cfg.Token == "" { + cfg.Token = saved.Token + } + if cfg.QueueID == "" { + cfg.QueueID = saved.QueueID + } + } + if cfg.URL == "" { + return sampleConfig{}, errors.New("-url or ORBIT_URL is required") + } + if cfg.Token == "" { + return sampleConfig{}, errors.New("-token, ORBIT_TOKEN or ORBIT_CREDENTIALS_FILE is required") + } + if !validID(cfg.QueueID) { + return sampleConfig{}, errors.New("-queue or ORBIT_QUEUE_ID must be a 32-character lowercase hexadecimal ID") + } + if cfg.Requests < 1 || cfg.Requests > 100000 { + return sampleConfig{}, errors.New("-requests must be between 1 and 100000") + } + if cfg.Concurrency < 1 || cfg.Concurrency > 1024 { + return sampleConfig{}, errors.New("-concurrency must be between 1 and 1024") + } + if cfg.Timeout <= 0 || cfg.Timeout > time.Hour { + return sampleConfig{}, errors.New("-timeout must be greater than zero and no more than one hour") + } + for name, value := range map[string]string{ + "handler": cfg.Handler, + "handler-version": cfg.HandlerVersion, + "payload-version": cfg.PayloadVersion, + } { + if strings.TrimSpace(value) == "" || strings.ContainsAny(value, "\r\n") { + return sampleConfig{}, fmt.Errorf("-%s must be a non-empty single line", name) + } + } + if cfg.Concurrency > cfg.Requests { + cfg.Concurrency = cfg.Requests + } + return cfg, nil +} + +func runSamples(client *orbit.Client, cfg sampleConfig, runID string) []sampleResult { + jobs := make(chan int) + results := make(chan sampleResult, cfg.Requests) + var wait sync.WaitGroup + for worker := 0; worker < cfg.Concurrency; worker++ { + wait.Add(1) + go func() { + defer wait.Done() + for sequence := range jobs { + start := time.Now() + requestCtx, cancel := context.WithTimeout(context.Background(), cfg.Timeout) + payload, _ := json.Marshal(struct { + RunID string `json:"run_id"` + Sequence int `json:"sequence"` + }{RunID: runID, Sequence: sequence}) + _, err := client.SubmitTask(requestCtx, fmt.Sprintf("orbit-load-%s-%06d", runID, sequence), orbit.SubmitTaskRequest{ + QueueID: cfg.QueueID, + Handler: cfg.Handler, + HandlerVersion: cfg.HandlerVersion, + PayloadVersion: cfg.PayloadVersion, + Payload: payload, + TaskTimeoutMS: 60 * 60 * 1000, + RecoverySafety: orbit.RecoveryReconcileOnly, + Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 10 * 1000}, + }) + cancel() + code := "" + if err != nil { + code = errorCode(err) + } + results <- sampleResult{duration: time.Since(start), code: code} + } + }() + } + for sequence := 0; sequence < cfg.Requests; sequence++ { + jobs <- sequence + } + close(jobs) + wait.Wait() + close(results) + out := make([]sampleResult, 0, cfg.Requests) + for result := range results { + out = append(out, result) + } + return out +} + +func makeReport(cfg sampleConfig, started, finished time.Time, results []sampleResult) loadReport { + durations := make([]time.Duration, 0, len(results)) + errorCodes := make(map[string]int) + success := 0 + for _, result := range results { + durations = append(durations, result.duration) + if result.code == "" { + success++ + } else { + errorCodes[result.code]++ + } + } + sort.Slice(durations, func(i, j int) bool { return durations[i] < durations[j] }) + return loadReport{ + StartedAt: started, FinishedAt: finished, + Requests: cfg.Requests, Concurrency: cfg.Concurrency, + Success: success, Errors: len(results) - success, + P50MS: percentileMS(durations, 0.50), P95MS: percentileMS(durations, 0.95), + P99MS: percentileMS(durations, 0.99), MaxMS: percentileMS(durations, 1), + ErrorCodes: errorCodes, + } +} + +func percentileMS(values []time.Duration, fraction float64) float64 { + if len(values) == 0 { + return 0 + } + index := int(math.Ceil(float64(len(values))*fraction)) - 1 + if index < 0 { + index = 0 + } + if index >= len(values) { + index = len(values) - 1 + } + return float64(values[index]) / float64(time.Millisecond) +} + +func errorCode(err error) string { + var apiErr *orbit.APIError + if errors.As(err, &apiErr) { + if apiErr.Code != "" { + return apiErr.Code + } + return fmt.Sprintf("HTTP_%d", apiErr.StatusCode) + } + var httpErr *orbit.HTTPError + if errors.As(err, &httpErr) { + return fmt.Sprintf("HTTP_%d", httpErr.StatusCode) + } + if errors.Is(err, context.DeadlineExceeded) { + return "TIMEOUT" + } + return "TRANSPORT" +} + +func newRunID() (string, error) { + var bytes [16]byte + if _, err := rand.Read(bytes[:]); err != nil { + return "", fmt.Errorf("generate load run id: %w", err) + } + return hex.EncodeToString(bytes[:]), nil +} + +func validID(value string) bool { + if len(value) != 32 { + return false + } + for _, ch := range value { + if !(ch >= '0' && ch <= '9' || ch >= 'a' && ch <= 'f') { + return false + } + } + return true +} diff --git a/sdk/go/cmd/orbit-load/main_test.go b/sdk/go/cmd/orbit-load/main_test.go new file mode 100644 index 0000000..8cf2c4d --- /dev/null +++ b/sdk/go/cmd/orbit-load/main_test.go @@ -0,0 +1,79 @@ +package main + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +func TestPercentileMSUsesNearestRankAndEmptyIsZero(t *testing.T) { + values := []time.Duration{10 * time.Millisecond, 20 * time.Millisecond, 30 * time.Millisecond, 40 * time.Millisecond} + if got := percentileMS(values, 0.50); got != 20 { + t.Fatalf("p50=%v, want 20", got) + } + if got := percentileMS(values, 0.95); got != 40 { + t.Fatalf("p95=%v, want 40", got) + } + if got := percentileMS(nil, 0.99); got != 0 { + t.Fatalf("empty percentile=%v, want 0", got) + } +} + +func TestParseConfigClampsWorkersAndRejectsInvalidQueue(t *testing.T) { + queue := "0123456789abcdef0123456789abcdef" + cfg, err := parseConfig([]string{"-url", "http://127.0.0.1:8080", "-token", "token", "-queue", queue, "-requests", "3", "-concurrency", "8"}) + if err != nil { + t.Fatal(err) + } + if cfg.Concurrency != 3 { + t.Fatalf("concurrency=%d, want clamp to requests", cfg.Concurrency) + } + if _, err := parseConfig([]string{"-url", "http://127.0.0.1:8080", "-token", "token", "-queue", "bad"}); err == nil { + t.Fatal("invalid queue accepted") + } +} + +func TestErrorCodeDoesNotExposeTransportText(t *testing.T) { + if got := errorCode(&orbit.APIError{StatusCode: 409, Code: "IDEMPOTENCY_CONFLICT"}); got != "IDEMPOTENCY_CONFLICT" { + t.Fatalf("api code=%q", got) + } + if got := errorCode(context.DeadlineExceeded); got != "TIMEOUT" { + t.Fatalf("deadline code=%q", got) + } + if got := errorCode(errors.New("secret transport detail")); got != "TRANSPORT" { + t.Fatalf("transport code=%q", got) + } +} + +func TestRunSamplesUsesOneRequestPerSequenceWithoutRetry(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/v1/tasks" { + t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path) + } + requests.Add(1) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"request_id":"0123456789abcdef0123456789abcdef","data":{"task":{},"deduplicated":false}}`)) + })) + defer server.Close() + client, err := orbit.NewClient(server.URL, orbit.Options{Token: "test-token"}) + if err != nil { + t.Fatal(err) + } + cfg := sampleConfig{QueueID: "0123456789abcdef0123456789abcdef", Requests: 7, Concurrency: 3, Timeout: time.Second, Handler: "load.noop", HandlerVersion: "v1", PayloadVersion: "v1"} + results := runSamples(client, cfg, "0123456789abcdef0123456789abcdef") + if len(results) != cfg.Requests || int(requests.Load()) != cfg.Requests { + t.Fatalf("results=%d requests=%d, want %d", len(results), requests.Load(), cfg.Requests) + } + for _, result := range results { + if result.code != "" { + t.Fatalf("unexpected request failure: %s", result.code) + } + } +}