Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions pkg/domain/search.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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()
}
Expand Down
83 changes: 83 additions & 0 deletions pkg/domain/search_test.go
Original file line number Diff line number Diff line change
@@ -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()
})
})
})