mirror of
https://github.com/OpenListTeam/OpenList.git
synced 2026-10-10 04:53:09 +08:00
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>
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user