hy/hyapp-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(room): optimize robot outbox claim query

Browse Source
This commit is contained in:
zhx 2026-06-26 16:08:31 +08:00
parent 0039c7f2ee
commit 538668014b
1 changed files with 6 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
services/room-service/internal/storage/mysql/outbox.go
View File

@ -161,14 +161,17 @@ func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table stri
nowMS := time.Now().UTC().UnixMilli() nowMS := time.Now().UTC().UnixMilli()
appCode := appcode.FromContext(ctx) appCode := appcode.FromContext(ctx)
records := make([]outbox.Record, 0, limit) records := make([]outbox.Record, 0, limit)
// pending 新事件的 next_retry_at_ms 固定为 0，真正决定可见性的是未被抢占的 lock_until_ms；
// 因此 pending 分支必须走 claim 索引按 created_at_ms 直接取最早记录，避免在机器人 outbox 堆积时用
// next_retry_at_ms range 扫描数百万历史行再 filesort。retryable 才按 next_retry_at_ms 到期时间排序。
claimBranches := []struct { claimBranches := []struct {
query string query string
args []any args []any
}{ }{
{ {
query: `SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '') query: `SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '')
FROM ` + table + ` FORCE INDEX (` + pendingIndex + `) FROM ` + table + ` FORCE INDEX (` + claimIndex + `)
WHERE app_code = ? AND status = ? AND next_retry_at_ms <= ? AND lock_until_ms IS NULL WHERE app_code = ? AND status = ? AND lock_until_ms IS NULL AND next_retry_at_ms <= ?
ORDER BY created_at_ms ASC, event_id ASC ORDER BY created_at_ms ASC, event_id ASC
LIMIT ? FOR UPDATE SKIP LOCKED`, LIMIT ? FOR UPDATE SKIP LOCKED`,
args: []any{appCode, outbox.StatusPending, nowMS}, args: []any{appCode, outbox.StatusPending, nowMS},
@ -177,7 +180,7 @@ func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table stri
query: `SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '') query: `SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '')
FROM ` + table + ` FORCE INDEX (` + pendingIndex + `) FROM ` + table + ` FORCE INDEX (` + pendingIndex + `)
WHERE app_code = ? AND status = ? AND next_retry_at_ms <= ? AND lock_until_ms IS NULL WHERE app_code = ? AND status = ? AND next_retry_at_ms <= ? AND lock_until_ms IS NULL
ORDER BY created_at_ms ASC, event_id ASC ORDER BY next_retry_at_ms ASC, created_at_ms ASC, event_id ASC
LIMIT ? FOR UPDATE SKIP LOCKED`, LIMIT ? FOR UPDATE SKIP LOCKED`,
args: []any{appCode, outbox.StatusRetryable, nowMS}, args: []any{appCode, outbox.StatusRetryable, nowMS},
}, },