144 lines
4.8 KiB
Go
144 lines
4.8 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
const (
|
||
// outboxRetryFixedInterval 是 V1 唯一支持的补偿策略，失败记录下一轮继续扫描。
|
||
outboxRetryFixedInterval = "fixed_interval"
|
||
// outboxWorkerName 固定写入日志，便于后续按 worker 聚合检索。
|
||
outboxWorkerName = "room_outbox"
|
||
)
|
||
|
||
// OutboxWorkerOptions 控制 room_outbox pending 事件的持续补偿循环。
|
||
type OutboxWorkerOptions struct {
|
||
// PollInterval 是两轮 pending 扫描之间的间隔；非正数会被归一到安全默认值。
|
||
PollInterval time.Duration
|
||
// BatchSize 是单轮最多处理条数；非正数会被归一，避免全表扫描。
|
||
BatchSize int
|
||
// PublishTimeout 限制单条外部投递耗时，shutdown 时当前事件最多阻塞该时长。
|
||
PublishTimeout time.Duration
|
||
// RetryStrategy 当前只支持 fixed_interval。
|
||
RetryStrategy string
|
||
}
|
||
|
||
func defaultOutboxWorkerOptions() OutboxWorkerOptions {
|
||
return OutboxWorkerOptions{
|
||
PollInterval: time.Second,
|
||
BatchSize: 100,
|
||
PublishTimeout: 3 * time.Second,
|
||
RetryStrategy: outboxRetryFixedInterval,
|
||
}
|
||
}
|
||
|
||
func normalizeOutboxWorkerOptions(options OutboxWorkerOptions) OutboxWorkerOptions {
|
||
defaults := defaultOutboxWorkerOptions()
|
||
if options.PollInterval <= 0 {
|
||
options.PollInterval = defaults.PollInterval
|
||
}
|
||
if options.BatchSize <= 0 {
|
||
options.BatchSize = defaults.BatchSize
|
||
}
|
||
if options.PublishTimeout <= 0 {
|
||
options.PublishTimeout = defaults.PublishTimeout
|
||
}
|
||
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) {
|
||
options = normalizeOutboxWorkerOptions(options)
|
||
if options.RetryStrategy != outboxRetryFixedInterval {
|
||
// 配置层会拒绝未知策略；这里兜底避免测试或手写构造静默改变语义。
|
||
log.Printf("worker=%s node_id=%s status=stopped error=%q", outboxWorkerName, s.nodeID, "unsupported retry_strategy")
|
||
return
|
||
}
|
||
|
||
if err := s.ProcessPendingOutbox(ctx, options); err != nil && ctx.Err() == nil {
|
||
log.Printf("worker=%s node_id=%s status=batch_failed error=%q", outboxWorkerName, s.nodeID, trimOutboxError(err.Error()))
|
||
}
|
||
|
||
ticker := time.NewTicker(options.PollInterval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
if err := s.ProcessPendingOutbox(ctx, options); err != nil && ctx.Err() == nil {
|
||
log.Printf("worker=%s node_id=%s status=batch_failed error=%q", outboxWorkerName, s.nodeID, trimOutboxError(err.Error()))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// ProcessPendingOutbox 把待投递事件补偿性地推送给异步消费者。
|
||
func (s *Service) ProcessPendingOutbox(ctx context.Context, options OutboxWorkerOptions) error {
|
||
options = normalizeOutboxWorkerOptions(options)
|
||
if options.RetryStrategy != outboxRetryFixedInterval {
|
||
return fmt.Errorf("unsupported outbox retry_strategy %q", options.RetryStrategy)
|
||
}
|
||
|
||
// 补偿 worker 从 repository 扫描 pending 事件；同步 room->im 失败不会回滚房间状态。
|
||
records, err := s.repository.ListPendingOutbox(ctx, options.BatchSize)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
for _, record := range records {
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
default:
|
||
}
|
||
|
||
publishCtx, cancel := context.WithTimeout(context.Background(), options.PublishTimeout)
|
||
err := s.outboxPublisher.PublishOutboxEvent(publishCtx, record.Envelope)
|
||
cancel()
|
||
if err != nil {
|
||
// 投递失败保持 pending，并记录错误和重试次数，等待下一轮 worker 再试。
|
||
markCtx, markCancel := context.WithTimeout(context.Background(), options.PublishTimeout)
|
||
markErr := s.repository.MarkOutboxFailed(markCtx, record.EventID, trimOutboxError(err.Error()))
|
||
markCancel()
|
||
if markErr != nil {
|
||
return markErr
|
||
}
|
||
|
||
log.Printf("worker=%s node_id=%s event_id=%s event_type=%s room_id=%s retry_count=%d status=pending error=%q", outboxWorkerName, s.nodeID, record.EventID, record.EventType, record.RoomID, record.RetryCount+1, trimOutboxError(err.Error()))
|
||
continue
|
||
}
|
||
|
||
// 只有外部发布成功后才能标记 published。
|
||
markCtx, markCancel := context.WithTimeout(context.Background(), options.PublishTimeout)
|
||
markErr := s.repository.MarkOutboxPublished(markCtx, record.EventID)
|
||
markCancel()
|
||
if markErr != nil {
|
||
return markErr
|
||
}
|
||
|
||
log.Printf("worker=%s node_id=%s event_id=%s event_type=%s room_id=%s retry_count=%d status=published", outboxWorkerName, s.nodeID, record.EventID, record.EventType, record.RoomID, record.RetryCount)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func trimOutboxError(value string) string {
|
||
value = strings.TrimSpace(value)
|
||
if len(value) <= 512 {
|
||
return value
|
||
}
|
||
|
||
return value[:512]
|
||
}
|