diff --git a/internal/search/build.go b/internal/search/build.go index bbed1f853..8dbe7d498 100644 --- a/internal/search/build.go +++ b/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) diff --git a/internal/search/build_test.go b/internal/search/build_test.go new file mode 100644 index 000000000..fed2e155e --- /dev/null +++ b/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") + } +} diff --git a/internal/search/update_lock.go b/internal/search/update_lock.go new file mode 100644 index 000000000..7ca8d28dd --- /dev/null +++ b/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() + } +}