diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_bus_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_bus_test.go new file mode 100644 index 000000000..335a2ed02 --- /dev/null +++ b/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) + } +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go index f9725b774..ea3b3f3f9 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go +++ b/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 } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go index 0a248cb9b..72ac00f95 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go +++ b/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) + } +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/move.go b/modules/mesh-knowledge/cmd/mesh-knowledge/move.go index 531f41828..f0be9b692 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/move.go +++ b/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 }