diff --git a/pkg/domain/search.go b/pkg/domain/search.go index 5cc1c62..2559346 100644 --- a/pkg/domain/search.go +++ b/pkg/domain/search.go @@ -4,12 +4,19 @@ import ( "fmt" "os" "path/filepath" + "sync" "github.com/blevesearch/bleve/v2" ) // Searcher handles full-text search using bleve type Searcher struct { + // mu guards index (and its indexPath backing files) so a concurrent + // IndexSkills rebuild cannot Close an index another goroutine is + // closing/searching. Without it, two overlapping RebuildIndex calls + // (one GitSyncer goroutine per repo) double-close the underlying bleve + // index and panic with "close of closed channel", taking the process down. + mu sync.RWMutex indexPath string index bleve.Index } @@ -37,6 +44,9 @@ func NewSearcher(skillsDir string) (*Searcher, error) { // IndexSkills indexes a list of skills func (s *Searcher) IndexSkills(skills []Skill) error { + s.mu.Lock() + defer s.mu.Unlock() + // Clear existing index by deleting and recreating s.index.Close() os.RemoveAll(s.indexPath) @@ -76,6 +86,11 @@ func (s *Searcher) IndexSkills(skills []Skill) error { // Search performs a full-text search and returns matching skills func (s *Searcher) Search(query string) ([]Skill, error) { + // RLock held for the whole search so an in-flight IndexSkills cannot + // Close the index out from under s.index.Search below. + s.mu.RLock() + defer s.mu.RUnlock() + if s.index == nil { return []Skill{}, nil } @@ -120,6 +135,9 @@ func (s *Searcher) Search(query string) ([]Skill, error) { // Close closes the search index func (s *Searcher) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.index != nil { return s.index.Close() } diff --git a/pkg/domain/search_test.go b/pkg/domain/search_test.go new file mode 100644 index 0000000..3a56dbf --- /dev/null +++ b/pkg/domain/search_test.go @@ -0,0 +1,83 @@ +package domain_test + +import ( + "os" + "sync" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/skillserver/pkg/domain" +) + +var _ = Describe("Searcher", func() { + var ( + searcher *domain.Searcher + tempDir string + err error + ) + + BeforeEach(func() { + tempDir, err = os.MkdirTemp("", "skillserver-search-test") + Expect(err).NotTo(HaveOccurred()) + searcher, err = domain.NewSearcher(tempDir) + Expect(err).NotTo(HaveOccurred()) + }) + + AfterEach(func() { + _ = searcher.Close() + os.RemoveAll(tempDir) + }) + + Context("Concurrent rebuilds", func() { + It("does not double-close the index when IndexSkills runs concurrently", func() { + skills := []domain.Skill{ + {Name: "alpha", Content: "docker guide"}, + {Name: "beta", Content: "kubernetes guide"}, + } + + // Regression: one GitSyncer goroutine per configured repo calls + // RebuildIndex -> IndexSkills. Without a mutex around index + // Close/recreate, overlapping calls close an already-closed bleve + // index and panic with "close of closed channel", crashing the host + // process. GinkgoRecover surfaces such a panic as a spec failure. + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer GinkgoRecover() + defer wg.Done() + Expect(searcher.IndexSkills(skills)).To(Succeed()) + }() + } + wg.Wait() + + // Index is still consistent and searchable afterwards. + results, err := searcher.Search("docker") + Expect(err).NotTo(HaveOccurred()) + Expect(results).NotTo(BeNil()) + }) + + It("stays searchable while rebuilds and searches interleave", func() { + skills := []domain.Skill{{Name: "gamma", Content: "terraform guide"}} + Expect(searcher.IndexSkills(skills)).To(Succeed()) + + var wg sync.WaitGroup + for i := 0; i < 6; i++ { + wg.Add(2) + go func() { + defer GinkgoRecover() + defer wg.Done() + Expect(searcher.IndexSkills(skills)).To(Succeed()) + }() + go func() { + defer GinkgoRecover() + defer wg.Done() + _, err := searcher.Search("terraform") + Expect(err).NotTo(HaveOccurred()) + }() + } + wg.Wait() + }) + }) +})