fix: lucky_gift_outbox 抢占改单 status + SKIP LOCKED，补 delivered 保留期清理

- ClaimPendingLuckyGiftOutbox 把 status IN ('pending','retryable') 拆成单
  status 查询并加 FOR UPDATE SKIP LOCKED：IN 跨 status 会 filesort 全部匹配行
  并逐行加锁，积压时两实例互相阻塞（参照 chatapp3-golang 523cfe9 的修法）；
  ORDER BY 对齐 idx_lucky_gift_outbox_claim_retry/claim_lock 列序，索引序扫描
  LIMIT 提前终止，只锁认领行。pending 留 10% 批次额度给 retryable 防饿死。
- 新增 delivered 行保留期清理 worker（默认开启，保留 30 天、5 分钟一轮、
  单批 500）：线上已积累 15.4 万只增不减的 delivered 行。清理按 app_code
  迭代走 idx_lucky_gift_outbox_retention，只删 delivered 终态。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

services/lucky-gift-service/internal/app/app.go

@@ -113,6 +113,16 @@ func (a *App) Run() error {
 			a.service.RunWorker(ctx, options)
 		})
 	}
+	if a.cfg.LuckyGiftWorker.Retention.Enabled && a.service != nil {
+		retention := a.cfg.LuckyGiftWorker.Retention
+		a.workers.Go(func(ctx context.Context) {
+			a.service.RunOutboxRetentionWorker(ctx, luckygiftservice.OutboxRetentionOptions{
+				ScanInterval: retention.ScanInterval,
+				MaxAge:       retention.MaxAge,
+				BatchSize:    retention.BatchSize,
+			})
+		})
+	}
 	err := a.server.Serve(a.listener)
 	if a.workers != nil {
 		a.workers.StopAndWait()

services/lucky-gift-service/internal/config/config.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"fmt"
 	"strings"
 	"time"
 
@@ -35,6 +36,21 @@ type LuckyGiftWorkerConfig struct {
 	PublishTimeout       time.Duration `yaml:"publish_timeout"`
 	StatsRefreshInterval time.Duration `yaml:"stats_refresh_interval"`
 	StatsBatchSize       int           `yaml:"stats_batch_size"`
+	// Retention 控制 delivered 历史事件的定期清理；它独立于 Enabled，临时关闭补偿投递不会停掉清理。
+	Retention OutboxRetentionConfig `yaml:"retention"`
+}
+
+// OutboxRetentionConfig 控制 lucky_gift_outbox 中 delivered 历史事件的小批量定期删除。
+// 只清理 delivered：pending/retryable/delivering 仍在投递生命周期内，failed 留给人工补偿。
+// 抽奖事实和审计都在 lucky_draw_records，outbox 没有下游读方，默认开启。
+type OutboxRetentionConfig struct {
+	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"`
 }
 
 func Default() Config {
@@ -57,6 +73,12 @@ func Default() Config {
 			PublishTimeout:       5 * time.Second,
 			StatsRefreshInterval: 10 * time.Minute,
 			StatsBatchSize:       5000,
+			Retention: OutboxRetentionConfig{
+				Enabled:      true,
+				MaxAge:       30 * 24 * time.Hour,
+				ScanInterval: 5 * time.Minute,
+				BatchSize:    500,
+			},
 		},
 		Log: logx.Config{
 			Level:           "info",
@@ -122,5 +144,19 @@ func Load(path string) (Config, error) {
 	if cfg.LuckyGiftWorker.StatsBatchSize <= 0 {
 		cfg.LuckyGiftWorker.StatsBatchSize = 5000
 	}
+	defaults := Default().LuckyGiftWorker.Retention
+	if cfg.LuckyGiftWorker.Retention.MaxAge <= 0 {
+		cfg.LuckyGiftWorker.Retention.MaxAge = defaults.MaxAge
+	}
+	if cfg.LuckyGiftWorker.Retention.ScanInterval <= 0 {
+		cfg.LuckyGiftWorker.Retention.ScanInterval = defaults.ScanInterval
+	}
+	if cfg.LuckyGiftWorker.Retention.BatchSize <= 0 {
+		cfg.LuckyGiftWorker.Retention.BatchSize = defaults.BatchSize
+	}
+	if cfg.LuckyGiftWorker.Retention.Enabled && cfg.LuckyGiftWorker.Retention.MaxAge < 24*time.Hour {
+		// 保留窗口低于一天基本可以断定是把单位写错（比如 30 写成 30ns/30s），宁可拒绝启动。
+		return Config{}, fmt.Errorf("lucky_gift_worker retention max_age %s is below the 24h safety floor", cfg.LuckyGiftWorker.Retention.MaxAge)
+	}
 	return cfg, nil
 }

services/lucky-gift-service/internal/config/config_test.go

@@ -0,0 +1,45 @@
+package config
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestLoadDefaultsEnableOutboxRetention(t *testing.T) {
+	cfg, err := Load("")
+	if err != nil {
+		t.Fatalf("Load(empty) failed: %v", err)
+	}
+	retention := cfg.LuckyGiftWorker.Retention
+	if !retention.Enabled {
+		t.Fatal("retention should be enabled by default")
+	}
+	if retention.MaxAge != 30*24*time.Hour {
+		t.Fatalf("retention.MaxAge = %s, want 720h", retention.MaxAge)
+	}
+	if retention.ScanInterval != 5*time.Minute {
+		t.Fatalf("retention.ScanInterval = %s, want 5m", retention.ScanInterval)
+	}
+	if retention.BatchSize != 500 {
+		t.Fatalf("retention.BatchSize = %d, want 500", retention.BatchSize)
+	}
+}
+
+func TestLoadRejectsRetentionMaxAgeBelowSafetyFloor(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "config.yaml")
+	content := `
+lucky_gift_worker:
+  retention:
+    enabled: true
+    max_age: 30s
+`
+	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
+		t.Fatalf("write config: %v", err)
+	}
+	if _, err := Load(path); err == nil || !strings.Contains(err.Error(), "safety floor") {
+		t.Fatalf("Load should reject max_age below 24h, got err=%v", err)
+	}
+}

services/lucky-gift-service/internal/service/luckygift/outbox_retention.go

@@ -0,0 +1,110 @@
+package luckygift
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"hyapp/pkg/logx"
+)
+
+// outboxRetentionBatchPause 是同一轮内两批 DELETE 之间的让路间隔，避免清理积压时长时间占用行锁和 IO。
+const outboxRetentionBatchPause = 100 * time.Millisecond
+
+// OutboxRetentionOptions 控制 lucky_gift_outbox 中 delivered 历史事件的定期清理循环。
+type OutboxRetentionOptions struct {
+	// ScanInterval 是清空一轮可删事件后到下一轮扫描的间隔；非正数会被归一到安全默认值。
+	ScanInterval time.Duration
+	// MaxAge 是 delivered 事件按 updated_at_ms 计算的保留时长。
+	MaxAge time.Duration
+	// BatchSize 是单条 DELETE 的行数上限。
+	BatchSize int
+}
+
+func defaultOutboxRetentionOptions() OutboxRetentionOptions {
+	// 默认值和 config.Default 的 retention 保持一致，防止手动构造绕过配置归一化后误删新事件。
+	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 且超过保留窗口的 lucky_gift_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, "lucky_gift_outbox_retention_failed", slog.String("error", err.Error()))
+	}
+}
+
+// ProcessOutboxRetention 执行一轮清理：按批删除到没有可删事件为止，返回本轮总删除行数。
+func (s *Service) ProcessOutboxRetention(ctx context.Context, options OutboxRetentionOptions) (int64, error) {
+	if err := s.requireRepository(); err != nil {
+		return 0, err
+	}
+	options = normalizeOutboxRetentionOptions(options)
+	cutoffMS := s.now().UTC().Add(-options.MaxAge).UnixMilli()
+
+	var total int64
+	for {
+		deleted, err := s.repository.DeleteDeliveredLuckyGiftOutboxBefore(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, "lucky_gift_outbox_retention_purged",
+			slog.Int64("rows_deleted", total),
+			slog.Int64("cutoff_ms", cutoffMS),
+		)
+	}
+	return total, nil
+}

services/lucky-gift-service/internal/service/luckygift/outbox_retention_test.go

@@ -0,0 +1,23 @@
+package luckygift
+
+import (
+	"testing"
+	"time"
+)
+
+func TestNormalizeOutboxRetentionOptionsFillsSafeDefaults(t *testing.T) {
+	got := normalizeOutboxRetentionOptions(OutboxRetentionOptions{})
+	want := defaultOutboxRetentionOptions()
+	if got != want {
+		t.Fatalf("normalizeOutboxRetentionOptions(zero) = %+v, want %+v", got, want)
+	}
+
+	custom := OutboxRetentionOptions{
+		ScanInterval: time.Minute,
+		MaxAge:       7 * 24 * time.Hour,
+		BatchSize:    100,
+	}
+	if got := normalizeOutboxRetentionOptions(custom); got != custom {
+		t.Fatalf("normalizeOutboxRetentionOptions(custom) = %+v, want unchanged %+v", got, custom)
+	}
+}

services/lucky-gift-service/internal/service/luckygift/service.go

@@ -35,6 +35,7 @@ type Repository interface {
 	ListLuckyGiftPoolBalances(ctx context.Context, query domain.PoolBalanceQuery) ([]domain.PoolBalance, error)
 	GetLuckyGiftDrawRewardState(ctx context.Context, appCode string, drawIDs []string) (domain.DrawRewardState, error)
 	ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int) ([]domain.DrawOutbox, error)
+	DeleteDeliveredLuckyGiftOutboxBefore(ctx context.Context, cutoffMS int64, limit int) (int64, error)
 	MarkLuckyGiftOutboxDelivered(ctx context.Context, event domain.DrawOutbox, nowMS int64) error
 	MarkLuckyGiftOutboxRetryable(ctx context.Context, event domain.DrawOutbox, retryCount int, nextRetryAtMS int64, lastErr string, nowMS int64) error
 	MarkLuckyGiftOutboxFailed(ctx context.Context, event domain.DrawOutbox, retryCount int, lastErr string, nowMS int64) error

services/lucky-gift-service/internal/storage/mysql/lucky_gift_outbox_claim_test.go

@@ -0,0 +1,25 @@
+package mysql
+
+import "testing"
+
+func TestLuckyGiftPendingClaimLimitReservesRetryableQuota(t *testing.T) {
+	tests := []struct {
+		name      string
+		batchSize int
+		want      int
+	}{
+		{name: "empty", batchSize: 0, want: 0},
+		{name: "single slot goes to pending", batchSize: 1, want: 1},
+		{name: "small batch keeps one retryable slot", batchSize: 4, want: 3},
+		{name: "default batch reserves ten percent", batchSize: 100, want: 90},
+		{name: "max batch reserves ten percent", batchSize: 500, want: 450},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := luckyGiftPendingClaimLimit(tt.batchSize); got != tt.want {
+				t.Fatalf("luckyGiftPendingClaimLimit(%d) = %d, want %d", tt.batchSize, got, tt.want)
+			}
+		})
+	}
+}

services/lucky-gift-service/internal/storage/mysql/lucky_gift_repository.go

@@ -1393,17 +1393,42 @@ func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID s
 		domain.EventTypeExternalLuckyGiftDrawn,
 	}
 	for _, eventType := range priorityTypes {
-		candidates, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, `
+		candidates := make([]domain.DrawOutbox, 0, batchSize)
-			SELECT app_code, outbox_id, event_type, COALESCE(CAST(payload AS CHAR), '{}'), retry_count, created_at_ms
+		// pending 和 retryable 分开抢占：IN 跨 status 会让索引无法消除排序，积压时 filesort 并
-			FROM lucky_gift_outbox
+		// FOR UPDATE 锁住全部匹配行，多实例互相阻塞。单 status 查询按 idx_lucky_gift_outbox_claim_retry
-			WHERE event_type = ?
+		// (event_type, status, next_retry_at_ms, created_at_ms, outbox_id) 的列序扫描，LIMIT 提前终止；
-			  AND status IN ('pending', 'retryable')
+		// pending 只拿部分额度，给 retryable 保留重试配额避免洪峰饿死补偿。
-			  AND next_retry_at_ms <= ?
+		for _, claim := range []struct {
-			ORDER BY created_at_ms ASC, outbox_id ASC
+			status string
-			LIMIT ?
+			limit  int
-			FOR UPDATE`, batchSize, eventType, nowMS, batchSize)
+		}{
-		if err != nil {
+			{status: "pending", limit: luckyGiftPendingClaimLimit(batchSize)},
-			return nil, err
+			{status: "retryable", limit: batchSize},
+		} {
+			remaining := batchSize - len(candidates)
+			if remaining <= 0 {
+				break
+			}
+			limit := claim.limit
+			if limit > remaining {
+				limit = remaining
+			}
+			if limit <= 0 {
+				continue
+			}
+			statusRows, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, `
+				SELECT app_code, outbox_id, event_type, COALESCE(CAST(payload AS CHAR), '{}'), retry_count, created_at_ms
+				FROM lucky_gift_outbox
+				WHERE event_type = ?
+				  AND status = ?
+				  AND next_retry_at_ms <= ?
+				ORDER BY next_retry_at_ms ASC, created_at_ms ASC, outbox_id ASC
+				LIMIT ?
+				FOR UPDATE SKIP LOCKED`, limit, eventType, claim.status, nowMS, limit)
+			if err != nil {
+				return nil, err
+			}
+			candidates = append(candidates, statusRows...)
 		}
 		if len(candidates) > 0 {
 			events = append(events, candidates...)
@@ -1412,15 +1437,17 @@ func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID s
 	}
 	if len(events) == 0 {
 		for _, eventType := range priorityTypes {
+			// 排序对齐 idx_lucky_gift_outbox_claim_lock (event_type, status, lock_until_ms, created_at_ms, outbox_id)，
+			// SKIP LOCKED 让两实例的超时接管互不阻塞。
 			retryEvents, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, `
 				SELECT app_code, outbox_id, event_type, COALESCE(CAST(payload AS CHAR), '{}'), retry_count, created_at_ms
 				FROM lucky_gift_outbox
 				WHERE event_type = ?
 				  AND status = 'delivering'
 				  AND lock_until_ms <= ?
-				ORDER BY created_at_ms ASC, outbox_id ASC
+				ORDER BY lock_until_ms ASC, created_at_ms ASC, outbox_id ASC
 				LIMIT ?
-				FOR UPDATE`, batchSize, eventType, nowMS, batchSize)
+				FOR UPDATE SKIP LOCKED`, batchSize, eventType, nowMS, batchSize)
 			if err != nil {
 				return nil, err
 			}
@@ -1447,6 +1474,18 @@ func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID s
 	return events, nil
 }
 
+// luckyGiftPendingClaimLimit 是单轮批次里 pending 可占用的额度上限，剩余部分留给 retryable。
+func luckyGiftPendingClaimLimit(batchSize int) int {
+	if batchSize <= 1 {
+		return batchSize
+	}
+	reserve := batchSize / 10
+	if reserve < 1 {
+		reserve = 1
+	}
+	return batchSize - reserve
+}
+
 func (r *Repository) queryLuckyGiftOutboxCandidates(ctx context.Context, tx *sql.Tx, query string, capacity int, args ...any) ([]domain.DrawOutbox, error) {
 	rows, err := tx.QueryContext(ctx, query, args...)
 	if err != nil {
@@ -1504,6 +1543,94 @@ func (r *Repository) MarkLuckyGiftOutboxFailed(ctx context.Context, event domain
 	return err
 }
 
+// DeleteDeliveredLuckyGiftOutboxBefore 删除一小批 delivered 且 updated_at_ms 早于 cutoffMS 的历史事件，返回本批删除总行数。
+// 只清理 delivered 终态：pending/retryable/delivering 仍在投递生命周期内，failed 留给人工补偿排查。
+// 抽奖事实在 lucky_draw_records，outbox 仅承载补偿投递，delivered 行没有下游读方。
+func (r *Repository) DeleteDeliveredLuckyGiftOutboxBefore(ctx context.Context, cutoffMS int64, limit int) (int64, error) {
+	if r == nil || r.db == nil {
+		return 0, xerr.New(xerr.Unavailable, "mysql repository is not configured")
+	}
+	if limit <= 0 {
+		// 默认批量上限避免调用方传 0 导致单事务删除失控。
+		limit = 500
+	}
+	// 清理跨 App 而 idx_lucky_gift_outbox_retention 以 app_code 开头；先用主键前缀枚举 app_code，
+	// 再按 App 走 retention 索引定位最旧一批主键后删除，避免优化器退化成全索引扫描。
+	appCodeRows, err := r.db.QueryContext(ctx, `SELECT DISTINCT app_code FROM lucky_gift_outbox`)
+	if err != nil {
+		return 0, err
+	}
+	appCodes := make([]string, 0, 4)
+	for appCodeRows.Next() {
+		var appCode string
+		if err := appCodeRows.Scan(&appCode); err != nil {
+			_ = appCodeRows.Close()
+			return 0, err
+		}
+		appCodes = append(appCodes, appCode)
+	}
+	if err := appCodeRows.Err(); err != nil {
+		_ = appCodeRows.Close()
+		return 0, err
+	}
+	_ = appCodeRows.Close()
+
+	var total int64
+	for _, appCode := range appCodes {
+		remaining := limit - int(total)
+		if remaining <= 0 {
+			break
+		}
+		outboxIDs, err := r.listDeliveredLuckyGiftOutboxIDs(ctx, appCode, cutoffMS, remaining)
+		if err != nil {
+			return total, err
+		}
+		if len(outboxIDs) == 0 {
+			continue
+		}
+		// delivered 是终态不会回流，select 到 delete 之间无需事务；status 条件兜底并发下的重复认领。
+		args := make([]any, 0, len(outboxIDs)+1)
+		args = append(args, appCode)
+		for _, outboxID := range outboxIDs {
+			args = append(args, outboxID)
+		}
+		result, err := r.db.ExecContext(ctx,
+			`DELETE FROM lucky_gift_outbox WHERE app_code = ? AND status = 'delivered' AND outbox_id IN (`+luckySQLPlaceholders(len(outboxIDs))+`)`,
+			args...,
+		)
+		if err != nil {
+			return total, err
+		}
+		deleted, err := result.RowsAffected()
+		if err != nil {
+			return total, err
+		}
+		total += deleted
+	}
+	return total, nil
+}
+
+func (r *Repository) listDeliveredLuckyGiftOutboxIDs(ctx context.Context, appCode string, cutoffMS int64, limit int) ([]string, error) {
+	rows, err := r.db.QueryContext(ctx, `
+		SELECT outbox_id FROM lucky_gift_outbox FORCE INDEX (idx_lucky_gift_outbox_retention)
+		WHERE app_code = ? AND status = 'delivered' AND updated_at_ms < ?
+		ORDER BY updated_at_ms ASC, outbox_id ASC
+		LIMIT ?`, appCode, cutoffMS, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	outboxIDs := make([]string, 0, limit)
+	for rows.Next() {
+		var outboxID string
+		if err := rows.Scan(&outboxID); err != nil {
+			return nil, err
+		}
+		outboxIDs = append(outboxIDs, outboxID)
+	}
+	return outboxIDs, rows.Err()
+}
+
 func (r *Repository) MarkLuckyGiftDrawGranted(ctx context.Context, appCode string, drawID string, walletTransactionID string, nowMS int64) error {
 	return r.MarkLuckyGiftDrawsGranted(ctx, appCode, []string{drawID}, walletTransactionID, nowMS)
 }