diff --git a/services/notice-service/internal/modules/cpnotice/service.go b/services/notice-service/internal/modules/cpnotice/service.go index 3712a5fd..abe70d60 100644 --- a/services/notice-service/internal/modules/cpnotice/service.go +++ b/services/notice-service/internal/modules/cpnotice/service.go @@ -86,9 +86,15 @@ func (s *Service) ProcessUserOutboxMessage(ctx context.Context, message usermq.U } for _, delivery := range deliveries { claimedDelivery, claimed, err := s.repository.ClaimCPDelivery(ctx, options.WorkerID, delivery, options.LockTTL) - if err != nil || !claimed { + if err != nil { return true, err } + if !claimed { + // 一条 user_outbox 事实会展开为多个按 channel 独立幂等的投递。当前 channel 已终态、 + // 被其他 worker 持有或尚未到重试时间时,只能跳过这一项;提前返回会把整条 MQ + // 消息确认掉,导致其后尚未落库或仍需补发的 channel 永久漏发。 + continue + } if err := s.publishDelivery(ctx, claimedDelivery, options); err != nil { return true, err } diff --git a/services/notice-service/internal/modules/cpnotice/service_test.go b/services/notice-service/internal/modules/cpnotice/service_test.go index 590e27f0..0821b35b 100644 --- a/services/notice-service/internal/modules/cpnotice/service_test.go +++ b/services/notice-service/internal/modules/cpnotice/service_test.go @@ -47,6 +47,39 @@ func TestProcessApplicationCreatedPublishesC2CAndBothRoomIMs(t *testing.T) { } } +func TestProcessApplicationCreatedContinuesAfterAlreadyFinalChannel(t *testing.T) { + // 模拟同一事实的 C2C 已在上一次消费中成功,而两个房间 channel 尚未处理;重放必须 + // 依赖 channel 级幂等跳过 C2C,并继续补齐后续副作用,不能把首个终态误判为整条完成。 + repo := &fakeRepository{notClaimedChannels: map[string]bool{channelTencentIMC2C: true}} + publisher := &fakePublisher{} + service := New(Config{NodeID: "notice-test"}, repo, publisher) + + handled, err := service.ProcessUserOutboxMessage(context.Background(), usermq.UserOutboxMessage{ + AppCode: "lalu", + EventID: "cp_application:created:1002:partial-replay", + EventType: eventApplicationCreated, + AggregateType: "cp_application", + AggregateID: 1002, + PayloadJSON: applicationPayloadJSON(), + OccurredAtMS: 1700000000000, + }, CPNoticeWorkerOptions{PublishTimeout: time.Second}) + if err != nil || !handled { + t.Fatalf("ProcessUserOutboxMessage failed handled=%v err=%v", handled, err) + } + if len(repo.claimedChannels) != 3 { + t.Fatalf("expected every channel to be inspected, got %+v", repo.claimedChannels) + } + if len(publisher.userMessages) != 0 { + t.Fatalf("already delivered c2c must not be published again: %+v", publisher.userMessages) + } + if len(publisher.groupMessages) != 2 || publisher.groupMessages[0].GroupID != "room-100" || publisher.groupMessages[1].GroupID != "room-100" { + t.Fatalf("expected both remaining room deliveries, got %+v", publisher.groupMessages) + } + if len(repo.delivered) != 2 || repo.delivered[0].Channel != channelTencentIMRoom || repo.delivered[1].Channel != channelTencentIMRoomSender { + t.Fatalf("expected only remaining channels to become delivered, got %+v", repo.delivered) + } +} + func TestProcessApplicationAcceptedPublishesMutualC2CAndRoomIM(t *testing.T) { repo := &fakeRepository{} publisher := &fakePublisher{} @@ -187,12 +220,18 @@ func TestPublishFailureMarksRetryable(t *testing.T) { } type fakeRepository struct { - delivered []CPDelivery - retryable []CPDelivery - failed []CPDelivery + notClaimedChannels map[string]bool + claimedChannels []string + delivered []CPDelivery + retryable []CPDelivery + failed []CPDelivery } func (r *fakeRepository) ClaimCPDelivery(_ context.Context, _ string, delivery CPDelivery, _ time.Duration) (CPDelivery, bool, error) { + r.claimedChannels = append(r.claimedChannels, delivery.Channel) + if r.notClaimedChannels[delivery.Channel] { + return CPDelivery{}, false, nil + } return delivery, true, nil }