fix(nginx_log): synchronize searcher shard swaps

CLAUDE.md

@@ -28,7 +28,8 @@ This project is a web-based NGINX management interface built with Go backend and
 - Follow Cosy Error Handler best practices for error management
 - Implement standardized CRUD operations using Cosy framework
 - Apply efficient database pagination for large datasets
-- Validate changes with `go test ./... -race -cover` before pushing
+- Before committing backend changes, run the CI-equivalent unit test command: `GOWORK=off go test -tags=unembed -race -cover -count=1 ./...`
+- Keep `-count=1` in pre-commit unit tests so cached results cannot hide race conditions or flaky failures
 - Keep files modular and well-organized by functionality
 - **All comments and documentation must be in English**
 
@@ -76,7 +77,7 @@ This project is a web-based NGINX management interface built with Go backend and
 
 ## Development Commands
 - **Frontend**: `bun run dev`, `bun run lint`, `bun run typecheck`, `bun run build`
-- **Backend**: `go generate ./...`, `go build ./...`, run `go test ./... -race -cover`; for release artifacts reuse the README command with `-tags=jsoniter -ldflags "$LD_FLAGS ..."`.
+- **Backend**: `go generate ./...`, `go build ./...`, run `GOWORK=off go test -tags=unembed -race -cover -count=1 ./...`; for release artifacts reuse the README command with `-tags=jsoniter -ldflags "$LD_FLAGS ..."`.
 - **Demo stack**: `docker-compose -f docker-compose-demo.yml up` to bootstrap the sample environment
 
 

internal/nginx_log/searcher/hot_swap_test.go

@@ -2,7 +2,9 @@ package searcher
 
 import (
 	"context"
+	"fmt"
 	"path/filepath"
+	"sync"
 	"testing"
 
 	"github.com/blevesearch/bleve/v2"
@@ -40,7 +42,7 @@ func TestDistributedSearcher_SwapShards(t *testing.T) {
 
 	err = shard1.Index("doc1", doc1)
 	require.NoError(t, err)
-	
+
 	err = shard2.Index("doc2", doc2)
 	require.NoError(t, err)
 
@@ -87,6 +89,85 @@ func TestDistributedSearcher_SwapShards(t *testing.T) {
 	assert.LessOrEqual(t, result.TotalHits, uint64(2))    // But no more than two
 }
 
+func TestDistributedSearcher_ConcurrentSearchAndSwap(t *testing.T) {
+	tempDir := t.TempDir()
+	mapping := bleve.NewIndexMapping()
+
+	shard1, err := bleve.New(filepath.Join(tempDir, "shard1.bleve"), mapping)
+	require.NoError(t, err)
+	defer shard1.Close()
+	require.NoError(t, shard1.Index("doc1", map[string]interface{}{
+		"content": "test document one",
+	}))
+
+	shard2, err := bleve.New(filepath.Join(tempDir, "shard2.bleve"), mapping)
+	require.NoError(t, err)
+	defer shard2.Close()
+	require.NoError(t, shard2.Index("doc2", map[string]interface{}{
+		"content": "test document two",
+	}))
+
+	config := DefaultSearcherConfig()
+	config.EnableCache = false
+	distributedSearcher := NewSearcher(config, []bleve.Index{shard1})
+	defer distributedSearcher.Stop()
+
+	stopSearches := make(chan struct{})
+	searchErrors := make(chan error, 1)
+	searchStarted := make(chan struct{})
+	var searchWaitGroup sync.WaitGroup
+	searchWaitGroup.Add(1)
+	go func() {
+		defer searchWaitGroup.Done()
+		close(searchStarted)
+		for {
+			select {
+			case <-stopSearches:
+				return
+			default:
+			}
+
+			result, searchErr := distributedSearcher.Search(context.Background(), &SearchRequest{
+				Query: "test",
+				Limit: 10,
+			})
+			if searchErr != nil {
+				searchErrors <- searchErr
+				return
+			}
+			if result.TotalHits == 0 {
+				searchErrors <- fmt.Errorf("concurrent search returned no hits")
+				return
+			}
+
+			_ = distributedSearcher.IsHealthy()
+			_ = distributedSearcher.GetShards()
+		}
+	}()
+
+	<-searchStarted
+	for i := 0; i < 25; i++ {
+		shards := []bleve.Index{shard1}
+		if i%2 == 1 {
+			shards = append(shards, shard2)
+		}
+		require.NoError(t, distributedSearcher.SwapShards(shards))
+	}
+
+	close(stopSearches)
+	searchWaitGroup.Wait()
+	select {
+	case searchErr := <-searchErrors:
+		require.NoError(t, searchErr)
+	default:
+	}
+
+	shardSnapshot := distributedSearcher.GetShards()
+	require.NotEmpty(t, shardSnapshot)
+	shardSnapshot[0] = nil
+	require.NotNil(t, distributedSearcher.GetShards()[0], "GetShards must return an isolated snapshot")
+}
+
 func TestDistributedSearcher_SwapShards_NotRunning(t *testing.T) {
 	tempDir := t.TempDir()
 
@@ -101,7 +182,7 @@ func TestDistributedSearcher_SwapShards_NotRunning(t *testing.T) {
 	config := DefaultSearcherConfig()
 	searcher := NewSearcher(config, []bleve.Index{shard})
 	require.NotNil(t, searcher)
-	
+
 	err = searcher.Stop()
 	require.NoError(t, err)
 
@@ -146,7 +227,7 @@ func TestDistributedSearcher_HotSwap_ZeroDowntime(t *testing.T) {
 	gen2Path := filepath.Join(tempDir, "gen2.bleve")
 
 	mapping := bleve.NewIndexMapping()
-	
+
 	// Generation 1 index
 	gen1Index, err := bleve.New(gen1Path, mapping)
 	require.NoError(t, err)
@@ -171,10 +252,10 @@ func TestDistributedSearcher_HotSwap_ZeroDowntime(t *testing.T) {
 
 	err = gen1Index.Index("old_doc", gen1Doc)
 	require.NoError(t, err)
-	
+
 	err = gen2Index.Index("new_doc", gen2Doc)
 	require.NoError(t, err)
-	
+
 	// Ensure both indexes are flushed
 	err = gen1Index.SetInternal([]byte("_flush"), []byte("true"))
 	require.NoError(t, err)
@@ -249,7 +330,7 @@ func TestDistributedSearcher_SwapShards_StatsUpdate(t *testing.T) {
 	// Check stats after swap
 	stats = searcher.GetStats()
 	assert.Len(t, stats.ShardStats, 2)
-	
+
 	// Verify shard IDs are correct
 	shardIDs := make([]int, len(stats.ShardStats))
 	for i, stat := range stats.ShardStats {
@@ -257,4 +338,4 @@ func TestDistributedSearcher_SwapShards_StatsUpdate(t *testing.T) {
 	}
 	assert.Contains(t, shardIDs, 0)
 	assert.Contains(t, shardIDs, 1)
-}
+}

internal/nginx_log/searcher/searcher.go

@@ -26,6 +26,8 @@ type Searcher struct {
 
 	// Concurrency control
 	semaphore chan struct{}
+	stateMu   sync.RWMutex // protects shards and indexAlias
+	swapMu    sync.Mutex   // serializes shard swaps with shutdown
 
 	// State
 	running int32
@@ -154,6 +156,9 @@ func (s *Searcher) Search(ctx context.Context, req *SearchRequest) (*SearchResul
 		return nil, fmt.Errorf("failed to build query: %w", err)
 	}
 	indexAlias, releaseAlias := s.indexAliasForRequest(req)
+	if indexAlias == nil {
+		return nil, fmt.Errorf("searcher is not running")
+	}
 	defer releaseAlias()
 
 	// Execute search across shards
@@ -246,7 +251,12 @@ func (s *Searcher) executeGlobalScoringSearch(
 }
 
 func (s *Searcher) indexAliasForRequest(req *SearchRequest) (bleve.IndexAlias, func()) {
-	return indexAliasForLogPaths(s.indexAlias, s.shards, req.UseMainLogPath, req.LogPaths)
+	s.stateMu.RLock()
+	defaultAlias := s.indexAlias
+	shards := append([]bleve.Index(nil), s.shards...)
+	s.stateMu.RUnlock()
+
+	return indexAliasForLogPaths(defaultAlias, shards, req.UseMainLogPath, req.LogPaths)
 }
 
 // convertBleveResult converts a Bleve SearchResult to our SearchResult format
@@ -444,8 +454,12 @@ func (s *Searcher) setRequestDefaults(req *SearchRequest) {
 func (s *Searcher) getHealthyShards() []int {
 	// With IndexAlias, Bleve handles shard health internally
 	// Return all shard IDs since the alias will route correctly
-	healthy := make([]int, len(s.shards))
-	for i := range s.shards {
+	s.stateMu.RLock()
+	shardCount := len(s.shards)
+	s.stateMu.RUnlock()
+
+	healthy := make([]int, shardCount)
+	for i := range healthy {
 		healthy[i] = i
 	}
 	return healthy
@@ -530,41 +544,44 @@ func (s *Searcher) GetConfig() *Config {
 	return s.config
 }
 
-// GetShards returns the underlying shards for cardinality counting
+// GetShards returns an isolated shard snapshot for cardinality counting.
 func (s *Searcher) GetShards() []bleve.Index {
-	return s.shards
+	s.stateMu.RLock()
+	defer s.stateMu.RUnlock()
+
+	return append([]bleve.Index(nil), s.shards...)
 }
 
 // SwapShards atomically replaces the current shards with new ones using IndexAlias.Swap()
 // This follows Bleve best practices for zero-downtime index updates
 func (s *Searcher) SwapShards(newShards []bleve.Index) error {
+	s.swapMu.Lock()
+	defer s.swapMu.Unlock()
+
 	if atomic.LoadInt32(&s.running) == 0 {
 		return fmt.Errorf("searcher is not running")
 	}
 
+	newShards = wrapRecoveringIndexes(newShards)
+
+	s.stateMu.Lock()
 	if s.indexAlias == nil {
+		s.stateMu.Unlock()
 		return fmt.Errorf("indexAlias is nil")
 	}
 
-	newShards = wrapRecoveringIndexes(newShards)
-
-	// Store old shards for logging
 	oldShards := s.shards
-
-	// Perform atomic swap using IndexAlias - this is the key Bleve operation
-	// that provides zero-downtime index updates
 	logger.Debugf("SwapShards: Starting atomic swap - old=%d, new=%d", len(oldShards), len(newShards))
 
 	swapStartTime := time.Now()
 	s.indexAlias.Swap(newShards, oldShards)
 	swapDuration := time.Since(swapStartTime)
+	s.shards = newShards
+	s.stateMu.Unlock()
 
 	logger.Infof("IndexAlias.Swap completed in %v (old=%d shards, new=%d shards)",
 		swapDuration, len(oldShards), len(newShards))
 
-	// Update internal shards reference to match the IndexAlias
-	s.shards = newShards
-
 	// Clear cache after shard swap to prevent stale results
 	// Use goroutine to avoid potential deadlock during shard swap
 	if s.cache != nil {
@@ -620,9 +637,15 @@ func (s *Searcher) Stop() error {
 	var err error
 
 	s.closeOnce.Do(func() {
+		s.swapMu.Lock()
+		defer s.swapMu.Unlock()
+
 		// Set running to 0
 		atomic.StoreInt32(&s.running, 0)
 
+		s.stateMu.Lock()
+		defer s.stateMu.Unlock()
+
 		// Close the index alias first (this doesn't close underlying indexes)
 		if s.indexAlias != nil {
 			if closeErr := s.indexAlias.Close(); closeErr != nil {