From ec35fcad5e0e0d36bb797b7800da529ac26c984e Mon Sep 17 00:00:00 2001 From: zhx Date: Sat, 4 Jul 2026 01:12:25 +0800 Subject: [PATCH] =?UTF-8?q?room=5Foutbox=20=E5=A2=9E=E5=8A=A0=20delivered?= =?UTF-8?q?=20=E5=8E=86=E5=8F=B2=E4=BA=8B=E4=BB=B6=E5=AE=9A=E6=9C=9F?= =?UTF-8?q?=E6=B8=85=E7=90=86=E5=B9=B6=E7=A7=BB=E9=99=A4=E9=87=8D=E5=A4=8D?= =?UTF-8?q?=E7=B4=A2=E5=BC=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- .../configs/config.tencent.example.yaml | 8 ++ .../deploy/mysql/initdb/001_room_service.sql | 1 - services/room-service/internal/app/app.go | 10 ++ .../room-service/internal/config/config.go | 37 ++++++ .../internal/config/config_test.go | 28 +++++ .../internal/room/service/outbox_retention.go | 108 ++++++++++++++++++ .../room/service/outbox_retention_test.go | 86 ++++++++++++++ .../internal/room/service/repository.go | 2 + .../internal/storage/mysql/outbox.go | 60 ++++++++++ .../storage/mysql/outbox_retention_test.go | 105 +++++++++++++++++ .../internal/storage/mysql/schema.go | 12 +- .../storage/mysql/schema_outbox_index_test.go | 38 ++++++ .../internal/testutil/mysqltest/mysqltest.go | 15 +++ 13 files changed, 508 insertions(+), 2 deletions(-) create mode 100644 services/room-service/internal/room/service/outbox_retention.go create mode 100644 services/room-service/internal/room/service/outbox_retention_test.go create mode 100644 services/room-service/internal/storage/mysql/outbox_retention_test.go diff --git a/services/room-service/configs/config.tencent.example.yaml b/services/room-service/configs/config.tencent.example.yaml index 2ab732e6..ab5c0a31 100644 --- a/services/room-service/configs/config.tencent.example.yaml +++ b/services/room-service/configs/config.tencent.example.yaml @@ -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 diff --git a/services/room-service/deploy/mysql/initdb/001_room_service.sql b/services/room-service/deploy/mysql/initdb/001_room_service.sql index 18b9c951..56481a3c 100644 --- a/services/room-service/deploy/mysql/initdb/001_room_service.sql +++ b/services/room-service/deploy/mysql/initdb/001_room_service.sql @@ -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), diff --git a/services/room-service/internal/app/app.go b/services/room-service/internal/app/app.go index b38e926e..a7999ffa 100644 --- a/services/room-service/internal/app/app.go +++ b/services/room-service/internal/app/app.go @@ -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) }) diff --git a/services/room-service/internal/config/config.go b/services/room-service/internal/config/config.go index 56ccafff..23335916 100644 --- a/services/room-service/internal/config/config.go +++ b/services/room-service/internal/config/config.go @@ -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 } diff --git a/services/room-service/internal/config/config_test.go b/services/room-service/internal/config/config_test.go index 4667a034..d6d82645 100644 --- a/services/room-service/internal/config/config_test.go +++ b/services/room-service/internal/config/config_test.go @@ -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") + } +} diff --git a/services/room-service/internal/room/service/outbox_retention.go b/services/room-service/internal/room/service/outbox_retention.go new file mode 100644 index 00000000..adb7432e --- /dev/null +++ b/services/room-service/internal/room/service/outbox_retention.go @@ -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 +} diff --git a/services/room-service/internal/room/service/outbox_retention_test.go b/services/room-service/internal/room/service/outbox_retention_test.go new file mode 100644 index 00000000..ac45ddc2 --- /dev/null +++ b/services/room-service/internal/room/service/outbox_retention_test.go @@ -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) + } +} diff --git a/services/room-service/internal/room/service/repository.go b/services/room-service/internal/room/service/repository.go index 9a7d50f3..212a12ba 100644 --- a/services/room-service/internal/room/service/repository.go +++ b/services/room-service/internal/room/service/repository.go @@ -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 内存。 diff --git a/services/room-service/internal/storage/mysql/outbox.go b/services/room-service/internal/storage/mysql/outbox.go index 16f94a13..5337173f 100644 --- a/services/room-service/internal/storage/mysql/outbox.go +++ b/services/room-service/internal/storage/mysql/outbox.go @@ -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, diff --git a/services/room-service/internal/storage/mysql/outbox_retention_test.go b/services/room-service/internal/storage/mysql/outbox_retention_test.go new file mode 100644 index 00000000..124d2d51 --- /dev/null +++ b/services/room-service/internal/storage/mysql/outbox_retention_test.go @@ -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) + } + } +} diff --git a/services/room-service/internal/storage/mysql/schema.go b/services/room-service/internal/storage/mysql/schema.go index 2659153e..8152ac32 100644 --- a/services/room-service/internal/storage/mysql/schema.go +++ b/services/room-service/internal/storage/mysql/schema.go @@ -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 } diff --git a/services/room-service/internal/storage/mysql/schema_outbox_index_test.go b/services/room-service/internal/storage/mysql/schema_outbox_index_test.go index dc6bfbbe..d85f4b1b 100644 --- a/services/room-service/internal/storage/mysql/schema_outbox_index_test.go +++ b/services/room-service/internal/storage/mysql/schema_outbox_index_test.go @@ -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") + } +} diff --git a/services/room-service/internal/testutil/mysqltest/mysqltest.go b/services/room-service/internal/testutil/mysqltest/mysqltest.go index 0425fdde..944ff4d3 100644 --- a/services/room-service/internal/testutil/mysqltest/mysqltest.go +++ b/services/room-service/internal/testutil/mysqltest/mysqltest.go @@ -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()