CAT-K13: the knowledge base runs whole on the ledger over a real bus, and a restart finds every record

One integration test: the module's states are JetStream buckets, its events are published under their message ids to a stream
that drops a repeat; it moves its records, proposes, is reviewed and accepts a record, loses an event between an operation and its
send, restarts, sends the lost event exactly once, builds its index and checks the database within 40 seconds, and is healthy. The
test found that a restart probed the ledger number by number and stopped at a gap of 50, finding 3 of 14 records: the store now
keeps a key held.<kind>.<number> in the state findings for each record it creates and reads them at start (the old probing, with a
gap of 1000, only for a ledger written before the keys).

Issue 1966.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

modules/mesh-knowledge/cmd/mesh-knowledge/ledger_bus_test.go

@@ -0,0 +1,196 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// jsKV is one of the module's states on a real JetStream bucket.
+func jsonMarshal(v any) ([]byte, error) { return json.Marshal(v) }
+
+type jsKV struct {
+	kv jetstream.KeyValue
+}
+
+func changed(err error) error {
+	if errors.Is(err, jetstream.ErrKeyExists) || (err != nil && strings.Contains(err.Error(), "wrong last sequence")) {
+		return stdio.ErrChanged
+	}
+	return err
+}
+
+func (k jsKV) Get(key string) (*stdio.StateEntry, error) {
+	e, err := k.kv.Get(context.Background(), key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &stdio.StateEntry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil
+}
+func (k jsKV) Put(key string, v any) (uint64, error) {
+	b, _ := jsonMarshal(v)
+	return k.kv.Put(context.Background(), key, b)
+}
+func (k jsKV) Create(key string, v any) (uint64, error) {
+	b, _ := jsonMarshal(v)
+	r, err := k.kv.Create(context.Background(), key, b)
+	return r, changed(err)
+}
+func (k jsKV) Update(key string, v any, rev uint64) (uint64, error) {
+	b, _ := jsonMarshal(v)
+	r, err := k.kv.Update(context.Background(), key, b, rev)
+	return r, changed(err)
+}
+func (k jsKV) Delete(key string) error { return k.kv.Delete(context.Background(), key) }
+func (k jsKV) Keys() ([]string, error) {
+	keys, err := k.kv.Keys(context.Background())
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil, nil
+	}
+	return keys, err
+}
+
+type jsStates struct {
+	mu sync.Mutex
+	js jetstream.JetStream
+	by map[string]jsKV
+	t  *testing.T
+}
+
+func (s *jsStates) State(name string) KV {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if k, ok := s.by[name]; ok {
+		return k
+	}
+	kv, err := s.js.CreateOrUpdateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: "mesh_knowledge_" + name, History: 1})
+	if err != nil {
+		s.t.Fatal(err)
+	}
+	k := jsKV{kv}
+	s.by[name] = k
+	return k
+}
+
+// The knowledge base running whole on the ledger against a real bus (to-be 55 §10, AC57): the states are JetStream buckets,
+// its events are published under their message ids to a stream that drops a repeat, the ledger is played with the module's
+// own types. It moves its records, proposes, is reviewed and accepts a record, loses an event between an operation and its send,
+// restarts, sends what the stop lost exactly once, builds its index and checks the database within 40 seconds, and is healthy.
+func TestTheKnowledgeBaseRunsWholeOnTheLedgerOverARealBus(t *testing.T) {
+	s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(30 * time.Second) {
+		t.Fatal("no bus within 30s")
+	}
+	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
+	nc, err := nats.Connect(s.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, _ := jetstream.New(nc)
+	ctx := context.Background()
+	events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "KB_EVENTS", Subjects: []string{"kb.event.>"}, Duplicates: 24 * time.Hour})
+	if err != nil {
+		t.Fatal(err)
+	}
+	states := &jsStates{js: js, by: map[string]jsKV{}, t: t}
+	publish := func(event string, body any, id string) error {
+		b, _ := jsonMarshal(body)
+		_, err := js.Publish(ctx, "kb.event."+event, b, jetstream.WithMsgID(id))
+		return err
+	}
+	fakeBus := newFakeLedger(t) // the ledger of the tool runner, played with the module's own types
+	database := newMemStore()
+	seed := imported(t, database)
+	seed.accepted(t, "author@g14", valid(seed.reserved(t, "author@g14")))
+
+	start := func() (*world, *movableStore) {
+		led := newLedgerStore(fakeBus, states)
+		m := newMovableStore(database, led, states)
+		w := newWorld(t, m, fixture())
+		w.k.States = states
+		w.k.EmitID = publish
+		w.tracker.next[KindDecision] = 400
+		return w, m
+	}
+	w, m := start()
+	moved, err := m.Move(ctx, MoveAsk{Commit: true}, nil)
+	if err != nil || !moved.Switched {
+		t.Fatalf("%v %+v", err, moved)
+	}
+	// A record is proposed, reviewed in parts and accepted, writing through the ledger and saying its events on the bus.
+	n := w.reserved(t, "author@g14")
+	w.accepted(t, "author@g14", valid(n))
+	if info, _ := events.Info(ctx); info.State.Msgs == 0 {
+		t.Fatal("no event reached the bus")
+	}
+	// An event lost between the operation and its send: the bus is cut, the operation is in the history.
+	cut := w.k.EmitID
+	w.k.EmitID = func(string, any, string) error { return errors.New("the bus is cut") }
+	lost := w.reserved(t, "author@g14")
+	if w.k.sentUpTo(Key{KindDecision, lost}) != 0 {
+		t.Fatal("an event never sent was marked sent")
+	}
+	w.k.EmitID = cut
+	before, _ := events.Info(ctx)
+
+	// Restart: new objects over the same ledger and buckets. The start builds the index, checks, and sends what the stop lost.
+	began := time.Now()
+	w2, m2 := start()
+	if m2.Source() != SourceLedger {
+		t.Fatalf("a restart serves from %s: the switch key was not kept", m2.Source())
+	}
+	if err := m2.led.BuildIndex(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if diffs, err := m2.Check(ctx); err != nil || len(diffs) != 0 {
+		t.Fatalf("%v %v", err, diffs)
+	}
+	sent, err := w2.k.ResendEvents(ctx)
+	if err != nil || sent != 1 {
+		t.Fatalf("a restart sent %d events: %v", sent, err)
+	}
+	if took := time.Since(began); took > 40*time.Second {
+		t.Fatalf("the start took %s; at most 40 s", took)
+	}
+	after, _ := events.Info(ctx)
+	if after.State.Msgs != before.State.Msgs+1 {
+		t.Fatalf("the bus holds %d messages, was %d: the lost event is sent once", after.State.Msgs, before.State.Msgs)
+	}
+	// Sent again under the same id (the mark lost too), the bus drops the repeat.
+	_ = states.State(sentState).Delete(sentKey(Key{KindDecision, lost}))
+	_, _ = w2.k.ResendEvents(ctx)
+	again, _ := events.Info(ctx)
+	if again.State.Msgs != after.State.Msgs {
+		t.Fatalf("a repeat was kept: %d messages, was %d", again.State.Msgs, after.State.Msgs)
+	}
+	// The records are served, searched and healthy.
+	if e, err := w2.k.Store.Get(ctx, Key{KindDecision, n}); err != nil || e == nil || !e.Current.Accepted() {
+		t.Fatalf("%v %v", err, e)
+	}
+	if hits, err := m2.Search(ctx, "merge window", 3); err != nil || len(hits) == 0 {
+		t.Fatalf("%v %v", err, hits)
+	}
+	fakeBus.fill.MaxBytes = 512 << 20
+	if h := w2.k.ledgerHealth(HealthAnswer{State: "healthy", Healthy: true}); !h.Healthy || !strings.Contains(h.Why, "served from the ledger") {
+		t.Fatalf("%+v", h)
+	}
+}

modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go

@@ -150,26 +150,63 @@ func (s *ledgerStore) fresh(k Key) (*headState, error) {
 	return h, nil
 }
 
-// scan reads every instance the ledger holds, once: the numbers are probed in order until a long run of none, because the
-// ledger answers an instance by its path and not a list of them.
+// heldPrefix starts the key of the state `findings` that says a record is held: the ledger answers an instance by its path and
+// lists none, so the store keeps one small key for each record it creates, and a restart reads them to know what to ask for.
+const heldPrefix = "held."
+
+func heldKey(k Key) string { return fmt.Sprintf("%s%s.%04d", heldPrefix, k.Kind, k.Number) }
+
+func (s *ledgerStore) hold(k Key) {
+	if _, err := s.states.State(findingsState).Put(heldKey(k), 1); err != nil && s.logf != nil {
+		s.logf("[mesh-knowledge] that %s is held could not be kept: %v", k, err)
+	}
+}
+
+// scan reads every instance the ledger holds, once: the keys that say a record is held name them; a ledger written before
+// those keys existed is probed number by number, a long run of none ending it, and the keys are then written.
 func (s *ledgerStore) scan() error {
 	if s.scanned {
 		return nil
 	}
-	const gap = 50
-	for _, kind := range Kinds {
-		misses := 0
-		for n := 1; misses < gap; n++ {
-			k := Key{kind, n}
-			h, err := s.fresh(k)
-			if err != nil {
-				return err
+	keys, err := s.states.State(findingsState).Keys()
+	if err != nil {
+		return err
+	}
+	found := 0
+	for _, key := range keys {
+		rest, ok := strings.CutPrefix(key, heldPrefix)
+		if !ok {
+			continue
+		}
+		var k Key
+		var n int
+		kind, num, ok := strings.Cut(rest, ".")
+		if _, err := fmt.Sscanf(num, "%d", &n); !ok || err != nil {
+			continue
+		}
+		k = Key{kind, n}
+		if _, err := s.fresh(k); err != nil {
+			return err
+		}
+		found++
+	}
+	if found == 0 {
+		const gap = 1000
+		for _, kind := range Kinds {
+			misses := 0
+			for n := 1; misses < gap; n++ {
+				k := Key{kind, n}
+				h, err := s.fresh(k)
+				if err != nil {
+					return err
+				}
+				if h == nil {
+					misses++
+					continue
+				}
+				misses = 0
+				s.hold(k)
 			}
-			if h == nil {
-				misses++
-				continue
-			}
-			misses = 0
 		}
 	}
 	s.scanned = true
@@ -377,6 +414,9 @@ 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)
+	if e.Versions == 1 {
+		s.hold(k)
+	}
 	s.indexed(v.Record, current)
 	return nil
 }

modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go

@@ -160,3 +160,28 @@ func mustJSON(t *testing.T, v any) json.RawMessage {
 }
 
 var _ = buscalls.LedgerApply
+
+// A restart finds every record the ledger holds, however far apart their numbers are: the ledger lists none, so the store
+// keeps a key for each record it creates.
+func TestARestartFindsEveryRecordHoweverSparseTheirNumbers(t *testing.T) {
+	f := newFakeLedger(t)
+	st := newFakeStates()
+	s := newLedgerStore(f, st)
+	for _, n := range []int{11, 127, 900} {
+		if err := s.Create(context.Background(), Version{By: "a@n", Record: ledgerRecord(n, StatusProposed, Section{"Overview", "x"})}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	all, err := newLedgerStore(f, st).All(context.Background())
+	if err != nil || len(all) != 3 || all[2].Current.Number != 900 {
+		t.Fatalf("%v %d", err, len(all))
+	}
+	// A ledger written before the keys existed is probed, and the keys are written.
+	bare := newFakeStates()
+	if all, err := newLedgerStore(f, bare).All(context.Background()); err != nil || len(all) != 3 {
+		t.Fatalf("%v %d", err, len(all))
+	}
+	if keys, _ := bare.State(findingsState).Keys(); len(keys) != 3 {
+		t.Fatalf("%v", keys)
+	}
+}

modules/mesh-knowledge/cmd/mesh-knowledge/move.go

@@ -260,6 +260,8 @@ func (m *movableStore) Move(ctx context.Context, a MoveAsk, say func(string)) (*
 		if err := m.led.Import(ctx, p.e, p.versions); err != nil {
 			return nil, fmt.Errorf("%s could not be copied (%v); %d records copied, the rest not, and nothing is switched", k, err, copied)
 		}
+		// The events of these versions were said when they were written in the database: none is owed.
+		_, _ = m.st.State(sentState).Put(sentKey(k), len(p.versions))
 		copied++
 		if say != nil && copied%100 == 0 {
 			say(fmt.Sprintf("copied %d of %d records", copied, len(plans)))
@@ -420,6 +422,7 @@ func (s *ledgerStore) Import(ctx context.Context, e *Entry, versions []Version)
 		}
 		s.heads[k] = &headState{Kind: k.Kind, Number: k.Number, Entry: head, Sequence: seq}
 	}
+	s.hold(k)
 	s.indexed(e.Current, true)
 	return nil
 }