diff --git a/services/room-service/internal/storage/mysql/outbox.go b/services/room-service/internal/storage/mysql/outbox.go index acb371b5..b37f094f 100644 --- a/services/room-service/internal/storage/mysql/outbox.go +++ b/services/room-service/internal/storage/mysql/outbox.go @@ -161,14 +161,17 @@ func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table stri nowMS := time.Now().UTC().UnixMilli() appCode := appcode.FromContext(ctx) 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 { query string 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, '') - FROM ` + table + ` FORCE INDEX (` + pendingIndex + `) - WHERE app_code = ? AND status = ? AND next_retry_at_ms <= ? AND lock_until_ms IS NULL + FROM ` + table + ` FORCE INDEX (` + claimIndex + `) + 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 LIMIT ? FOR UPDATE SKIP LOCKED`, 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, '') FROM ` + table + ` FORCE INDEX (` + pendingIndex + `) 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`, args: []any{appCode, outbox.StatusRetryable, nowMS}, },