test(concurrency): 修复异步测试仓库与事件计数竞争

internal/application/service/tenant_api_key_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -13,6 +14,7 @@ import (
 )
 
 type fakeTenantAPIKeyRepo struct {
+	mu                  sync.Mutex
 	byHash              map[string]*types.TenantAPIKey
 	nextID              uint64
 	lastUsedUpdateCount int
@@ -43,6 +45,8 @@ func newFakeTenantAPIKeyRepo() *fakeTenantAPIKeyRepo {
 }
 
 func (r *fakeTenantAPIKeyRepo) CreateAPIKey(_ context.Context, key *types.TenantAPIKey) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	if _, ok := r.byHash[key.KeyHash]; ok {
 		return errors.New("duplicate key hash")
 	}
@@ -55,6 +59,8 @@ func (r *fakeTenantAPIKeyRepo) CreateAPIKey(_ context.Context, key *types.Tenant
 }
 
 func (r *fakeTenantAPIKeyRepo) GetAPIKeyByHash(_ context.Context, hash string) (*types.TenantAPIKey, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	key, ok := r.byHash[hash]
 	if !ok {
 		return nil, apprepo.ErrTenantAPIKeyNotFound
@@ -64,6 +70,8 @@ func (r *fakeTenantAPIKeyRepo) GetAPIKeyByHash(_ context.Context, hash string) (
 }
 
 func (r *fakeTenantAPIKeyRepo) ListAPIKeys(_ context.Context, tenantID uint64) ([]*types.TenantAPIKey, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	out := []*types.TenantAPIKey{}
 	for _, key := range r.byHash {
 		if key.TenantIDValue() == tenantID && key.RevokedAt == nil {
@@ -75,6 +83,8 @@ func (r *fakeTenantAPIKeyRepo) ListAPIKeys(_ context.Context, tenantID uint64) (
 }
 
 func (r *fakeTenantAPIKeyRepo) ListPlatformAPIKeys(_ context.Context) ([]*types.TenantAPIKey, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	out := []*types.TenantAPIKey{}
 	for _, key := range r.byHash {
 		if key.IsPlatform() && key.RevokedAt == nil {
@@ -86,6 +96,8 @@ func (r *fakeTenantAPIKeyRepo) ListPlatformAPIKeys(_ context.Context) ([]*types.
 }
 
 func (r *fakeTenantAPIKeyRepo) RevokeAPIKey(_ context.Context, tenantID uint64, id uint64) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	now := time.Now()
 	for _, key := range r.byHash {
 		if key.ID == id && key.TenantIDValue() == tenantID && key.RevokedAt == nil {
@@ -97,6 +109,8 @@ func (r *fakeTenantAPIKeyRepo) RevokeAPIKey(_ context.Context, tenantID uint64,
 }
 
 func (r *fakeTenantAPIKeyRepo) RevokePlatformAPIKey(_ context.Context, id uint64) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	now := time.Now()
 	for _, key := range r.byHash {
 		if key.ID == id && key.IsPlatform() && key.RevokedAt == nil {
@@ -108,6 +122,8 @@ func (r *fakeTenantAPIKeyRepo) RevokePlatformAPIKey(_ context.Context, id uint64
 }
 
 func (r *fakeTenantAPIKeyRepo) UpdateAPIKeyHash(_ context.Context, id uint64, hash string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	for oldHash, key := range r.byHash {
 		if key.ID == id && key.RevokedAt == nil {
 			delete(r.byHash, oldHash)
@@ -120,6 +136,8 @@ func (r *fakeTenantAPIKeyRepo) UpdateAPIKeyHash(_ context.Context, id uint64, ha
 }
 
 func (r *fakeTenantAPIKeyRepo) HasKeysWithPlaceholderHash(_ context.Context) (bool, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	for _, key := range r.byHash {
 		if key.RevokedAt == nil && strings.HasPrefix(key.KeyHash, "migrated-tenant-") {
 			return true, nil
@@ -129,6 +147,8 @@ func (r *fakeTenantAPIKeyRepo) HasKeysWithPlaceholderHash(_ context.Context) (bo
 }
 
 func (r *fakeTenantAPIKeyRepo) ListKeysWithPlaceholderHash(_ context.Context) ([]*types.TenantAPIKey, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	out := []*types.TenantAPIKey{}
 	for _, key := range r.byHash {
 		if key.RevokedAt == nil && strings.HasPrefix(key.KeyHash, "migrated-tenant-") {
@@ -140,6 +160,8 @@ func (r *fakeTenantAPIKeyRepo) ListKeysWithPlaceholderHash(_ context.Context) ([
 }
 
 func (r *fakeTenantAPIKeyRepo) UpdateAPIKeyLastUsed(_ context.Context, id uint64, at time.Time) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	r.lastUsedUpdateCount++
 	for _, key := range r.byHash {
 		if key.ID == id && key.RevokedAt == nil {
@@ -253,11 +275,11 @@ func TestTenantAPIKeyServiceAuthenticateThrottlesLastUsedUpdates(t *testing.T) {
 	}
 
 	deadline := time.Now().Add(500 * time.Millisecond)
-	for repo.lastUsedUpdateCount == 0 && time.Now().Before(deadline) {
+	for repo.lastUsedCount() == 0 && time.Now().Before(deadline) {
 		time.Sleep(10 * time.Millisecond)
 	}
-	if repo.lastUsedUpdateCount != 1 {
-		t.Fatalf("last_used update count = %d, want 1 (throttled async write)", repo.lastUsedUpdateCount)
+	if count := repo.lastUsedCount(); count != 1 {
+		t.Fatalf("last_used update count = %d, want 1 (throttled async write)", count)
 	}
 }
 
@@ -279,3 +301,9 @@ func TestTenantAPIKeyServiceAuthenticateRejectsExpiredKey(t *testing.T) {
 		t.Fatal("expired key should not authenticate")
 	}
 }
+
+func (r *fakeTenantAPIKeyRepo) lastUsedCount() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.lastUsedUpdateCount
+}

internal/event/example_test.go

@@ -3,6 +3,7 @@ package event
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -201,13 +202,13 @@ func TestEventBus_EmitAndWait(t *testing.T) {
 	ctx := context.Background()
 	bus := NewAsyncEventBus()
 
-	counter := 0
+	var counter atomic.Int32
 
 	// Register handlers
 	for i := 0; i < 3; i++ {
 		bus.On(EventQueryReceived, func(ctx context.Context, event Event) error {
 			time.Sleep(50 * time.Millisecond)
-			counter++
+			counter.Add(1)
 			return nil
 		})
 	}
@@ -222,8 +223,8 @@ func TestEventBus_EmitAndWait(t *testing.T) {
 		t.Errorf("EmitAndWait failed: %v", err)
 	}
 
-	if counter != 3 {
-		t.Errorf("Expected 3 handlers to complete, got %d", counter)
+	if got := counter.Load(); got != 3 {
+		t.Errorf("Expected 3 handlers to complete, got %d", got)
 	}
 }