From 815275676ee8d7aac5203a9471620a841111895d Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 1 Jun 2026 12:38:30 +0800 Subject: [PATCH] fix: reduce outbox worker mysql scans --- .../configs/config.tencent.example.yaml | 4 +- .../mysql/initdb/001_activity_service.sql | 4 +- .../storage/mysql/lucky_gift_repository.go | 55 +++++++++----- .../internal/storage/mysql/repository.go | 26 +++++++ .../configs/config.tencent.example.yaml | 4 +- .../deploy/mysql/initdb/001_game_service.sql | 1 + .../internal/storage/mysql/repository.go | 36 ++++++--- .../wallet-service/configs/config.docker.yaml | 5 ++ .../configs/config.tencent.example.yaml | 5 ++ services/wallet-service/configs/config.yaml | 5 ++ .../mysql/initdb/001_wallet_service.sql | 21 ++++++ services/wallet-service/internal/app/app.go | 19 +++-- .../wallet-service/internal/config/config.go | 28 +++++++ .../storage/mysql/badge_projection_relay.go | 74 +++++++++++++++---- .../storage/mysql/gift_wall_projection.go | 71 ++++++++++++++---- 15 files changed, 292 insertions(+), 66 deletions(-) diff --git a/services/activity-service/configs/config.tencent.example.yaml b/services/activity-service/configs/config.tencent.example.yaml index c186e4ca..1687a4b2 100644 --- a/services/activity-service/configs/config.tencent.example.yaml +++ b/services/activity-service/configs/config.tencent.example.yaml @@ -19,8 +19,8 @@ red_packet_broadcast_worker: enabled: true lucky_gift_worker: enabled: true - worker_poll_interval: "1s" - worker_batch_size: 100 + worker_poll_interval: "60s" + worker_batch_size: 10 worker_concurrency: 4 worker_lock_ttl: "30s" worker_max_retry: 8 diff --git a/services/activity-service/deploy/mysql/initdb/001_activity_service.sql b/services/activity-service/deploy/mysql/initdb/001_activity_service.sql index 8ac41c96..93f339d3 100644 --- a/services/activity-service/deploy/mysql/initdb/001_activity_service.sql +++ b/services/activity-service/deploy/mysql/initdb/001_activity_service.sql @@ -44,7 +44,9 @@ CREATE TABLE IF NOT EXISTS activity_outbox ( created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', PRIMARY KEY (app_code, outbox_id), - KEY idx_activity_outbox_status_retry (app_code, status, next_retry_at_ms) + KEY idx_activity_outbox_status_retry (app_code, status, next_retry_at_ms), + KEY idx_activity_outbox_lucky_retry (app_code, event_type, status, next_retry_at_ms, created_at_ms, outbox_id), + KEY idx_activity_outbox_lucky_lock (app_code, event_type, status, lock_until_ms, created_at_ms, outbox_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='活动事件 outbox 表'; CREATE TABLE IF NOT EXISTS lucky_gift_rules ( diff --git a/services/activity-service/internal/storage/mysql/lucky_gift_repository.go b/services/activity-service/internal/storage/mysql/lucky_gift_repository.go index 3591d6a4..55f971e3 100644 --- a/services/activity-service/internal/storage/mysql/lucky_gift_repository.go +++ b/services/activity-service/internal/storage/mysql/lucky_gift_repository.go @@ -1359,31 +1359,35 @@ func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID s } defer func() { _ = tx.Rollback() }() - rows, err := tx.QueryContext(ctx, ` + appCode := appcode.FromContext(ctx) + events, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, ` SELECT app_code, outbox_id, event_type, COALESCE(CAST(payload AS CHAR), '{}'), retry_count, created_at_ms - FROM activity_outbox - WHERE event_type = 'LuckyGiftDrawn' - AND ( - (status IN ('pending', 'retryable') AND next_retry_at_ms <= ?) - OR (status = 'delivering' AND lock_until_ms <= ?) - ) + FROM activity_outbox FORCE INDEX (idx_activity_outbox_lucky_retry) + WHERE app_code = ? + AND event_type = 'LuckyGiftDrawn' + AND status IN ('pending', 'retryable') + AND next_retry_at_ms <= ? ORDER BY created_at_ms ASC, outbox_id ASC LIMIT ? - FOR UPDATE`, nowMS, nowMS, batchSize) + FOR UPDATE`, batchSize, appCode, nowMS, batchSize) if err != nil { return nil, err } - defer rows.Close() - events := make([]domain.DrawOutbox, 0, batchSize) - for rows.Next() { - var event domain.DrawOutbox - if err := rows.Scan(&event.AppCode, &event.OutboxID, &event.EventType, &event.PayloadJSON, &event.RetryCount, &event.CreatedAtMS); err != nil { + if remaining := batchSize - len(events); remaining > 0 { + retryEvents, err := r.queryLuckyGiftOutboxCandidates(ctx, tx, ` + SELECT app_code, outbox_id, event_type, COALESCE(CAST(payload AS CHAR), '{}'), retry_count, created_at_ms + FROM activity_outbox FORCE INDEX (idx_activity_outbox_lucky_lock) + WHERE app_code = ? + AND event_type = 'LuckyGiftDrawn' + AND status = 'delivering' + AND lock_until_ms <= ? + ORDER BY created_at_ms ASC, outbox_id ASC + LIMIT ? + FOR UPDATE`, remaining, appCode, nowMS, remaining) + if err != nil { return nil, err } - events = append(events, event) - } - if err := rows.Err(); err != nil { - return nil, err + events = append(events, retryEvents...) } lockUntilMS := nowMS + lockTTL.Milliseconds() for _, event := range events { @@ -1402,6 +1406,23 @@ func (r *Repository) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID s return events, nil } +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 { + return nil, err + } + defer rows.Close() + events := make([]domain.DrawOutbox, 0, capacity) + for rows.Next() { + var event domain.DrawOutbox + if err := rows.Scan(&event.AppCode, &event.OutboxID, &event.EventType, &event.PayloadJSON, &event.RetryCount, &event.CreatedAtMS); err != nil { + return nil, err + } + events = append(events, event) + } + return events, rows.Err() +} + func (r *Repository) MarkLuckyGiftOutboxDelivered(ctx context.Context, event domain.DrawOutbox, nowMS int64) error { if r == nil || r.db == nil { return xerr.New(xerr.Unavailable, "mysql repository is not configured") diff --git a/services/activity-service/internal/storage/mysql/repository.go b/services/activity-service/internal/storage/mysql/repository.go index da7eb6af..c645e3f9 100644 --- a/services/activity-service/internal/storage/mysql/repository.go +++ b/services/activity-service/internal/storage/mysql/repository.go @@ -48,6 +48,9 @@ func (r *Repository) Migrate(ctx context.Context) error { if err := r.ensureLuckyGiftOutboxColumns(ctx); err != nil { return err } + if err := r.ensureLuckyGiftOutboxIndexes(ctx); err != nil { + return err + } if err := r.ensureLuckyDrawRewardSettlementColumns(ctx); err != nil { return err } @@ -299,6 +302,29 @@ func (r *Repository) ensureLuckyGiftOutboxColumns(ctx context.Context) error { return nil } +func (r *Repository) ensureLuckyGiftOutboxIndexes(ctx context.Context) error { + // LuckyGiftDrawn worker 的 pending 和 lock-expired 两条路径分开查，索引列顺序要和各自 WHERE 匹配。 + additions := []struct { + name string + sql string + }{ + {"idx_activity_outbox_lucky_retry", `CREATE INDEX idx_activity_outbox_lucky_retry ON activity_outbox (app_code, event_type, status, next_retry_at_ms, created_at_ms, outbox_id)`}, + {"idx_activity_outbox_lucky_lock", `CREATE INDEX idx_activity_outbox_lucky_lock ON activity_outbox (app_code, event_type, status, lock_until_ms, created_at_ms, outbox_id)`}, + } + for _, addition := range additions { + exists, err := r.indexExists(ctx, "activity_outbox", addition.name) + if err != nil { + return err + } + if !exists { + if _, err := r.db.ExecContext(ctx, addition.sql); err != nil { + return err + } + } + } + return nil +} + func (r *Repository) ensureLuckyDrawRewardSettlementColumns(ctx context.Context) error { additions := []struct { name string diff --git a/services/game-service/configs/config.tencent.example.yaml b/services/game-service/configs/config.tencent.example.yaml index e8351cbc..e66fc900 100644 --- a/services/game-service/configs/config.tencent.example.yaml +++ b/services/game-service/configs/config.tencent.example.yaml @@ -23,8 +23,8 @@ rocketmq: producer_group: "hyapp-game-outbox-producer" outbox_worker: enabled: true - poll_interval: "1s" - batch_size: 100 + poll_interval: "60s" + batch_size: 10 publish_timeout: "3s" log: level: info diff --git a/services/game-service/deploy/mysql/initdb/001_game_service.sql b/services/game-service/deploy/mysql/initdb/001_game_service.sql index 655fa794..25c18f5f 100644 --- a/services/game-service/deploy/mysql/initdb/001_game_service.sql +++ b/services/game-service/deploy/mysql/initdb/001_game_service.sql @@ -167,6 +167,7 @@ CREATE TABLE IF NOT EXISTS game_level_event_outbox ( PRIMARY KEY(app_code, event_id), UNIQUE KEY uk_game_level_event_order(app_code, order_id), KEY idx_game_level_event_pending(app_code, status, next_retry_at_ms, created_at_ms), + KEY idx_game_level_event_claim(app_code, status, locked_until_ms, next_retry_at_ms, created_at_ms, event_id), KEY idx_game_level_event_user(app_code, user_id, created_at_ms) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='游戏等级事件 outbox 表'; diff --git a/services/game-service/internal/storage/mysql/repository.go b/services/game-service/internal/storage/mysql/repository.go index 79de8bd6..244a7a72 100644 --- a/services/game-service/internal/storage/mysql/repository.go +++ b/services/game-service/internal/storage/mysql/repository.go @@ -214,6 +214,7 @@ func (r *Repository) Migrate(ctx context.Context) error { PRIMARY KEY(app_code, event_id), UNIQUE KEY uk_game_level_event_order(app_code, order_id), KEY idx_game_level_event_pending(app_code, status, next_retry_at_ms, created_at_ms), + KEY idx_game_level_event_claim(app_code, status, locked_until_ms, next_retry_at_ms, created_at_ms, event_id), KEY idx_game_level_event_user(app_code, user_id, created_at_ms) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS game_outbox ( @@ -749,20 +750,28 @@ func (r *Repository) ClaimPendingLevelEvents(ctx context.Context, workerID strin return nil, err } defer func() { _ = tx.Rollback() }() - rows, err := tx.QueryContext(ctx, levelEventOutboxSelectSQL()+` + + appCode := appcode.FromContext(ctx) + events, err := r.queryPendingLevelEvents(ctx, tx, ` WHERE app_code = ? AND status IN ('pending', 'failed') AND next_retry_at_ms <= ? - AND (locked_until_ms = 0 OR locked_until_ms <= ?) + AND locked_until_ms = 0 ORDER BY created_at_ms ASC, event_id ASC LIMIT ? FOR UPDATE SKIP LOCKED`, - appcode.FromContext(ctx), nowMS, nowMS, batchSize, - ) + appCode, nowMS, batchSize) if err != nil { return nil, err } - events, err := scanLevelEventOutboxRows(rows) - _ = rows.Close() - if err != nil { - return nil, err + if remaining := batchSize - len(events); remaining > 0 { + retryEvents, err := r.queryPendingLevelEvents(ctx, tx, ` + WHERE app_code = ? AND status IN ('pending', 'failed') AND next_retry_at_ms <= ? + AND locked_until_ms > 0 AND locked_until_ms <= ? + ORDER BY created_at_ms ASC, event_id ASC + LIMIT ? FOR UPDATE SKIP LOCKED`, + appCode, nowMS, nowMS, remaining) + if err != nil { + return nil, err + } + events = append(events, retryEvents...) } lockUntil := nowMS + lockTTL.Milliseconds() for _, event := range events { @@ -781,6 +790,15 @@ func (r *Repository) ClaimPendingLevelEvents(ctx context.Context, workerID strin return events, nil } +func (r *Repository) queryPendingLevelEvents(ctx context.Context, tx *sql.Tx, whereSQL string, args ...any) ([]gamedomain.LevelEventOutbox, error) { + rows, err := tx.QueryContext(ctx, levelEventOutboxSelectSQL()+whereSQL, args...) + if err != nil { + return nil, err + } + defer rows.Close() + return scanLevelEventOutboxRows(rows) +} + func (r *Repository) MarkLevelEventDelivered(ctx context.Context, eventID string, nowMS int64) error { if r == nil || r.db == nil { return xerr.New(xerr.Unavailable, "mysql repository is not configured") @@ -1052,7 +1070,7 @@ func levelEventOutboxSelectSQL() string { provider_round_id, coin_amount, wallet_transaction_id, COALESCE(CAST(payload_json AS CHAR), '{}'), status, attempt_count, failure_reason, occurred_at_ms, created_at_ms, updated_at_ms - FROM game_level_event_outbox ` + FROM game_level_event_outbox FORCE INDEX (idx_game_level_event_claim) ` } func scanCatalog(scanner catalogScanner) (gamedomain.CatalogItem, error) { diff --git a/services/wallet-service/configs/config.docker.yaml b/services/wallet-service/configs/config.docker.yaml index ae296fd2..eaf79a1f 100644 --- a/services/wallet-service/configs/config.docker.yaml +++ b/services/wallet-service/configs/config.docker.yaml @@ -17,6 +17,11 @@ red_packet_expiry_worker: app_code: "lalu" poll_interval: "5s" batch_size: 50 +projection_worker: + enabled: true + poll_interval: "60s" + batch_size: 10 + lock_ttl: "30s" rocketmq: # Docker 本地使用 compose RocketMQ 发布 wallet_outbox 事实。 enabled: true diff --git a/services/wallet-service/configs/config.tencent.example.yaml b/services/wallet-service/configs/config.tencent.example.yaml index 967d7a30..1c9e5935 100644 --- a/services/wallet-service/configs/config.tencent.example.yaml +++ b/services/wallet-service/configs/config.tencent.example.yaml @@ -17,6 +17,11 @@ red_packet_expiry_worker: app_code: "lalu" poll_interval: "5s" batch_size: 50 +projection_worker: + enabled: true + poll_interval: "60s" + batch_size: 10 + lock_ttl: "30s" rocketmq: enabled: true name_servers: diff --git a/services/wallet-service/configs/config.yaml b/services/wallet-service/configs/config.yaml index cc0cecb3..0ebeff83 100644 --- a/services/wallet-service/configs/config.yaml +++ b/services/wallet-service/configs/config.yaml @@ -17,6 +17,11 @@ red_packet_expiry_worker: app_code: "lalu" poll_interval: "5s" batch_size: 50 +projection_worker: + enabled: true + poll_interval: "60s" + batch_size: 10 + lock_ttl: "30s" rocketmq: # 本地默认关闭；线上由 wallet-service outbox worker 投递账务事实到 MQ。 enabled: false diff --git a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql index 6572be03..b61d16cf 100644 --- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql +++ b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql @@ -75,6 +75,8 @@ CREATE TABLE IF NOT EXISTS wallet_outbox ( PRIMARY KEY (app_code, event_id), KEY idx_wallet_outbox_pending (app_code, status, next_retry_at_ms, created_at_ms), KEY idx_wallet_outbox_claim (app_code, status, lock_until_ms, created_at_ms), + KEY idx_wallet_outbox_event_created (app_code, event_type, created_at_ms, event_id), + KEY idx_wallet_outbox_asset_event_created (app_code, asset_type, event_type, created_at_ms, event_id), KEY idx_wallet_outbox_tx (app_code, transaction_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='钱包事件 outbox 表'; @@ -338,6 +340,25 @@ PREPARE stmt FROM @ddl; EXECUTE stmt; DEALLOCATE PREPARE stmt; +-- projection worker 只扫描单租户、单事件类型的 wallet_outbox；缺少该索引会退化成百万级全表扫描。 +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'wallet_outbox' AND INDEX_NAME = 'idx_wallet_outbox_event_created') = 0, + 'ALTER TABLE wallet_outbox ADD INDEX idx_wallet_outbox_event_created (app_code, event_type, created_at_ms, event_id)', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl := IF( + (SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'wallet_outbox' AND INDEX_NAME = 'idx_wallet_outbox_asset_event_created') = 0, + 'ALTER TABLE wallet_outbox ADD INDEX idx_wallet_outbox_asset_event_created (app_code, asset_type, event_type, created_at_ms, event_id)', + 'SELECT 1' +); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + CREATE TABLE IF NOT EXISTS wallet_gift_prices ( app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码，用于多租户隔离', gift_id VARCHAR(96) NOT NULL COMMENT '礼物 ID', diff --git a/services/wallet-service/internal/app/app.go b/services/wallet-service/internal/app/app.go index 7cdcb2a5..8290ced1 100644 --- a/services/wallet-service/internal/app/app.go +++ b/services/wallet-service/internal/app/app.go @@ -40,6 +40,7 @@ type App struct { activityConn *grpc.ClientConn outboxProducer *rocketmqx.Producer outboxWorkerCfg config.OutboxWorkerConfig + projectionWorkerCfg config.ProjectionWorkerConfig walletOutboxTopic string nodeID string redPacketExpiryWorkerCfg config.RedPacketExpiryWorkerConfig @@ -124,6 +125,7 @@ func New(cfg config.Config) (*App, error) { activityConn: activityConn, outboxProducer: outboxProducer, outboxWorkerCfg: cfg.OutboxWorker, + projectionWorkerCfg: cfg.ProjectionWorker, walletOutboxTopic: cfg.RocketMQ.WalletOutbox.Topic, nodeID: cfg.NodeID, redPacketExpiryWorkerCfg: cfg.RedPacketExpiryWorker, @@ -169,25 +171,28 @@ func (a *App) Close() { } func (a *App) runGiftWallProjectionWorker() { - if a.walletSvc == nil { + if a.walletSvc == nil || !a.projectionWorkerCfg.Enabled { return } ctx, cancel := context.WithCancel(context.Background()) a.stopWorker = cancel giftWallWorkerID := "gift-wall-" + a.nodeID + pollInterval := a.projectionWorkerCfg.PollInterval + batchSize := a.projectionWorkerCfg.BatchSize + lockTTL := a.projectionWorkerCfg.LockTTL a.workers.Add(1) go func() { defer a.workers.Done() - ticker := time.NewTicker(2 * time.Second) + ticker := time.NewTicker(pollInterval) defer ticker.Stop() for { // 礼物墙是展示读模型，worker 常驻在 wallet-service 内消费账务 outbox。 // 服务重启会回滚未提交的投影事务；已完成事件由 projection done 状态防重。 - processed, err := a.walletSvc.ProjectPendingGiftWallEvents(ctx, giftWallWorkerID, 50, 30*time.Second) + processed, err := a.walletSvc.ProjectPendingGiftWallEvents(ctx, giftWallWorkerID, batchSize, lockTTL) if err != nil && !errors.Is(err, context.Canceled) { logx.Error(ctx, "gift_wall_projection_failed", err, slog.String("worker_id", giftWallWorkerID)) } - if processed >= 50 { + if processed >= batchSize { // 批次打满说明还有积压，立即继续 drain，避免固定 tick 放大展示延迟。 continue } @@ -202,15 +207,15 @@ func (a *App) runGiftWallProjectionWorker() { a.workers.Add(1) go func() { defer a.workers.Done() - ticker := time.NewTicker(2 * time.Second) + ticker := time.NewTicker(pollInterval) defer ticker.Stop() for { // 徽章展示槽位属于 activity-service 读模型；wallet 只 relay 自己 outbox 中的资源赠送事实。 - processed, err := a.walletSvc.ProjectPendingBadgeGrantEvents(ctx, badgeWorkerID, 50, 30*time.Second) + processed, err := a.walletSvc.ProjectPendingBadgeGrantEvents(ctx, badgeWorkerID, batchSize, lockTTL) if err != nil && !errors.Is(err, context.Canceled) { logx.Error(ctx, "badge_grant_projection_failed", err, slog.String("worker_id", badgeWorkerID)) } - if processed >= 50 { + if processed >= batchSize { continue } select { diff --git a/services/wallet-service/internal/config/config.go b/services/wallet-service/internal/config/config.go index eb6bb235..68d7fc68 100644 --- a/services/wallet-service/internal/config/config.go +++ b/services/wallet-service/internal/config/config.go @@ -29,6 +29,8 @@ type Config struct { GooglePlay GooglePlayConfig `yaml:"google_play"` // OutboxWorker 控制 wallet_outbox 到 MQ 的补偿投递。 OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"` + // ProjectionWorker 控制 wallet_outbox 到本服务读模型的投影补偿扫描。 + ProjectionWorker ProjectionWorkerConfig `yaml:"projection_worker"` // MySQLAutoMigrate 保留给显式本地迁移；线上 schema 由发布流程或 initdb 管理。 MySQLAutoMigrate bool `yaml:"mysql_auto_migrate"` Log logx.Config `yaml:"log"` @@ -75,6 +77,14 @@ type OutboxWorkerConfig struct { MaxBackoff time.Duration `yaml:"max_backoff"` } +// ProjectionWorkerConfig 控制礼物墙和 badge 展示读模型从 wallet_outbox 补偿投影的节奏。 +type ProjectionWorkerConfig struct { + Enabled bool `yaml:"enabled"` + PollInterval time.Duration `yaml:"poll_interval"` + BatchSize int `yaml:"batch_size"` + LockTTL time.Duration `yaml:"lock_ttl"` +} + // GooglePlayConfig 保存 Google Play Developer API 服务账号和网络配置。 type GooglePlayConfig struct { Enabled bool `yaml:"enabled"` @@ -120,6 +130,12 @@ func Default() Config { InitialBackoff: 5 * time.Second, MaxBackoff: 5 * time.Minute, }, + ProjectionWorker: ProjectionWorkerConfig{ + Enabled: true, + PollInterval: 60 * time.Second, + BatchSize: 10, + LockTTL: 30 * time.Second, + }, MySQLAutoMigrate: false, Log: logx.Config{ Level: "info", @@ -234,6 +250,18 @@ func Load(path string) (Config, error) { if cfg.OutboxWorker.Enabled && !cfg.RocketMQ.WalletOutbox.Enabled { return Config{}, errors.New("outbox_worker requires rocketmq.wallet_outbox.enabled") } + if cfg.ProjectionWorker.PollInterval <= 0 { + cfg.ProjectionWorker.PollInterval = Default().ProjectionWorker.PollInterval + } + if cfg.ProjectionWorker.BatchSize <= 0 { + cfg.ProjectionWorker.BatchSize = Default().ProjectionWorker.BatchSize + } + if cfg.ProjectionWorker.BatchSize > 100 { + cfg.ProjectionWorker.BatchSize = 100 + } + if cfg.ProjectionWorker.LockTTL <= 0 { + cfg.ProjectionWorker.LockTTL = Default().ProjectionWorker.LockTTL + } return cfg, nil } diff --git a/services/wallet-service/internal/storage/mysql/badge_projection_relay.go b/services/wallet-service/internal/storage/mysql/badge_projection_relay.go index 4a2a1d64..3196dfe8 100644 --- a/services/wallet-service/internal/storage/mysql/badge_projection_relay.go +++ b/services/wallet-service/internal/storage/mysql/badge_projection_relay.go @@ -5,6 +5,7 @@ import ( "database/sql" "encoding/json" "errors" + "sort" "strings" "time" @@ -21,8 +22,9 @@ const ( ) type badgeGrantProjectionCandidate struct { - AppCode string - EventID string + AppCode string + EventID string + CreatedAtMS int64 } type badgeGrantOutboxPayload struct { @@ -74,37 +76,81 @@ func (r *Repository) ClaimPendingBadgeGrantEvents(ctx context.Context, workerID } func (r *Repository) listBadgeGrantProjectionCandidates(ctx context.Context, nowMS int64, limit int) ([]badgeGrantProjectionCandidate, error) { + appCode := appcode.FromContext(ctx) rows, err := r.db.QueryContext(ctx, ` - SELECT wo.app_code, wo.event_id - FROM wallet_outbox wo + SELECT wo.app_code, wo.event_id, wo.created_at_ms + FROM wallet_outbox wo FORCE INDEX (idx_wallet_outbox_asset_event_created) LEFT JOIN wallet_projection_events pe ON pe.projection_name = ? AND pe.app_code = wo.app_code AND pe.event_id = wo.event_id - WHERE wo.asset_type = ? + WHERE wo.app_code = ? + AND wo.asset_type = ? AND wo.event_type IN (?, ?) - AND ( - pe.event_id IS NULL - OR (pe.status IN (?, ?) AND pe.locked_until_ms <= ?) - ) + AND pe.event_id IS NULL ORDER BY wo.created_at_ms ASC, wo.event_id ASC LIMIT ?`, badgeDisplayProjectionName, + appCode, resourceOutboxAsset, walletResourceGrantedEvent, walletResourceGroupGrantedEvent, - projectionStatusFailed, - projectionStatusProcessing, - nowMS, limit, ) if err != nil { return nil, err } - defer rows.Close() + candidates, err := scanBadgeGrantProjectionCandidates(rows, limit) + if err != nil { + return nil, err + } + rows, err = r.db.QueryContext(ctx, ` + SELECT wo.app_code, wo.event_id, wo.created_at_ms + FROM wallet_projection_events pe + JOIN wallet_outbox wo + ON wo.app_code = pe.app_code AND wo.event_id = pe.event_id + WHERE pe.projection_name = ? + AND pe.app_code = ? + AND pe.status IN (?, ?) + AND pe.locked_until_ms <= ? + AND wo.asset_type = ? + AND wo.event_type IN (?, ?) + ORDER BY wo.created_at_ms ASC, wo.event_id ASC + LIMIT ?`, + badgeDisplayProjectionName, + appCode, + projectionStatusFailed, + projectionStatusProcessing, + nowMS, + resourceOutboxAsset, + walletResourceGrantedEvent, + walletResourceGroupGrantedEvent, + limit, + ) + if err != nil { + return nil, err + } + retryCandidates, err := scanBadgeGrantProjectionCandidates(rows, limit) + if err != nil { + return nil, err + } + candidates = append(candidates, retryCandidates...) + sort.SliceStable(candidates, func(i, j int) bool { + if candidates[i].CreatedAtMS == candidates[j].CreatedAtMS { + return candidates[i].EventID < candidates[j].EventID + } + return candidates[i].CreatedAtMS < candidates[j].CreatedAtMS + }) + if len(candidates) > limit { + candidates = candidates[:limit] + } + return candidates, nil +} +func scanBadgeGrantProjectionCandidates(rows *sql.Rows, limit int) ([]badgeGrantProjectionCandidate, error) { + defer rows.Close() candidates := make([]badgeGrantProjectionCandidate, 0, limit) for rows.Next() { var candidate badgeGrantProjectionCandidate - if err := rows.Scan(&candidate.AppCode, &candidate.EventID); err != nil { + if err := rows.Scan(&candidate.AppCode, &candidate.EventID, &candidate.CreatedAtMS); err != nil { return nil, err } candidates = append(candidates, candidate) diff --git a/services/wallet-service/internal/storage/mysql/gift_wall_projection.go b/services/wallet-service/internal/storage/mysql/gift_wall_projection.go index 4fd54687..9d7c878d 100644 --- a/services/wallet-service/internal/storage/mysql/gift_wall_projection.go +++ b/services/wallet-service/internal/storage/mysql/gift_wall_projection.go @@ -5,6 +5,7 @@ import ( "database/sql" "encoding/json" "errors" + "sort" "strings" "time" @@ -26,8 +27,9 @@ const ( ) type giftWallProjectionCandidate struct { - AppCode string - EventID string + AppCode string + EventID string + CreatedAtMS int64 } type giftWallProjectionEvent struct { @@ -79,34 +81,75 @@ func (r *Repository) ProjectPendingGiftWallEvents(ctx context.Context, workerID func (r *Repository) listGiftWallProjectionCandidates(ctx context.Context, limit int) ([]giftWallProjectionCandidate, error) { nowMs := time.Now().UnixMilli() + appCode := appcode.FromContext(ctx) rows, err := r.db.QueryContext(ctx, ` - SELECT wo.app_code, wo.event_id - FROM wallet_outbox wo + SELECT wo.app_code, wo.event_id, wo.created_at_ms + FROM wallet_outbox wo FORCE INDEX (idx_wallet_outbox_event_created) LEFT JOIN wallet_projection_events pe ON pe.projection_name = ? AND pe.app_code = wo.app_code AND pe.event_id = wo.event_id - WHERE wo.event_type = ? - AND ( - pe.event_id IS NULL - OR (pe.status IN (?, ?) AND pe.locked_until_ms <= ?) - ) + WHERE wo.app_code = ? + AND wo.event_type = ? + AND pe.event_id IS NULL ORDER BY wo.created_at_ms ASC, wo.event_id ASC LIMIT ?`, giftWallProjectionName, + appCode, walletGiftDebitedEvent, - projectionStatusFailed, - projectionStatusProcessing, - nowMs, limit, ) if err != nil { return nil, err } - defer rows.Close() + candidates, err := scanGiftWallProjectionCandidates(rows, limit) + if err != nil { + return nil, err + } + rows, err = r.db.QueryContext(ctx, ` + SELECT wo.app_code, wo.event_id, wo.created_at_ms + FROM wallet_projection_events pe + JOIN wallet_outbox wo + ON wo.app_code = pe.app_code AND wo.event_id = pe.event_id + WHERE pe.projection_name = ? + AND pe.app_code = ? + AND pe.status IN (?, ?) + AND pe.locked_until_ms <= ? + AND wo.event_type = ? + ORDER BY wo.created_at_ms ASC, wo.event_id ASC + LIMIT ?`, + giftWallProjectionName, + appCode, + projectionStatusFailed, + projectionStatusProcessing, + nowMs, + walletGiftDebitedEvent, + limit, + ) + if err != nil { + return nil, err + } + retryCandidates, err := scanGiftWallProjectionCandidates(rows, limit) + if err != nil { + return nil, err + } + candidates = append(candidates, retryCandidates...) + sort.SliceStable(candidates, func(i, j int) bool { + if candidates[i].CreatedAtMS == candidates[j].CreatedAtMS { + return candidates[i].EventID < candidates[j].EventID + } + return candidates[i].CreatedAtMS < candidates[j].CreatedAtMS + }) + if len(candidates) > limit { + candidates = candidates[:limit] + } + return candidates, nil +} +func scanGiftWallProjectionCandidates(rows *sql.Rows, limit int) ([]giftWallProjectionCandidate, error) { + defer rows.Close() candidates := make([]giftWallProjectionCandidate, 0, limit) for rows.Next() { var candidate giftWallProjectionCandidate - if err := rows.Scan(&candidate.AppCode, &candidate.EventID); err != nil { + if err := rows.Scan(&candidate.AppCode, &candidate.EventID, &candidate.CreatedAtMS); err != nil { return nil, err } candidates = append(candidates, candidate)