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

internal/application/service/knowledge_process.go

@@ -457,29 +457,7 @@ func (s *knowledgeService) processChunks(ctx context.Context,
 		return insertChunks[i].ChunkIndex < insertChunks[j].ChunkIndex
 	})
 
-	// 仅为文本类型的Chunk设置前后关系（child chunks only, parents already linked above）
-	textChunks := make([]*types.Chunk, 0, len(chunks))
-	for _, chunk := range insertChunks {
-		if chunk.ChunkType == types.ChunkTypeText && chunk.ParentChunkID != "" {
-			// This is a child chunk in parent-child mode
-			textChunks = append(textChunks, chunk)
-		} else if chunk.ChunkType == types.ChunkTypeText && !hasParentChild {
-			// Normal flat chunk (no parent-child mode)
-			textChunks = append(textChunks, chunk)
-		}
-	}
-
-	// 设置文本Chunk之间的前后关系 (skip if parent-child, children don't need prev/next links)
-	if !hasParentChild {
-		for i, chunk := range textChunks {
-			if i > 0 {
-				textChunks[i-1].NextChunkID = chunk.ID
-			}
-			if i < len(textChunks)-1 {
-				textChunks[i+1].PreChunkID = chunk.ID
-			}
-		}
-	}
+	textChunks := processTextChunks(insertChunks, hasParentChild)
 
 	// Check if knowledge is being deleted/cancelled before writing chunks.
 	// Nothing has been persisted yet, so both branches just bail.

internal/application/service/knowledge_text_chunks.go

@@ -0,0 +1,27 @@
+package service
+
+import "github.com/Tencent/WeKnora/internal/types"
+
+// processTextChunks selects indexing inputs without linking parent-child context
+// as if it were a flat document.
+func processTextChunks(chunks []*types.Chunk, hasParentChild bool) []*types.Chunk {
+	textChunks := make([]*types.Chunk, 0, len(chunks))
+	for _, chunk := range chunks {
+		// Parent-less text can coexist with parent-child output (for example
+		// table chunks). Persisting it without indexing silently loses recall.
+		if chunk.ChunkType == types.ChunkTypeText {
+			textChunks = append(textChunks, chunk)
+		}
+	}
+	if !hasParentChild {
+		for i, chunk := range textChunks {
+			if i > 0 {
+				textChunks[i-1].NextChunkID = chunk.ID
+			}
+			if i < len(textChunks)-1 {
+				textChunks[i+1].PreChunkID = chunk.ID
+			}
+		}
+	}
+	return textChunks
+}

internal/application/service/knowledge_text_chunks_test.go

@@ -0,0 +1,43 @@
+package service
+
+import (
+	"github.com/Tencent/WeKnora/internal/types"
+	"testing"
+)
+
+func TestProcessTextChunksIncludesUnparentedText(t *testing.T) {
+	parent := &types.Chunk{ID: "parent", ChunkType: types.ChunkTypeParentText}
+	child := &types.Chunk{ID: "child", ChunkType: types.ChunkTypeText, ParentChunkID: "parent"}
+	standalone := &types.Chunk{ID: "standalone", ChunkType: types.ChunkTypeText}
+	got := processTextChunks([]*types.Chunk{parent, child, standalone}, true)
+	if len(got) != 2 {
+		t.Fatalf("indexed text count = %d, want 2; unparented text must not disappear in parent-child mode", len(got))
+	}
+	if got[0] != child || got[1] != standalone {
+		t.Fatal("must preserve text identity/order and exclude parent_text")
+	}
+	if child.NextChunkID != "" || standalone.PreChunkID != "" {
+		t.Fatal("parent-child mode must not create flat adjacency links")
+	}
+}
+
+func TestProcessTextChunksFlatLinks(t *testing.T) {
+	first := &types.Chunk{ID: "first", ChunkType: types.ChunkTypeText}
+	second := &types.Chunk{ID: "second", ChunkType: types.ChunkTypeText}
+	got := processTextChunks([]*types.Chunk{first, {ID: "summary", ChunkType: types.ChunkTypeSummary}, second}, false)
+	if len(got) != 2 || got[0] != first || got[1] != second {
+		t.Fatal("flat text identity/order changed")
+	}
+	if first.NextChunkID != "second" || second.PreChunkID != "first" {
+		t.Fatal("flat adjacency links missing")
+	}
+}
+
+func TestProcessTextChunksParentsOnly(t *testing.T) {
+	if got := processTextChunks([]*types.Chunk{{ID: "parent", ChunkType: types.ChunkTypeParentText}}, true); len(got) != 0 {
+		t.Fatal("parent_text must remain stored-only")
+	}
+	if got := processTextChunks(nil, false); len(got) != 0 {
+		t.Fatal("empty input must remain empty")
+	}
+}

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)
 	}
 }