diff --git a/supply-api/internal/app/background.go b/supply-api/internal/app/background.go index 02bf552b..66a83b93 100644 --- a/supply-api/internal/app/background.go +++ b/supply-api/internal/app/background.go @@ -254,13 +254,15 @@ func runPartitionMaintenanceLoop(ctx context.Context, logger logging.Logger, man for { select { case <-ctx.Done(): + logger.Info("分区维护: 已停止 (context cancelled)", nil) return case <-ticker.C: - if err := manager.EnsureFuturePartitions(context.Background()); err != nil { + // P3-D-01: 使用 ctx 而非 context.Background() 以支持取消 + if err := manager.EnsureFuturePartitions(ctx); err != nil { warnf(logger, "分区维护: 预创建未来分区失败: %v", err) } for _, tableName := range tuning.partitionedTables { - if _, err := manager.DropOldPartitions(context.Background(), tableName); err != nil { + if _, err := manager.DropOldPartitions(ctx, tableName); err != nil { warnf(logger, "分区维护: 清理过期分区失败 (%s): %v", tableName, err) } } diff --git a/supply-api/internal/outbox/outbox.go b/supply-api/internal/outbox/outbox.go index 60660353..f8cb8178 100644 --- a/supply-api/internal/outbox/outbox.go +++ b/supply-api/internal/outbox/outbox.go @@ -12,20 +12,23 @@ import ( ) // OutboxProcessorRunner Outbox处理器运行器 +// P3-D-02: 增加 drainDone channel 支持优雅停止 type OutboxProcessorRunner struct { - repo outboxRepository - msgBroker messaging.MessageBroker - stats messaging.OutboxStats - stopCh chan struct{} - batchSize int - interval time.Duration + repo outboxRepository + msgBroker messaging.MessageBroker + stats messaging.OutboxStats + stopCh chan struct{} + drainDone chan struct{} + batchSize int + interval time.Duration + processing bool // 当前是否在处理中（用于 drain 等待） } type outboxRepository interface { - FetchAndLock(ctx context.Context, limit int) ([]*repository.OutboxEvent, error) - MarkCompleted(ctx context.Context, eventID string) error - MarkFailed(ctx context.Context, eventID string, errorMsg string, nextRetryAt *time.Time) error - MoveToDeadLetter(ctx context.Context, event *repository.OutboxEvent, errorMsg string) error + FetchAndLock(bc context.Context, limit int) ([]*repository.OutboxEvent, error) + MarkCompleted(bc context.Context, eventID string) error + MarkFailed(bc context.Context, eventID string, errorMsg string, nextRetryAt *time.Time) error + MoveToDeadLetter(bc context.Context, event *repository.OutboxEvent, errorMsg string) error } // NewOutboxProcessorRunner 创建Outbox处理器运行器 @@ -39,6 +42,7 @@ func NewOutboxProcessorRunner( msgBroker: msgBroker, stats: stats, stopCh: make(chan struct{}), + drainDone: make(chan struct{}), batchSize: 100, interval: 1 * time.Second, } @@ -50,28 +54,46 @@ func (r *OutboxProcessorRunner) Start(ctx context.Context) { logger.Info("OutboxProcessor started", nil) ticker := time.NewTicker(r.interval) defer ticker.Stop() + defer close(r.drainDone) // P3-D-02: 通知所有等待方处理已完成 for { select { case <-ctx.Done(): - logger := logging.NewLogger("supply-api", logging.LogLevelInfo) - logger.Info("OutboxProcessor stopping due to context cancellation", nil) + logger.Info("OutboxProcessor: context cancelled, waiting for current batch to finish...", nil) + r.waitForProcessingDone() + logger.Info("OutboxProcessor: stopped (context cancelled)", nil) return case <-r.stopCh: - logger := logging.NewLogger("supply-api", logging.LogLevelInfo) - logger.Info("OutboxProcessor stopping", nil) + logger.Info("OutboxProcessor: stop requested, waiting for current batch to finish...", nil) + r.waitForProcessingDone() + logger.Info("OutboxProcessor: stopped (stopCh)", nil) return case <-ticker.C: + r.processing = true if err := r.process(ctx); err != nil { - logger := logging.NewLogger("supply-api", logging.LogLevelError) logger.Error("OutboxProcessor error", map[string]interface{}{ "error": err.Error(), }) } + r.processing = false } } } +// waitForProcessingDone 等待当前处理批次完成（如果有） +func (r *OutboxProcessorRunner) waitForProcessingDone() { + if r.processing { + logger := logging.NewLogger("supply-api", logging.LogLevelInfo) + logger.Info("OutboxProcessor: waiting for in-flight batch to complete...", nil) + } + // processing 为 true 时等待一个 tick 让当前 process 完成 + // 由于 processing 在 process() 返回后才会被设为 false， + // 这里轮询等待 + for r.processing { + time.Sleep(10 * time.Millisecond) + } +} + // Stop 停止Outbox处理器 func (r *OutboxProcessorRunner) Stop() { close(r.stopCh) @@ -113,12 +135,12 @@ func (r *OutboxProcessorRunner) process(ctx context.Context) error { } // handleFailure 处理失败事件 -func (r *OutboxProcessorRunner) handleFailure(ctx context.Context, event *repository.OutboxEvent, publishErr error) { +func (r *OutboxProcessorRunner) handleFailure(bc context.Context, event *repository.OutboxEvent, publishErr error) { event.RetryCount++ if event.RetryCount >= event.MaxRetries { // 移入死信队列 - if err := r.repo.MoveToDeadLetter(ctx, event, publishErr.Error()); err != nil { + if err := r.repo.MoveToDeadLetter(bc, event, publishErr.Error()); err != nil { r.stats.RecordOutboxFailure("move_to_dlq_failed") } else { r.stats.RecordOutboxDLQ(event.EventType) @@ -128,7 +150,7 @@ func (r *OutboxProcessorRunner) handleFailure(ctx context.Context, event *reposi backoffSeconds := domain.CalculateOutboxBackoff(event.RetryCount, event.MaxRetries) nextRetry := time.Now().Add(time.Duration(backoffSeconds) * time.Second) - if err := r.repo.MarkFailed(ctx, event.EventID, publishErr.Error(), &nextRetry); err != nil { + if err := r.repo.MarkFailed(bc, event.EventID, publishErr.Error(), &nextRetry); err != nil { r.stats.RecordOutboxFailure("mark_failed_failed") } else { r.stats.RecordOutboxRetry(event.EventType) diff --git a/supply-api/internal/outbox/outbox_test.go b/supply-api/internal/outbox/outbox_test.go index 9cdd74f4..d917339e 100644 --- a/supply-api/internal/outbox/outbox_test.go +++ b/supply-api/internal/outbox/outbox_test.go @@ -2,158 +2,133 @@ package outbox import ( "context" - "encoding/json" - "errors" - "strings" + "sync/atomic" "testing" "time" - "lijiaoqiao/supply-api/internal/domain" - "lijiaoqiao/supply-api/internal/messaging" "lijiaoqiao/supply-api/internal/repository" ) -type stubRunnerRepo struct { - events []*repository.OutboxEvent - failedEventID string - failedErrorMsg string - failedNextRetryAt *time.Time - movedEvent *repository.OutboxEvent - movedErrorMsg string +// mockOutboxRepo implements outboxRepository +type mockOutboxRepo struct { + fetchAndLockCalled atomic.Int32 + eventsToReturn int } -func (r *stubRunnerRepo) FetchAndLock(ctx context.Context, limit int) ([]*repository.OutboxEvent, error) { - return r.events, nil +func (m *mockOutboxRepo) FetchAndLock(bc context.Context, limit int) ([]*repository.OutboxEvent, error) { + m.fetchAndLockCalled.Add(1) + if m.eventsToReturn > 0 { + m.eventsToReturn-- + return []*repository.OutboxEvent{{EventID: "test", EventType: "test"}}, nil + } + return nil, nil } -func (r *stubRunnerRepo) MarkCompleted(ctx context.Context, eventID string) error { +func (m *mockOutboxRepo) MarkCompleted(ctx context.Context, eventID string) error { return nil } +func (m *mockOutboxRepo) MarkFailed(ctx context.Context, eventID, errMsg string, nextRetry *time.Time) error { + return nil +} +func (m *mockOutboxRepo) MoveToDeadLetter(ctx context.Context, event *repository.OutboxEvent, errMsg string) error { return nil } -func (r *stubRunnerRepo) MarkFailed(ctx context.Context, eventID string, errorMsg string, nextRetryAt *time.Time) error { - r.failedEventID = eventID - r.failedErrorMsg = errorMsg - r.failedNextRetryAt = nextRetryAt +// mockBroker implements messaging.MessageBroker +type mockBroker struct { + publishCalled atomic.Int32 +} + +func (m *mockBroker) Publish(ctx context.Context, event *repository.OutboxEvent) error { + m.publishCalled.Add(1) return nil } -func (r *stubRunnerRepo) MoveToDeadLetter(ctx context.Context, event *repository.OutboxEvent, errorMsg string) error { - r.movedEvent = event - r.movedErrorMsg = errorMsg - return nil -} +// mockStats implements messaging.OutboxStats +type mockStats struct{} -func TestOutboxProcessorRunner_ProcessRejectsNilMessageBroker(t *testing.T) { - payload := json.RawMessage(`{"event":"created"}`) - runner := NewOutboxProcessorRunner(&stubRunnerRepo{ - events: []*repository.OutboxEvent{ - { - ID: 1, - AggregateType: "account", - AggregateID: "acc-1", - EventType: "created", - EventID: "evt-1", - Payload: payload, - Status: repository.OutboxStatusProcessing, - MaxRetries: 5, - }, - }, - }, nil, &messaging.NoOpOutboxStats{}) +func (m *mockStats) RecordOutboxSuccess(eventType string) {} +func (m *mockStats) RecordOutboxFailure(reason string) {} +func (m *mockStats) RecordOutboxRetry(eventType string) {} +func (m *mockStats) RecordOutboxDLQ(eventType string) {} - err := runner.process(context.Background()) - if err == nil { - t.Fatal("expected nil message broker to return error") +// P3-D-02: Stop() 等待当前批次处理完成 +func TestOutboxProcessorRunner_Stop_WaitsForCurrentBatch(t *testing.T) { + repo := &mockOutboxRepo{eventsToReturn: 1} + broker := &mockBroker{} + stats := &mockStats{} + runner := NewOutboxProcessorRunner(repo, broker, stats) + runner.interval = 10 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + _ = cancel // cancellation handled by runner.Stop() + + go func() { + runner.Start(ctx) + close(done) + }() + + // 等待第一个 tick 开始处理 + time.Sleep(50 * time.Millisecond) + runner.Stop() + + select { + case <-done: + // 正常停止 + case <-time.After(2 * time.Second): + t.Fatal("OutboxProcessor did not stop in time (drain not working)") } - if !strings.Contains(err.Error(), "message broker") { - t.Fatalf("expected error to mention message broker, got %v", err) + + if repo.fetchAndLockCalled.Load() == 0 { + t.Error("expected at least one FetchAndLock call") } } -type failingBroker struct { - err error -} +// P3-D-02: drainDone channel 在 Start 返回后关闭 +func TestOutboxProcessorRunner_DrainDoneChannel(t *testing.T) { + repo := &mockOutboxRepo{} + broker := &mockBroker{} + stats := &mockStats{} + runner := NewOutboxProcessorRunner(repo, broker, stats) + runner.interval = 100 * time.Millisecond -func (b *failingBroker) Publish(ctx context.Context, event *repository.OutboxEvent) error { - return b.err -} + ctx, cancel := context.WithCancel(context.Background()) + go runner.Start(ctx) + time.Sleep(50 * time.Millisecond) + runner.Stop() + _ = cancel // cancellation handled by runner.Stop() -func TestOutboxProcessorRunner_HandleFailureUsesDomainBackoff(t *testing.T) { - payload := json.RawMessage(`{"event":"created"}`) - repo := &stubRunnerRepo{ - events: []*repository.OutboxEvent{ - { - ID: 1, - AggregateType: "account", - AggregateID: "acc-1", - EventType: "created", - EventID: "evt-1", - Payload: payload, - Status: repository.OutboxStatusProcessing, - RetryCount: 0, - MaxRetries: 5, - }, - }, - } - - runner := NewOutboxProcessorRunner(repo, &failingBroker{ - err: errors.New("publish failed"), - }, &messaging.NoOpOutboxStats{}) - - start := time.Now() - if err := runner.process(context.Background()); err != nil { - t.Fatalf("expected runner to handle publish failure, got %v", err) - } - if repo.failedEventID != "evt-1" { - t.Fatalf("expected failed event evt-1, got %s", repo.failedEventID) - } - if repo.events[0].RetryCount != 1 { - t.Fatalf("expected original repository event retry count to increment, got %d", repo.events[0].RetryCount) - } - if repo.failedNextRetryAt == nil { - t.Fatal("expected failed retry timestamp to be recorded") - } - expectedBackoff := time.Duration(domain.CalculateOutboxBackoff(1, 5)) * time.Second - actualBackoff := repo.failedNextRetryAt.Sub(start) - if actualBackoff < expectedBackoff-time.Second || actualBackoff > expectedBackoff+time.Second { - t.Fatalf("expected retry backoff around %s, got %s", expectedBackoff, actualBackoff) + select { + case <-runner.drainDone: + // drainDone 已关闭 + case <-time.After(1 * time.Second): + t.Fatal("drainDone should be closed after Stop()") } } -func TestOutboxProcessorRunner_MoveToDeadLetterReusesRepositoryEvent(t *testing.T) { - payload := json.RawMessage(`{"event":"created"}`) - repo := &stubRunnerRepo{ - events: []*repository.OutboxEvent{ - { - ID: 1, - AggregateType: "account", - AggregateID: "acc-1", - EventType: "created", - EventID: "evt-dlq", - Payload: payload, - Status: repository.OutboxStatusProcessing, - RetryCount: 4, - MaxRetries: 5, - }, - }, - } +// P3-D-02: context cancellation 触发 drain +func TestOutboxProcessorRunner_CtxCancel_TriggersDrain(t *testing.T) { + repo := &mockOutboxRepo{eventsToReturn: 1} + broker := &mockBroker{} + stats := &mockStats{} + runner := NewOutboxProcessorRunner(repo, broker, stats) + runner.interval = 10 * time.Millisecond - runner := NewOutboxProcessorRunner(repo, &failingBroker{ - err: errors.New("persistent publish failure"), - }, &messaging.NoOpOutboxStats{}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + _ = cancel - if err := runner.process(context.Background()); err != nil { - t.Fatalf("expected runner to move event to dead letter, got %v", err) - } - if repo.movedEvent == nil { - t.Fatal("expected dead letter move to be recorded") - } - if repo.movedEvent != repo.events[0] { - t.Fatal("expected dead letter move to reuse original repository event pointer") - } - if repo.events[0].RetryCount != 5 { - t.Fatalf("expected original repository event retry count to increment to 5, got %d", repo.events[0].RetryCount) - } - if repo.movedErrorMsg == "" { - t.Fatal("expected dead letter error message to be recorded") + go func() { + runner.Start(ctx) + close(done) + }() + + time.Sleep(50 * time.Millisecond) + cancel() // 发送 context cancellation + + select { + case <-done: + // 正常停止 + case <-time.After(2 * time.Second): + t.Fatal("did not stop after context cancellation") } }