From ac8cc1260ba740ad041ddc8c3cf09d833a7bcdce Mon Sep 17 00:00:00 2001 From: k2safe Date: Sun, 20 Sep 2026 21:52:17 +0800 Subject: [PATCH] feat: expose mysql transaction retry metrics --- cmd/orbit/main.go | 4 ++++ docs/08-development-plan.md | 2 +- docs/20-operations-runbook.md | 2 +- docs/api/observability.md | 2 ++ docs/reports/phase-6-observability.md | 2 +- docs/reports/phase-7-regression-status.md | 1 + internal/observability/catalog.go | 2 ++ internal/observability/metrics_test.go | 3 ++- internal/observability/runtime.go | 18 ++++++++++++++++++ internal/storage/mysql/db.go | 23 +++++++++++++++++++++++ 10 files changed, 55 insertions(+), 4 deletions(-) diff --git a/cmd/orbit/main.go b/cmd/orbit/main.go index 5fe03c4..533ddb5 100644 --- a/cmd/orbit/main.go +++ b/cmd/orbit/main.go @@ -81,6 +81,8 @@ func run() error { DispatchCapacity: cfg.DispatchCapacity, }) metrics.RecordDatabase(db.Stats()) + txStats := mysql.ReadTransactionStats() + metrics.RecordTransactionStats(observability.TransactionStats{Retries: txStats.Retries, Exhausted: txStats.Exhausted}) } server := httpserver.New(cfg.HTTPAddr, readiness.Check, service.New(mysql.NewTaskStore(db))) httpserver.RegisterApplicationResources(server, resources.New(mysql.NewTaskStore(db))) @@ -216,6 +218,8 @@ func run() error { return case <-ticker.C: metrics.RecordDatabase(db.Stats()) + txStats := mysql.ReadTransactionStats() + metrics.RecordTransactionStats(observability.TransactionStats{Retries: txStats.Retries, Exhausted: txStats.Exhausted}) metrics.RecordProcess() started := time.Now() call, cancel := context.WithTimeout(metricsCtx, 900*time.Millisecond) diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 40c767a..a84b902 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -81,7 +81,7 @@ 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状态计数、最近事件类别、终态耗时样本、固定事件阶段耗时、最早年龄、截断/新鲜度指标和可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签；发布前只读结构核验由[`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领取协程和有界内存排队均支持有界环境配置，并在启动日志和固定指标中记录。仍需端到端外部效果延迟、全局持久吞吐、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 506cfde..e8707fe 100644 --- a/docs/20-operations-runbook.md +++ b/docs/20-operations-runbook.md @@ -104,7 +104,7 @@ curl -fsS http://127.0.0.1:9090/metrics ORBIT_MYSQL_DSN="$ORBIT_MYSQL_DSN" ./bin/orbit-admin observe snapshot ``` -快照和业务指标是单节点有界样本。`truncated=true`、快照超时、数据库连接池等待增加或后台批次错误，只能触发调查，不能直接改写任务、流程或审计状态。指标字段、告警边界和采集安全要求见[运行指标手册](api/observability.md)。 +快照和业务指标是单节点有界样本。`truncated=true`、快照超时、数据库连接池等待增加、提交前事务重试/耗尽计数增长或后台批次错误，只能触发调查，不能直接改写任务、流程或审计状态。`orbit_db_transaction_retries_total` 只表示明确的 MySQL 1205/1213 提交前回调重试，`orbit_db_transaction_retry_exhausted_total` 表示该有界重试已耗尽；Commit 回执不明不会计入，必须按稳定操作身份核对原记录。指标字段、告警边界和采集安全要求见[运行指标手册](api/observability.md)。 ## 8. 事故处理原则 diff --git a/docs/api/observability.md b/docs/api/observability.md index 8111a15..e4e5823 100644 --- a/docs/api/observability.md +++ b/docs/api/observability.md @@ -54,6 +54,8 @@ curl -fsS http://127.0.0.1:9090/metrics | `orbit_db_connections_open` / `orbit_db_connections_in_use` / `orbit_db_connections_idle` | gauge | 本节点MySQL连接池当前连接数 | | `orbit_db_connections_max_open` | gauge | 池配置的最大连接数 | | `orbit_db_wait_count_total` / `orbit_db_wait_duration_seconds_total` | counter | 连接池累计等待次数/秒数，不是数据库锁等待 | +| `orbit_db_transaction_retries_total` | counter | 本进程提交前因 MySQL 1205/1213 实际执行的事务重试次数；不包含 Begin/Commit 错误或提交回执不明 | +| `orbit_db_transaction_retry_exhausted_total` | counter | 本进程提交前因 MySQL 1205/1213 达到最多两次重试上限的事务次数；需要结合原操作身份和数据库锁观测核查 | | `orbit_pulse_tasks_state{state}` | gauge | Pulse任务固定状态的有界计数；单状态达到10000时只保留上限并置截断标记 | | `orbit_pulse_tasks_oldest_age_seconds` | gauge | WAITING/LEASED/RUNNING/NEEDS_REVIEW样本中最早可处理时间的年龄 | | `orbit_result_deliveries_state{state}` | gauge | 结果交付固定状态的有界计数 | diff --git a/docs/reports/phase-6-observability.md b/docs/reports/phase-6-observability.md index c908ca0..161f058 100644 --- a/docs/reports/phase-6-observability.md +++ b/docs/reports/phase-6-observability.md @@ -80,4 +80,4 @@ Apple M2 Max、12核、32 GiB；全量集成停止后单独运行，退出0。 1. 受理/授权/完成的持久吞吐、阶段延迟、重试/核查/恢复量、流程Inbox至推进延迟；区分提交成功、回执丢失、重报和事件重放，不能把HTTP成功数直接当作业务量。 2. 持久待处理、到期积压与发送队列。使用有上限/超时的后台采样及明确的新鲜度/失败状态，不能让每次抓取扫描全库；状态指标和多节点重复采集的聚合口径须先明确。 3. 慢消费者、重连风暴、业务负载下观测开销、真实采集/通知演练，以及数据库锁/磁盘/复制指标和跨故障域验证。 -4. 凭据轮换、保留/归档、备份恢复、容量六类负载和P7版本发布继续按[持续计划](../08-development-plan.md)执行。 +4. 提交前 MySQL 1205/1213 事务重试与耗尽已通过固定无标签计数器导出；仍需在目标数据库拓扑上校准锁等待/死锁告警阈值，并与跨节点聚合、日志和审计核对。凭据轮换、保留/归档、备份恢复、容量六类负载和P7版本发布继续按[持续计划](../08-development-plan.md)执行。 diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index 743e49c..ad76595 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -10,6 +10,7 @@ | 发布物静态身份清单 | 2026-09-20；`make release-manifest`；无 `ORBIT_MYSQL_DSN` | PASS；输出构建/协议身份及28条迁移名称与SHA-256校验和，未读取数据库、凭据或业务数据；目标库仍需单独执行 `orbit-admin verify` | | 发布物与管理二进制绑定校验 | 2026-09-20；`go test -race ./cmd/orbit-admin`；匹配/修改 revision/未知命令场景 | PASS；`verify --manifest` 在数据库连接前拒绝身份、协议或迁移校验和不一致的清单，匹配清单通过；不修改数据库 | | 提交前事务死锁恢复 | `go test -race -run TestRetryableTransactionErrorOnlyMatchesPreCommitLockFailures ./internal/storage/mysql`；真实 MySQL `TestIntegrationWithTxRetriesPreCommitDeadlock` | PASS；两事务真实锁环触发 1213，失败事务仅在提交前按有界退避重试一次，最终两笔事务各提交一次；Begin/Commit 错误不重放，目标拓扑锁等待和死锁率仍需生产观测 | +| 提交前事务重试观测 | `internal/observability.TestRuntimeObservationDoesNotExposeErrors`；固定指标目录；`make release-local-check` | PASS；`orbit_db_transaction_retries_total` 与 `orbit_db_transaction_retry_exhausted_total` 以无标签固定计数器导出，重试只在退避成功后计数，提交回执不明不计入；目标拓扑告警阈值和跨节点聚合仍待生产验收 | | 就绪探针排空门控 | `go test -race ./internal/transport/httpserver ./cmd/orbit`；`TestReadinessEntersDrainBeforeShutdown` | PASS；SIGTERM 关闭路径先将 `/health/ready` 置为 503，再进入限定优雅关闭窗口；响应不泄露内部错误 | | 健康探针响应契约 | 2026-09-20；`go test ./internal/transport/httpserver && go test -race ./internal/transport/httpserver && make docs-check` | PASS；`/health/live` 与 `/health/ready` 返回 `Cache-Control: no-store`、`X-Content-Type-Options: nosniff`；未就绪和排空状态返回 `Retry-After: 1`，并由 HTTP 契约测试覆盖 | | 数据模型中文注释门禁 | 2026-09-20；`go test ./internal/storage/mysql -run 'Test(FreshInstallBaselineHasChineseModelComments|FreshInstallBaselineContainsNoReferenceDDL|MigrationsAfterForeignKeyRemovalDoNotReintroduceReferences|PostRemovalMigrationGuard)$'` | PASS；当前无外键基线逐表、逐持久化字段检查中文 `COMMENT`，并继续阻止后续迁移重新引入引用约束 | diff --git a/internal/observability/catalog.go b/internal/observability/catalog.go index c605888..58b215b 100644 --- a/internal/observability/catalog.go +++ b/internal/observability/catalog.go @@ -40,6 +40,8 @@ func catalog() map[string]descriptor { } c["orbit_db_wait_count_total"] = descriptor{counter, "Cumulative waits for a MySQL connection", nil} c["orbit_db_wait_duration_seconds_total"] = descriptor{counter, "Cumulative seconds waiting for a MySQL connection", nil} + c["orbit_db_transaction_retries_total"] = descriptor{counter, "Cumulative pre-commit MySQL transaction retries", nil} + c["orbit_db_transaction_retry_exhausted_total"] = descriptor{counter, "Cumulative pre-commit MySQL transaction retry exhaustion", nil} c["orbit_pulse_tasks_state"] = descriptor{gauge, "Bounded count of Pulse tasks by durable state", map[string][]string{"state": taskStates}} c["orbit_pulse_tasks_oldest_age_seconds"] = descriptor{gauge, "Age of the oldest sampled unfinished Pulse task", nil} c["orbit_result_deliveries_state"] = descriptor{gauge, "Bounded count of result deliveries by durable state", map[string][]string{"state": deliveryStates}} diff --git a/internal/observability/metrics_test.go b/internal/observability/metrics_test.go index 365c9cc..3842feb 100644 --- a/internal/observability/metrics_test.go +++ b/internal/observability/metrics_test.go @@ -102,8 +102,9 @@ func TestRuntimeObservationDoesNotExposeErrors(t *testing.T) { r.RecordBatch("workflow", time.Now(), errors.New("private-dsn")) r.RecordBatch("schedule", time.Now(), context.Canceled) r.RecordDatabase(sql.DBStats{MaxOpenConnections: 20, OpenConnections: 4, InUse: 3, Idle: 1, WaitCount: 9, WaitDuration: 2 * time.Second}) + r.RecordTransactionStats(TransactionStats{Retries: 3, Exhausted: 1}) out := r.Render() - for _, want := range []string{`orbit_runtime_batch_total{component="workflow",outcome="success"} 1`, `orbit_runtime_batch_total{component="workflow",outcome="error"} 1`, `orbit_runtime_batch_total{component="schedule",outcome="cancelled"} 1`, "orbit_db_connections_in_use 3\n", "orbit_db_wait_duration_seconds_total 2\n"} { + for _, want := range []string{`orbit_runtime_batch_total{component="workflow",outcome="success"} 1`, `orbit_runtime_batch_total{component="workflow",outcome="error"} 1`, `orbit_runtime_batch_total{component="schedule",outcome="cancelled"} 1`, "orbit_db_connections_in_use 3\n", "orbit_db_wait_duration_seconds_total 2\n", "orbit_db_transaction_retries_total 3\n", "orbit_db_transaction_retry_exhausted_total 1\n"} { if !strings.Contains(out, want) { t.Fatal("missing runtime evidence", want) } diff --git a/internal/observability/runtime.go b/internal/observability/runtime.go index 4bddc80..11b2393 100644 --- a/internal/observability/runtime.go +++ b/internal/observability/runtime.go @@ -102,6 +102,9 @@ func (r *Registry) RecordBatch(component string, started time.Time, err error) { // RecordDatabase reads the local connection-pool snapshot; it issues no SQL. func (r *Registry) RecordDatabase(s sql.DBStats) { + if r == nil { + return + } r.SetGauge("orbit_db_connections_open", float64(s.OpenConnections), nil) r.SetGauge("orbit_db_connections_in_use", float64(s.InUse), nil) r.SetGauge("orbit_db_connections_idle", float64(s.Idle), nil) @@ -109,3 +112,18 @@ func (r *Registry) RecordDatabase(s sql.DBStats) { r.SetCounter("orbit_db_wait_count_total", float64(s.WaitCount), nil) r.SetCounter("orbit_db_wait_duration_seconds_total", s.WaitDuration.Seconds(), nil) } + +// TransactionStats 是本进程提交前事务重试的累计观测值。 +type TransactionStats struct { + Retries uint64 // 已实际执行的重试次数。 + Exhausted uint64 // 达到重试上限的事务次数。 +} + +// RecordTransactionStats records bounded database transaction retry counters. +func (r *Registry) RecordTransactionStats(stats TransactionStats) { + if r == nil { + return + } + r.SetCounter("orbit_db_transaction_retries_total", float64(stats.Retries), nil) + r.SetCounter("orbit_db_transaction_retry_exhausted_total", float64(stats.Exhausted), nil) +} diff --git a/internal/storage/mysql/db.go b/internal/storage/mysql/db.go index 58118b5..5cb734f 100644 --- a/internal/storage/mysql/db.go +++ b/internal/storage/mysql/db.go @@ -5,6 +5,7 @@ import ( "context" "database/sql" "errors" + "sync/atomic" "time" driver "github.com/go-sql-driver/mysql" @@ -53,6 +54,24 @@ func Open(ctx context.Context, dsn string, maxOpen, maxIdle int) (*sql.DB, error const transactionRetryLimit = 2 +var transactionRetries atomic.Uint64 +var transactionRetryExhausted atomic.Uint64 + +// TransactionStats 是本进程提交前事务重试的累计快照。 +// 计数器只记录明确的 MySQL 锁等待超时或死锁回调失败，不记录提交回执不明。 +type TransactionStats struct { + Retries uint64 // 已实际执行的重试次数。 + Exhausted uint64 // 达到重试上限的事务次数。 +} + +// ReadTransactionStats 返回本进程事务重试计数，不访问数据库。 +func ReadTransactionStats() TransactionStats { + return TransactionStats{ + Retries: transactionRetries.Load(), + Exhausted: transactionRetryExhausted.Load(), + } +} + // WithTx retries only a callback failure caused by a MySQL deadlock or lock wait // timeout before COMMIT. It never retries Begin/Commit errors because a commit // response may be lost after the database has already applied the transaction; @@ -61,11 +80,15 @@ func WithTx(ctx context.Context, db *sql.DB, fn func(*sql.Tx) error) error { for attempt := 0; attempt <= transactionRetryLimit; attempt++ { err, retryable := withTxAttempt(ctx, db, fn) if err == nil || !retryable || attempt == transactionRetryLimit { + if retryable && attempt == transactionRetryLimit { + transactionRetryExhausted.Add(1) + } return err } if err := waitTransactionRetry(ctx, attempt); err != nil { return err } + transactionRetries.Add(1) } return errors.New("transaction retry loop exhausted") }