From 63c1e3faa27dfc5d8ea113eef20ab6f41d2961ab Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 13:42:33 +0100 Subject: [PATCH] Read a ledger: an instance's state, its history oldest first with sequence and author, a watch of its states, a declared index (issue 1966 T3) --- node-tools/internal/bus/ledger.go | 114 +++++++++++++++++ node-tools/internal/ledger/ledger.go | 2 + node-tools/internal/ledger/memory.go | 49 +++++++ node-tools/internal/ledger/read.go | 155 +++++++++++++++++++++++ node-tools/internal/ledger/read_test.go | 67 ++++++++++ node-tools/internal/ledger/types_test.go | 5 +- 6 files changed, 390 insertions(+), 2 deletions(-) create mode 100644 node-tools/internal/ledger/read.go create mode 100644 node-tools/internal/ledger/read_test.go diff --git a/node-tools/internal/bus/ledger.go b/node-tools/internal/bus/ledger.go index bf46d53..5591379 100644 --- a/node-tools/internal/bus/ledger.go +++ b/node-tools/internal/bus/ledger.go @@ -5,9 +5,12 @@ import ( "encoding/json" "errors" "fmt" + "sort" "strconv" "strings" + "time" + "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/membership" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -227,3 +230,114 @@ func (s *busSnapshots) Put(key string, value []byte, expect uint64) error { _, err := publishExpecting(s.write, s.prefix+"."+key, value, expect) return err } + +// Watch hands deliver every instance's record under a key pattern as it is, then each change; a removed key is handed as +// nil. +func (s *busState) Watch(pattern string, deliver func(key string, value []byte, current bool) error) (func(), error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + kv, err := s.kv(ctx) + cancel() + if err != nil { + return nil, err + } + w, err := kv.Watch(context.Background(), pattern) + if err != nil { + return nil, err + } + hand := func(e jetstream.KeyValueEntry, current bool) { + var value []byte + if e.Operation() == jetstream.KeyValuePut { + value = e.Value() + } + if err := deliver(e.Key(), value, current); err != nil { + s.c.Logf("[mesh-tools] a ledger watch of %s was not taken: %v", s.bucket, err) + } + } + timer := time.NewTimer(StateTimeout) + defer timer.Stop() + for current := true; current; { + select { + case e, ok := <-w.Updates(): + if !ok { + return nil, errors.New("the ledger's state watch ended before its current states were handed over") + } + if e == nil { + current = false + continue + } + hand(e, true) + case <-timer.C: + _ = w.Stop() + return nil, fmt.Errorf("the ledger's state %s did not hand its current states in %s", s.bucket, StateTimeout) + } + } + go func() { + for e := range w.Updates() { + if e != nil { + hand(e, false) + } + } + }() + return func() { _ = w.Stop() }, nil +} + +// LedgerIndexes are the buckets of a ledger binding's declared indexes, by index name, on the runner's own connection. +func (c *Conn) LedgerIndexes(b membership.BusBinding) map[string]ledger.Index { + out := map[string]ledger.Index{} + for name, bucket := range b.Indexes { + out[name] = &busState{c: c, bucket: bucket} + } + return out +} + +func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) { + if pattern == "" { + pattern = ">" + } + keys, err := s.Keys() + if err != nil { + return nil, err + } + sort.Strings(keys) + var out []buscalls.IndexEntry + for _, k := range keys { + if limit > 0 && len(out) >= limit { + break + } + if !SubjectMatches(pattern, k) { + continue + } + v, _, err := s.Get(k) + if err != nil { + return out, err + } + if v != nil { + out = append(out, buscalls.IndexEntry{Key: k, Value: valueOf(v)}) + } + } + return out, nil +} + +func (s *busState) Put(key string, value []byte) error { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := s.kv(ctx) + if err != nil { + return err + } + _, err = kv.Put(ctx, key, value) + return err +} + +func (s *busState) Delete(key string) error { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := s.kv(ctx) + if err != nil { + return err + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + return nil +} diff --git a/node-tools/internal/ledger/ledger.go b/node-tools/internal/ledger/ledger.go index e3b055b..f39b889 100644 --- a/node-tools/internal/ledger/ledger.go +++ b/node-tools/internal/ledger/ledger.go @@ -87,6 +87,8 @@ type Ledger struct { History History State State Snapshots Snapshots + // Indexes are its declared indexes' buckets, by index name. + Indexes map[string]Index // Now is the clock an operation is stamped by; time.Now when nil. Now func() time.Time // Logf says what an apply could not do after its operation was taken; nothing is said when nil. diff --git a/node-tools/internal/ledger/memory.go b/node-tools/internal/ledger/memory.go index c038888..240a225 100644 --- a/node-tools/internal/ledger/memory.go +++ b/node-tools/internal/ledger/memory.go @@ -1,9 +1,12 @@ package ledger import ( + "encoding/json" "sort" "strings" "sync" + + "git.novox.be/novox/mesh-sdk/go/buscalls" ) // The memory stores keep what the bus keeps, as the bus answers it, for a test of a ledger without a bus: a stream whose @@ -156,3 +159,49 @@ func (s *MemorySnapshots) Put(key string, value []byte, expect uint64) error { s.kv[key] = memoryValue{value, s.at} return nil } + +// MemoryIndex is an index's bucket in memory. +type MemoryIndex struct { + mu sync.Mutex + kv map[string][]byte +} + +func (x *MemoryIndex) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) { + x.mu.Lock() + defer x.mu.Unlock() + if pattern == "" { + pattern = ">" + } + var keys []string + for k := range x.kv { + if Matches(pattern, k) { + keys = append(keys, k) + } + } + sort.Strings(keys) + var out []buscalls.IndexEntry + for _, k := range keys { + if limit > 0 && len(out) >= limit { + break + } + out = append(out, buscalls.IndexEntry{Key: k, Value: json.RawMessage(x.kv[k])}) + } + return out, nil +} + +func (x *MemoryIndex) Put(key string, value []byte) error { + x.mu.Lock() + defer x.mu.Unlock() + if x.kv == nil { + x.kv = map[string][]byte{} + } + x.kv[key] = value + return nil +} + +func (x *MemoryIndex) Delete(key string) error { + x.mu.Lock() + defer x.mu.Unlock() + delete(x.kv, key) + return nil +} diff --git a/node-tools/internal/ledger/read.go b/node-tools/internal/ledger/read.go new file mode 100644 index 0000000..08c2e1c --- /dev/null +++ b/node-tools/internal/ledger/read.go @@ -0,0 +1,155 @@ +package ledger + +import ( + "encoding/json" + "errors" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// Reading a ledger (ADR 0597 §10, design 25 § 12.5; issue 1966 T3): an instance's state, its history, a watch of its +// states, and a declared index. Only the owning module reads its ledger: the runtime opens a ledger for the module whose +// bundle asks, never one the request names. + +// Index is one declared index's bucket, kept by a watcher in the owning module's process and rebuilt by reindex (ADR +// 0597 §4). +type Index interface { + // Entries is the entries whose key a pattern names (`*` one name, `>` the rest; empty is every key), at most limit + // (every one when 0). + Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) + // Put writes one entry; Delete removes one. + Put(key string, value []byte) error + Delete(key string) error +} + +// Watcher is a state bucket that hands each instance's record as it is and as it changes, under a key pattern. +type Watcher interface { + Watch(pattern string, deliver func(key string, value []byte, current bool) error) (func(), error) +} + +// HistoryMost is the most operations one history read answers. +const HistoryMost = 1000 + +// Get is an instance's state as the state holds it, and the sequence of its last operation in it; null and 0 for an +// instance with none. +func (l *Ledger) Get(r buscalls.InstanceRequest) (buscalls.InstanceState, error) { + key, err := KeyOf(r.Instance) + if err != nil { + return buscalls.InstanceState{}, err + } + rec, _, err := l.readState(key) + if err != nil { + return buscalls.InstanceState{}, err + } + state := rec.State + if len(state) == 0 { + state = json.RawMessage("null") + } + return buscalls.InstanceState{Instance: r.Instance, State: state, Sequence: rec.Sequence}, nil +} + +// operationsOf reads kept operations as the history answers them, each with its sequence. +func operationsOf(kept []Kept) ([]buscalls.Operation, error) { + out := make([]buscalls.Operation, 0, len(kept)) + for _, k := range kept { + var o buscalls.Operation + if err := json.Unmarshal(k.Data, &o); err != nil { + return nil, fmt.Errorf("the history's operation %d is not one: %w", k.Sequence, err) + } + o.Sequence = k.Sequence + out = append(out, o) + } + return out, nil +} + +// ReadHistory is an instance's operations from a sequence on, oldest first, each with its sequence and its proven author. +func (l *Ledger) ReadHistory(r buscalls.HistoryRequest) (buscalls.HistoryPage, error) { + key, err := KeyOf(r.Instance) + if err != nil { + return buscalls.HistoryPage{}, err + } + limit := r.Limit + if limit <= 0 || limit > HistoryMost { + limit = HistoryMost + } + last, err := l.History.Last(key) + if err != nil { + return buscalls.HistoryPage{}, err + } + kept, err := l.History.Read(key, r.From, limit+1) + if err != nil { + return buscalls.HistoryPage{}, err + } + page := buscalls.HistoryPage{Last: last} + if len(kept) > limit { + page.Next = kept[limit].Sequence + kept = kept[:limit] + } + if page.Operations, err = operationsOf(kept); err != nil { + return buscalls.HistoryPage{}, err + } + return page, nil +} + +// Watch hands deliver the state of every instance Instances names (`*` one name, `>` the rest, in an instance's path), +// the current ones first, then each change. +func (l *Ledger) Watch(r buscalls.LedgerWatchRequest, deliver func(buscalls.InstanceChange) error) (func(), error) { + w, ok := l.State.(Watcher) + if !ok { + return nil, errors.New("this ledger's state cannot be watched") + } + pattern := ">" + if r.Instances != "" { + pattern = keyPattern(r.Instances) + } + return w.Watch(pattern, func(key string, value []byte, current bool) error { + var rec Record + if value != nil { + if err := json.Unmarshal(value, &rec); err != nil { + return fmt.Errorf("the state of %s is not a record: %w", key, err) + } + } + state := rec.State + if len(state) == 0 { + state = json.RawMessage("null") + } + return deliver(buscalls.InstanceChange{Watch: r.Watch, Instance: InstanceOf(key), State: state, Sequence: rec.Sequence, + Current: current}) + }) +} + +// keyPattern is an instance pattern as a key pattern: its path's parts joined by dots. +func keyPattern(instances string) string { + out := []byte(instances) + for i, c := range out { + if c == '/' { + out[i] = '.' + } + } + return string(out) +} + +// index is a declared index by its name, or the reason there is none. +func (l *Ledger) index(name string) (Index, error) { + if ix, ok := l.Indexes[name]; ok { + return ix, nil + } + return nil, fmt.Errorf("%s's ledger %s declares no index %q", l.Module, l.Name, name) +} + +// Query is a declared index's entries whose key Key names. +func (l *Ledger) Query(r buscalls.QueryRequest) (buscalls.QueryAnswer, error) { + ix, err := l.index(r.Index) + if err != nil { + return buscalls.QueryAnswer{}, err + } + entries, err := ix.Entries(r.Key, r.Limit) + if err != nil { + return buscalls.QueryAnswer{}, err + } + if entries == nil { + entries = []buscalls.IndexEntry{} + } + return buscalls.QueryAnswer{Entries: entries}, nil +} diff --git a/node-tools/internal/ledger/read_test.go b/node-tools/internal/ledger/read_test.go new file mode 100644 index 0000000..17584db --- /dev/null +++ b/node-tools/internal/ledger/read_test.go @@ -0,0 +1,67 @@ +package ledger + +import ( + "fmt" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// 1966/AC15 and the other reads of a ledger: get, query (ADR 0597 §10; design 25 § 12.5). +func TestReadsOfALedger(t *testing.T) { + cases := []struct { + name string + run func(t *testing.T, l *Ledger, f fakes) + }{ + {"1966/AC15 history answers an instance's operations oldest first with sequence and author", func(t *testing.T, l *Ledger, f fakes) { + op := buscalls.Author{Account: "ledger.anchor.m", Caller: "op@shanks"} + applied(t, l, "decision/1", "note", `{"says":"one"}`, owner) + applied(t, l, "decision/2", "note", `{"says":"elsewhere"}`, owner) + applied(t, l, "decision/1", "title", `{"title":"Two"}`, op) + applied(t, l, "decision/1", "note", `{"says":"three"}`, owner) + page, err := l.ReadHistory(buscalls.HistoryRequest{Instance: "decision/1", Limit: 2}) + if err != nil { + t.Fatal(err) + } + var got []string + for _, o := range page.Operations { + got = append(got, fmt.Sprintf("%s@%d %s %s", o.Operation, o.Sequence, o.Author.Account, o.Author.Caller)) + } + if strings.Join(got, "; ") != "note@1 ledger.anchor.m ; title@3 ledger.anchor.m op@shanks" || page.Next != 4 || page.Last != 4 { + t.Fatalf("the history is %q, next %d, last %d", got, page.Next, page.Last) + } + rest, _ := l.ReadHistory(buscalls.HistoryRequest{Instance: "decision/1", From: page.Next}) + if len(rest.Operations) != 1 || rest.Operations[0].Sequence != 4 || rest.Next != 0 { + t.Fatalf("read on: %+v", rest) + } + }}, + {"get answers the state and its sequence, null for an instance with none", func(t *testing.T, l *Ledger, f fakes) { + applied(t, l, "decision/1", "title", `{"title":"One"}`, owner) + got, err := l.Get(buscalls.InstanceRequest{Instance: "decision/1"}) + if err != nil || got.Sequence != 1 || !strings.Contains(string(got.State), `"title":"One"`) { + t.Fatalf("got %+v, %v", got, err) + } + if none, err := l.Get(buscalls.InstanceRequest{Instance: "decision/9"}); err != nil || string(none.State) != "null" { + t.Fatalf("none is %+v, %v", none, err) + } + }}, + {"query reads a declared index by key, and an undeclared one is refused", func(t *testing.T, l *Ledger, f fakes) { + _ = f.index.Put("One.decision.1", []byte(`{"instance":"decision/1"}`)) + _ = f.index.Put("Two.decision.2", []byte(`{"instance":"decision/2"}`)) + got, err := l.Query(buscalls.QueryRequest{Index: "by-title", Key: "One.>"}) + if err != nil || len(got.Entries) != 1 || string(got.Entries[0].Value) != `{"instance":"decision/1"}` { + t.Fatalf("queried %+v, %v", got, err) + } + if _, err := l.Query(buscalls.QueryRequest{Index: "by-date"}); err == nil || !strings.Contains(err.Error(), `no index "by-date"`) { + t.Fatalf("an undeclared index: %v", err) + } + }}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + l, f := newLedger() + c.run(t, l, f) + }) + } +} diff --git a/node-tools/internal/ledger/types_test.go b/node-tools/internal/ledger/types_test.go index 7d8908a..1889063 100644 --- a/node-tools/internal/ledger/types_test.go +++ b/node-tools/internal/ledger/types_test.go @@ -29,12 +29,13 @@ type fakes struct { history *MemoryHistory state *MemoryState snapshots *MemorySnapshots + index *MemoryIndex } func newLedger() (*Ledger, fakes) { - f := fakes{&MemoryHistory{}, &MemoryState{}, &MemorySnapshots{}} + f := fakes{&MemoryHistory{}, &MemoryState{}, &MemorySnapshots{}, &MemoryIndex{}} return &Ledger{Module: "m", Name: "records", Types: []catalogue.LedgerType{records}, History: f.history, State: f.state, - Snapshots: f.snapshots}, f + Snapshots: f.snapshots, Indexes: map[string]Index{"by-title": f.index}}, f } // applied applies an operation and fails the test on a refusal.