P3-D: supply-api后台worker shutdown纪律 - partition维护取消/outbox优雅停止

2026-04-21 18:53:01 +08:00
parent e249a9160b
commit 8c5ab32e2e
3 changed files with 143 additions and 144 deletions
--- 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)
 				}
 			}
--- 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)
--- 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")
 	}
 }