From 2e061101b3c8e9abb9bbf8993f10ffa471e7e020 Mon Sep 17 00:00:00 2001 From: zhx Date: Fri, 26 Jun 2026 23:21:22 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=A0=E9=99=A4=E6=8E=89=E6=9C=BA=E5=99=A8?= =?UTF-8?q?=E4=BA=BAoutbox?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 1 - scripts/apply-local-rocketmq-topics.sh | 1 - .../room-service/configs/config.docker.yaml | 30 +-- .../configs/config.tencent.example.yaml | 30 +-- services/room-service/configs/config.yaml | 30 +-- .../deploy/mysql/initdb/001_room_service.sql | 21 -- services/room-service/internal/app/app.go | 111 ++------- .../room-service/internal/config/config.go | 61 +---- .../internal/config/config_test.go | 34 ++- .../internal/integration/outbox_sharded.go | 53 +++++ .../integration/outbox_sharded_test.go | 89 +++++++ .../internal/room/service/create_room.go | 2 + .../internal/room/service/gift.go | 46 ++-- .../internal/room/service/outbox_worker.go | 222 ++++++------------ .../service/outbox_worker_internal_test.go | 27 +++ .../room/service/outbox_worker_test.go | 154 ------------ .../internal/room/service/pipeline.go | 11 +- .../internal/room/service/repository.go | 20 +- .../room/service/robot_display_publisher.go | 68 ++++++ ...ox_sampler.go => robot_display_sampler.go} | 28 +-- .../service/room_rocket_progress_coalescer.go | 57 +++-- .../internal/room/service/service.go | 27 +-- .../room/service/user_gift_value_test.go | 139 ++++++----- .../internal/storage/mysql/command_log.go | 12 +- .../internal/storage/mysql/outbox.go | 203 +--------------- .../internal/storage/mysql/repository.go | 3 +- .../internal/storage/mysql/schema.go | 20 -- .../internal/testutil/mysqltest/mysqltest.go | 24 -- 28 files changed, 569 insertions(+), 955 deletions(-) create mode 100644 services/room-service/internal/integration/outbox_sharded.go create mode 100644 services/room-service/internal/integration/outbox_sharded_test.go create mode 100644 services/room-service/internal/room/service/outbox_worker_internal_test.go delete mode 100644 services/room-service/internal/room/service/outbox_worker_test.go create mode 100644 services/room-service/internal/room/service/robot_display_publisher.go rename services/room-service/internal/room/service/{robot_outbox_sampler.go => robot_display_sampler.go} (58%) diff --git a/README.md b/README.md index 14e9f62d..d17e163e 100644 --- a/README.md +++ b/README.md @@ -193,7 +193,6 @@ docs/flutter对接/Flutter App 通知与钱包余额对接.md | `hyapp_room_outbox` | `hyapp-room-outbox-producer` | `hyapp-activity-room-outbox` | activity-service 消费 `RoomGiftSent`、`RoomRocketIgnited` 等房间事实 | | `hyapp_room_outbox` | `hyapp-room-outbox-producer` | `hyapp-notice-room-outbox` | notice-service 消费 `RoomUserKicked` 等私有通知事实 | | `hyapp_room_outbox` | `hyapp-room-outbox-producer` | `hyapp-room-im-bridge` | room-service IM bridge 消费房间群系统消息和群成员控制事实 | -| `hyapp_robot_room_outbox` | `hyapp-room-robot-outbox-producer` | `hyapp-test-room-robot-im-bridge` / `hyapp-room-robot-im-bridge` | room-service IM bridge 只消费机器人房间礼物、热度和麦位热度展示事实；本地/testbox 用 test group，线上用正式 group | | `hyapp_room_outbox` | `hyapp-room-outbox-producer` | `hyapp-user-mictime-room-outbox` | user-service 消费 `RoomMicChanged` 并更新用户麦时读模型 | | `hyapp_room_outbox` | `hyapp-room-outbox-producer` | `hyapp-statistics-room-outbox` | statistics-service 消费 `RoomGiftSent`、`RoomUserJoined` 并更新聚合表 | | `hyapp_user_outbox` | `hyapp-user-outbox-producer` | `hyapp-statistics-user-outbox` | statistics-service 消费 `UserRegistered` 并更新注册 cohort | diff --git a/scripts/apply-local-rocketmq-topics.sh b/scripts/apply-local-rocketmq-topics.sh index 47ba75ff..d1f2288f 100755 --- a/scripts/apply-local-rocketmq-topics.sh +++ b/scripts/apply-local-rocketmq-topics.sh @@ -22,7 +22,6 @@ TOPICS=( "hyapp_wallet_outbox" "hyapp_wallet_realtime_outbox" "hyapp_room_outbox" - "hyapp_robot_room_outbox" "hyapp_room_rocket_launch" "hyapp_user_outbox" "hyapp_message_action_outbox" diff --git a/services/room-service/configs/config.docker.yaml b/services/room-service/configs/config.docker.yaml index 98988e70..85928091 100644 --- a/services/room-service/configs/config.docker.yaml +++ b/services/room-service/configs/config.docker.yaml @@ -70,13 +70,8 @@ rocketmq: tencent_im_consumer_enabled: true tencent_im_consumer_group: "hyapp-room-im-bridge" consumer_max_reconsume_times: 16 - robot_room_outbox: - enabled: true - topic: "hyapp_robot_room_outbox" - producer_group: "hyapp-room-robot-outbox-producer" - tencent_im_consumer_enabled: true - tencent_im_consumer_group: "hyapp-test-room-robot-im-bridge" - consumer_max_reconsume_times: 16 + consumer_goroutines: 16 + im_publish_concurrency: 16 rocket_launch: enabled: true topic: "hyapp_room_rocket_launch" @@ -92,28 +87,11 @@ outbox_worker: # Docker 和 testbox 走 MQ；room-service 同进程启动 IM bridge consumer 补偿房间群消息。 enabled: true publish_mode: "mq" - poll_interval: "1s" - batch_size: 100 + poll_interval: "200ms" + batch_size: 50 concurrency: 16 publish_timeout: "15s" retry_strategy: "exponential_backoff" max_retry_count: 10 initial_backoff: "5s" max_backoff: "5m" -robot_outbox_worker: - # Docker 本地单独投递机器人展示事件，避免压住主 room_outbox fanout。 - enabled: true - publish_mode: "mq" - poll_interval: "200ms" - batch_size: 100 - concurrency: 1 - publish_timeout: "15s" - retry_strategy: "exponential_backoff" - max_retry_count: 10 - initial_backoff: "5s" - max_backoff: "5m" - stale_discard_after: "60s" - cleanup_interval: "30s" - cleanup_active_after: "5m" - cleanup_terminal_after: "1h" - cleanup_batch_size: 50000 diff --git a/services/room-service/configs/config.tencent.example.yaml b/services/room-service/configs/config.tencent.example.yaml index 232d9236..a2305087 100644 --- a/services/room-service/configs/config.tencent.example.yaml +++ b/services/room-service/configs/config.tencent.example.yaml @@ -80,13 +80,8 @@ rocketmq: tencent_im_consumer_enabled: true tencent_im_consumer_group: "hyapp-room-im-bridge" consumer_max_reconsume_times: 16 - robot_room_outbox: - enabled: true - topic: "hyapp_robot_room_outbox" - producer_group: "hyapp-room-robot-outbox-producer" - tencent_im_consumer_enabled: true - tencent_im_consumer_group: "hyapp-room-robot-im-bridge" - consumer_max_reconsume_times: 16 + consumer_goroutines: 16 + im_publish_concurrency: 16 rocket_launch: enabled: true topic: "hyapp_room_rocket_launch" @@ -102,28 +97,11 @@ outbox_worker: # mq 模式下 outbox worker 只把事实投 MQ，并顺带安排火箭延迟发射唤醒。 enabled: true publish_mode: "mq" - poll_interval: "1s" - batch_size: 100 + poll_interval: "200ms" + batch_size: 50 concurrency: 16 publish_timeout: "15s" retry_strategy: "exponential_backoff" max_retry_count: 10 initial_backoff: "5s" max_backoff: "5m" -robot_outbox_worker: - # 机器人房间展示事件独立投递，只服务 IM 展示，不投 activity/statistics 主业务消费者。 - enabled: true - publish_mode: "mq" - poll_interval: "200ms" - batch_size: 100 - concurrency: 1 - publish_timeout: "15s" - retry_strategy: "exponential_backoff" - max_retry_count: 10 - initial_backoff: "5s" - max_backoff: "5m" - stale_discard_after: "60s" - cleanup_interval: "30s" - cleanup_active_after: "5m" - cleanup_terminal_after: "1h" - cleanup_batch_size: 50000 diff --git a/services/room-service/configs/config.yaml b/services/room-service/configs/config.yaml index 7ebeca18..ee93cca8 100644 --- a/services/room-service/configs/config.yaml +++ b/services/room-service/configs/config.yaml @@ -73,13 +73,8 @@ rocketmq: tencent_im_consumer_enabled: true tencent_im_consumer_group: "hyapp-room-im-bridge" consumer_max_reconsume_times: 16 - robot_room_outbox: - enabled: true - topic: "hyapp_robot_room_outbox" - producer_group: "hyapp-room-robot-outbox-producer" - tencent_im_consumer_enabled: true - tencent_im_consumer_group: "hyapp-test-room-robot-im-bridge" - consumer_max_reconsume_times: 16 + consumer_goroutines: 16 + im_publish_concurrency: 16 rocket_launch: enabled: true topic: "hyapp_room_rocket_launch" @@ -95,28 +90,11 @@ outbox_worker: # room_outbox 是腾讯云 IM 和 activity-service 的异步投递通道；本地也走 MQ，避免和测试/线上链路漂移。 enabled: true publish_mode: "mq" - poll_interval: "1s" - batch_size: 100 + poll_interval: "200ms" + batch_size: 50 concurrency: 16 publish_timeout: "15s" retry_strategy: "exponential_backoff" max_retry_count: 10 initial_backoff: "5s" max_backoff: "5m" -robot_outbox_worker: - # 机器人展示事件走独立 outbox/topic，只投房间 IM，不占用真人 room_outbox 主流程。 - enabled: true - publish_mode: "mq" - poll_interval: "200ms" - batch_size: 100 - concurrency: 1 - publish_timeout: "15s" - retry_strategy: "exponential_backoff" - max_retry_count: 10 - initial_backoff: "5s" - max_backoff: "5m" - stale_discard_after: "60s" - cleanup_interval: "30s" - cleanup_active_after: "5m" - cleanup_terminal_after: "1h" - cleanup_batch_size: 50000 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 5cc31ce3..484f2463 100644 --- a/services/room-service/deploy/mysql/initdb/001_room_service.sql +++ b/services/room-service/deploy/mysql/initdb/001_room_service.sql @@ -156,27 +156,6 @@ CREATE TABLE IF NOT EXISTS room_outbox ( KEY idx_room_outbox_retention (app_code, status, updated_at_ms, event_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='房间事件 outbox 表'; -CREATE TABLE IF NOT EXISTS room_robot_outbox ( - app_code VARCHAR(32) NOT NULL COMMENT '应用编码，用于多租户隔离', - event_id VARCHAR(128) NOT NULL COMMENT '事件幂等 ID', - event_type VARCHAR(64) NOT NULL COMMENT '事件类型', - room_id VARCHAR(64) NOT NULL COMMENT '机器人房间 ID', - status VARCHAR(32) NOT NULL COMMENT '业务状态', - worker_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT 'worker ID', - lock_until_ms BIGINT NULL COMMENT '锁定过期时间，UTC epoch ms', - envelope LONGBLOB NOT NULL COMMENT '事件信封', - created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', - retry_count INT NOT NULL DEFAULT 0 COMMENT '重试次数', - next_retry_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '下一次重试时间，UTC epoch ms', - last_error TEXT NULL COMMENT '最后一次失败原因', - updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', - PRIMARY KEY (app_code, event_id), - KEY idx_room_robot_outbox_pending (app_code, status, next_retry_at_ms, created_at_ms, event_id), - KEY idx_room_robot_outbox_claim (app_code, status, lock_until_ms, created_at_ms, event_id), - KEY idx_room_robot_outbox_event_created (app_code, event_type, created_at_ms, event_id), - KEY idx_room_robot_outbox_retention (app_code, status, updated_at_ms, event_id) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='机器人房间展示事件 outbox 表'; - CREATE TABLE IF NOT EXISTS room_user_feed_entries ( app_code VARCHAR(32) NOT NULL COMMENT '应用编码，用于多租户隔离', user_id BIGINT NOT NULL COMMENT '用户 ID', diff --git a/services/room-service/internal/app/app.go b/services/room-service/internal/app/app.go index 8c2998d8..6cf51448 100644 --- a/services/room-service/internal/app/app.go +++ b/services/room-service/internal/app/app.go @@ -174,10 +174,9 @@ func New(cfg config.Config) (*App, error) { rtcUserRemover = rtcClient } - mqProducers := make([]*rocketmqx.Producer, 0, 3) - mqConsumers := make([]*rocketmqx.Consumer, 0, 4) + mqProducers := make([]*rocketmqx.Producer, 0, 2) + mqConsumers := make([]*rocketmqx.Consumer, 0, 3) var rocketLaunchScheduler integration.RoomRocketLaunchScheduler - robotOutboxPublishers := make([]integration.OutboxPublisher, 0, 1) if cfg.OutboxWorker.PublishMode == config.OutboxPublishModeDirect || cfg.OutboxWorker.PublishMode == config.OutboxPublishModeDual { if activityConn != nil { activityPublisher := integration.NewActivityOutboxPublisher(activityv1.NewRoomEventConsumerServiceClient(activityConn), time.Second) @@ -187,12 +186,6 @@ func New(cfg config.Config) (*App, error) { outboxPublishers = append(outboxPublishers, tencentPublisher) } } - if cfg.RobotOutboxWorker.PublishMode == config.OutboxPublishModeDirect || cfg.RobotOutboxWorker.PublishMode == config.OutboxPublishModeDual { - if tencentPublisher != nil { - // 机器人展示事件只需要 IM 房间消息；不投 activity/statistics，避免污染真人房间主业务链路。 - robotOutboxPublishers = append(robotOutboxPublishers, tencentPublisher) - } - } if cfg.OutboxWorker.PublishMode == config.OutboxPublishModeMQ || cfg.OutboxWorker.PublishMode == config.OutboxPublishModeDual { outboxProducer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.RoomOutbox.ProducerGroup)) if err != nil { @@ -205,18 +198,6 @@ func New(cfg config.Config) (*App, error) { mqProducers = append(mqProducers, outboxProducer) outboxPublishers = append(outboxPublishers, integration.NewRocketMQOutboxPublisher(outboxProducer, cfg.RocketMQ.RoomOutbox.Topic)) } - if cfg.RobotOutboxWorker.PublishMode == config.OutboxPublishModeMQ || cfg.RobotOutboxWorker.PublishMode == config.OutboxPublishModeDual { - robotOutboxProducer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.RobotRoomOutbox.ProducerGroup)) - if err != nil { - _ = repository.Close() - closeClientConn(activityConn) - _ = walletConn.Close() - _ = redisClient.Close() - return nil, err - } - mqProducers = append(mqProducers, robotOutboxProducer) - robotOutboxPublishers = append(robotOutboxPublishers, integration.NewRocketMQOutboxPublisher(robotOutboxProducer, cfg.RocketMQ.RobotRoomOutbox.Topic)) - } if cfg.RocketMQ.RocketLaunch.Enabled { rocketProducer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.RocketLaunch.ProducerGroup)) if err != nil { @@ -230,7 +211,6 @@ func New(cfg config.Config) (*App, error) { rocketLaunchScheduler = integration.NewRocketMQRocketLaunchScheduler(rocketProducer, cfg.RocketMQ.RocketLaunch.Topic) } outboxPublisher := integration.NewCompositeOutboxPublisher(outboxPublishers...) - robotOutboxPublisher := integration.NewCompositeOutboxPublisher(robotOutboxPublishers...) var luckyGiftClient integration.LuckyGiftClient if activityConn != nil { luckyGiftClient = integration.NewGRPCLuckyGiftClient(activityConn) @@ -246,7 +226,7 @@ func New(cfg config.Config) (*App, error) { MicPublishTimeout: cfg.MicPublishTimeout, RTCUserRemover: rtcUserRemover, RoomRocketLaunchScheduler: rocketLaunchScheduler, - RobotOutboxPublisher: robotOutboxPublisher, + RobotDisplayPublisher: tencentPublisher, HumanRobotPoolProvider: integration.NewGRPCHumanRoomRobotPoolProvider(robotConn, userConn), RoomGiftLeaderboard: roomservice.NewRedisRoomGiftLeaderboardStore(redisClient), RoomListCache: roomservice.NewRedisRoomListCacheStore(redisClient), @@ -254,7 +234,7 @@ func New(cfg config.Config) (*App, error) { LuckyGiftSendLockTTL: cfg.LuckyGiftSendLockTTL, }, directory, repository, integration.NewGRPCWalletClient(walletConn), roomPublisher, outboxPublisher, luckyGiftClient) if cfg.RocketMQ.RocketLaunch.Enabled { - rocketConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.RocketLaunch.ConsumerGroup, cfg.RocketMQ.RocketLaunch.ConsumerMaxReconsumeTimes)) + rocketConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.RocketLaunch.ConsumerGroup, cfg.RocketMQ.RocketLaunch.ConsumerMaxReconsumeTimes, 0)) if err != nil { _ = repository.Close() closeClientConn(activityConn) @@ -285,7 +265,8 @@ func New(cfg config.Config) (*App, error) { return nil, errors.New("rocketmq room_outbox tencent_im consumer requires tencent_im.enabled") } if cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled && tencentPublisher != nil { - imConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.RoomOutbox.TencentIMConsumerGroup, cfg.RocketMQ.RoomOutbox.ConsumerMaxReconsumeTimes)) + imPublisher := integration.NewShardedOutboxPublisher(tencentPublisher, cfg.RocketMQ.RoomOutbox.IMPublishConcurrency) + imConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.RoomOutbox.TencentIMConsumerGroup, cfg.RocketMQ.RoomOutbox.ConsumerMaxReconsumeTimes, cfg.RocketMQ.RoomOutbox.ConsumerGoroutines)) if err != nil { _ = repository.Close() closeClientConn(activityConn) @@ -298,7 +279,7 @@ func New(cfg config.Config) (*App, error) { if err != nil { return err } - return tencentPublisher.PublishOutboxEvent(appcode.WithContext(ctx, envelope.GetAppCode()), envelope) + return imPublisher.PublishOutboxEvent(appcode.WithContext(ctx, envelope.GetAppCode()), envelope) }); err != nil { _ = repository.Close() closeClientConn(activityConn) @@ -308,39 +289,8 @@ func New(cfg config.Config) (*App, error) { } mqConsumers = append(mqConsumers, imConsumer) } - if cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerEnabled && tencentPublisher == nil { - _ = repository.Close() - closeClientConn(activityConn) - _ = walletConn.Close() - _ = redisClient.Close() - return nil, errors.New("rocketmq robot_room_outbox tencent_im consumer requires tencent_im.enabled") - } - if cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerEnabled && tencentPublisher != nil { - robotIMConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerGroup, cfg.RocketMQ.RobotRoomOutbox.ConsumerMaxReconsumeTimes)) - if err != nil { - _ = repository.Close() - closeClientConn(activityConn) - _ = walletConn.Close() - _ = redisClient.Close() - return nil, err - } - if err := robotIMConsumer.Subscribe(cfg.RocketMQ.RobotRoomOutbox.Topic, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { - envelope, _, err := roommq.DecodeRoomOutboxMessage(message.Body) - if err != nil { - return err - } - return tencentPublisher.PublishOutboxEvent(appcode.WithContext(ctx, envelope.GetAppCode()), envelope) - }); err != nil { - _ = repository.Close() - closeClientConn(activityConn) - _ = walletConn.Close() - _ = redisClient.Close() - return nil, err - } - mqConsumers = append(mqConsumers, robotIMConsumer) - } if cfg.RocketMQ.UserOutbox.Enabled { - userConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.UserOutbox.ConsumerGroup, cfg.RocketMQ.UserOutbox.ConsumerMaxReconsumeTimes)) + userConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.UserOutbox.ConsumerGroup, cfg.RocketMQ.UserOutbox.ConsumerMaxReconsumeTimes, 0)) if err != nil { _ = repository.Close() closeClientConn(activityConn) @@ -451,7 +401,7 @@ func (a *App) Run() error { }) } a.workers.Go(func(ctx context.Context) { - // 火箭燃料进度允许 1 秒展示延迟；100ms 扫描让到期 bucket 尽快写回 outbox，后续仍走原 MQ/IM 补偿链路。 + // 火箭燃料进度允许 1 秒展示延迟；普通房写主 outbox，机器人展示到期后 best-effort 直发腾讯 IM。 a.service.RunRoomRocketProgressFlushWorker(ctx, 100*time.Millisecond) }) if a.cfg.RoomListCacheRefreshInterval > 0 { @@ -463,38 +413,14 @@ func (a *App) Run() error { if a.cfg.OutboxWorker.Enabled { a.workers.Go(func(ctx context.Context) { a.service.RunOutboxWorker(ctx, roomservice.OutboxWorkerOptions{ - PollInterval: a.cfg.OutboxWorker.PollInterval, - BatchSize: a.cfg.OutboxWorker.BatchSize, - Concurrency: a.cfg.OutboxWorker.Concurrency, - PublishTimeout: a.cfg.OutboxWorker.PublishTimeout, - RetryStrategy: a.cfg.OutboxWorker.RetryStrategy, - MaxRetryCount: a.cfg.OutboxWorker.MaxRetryCount, - InitialBackoff: a.cfg.OutboxWorker.InitialBackoff, - MaxBackoff: a.cfg.OutboxWorker.MaxBackoff, - StaleDiscardAfter: a.cfg.OutboxWorker.StaleDiscardAfter, - CleanupInterval: a.cfg.OutboxWorker.CleanupInterval, - CleanupActiveAfter: a.cfg.OutboxWorker.CleanupActiveAfter, - CleanupTerminalAfter: a.cfg.OutboxWorker.CleanupTerminalAfter, - CleanupBatchSize: a.cfg.OutboxWorker.CleanupBatchSize, - }) - }) - } - if a.cfg.RobotOutboxWorker.Enabled { - a.workers.Go(func(ctx context.Context) { - a.service.RunRobotOutboxWorker(ctx, roomservice.OutboxWorkerOptions{ - PollInterval: a.cfg.RobotOutboxWorker.PollInterval, - BatchSize: a.cfg.RobotOutboxWorker.BatchSize, - Concurrency: a.cfg.RobotOutboxWorker.Concurrency, - PublishTimeout: a.cfg.RobotOutboxWorker.PublishTimeout, - RetryStrategy: a.cfg.RobotOutboxWorker.RetryStrategy, - MaxRetryCount: a.cfg.RobotOutboxWorker.MaxRetryCount, - InitialBackoff: a.cfg.RobotOutboxWorker.InitialBackoff, - MaxBackoff: a.cfg.RobotOutboxWorker.MaxBackoff, - StaleDiscardAfter: a.cfg.RobotOutboxWorker.StaleDiscardAfter, - CleanupInterval: a.cfg.RobotOutboxWorker.CleanupInterval, - CleanupActiveAfter: a.cfg.RobotOutboxWorker.CleanupActiveAfter, - CleanupTerminalAfter: a.cfg.RobotOutboxWorker.CleanupTerminalAfter, - CleanupBatchSize: a.cfg.RobotOutboxWorker.CleanupBatchSize, + PollInterval: a.cfg.OutboxWorker.PollInterval, + BatchSize: a.cfg.OutboxWorker.BatchSize, + Concurrency: a.cfg.OutboxWorker.Concurrency, + PublishTimeout: a.cfg.OutboxWorker.PublishTimeout, + RetryStrategy: a.cfg.OutboxWorker.RetryStrategy, + MaxRetryCount: a.cfg.OutboxWorker.MaxRetryCount, + InitialBackoff: a.cfg.OutboxWorker.InitialBackoff, + MaxBackoff: a.cfg.OutboxWorker.MaxBackoff, }) }) } @@ -655,7 +581,7 @@ func rocketMQProducerConfig(cfg config.RocketMQConfig, groupName string) rocketm } } -func rocketMQConsumerConfig(cfg config.RocketMQConfig, groupName string, maxReconsumeTimes int32) rocketmqx.ConsumerConfig { +func rocketMQConsumerConfig(cfg config.RocketMQConfig, groupName string, maxReconsumeTimes int32, consumeGoroutines int) rocketmqx.ConsumerConfig { return rocketmqx.ConsumerConfig{ EndpointConfig: rocketMQEndpointConfig(cfg), GroupName: groupName, @@ -663,6 +589,7 @@ func rocketMQConsumerConfig(cfg config.RocketMQConfig, groupName string, maxReco ConsumeRetryDelay: time.Second, ConsumePullBatch: 32, ConsumePullTimeout: 15 * time.Minute, + ConsumeGoroutines: consumeGoroutines, } } diff --git a/services/room-service/internal/config/config.go b/services/room-service/internal/config/config.go index b307a431..b8c25f60 100644 --- a/services/room-service/internal/config/config.go +++ b/services/room-service/internal/config/config.go @@ -90,8 +90,6 @@ type Config struct { RocketMQ RocketMQConfig `yaml:"rocketmq"` // OutboxWorker 控制 room_outbox 到腾讯云 IM 补偿投递 worker。 OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"` - // RobotOutboxWorker 控制机器人房间展示 outbox，避免机器人流量抢占主 room_outbox。 - RobotOutboxWorker OutboxWorkerConfig `yaml:"robot_outbox_worker"` // Log 控制 stdout 结构化日志格式、等级和 access body 策略。 Log logx.Config `yaml:"log"` } @@ -118,16 +116,6 @@ type OutboxWorkerConfig struct { InitialBackoff time.Duration `yaml:"initial_backoff"` // MaxBackoff 是指数退避的最大等待时间。 MaxBackoff time.Duration `yaml:"max_backoff"` - // StaleDiscardAfter 只用于 robot outbox；超过该窗口的展示事件直接删除，不再补发 MQ/IM。 - StaleDiscardAfter time.Duration `yaml:"stale_discard_after"` - // CleanupInterval 控制 robot outbox 后台清理频率；非正数表示不启动清理。 - CleanupInterval time.Duration `yaml:"cleanup_interval"` - // CleanupActiveAfter 控制 pending/retryable/delivering 机器人展示事件的最大保留时长。 - CleanupActiveAfter time.Duration `yaml:"cleanup_active_after"` - // CleanupTerminalAfter 控制 delivered/failed 机器人展示事件的最大保留时长。 - CleanupTerminalAfter time.Duration `yaml:"cleanup_terminal_after"` - // CleanupBatchSize 控制单轮清理最多删除多少行，避免长事务拖慢主库。 - CleanupBatchSize int `yaml:"cleanup_batch_size"` } // RocketMQConfig 描述 room-service 使用的 RocketMQ 集群和业务 topic。 @@ -143,7 +131,6 @@ type RocketMQConfig struct { SendTimeout time.Duration `yaml:"send_timeout"` Retry int `yaml:"retry"` RoomOutbox RoomOutboxMQConfig `yaml:"room_outbox"` - RobotRoomOutbox RoomOutboxMQConfig `yaml:"robot_room_outbox"` RocketLaunch RocketLaunchMQConfig `yaml:"rocket_launch"` UserOutbox UserOutboxMQConfig `yaml:"user_outbox"` } @@ -156,6 +143,10 @@ type RoomOutboxMQConfig struct { TencentIMConsumerEnabled bool `yaml:"tencent_im_consumer_enabled"` TencentIMConsumerGroup string `yaml:"tencent_im_consumer_group"` ConsumerMaxReconsumeTimes int32 `yaml:"consumer_max_reconsume_times"` + // ConsumerGoroutines 控制 RocketMQ push consumer 的并发回调数；同房间顺序仍由 IM publish 分片锁保证。 + ConsumerGoroutines int `yaml:"consumer_goroutines"` + // IMPublishConcurrency 控制 MQ -> Tencent IM 桥接按 room_id 分片后的最大并行房间数。 + IMPublishConcurrency int `yaml:"im_publish_concurrency"` } // RocketLaunchMQConfig 控制火箭延迟发射唤醒 topic。 @@ -296,7 +287,6 @@ func Default() Config { RoomListCacheRefreshInterval: 5 * time.Minute, RocketMQ: defaultRocketMQConfig(), OutboxWorker: defaultOutboxWorkerConfig(), - RobotOutboxWorker: defaultOutboxWorkerConfig(), Log: logx.Config{ Level: "info", Format: "json", @@ -333,14 +323,7 @@ func defaultRocketMQConfig() RocketMQConfig { TencentIMConsumerEnabled: false, TencentIMConsumerGroup: "hyapp-room-im-bridge", ConsumerMaxReconsumeTimes: 16, - }, - RobotRoomOutbox: RoomOutboxMQConfig{ - Enabled: false, - Topic: "hyapp_robot_room_outbox", - ProducerGroup: "hyapp-room-robot-outbox-producer", - TencentIMConsumerEnabled: false, - TencentIMConsumerGroup: "hyapp-room-robot-im-bridge", - ConsumerMaxReconsumeTimes: 16, + IMPublishConcurrency: 1, }, RocketLaunch: RocketLaunchMQConfig{ Enabled: false, @@ -440,12 +423,7 @@ func Normalize(cfg Config) (Config, error) { return Config{}, err } cfg.OutboxWorker = outboxWorker - robotOutboxWorker, err := normalizeOutboxWorkerConfig(cfg.RobotOutboxWorker) - if err != nil { - return Config{}, err - } - cfg.RobotOutboxWorker = robotOutboxWorker - rocketMQ, err := normalizeRocketMQConfig(cfg.RocketMQ, cfg.OutboxWorker.PublishMode, cfg.RobotOutboxWorker.PublishMode) + rocketMQ, err := normalizeRocketMQConfig(cfg.RocketMQ, cfg.OutboxWorker.PublishMode) if err != nil { return Config{}, err } @@ -494,10 +472,6 @@ func normalizeOutboxWorkerConfig(cfg OutboxWorkerConfig) (OutboxWorkerConfig, er if cfg.MaxBackoff < cfg.InitialBackoff { cfg.MaxBackoff = cfg.InitialBackoff } - if cfg.CleanupBatchSize <= 0 && (cfg.CleanupActiveAfter > 0 || cfg.CleanupTerminalAfter > 0) { - // 清理开启时必须有批量上限；默认和线上一次性清理批大小保持一致。 - cfg.CleanupBatchSize = 50000 - } cfg.RetryStrategy = strings.TrimSpace(cfg.RetryStrategy) if cfg.RetryStrategy == "" { cfg.RetryStrategy = defaults.RetryStrategy @@ -519,7 +493,7 @@ func normalizeOutboxWorkerConfig(cfg OutboxWorkerConfig) (OutboxWorkerConfig, er return cfg, nil } -func normalizeRocketMQConfig(cfg RocketMQConfig, publishMode string, robotPublishMode string) (RocketMQConfig, error) { +func normalizeRocketMQConfig(cfg RocketMQConfig, publishMode string) (RocketMQConfig, error) { defaults := defaultRocketMQConfig() cfg.NameServers = normalizeStringSlice(cfg.NameServers) cfg.NameServerDomain = strings.TrimSpace(cfg.NameServerDomain) @@ -545,17 +519,11 @@ func normalizeRocketMQConfig(cfg RocketMQConfig, publishMode string, robotPublis if cfg.RoomOutbox.ConsumerMaxReconsumeTimes <= 0 { cfg.RoomOutbox.ConsumerMaxReconsumeTimes = defaults.RoomOutbox.ConsumerMaxReconsumeTimes } - if cfg.RobotRoomOutbox.Topic = strings.TrimSpace(cfg.RobotRoomOutbox.Topic); cfg.RobotRoomOutbox.Topic == "" { - cfg.RobotRoomOutbox.Topic = defaults.RobotRoomOutbox.Topic + if cfg.RoomOutbox.ConsumerGoroutines < 0 { + cfg.RoomOutbox.ConsumerGoroutines = 0 } - if cfg.RobotRoomOutbox.ProducerGroup = strings.TrimSpace(cfg.RobotRoomOutbox.ProducerGroup); cfg.RobotRoomOutbox.ProducerGroup == "" { - cfg.RobotRoomOutbox.ProducerGroup = defaults.RobotRoomOutbox.ProducerGroup - } - if cfg.RobotRoomOutbox.TencentIMConsumerGroup = strings.TrimSpace(cfg.RobotRoomOutbox.TencentIMConsumerGroup); cfg.RobotRoomOutbox.TencentIMConsumerGroup == "" { - cfg.RobotRoomOutbox.TencentIMConsumerGroup = defaults.RobotRoomOutbox.TencentIMConsumerGroup - } - if cfg.RobotRoomOutbox.ConsumerMaxReconsumeTimes <= 0 { - cfg.RobotRoomOutbox.ConsumerMaxReconsumeTimes = defaults.RobotRoomOutbox.ConsumerMaxReconsumeTimes + if cfg.RoomOutbox.IMPublishConcurrency <= 0 { + cfg.RoomOutbox.IMPublishConcurrency = defaults.RoomOutbox.IMPublishConcurrency } if cfg.RocketLaunch.Topic = strings.TrimSpace(cfg.RocketLaunch.Topic); cfg.RocketLaunch.Topic == "" { cfg.RocketLaunch.Topic = defaults.RocketLaunch.Topic @@ -578,7 +546,7 @@ func normalizeRocketMQConfig(cfg RocketMQConfig, publishMode string, robotPublis if cfg.UserOutbox.ConsumerMaxReconsumeTimes <= 0 { cfg.UserOutbox.ConsumerMaxReconsumeTimes = defaults.UserOutbox.ConsumerMaxReconsumeTimes } - if cfg.RoomOutbox.Enabled || cfg.RobotRoomOutbox.Enabled || cfg.RocketLaunch.Enabled || cfg.UserOutbox.Enabled || cfg.RoomOutbox.TencentIMConsumerEnabled || cfg.RobotRoomOutbox.TencentIMConsumerEnabled { + if cfg.RoomOutbox.Enabled || cfg.RocketLaunch.Enabled || cfg.UserOutbox.Enabled || cfg.RoomOutbox.TencentIMConsumerEnabled { cfg.Enabled = true } if publishMode == OutboxPublishModeMQ || publishMode == OutboxPublishModeDual { @@ -586,11 +554,6 @@ func normalizeRocketMQConfig(cfg RocketMQConfig, publishMode string, robotPublis return RocketMQConfig{}, fmt.Errorf("outbox_worker publish_mode %q requires rocketmq.room_outbox.enabled", publishMode) } } - if robotPublishMode == OutboxPublishModeMQ || robotPublishMode == OutboxPublishModeDual { - if !cfg.RobotRoomOutbox.Enabled { - return RocketMQConfig{}, fmt.Errorf("robot_outbox_worker publish_mode %q requires rocketmq.robot_room_outbox.enabled", robotPublishMode) - } - } if cfg.Enabled { if len(cfg.NameServers) == 0 && cfg.NameServerDomain == "" { return RocketMQConfig{}, fmt.Errorf("rocketmq name_servers or name_server_domain is required") diff --git a/services/room-service/internal/config/config_test.go b/services/room-service/internal/config/config_test.go index fa1ba7a6..c52a0071 100644 --- a/services/room-service/internal/config/config_test.go +++ b/services/room-service/internal/config/config_test.go @@ -40,18 +40,15 @@ func TestLoad(t *testing.T) { if cfg.TencentIM.SDKAppID != 20040101 || cfg.TencentIM.SecretKey == "" || cfg.TencentIM.AdminIdentifier != "administrator" { t.Fatalf("local room-service must use test Tencent IM account: %+v", cfg.TencentIM) } - if !cfg.OutboxWorker.Enabled || cfg.OutboxWorker.PublishMode != OutboxPublishModeMQ || cfg.OutboxWorker.PollInterval != time.Second || cfg.OutboxWorker.BatchSize != 100 || cfg.OutboxWorker.Concurrency != 16 || cfg.OutboxWorker.PublishTimeout != 15*time.Second || cfg.OutboxWorker.RetryStrategy != "exponential_backoff" || cfg.OutboxWorker.MaxRetryCount != 10 || cfg.OutboxWorker.InitialBackoff != 5*time.Second || cfg.OutboxWorker.MaxBackoff != 5*time.Minute { + if !cfg.OutboxWorker.Enabled || cfg.OutboxWorker.PublishMode != OutboxPublishModeMQ || cfg.OutboxWorker.PollInterval != 200*time.Millisecond || cfg.OutboxWorker.BatchSize != 50 || cfg.OutboxWorker.Concurrency != 16 || cfg.OutboxWorker.PublishTimeout != 15*time.Second || cfg.OutboxWorker.RetryStrategy != "exponential_backoff" || cfg.OutboxWorker.MaxRetryCount != 10 || cfg.OutboxWorker.InitialBackoff != 5*time.Second || cfg.OutboxWorker.MaxBackoff != 5*time.Minute { t.Fatalf("unexpected outbox worker config: %+v", cfg.OutboxWorker) } - if !cfg.RobotOutboxWorker.Enabled || cfg.RobotOutboxWorker.PublishMode != OutboxPublishModeMQ || cfg.RobotOutboxWorker.PollInterval != 200*time.Millisecond || cfg.RobotOutboxWorker.BatchSize != 100 || cfg.RobotOutboxWorker.Concurrency != 1 || cfg.RobotOutboxWorker.PublishTimeout != 15*time.Second || cfg.RobotOutboxWorker.RetryStrategy != "exponential_backoff" || cfg.RobotOutboxWorker.MaxRetryCount != 10 || cfg.RobotOutboxWorker.InitialBackoff != 5*time.Second || cfg.RobotOutboxWorker.MaxBackoff != 5*time.Minute || cfg.RobotOutboxWorker.StaleDiscardAfter != 60*time.Second || cfg.RobotOutboxWorker.CleanupInterval != 30*time.Second || cfg.RobotOutboxWorker.CleanupActiveAfter != 5*time.Minute || cfg.RobotOutboxWorker.CleanupTerminalAfter != time.Hour || cfg.RobotOutboxWorker.CleanupBatchSize != 50000 { - t.Fatalf("unexpected robot outbox worker config: %+v", cfg.RobotOutboxWorker) - } // 本地默认也走 room_outbox MQ、IM bridge 和火箭延迟唤醒，避免本地直投和线上 MQ 语义分叉。 - if !cfg.RocketMQ.Enabled || !cfg.RocketMQ.RoomOutbox.Enabled || !cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled || !cfg.RocketMQ.RobotRoomOutbox.Enabled || !cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerEnabled || !cfg.RocketMQ.RocketLaunch.Enabled || !cfg.RocketMQ.UserOutbox.Enabled { + if !cfg.RocketMQ.Enabled || !cfg.RocketMQ.RoomOutbox.Enabled || !cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled || !cfg.RocketMQ.RocketLaunch.Enabled || !cfg.RocketMQ.UserOutbox.Enabled { t.Fatalf("local config must enable room MQ fanout and IM bridge: %+v", cfg.RocketMQ) } - if cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerGroup != "hyapp-test-room-robot-im-bridge" { - t.Fatalf("local robot IM bridge group mismatch: %s", cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerGroup) + if cfg.RocketMQ.RoomOutbox.ConsumerGoroutines != 16 || cfg.RocketMQ.RoomOutbox.IMPublishConcurrency != 16 { + t.Fatalf("local IM bridge concurrency mismatch: room=%+v", cfg.RocketMQ.RoomOutbox) } } @@ -83,17 +80,14 @@ func TestLoadTencentExample(t *testing.T) { if cfg.AdvertiseAddr == "" || cfg.AdvertiseAddr == "room-service.internal:13001" { t.Fatalf("tencent example must use per-instance advertise_addr, got %q", cfg.AdvertiseAddr) } - if !cfg.OutboxWorker.Enabled || cfg.OutboxWorker.Concurrency != 16 || cfg.OutboxWorker.PublishTimeout != 15*time.Second || cfg.OutboxWorker.RetryStrategy != "exponential_backoff" || cfg.OutboxWorker.MaxRetryCount != 10 { + if !cfg.OutboxWorker.Enabled || cfg.OutboxWorker.PollInterval != 200*time.Millisecond || cfg.OutboxWorker.BatchSize != 50 || cfg.OutboxWorker.Concurrency != 16 || cfg.OutboxWorker.PublishTimeout != 15*time.Second || cfg.OutboxWorker.RetryStrategy != "exponential_backoff" || cfg.OutboxWorker.MaxRetryCount != 10 { t.Fatalf("tencent example must configure outbox worker: %+v", cfg.OutboxWorker) } - if !cfg.RobotOutboxWorker.Enabled || cfg.RobotOutboxWorker.PublishMode != OutboxPublishModeMQ || cfg.RobotOutboxWorker.PollInterval != 200*time.Millisecond || cfg.RobotOutboxWorker.BatchSize != 100 || cfg.RobotOutboxWorker.Concurrency != 1 || cfg.RobotOutboxWorker.PublishTimeout != 15*time.Second || cfg.RobotOutboxWorker.StaleDiscardAfter != 60*time.Second || cfg.RobotOutboxWorker.CleanupActiveAfter != 5*time.Minute || cfg.RobotOutboxWorker.CleanupTerminalAfter != time.Hour { - t.Fatalf("tencent example must configure robot outbox worker: %+v", cfg.RobotOutboxWorker) - } - if cfg.OutboxWorker.PublishMode != OutboxPublishModeMQ || !cfg.RocketMQ.RoomOutbox.Enabled || !cfg.RocketMQ.RobotRoomOutbox.Enabled || !cfg.RocketMQ.RocketLaunch.Enabled { + if cfg.OutboxWorker.PublishMode != OutboxPublishModeMQ || !cfg.RocketMQ.RoomOutbox.Enabled || !cfg.RocketMQ.RocketLaunch.Enabled { t.Fatalf("tencent example must route room outbox and rocket launch through RocketMQ: mode=%s mq=%+v", cfg.OutboxWorker.PublishMode, cfg.RocketMQ) } - if cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerGroup != "hyapp-room-robot-im-bridge" { - t.Fatalf("prod robot IM bridge group mismatch: %s", cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerGroup) + if cfg.RocketMQ.RoomOutbox.ConsumerGoroutines != 16 || cfg.RocketMQ.RoomOutbox.IMPublishConcurrency != 16 { + t.Fatalf("prod IM bridge concurrency mismatch: room=%+v", cfg.RocketMQ.RoomOutbox) } } @@ -106,8 +100,8 @@ func TestLoadDockerConfigStartsRoomIMBridgeWhenOutboxUsesMQ(t *testing.T) { if cfg.OutboxWorker.PublishMode != OutboxPublishModeMQ { t.Fatalf("docker config should publish room outbox through MQ, got %q", cfg.OutboxWorker.PublishMode) } - if cfg.OutboxWorker.Concurrency != 16 || cfg.OutboxWorker.PublishTimeout != 15*time.Second { - t.Fatalf("docker config must run room outbox with 16 workers and a longer claim window: %+v", cfg.OutboxWorker) + if cfg.OutboxWorker.PollInterval != 200*time.Millisecond || cfg.OutboxWorker.BatchSize != 50 || cfg.OutboxWorker.Concurrency != 16 || cfg.OutboxWorker.PublishTimeout != 15*time.Second { + t.Fatalf("docker config must run room outbox with 16 workers, short polling, and a bounded claim window: %+v", cfg.OutboxWorker) } if !cfg.RocketMQ.RoomOutbox.Enabled { t.Fatalf("docker config must enable room outbox MQ") @@ -115,11 +109,11 @@ func TestLoadDockerConfigStartsRoomIMBridgeWhenOutboxUsesMQ(t *testing.T) { if !cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled { t.Fatalf("docker mq room outbox must start Tencent IM bridge consumer") } - if cfg.RobotOutboxWorker.PublishMode != OutboxPublishModeMQ || !cfg.RocketMQ.RobotRoomOutbox.Enabled || !cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerEnabled { - t.Fatalf("docker config must start robot outbox MQ and IM bridge: worker=%+v mq=%+v", cfg.RobotOutboxWorker, cfg.RocketMQ.RobotRoomOutbox) + if cfg.RocketMQ.RoomOutbox.ConsumerGoroutines != 16 || cfg.RocketMQ.RoomOutbox.IMPublishConcurrency != 16 { + t.Fatalf("docker room IM bridge concurrency mismatch: %+v", cfg.RocketMQ.RoomOutbox) } - if cfg.TencentIM.SDKAppID != 20040101 || cfg.TencentIM.AdminIdentifier != "administrator" || cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerGroup != "hyapp-test-room-robot-im-bridge" { - t.Fatalf("docker config must use test IM account and robot group: im=%+v group=%s", cfg.TencentIM, cfg.RocketMQ.RobotRoomOutbox.TencentIMConsumerGroup) + if cfg.TencentIM.SDKAppID != 20040101 || cfg.TencentIM.AdminIdentifier != "administrator" { + t.Fatalf("docker config must use test IM account: im=%+v", cfg.TencentIM) } if !cfg.TencentIM.Enabled { t.Fatalf("Tencent IM bridge consumer requires tencent_im.enabled") diff --git a/services/room-service/internal/integration/outbox_sharded.go b/services/room-service/internal/integration/outbox_sharded.go new file mode 100644 index 00000000..4d2e119a --- /dev/null +++ b/services/room-service/internal/integration/outbox_sharded.go @@ -0,0 +1,53 @@ +package integration + +import ( + "context" + "hash/fnv" + "sync" + + roomeventsv1 "hyapp.local/api/proto/events/room/v1" +) + +// ShardedOutboxPublisher 只在 IM bridge 侧使用：不同 room_id 可以并发，同一个房间仍串行调用下游发布器。 +type ShardedOutboxPublisher struct { + inner OutboxPublisher + locks []sync.Mutex +} + +// NewShardedOutboxPublisher 用 room_id 分片保护下游发布器的房间内顺序；concurrency<=1 时保持历史串行实现。 +func NewShardedOutboxPublisher(inner OutboxPublisher, concurrency int) OutboxPublisher { + if inner == nil { + return NewNoopOutboxPublisher() + } + if concurrency <= 1 { + return inner + } + return &ShardedOutboxPublisher{ + inner: inner, + locks: make([]sync.Mutex, concurrency), + } +} + +// PublishOutboxEvent 对同一 room_id 使用同一把锁，避免 RocketMQ consumer 并发后把礼物和麦位展示乱序发给腾讯 IM。 +func (p *ShardedOutboxPublisher) PublishOutboxEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) error { + if p == nil || p.inner == nil || envelope == nil { + return nil + } + if len(p.locks) == 0 { + return p.inner.PublishOutboxEvent(ctx, envelope) + } + lock := &p.locks[p.shardIndex(envelope.GetRoomId())] + lock.Lock() + defer lock.Unlock() + + return p.inner.PublishOutboxEvent(ctx, envelope) +} + +func (p *ShardedOutboxPublisher) shardIndex(roomID string) uint32 { + if len(p.locks) == 1 { + return 0 + } + hash := fnv.New32a() + _, _ = hash.Write([]byte(roomID)) + return hash.Sum32() % uint32(len(p.locks)) +} diff --git a/services/room-service/internal/integration/outbox_sharded_test.go b/services/room-service/internal/integration/outbox_sharded_test.go new file mode 100644 index 00000000..6371c799 --- /dev/null +++ b/services/room-service/internal/integration/outbox_sharded_test.go @@ -0,0 +1,89 @@ +package integration + +import ( + "context" + "sync" + "testing" + "time" + + roomeventsv1 "hyapp.local/api/proto/events/room/v1" +) + +type blockingOutboxPublisher struct { + entered chan struct{} + release chan struct{} + + mu sync.Mutex + inFlight int + maxInFlight int + count int +} + +func newBlockingOutboxPublisher() *blockingOutboxPublisher { + return &blockingOutboxPublisher{ + entered: make(chan struct{}, 2), + release: make(chan struct{}), + } +} + +func (p *blockingOutboxPublisher) PublishOutboxEvent(ctx context.Context, _ *roomeventsv1.EventEnvelope) error { + p.mu.Lock() + p.inFlight++ + if p.inFlight > p.maxInFlight { + p.maxInFlight = p.inFlight + } + p.count++ + p.mu.Unlock() + + p.entered <- struct{}{} + select { + case <-ctx.Done(): + return ctx.Err() + case <-p.release: + } + + p.mu.Lock() + p.inFlight-- + p.mu.Unlock() + return nil +} + +func (p *blockingOutboxPublisher) snapshot() (int, int) { + p.mu.Lock() + defer p.mu.Unlock() + return p.count, p.maxInFlight +} + +func TestShardedOutboxPublisherSerializesSameRoom(t *testing.T) { + inner := newBlockingOutboxPublisher() + publisher := NewShardedOutboxPublisher(inner, 8) + envelope := &roomeventsv1.EventEnvelope{RoomId: "room-same"} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + // 两个同房间事件同时进入时，分片锁必须让第二条等第一条完成后再调用真实 IM 发布器。 + errs := make(chan error, 2) + go func() { errs <- publisher.PublishOutboxEvent(ctx, envelope) }() + <-inner.entered + go func() { errs <- publisher.PublishOutboxEvent(ctx, envelope) }() + + select { + case <-inner.entered: + t.Fatal("same room event entered inner publisher before the first event finished") + case <-time.After(50 * time.Millisecond): + } + + inner.release <- struct{}{} + if err := <-errs; err != nil { + t.Fatalf("first publish failed: %v", err) + } + <-inner.entered + inner.release <- struct{}{} + if err := <-errs; err != nil { + t.Fatalf("second publish failed: %v", err) + } + + if count, maxInFlight := inner.snapshot(); count != 2 || maxInFlight != 1 { + t.Fatalf("same room publishes must be serialized, count=%d max_in_flight=%d", count, maxInFlight) + } +} diff --git a/services/room-service/internal/room/service/create_room.go b/services/room-service/internal/room/service/create_room.go index 341e1535..ef073fe0 100644 --- a/services/room-service/internal/room/service/create_room.go +++ b/services/room-service/internal/room/service/create_room.go @@ -245,6 +245,8 @@ func (s *Service) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest) return nil, err } saveMutationMS = elapsedMS(saveStartedAt) + // SaveMutation 成功代表 RoomCreated outbox 已经随命令事务提交，可以立即唤醒 worker 解除 1 秒轮询下限。 + s.notifyOutboxCommitted([]outbox.Record{createdEvent}) snapshot := snapshotWithApp(ctx, roomState.ToProto()) snapshotStartedAt := time.Now() diff --git a/services/room-service/internal/room/service/gift.go b/services/room-service/internal/room/service/gift.go index 8e28228b..3957609b 100644 --- a/services/room-service/internal/room/service/gift.go +++ b/services/room-service/internal/room/service/gift.go @@ -294,11 +294,11 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob } records := make([]outbox.Record, 0, len(giftEvents)+4) - robotRecords := make([]outbox.Record, 0, len(giftEvents)+5) - // RobotWalletGift 只表示这笔礼物由机器人钱包结算；真实房机器人仍会同步写房间贡献、火箭和麦位热度，但展示事件必须走独立 outbox，避免机器人流量抢占真人主 outbox。 - robotOutboxGift := cmd.RobotWalletGift - if robotOutboxGift { - robotRecords = append(robotRecords, giftEvents...) + robotDisplayRecords := make([]outbox.Record, 0, len(giftEvents)+5) + // RobotWalletGift 只表示这笔礼物由机器人钱包结算；真实房机器人仍会同步写房间贡献、火箭和麦位热度，但展示事件改为提交后直发 IM，避免机器人流量抢占真人主 outbox。 + robotDisplayGift := cmd.RobotWalletGift + if robotDisplayGift { + robotDisplayRecords = append(robotDisplayRecords, giftEvents...) } else { records = append(records, giftEvents...) } @@ -319,15 +319,15 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob if err != nil { return mutationResult{}, nil, err } - if robotOutboxGift { + if robotDisplayGift { // 房间热度和麦位收礼热度仍要给客户端展示；是否真实计入统计由 RobotGift 语义单独控制，不能再拿投递通道反推业务真实性。 - robotRecords = append(robotRecords, heatEvent, rankEvent) + robotDisplayRecords = append(robotDisplayRecords, heatEvent, rankEvent) } else { records = append(records, heatEvent, rankEvent) } var rocketProgress *roomRocketProgressPending if !cmd.RobotGift { - // 真实房间礼物才广播火箭进度；真人房机器人送礼使用机器人钱包结算，但仍是真实房间贡献，所以火箭事件跟随机器人通道投递。 + // 真实房间礼物才广播火箭进度；真人房机器人送礼使用机器人钱包结算，但仍是真实房间贡献，所以火箭事件跟随机器人展示直发通道投递。 if rocketApply.progressEvent != nil { progressRecord, err := outbox.Build(current.RoomID, "RoomRocketFuelChanged", current.Version, now, rocketApply.progressEvent) if err != nil { @@ -340,7 +340,7 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob // 燃料变化是高频展示事件，不逐笔写 outbox；命令提交成功后由内存合并器按 1 秒窗口只落最大进度。 rocketProgress = &roomRocketProgressPending{ record: progressRecord, - robotLane: robotOutboxGift, + robotLane: robotDisplayGift, rocketID: rocketApply.progressEvent.GetRocketId(), level: rocketApply.progressEvent.GetLevel(), currentFuel: rocketApply.progressEvent.GetCurrentFuel(), @@ -352,8 +352,8 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob if err != nil { return mutationResult{}, nil, err } - if robotOutboxGift { - robotRecords = append(robotRecords, ignitedEvent) + if robotDisplayGift { + robotDisplayRecords = append(robotDisplayRecords, ignitedEvent) } else { records = append(records, ignitedEvent) } @@ -381,8 +381,8 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob if err != nil { return mutationResult{}, nil, err } - if robotOutboxGift { - robotRecords = append(robotRecords, drawEvent) + if robotDisplayGift { + robotDisplayRecords = append(robotDisplayRecords, drawEvent) } else { records = append(records, drawEvent) } @@ -390,16 +390,16 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob firstTargetDisplayProfile := giftDisplayProfileForUser(cmd.TargetDisplayProfiles, cmd.TargetUserID) result := mutationResult{ - snapshot: current.ToProto(), - billingReceiptID: billing.GetBillingReceiptId(), - roomHeat: current.Heat, - giftRank: cloneProtoRank(current.GiftRank), - luckyGift: luckyGift, - luckyGifts: luckyGifts, - commandPayload: commandPayload, - walletDebitMS: walletDebitMS, - robotOutboxRecords: robotRecords, - roomRocketProgress: rocketProgress, + snapshot: current.ToProto(), + billingReceiptID: billing.GetBillingReceiptId(), + roomHeat: current.Heat, + giftRank: cloneProtoRank(current.GiftRank), + luckyGift: luckyGift, + luckyGifts: luckyGifts, + commandPayload: commandPayload, + walletDebitMS: walletDebitMS, + robotDisplayRecords: robotDisplayRecords, + roomRocketProgress: rocketProgress, roomGiftLeaderboard: &RoomGiftLeaderboardIncrement{ AppCode: appcode.FromContext(ctx), RoomID: current.RoomID, diff --git a/services/room-service/internal/room/service/outbox_worker.go b/services/room-service/internal/room/service/outbox_worker.go index 5b9e82ae..906b9466 100644 --- a/services/room-service/internal/room/service/outbox_worker.go +++ b/services/room-service/internal/room/service/outbox_worker.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "sort" "strings" "sync" "time" @@ -18,8 +19,6 @@ const ( outboxRetryExponentialBackoff = "exponential_backoff" // outboxWorkerName 固定写入日志，便于后续按 worker 聚合检索。 outboxWorkerName = "room_outbox" - // robotOutboxWorkerName 固定标识机器人展示事件通道，避免和主 room_outbox 监控混淆。 - robotOutboxWorkerName = "room_robot_outbox" ) // OutboxWorkerOptions 控制 room_outbox pending 事件的持续补偿循环。 @@ -42,18 +41,8 @@ type OutboxWorkerOptions struct { MaxBackoff time.Duration // WorkerID 写入 outbox worker_id，空值时使用节点 ID；并发启动时会自动带上序号。 WorkerID string - // WorkerName 写入日志 worker 字段，空值时使用主通道或机器人通道的稳定名称。 + // WorkerName 写入日志 worker 字段，空值时使用主通道的稳定名称。 WorkerName string - // StaleDiscardAfter 只用于 robot outbox；超过该窗口的展示事件直接删除，不再补发 MQ/IM。 - StaleDiscardAfter time.Duration - // CleanupInterval 控制 robot outbox 后台清理频率；非正数表示不启动清理。 - CleanupInterval time.Duration - // CleanupActiveAfter 控制 pending/retryable/delivering 机器人展示事件的最大保留时长。 - CleanupActiveAfter time.Duration - // CleanupTerminalAfter 控制 delivered/failed 机器人展示事件的最大保留时长。 - CleanupTerminalAfter time.Duration - // CleanupBatchSize 控制单轮清理最多删除多少行，避免长事务拖慢主库。 - CleanupBatchSize int } func defaultOutboxWorkerOptions() OutboxWorkerOptions { @@ -70,6 +59,16 @@ func defaultOutboxWorkerOptions() OutboxWorkerOptions { } } +func firstNonEmpty(values ...string) string { + // 配置和事件里经常需要“显式值优先，运行上下文兜底”；集中处理可避免空字符串穿透到 worker_id、app_code 等排障关键字段。 + for _, value := range values { + if strings.TrimSpace(value) != "" { + return value + } + } + return "" +} + func normalizeOutboxWorkerOptions(options OutboxWorkerOptions) OutboxWorkerOptions { // worker 层再次归一化，防止测试或手动构造绕过 config.Normalize。 defaults := defaultOutboxWorkerOptions() @@ -101,10 +100,6 @@ func normalizeOutboxWorkerOptions(options OutboxWorkerOptions) OutboxWorkerOptio if options.MaxBackoff < options.InitialBackoff { options.MaxBackoff = options.InitialBackoff } - if options.CleanupBatchSize <= 0 && (options.CleanupActiveAfter > 0 || options.CleanupTerminalAfter > 0) { - // robot outbox 清理开启时必须有批量上限；默认 5 万和线上一次性清理批量保持一致。 - options.CleanupBatchSize = 50000 - } options.RetryStrategy = strings.TrimSpace(options.RetryStrategy) if options.RetryStrategy == "" { // 空策略按指数退避处理，保持和配置默认值一致。 @@ -116,15 +111,10 @@ func normalizeOutboxWorkerOptions(options OutboxWorkerOptions) OutboxWorkerOptio // RunOutboxWorker 持续扫描 pending outbox；它不执行房间命令，只投递已持久化事件。 func (s *Service) RunOutboxWorker(ctx context.Context, options OutboxWorkerOptions) { - s.runOutboxWorker(ctx, options, outboxWorkerName, s.ProcessPendingOutbox) + s.runOutboxWorker(ctx, options, outboxWorkerName, s.outboxWake, s.ProcessPendingOutbox) } -// RunRobotOutboxWorker 持续扫描机器人房间展示 outbox；它和主 room_outbox 完全分开抢占。 -func (s *Service) RunRobotOutboxWorker(ctx context.Context, options OutboxWorkerOptions) { - s.runOutboxWorker(ctx, options, robotOutboxWorkerName, s.ProcessPendingRobotOutbox) -} - -func (s *Service) runOutboxWorker(ctx context.Context, options OutboxWorkerOptions, workerName string, process func(context.Context, OutboxWorkerOptions) error) { +func (s *Service) runOutboxWorker(ctx context.Context, options OutboxWorkerOptions, workerName string, wake <-chan struct{}, process func(context.Context, OutboxWorkerOptions) error) { options = normalizeOutboxWorkerOptions(options) if options.RetryStrategy != outboxRetryExponentialBackoff { // 配置层会拒绝未知策略；这里兜底避免测试或手写构造静默改变语义。 @@ -135,7 +125,7 @@ func (s *Service) runOutboxWorker(ctx context.Context, options OutboxWorkerOptio // 单 worker 保持历史 worker_id 和日志名称，避免线上排障查询字段无意义变化。 options.WorkerID = firstNonEmpty(strings.TrimSpace(options.WorkerID), s.nodeID) options.WorkerName = firstNonEmpty(strings.TrimSpace(options.WorkerName), workerName) - s.runOutboxWorkerLoop(ctx, options, process) + s.runOutboxWorkerLoop(ctx, options, wake, process) return } @@ -148,18 +138,16 @@ func (s *Service) runOutboxWorker(ctx context.Context, options OutboxWorkerOptio wg.Add(1) go func() { defer wg.Done() - s.runOutboxWorkerLoop(ctx, workerOptions, process) + s.runOutboxWorkerLoop(ctx, workerOptions, wake, process) }() } wg.Wait() } -func (s *Service) runOutboxWorkerLoop(ctx context.Context, options OutboxWorkerOptions, process func(context.Context, OutboxWorkerOptions) error) { +func (s *Service) runOutboxWorkerLoop(ctx context.Context, options OutboxWorkerOptions, wake <-chan struct{}, process func(context.Context, OutboxWorkerOptions) error) { workerName := firstNonEmpty(strings.TrimSpace(options.WorkerName), outboxWorkerName) // 启动后先执行一轮，避免服务刚恢复时还要等一个 poll interval 才补偿历史事件。 - if err := process(ctx, options); err != nil && ctx.Err() == nil { - logx.Error(ctx, "worker_batch_failed", err, slog.String("worker", workerName), slog.String("node_id", s.nodeID), slog.Int("batch_size", options.BatchSize)) - } + s.runOutboxWorkerBatch(ctx, options, workerName, process) ticker := time.NewTicker(options.PollInterval) defer ticker.Stop() @@ -169,39 +157,54 @@ func (s *Service) runOutboxWorkerLoop(ctx context.Context, options OutboxWorkerO case <-ctx.Done(): return case <-ticker.C: - if err := process(ctx, options); err != nil && ctx.Err() == nil { - logx.Error(ctx, "worker_batch_failed", err, slog.String("worker", workerName), slog.String("node_id", s.nodeID), slog.Int("batch_size", options.BatchSize)) - } + s.runOutboxWorkerBatch(ctx, options, workerName, process) + case <-wake: + // 提交唤醒只负责打破 poll_interval 下限；真正领取仍靠 MySQL claim 和 lock_until_ms。 + s.runOutboxWorkerBatch(ctx, options, workerName, process) } } } +func (s *Service) runOutboxWorkerBatch(ctx context.Context, options OutboxWorkerOptions, workerName string, process func(context.Context, OutboxWorkerOptions) error) { + if err := process(ctx, options); err != nil && ctx.Err() == nil { + logx.Error(ctx, "worker_batch_failed", err, slog.String("worker", workerName), slog.String("node_id", s.nodeID), slog.Int("batch_size", options.BatchSize)) + } +} + +func (s *Service) notifyOutboxCommitted(outboxRecords []outbox.Record) { + // 唤醒信号必须在 MySQL 事务成功后发送；它只降低新事件延迟，丢失时仍由周期扫描兜底。 + s.notifyOutboxWake(s.outboxWake, len(outboxRecords)) +} + +func (s *Service) notifyOutboxWake(wake chan struct{}, recordCount int) { + if wake == nil || recordCount <= 0 { + return + } + select { + case wake <- struct{}{}: + default: + // channel 满说明已经有 worker 被唤醒或待唤醒；丢弃信号不会丢事件，MySQL pending 仍可扫描。 + } +} + // ProcessPendingOutbox 把待投递事件补偿性地推送给异步消费者。 func (s *Service) ProcessPendingOutbox(ctx context.Context, options OutboxWorkerOptions) error { workerName := firstNonEmpty(strings.TrimSpace(options.WorkerName), outboxWorkerName) - return s.processPendingOutbox(ctx, options, workerName, false) + return s.processPendingOutbox(ctx, options, workerName) } -// ProcessPendingRobotOutbox 把机器人展示事件补偿性地推送给独立机器人异步消费者。 -func (s *Service) ProcessPendingRobotOutbox(ctx context.Context, options OutboxWorkerOptions) error { - workerName := firstNonEmpty(strings.TrimSpace(options.WorkerName), robotOutboxWorkerName) - if err := s.maybeCleanupRobotOutbox(ctx, normalizeOutboxWorkerOptions(options)); err != nil { - return err - } - return s.processPendingOutbox(ctx, options, workerName, true) -} - -func (s *Service) processPendingOutbox(ctx context.Context, options OutboxWorkerOptions, workerName string, robotLane bool) error { +func (s *Service) processPendingOutbox(ctx context.Context, options OutboxWorkerOptions, workerName string) error { options = normalizeOutboxWorkerOptions(options) if options.RetryStrategy != outboxRetryExponentialBackoff { return fmt.Errorf("unsupported outbox retry_strategy %q", options.RetryStrategy) } // 补偿 worker 先抢占 pending/retryable 事件，再投递，避免多机同时处理同一批。 - records, err := s.claimPendingOutbox(ctx, robotLane, options) + records, err := s.claimPendingOutbox(ctx, options) if err != nil { return err } + sortOutboxRecords(records) for _, record := range records { select { @@ -210,29 +213,10 @@ func (s *Service) processPendingOutbox(ctx context.Context, options OutboxWorker return ctx.Err() default: } - if robotLane && staleRobotOutboxRecord(record, options.StaleDiscardAfter) { - // 机器人展示事件超过 TTL 后再补发会制造“迟到礼物”体验，直接删除并释放 MQ/IM 资源。 - deleteCtx, deleteCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout) - deleteErr := s.repository.DeleteRobotOutbox(deleteCtx, record.EventID) - deleteCancel() - if deleteErr != nil { - return deleteErr - } - logx.Warn(ctx, "robot_outbox_stale_discarded", - slog.String("worker", workerName), - slog.String("node_id", s.nodeID), - slog.String("event_id", record.EventID), - slog.String("event_type", record.EventType), - slog.String("room_id", record.RoomID), - slog.Int64("created_at_ms", record.CreatedAtMS), - slog.Int64("stale_discard_after_ms", options.StaleDiscardAfter.Milliseconds()), - ) - continue - } if record.RetryCount >= options.MaxRetryCount { // 已达到最大失败次数的历史事件直接转死信，防止每轮 worker 继续打外部系统。 markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout) - markErr := s.markOutboxDead(markCtx, robotLane, record.EventID, deadLetterReason(record)) + markErr := s.markOutboxDead(markCtx, record.EventID, deadLetterReason(record)) markCancel() if markErr != nil { return markErr @@ -251,13 +235,13 @@ func (s *Service) processPendingOutbox(ctx context.Context, options OutboxWorker publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout) // 发布上下文不继承 worker ctx，避免 shutdown 取消导致已开始的单条外部投递无界处于未知状态。 - err := s.publishOutboxRecord(publishCtx, record, robotLane) + err := s.publishOutboxRecord(publishCtx, record) cancel() if err != nil { // 投递失败转为 retryable，并写入指数退避时间，等待下一轮 worker 到期后再抢占。 nextRetryAtMS := time.Now().UTC().Add(outboxBackoff(record.RetryCount+1, options)).UnixMilli() markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout) - markErr := s.markOutboxRetryable(markCtx, robotLane, record.EventID, trimOutboxError(err.Error()), nextRetryAtMS) + markErr := s.markOutboxRetryable(markCtx, record.EventID, trimOutboxError(err.Error()), nextRetryAtMS) markCancel() if markErr != nil { return markErr @@ -278,7 +262,7 @@ func (s *Service) processPendingOutbox(ctx context.Context, options OutboxWorker // 只有外部发布成功后才能标记 delivered。 markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout) - markErr := s.markOutboxDelivered(markCtx, robotLane, record.EventID) + markErr := s.markOutboxDelivered(markCtx, record.EventID) markCancel() if markErr != nil { return markErr @@ -297,96 +281,55 @@ func (s *Service) processPendingOutbox(ctx context.Context, options OutboxWorker return nil } -func (s *Service) maybeCleanupRobotOutbox(ctx context.Context, options OutboxWorkerOptions) error { - if s == nil || s.repository == nil || options.CleanupInterval <= 0 || (options.CleanupActiveAfter <= 0 && options.CleanupTerminalAfter <= 0) { - return nil - } - - now := time.Now().UTC() - s.robotOutboxCleanupMu.Lock() - if !s.robotOutboxLastCleanup.IsZero() && now.Sub(s.robotOutboxLastCleanup) < options.CleanupInterval { - s.robotOutboxCleanupMu.Unlock() - return nil - } - // 先记录本次尝试时间，避免多 worker 或异常重试在同一秒内反复打 DELETE。 - s.robotOutboxLastCleanup = now - s.robotOutboxCleanupMu.Unlock() - - var activeBeforeMS int64 - if options.CleanupActiveAfter > 0 { - activeBeforeMS = now.Add(-options.CleanupActiveAfter).UnixMilli() - } - var terminalBeforeMS int64 - if options.CleanupTerminalAfter > 0 { - terminalBeforeMS = now.Add(-options.CleanupTerminalAfter).UnixMilli() - } - cleanupCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), appcode.FromContext(ctx)), options.PublishTimeout) - activeDeleted, terminalDeleted, err := s.repository.CleanupRobotOutbox(cleanupCtx, activeBeforeMS, terminalBeforeMS, options.CleanupBatchSize) - cancel() - if err != nil { - return err - } - if activeDeleted > 0 || terminalDeleted > 0 { - logx.Info(ctx, "robot_outbox_cleanup_deleted", - slog.String("worker", robotOutboxWorkerName), - slog.String("node_id", s.nodeID), - slog.Int64("active_deleted", activeDeleted), - slog.Int64("terminal_deleted", terminalDeleted), - slog.Int("batch_size", options.CleanupBatchSize), - ) - } - return nil +func sortOutboxRecords(records []outbox.Record) { + // DB claim 负责并发安全；批内稳定排序只把礼物、上下麦和进退房展示事件排到热度/榜单前面，降低发送方可感知尾延迟。 + sort.SliceStable(records, func(i, j int) bool { + leftPriority := outboxEventPriority(records[i].EventType) + rightPriority := outboxEventPriority(records[j].EventType) + if leftPriority != rightPriority { + return leftPriority < rightPriority + } + if records[i].CreatedAtMS != records[j].CreatedAtMS { + return records[i].CreatedAtMS < records[j].CreatedAtMS + } + return records[i].EventID < records[j].EventID + }) } -func staleRobotOutboxRecord(record outbox.Record, ttl time.Duration) bool { - if ttl <= 0 || record.CreatedAtMS <= 0 { - return false +func outboxEventPriority(eventType string) int { + switch eventType { + case "RoomGiftSent", "RoomMicChanged", "RoomUserJoined", "RoomUserLeft": + return 0 + case "RoomHeatChanged", "RoomRankChanged", "RoomRocketFuelChanged": + return 2 + default: + return 1 } - // 使用当前 UTC 毫秒判断 TTL，和 outbox created_at_ms 的持久化口径保持一致。 - return time.Now().UTC().UnixMilli()-record.CreatedAtMS >= ttl.Milliseconds() } -func (s *Service) claimPendingOutbox(ctx context.Context, robotLane bool, options OutboxWorkerOptions) ([]outbox.Record, error) { +func (s *Service) claimPendingOutbox(ctx context.Context, options OutboxWorkerOptions) ([]outbox.Record, error) { lockUntilMS := time.Now().UTC().Add(options.PublishTimeout).UnixMilli() workerID := firstNonEmpty(strings.TrimSpace(options.WorkerID), s.nodeID) - if robotLane { - return s.repository.ClaimPendingRobotOutbox(ctx, workerID, options.BatchSize, lockUntilMS) - } return s.repository.ClaimPendingOutbox(ctx, workerID, options.BatchSize, lockUntilMS) } -func (s *Service) markOutboxDelivered(ctx context.Context, robotLane bool, eventID string) error { - if robotLane { - return s.repository.MarkRobotOutboxDelivered(ctx, eventID) - } +func (s *Service) markOutboxDelivered(ctx context.Context, eventID string) error { return s.repository.MarkOutboxDelivered(ctx, eventID) } -func (s *Service) markOutboxRetryable(ctx context.Context, robotLane bool, eventID string, lastErr string, nextRetryAtMS int64) error { - if robotLane { - return s.repository.MarkRobotOutboxRetryable(ctx, eventID, lastErr, nextRetryAtMS) - } +func (s *Service) markOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error { return s.repository.MarkOutboxRetryable(ctx, eventID, lastErr, nextRetryAtMS) } -func (s *Service) markOutboxDead(ctx context.Context, robotLane bool, eventID string, lastErr string) error { - if robotLane { - return s.repository.MarkRobotOutboxDead(ctx, eventID, lastErr) - } +func (s *Service) markOutboxDead(ctx context.Context, eventID string, lastErr string) error { return s.repository.MarkOutboxDead(ctx, eventID, lastErr) } -func (s *Service) publishOutboxRecord(ctx context.Context, record outbox.Record, robotLane bool) error { +func (s *Service) publishOutboxRecord(ctx context.Context, record outbox.Record) error { // Ignited 先安排延迟发射，再对外广播事件；如果 MQ 延迟调度不可用，本条 outbox 保持 retryable。 if err := s.scheduleRoomRocketLaunchFromEnvelope(ctx, record.Envelope); err != nil { return err } - if robotLane { - if s.robotOutboxPublisher == nil { - return nil - } - return s.robotOutboxPublisher.PublishOutboxEvent(ctx, record.Envelope) - } if s.outboxPublisher == nil { return nil } @@ -430,12 +373,3 @@ func trimOutboxError(value string) string { // V1 只保留错误前缀，避免外部响应过大撑爆日志或 TEXT 字段查询成本。 return value[:512] } - -func firstNonEmpty(values ...string) string { - for _, value := range values { - if strings.TrimSpace(value) != "" { - return strings.TrimSpace(value) - } - } - return "" -} diff --git a/services/room-service/internal/room/service/outbox_worker_internal_test.go b/services/room-service/internal/room/service/outbox_worker_internal_test.go new file mode 100644 index 00000000..a6be334c --- /dev/null +++ b/services/room-service/internal/room/service/outbox_worker_internal_test.go @@ -0,0 +1,27 @@ +package service + +import ( + "testing" + + "hyapp/services/room-service/internal/room/outbox" +) + +func TestSortOutboxRecordsPrioritizesGiftDisplay(t *testing.T) { + // 低优先级事件故意给更早 created_at_ms，验证 worker 不会让热度/榜单压住礼物展示。 + records := []outbox.Record{ + {EventID: "rank", EventType: "RoomRankChanged", CreatedAtMS: 1}, + {EventID: "gift", EventType: "RoomGiftSent", CreatedAtMS: 2}, + {EventID: "other", EventType: "RoomNoticeCreated", CreatedAtMS: 0}, + {EventID: "heat", EventType: "RoomHeatChanged", CreatedAtMS: 0}, + } + + sortOutboxRecords(records) + + got := []string{records[0].EventID, records[1].EventID, records[2].EventID, records[3].EventID} + want := []string{"gift", "other", "heat", "rank"} + for index := range want { + if got[index] != want[index] { + t.Fatalf("priority order mismatch got=%v want=%v", got, want) + } + } +} diff --git a/services/room-service/internal/room/service/outbox_worker_test.go b/services/room-service/internal/room/service/outbox_worker_test.go deleted file mode 100644 index c9b1fd87..00000000 --- a/services/room-service/internal/room/service/outbox_worker_test.go +++ /dev/null @@ -1,154 +0,0 @@ -package service_test - -import ( - "context" - "sync" - "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" -) - -type recordingOutboxPublisher struct { - mu sync.Mutex - envelopes []*roomeventsv1.EventEnvelope -} - -func (p *recordingOutboxPublisher) PublishOutboxEvent(_ context.Context, envelope *roomeventsv1.EventEnvelope) error { - p.mu.Lock() - defer p.mu.Unlock() - p.envelopes = append(p.envelopes, envelope) - return nil -} - -func (p *recordingOutboxPublisher) count() int { - p.mu.Lock() - defer p.mu.Unlock() - return len(p.envelopes) -} - -func TestRobotOutboxWorkerDropsStaleRecordsWithoutPublishing(t *testing.T) { - ctx := appcode.WithContext(context.Background(), appcode.Default) - repository := mysqltest.NewRepository(t) - publisher := &recordingOutboxPublisher{} - svc := roomservice.New(roomservice.Config{ - NodeID: "node-robot-stale-worker", - LeaseTTL: 10 * time.Second, - RobotOutboxPublisher: publisher, - }, router.NewMemoryDirectory(), repository, nil, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) - - stale := robotOutboxRecordForTest(t, "room-robot-stale", time.Now().UTC().Add(-2*time.Minute)) - if err := repository.SaveRobotOutbox(ctx, []outbox.Record{stale}); err != nil { - t.Fatalf("save stale robot outbox failed: %v", err) - } - - if err := svc.ProcessPendingRobotOutbox(ctx, roomservice.OutboxWorkerOptions{ - BatchSize: 10, - PublishTimeout: time.Second, - StaleDiscardAfter: 60 * time.Second, - }); err != nil { - t.Fatalf("process stale robot outbox failed: %v", err) - } - if publisher.count() != 0 { - t.Fatalf("stale robot outbox must not publish, got %d publishes", publisher.count()) - } - if _, exists := repository.RobotOutboxRecord(stale.EventID); exists { - t.Fatalf("stale robot outbox must be deleted") - } -} - -func TestRobotOutboxWorkerPublishesFreshRecords(t *testing.T) { - ctx := appcode.WithContext(context.Background(), appcode.Default) - repository := mysqltest.NewRepository(t) - publisher := &recordingOutboxPublisher{} - svc := roomservice.New(roomservice.Config{ - NodeID: "node-robot-fresh-worker", - LeaseTTL: 10 * time.Second, - RobotOutboxPublisher: publisher, - }, router.NewMemoryDirectory(), repository, nil, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) - - fresh := robotOutboxRecordForTest(t, "room-robot-fresh", time.Now().UTC()) - if err := repository.SaveRobotOutbox(ctx, []outbox.Record{fresh}); err != nil { - t.Fatalf("save fresh robot outbox failed: %v", err) - } - - if err := svc.ProcessPendingRobotOutbox(ctx, roomservice.OutboxWorkerOptions{ - BatchSize: 10, - PublishTimeout: time.Second, - StaleDiscardAfter: 60 * time.Second, - }); err != nil { - t.Fatalf("process fresh robot outbox failed: %v", err) - } - if publisher.count() != 1 { - t.Fatalf("fresh robot outbox must publish once, got %d publishes", publisher.count()) - } - record, exists := repository.RobotOutboxRecord(fresh.EventID) - if !exists || record.Status != outbox.StatusDelivered { - t.Fatalf("fresh robot outbox must be delivered, exists=%v record=%+v", exists, record) - } -} - -func TestCleanupRobotOutboxDeletesExpiredActiveAndTerminalRecords(t *testing.T) { - ctx := appcode.WithContext(context.Background(), appcode.Default) - repository := mysqltest.NewRepository(t) - now := time.Now().UTC() - oldCreated := now.Add(-10 * time.Minute) - freshCreated := now.Add(-30 * time.Second) - oldUpdatedMS := now.Add(-2 * time.Hour).UnixMilli() - freshUpdatedMS := now.UnixMilli() - expiredLockMS := now.Add(-time.Minute).UnixMilli() - futureLockMS := now.Add(time.Hour).UnixMilli() - - oldPending := robotOutboxRecordForTest(t, "room-robot-cleanup", oldCreated) - oldRetryable := robotOutboxRecordForTest(t, "room-robot-cleanup", oldCreated.Add(time.Millisecond)) - oldDeliveringExpired := robotOutboxRecordForTest(t, "room-robot-cleanup", oldCreated.Add(2*time.Millisecond)) - oldDeliveringLocked := robotOutboxRecordForTest(t, "room-robot-cleanup", oldCreated.Add(3*time.Millisecond)) - oldDelivered := robotOutboxRecordForTest(t, "room-robot-cleanup", oldCreated.Add(4*time.Millisecond)) - freshPending := robotOutboxRecordForTest(t, "room-robot-cleanup", freshCreated) - freshDelivered := robotOutboxRecordForTest(t, "room-robot-cleanup", freshCreated.Add(time.Millisecond)) - records := []outbox.Record{oldPending, oldRetryable, oldDeliveringExpired, oldDeliveringLocked, oldDelivered, freshPending, freshDelivered} - if err := repository.SaveRobotOutbox(ctx, records); err != nil { - t.Fatalf("save cleanup robot outbox records failed: %v", err) - } - repository.SetRobotOutboxStatus(oldRetryable.EventID, outbox.StatusRetryable, oldUpdatedMS, nil) - repository.SetRobotOutboxStatus(oldDeliveringExpired.EventID, outbox.StatusDelivering, oldUpdatedMS, &expiredLockMS) - repository.SetRobotOutboxStatus(oldDeliveringLocked.EventID, outbox.StatusDelivering, oldUpdatedMS, &futureLockMS) - repository.SetRobotOutboxStatus(oldDelivered.EventID, outbox.StatusDelivered, oldUpdatedMS, nil) - repository.SetRobotOutboxStatus(freshDelivered.EventID, outbox.StatusDelivered, freshUpdatedMS, nil) - - activeDeleted, terminalDeleted, err := repository.CleanupRobotOutbox(ctx, now.Add(-5*time.Minute).UnixMilli(), now.Add(-time.Hour).UnixMilli(), 50000) - if err != nil { - t.Fatalf("cleanup robot outbox failed: %v", err) - } - if activeDeleted != 3 || terminalDeleted != 1 { - t.Fatalf("cleanup counts mismatch active=%d terminal=%d", activeDeleted, terminalDeleted) - } - for _, record := range []outbox.Record{oldPending, oldRetryable, oldDeliveringExpired, oldDelivered} { - if _, exists := repository.RobotOutboxRecord(record.EventID); exists { - t.Fatalf("expired robot outbox %s should be deleted", record.EventID) - } - } - for _, record := range []outbox.Record{oldDeliveringLocked, freshPending, freshDelivered} { - if _, exists := repository.RobotOutboxRecord(record.EventID); !exists { - t.Fatalf("fresh or locked robot outbox %s should remain", record.EventID) - } - } -} - -func robotOutboxRecordForTest(t *testing.T, roomID string, occurredAt time.Time) outbox.Record { - t.Helper() - record, err := outbox.Build(roomID, "RoomHeatChanged", 1, occurredAt, &roomeventsv1.RoomHeatChanged{ - Delta: 1, - CurrentHeat: 1, - }) - if err != nil { - t.Fatalf("build robot outbox record failed: %v", err) - } - return record -} diff --git a/services/room-service/internal/room/service/pipeline.go b/services/room-service/internal/room/service/pipeline.go index de195eb4..7be14d51 100644 --- a/services/room-service/internal/room/service/pipeline.go +++ b/services/room-service/internal/room/service/pipeline.go @@ -77,8 +77,8 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl if err := s.ensureLeaseStillOwned(ctx, cmd.RoomID(), lease); err != nil { return nil, err } - // robot outbox 只是 App 展示补偿，进入持久化前按短窗口降采样；命令日志、房间状态和主 outbox 不受影响。 - robotOutboxRecords := s.sampleRobotOutboxRecords(result.robotOutboxRecords) + // 机器人展示只是 App 视觉补偿，提交前按短窗口降采样；命令日志、房间状态和主 outbox 不受影响。 + robotDisplayRecords := s.sampleRobotDisplayRecords(result.robotDisplayRecords) saveStartedAt := time.Now() if err := s.repository.SaveMutation(ctx, MutationCommit{ Command: CommandRecord{ @@ -94,7 +94,6 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl CreatedAtMS: now.UnixMilli(), }, OutboxRecords: outboxRecords, - RobotOutboxRecords: robotOutboxRecords, RoomStatus: result.roomStatus, RoomSeatCount: result.roomSeatCount, RoomPasswordHash: result.roomPasswordHash, @@ -109,7 +108,11 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl saveMutationMS += elapsedMS(saveStartedAt) // 保存已提交 outbox 记录，出 Cell 锁后由 outbox worker 异步投递 IM/activity，避免外部慢调用占住房间串行执行队列。 result.outboxRecords = outboxRecords - result.robotOutboxRecords = robotOutboxRecords + result.robotDisplayRecords = robotDisplayRecords + // MySQL 事务成功提交后再发本进程唤醒信号；信号丢失只影响延迟，不影响后续周期扫描补偿。 + s.notifyOutboxCommitted(outboxRecords) + // 机器人展示事件允许极小概率丢失，不写 MySQL outbox；提交成功后后台直发 IM，失败只记录日志。 + s.publishRobotDisplayRecordsBestEffort(ctx, robotDisplayRecords) // 持久化成功后再替换内存状态，避免内存提交领先于恢复来源。 current.ReplaceWith(nextState) diff --git a/services/room-service/internal/room/service/repository.go b/services/room-service/internal/room/service/repository.go index 0a07e2dd..b35e5fc0 100644 --- a/services/room-service/internal/room/service/repository.go +++ b/services/room-service/internal/room/service/repository.go @@ -59,8 +59,6 @@ type MutationCommit struct { Command CommandRecord // OutboxRecords 是本次命令产生的房间外事件。 OutboxRecords []outbox.Record - // RobotOutboxRecords 是机器人房间必要展示事件，必须和命令日志同事务写入独立 robot outbox。 - RobotOutboxRecords []outbox.Record // RoomStatus 非空时同步更新 rooms 和 room_list_entries 的生命周期状态。 RoomStatus string // RoomSeatCount 非 nil 时同步更新 rooms 元数据中的麦位总数。 @@ -733,36 +731,20 @@ type RoomConfigStore interface { UpsertRoomRocketConfig(ctx context.Context, config RoomRocketConfig) error } -// OutboxStore 管理 room_outbox 和 robot outbox 的补偿投递生命周期。 +// OutboxStore 管理 room_outbox 的补偿投递生命周期。 type OutboxStore interface { // SaveOutbox 写入房间外事件，必须和成功命令保持同一提交语义。 SaveOutbox(ctx context.Context, records []outbox.Record) error - // SaveRobotOutbox 写入机器人房间展示事件；火箭进度合并器 flush 后仍按 lane 回到独立 robot outbox。 - SaveRobotOutbox(ctx context.Context, records []outbox.Record) error // ListPendingOutbox 扫描待补偿投递事件。 ListPendingOutbox(ctx context.Context, limit int) ([]outbox.Record, error) - // ListPendingRobotOutbox 扫描机器人房间待补偿展示事件。 - ListPendingRobotOutbox(ctx context.Context, limit int) ([]outbox.Record, error) // ClaimPendingOutbox 抢占一批待投递事件，避免多 worker 同时投递同一批。 ClaimPendingOutbox(ctx context.Context, workerID string, limit int, lockUntilMS int64) ([]outbox.Record, error) - // ClaimPendingRobotOutbox 抢占机器人房间展示事件，避免机器人流量占用主 room_outbox worker。 - ClaimPendingRobotOutbox(ctx context.Context, workerID string, limit int, lockUntilMS int64) ([]outbox.Record, error) // MarkOutboxDelivered 标记事件已经投递成功。 MarkOutboxDelivered(ctx context.Context, eventID string) error - // MarkRobotOutboxDelivered 标记机器人展示事件已经投递成功。 - MarkRobotOutboxDelivered(ctx context.Context, eventID string) error // MarkOutboxRetryable 记录一次投递失败，并写入下一次允许抢占的退避时间。 MarkOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error - // MarkRobotOutboxRetryable 记录机器人展示事件投递失败，并写入独立重试退避。 - MarkRobotOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error // MarkOutboxDead 把超过最大重试次数的事件转入 failed，避免固定污染日志和外部依赖。 MarkOutboxDead(ctx context.Context, eventID string, lastErr string) error - // MarkRobotOutboxDead 把超过最大重试次数的机器人展示事件转入 failed。 - MarkRobotOutboxDead(ctx context.Context, eventID string, lastErr string) error - // DeleteRobotOutbox 直接删除机器人展示事件；过期展示没有补偿价值，不能再占用 MQ/IM 通道。 - DeleteRobotOutbox(ctx context.Context, eventID string) error - // CleanupRobotOutbox 批量删除过期机器人展示事件；非终态和终态分开计数，便于上线后验证清理效果。 - CleanupRobotOutbox(ctx context.Context, activeBeforeMS int64, terminalBeforeMS int64, limit int) (activeDeleted int64, terminalDeleted int64, err error) } // RoomListStore 管理发现页、Mine 页和关系房间流读模型；它不读取 Room Cell 内存。 diff --git a/services/room-service/internal/room/service/robot_display_publisher.go b/services/room-service/internal/room/service/robot_display_publisher.go new file mode 100644 index 00000000..ae1b7f09 --- /dev/null +++ b/services/room-service/internal/room/service/robot_display_publisher.go @@ -0,0 +1,68 @@ +package service + +import ( + "context" + "log/slog" + "time" + + "hyapp/pkg/appcode" + "hyapp/pkg/logx" + "hyapp/services/room-service/internal/room/outbox" +) + +const ( + // robotDisplayPublishTimeout 是机器人展示直发 IM 的单条上限；超时或失败只记日志，不影响已提交房间状态。 + robotDisplayPublishTimeout = 3 * time.Second +) + +func (s *Service) publishRobotDisplayRecordsBestEffort(ctx context.Context, records []outbox.Record) { + if s == nil || len(records) == 0 { + return + } + if s.robotDisplayPublisher == nil { + // 缺少腾讯 IM publisher 时不能阻断房间命令，只把整批展示事件视为 best-effort 投递失败。 + logx.Warn(ctx, "robot_display_im_publish_skipped", + slog.String("node_id", s.nodeID), + slog.Int("record_count", len(records)), + slog.String("reason", "publisher_not_configured"), + ) + return + } + scopedApp := appcode.FromContext(ctx) + copied := append([]outbox.Record(nil), records...) + for index := range copied { + copied[index].AppCode = appcode.Normalize(firstNonEmpty(copied[index].AppCode, scopedApp)) + if copied[index].Envelope != nil { + copied[index].Envelope.AppCode = copied[index].AppCode + } + } + sortOutboxRecords(copied) + + go func(records []outbox.Record) { + for _, record := range records { + if record.Envelope == nil { + logx.Warn(ctx, "robot_display_im_publish_skipped", + slog.String("node_id", s.nodeID), + slog.String("event_id", record.EventID), + slog.String("event_type", record.EventType), + slog.String("room_id", record.RoomID), + slog.String("reason", "nil_envelope"), + ) + continue + } + publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), robotDisplayPublishTimeout) + err := s.robotDisplayPublisher.PublishOutboxEvent(publishCtx, record.Envelope) + cancel() + if err != nil { + // 机器人展示事件允许极小概率丢失；这里只暴露日志，客户端靠房间快照恢复最终状态。 + logx.Warn(ctx, "robot_display_im_publish_failed", + slog.String("node_id", s.nodeID), + slog.String("event_id", record.EventID), + slog.String("event_type", record.EventType), + slog.String("room_id", record.RoomID), + slog.String("error", trimOutboxError(err.Error())), + ) + } + } + }(copied) +} diff --git a/services/room-service/internal/room/service/robot_outbox_sampler.go b/services/room-service/internal/room/service/robot_display_sampler.go similarity index 58% rename from services/room-service/internal/room/service/robot_outbox_sampler.go rename to services/room-service/internal/room/service/robot_display_sampler.go index cea0f412..3bf30608 100644 --- a/services/room-service/internal/room/service/robot_outbox_sampler.go +++ b/services/room-service/internal/room/service/robot_display_sampler.go @@ -9,33 +9,33 @@ import ( ) const ( - // robotOutboxSampleWindow 是 v1 展示降采样窗口：同房间同事件类型 2 秒内只保留一条 robot outbox。 - robotOutboxSampleWindow = 2 * time.Second + // robotDisplaySampleWindow 是 v1 展示降采样窗口：同房间同事件类型 2 秒内只直发一条机器人展示 IM。 + robotDisplaySampleWindow = 2 * time.Second ) -type robotOutboxSampleKey struct { +type robotDisplaySampleKey struct { appCode string roomID string eventType string } -type robotOutboxSampler struct { +type robotDisplaySampler struct { mu sync.Mutex - lastWindow map[robotOutboxSampleKey]int64 + lastWindow map[robotDisplaySampleKey]int64 } -func newRobotOutboxSampler() *robotOutboxSampler { - return &robotOutboxSampler{lastWindow: make(map[robotOutboxSampleKey]int64)} +func newRobotDisplaySampler() *robotDisplaySampler { + return &robotDisplaySampler{lastWindow: make(map[robotDisplaySampleKey]int64)} } -func (s *Service) sampleRobotOutboxRecords(records []outbox.Record) []outbox.Record { - if len(records) == 0 || s == nil || s.robotOutboxSampler == nil { +func (s *Service) sampleRobotDisplayRecords(records []outbox.Record) []outbox.Record { + if len(records) == 0 || s == nil || s.robotDisplaySampler == nil { return records } - return s.robotOutboxSampler.filter(records, robotOutboxSampleWindow) + return s.robotDisplaySampler.filter(records, robotDisplaySampleWindow) } -func (s *robotOutboxSampler) filter(records []outbox.Record, window time.Duration) []outbox.Record { +func (s *robotDisplaySampler) filter(records []outbox.Record, window time.Duration) []outbox.Record { if s == nil || window <= 0 || len(records) == 0 { return records } @@ -47,7 +47,7 @@ func (s *robotOutboxSampler) filter(records []outbox.Record, window time.Duratio var newestWindow int64 for _, record := range records { - key := robotOutboxSampleKey{ + key := robotDisplaySampleKey{ appCode: appcode.Normalize(record.AppCode), roomID: record.RoomID, eventType: record.EventType, @@ -61,7 +61,7 @@ func (s *robotOutboxSampler) filter(records []outbox.Record, window time.Duratio newestWindow = windowStartMS } if last, exists := s.lastWindow[key]; exists && windowStartMS <= last { - // 展示事件过密时只丢 robot outbox 记录；Room Cell 命令、房间热度、榜单和火箭状态已经在进入这里前完成计算。 + // 展示事件过密时只丢机器人 IM 展示；Room Cell 命令、房间热度、榜单和火箭状态已经在进入这里前完成计算。 continue } s.lastWindow[key] = windowStartMS @@ -75,7 +75,7 @@ func (s *robotOutboxSampler) filter(records []outbox.Record, window time.Duratio return out } -func (s *robotOutboxSampler) pruneLocked(beforeWindowMS int64) { +func (s *robotDisplaySampler) pruneLocked(beforeWindowMS int64) { for key, windowStartMS := range s.lastWindow { if windowStartMS < beforeWindowMS { delete(s.lastWindow, key) diff --git a/services/room-service/internal/room/service/room_rocket_progress_coalescer.go b/services/room-service/internal/room/service/room_rocket_progress_coalescer.go index c86c55ed..81569313 100644 --- a/services/room-service/internal/room/service/room_rocket_progress_coalescer.go +++ b/services/room-service/internal/room/service/room_rocket_progress_coalescer.go @@ -2,7 +2,6 @@ package service import ( "context" - "errors" "log/slog" "sync" "time" @@ -95,20 +94,21 @@ func (c *roomRocketProgressCoalescer) enqueue(progress roomRocketProgressPending c.mergeLocked(entry) } -func (c *roomRocketProgressCoalescer) flushDue(ctx context.Context, nowMS int64) error { +func (c *roomRocketProgressCoalescer) flushDue(ctx context.Context, nowMS int64) ([]outbox.Record, error) { if c == nil || c.repository == nil { - return nil + return nil, nil } due := c.takeDue(nowMS) if len(due) == 0 { - return nil + return nil, nil } - if err := c.saveDue(ctx, due); err != nil { - // DB 或上下文失败时把事件放回队列，下一轮继续尝试；新进度已经入队时不会被旧版本覆盖。 - c.requeue(due, nowMS+roomRocketProgressFlushInterval.Milliseconds()) - return err + normalEntries, robotRecords := splitRoomRocketProgressDue(due) + if err := c.saveNormalDue(ctx, normalEntries); err != nil { + // 主 outbox 保存失败时只把主链路进度放回队列；机器人展示允许丢失，仍交给直发 IM best-effort 处理。 + c.requeue(normalEntries, nowMS+roomRocketProgressFlushInterval.Milliseconds()) + return robotRecords, err } - return nil + return robotRecords, nil } func (c *roomRocketProgressCoalescer) takeDue(nowMS int64) []roomRocketProgressEntry { @@ -143,9 +143,9 @@ func (c *roomRocketProgressCoalescer) mergeLocked(entry roomRocketProgressEntry) } } -func (c *roomRocketProgressCoalescer) saveDue(ctx context.Context, entries []roomRocketProgressEntry) error { - normal := make(map[string][]outbox.Record) - robot := make(map[string][]outbox.Record) +func splitRoomRocketProgressDue(entries []roomRocketProgressEntry) ([]roomRocketProgressEntry, []outbox.Record) { + normal := make([]roomRocketProgressEntry, 0, len(entries)) + robotRecords := make([]outbox.Record, 0) for _, entry := range entries { record := entry.record record.AppCode = appcode.Normalize(record.AppCode) @@ -153,21 +153,32 @@ func (c *roomRocketProgressCoalescer) saveDue(ctx context.Context, entries []roo record.Envelope.AppCode = record.AppCode } if entry.key.robotLane { - robot[record.AppCode] = append(robot[record.AppCode], record) + robotRecords = append(robotRecords, record) continue } + normal = append(normal, entry) + } + return normal, robotRecords +} + +func (c *roomRocketProgressCoalescer) saveNormalDue(ctx context.Context, entries []roomRocketProgressEntry) error { + normal := make(map[string][]outbox.Record) + for _, entry := range entries { + record := entry.record + record.AppCode = appcode.Normalize(record.AppCode) + if record.Envelope != nil { + record.Envelope.AppCode = record.AppCode + } normal[record.AppCode] = append(normal[record.AppCode], record) } - var joined error for scopedApp, records := range normal { // repository 会按 context app_code 二次归一，分 app 批量写避免跨租户记录被同一个上下文覆盖。 - joined = errors.Join(joined, c.repository.SaveOutbox(appcode.WithContext(ctx, scopedApp), records)) + if err := c.repository.SaveOutbox(appcode.WithContext(ctx, scopedApp), records); err != nil { + return err + } } - for scopedApp, records := range robot { - joined = errors.Join(joined, c.repository.SaveRobotOutbox(appcode.WithContext(ctx, scopedApp), records)) - } - return joined + return nil } func roomRocketProgressEntryNewer(candidate roomRocketProgressEntry, current roomRocketProgressEntry) bool { @@ -188,15 +199,17 @@ func roomRocketProgressWindowStartMS(createdAtMS int64) int64 { return createdAtMS / windowMS * windowMS } -// FlushRoomRocketProgress 把已经超过 1 秒窗口的火箭进度写回 outbox；测试和后台 worker 共用同一条路径。 +// FlushRoomRocketProgress 把已经超过 1 秒窗口的火箭进度写回主 outbox；机器人进度只 best-effort 直发 IM。 func (s *Service) FlushRoomRocketProgress(ctx context.Context) error { if s == nil || s.roomRocketProgressCoalescer == nil { return nil } - return s.roomRocketProgressCoalescer.flushDue(ctx, s.clock.Now().UTC().UnixMilli()) + robotRecords, err := s.roomRocketProgressCoalescer.flushDue(ctx, s.clock.Now().UTC().UnixMilli()) + s.publishRobotDisplayRecordsBestEffort(ctx, robotRecords) + return err } -// RunRoomRocketProgressFlushWorker 周期性把内存合并后的火箭进度落入 outbox，实际 MQ/IM 投递仍由原 outbox worker 负责。 +// RunRoomRocketProgressFlushWorker 周期性把内存合并后的火箭进度落入主 outbox；机器人展示进度不经过 MySQL。 func (s *Service) RunRoomRocketProgressFlushWorker(ctx context.Context, interval time.Duration) { if interval <= 0 { interval = roomRocketProgressFlushInterval diff --git a/services/room-service/internal/room/service/service.go b/services/room-service/internal/room/service/service.go index 80427ab5..05d48611 100644 --- a/services/room-service/internal/room/service/service.go +++ b/services/room-service/internal/room/service/service.go @@ -35,8 +35,8 @@ type Config struct { RTCUserRemover integration.RTCUserRemover // RoomRocketLaunchScheduler 把倒计时发射唤醒交给外部延迟消息，避免只靠已加载 Cell 扫描。 RoomRocketLaunchScheduler integration.RoomRocketLaunchScheduler - // RobotOutboxPublisher 专门投递机器人房间展示事件，避免机器人流量进入真人房间 room_outbox 通道。 - RobotOutboxPublisher integration.OutboxPublisher + // RobotDisplayPublisher 专门 best-effort 投递机器人房间展示事件；失败只打日志，不进入 MySQL outbox。 + RobotDisplayPublisher integration.OutboxPublisher // HumanRobotPoolProvider 自动读取全站机器人账号和用户国家，用于真人房间机器人运行时按国家分组。 HumanRobotPoolProvider integration.HumanRoomRobotPoolProvider // RoomGiftLeaderboard 是跨房间贡献榜读模型，SendGift 提交后 best-effort 写入。 @@ -79,16 +79,18 @@ type Service struct { syncPublisher integration.RoomEventPublisher // outboxPublisher 是补偿 worker 对外部消费者的异步投递抽象。 outboxPublisher integration.OutboxPublisher - // robotOutboxPublisher 只服务机器人房间必要展示事件，和主 room_outbox 发布器隔离。 - robotOutboxPublisher integration.OutboxPublisher + // robotDisplayPublisher 只服务机器人房间展示事件，和主 room_outbox 持久事实隔离。 + robotDisplayPublisher integration.OutboxPublisher + // outboxWake 提交成功后唤醒主 outbox worker；它只是低延迟提示，事实源仍然是 MySQL room_outbox。 + outboxWake chan struct{} // humanRobotPoolProvider 运行时自动获取全站机器人池，避免后台手工维护国家机器人池。 humanRobotPoolProvider integration.HumanRoomRobotPoolProvider // roomRocketLaunchScheduler 在 ignited outbox 投递成功后安排 launch_at_ms 唤醒。 roomRocketLaunchScheduler integration.RoomRocketLaunchScheduler // roomRocketProgressCoalescer 把高频燃料变化压成每秒每轮一条 outbox，降低 room_outbox/MQ/IM 写放大。 roomRocketProgressCoalescer *roomRocketProgressCoalescer - // robotOutboxSampler 把同房间同展示类型在短窗口内收敛成一条 robot outbox，房间命令和状态仍完整执行。 - robotOutboxSampler *robotOutboxSampler + // robotDisplaySampler 把同房间同展示类型在短窗口内收敛成一条直发 IM，房间命令和状态仍完整执行。 + robotDisplaySampler *robotDisplaySampler // roomGiftLeaderboard 只保存房间维度贡献值 zset，不承载 Room Cell 核心状态。 roomGiftLeaderboard RoomGiftLeaderboardStore // roomListCache 只承载发现页展示顺序；MySQL 仍然是房间列表和置顶事实 owner。 @@ -115,10 +117,6 @@ type Service struct { humanRobotRuntimeMu sync.Mutex // humanRobotRuntimes 保存 room_id -> runtime，避免同一真人房间被重复补机器人。 humanRobotRuntimes map[string]robotRoomRuntime - // robotOutboxCleanupMu 保护清理 worker 的最近执行时间，避免并发 worker 同一时刻重复清理。 - robotOutboxCleanupMu sync.Mutex - // robotOutboxLastCleanup 记录本进程上一轮 robot outbox 清理时间。 - robotOutboxLastCleanup time.Time } type robotRoomRuntime struct { @@ -166,8 +164,8 @@ type mutationResult struct { syncEvent *tencentim.RoomEvent // outboxRecords 是本次命令已经持久化的房间事实，提交后可投给 activity 等外部消费者。 outboxRecords []outbox.Record - // robotOutboxRecords 是机器人房间必要展示事实，只进入 robot outbox/worker，不占主 room_outbox。 - robotOutboxRecords []outbox.Record + // robotDisplayRecords 是机器人房间展示事实，提交成功后 best-effort 直发 Tencent IM，不占主 room_outbox。 + robotDisplayRecords []outbox.Record // roomRocketProgress 是提交成功后才进入内存合并器的火箭进度展示事件，不和送礼命令事务逐笔写 outbox。 roomRocketProgress *roomRocketProgressPending // roomStatus 非空时代表命令提交时需要同步更新 rooms 和列表投影生命周期状态。 @@ -244,11 +242,12 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i luckyGift: luckyGiftClient, syncPublisher: syncPublisher, outboxPublisher: outboxPublisher, - robotOutboxPublisher: cfg.RobotOutboxPublisher, + robotDisplayPublisher: cfg.RobotDisplayPublisher, + outboxWake: make(chan struct{}, 1024), humanRobotPoolProvider: cfg.HumanRobotPoolProvider, roomRocketLaunchScheduler: cfg.RoomRocketLaunchScheduler, roomRocketProgressCoalescer: newRoomRocketProgressCoalescer(repository), - robotOutboxSampler: newRobotOutboxSampler(), + robotDisplaySampler: newRobotDisplaySampler(), roomGiftLeaderboard: cfg.RoomGiftLeaderboard, roomListCache: cfg.RoomListCache, luckyGiftSendLocker: cfg.LuckyGiftSendLocker, diff --git a/services/room-service/internal/room/service/user_gift_value_test.go b/services/room-service/internal/room/service/user_gift_value_test.go index 00e0ca81..047df9e8 100644 --- a/services/room-service/internal/room/service/user_gift_value_test.go +++ b/services/room-service/internal/room/service/user_gift_value_test.go @@ -2,6 +2,7 @@ package service_test import ( "context" + "sync" "testing" "time" @@ -110,7 +111,8 @@ func TestRobotGiftSkipsMainHeatAndRankOutbox(t *testing.T) { ChargeAssetType: "ROBOT_COIN", GiftTypeCode: "robot", }}} - svc := newUserGiftValueTestService(repository, wallet, now, "node-robot-gift-outbox-boundary") + robotDisplay := newRecordingRobotDisplayPublisher() + svc := newUserGiftValueTestService(repository, wallet, now, "node-robot-gift-display-boundary", robotDisplay) roomID := "robot-room-outbox-boundary" senderID := int64(63001) @@ -118,7 +120,6 @@ func TestRobotGiftSkipsMainHeatAndRankOutbox(t *testing.T) { createRocketRoom(t, ctx, svc, roomID, senderID, 9001) joinRocketRoom(t, ctx, svc, roomID, targetID) beforeMain := outboxEventCounts(t, ctx, repository) - beforeRobot := robotOutboxEventCounts(t, ctx, repository) if _, err := svc.RobotSendGift(ctx, roomservice.RobotSendGiftInput{ Meta: rocketMeta(roomID, senderID, "robot-display-only"), @@ -141,24 +142,20 @@ func TestRobotGiftSkipsMainHeatAndRankOutbox(t *testing.T) { t.Fatalf("robot gift must not create main rank outbox, delta=%d counts=%+v", delta, afterMain) } - afterRobot := robotOutboxEventCounts(t, ctx, repository) - if delta := afterRobot["RoomGiftSent"] - beforeRobot["RoomGiftSent"]; delta != 1 { - t.Fatalf("robot gift must keep one robot gift display outbox, delta=%d counts=%+v", delta, afterRobot) - } - if delta := afterRobot["RoomHeatChanged"] - beforeRobot["RoomHeatChanged"]; delta != 1 { - t.Fatalf("robot gift must keep robot heat display outbox, delta=%d counts=%+v", delta, afterRobot) - } - if delta := afterRobot["RoomRankChanged"] - beforeRobot["RoomRankChanged"]; delta != 1 { - t.Fatalf("robot gift must keep robot rank display outbox, delta=%d counts=%+v", delta, afterRobot) - } + waitForRobotDisplayCounts(t, robotDisplay, map[string]int{ + "RoomGiftSent": 1, + "RoomHeatChanged": 1, + "RoomRankChanged": 1, + }) } -func TestRobotGiftDownsamplesRobotOutboxWithoutSkippingRoomState(t *testing.T) { +func TestRobotGiftDownsamplesRobotDisplayWithoutSkippingRoomState(t *testing.T) { ctx := context.Background() repository := mysqltest.NewRepository(t) now := &fixedRoomRocketClock{now: time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)} wallet := &rocketTestWallet{} - svc := newUserGiftValueTestService(repository, wallet, now, "node-robot-gift-outbox-sampler") + robotDisplay := newRecordingRobotDisplayPublisher() + svc := newUserGiftValueTestService(repository, wallet, now, "node-robot-gift-display-sampler", robotDisplay) roomID := "robot-room-outbox-sampler" senderID := int64(63101) @@ -166,7 +163,6 @@ func TestRobotGiftDownsamplesRobotOutboxWithoutSkippingRoomState(t *testing.T) { createRocketRoom(t, ctx, svc, roomID, senderID, 9001) joinRocketRoom(t, ctx, svc, roomID, targetID) beforeMain := outboxEventCounts(t, ctx, repository) - beforeRobot := robotOutboxEventCounts(t, ctx, repository) firstResp, err := svc.RobotSendGift(ctx, roomservice.RobotSendGiftInput{ Meta: rocketMeta(roomID, senderID, "robot-sampler-first"), @@ -205,20 +201,14 @@ func TestRobotGiftDownsamplesRobotOutboxWithoutSkippingRoomState(t *testing.T) { t.Fatalf("robot sampler must not write %s to main outbox, delta=%d counts=%+v", eventType, delta, afterMain) } } - afterRobot := robotOutboxEventCounts(t, ctx, repository) - wantRobotDeltas := map[string]int{ + waitForRobotDisplayCounts(t, robotDisplay, map[string]int{ "RoomGiftSent": 1, "RoomHeatChanged": 1, "RoomRankChanged": 1, - } - for eventType, want := range wantRobotDeltas { - if delta := afterRobot[eventType] - beforeRobot[eventType]; delta != want { - t.Fatalf("robot sampler %s delta=%d want=%d counts=%+v", eventType, delta, want, afterRobot) - } - } + }) } -func TestRealRoomRobotGiftKeepsRoomContributionAndUsesRobotOutbox(t *testing.T) { +func TestRealRoomRobotGiftKeepsRoomContributionAndDirectPublishesRobotDisplay(t *testing.T) { ctx := context.Background() repository := mysqltest.NewRepository(t) now := &fixedRoomRocketClock{now: time.Date(2026, 6, 8, 11, 30, 0, 0, time.UTC)} @@ -231,13 +221,15 @@ func TestRealRoomRobotGiftKeepsRoomContributionAndUsesRobotOutbox(t *testing.T) GiftTypeCode: "normal", }}} leaderboard := &recordingRoomGiftLeaderboard{} + robotDisplay := newRecordingRobotDisplayPublisher() svc := roomservice.New(roomservice.Config{ - NodeID: "node-real-room-robot-outbox", - LeaseTTL: 10 * time.Second, - RankLimit: 20, - SnapshotEveryN: 100, - Clock: now, - RoomGiftLeaderboard: leaderboard, + NodeID: "node-real-room-robot-display", + LeaseTTL: 10 * time.Second, + RankLimit: 20, + SnapshotEveryN: 100, + Clock: now, + RoomGiftLeaderboard: leaderboard, + RobotDisplayPublisher: robotDisplay, }, router.NewMemoryDirectory(), repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) writeRocketConfig(t, ctx, repository, rocketConfigForTest(100, 1_000)) @@ -253,7 +245,6 @@ func TestRealRoomRobotGiftKeepsRoomContributionAndUsesRobotOutbox(t *testing.T) t.Fatalf("target mic up failed: %v", err) } beforeMain := outboxEventCounts(t, ctx, repository) - beforeRobot := robotOutboxEventCounts(t, ctx, repository) resp, err := svc.RobotSendGift(ctx, roomservice.RobotSendGiftInput{ Meta: rocketMeta(roomID, senderID, "real-room-robot-gift"), @@ -295,20 +286,14 @@ func TestRealRoomRobotGiftKeepsRoomContributionAndUsesRobotOutbox(t *testing.T) t.Fatalf("real room robot gift must not write %s to main outbox, delta=%d counts=%+v", eventType, delta, afterMain) } } - afterRobot := robotOutboxEventCounts(t, ctx, repository) - wantRobotDeltas := map[string]int{ + waitForRobotDisplayCounts(t, robotDisplay, map[string]int{ "RoomGiftSent": 1, "RoomHeatChanged": 1, "RoomRankChanged": 1, "RoomRobotLuckyGiftDrawn": 1, "RoomRocketFuelChanged": 1, "RoomRocketIgnited": 1, - } - for eventType, want := range wantRobotDeltas { - if delta := afterRobot[eventType] - beforeRobot[eventType]; delta != want { - t.Fatalf("real room robot gift robot outbox %s delta=%d want=%d counts=%+v", eventType, delta, want, afterRobot) - } - } + }) } func TestRoomUserGiftValueRecoversFromCommandLog(t *testing.T) { @@ -385,26 +370,74 @@ func outboxEventCounts(t *testing.T, ctx context.Context, repository *mysqltest. return counts } -func robotOutboxEventCounts(t *testing.T, ctx context.Context, repository *mysqltest.Repository) map[string]int { - t.Helper() - records, err := repository.ListPendingRobotOutbox(ctx, 200) - if err != nil { - t.Fatalf("list pending robot outbox failed: %v", err) +type recordingRobotDisplayPublisher struct { + mu sync.Mutex + envelopes []*roomeventsv1.EventEnvelope + wake chan struct{} +} + +func newRecordingRobotDisplayPublisher() *recordingRobotDisplayPublisher { + return &recordingRobotDisplayPublisher{wake: make(chan struct{}, 32)} +} + +func (p *recordingRobotDisplayPublisher) PublishOutboxEvent(_ context.Context, envelope *roomeventsv1.EventEnvelope) error { + p.mu.Lock() + p.envelopes = append(p.envelopes, proto.Clone(envelope).(*roomeventsv1.EventEnvelope)) + p.mu.Unlock() + select { + case p.wake <- struct{}{}: + default: } - counts := make(map[string]int, len(records)) - for _, record := range records { - counts[record.EventType]++ + return nil +} + +func (p *recordingRobotDisplayPublisher) counts() map[string]int { + p.mu.Lock() + defer p.mu.Unlock() + counts := make(map[string]int, len(p.envelopes)) + for _, envelope := range p.envelopes { + counts[envelope.GetEventType()]++ } return counts } -func newUserGiftValueTestService(repository *mysqltest.Repository, wallet *rocketTestWallet, clock *fixedRoomRocketClock, nodeID string) *roomservice.Service { +func waitForRobotDisplayCounts(t *testing.T, publisher *recordingRobotDisplayPublisher, want map[string]int) map[string]int { + t.Helper() + + deadline := time.NewTimer(2 * time.Second) + defer deadline.Stop() + for { + counts := publisher.counts() + matched := true + for eventType, wantCount := range want { + if counts[eventType] < wantCount { + matched = false + break + } + } + if matched { + return counts + } + select { + case <-publisher.wake: + case <-deadline.C: + t.Fatalf("robot display IM events mismatch: want at least %+v got %+v", want, counts) + } + } +} + +func newUserGiftValueTestService(repository *mysqltest.Repository, wallet *rocketTestWallet, clock *fixedRoomRocketClock, nodeID string, robotDisplayPublisher ...integration.OutboxPublisher) *roomservice.Service { + var publisher integration.OutboxPublisher + if len(robotDisplayPublisher) > 0 { + publisher = robotDisplayPublisher[0] + } return roomservice.New(roomservice.Config{ - NodeID: nodeID, - LeaseTTL: 10 * time.Second, - RankLimit: 20, - SnapshotEveryN: 100, - Clock: clock, + NodeID: nodeID, + LeaseTTL: 10 * time.Second, + RankLimit: 20, + SnapshotEveryN: 100, + Clock: clock, + RobotDisplayPublisher: publisher, }, router.NewMemoryDirectory(), repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) } diff --git a/services/room-service/internal/storage/mysql/command_log.go b/services/room-service/internal/storage/mysql/command_log.go index cf16cd16..4ae86148 100644 --- a/services/room-service/internal/storage/mysql/command_log.go +++ b/services/room-service/internal/storage/mysql/command_log.go @@ -89,16 +89,11 @@ func (r *Repository) SaveMutation(ctx context.Context, commit roomservice.Mutati return err } - if err := r.insertRoomOutboxRecords(ctx, tx, roomOutboxTable, commit.OutboxRecords); err != nil { + if err := r.insertRoomOutboxRecords(ctx, tx, commit.OutboxRecords); err != nil { // 任一事件失败就回滚整批，避免 GiftSent/HeatChanged/RankChanged 部分缺失。 _ = tx.Rollback() return err } - if err := r.insertRoomOutboxRecords(ctx, tx, roomRobotOutboxTable, commit.RobotOutboxRecords); err != nil { - // robot outbox 也和命令日志同事务提交，避免机器人礼物状态变了但客户端展示事件丢失。 - _ = tx.Rollback() - return err - } if strings.TrimSpace(commit.RoomStatus) != "" { if _, err := tx.ExecContext(ctx, @@ -281,8 +276,7 @@ func (r *Repository) SaveMutation(ctx context.Context, commit roomservice.Mutati return tx.Commit() } -func (r *Repository) insertRoomOutboxRecords(ctx context.Context, tx *sql.Tx, table string, records []outbox.Record) error { - table = roomOutboxTableName(table) +func (r *Repository) insertRoomOutboxRecords(ctx context.Context, tx *sql.Tx, records []outbox.Record) error { if len(records) == 0 { return nil } @@ -297,7 +291,7 @@ func (r *Repository) insertRoomOutboxRecords(ctx context.Context, tx *sql.Tx, ta return err } if _, err := tx.ExecContext(ctx, - `INSERT INTO `+table+` (app_code, event_id, event_type, room_id, status, envelope, created_at_ms, retry_count, next_retry_at_ms, last_error, updated_at_ms) + `INSERT INTO room_outbox (app_code, event_id, event_type, room_id, status, envelope, created_at_ms, retry_count, next_retry_at_ms, last_error, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE event_id = event_id`, record.AppCode, diff --git a/services/room-service/internal/storage/mysql/outbox.go b/services/room-service/internal/storage/mysql/outbox.go index 4fd7e66b..16f94a13 100644 --- a/services/room-service/internal/storage/mysql/outbox.go +++ b/services/room-service/internal/storage/mysql/outbox.go @@ -13,24 +13,14 @@ import ( "hyapp/services/room-service/internal/room/outbox" ) -// SaveOutbox 追加待投递 outbox 事件。 +// SaveOutbox 追加待投递 room_outbox 事件；机器人展示事件已经改为提交后 best-effort 直发 IM，不再进入 MySQL。 func (r *Repository) SaveOutbox(ctx context.Context, records []outbox.Record) error { - return r.saveOutboxInTable(ctx, roomOutboxTable, records) -} - -// SaveRobotOutbox 追加机器人房间展示 outbox 事件。 -func (r *Repository) SaveRobotOutbox(ctx context.Context, records []outbox.Record) error { - return r.saveOutboxInTable(ctx, roomRobotOutboxTable, records) -} - -func (r *Repository) saveOutboxInTable(ctx context.Context, table string, records []outbox.Record) error { - table = roomOutboxTableName(table) if len(records) == 0 { // 没有事件时保持无副作用，方便领域层统一调用。 return nil } - // 多条事件用一个事务写入，保证同一命令产生的事件不会部分入库。 + // 多条事件用一个事务写入，保证同一命令产生的主链路事件不会部分入库。 tx, err := r.db.BeginTx(ctx, nil) if err != nil { return err @@ -49,7 +39,7 @@ func (r *Repository) saveOutboxInTable(ctx context.Context, table string, record } if _, err := tx.ExecContext(ctx, - `INSERT INTO `+table+` (app_code, event_id, event_type, room_id, status, envelope, created_at_ms, retry_count, next_retry_at_ms, last_error, updated_at_ms) + `INSERT INTO room_outbox (app_code, event_id, event_type, room_id, status, envelope, created_at_ms, retry_count, next_retry_at_ms, last_error, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE event_id = event_id`, record.AppCode, @@ -76,27 +66,16 @@ func (r *Repository) saveOutboxInTable(ctx context.Context, table string, record // ListPendingOutbox 返回待投递事件。 func (r *Repository) ListPendingOutbox(ctx context.Context, limit int) ([]outbox.Record, error) { - return r.listPendingOutboxFromTable(ctx, roomOutboxTable, limit) -} - -// ListPendingRobotOutbox 返回机器人房间待投递展示事件。 -func (r *Repository) ListPendingRobotOutbox(ctx context.Context, limit int) ([]outbox.Record, error) { - return r.listPendingOutboxFromTable(ctx, roomRobotOutboxTable, limit) -} - -func (r *Repository) listPendingOutboxFromTable(ctx context.Context, table string, limit int) ([]outbox.Record, error) { if limit <= 0 { // 默认批量上限避免调用方传 0 导致全表扫描。 limit = 100 } - table = roomOutboxTableName(table) - pendingIndex, _ := roomOutboxIndexes(table) // 按创建时间顺序扫描 pending，尽量保持事件投递顺序。 nowMS := time.Now().UTC().UnixMilli() rows, err := r.db.QueryContext(ctx, `SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '') - FROM `+table+` FORCE INDEX (`+pendingIndex+`) + FROM room_outbox FORCE INDEX (idx_room_outbox_pending) WHERE app_code = ? AND status IN (?, ?) AND next_retry_at_ms <= ? ORDER BY created_at_ms ASC @@ -145,23 +124,12 @@ func (r *Repository) listPendingOutboxFromTable(ctx context.Context, table strin // ClaimPendingOutbox 抢占一批待投递事件，避免多 worker 同时投递同一批。 func (r *Repository) ClaimPendingOutbox(ctx context.Context, workerID string, limit int, lockUntilMS int64) ([]outbox.Record, error) { - return r.claimPendingOutboxFromTable(ctx, roomOutboxTable, workerID, limit, lockUntilMS) -} - -// ClaimPendingRobotOutbox 抢占机器人房间展示事件，避免机器人流量占用主 room_outbox worker。 -func (r *Repository) ClaimPendingRobotOutbox(ctx context.Context, workerID string, limit int, lockUntilMS int64) ([]outbox.Record, error) { - return r.claimPendingOutboxFromTable(ctx, roomRobotOutboxTable, workerID, limit, lockUntilMS) -} - -func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table string, workerID string, limit int, lockUntilMS int64) ([]outbox.Record, error) { if limit <= 0 { limit = 100 } if strings.TrimSpace(workerID) == "" { workerID = "room-outbox-worker" } - table = roomOutboxTableName(table) - pendingIndex, claimIndex := roomOutboxIndexes(table) tx, err := r.db.BeginTx(ctx, nil) if err != nil { @@ -172,15 +140,14 @@ func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table stri appCode := appcode.FromContext(ctx) records := make([]outbox.Record, 0, limit) // pending 新事件的 next_retry_at_ms 固定为 0，真正决定可见性的是未被抢占的 lock_until_ms； - // 因此 pending 分支必须走 claim 索引按 created_at_ms 直接取最早记录，避免在机器人 outbox 堆积时用 - // next_retry_at_ms range 扫描数百万历史行再 filesort。retryable 才按 next_retry_at_ms 到期时间排序。 + // 因此 pending 分支必须走 claim 索引按 created_at_ms 直接取最早记录，避免堆积时 range 扫描历史行。 claimBranches := []struct { query string args []any }{ { query: `SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '') - FROM ` + table + ` FORCE INDEX (` + claimIndex + `) + FROM room_outbox FORCE INDEX (idx_room_outbox_claim) WHERE app_code = ? AND status = ? AND lock_until_ms IS NULL AND next_retry_at_ms <= ? ORDER BY created_at_ms ASC, event_id ASC LIMIT ? FOR UPDATE SKIP LOCKED`, @@ -188,7 +155,7 @@ func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table stri }, { query: `SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '') - FROM ` + table + ` FORCE INDEX (` + pendingIndex + `) + FROM room_outbox FORCE INDEX (idx_room_outbox_pending) WHERE app_code = ? AND status = ? AND next_retry_at_ms <= ? AND lock_until_ms IS NULL ORDER BY next_retry_at_ms ASC, created_at_ms ASC, event_id ASC LIMIT ? FOR UPDATE SKIP LOCKED`, @@ -196,7 +163,7 @@ func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table stri }, { query: `SELECT app_code, event_id, event_type, room_id, status, worker_id, lock_until_ms, envelope, created_at_ms, retry_count, next_retry_at_ms, COALESCE(last_error, '') - FROM ` + table + ` FORCE INDEX (` + claimIndex + `) + FROM room_outbox FORCE INDEX (idx_room_outbox_claim) WHERE app_code = ? AND status = ? AND lock_until_ms <= ? ORDER BY created_at_ms ASC, event_id ASC LIMIT ? FOR UPDATE SKIP LOCKED`, @@ -219,7 +186,7 @@ func (r *Repository) claimPendingOutboxFromTable(ctx context.Context, table stri for _, record := range records { if _, err := tx.ExecContext(ctx, - `UPDATE `+table+` + `UPDATE room_outbox SET status = ?, worker_id = ?, lock_until_ms = ?, updated_at_ms = ? WHERE app_code = ? AND event_id = ?`, outbox.StatusDelivering, @@ -283,19 +250,9 @@ func (r *Repository) queryRoomOutboxRecords(ctx context.Context, tx *sql.Tx, que // MarkOutboxDelivered 标记事件投递成功。 func (r *Repository) MarkOutboxDelivered(ctx context.Context, eventID string) error { - return r.markOutboxDeliveredInTable(ctx, roomOutboxTable, eventID) -} - -// MarkRobotOutboxDelivered 标记机器人展示事件已经投递成功。 -func (r *Repository) MarkRobotOutboxDelivered(ctx context.Context, eventID string) error { - return r.markOutboxDeliveredInTable(ctx, roomRobotOutboxTable, eventID) -} - -func (r *Repository) markOutboxDeliveredInTable(ctx context.Context, table string, eventID string) error { - table = roomOutboxTableName(table) // 成功后清空 last_error，后续扫描不再返回该事件。 _, err := r.db.ExecContext(ctx, - `UPDATE `+table+` + `UPDATE room_outbox SET status = ?, worker_id = '', lock_until_ms = NULL, last_error = NULL, updated_at_ms = ? WHERE app_code = ? AND event_id = ?`, outbox.StatusDelivered, @@ -309,19 +266,9 @@ func (r *Repository) markOutboxDeliveredInTable(ctx context.Context, table strin // MarkOutboxRetryable 标记事件投递失败并写入下一次重试时间。 func (r *Repository) MarkOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error { - return r.markOutboxRetryableInTable(ctx, roomOutboxTable, eventID, lastErr, nextRetryAtMS) -} - -// MarkRobotOutboxRetryable 记录机器人展示事件投递失败，并写入独立重试退避。 -func (r *Repository) MarkRobotOutboxRetryable(ctx context.Context, eventID string, lastErr string, nextRetryAtMS int64) error { - return r.markOutboxRetryableInTable(ctx, roomRobotOutboxTable, eventID, lastErr, nextRetryAtMS) -} - -func (r *Repository) markOutboxRetryableInTable(ctx context.Context, table string, eventID string, lastErr string, nextRetryAtMS int64) error { - table = roomOutboxTableName(table) // 失败转入 retryable，worker 只有到 next_retry_at_ms 后才会重新抢占。 _, err := r.db.ExecContext(ctx, - `UPDATE `+table+` + `UPDATE room_outbox SET status = ?, worker_id = '', lock_until_ms = NULL, retry_count = retry_count + 1, next_retry_at_ms = ?, last_error = ?, updated_at_ms = ? WHERE app_code = ? AND event_id = ?`, outbox.StatusRetryable, @@ -337,18 +284,8 @@ func (r *Repository) markOutboxRetryableInTable(ctx context.Context, table strin // MarkOutboxDead 把超过最大重试次数的事件转入 failed。 func (r *Repository) MarkOutboxDead(ctx context.Context, eventID string, lastErr string) error { - return r.markOutboxDeadInTable(ctx, roomOutboxTable, eventID, lastErr) -} - -// MarkRobotOutboxDead 把超过最大重试次数的机器人展示事件转入 failed。 -func (r *Repository) MarkRobotOutboxDead(ctx context.Context, eventID string, lastErr string) error { - return r.markOutboxDeadInTable(ctx, roomRobotOutboxTable, eventID, lastErr) -} - -func (r *Repository) markOutboxDeadInTable(ctx context.Context, table string, eventID string, lastErr string) error { - table = roomOutboxTableName(table) _, err := r.db.ExecContext(ctx, - `UPDATE `+table+` + `UPDATE room_outbox SET status = ?, worker_id = '', lock_until_ms = NULL, last_error = ?, updated_at_ms = ? WHERE app_code = ? AND event_id = ?`, outbox.StatusFailed, @@ -360,119 +297,3 @@ func (r *Repository) markOutboxDeadInTable(ctx context.Context, table string, ev return err } - -// DeleteRobotOutbox 直接删除一条机器人展示事件。 -func (r *Repository) DeleteRobotOutbox(ctx context.Context, eventID string) error { - // robot outbox 只承载 App 展示补偿；超过业务展示窗口后删除比标记 failed 更能降低后续扫描成本。 - _, err := r.db.ExecContext(ctx, - `DELETE FROM `+roomRobotOutboxTable+` - WHERE app_code = ? AND event_id = ?`, - appcode.FromContext(ctx), - eventID, - ) - return err -} - -// CleanupRobotOutbox 批量删除机器人展示 outbox 的过期记录，不做归档。 -func (r *Repository) CleanupRobotOutbox(ctx context.Context, activeBeforeMS int64, terminalBeforeMS int64, limit int) (int64, int64, error) { - if limit <= 0 { - // 清理必须有批量上限，避免误配置导致一次 DELETE 长时间持锁。 - limit = 50000 - } - - appCode := appcode.FromContext(ctx) - nowMS := time.Now().UTC().UnixMilli() - var activeDeleted int64 - var terminalDeleted int64 - if activeBeforeMS > 0 { - // pending/retryable 没有正在处理的 worker，超过展示窗口后可以直接删除。 - deleted, err := r.deleteRobotOutboxBatch(ctx, ` - DELETE FROM `+roomRobotOutboxTable+` - WHERE app_code = ? - AND status IN (?, ?) - AND created_at_ms < ? - ORDER BY created_at_ms ASC, event_id ASC - LIMIT ?`, - appCode, - outbox.StatusPending, - outbox.StatusRetryable, - activeBeforeMS, - limit, - ) - if err != nil { - return activeDeleted, terminalDeleted, err - } - activeDeleted += deleted - - // delivering 可能正被其他 worker 处理，只删除锁已过期或历史异常无锁的旧记录。 - deleted, err = r.deleteRobotOutboxBatch(ctx, ` - DELETE FROM `+roomRobotOutboxTable+` - WHERE app_code = ? - AND status = ? - AND created_at_ms < ? - AND (lock_until_ms IS NULL OR lock_until_ms <= ?) - ORDER BY created_at_ms ASC, event_id ASC - LIMIT ?`, - appCode, - outbox.StatusDelivering, - activeBeforeMS, - nowMS, - limit, - ) - if err != nil { - return activeDeleted, terminalDeleted, err - } - activeDeleted += deleted - } - - if terminalBeforeMS > 0 { - // delivered/failed 对机器人展示没有审计价值，保留短窗口方便排障后即可删除。 - deleted, err := r.deleteRobotOutboxBatch(ctx, ` - DELETE FROM `+roomRobotOutboxTable+` - WHERE app_code = ? - AND status IN (?, ?) - AND updated_at_ms < ? - ORDER BY updated_at_ms ASC, event_id ASC - LIMIT ?`, - appCode, - outbox.StatusDelivered, - outbox.StatusFailed, - terminalBeforeMS, - limit, - ) - if err != nil { - return activeDeleted, terminalDeleted, err - } - terminalDeleted += deleted - } - - return activeDeleted, terminalDeleted, nil -} - -func (r *Repository) deleteRobotOutboxBatch(ctx context.Context, query string, args ...any) (int64, error) { - result, err := r.db.ExecContext(ctx, query, args...) - if err != nil { - return 0, err - } - deleted, err := result.RowsAffected() - if err != nil { - return 0, err - } - return deleted, nil -} - -func roomOutboxTableName(table string) string { - switch table { - case roomRobotOutboxTable: - return roomRobotOutboxTable - default: - return roomOutboxTable - } -} - -func roomOutboxIndexes(table string) (pending string, claim string) { - if roomOutboxTableName(table) == roomRobotOutboxTable { - return "idx_room_robot_outbox_pending", "idx_room_robot_outbox_claim" - } - return "idx_room_outbox_pending", "idx_room_outbox_claim" -} diff --git a/services/room-service/internal/storage/mysql/repository.go b/services/room-service/internal/storage/mysql/repository.go index 3f3ea5e8..5a49dab8 100644 --- a/services/room-service/internal/storage/mysql/repository.go +++ b/services/room-service/internal/storage/mysql/repository.go @@ -15,8 +15,7 @@ import ( ) const ( - roomOutboxTable = "room_outbox" - roomRobotOutboxTable = "room_robot_outbox" + roomOutboxTable = "room_outbox" ) // Repository 把 room-service 的权威状态持久化到 MySQL。 diff --git a/services/room-service/internal/storage/mysql/schema.go b/services/room-service/internal/storage/mysql/schema.go index 90b8d2f1..99952b63 100644 --- a/services/room-service/internal/storage/mysql/schema.go +++ b/services/room-service/internal/storage/mysql/schema.go @@ -146,26 +146,6 @@ func (r *Repository) Migrate(ctx context.Context) error { 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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, - `CREATE TABLE IF NOT EXISTS room_robot_outbox ( - app_code VARCHAR(32) NOT NULL, - event_id VARCHAR(128) NOT NULL, - event_type VARCHAR(64) NOT NULL, - room_id VARCHAR(64) NOT NULL, - status VARCHAR(32) NOT NULL, - worker_id VARCHAR(128) NOT NULL DEFAULT '', - lock_until_ms BIGINT NULL, - envelope LONGBLOB NOT NULL, - created_at_ms BIGINT NOT NULL, - retry_count INT NOT NULL DEFAULT 0, - next_retry_at_ms BIGINT NOT NULL DEFAULT 0, - last_error TEXT NULL, - updated_at_ms BIGINT NOT NULL, - PRIMARY KEY (app_code, event_id), - KEY idx_room_robot_outbox_pending (app_code, status, next_retry_at_ms, created_at_ms, event_id), - KEY idx_room_robot_outbox_claim (app_code, status, lock_until_ms, created_at_ms, event_id), - KEY idx_room_robot_outbox_event_created (app_code, event_type, created_at_ms, event_id), - KEY idx_room_robot_outbox_retention (app_code, status, updated_at_ms, event_id) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS room_user_feed_entries ( app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, diff --git a/services/room-service/internal/testutil/mysqltest/mysqltest.go b/services/room-service/internal/testutil/mysqltest/mysqltest.go index b62bb615..457e89e9 100644 --- a/services/room-service/internal/testutil/mysqltest/mysqltest.go +++ b/services/room-service/internal/testutil/mysqltest/mysqltest.go @@ -145,13 +145,6 @@ func (r *Repository) OutboxRecord(eventID string) (outbox.Record, bool) { return r.outboxRecord("room_outbox", eventID) } -// RobotOutboxRecord reads one robot outbox row, including delivered rows that normal scans skip. -func (r *Repository) RobotOutboxRecord(eventID string) (outbox.Record, bool) { - r.t.Helper() - - return r.outboxRecord("room_robot_outbox", eventID) -} - func (r *Repository) outboxRecord(table string, eventID string) (outbox.Record, bool) { r.t.Helper() @@ -183,23 +176,6 @@ func (r *Repository) outboxRecord(table string, eventID string) (outbox.Record, return record, true } -// SetRobotOutboxStatus rewrites robot outbox lifecycle fields for cleanup and worker tests. -func (r *Repository) SetRobotOutboxStatus(eventID string, status string, updatedAtMS int64, lockUntilMS *int64) { - r.t.Helper() - - var lockUntil any - if lockUntilMS != nil { - lockUntil = *lockUntilMS - } - if _, err := r.schema.DB.ExecContext(context.Background(), ` - UPDATE room_robot_outbox - SET status = ?, updated_at_ms = ?, lock_until_ms = ? - WHERE app_code = ? AND event_id = ? - `, status, updatedAtMS, lockUntil, appcode.Default, eventID); err != nil { - r.t.Fatalf("update robot outbox status failed: %v", err) - } -} - func initDBPath(t testing.TB) string { t.Helper()