ec35fcad5e
- outbox_worker.retention 配置块（默认关闭，max_age<24h 拒绝启动） - 清理走预留的 idx_room_outbox_retention 索引，先选主键再小批量 DELETE - 移除与 idx_room_outbox_pending 完全重复的 idx_room_outbox_retry， auto_migrate=true 时 Migrate 自动 DROP - 开启清理前需和数据侧确认保留窗口覆盖统计重放/补数工具 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
109 lines
3.5 KiB
Go
109 lines
3.5 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"hyapp/pkg/logx"
|
|
)
|
|
|
|
// outboxRetentionBatchPause 是同一轮内两批 DELETE 之间的让路间隔，避免清理积压时长时间占用行锁和 IO。
|
|
const outboxRetentionBatchPause = 100 * time.Millisecond
|
|
|
|
// OutboxRetentionOptions 控制 delivered 历史事件的定期清理循环。
|
|
type OutboxRetentionOptions struct {
|
|
// ScanInterval 是清空一轮可删事件后到下一轮扫描的间隔；非正数会被归一到安全默认值。
|
|
ScanInterval time.Duration
|
|
// MaxAge 是 delivered 事件按 updated_at_ms 计算的保留时长。
|
|
MaxAge time.Duration
|
|
// BatchSize 是单条 DELETE 的行数上限。
|
|
BatchSize int
|
|
}
|
|
|
|
func defaultOutboxRetentionOptions() OutboxRetentionOptions {
|
|
// 默认值和 config.defaultOutboxWorkerConfig 保持一致，防止手动构造绕过配置归一化后误删新事件。
|
|
return OutboxRetentionOptions{
|
|
ScanInterval: 5 * time.Minute,
|
|
MaxAge: 30 * 24 * time.Hour,
|
|
BatchSize: 500,
|
|
}
|
|
}
|
|
|
|
func normalizeOutboxRetentionOptions(options OutboxRetentionOptions) OutboxRetentionOptions {
|
|
defaults := defaultOutboxRetentionOptions()
|
|
if options.ScanInterval <= 0 {
|
|
options.ScanInterval = defaults.ScanInterval
|
|
}
|
|
if options.MaxAge <= 0 {
|
|
options.MaxAge = defaults.MaxAge
|
|
}
|
|
if options.BatchSize <= 0 {
|
|
options.BatchSize = defaults.BatchSize
|
|
}
|
|
return options
|
|
}
|
|
|
|
// RunOutboxRetentionWorker 周期性删除 delivered 且超过保留窗口的 room_outbox 历史事件。
|
|
// 它只做数据保留治理，不参与投递；多节点同时运行时 DELETE 靠 MySQL 行锁串行，语义幂等。
|
|
func (s *Service) RunOutboxRetentionWorker(ctx context.Context, options OutboxRetentionOptions) {
|
|
if s == nil || s.repository == nil {
|
|
return
|
|
}
|
|
options = normalizeOutboxRetentionOptions(options)
|
|
|
|
// 启动先执行一轮，让积压清理不用等第一个 scan interval。
|
|
s.runOutboxRetentionRound(ctx, options)
|
|
|
|
ticker := time.NewTicker(options.ScanInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
s.runOutboxRetentionRound(ctx, options)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) runOutboxRetentionRound(ctx context.Context, options OutboxRetentionOptions) {
|
|
if _, err := s.ProcessOutboxRetention(ctx, options); err != nil && ctx.Err() == nil {
|
|
// 单轮失败只影响清理进度，下一轮扫描会继续；不升级为 Error 避免和投递故障混在告警里。
|
|
logx.Warn(ctx, "outbox_retention_failed", slog.String("error", err.Error()))
|
|
}
|
|
}
|
|
|
|
// ProcessOutboxRetention 执行一轮清理：按批删除到没有可删事件为止，返回本轮总删除行数。
|
|
func (s *Service) ProcessOutboxRetention(ctx context.Context, options OutboxRetentionOptions) (int64, error) {
|
|
options = normalizeOutboxRetentionOptions(options)
|
|
cutoffMS := s.clock.Now().UTC().Add(-options.MaxAge).UnixMilli()
|
|
|
|
var total int64
|
|
for {
|
|
deleted, err := s.repository.DeleteDeliveredOutboxBefore(ctx, cutoffMS, options.BatchSize)
|
|
total += deleted
|
|
if err != nil {
|
|
return total, err
|
|
}
|
|
if deleted < int64(options.BatchSize) {
|
|
break
|
|
}
|
|
// 整批删满说明还有积压，稍作停顿继续，直到删空；退出信号优先于继续清理。
|
|
select {
|
|
case <-ctx.Done():
|
|
return total, ctx.Err()
|
|
case <-time.After(outboxRetentionBatchPause):
|
|
}
|
|
}
|
|
|
|
if total > 0 {
|
|
logx.Info(ctx, "outbox_retention_purged",
|
|
slog.Int64("rows_deleted", total),
|
|
slog.Int64("cutoff_ms", cutoffMS),
|
|
slog.String("node_id", s.nodeID),
|
|
)
|
|
}
|
|
return total, nil
|
|
}
|