fix(search): serialize index updates by parent (#2827)

- Prevent concurrent updates from inserting duplicate index entries
- Preserve parallel indexing across different parent paths
- Add serialization, concurrency, and lock cleanup tests

Signed-off-by: jyxjjj <16695261+jyxjjj@users.noreply.github.com>
Co-authored-by: Codex <267193182+codex@users.noreply.github.com>
Co-authored-by: Pikachu Ren <40362270+PIKACHUIM@users.noreply.github.com>

Author: 3 people

internal/search/build.go

@@ -225,6 +225,9 @@ func Update(ctx context.Context, parent string, objs []model.Obj) {
 		return
 	}
 
+	unlock := lockUpdate(parent)
+	defer unlock()
+
 	nodes, err := instance.Get(ctx, parent)
 	if err != nil {
 		log.Errorf("update search index error while get nodes: %+v", err)

internal/search/build_test.go

@@ -0,0 +1,58 @@
+package search
+
+import (
+	"testing"
+	"time"
+)
+
+func TestLockUpdateSerializesSameParent(t *testing.T) {
+	unlockFirst := lockUpdate("/same-parent")
+	secondStarted := make(chan struct{})
+	secondAcquired := make(chan struct{})
+	secondReleased := make(chan struct{})
+	go func() {
+		close(secondStarted)
+		unlockSecond := lockUpdate("/same-parent")
+		close(secondAcquired)
+		unlockSecond()
+		close(secondReleased)
+	}()
+	<-secondStarted
+
+	select {
+	case <-secondAcquired:
+		t.Fatal("second update acquired the same parent lock")
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	unlockFirst()
+	select {
+	case <-secondReleased:
+	case <-time.After(time.Second):
+		t.Fatal("second update did not acquire the released parent lock")
+	}
+
+	updateLocksMu.Lock()
+	defer updateLocksMu.Unlock()
+	if len(updateLocks) != 0 {
+		t.Fatalf("update locks were not cleaned up: %d", len(updateLocks))
+	}
+}
+
+func TestLockUpdateAllowsDifferentParents(t *testing.T) {
+	unlockFirst := lockUpdate("/first-parent")
+	defer unlockFirst()
+
+	secondAcquired := make(chan struct{})
+	go func() {
+		unlockSecond := lockUpdate("/second-parent")
+		unlockSecond()
+		close(secondAcquired)
+	}()
+
+	select {
+	case <-secondAcquired:
+	case <-time.After(time.Second):
+		t.Fatal("update for a different parent was blocked")
+	}
+}

internal/search/update_lock.go

@@ -0,0 +1,38 @@
+package search
+
+import "sync"
+
+var (
+	updateLocksMu sync.Mutex
+	updateLocks   = make(map[string]*updateLock)
+)
+
+type updateLock struct {
+	mu   sync.Mutex
+	refs uint
+}
+
+// lockUpdate serializes index updates for the same parent while allowing
+// unrelated directories to update concurrently.
+func lockUpdate(parent string) func() {
+	updateLocksMu.Lock()
+	lock, ok := updateLocks[parent]
+	if !ok {
+		lock = &updateLock{}
+		updateLocks[parent] = lock
+	}
+	lock.refs++
+	updateLocksMu.Unlock()
+
+	lock.mu.Lock()
+	return func() {
+		lock.mu.Unlock()
+
+		updateLocksMu.Lock()
+		lock.refs--
+		if lock.refs == 0 {
+			delete(updateLocks, parent)
+		}
+		updateLocksMu.Unlock()
+	}
+}