463 lines
18 KiB
Go
463 lines
18 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"log/slog"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/services/room-service/internal/room/outbox"
|
||
)
|
||
|
||
const (
|
||
// outboxRetryExponentialBackoff 是房间事件补偿的默认策略，避免外部故障时每秒刷同一条错误。
|
||
outboxRetryExponentialBackoff = "exponential_backoff"
|
||
// outboxWorkerName 固定写入日志，便于后续按 worker 聚合检索。
|
||
outboxWorkerName = "room_outbox"
|
||
)
|
||
|
||
// OutboxWorkerOptions 控制 room_outbox pending 事件的持续补偿循环。
|
||
type OutboxWorkerOptions struct {
|
||
// PollInterval 是两轮 pending 扫描之间的间隔；非正数会被归一到安全默认值。
|
||
PollInterval time.Duration
|
||
// BatchSize 是单轮最多处理条数；非正数会被归一，避免全表扫描。
|
||
BatchSize int
|
||
// Concurrency 是本进程内独立抢占 pending outbox 的 worker 数；每条 worker 都依赖 MySQL SKIP LOCKED 隔离批次。
|
||
Concurrency int
|
||
// PublishTimeout 限制单条外部投递耗时，shutdown 时当前事件最多阻塞该时长。
|
||
PublishTimeout time.Duration
|
||
// RetryStrategy 当前只支持 exponential_backoff。
|
||
RetryStrategy string
|
||
// MaxRetryCount 是转入 failed 前允许的最大失败次数。
|
||
MaxRetryCount int
|
||
// InitialBackoff 是首次失败后的等待时间。
|
||
InitialBackoff time.Duration
|
||
// MaxBackoff 是指数退避上限。
|
||
MaxBackoff time.Duration
|
||
// WorkerID 写入 outbox worker_id，空值时使用节点 ID；并发启动时会自动带上序号。
|
||
WorkerID string
|
||
// WorkerName 写入日志 worker 字段，空值时使用主通道的稳定名称。
|
||
WorkerName string
|
||
}
|
||
|
||
func defaultOutboxWorkerOptions() OutboxWorkerOptions {
|
||
// 默认值优先避免 busy loop、无限重试和日志污染。
|
||
return OutboxWorkerOptions{
|
||
PollInterval: time.Second,
|
||
BatchSize: 100,
|
||
Concurrency: 1,
|
||
PublishTimeout: 3 * time.Second,
|
||
RetryStrategy: outboxRetryExponentialBackoff,
|
||
MaxRetryCount: 10,
|
||
InitialBackoff: 5 * time.Second,
|
||
MaxBackoff: 5 * time.Minute,
|
||
}
|
||
}
|
||
|
||
func firstNonEmpty(values ...string) string {
|
||
// 配置和事件里经常需要“显式值优先，运行上下文兜底”；集中处理可避免空字符串穿透到 worker_id、app_code 等排障关键字段。
|
||
for _, value := range values {
|
||
if strings.TrimSpace(value) != "" {
|
||
return value
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func normalizeOutboxWorkerOptions(options OutboxWorkerOptions) OutboxWorkerOptions {
|
||
// worker 层再次归一化，防止测试或手动构造绕过 config.Normalize。
|
||
defaults := defaultOutboxWorkerOptions()
|
||
if options.PollInterval <= 0 {
|
||
// 非正间隔不能直接传给 ticker，否则会 panic 或形成忙循环。
|
||
options.PollInterval = defaults.PollInterval
|
||
}
|
||
if options.BatchSize <= 0 {
|
||
// 非正批量按安全默认值处理，避免 repository 扫描不受控。
|
||
options.BatchSize = defaults.BatchSize
|
||
}
|
||
if options.Concurrency <= 0 {
|
||
// 并发数为 0 等价于历史单 worker，避免老测试或手工构造配置意外关闭补偿。
|
||
options.Concurrency = defaults.Concurrency
|
||
}
|
||
if options.PublishTimeout <= 0 {
|
||
// 单条外部投递必须有 deadline，shutdown 才能等待有界。
|
||
options.PublishTimeout = defaults.PublishTimeout
|
||
}
|
||
if options.MaxRetryCount <= 0 {
|
||
options.MaxRetryCount = defaults.MaxRetryCount
|
||
}
|
||
if options.InitialBackoff <= 0 {
|
||
options.InitialBackoff = defaults.InitialBackoff
|
||
}
|
||
if options.MaxBackoff <= 0 {
|
||
options.MaxBackoff = defaults.MaxBackoff
|
||
}
|
||
if options.MaxBackoff < options.InitialBackoff {
|
||
options.MaxBackoff = options.InitialBackoff
|
||
}
|
||
options.RetryStrategy = strings.TrimSpace(options.RetryStrategy)
|
||
if options.RetryStrategy == "" {
|
||
// 空策略按指数退避处理，保持和配置默认值一致。
|
||
options.RetryStrategy = defaults.RetryStrategy
|
||
}
|
||
|
||
return options
|
||
}
|
||
|
||
// RunOutboxWorker 持续扫描 pending outbox；它不执行房间命令，只投递已持久化事件。
|
||
func (s *Service) RunOutboxWorker(ctx context.Context, options OutboxWorkerOptions) {
|
||
s.runOutboxWorker(ctx, options, outboxWorkerName, s.outboxWake, s.ProcessPendingOutbox)
|
||
}
|
||
|
||
func (s *Service) runOutboxWorker(ctx context.Context, options OutboxWorkerOptions, workerName string, wake <-chan struct{}, process func(context.Context, OutboxWorkerOptions) error) {
|
||
options = normalizeOutboxWorkerOptions(options)
|
||
if options.RetryStrategy != outboxRetryExponentialBackoff {
|
||
// 配置层会拒绝未知策略；这里兜底避免测试或手写构造静默改变语义。
|
||
logx.Error(ctx, "worker_stopped", fmt.Errorf("unsupported retry_strategy"), slog.String("worker", workerName), slog.String("node_id", s.nodeID))
|
||
return
|
||
}
|
||
if options.Concurrency == 1 {
|
||
// 单 worker 保持历史 worker_id 和日志名称，避免线上排障查询字段无意义变化。
|
||
options.WorkerID = firstNonEmpty(strings.TrimSpace(options.WorkerID), s.nodeID)
|
||
options.WorkerName = firstNonEmpty(strings.TrimSpace(options.WorkerName), workerName)
|
||
s.runOutboxWorkerLoop(ctx, options, wake, process)
|
||
return
|
||
}
|
||
|
||
var wg sync.WaitGroup
|
||
for index := 1; index <= options.Concurrency; index++ {
|
||
workerOptions := options
|
||
// 多 worker 共用同一个进程上下文，但必须写不同 worker_id，方便从 delivering 记录定位卡住的协程。
|
||
workerOptions.WorkerName = fmt.Sprintf("%s-%02d", workerName, index)
|
||
workerOptions.WorkerID = fmt.Sprintf("%s/%s", s.nodeID, workerOptions.WorkerName)
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
s.runOutboxWorkerLoop(ctx, workerOptions, wake, process)
|
||
}()
|
||
}
|
||
wg.Wait()
|
||
}
|
||
|
||
func (s *Service) runOutboxWorkerLoop(ctx context.Context, options OutboxWorkerOptions, wake <-chan struct{}, process func(context.Context, OutboxWorkerOptions) error) {
|
||
workerName := firstNonEmpty(strings.TrimSpace(options.WorkerName), outboxWorkerName)
|
||
// 启动后先执行一轮，避免服务刚恢复时还要等一个 poll interval 才补偿历史事件。
|
||
s.runOutboxWorkerBatch(ctx, options, workerName, process)
|
||
|
||
ticker := time.NewTicker(options.PollInterval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
s.runOutboxWorkerBatch(ctx, options, workerName, process)
|
||
case <-wake:
|
||
// 提交唤醒只负责打破 poll_interval 下限；真正领取仍靠 MySQL claim 和 lock_until_ms。
|
||
s.runOutboxWorkerBatch(ctx, options, workerName, process)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Service) runOutboxWorkerBatch(ctx context.Context, options OutboxWorkerOptions, workerName string, process func(context.Context, OutboxWorkerOptions) error) {
|
||
if err := process(ctx, options); err != nil && ctx.Err() == nil {
|
||
logx.Error(ctx, "worker_batch_failed", err, slog.String("worker", workerName), slog.String("node_id", s.nodeID), slog.Int("batch_size", options.BatchSize))
|
||
}
|
||
}
|
||
|
||
func (s *Service) notifyOutboxCommitted(outboxRecords []outbox.Record) {
|
||
// 唤醒信号必须在 MySQL 事务成功后发送；它只降低新事件延迟，丢失时仍由周期扫描兜底。
|
||
s.notifyOutboxWake(s.outboxWake, len(outboxRecords))
|
||
}
|
||
|
||
func (s *Service) notifyOutboxWake(wake chan struct{}, recordCount int) {
|
||
if wake == nil || recordCount <= 0 {
|
||
return
|
||
}
|
||
select {
|
||
case wake <- struct{}{}:
|
||
default:
|
||
// channel 满说明已经有 worker 被唤醒或待唤醒；丢弃信号不会丢事件，MySQL pending 仍可扫描。
|
||
}
|
||
}
|
||
|
||
// ProcessPendingOutbox 把待投递事件补偿性地推送给异步消费者。
|
||
func (s *Service) ProcessPendingOutbox(ctx context.Context, options OutboxWorkerOptions) error {
|
||
workerName := firstNonEmpty(strings.TrimSpace(options.WorkerName), outboxWorkerName)
|
||
return s.processPendingOutbox(ctx, options, workerName)
|
||
}
|
||
|
||
func (s *Service) processPendingOutbox(ctx context.Context, options OutboxWorkerOptions, workerName string) error {
|
||
options = normalizeOutboxWorkerOptions(options)
|
||
if options.RetryStrategy != outboxRetryExponentialBackoff {
|
||
return fmt.Errorf("unsupported outbox retry_strategy %q", options.RetryStrategy)
|
||
}
|
||
appCodes, err := s.roomOutboxAppCodes(ctx)
|
||
if err != nil || len(appCodes) == 0 {
|
||
return err
|
||
}
|
||
|
||
for processed := 0; processed < options.BatchSize; processed++ {
|
||
select {
|
||
case <-ctx.Done():
|
||
// 退出时不再领取新事件；已领取的单条事实会在有界 lease 后重新可见。
|
||
return ctx.Err()
|
||
default:
|
||
}
|
||
|
||
// RocketMQ/direct publisher 都是同步调用，因此每次只 claim 一条。旧实现一次
|
||
// claim 整批却只给一个 publish_timeout 的 lease，队尾尚未开始发送就可能被
|
||
// 其他实例接管并发重复发布。逐条 claim 同时避免把崩溃恢复时间放大为
|
||
// batch_size*publish_timeout，吞吐由配置的 worker concurrency 横向提供。
|
||
claimOptions := options
|
||
claimOptions.BatchSize = 1
|
||
records, err := s.claimPendingOutboxForApps(ctx, claimOptions, appCodes)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(records) == 0 {
|
||
return nil
|
||
}
|
||
record := records[0]
|
||
if record.RetryCount >= options.MaxRetryCount {
|
||
// 已达到最大失败次数的历史事件直接转死信，防止每轮 worker 继续打外部系统。
|
||
markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
|
||
markErr := s.markOutboxDead(markCtx, record.EventID, deadLetterReason(record))
|
||
markCancel()
|
||
if markErr != nil {
|
||
return markErr
|
||
}
|
||
logx.Error(ctx, "outbox_dead_letter", nil,
|
||
slog.String("worker", workerName),
|
||
slog.String("node_id", s.nodeID),
|
||
slog.String("event_id", record.EventID),
|
||
slog.String("event_type", record.EventType),
|
||
slog.String("room_id", record.RoomID),
|
||
slog.Int("retry_count", record.RetryCount),
|
||
slog.String("last_error", trimOutboxError(record.LastError)),
|
||
)
|
||
continue
|
||
}
|
||
|
||
publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
|
||
// 发布上下文不继承 worker ctx，避免 shutdown 取消导致已开始的单条外部投递无界处于未知状态。
|
||
err = s.publishOutboxRecord(publishCtx, record)
|
||
cancel()
|
||
if err != nil {
|
||
// 投递失败转为 retryable，并写入指数退避时间，等待下一轮 worker 到期后再抢占。
|
||
nextRetryAtMS := time.Now().UTC().Add(outboxBackoff(record.RetryCount+1, options)).UnixMilli()
|
||
markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
|
||
markErr := s.markOutboxRetryable(markCtx, record.EventID, trimOutboxError(err.Error()), nextRetryAtMS)
|
||
markCancel()
|
||
if markErr != nil {
|
||
return markErr
|
||
}
|
||
|
||
logx.Warn(ctx, "outbox_publish_retryable",
|
||
slog.String("worker", workerName),
|
||
slog.String("node_id", s.nodeID),
|
||
slog.String("event_id", record.EventID),
|
||
slog.String("event_type", record.EventType),
|
||
slog.String("room_id", record.RoomID),
|
||
slog.Int("retry_count", record.RetryCount+1),
|
||
slog.Int64("next_retry_at_ms", nextRetryAtMS),
|
||
slog.String("error", trimOutboxError(err.Error())),
|
||
)
|
||
continue
|
||
}
|
||
|
||
// 只有外部发布成功后才能标记 delivered。
|
||
markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
|
||
markErr := s.markOutboxDelivered(markCtx, record.EventID)
|
||
markCancel()
|
||
if markErr != nil {
|
||
return markErr
|
||
}
|
||
|
||
logx.Info(ctx, "outbox_delivered",
|
||
slog.String("worker", workerName),
|
||
slog.String("node_id", s.nodeID),
|
||
slog.String("event_id", record.EventID),
|
||
slog.String("event_type", record.EventType),
|
||
slog.String("room_id", record.RoomID),
|
||
slog.Int("retry_count", record.RetryCount),
|
||
)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func sortOutboxRecords(records []outbox.Record) {
|
||
// Direct IM 和机器人展示仍会在内存批次内调用此排序；主 MQ worker 改为
|
||
// 逐条 claim 后不依赖批内排序，但保留统一优先级可避免展示路径语义漂移。
|
||
sort.SliceStable(records, func(i, j int) bool {
|
||
leftPriority := outboxEventPriority(records[i].EventType)
|
||
rightPriority := outboxEventPriority(records[j].EventType)
|
||
if leftPriority != rightPriority {
|
||
return leftPriority < rightPriority
|
||
}
|
||
if records[i].CreatedAtMS != records[j].CreatedAtMS {
|
||
return records[i].CreatedAtMS < records[j].CreatedAtMS
|
||
}
|
||
return records[i].EventID < records[j].EventID
|
||
})
|
||
}
|
||
|
||
func outboxEventPriority(eventType string) int {
|
||
switch eventType {
|
||
case "RoomGiftSent", "RoomMicChanged", "RoomUserJoined", "RoomUserLeft":
|
||
return 0
|
||
case "RoomHeatChanged", "RoomRankChanged", "RoomRocketFuelChanged":
|
||
return 2
|
||
default:
|
||
return 1
|
||
}
|
||
}
|
||
|
||
func (s *Service) claimPendingOutbox(ctx context.Context, options OutboxWorkerOptions) ([]outbox.Record, error) {
|
||
appCodes, err := s.roomOutboxAppCodes(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return s.claimPendingOutboxForApps(ctx, options, appCodes)
|
||
}
|
||
|
||
func (s *Service) claimPendingOutboxForApps(ctx context.Context, options OutboxWorkerOptions, appCodes []string) ([]outbox.Record, error) {
|
||
if len(appCodes) == 0 {
|
||
return nil, nil
|
||
}
|
||
lockUntilMS := time.Now().UTC().Add(outboxSingleRecordLease(options.PublishTimeout)).UnixMilli()
|
||
workerID := firstNonEmpty(strings.TrimSpace(options.WorkerID), s.nodeID)
|
||
|
||
// 主 worker 每次只 claim 一条；轮转 App 起点可保留单条 lease 语义，同时避免某个高流量 App 独占所有轮次。
|
||
start := int(s.outboxClaimCursor.Add(1)-1) % len(appCodes)
|
||
for offset := 0; offset < len(appCodes); offset++ {
|
||
appCode := appCodes[(start+offset)%len(appCodes)]
|
||
records, err := s.repository.ClaimPendingOutbox(appcode.WithContext(ctx, appCode), workerID, options.BatchSize, lockUntilMS)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(records) > 0 {
|
||
return records, nil
|
||
}
|
||
}
|
||
return nil, nil
|
||
}
|
||
|
||
type roomOutboxAppCodeLister interface {
|
||
ListRoomOutboxAppCodes(ctx context.Context) ([]string, error)
|
||
}
|
||
|
||
func (s *Service) roomOutboxAppCodes(ctx context.Context) ([]string, error) {
|
||
// memory/fake repository 没有跨 App 枚举能力时沿用调用上下文，保持隔离测试和旧实现兼容。
|
||
lister, ok := s.repository.(roomOutboxAppCodeLister)
|
||
if !ok {
|
||
return []string{appcode.FromContext(ctx)}, nil
|
||
}
|
||
appCodes, err := lister.ListRoomOutboxAppCodes(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
seen := make(map[string]struct{}, len(appCodes))
|
||
normalized := make([]string, 0, len(appCodes))
|
||
for _, appCode := range appCodes {
|
||
appCode = appcode.Normalize(appCode)
|
||
if _, exists := seen[appCode]; exists {
|
||
continue
|
||
}
|
||
seen[appCode] = struct{}{}
|
||
normalized = append(normalized, appCode)
|
||
}
|
||
sort.Strings(normalized)
|
||
return normalized, nil
|
||
}
|
||
|
||
func outboxSingleRecordLease(publishTimeout time.Duration) time.Duration {
|
||
if publishTimeout <= 0 {
|
||
publishTimeout = defaultOutboxWorkerOptions().PublishTimeout
|
||
}
|
||
// 一条事实最多经历一次外部发布 deadline 和一次状态更新 deadline；额外
|
||
// 预留一个 deadline 吸收调度延迟。at-least-once 的 publish 成功/ACK 失败
|
||
// 窗口仍由下游 event_id 幂等收敛，但不会因批内队尾 lease 过期主动放大重复。
|
||
return 3 * publishTimeout
|
||
}
|
||
|
||
func (s *Service) markOutboxDelivered(ctx context.Context, eventID string) error {
|
||
return s.repository.MarkOutboxDelivered(ctx, eventID)
|
||
}
|
||
|
||
func (s *Service) markOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error {
|
||
return s.repository.MarkOutboxRetryable(ctx, eventID, lastErr, nextRetryAtMS)
|
||
}
|
||
|
||
func (s *Service) markOutboxDead(ctx context.Context, eventID string, lastErr string) error {
|
||
return s.repository.MarkOutboxDead(ctx, eventID, lastErr)
|
||
}
|
||
|
||
func (s *Service) publishOutboxRecord(ctx context.Context, record outbox.Record) error {
|
||
if record.EventType == roomLuckyGiftDrawnEventType {
|
||
// 真人中奖展示是 lucky-gift owner 事实在 room-service 的本地派生 outbox：只投腾讯云房间群。
|
||
// 若继续走通用 outboxPublisher，会把派生事件重新发到 hyapp_room_outbox，造成事实 owner 混淆和循环 fanout。
|
||
if record.Envelope == nil || record.Envelope.GetEventType() != roomLuckyGiftDrawnEventType {
|
||
return errors.New("room lucky gift display envelope is invalid")
|
||
}
|
||
if s.roomDisplayPublisher == nil {
|
||
return errors.New("room display publisher is not configured")
|
||
}
|
||
return s.roomDisplayPublisher.PublishOutboxEvent(ctx, record.Envelope)
|
||
}
|
||
// Ignited 在 outbox worker 内安排延迟发射，保证 MQ 调度失败能通过 room_outbox retry；客户端 IM 展示已在 Room Cell 提交后直发。
|
||
if err := s.scheduleRoomRocketLaunchFromEnvelope(ctx, record.Envelope); err != nil {
|
||
return err
|
||
}
|
||
if s.outboxPublisher == nil {
|
||
return nil
|
||
}
|
||
return s.outboxPublisher.PublishOutboxEvent(ctx, record.Envelope)
|
||
}
|
||
|
||
// outboxBackoff 根据失败次数计算指数退避，且永远不超过 MaxBackoff。
|
||
func outboxBackoff(retryCount int, options OutboxWorkerOptions) time.Duration {
|
||
if retryCount <= 1 {
|
||
return options.InitialBackoff
|
||
}
|
||
backoff := options.InitialBackoff
|
||
for i := 1; i < retryCount; i++ {
|
||
if backoff >= options.MaxBackoff/2 {
|
||
// 提前截断，避免 duration 乘 2 溢出或超过配置上限。
|
||
return options.MaxBackoff
|
||
}
|
||
backoff *= 2
|
||
}
|
||
if backoff > options.MaxBackoff {
|
||
return options.MaxBackoff
|
||
}
|
||
return backoff
|
||
}
|
||
|
||
// deadLetterReason 生成写入 failed 状态的稳定错误原因。
|
||
func deadLetterReason(record outbox.Record) string {
|
||
if strings.TrimSpace(record.LastError) == "" {
|
||
return "outbox retry limit exceeded"
|
||
}
|
||
return trimOutboxError("outbox retry limit exceeded: " + record.LastError)
|
||
}
|
||
|
||
func trimOutboxError(value string) string {
|
||
// last_error 会写入 MySQL 和日志，先去掉首尾空白避免不可见差异影响排障。
|
||
value = strings.TrimSpace(value)
|
||
if len(value) <= 512 {
|
||
return value
|
||
}
|
||
|
||
// V1 只保留错误前缀，避免外部响应过大撑爆日志或 TEXT 字段查询成本。
|
||
return value[:512]
|
||
}
|