CAT-K8: the search index is in the module's memory (bleve), built from the ledger and kept as it writes
ask and holds on the ledger answer from a full-text index of the heads and the current versions' text and clauses; it is built at start (index not built is the answer meanwhile), a new version replaces its record's documents, a reserved, withdrawn or retracted record has none. A new pure Go dependency, bleve. Issue 1966. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -0,0 +1,221 @@
|
||||
package main
|
||||
|
||||
// The search index (novox/hq to-be 55 §1, ADR 0597 §12): a full-text index in the module's own memory, written with the
|
||||
// pure Go library bleve. At start the module reads every head from the ledger and the bodies of every current version into
|
||||
// it, then keeps it current as it writes; `ask` and `holds` answer from it, and refuse with "index not built" while it is
|
||||
// built. Nothing but the module writes it and nothing outside it reads it.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/blevesearch/bleve/v2"
|
||||
"github.com/blevesearch/bleve/v2/analysis/lang/en"
|
||||
"github.com/blevesearch/bleve/v2/mapping"
|
||||
"github.com/blevesearch/bleve/v2/search/query"
|
||||
)
|
||||
|
||||
// ErrIndexNotBuilt is the answer of a question asked while the index is built.
|
||||
var ErrIndexNotBuilt = errors.New("index not built: the search index is being built from the ledger; ask again in a moment (get, list and propose answer at once)")
|
||||
|
||||
// searchIndex is the index and what it holds.
|
||||
type searchIndex struct {
|
||||
mu sync.Mutex
|
||||
idx bleve.Index
|
||||
built bool
|
||||
builtAt time.Time
|
||||
docs int
|
||||
// docsOf are the documents each record has in the index, so a new version replaces them.
|
||||
docsOf map[Key][]string
|
||||
}
|
||||
|
||||
func newSearchIndex() (*searchIndex, error) {
|
||||
m := bleve.NewIndexMapping()
|
||||
text := mapping.NewTextFieldMapping()
|
||||
text.Analyzer = en.AnalyzerName
|
||||
doc := bleve.NewDocumentMapping()
|
||||
for _, f := range []string{"title", "topic", "text", "clause"} {
|
||||
doc.AddFieldMappingsAt(f, text)
|
||||
}
|
||||
m.DefaultMapping = doc
|
||||
idx, err := bleve.NewMemOnly(m)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &searchIndex{idx: idx, docsOf: map[Key][]string{}}, nil
|
||||
}
|
||||
|
||||
type indexDoc struct {
|
||||
Title string `json:"title,omitempty"`
|
||||
Topic string `json:"topic,omitempty"`
|
||||
Text string `json:"text,omitempty"`
|
||||
Clause string `json:"clause,omitempty"`
|
||||
}
|
||||
|
||||
// put replaces the documents of a record by those of its served version: its title, topic and sections as one document,
|
||||
// and each clause in force or proposed as one of its own. A reserved, withdrawn or retracted record has none.
|
||||
func (s *searchIndex) put(r *Record) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
key := r.Key()
|
||||
b := s.idx.NewBatch()
|
||||
for _, id := range s.docsOf[key] {
|
||||
b.Delete(id)
|
||||
}
|
||||
delete(s.docsOf, key)
|
||||
if r.Status != StatusReserved && r.Status != StatusWithdrawn && r.Status != StatusRetracted {
|
||||
var body strings.Builder
|
||||
for _, sec := range r.Sections {
|
||||
body.WriteString(sec.Text + "\n")
|
||||
}
|
||||
id := fmt.Sprintf("%s/%d", key.Kind, key.Number)
|
||||
if err := b.Index(id, indexDoc{Title: r.Title, Topic: r.Topic, Text: body.String()}); err != nil {
|
||||
return err
|
||||
}
|
||||
ids := []string{id}
|
||||
for _, c := range r.Clauses {
|
||||
if c.Status == ClauseRetracted {
|
||||
continue
|
||||
}
|
||||
cid := fmt.Sprintf("%s#%d", id, c.N)
|
||||
if err := b.Index(cid, indexDoc{Clause: c.Text}); err != nil {
|
||||
return err
|
||||
}
|
||||
ids = append(ids, cid)
|
||||
}
|
||||
s.docsOf[key] = ids
|
||||
}
|
||||
s.docs = 0
|
||||
for _, ids := range s.docsOf {
|
||||
s.docs += len(ids)
|
||||
}
|
||||
return s.idx.Batch(b)
|
||||
}
|
||||
|
||||
// search ranks the documents against the question's words: a title weighs twice, as the in-memory ranking did.
|
||||
func (s *searchIndex) search(question string, limit int) ([]Hit, error) {
|
||||
s.mu.Lock()
|
||||
built := s.built
|
||||
s.mu.Unlock()
|
||||
if !built {
|
||||
return nil, ErrIndexNotBuilt
|
||||
}
|
||||
var words []string
|
||||
for _, w := range wordsOf(question) {
|
||||
if !askingWords[w] && len(w) >= 3 {
|
||||
words = append(words, w)
|
||||
}
|
||||
}
|
||||
if len(words) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
var should []query.Query
|
||||
for _, w := range words {
|
||||
for field, boost := range map[string]float64{"title": 2, "topic": 1, "text": 0.05, "clause": 1} {
|
||||
q := bleve.NewMatchQuery(w)
|
||||
q.SetField(field)
|
||||
q.SetBoost(boost)
|
||||
should = append(should, q)
|
||||
}
|
||||
}
|
||||
req := bleve.NewSearchRequestOptions(bleve.NewDisjunctionQuery(should...), 5000, 0, false)
|
||||
res, err := s.idx.Search(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var hits []Hit
|
||||
for _, h := range res.Hits {
|
||||
var kind string
|
||||
var number, clause int
|
||||
rest := h.ID
|
||||
if i := strings.IndexByte(rest, '#'); i >= 0 {
|
||||
if _, err := fmt.Sscanf(rest[i+1:], "%d", &clause); err != nil {
|
||||
continue
|
||||
}
|
||||
rest = rest[:i]
|
||||
}
|
||||
if i := strings.IndexByte(rest, '/'); i >= 0 {
|
||||
kind = rest[:i]
|
||||
if _, err := fmt.Sscanf(rest[i+1:], "%d", &number); err != nil {
|
||||
continue
|
||||
}
|
||||
}
|
||||
hits = append(hits, Hit{Key: Key{kind, number}, Clause: clause, Rank: h.Score})
|
||||
}
|
||||
sort.SliceStable(hits, func(i, j int) bool {
|
||||
if hits[i].Rank != hits[j].Rank {
|
||||
return hits[i].Rank > hits[j].Rank
|
||||
}
|
||||
if hits[i].Key != hits[j].Key {
|
||||
return hits[i].Key.Number > hits[j].Key.Number
|
||||
}
|
||||
return hits[i].Clause < hits[j].Clause
|
||||
})
|
||||
if limit > 0 && len(hits) > limit {
|
||||
hits = hits[:limit]
|
||||
}
|
||||
return hits, nil
|
||||
}
|
||||
|
||||
// IndexState says how the index stands: built or not, when, and how many documents it holds.
|
||||
type IndexState struct {
|
||||
Built bool `json:"built"`
|
||||
BuiltAt time.Time `json:"built-at,omitempty"`
|
||||
Docs int `json:"documents"`
|
||||
}
|
||||
|
||||
func (s *searchIndex) state() IndexState {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return IndexState{Built: s.built, BuiltAt: s.builtAt, Docs: s.docs}
|
||||
}
|
||||
|
||||
// BuildIndex reads every record from the ledger into the index and marks it built; asked again it builds it anew.
|
||||
func (s *ledgerStore) BuildIndex(ctx context.Context) error {
|
||||
idx, err := newSearchIndex()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
all, err := s.All(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, e := range all {
|
||||
if err := idx.put(e.Current); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
idx.mu.Lock()
|
||||
idx.built, idx.builtAt = true, time.Now().UTC()
|
||||
idx.mu.Unlock()
|
||||
s.mu.Lock()
|
||||
s.index = idx
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
// IndexState is how the module's index stands.
|
||||
func (s *ledgerStore) IndexState() IndexState {
|
||||
s.mu.Lock()
|
||||
idx := s.index
|
||||
s.mu.Unlock()
|
||||
if idx == nil {
|
||||
return IndexState{}
|
||||
}
|
||||
return idx.state()
|
||||
}
|
||||
|
||||
// indexed keeps the index current with a version the store wrote.
|
||||
func (s *ledgerStore) indexed(r *Record, current bool) {
|
||||
if s.index == nil || !current {
|
||||
return
|
||||
}
|
||||
if err := s.index.put(r); err != nil && s.logf != nil {
|
||||
s.logf("[mesh-knowledge] the search index could not take %s: %v", r.Key(), err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The search index (to-be 55 §1, AC52): built from the ledger at start, refusing "index not built" meanwhile, kept current
|
||||
// as the module writes, and answering the questions the knowledge base is asked as the in-memory ranking does: the records
|
||||
// the questions are about come first.
|
||||
func TestTheSearchIndexAnswersTheQuestionsAsTheStoreDoes(t *testing.T) {
|
||||
f := newFakeLedger(t)
|
||||
ls := newLedgerStore(f, newFakeStates())
|
||||
w := newWorld(t, ls, fixture())
|
||||
w.k.States = newFakeStates()
|
||||
ls.indexAsync = true
|
||||
if _, err := ls.Search(t.Context(), "merge window", 5); !errors.Is(err, ErrIndexNotBuilt) {
|
||||
t.Fatalf("a question before the index was built: %v", err)
|
||||
}
|
||||
if _, err := w.k.Import("import@test", ImportAsk{}, fixture()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ls.BuildIndex(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st := ls.IndexState(); !st.Built || st.Docs < 20 {
|
||||
t.Fatalf("%+v", st)
|
||||
}
|
||||
mem := imported(t, newMemStore())
|
||||
for _, q := range []string{"merge window", "how is the tool runtime chosen", "what holds about resolved names", "a knowledge base the mesh writes"} {
|
||||
want, _ := mem.k.Store.Search(t.Context(), q, 3)
|
||||
got, err := ls.Search(t.Context(), q, 8)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The ranking differs (the index weighs by how rare a word is), so the record the store puts first is among the
|
||||
// first few of the index's.
|
||||
found := false
|
||||
for _, h := range got {
|
||||
found = found || (len(want) > 0 && h.Key == want[0].Key)
|
||||
}
|
||||
if len(want) > 0 && !found {
|
||||
t.Errorf("%q: the store puts %v first, the index %v", q, want[0], got)
|
||||
}
|
||||
}
|
||||
// A new version is in the index at once; a withdrawn record is out of it.
|
||||
r, _ := w.k.ReserveNumber("author@g14", KindDecision, "a test record")
|
||||
if hits, _ := ls.Search(t.Context(), "zebra quantum", 5); len(hits) != 0 {
|
||||
t.Fatalf("%v", hits)
|
||||
}
|
||||
p := valid(r.Number)
|
||||
p.Title = "Zebra quantum processors"
|
||||
p.Sections[2].Text = "A zebra quantum processor."
|
||||
if _, err := w.k.Propose("author@g14", p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if hits, _ := ls.Search(t.Context(), "zebra quantum", 5); len(hits) == 0 || hits[0].Key.Number != r.Number {
|
||||
t.Fatalf("a proposed record is not in the index: %v", hits)
|
||||
}
|
||||
if _, err := w.k.WithdrawRecord("author@g14", Key{KindDecision, r.Number}, "not needed"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if hits, _ := ls.Search(t.Context(), "zebra quantum", 5); len(hits) != 0 {
|
||||
t.Fatalf("a withdrawn record is in the index: %v", hits)
|
||||
}
|
||||
}
|
||||
@@ -93,6 +93,10 @@ type ledgerStore struct {
|
||||
// drafted are the sections written for a draft and named by no version yet: a version that carries one reuses its
|
||||
// write-section and writes none again.
|
||||
drafted map[Key][]sectionRef
|
||||
// index is the search index of the module's memory, nil until built.
|
||||
index *searchIndex
|
||||
indexAsync bool
|
||||
logf func(format string, a ...any)
|
||||
}
|
||||
|
||||
// newLedgerStore is the store over a pipe and the module's states.
|
||||
@@ -350,13 +354,13 @@ func (s *ledgerStore) Create(ctx context.Context, v Version) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.commit(ctx, k, operationFor(nil, v, true), &entryHead{CurrentVersion: 1, Versions: 1, Current: ptr(stripped(v.Record, refs))}, v, refs)
|
||||
return s.commit(ctx, k, operationFor(nil, v, true), &entryHead{CurrentVersion: 1, Versions: 1, Current: ptr(stripped(v.Record, refs))}, v, refs, true)
|
||||
}
|
||||
|
||||
func ptr[T any](v T) *T { return &v }
|
||||
|
||||
// commit appends the operation that carries the new head and the version, and holds the head it derived.
|
||||
func (s *ledgerStore) commit(ctx context.Context, k Key, op string, e *entryHead, v Version, refs []sectionRef) error {
|
||||
func (s *ledgerStore) commit(ctx context.Context, k Key, op string, e *entryHead, v Version, refs []sectionRef, current bool) error {
|
||||
body := map[string]any{"kind": k.Kind, "number": k.Number, "entry": e,
|
||||
"version": versionHead{N: v.N, By: v.By, At: v.At, Why: v.Why, Insight: v.Insight, Record: stripped(v.Record, refs)},
|
||||
"pending-sections": map[string]any{}}
|
||||
@@ -366,6 +370,7 @@ func (s *ledgerStore) commit(ctx context.Context, k Key, op string, e *entryHead
|
||||
}
|
||||
s.heads[k] = &headState{Kind: k.Kind, Number: k.Number, Entry: e, Sequence: seq}
|
||||
delete(s.drafted, k)
|
||||
s.indexed(v.Record, current)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -455,7 +460,7 @@ func (s *ledgerStore) Save(ctx context.Context, v Version, current, clearPending
|
||||
if clearPending {
|
||||
e.Pending, e.PendingVersion = nil, 0
|
||||
}
|
||||
return s.commit(ctx, k, operationFor(old, v, current), &e, v, refs)
|
||||
return s.commit(ctx, k, operationFor(old, v, current), &e, v, refs, current)
|
||||
}
|
||||
|
||||
func (s *ledgerStore) Versions(_ context.Context, k Key) ([]Version, error) {
|
||||
@@ -493,14 +498,24 @@ func (s *ledgerStore) Versions(_ context.Context, k Key) ([]Version, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Search ranks the served versions as the in-memory store does, over every record; the index of the module's memory
|
||||
// replaces it (the search index of to-be 55 §1).
|
||||
// Search ranks the served versions from the index of the module's memory; it is built on the first question when no start
|
||||
// built it, and refuses with "index not built" while a start builds it.
|
||||
func (s *ledgerStore) Search(ctx context.Context, question string, limit int) ([]Hit, error) {
|
||||
all, err := s.All(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
s.mu.Lock()
|
||||
idx, async := s.index, s.indexAsync
|
||||
s.mu.Unlock()
|
||||
if idx == nil {
|
||||
if async {
|
||||
return nil, ErrIndexNotBuilt
|
||||
}
|
||||
if err := s.BuildIndex(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.mu.Lock()
|
||||
idx = s.index
|
||||
s.mu.Unlock()
|
||||
}
|
||||
return searchEntries(all, question, limit), nil
|
||||
return idx.search(question, limit)
|
||||
}
|
||||
|
||||
func (s *ledgerStore) Inbound(_ context.Context, k Key) ([]Inbound, error) {
|
||||
|
||||
Reference in New Issue
Block a user