room_outbox 增加 delivered 历史事件定期清理并移除重复索引
- 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>
This commit is contained in:
zhx
parent
41a65c244f
commit
ec35fcad5e
@ -100,3 +100,11 @@ outbox_worker:
|
||||
max_retry_count: 10
|
||||
initial_backoff: "5s"
|
||||
max_backoff: "5m"
|
||||
# retention 按 updated_at_ms 小批量删除 delivered 历史事件，防止 room_outbox 无限增长。
|
||||
# ⚠️ 开启前必须和数据侧确认 max_age 覆盖 replay-stat-tz/databi-real-flow-audit 等
|
||||
# 直读 room_outbox 的重放与补数窗口（或先归档再开启）；max_age 低于 24h 会拒绝启动。
|
||||
retention:
|
||||
enabled: false
|
||||
max_age: "720h"
|
||||
scan_interval: "5m"
|
||||
batch_size: 500
|
||||
|
||||
@ -150,7 +150,6 @@ CREATE TABLE IF NOT EXISTS room_outbox (
|
||||
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
|
||||
PRIMARY KEY (app_code, event_id),
|
||||
KEY idx_room_outbox_pending (app_code, status, next_retry_at_ms, created_at_ms, event_id),
|
||||
KEY idx_room_outbox_retry (app_code, status, next_retry_at_ms, created_at_ms, event_id),
|
||||
KEY idx_room_outbox_claim (app_code, status, lock_until_ms, created_at_ms, event_id),
|
||||
KEY idx_room_outbox_event_created (app_code, event_type, created_at_ms, event_id),
|
||||
KEY idx_room_outbox_retention (app_code, status, updated_at_ms, event_id),
|
||||
|
||||
@ -389,6 +389,16 @@ func (a *App) Run() error {
|
||||
})
|
||||
})
|
||||
}
|
||||
if a.cfg.OutboxWorker.Retention.Enabled {
|
||||
// 清理独立于补偿投递开关：outbox_worker.enabled=false 止血时历史 delivered 仍可继续回收。
|
||||
a.workers.Go(func(ctx context.Context) {
|
||||
a.service.RunOutboxRetentionWorker(ctx, roomservice.OutboxRetentionOptions{
|
||||
ScanInterval: a.cfg.OutboxWorker.Retention.ScanInterval,
|
||||
MaxAge: a.cfg.OutboxWorker.Retention.MaxAge,
|
||||
BatchSize: a.cfg.OutboxWorker.Retention.BatchSize,
|
||||
})
|
||||
})
|
||||
}
|
||||
a.workers.Go(func(ctx context.Context) {
|
||||
a.service.RunRobotRoomRuntimeManager(ctx, 10*time.Second)
|
||||
})
|
||||
|
||||
@ -116,6 +116,23 @@ type OutboxWorkerConfig struct {
|
||||
InitialBackoff time.Duration `yaml:"initial_backoff"`
|
||||
// MaxBackoff 是指数退避的最大等待时间。
|
||||
MaxBackoff time.Duration `yaml:"max_backoff"`
|
||||
// Retention 控制 delivered 历史事件的定期清理；它独立于 Enabled，临时关闭补偿投递不会停掉清理。
|
||||
Retention OutboxRetentionConfig `yaml:"retention"`
|
||||
}
|
||||
|
||||
// OutboxRetentionConfig 控制 room_outbox 中 delivered 历史事件的小批量定期删除。
|
||||
// 只清理 delivered：pending/retryable/delivering 仍在投递生命周期内，failed 留给人工补偿。
|
||||
type OutboxRetentionConfig struct {
|
||||
// Enabled 默认 false。statistics-service 的 replay-stat-tz/databi-real-flow-audit 和
|
||||
// notice-service 离线修复工具都按时间窗直读 room_outbox 历史，开启前必须和数据侧确认
|
||||
// MaxAge 覆盖所有重放/补数窗口（或先归档再删）。
|
||||
Enabled bool `yaml:"enabled"`
|
||||
// MaxAge 是 delivered 事件按 updated_at_ms 计算的保留时长，早于 now-MaxAge 的才会删除。
|
||||
MaxAge time.Duration `yaml:"max_age"`
|
||||
// ScanInterval 是清空一轮可删事件后到下一轮扫描的间隔。
|
||||
ScanInterval time.Duration `yaml:"scan_interval"`
|
||||
// BatchSize 是单条 DELETE 的行数上限，小批量避免大事务长时间持有行锁。
|
||||
BatchSize int `yaml:"batch_size"`
|
||||
}
|
||||
|
||||
// RocketMQConfig 描述 room-service 使用的 RocketMQ 集群和业务 topic。
|
||||
@ -309,6 +326,13 @@ func defaultOutboxWorkerConfig() OutboxWorkerConfig {
|
||||
MaxRetryCount: 10,
|
||||
InitialBackoff: 5 * time.Second,
|
||||
MaxBackoff: 5 * time.Minute,
|
||||
// 清理默认关闭：保留窗口必须先和数据侧确认，见 OutboxRetentionConfig.Enabled。
|
||||
Retention: OutboxRetentionConfig{
|
||||
Enabled: false,
|
||||
MaxAge: 30 * 24 * time.Hour,
|
||||
ScanInterval: 5 * time.Minute,
|
||||
BatchSize: 500,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@ -490,6 +514,19 @@ func normalizeOutboxWorkerConfig(cfg OutboxWorkerConfig) (OutboxWorkerConfig, er
|
||||
default:
|
||||
return OutboxWorkerConfig{}, fmt.Errorf("unsupported outbox_worker publish_mode %q", cfg.PublishMode)
|
||||
}
|
||||
if cfg.Retention.MaxAge <= 0 {
|
||||
cfg.Retention.MaxAge = defaults.Retention.MaxAge
|
||||
}
|
||||
if cfg.Retention.ScanInterval <= 0 {
|
||||
cfg.Retention.ScanInterval = defaults.Retention.ScanInterval
|
||||
}
|
||||
if cfg.Retention.BatchSize <= 0 {
|
||||
cfg.Retention.BatchSize = defaults.Retention.BatchSize
|
||||
}
|
||||
if cfg.Retention.Enabled && cfg.Retention.MaxAge < 24*time.Hour {
|
||||
// 保留窗口小于一天大概率是单位写错（如 720 被当秒），直接拒绝启动，避免静默清掉重放依赖的历史。
|
||||
return OutboxWorkerConfig{}, fmt.Errorf("outbox_worker retention max_age %s is below the 24h safety floor", cfg.Retention.MaxAge)
|
||||
}
|
||||
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
@ -157,3 +157,31 @@ func TestNormalizeOutboxWorkerKeepsSafeDefaults(t *testing.T) {
|
||||
t.Fatalf("outbox worker defaults not restored: %+v", normalized.OutboxWorker)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeOutboxRetentionKeepsSafeDefaults(t *testing.T) {
|
||||
cfg := Default()
|
||||
if cfg.OutboxWorker.Retention.Enabled {
|
||||
// 清理会删除统计重放依赖的历史事件，必须显式开启，默认配置绝不能自动生效。
|
||||
t.Fatal("outbox retention must default to disabled")
|
||||
}
|
||||
cfg.OutboxWorker.Retention = OutboxRetentionConfig{Enabled: true}
|
||||
|
||||
normalized, err := Normalize(cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("Normalize failed: %v", err)
|
||||
}
|
||||
retention := normalized.OutboxWorker.Retention
|
||||
if !retention.Enabled || retention.MaxAge != 30*24*time.Hour || retention.ScanInterval != 5*time.Minute || retention.BatchSize != 500 {
|
||||
t.Fatalf("outbox retention defaults not restored: %+v", retention)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeRejectsOutboxRetentionMaxAgeBelowSafetyFloor(t *testing.T) {
|
||||
cfg := Default()
|
||||
cfg.OutboxWorker.Retention.Enabled = true
|
||||
cfg.OutboxWorker.Retention.MaxAge = time.Hour
|
||||
|
||||
if _, err := Normalize(cfg); err == nil {
|
||||
t.Fatal("Normalize should reject outbox retention max_age below 24h")
|
||||
}
|
||||
}
|
||||
|
||||
108
services/room-service/internal/room/service/outbox_retention.go
Normal file
108
services/room-service/internal/room/service/outbox_retention.go
Normal file
@ -0,0 +1,108 @@
|
||||
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
|
||||
}
|
||||
@ -0,0 +1,86 @@
|
||||
package service_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||||
"hyapp/pkg/appcode"
|
||||
"hyapp/services/room-service/internal/integration"
|
||||
"hyapp/services/room-service/internal/room/outbox"
|
||||
roomservice "hyapp/services/room-service/internal/room/service"
|
||||
"hyapp/services/room-service/internal/router"
|
||||
"hyapp/services/room-service/internal/testutil/mysqltest"
|
||||
)
|
||||
|
||||
// 一轮清理必须按批删空所有过期 delivered 事件，同时保留窗口内 delivered 和未完成投递的事件。
|
||||
func TestProcessOutboxRetentionDrainsExpiredDelivered(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repository := mysqltest.NewRepository(t)
|
||||
svc := roomservice.New(roomservice.Config{
|
||||
NodeID: "node-outbox-retention-test",
|
||||
}, router.NewMemoryDirectory(), repository, followTestWallet{}, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher())
|
||||
|
||||
roomID := "room-outbox-retention"
|
||||
newRecord := func(eventID string) outbox.Record {
|
||||
return outbox.Record{
|
||||
AppCode: appcode.Default,
|
||||
EventID: eventID,
|
||||
EventType: "RoomGiftSent",
|
||||
RoomID: roomID,
|
||||
Status: outbox.StatusPending,
|
||||
Envelope: &roomeventsv1.EventEnvelope{
|
||||
AppCode: appcode.Default,
|
||||
EventId: eventID,
|
||||
RoomId: roomID,
|
||||
EventType: "RoomGiftSent",
|
||||
},
|
||||
CreatedAtMS: 1000,
|
||||
}
|
||||
}
|
||||
|
||||
expiredMS := time.Now().UTC().Add(-48 * time.Hour).UnixMilli()
|
||||
records := make([]outbox.Record, 0, 7)
|
||||
expired := make([]string, 0, 5)
|
||||
for index := 0; index < 5; index++ {
|
||||
eventID := fmt.Sprintf("evt-retention-expired-%d", index)
|
||||
expired = append(expired, eventID)
|
||||
records = append(records, newRecord(eventID))
|
||||
}
|
||||
records = append(records, newRecord("evt-retention-fresh"), newRecord("evt-retention-pending"))
|
||||
if err := repository.SaveOutbox(ctx, records); err != nil {
|
||||
t.Fatalf("save outbox failed: %v", err)
|
||||
}
|
||||
for _, eventID := range expired {
|
||||
repository.SetOutboxDeliveredAt(eventID, expiredMS)
|
||||
}
|
||||
if err := repository.MarkOutboxDelivered(ctx, "evt-retention-fresh"); err != nil {
|
||||
t.Fatalf("mark fresh delivered failed: %v", err)
|
||||
}
|
||||
|
||||
// BatchSize=2 强制一轮内分 3 批删完 5 条过期事件，覆盖“整批删满继续、删空退出”的循环边界。
|
||||
total, err := svc.ProcessOutboxRetention(ctx, roomservice.OutboxRetentionOptions{
|
||||
MaxAge: 24 * time.Hour,
|
||||
BatchSize: 2,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("process outbox retention failed: %v", err)
|
||||
}
|
||||
if total != int64(len(expired)) {
|
||||
t.Fatalf("retention deleted %d rows, want %d", total, len(expired))
|
||||
}
|
||||
|
||||
for _, eventID := range expired {
|
||||
if _, ok := repository.OutboxRecord(eventID); ok {
|
||||
t.Fatalf("expired delivered event %s must be purged", eventID)
|
||||
}
|
||||
}
|
||||
if record, ok := repository.OutboxRecord("evt-retention-fresh"); !ok || record.Status != outbox.StatusDelivered {
|
||||
t.Fatalf("fresh delivered event must survive: ok=%v record=%+v", ok, record)
|
||||
}
|
||||
if record, ok := repository.OutboxRecord("evt-retention-pending"); !ok || record.Status != outbox.StatusPending {
|
||||
t.Fatalf("pending event must survive: ok=%v record=%+v", ok, record)
|
||||
}
|
||||
}
|
||||
@ -750,6 +750,8 @@ type OutboxStore interface {
|
||||
MarkOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error
|
||||
// MarkOutboxDead 把超过最大重试次数的事件转入 failed，避免固定污染日志和外部依赖。
|
||||
MarkOutboxDead(ctx context.Context, eventID string, lastErr string) error
|
||||
// DeleteDeliveredOutboxBefore 删除一小批 delivered 且 updated_at_ms 早于 cutoffMS 的历史事件，返回本批删除行数。
|
||||
DeleteDeliveredOutboxBefore(ctx context.Context, cutoffMS int64, limit int) (int64, error)
|
||||
}
|
||||
|
||||
// RoomListStore 管理发现页、Mine 页和关系房间流读模型；它不读取 Room Cell 内存。
|
||||
|
||||
@ -282,6 +282,66 @@ func (r *Repository) MarkOutboxRetryable(ctx context.Context, eventID string, la
|
||||
return err
|
||||
}
|
||||
|
||||
// DeleteDeliveredOutboxBefore 删除一小批 delivered 且 updated_at_ms 早于 cutoffMS 的历史事件，返回本批删除行数。
|
||||
// 只允许清理 delivered：pending/retryable/delivering 仍在投递生命周期内，failed 留给人工补偿排查。
|
||||
// 调用方负责用足够大的保留窗口保护 replay-stat-tz 等直读 room_outbox 历史的重放/补数工具。
|
||||
func (r *Repository) DeleteDeliveredOutboxBefore(ctx context.Context, cutoffMS int64, limit int) (int64, error) {
|
||||
if limit <= 0 {
|
||||
// 默认批量上限避免调用方传 0 导致单事务删除失控。
|
||||
limit = 500
|
||||
}
|
||||
|
||||
// 单表 DELETE 不接受 index hint，先按 retention 索引选出最旧一批主键再删，避免优化器
|
||||
// 选到同前缀的 pending/claim 索引后对全部命中行 filesort。
|
||||
appCode := appcode.FromContext(ctx)
|
||||
rows, err := r.db.QueryContext(ctx,
|
||||
`SELECT event_id FROM room_outbox FORCE INDEX (idx_room_outbox_retention)
|
||||
WHERE app_code = ? AND status = ? AND updated_at_ms < ?
|
||||
ORDER BY updated_at_ms ASC, event_id ASC
|
||||
LIMIT ?`,
|
||||
appCode,
|
||||
outbox.StatusDelivered,
|
||||
cutoffMS,
|
||||
limit,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
eventIDs := make([]string, 0, limit)
|
||||
for rows.Next() {
|
||||
var eventID string
|
||||
if err := rows.Scan(&eventID); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
eventIDs = append(eventIDs, eventID)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if len(eventIDs) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// delivered 是终态不会回流，select 到 delete 之间无需事务；status 条件兜底并发下的重复认领。
|
||||
args := make([]any, 0, len(eventIDs)+2)
|
||||
args = append(args, appCode, outbox.StatusDelivered)
|
||||
for _, eventID := range eventIDs {
|
||||
args = append(args, eventID)
|
||||
}
|
||||
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(eventIDs)), ",")
|
||||
result, err := r.db.ExecContext(ctx,
|
||||
`DELETE FROM room_outbox WHERE app_code = ? AND status = ? AND event_id IN (`+placeholders+`)`,
|
||||
args...,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return result.RowsAffected()
|
||||
}
|
||||
|
||||
// MarkOutboxDead 把超过最大重试次数的事件转入 failed。
|
||||
func (r *Repository) MarkOutboxDead(ctx context.Context, eventID string, lastErr string) error {
|
||||
_, err := r.db.ExecContext(ctx,
|
||||
|
||||
@ -0,0 +1,105 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||||
"hyapp/internal/testutil/mysqlschema"
|
||||
"hyapp/pkg/appcode"
|
||||
"hyapp/services/room-service/internal/room/outbox"
|
||||
)
|
||||
|
||||
func retentionTestRecord(eventID string, createdAtMS int64) outbox.Record {
|
||||
return outbox.Record{
|
||||
AppCode: appcode.Default,
|
||||
EventID: eventID,
|
||||
EventType: "RoomGiftSent",
|
||||
RoomID: "room-retention",
|
||||
Status: outbox.StatusPending,
|
||||
Envelope: &roomeventsv1.EventEnvelope{
|
||||
AppCode: appcode.Default,
|
||||
EventId: eventID,
|
||||
RoomId: "room-retention",
|
||||
EventType: "RoomGiftSent",
|
||||
},
|
||||
CreatedAtMS: createdAtMS,
|
||||
}
|
||||
}
|
||||
|
||||
// 清理只允许命中 delivered 且早于 cutoff 的行；其他状态和窗口内的 delivered 必须原样保留，
|
||||
// 否则统计重放和人工补偿会丢事实。
|
||||
func TestDeleteDeliveredOutboxBeforeOnlyPurgesOldDelivered(t *testing.T) {
|
||||
schema := mysqlschema.New(t, mysqlschema.Config{
|
||||
EnvVar: "ROOM_SERVICE_MYSQL_TEST_DSN",
|
||||
InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_room_service.sql"),
|
||||
DatabasePrefix: "hyapp_room_retention_test",
|
||||
})
|
||||
ctx := context.Background()
|
||||
repo, err := Open(ctx, schema.DSN, 4, 4)
|
||||
if err != nil {
|
||||
t.Fatalf("open repository: %v", err)
|
||||
}
|
||||
defer repo.Close()
|
||||
|
||||
records := []outbox.Record{
|
||||
retentionTestRecord("evt-old-delivered-1", 1000),
|
||||
retentionTestRecord("evt-old-delivered-2", 2000),
|
||||
retentionTestRecord("evt-old-delivered-3", 3000),
|
||||
retentionTestRecord("evt-fresh-delivered", 4000),
|
||||
retentionTestRecord("evt-old-pending", 1000),
|
||||
retentionTestRecord("evt-old-failed", 1000),
|
||||
}
|
||||
if err := repo.SaveOutbox(ctx, records); err != nil {
|
||||
t.Fatalf("save outbox: %v", err)
|
||||
}
|
||||
const cutoffMS = int64(1_000_000)
|
||||
setStatusAt := func(eventID string, status string, updatedAtMS int64) {
|
||||
t.Helper()
|
||||
if _, err := schema.DB.ExecContext(ctx, `UPDATE room_outbox SET status = ?, updated_at_ms = ? WHERE app_code = ? AND event_id = ?`, status, updatedAtMS, appcode.Default, eventID); err != nil {
|
||||
t.Fatalf("rewrite outbox row %s: %v", eventID, err)
|
||||
}
|
||||
}
|
||||
setStatusAt("evt-old-delivered-1", outbox.StatusDelivered, 1000)
|
||||
setStatusAt("evt-old-delivered-2", outbox.StatusDelivered, 2000)
|
||||
setStatusAt("evt-old-delivered-3", outbox.StatusDelivered, 3000)
|
||||
setStatusAt("evt-fresh-delivered", outbox.StatusDelivered, cutoffMS+1)
|
||||
setStatusAt("evt-old-failed", outbox.StatusFailed, 1000)
|
||||
|
||||
// 批量上限必须生效：3 条可删行按 limit=2 分两批删完，第三批确认删空。
|
||||
for step, want := range []int64{2, 1, 0} {
|
||||
deleted, err := repo.DeleteDeliveredOutboxBefore(ctx, cutoffMS, 2)
|
||||
if err != nil {
|
||||
t.Fatalf("delete batch %d: %v", step, err)
|
||||
}
|
||||
if deleted != want {
|
||||
t.Fatalf("delete batch %d deleted %d rows, want %d", step, deleted, want)
|
||||
}
|
||||
}
|
||||
|
||||
var survivors []string
|
||||
rows, err := schema.DB.QueryContext(ctx, `SELECT event_id FROM room_outbox WHERE app_code = ? ORDER BY event_id`, appcode.Default)
|
||||
if err != nil {
|
||||
t.Fatalf("list survivors: %v", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var eventID string
|
||||
if err := rows.Scan(&eventID); err != nil {
|
||||
t.Fatalf("scan survivor: %v", err)
|
||||
}
|
||||
survivors = append(survivors, eventID)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatalf("iterate survivors: %v", err)
|
||||
}
|
||||
want := []string{"evt-fresh-delivered", "evt-old-failed", "evt-old-pending"}
|
||||
if len(survivors) != len(want) {
|
||||
t.Fatalf("survivors mismatch got=%v want=%v", survivors, want)
|
||||
}
|
||||
for index := range want {
|
||||
if survivors[index] != want[index] {
|
||||
t.Fatalf("survivors mismatch got=%v want=%v", survivors, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -627,7 +627,6 @@ func (r *Repository) ensureOutboxRetrySchema(ctx context.Context) error {
|
||||
statement string
|
||||
}{
|
||||
{"idx_room_outbox_pending", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_pending (app_code, status, next_retry_at_ms, created_at_ms, event_id)`},
|
||||
{"idx_room_outbox_retry", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_retry (app_code, status, next_retry_at_ms, created_at_ms, event_id)`},
|
||||
{"idx_room_outbox_claim", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_claim (app_code, status, lock_until_ms, created_at_ms, event_id)`},
|
||||
{"idx_room_outbox_event_created", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_event_created (app_code, event_type, created_at_ms, event_id)`},
|
||||
{"idx_room_outbox_retention", `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_retention (app_code, status, updated_at_ms, event_id)`},
|
||||
@ -646,6 +645,17 @@ func (r *Repository) ensureOutboxRetrySchema(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
}
|
||||
// 历史版本把 idx_room_outbox_retry 建成了和 idx_room_outbox_pending 完全相同的列组合，
|
||||
// 每次写 outbox 都要多维护一份索引；没有任何查询引用它，存在即删除。
|
||||
hasLegacyRetryIndex, err := r.indexExists(ctx, "room_outbox", "idx_room_outbox_retry")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if hasLegacyRetryIndex {
|
||||
if _, err := r.db.ExecContext(ctx, `ALTER TABLE room_outbox DROP INDEX idx_room_outbox_retry`); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@ -37,3 +37,41 @@ func TestMigrateBackfillsRoomOutboxRoomEventIndex(t *testing.T) {
|
||||
t.Fatal("Migrate did not backfill idx_room_outbox_room_event on an existing room_outbox table")
|
||||
}
|
||||
}
|
||||
|
||||
// 旧版本 initdb/ensure 链把 idx_room_outbox_retry 建成了和 idx_room_outbox_pending 完全相同的列组合；
|
||||
// Migrate 必须在 mysql_auto_migrate=true 时把这份重复索引删掉，消除多余写放大。
|
||||
func TestMigrateDropsDuplicateRoomOutboxRetryIndex(t *testing.T) {
|
||||
schema := mysqlschema.New(t, mysqlschema.Config{
|
||||
EnvVar: "ROOM_SERVICE_MYSQL_TEST_DSN",
|
||||
InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_room_service.sql"),
|
||||
DatabasePrefix: "hyapp_room_idx_test",
|
||||
})
|
||||
ctx := context.Background()
|
||||
if _, err := schema.DB.ExecContext(ctx, `ALTER TABLE room_outbox ADD INDEX idx_room_outbox_retry (app_code, status, next_retry_at_ms, created_at_ms, event_id)`); err != nil {
|
||||
t.Fatalf("add index to simulate legacy table: %v", err)
|
||||
}
|
||||
|
||||
repo, err := Open(ctx, schema.DSN, 4, 4)
|
||||
if err != nil {
|
||||
t.Fatalf("open repository: %v", err)
|
||||
}
|
||||
defer repo.Close()
|
||||
if err := repo.Migrate(ctx); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
|
||||
hasIndex, err := repo.indexExists(ctx, "room_outbox", "idx_room_outbox_retry")
|
||||
if err != nil {
|
||||
t.Fatalf("check index: %v", err)
|
||||
}
|
||||
if hasIndex {
|
||||
t.Fatal("Migrate did not drop the duplicate idx_room_outbox_retry index")
|
||||
}
|
||||
hasPending, err := repo.indexExists(ctx, "room_outbox", "idx_room_outbox_pending")
|
||||
if err != nil {
|
||||
t.Fatalf("check pending index: %v", err)
|
||||
}
|
||||
if !hasPending {
|
||||
t.Fatal("idx_room_outbox_pending must survive the duplicate index cleanup")
|
||||
}
|
||||
}
|
||||
|
||||
@ -149,6 +149,21 @@ func (r *Repository) SetRoomCreatedAt(roomID string, createdAtMS int64) {
|
||||
}
|
||||
}
|
||||
|
||||
// SetOutboxDeliveredAt rewrites one outbox row to delivered with a fixed updated_at_ms for retention tests.
|
||||
func (r *Repository) SetOutboxDeliveredAt(eventID string, updatedAtMS int64) {
|
||||
r.t.Helper()
|
||||
|
||||
result, err := r.schema.DB.ExecContext(context.Background(),
|
||||
`UPDATE room_outbox SET status = ?, worker_id = '', lock_until_ms = NULL, updated_at_ms = ? WHERE app_code = ? AND event_id = ?`,
|
||||
outbox.StatusDelivered, updatedAtMS, appcode.Default, eventID)
|
||||
if err != nil {
|
||||
r.t.Fatalf("mark outbox delivered failed: %v", err)
|
||||
}
|
||||
if affected, err := result.RowsAffected(); err != nil || affected != 1 {
|
||||
r.t.Fatalf("outbox row %s not rewritten (affected=%d err=%v)", eventID, affected, err)
|
||||
}
|
||||
}
|
||||
|
||||
// OutboxRecord reads one outbox row, including delivered rows that normal scans skip.
|
||||
func (r *Repository) OutboxRecord(eventID string) (outbox.Record, bool) {
|
||||
r.t.Helper()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user