novox/mesh-tools
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Key and answer a ledger index by mesh-sdk's IndexKey and IndexEntry, the one shape the owner's watcher also writes (issue 1966 T3)

Browse Source
This commit is contained in:
jschoubben
2026-11-03 15:15:00 +01:00
parent d206b0dde1
commit fadc73f73b
5 changed files with 40 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node-tools/internal/bus/ledger.go
+3 -4
View File
@@ -10,7 +10,6 @@ import (
"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"
@@ -290,7 +289,7 @@ func (c *Conn) LedgerIndexes(b membership.BusBinding) map[string]ledger.Index {
return out
}
func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) {
func (s *busState) Entries(pattern string, limit int) ([]ledger.Indexed, error) {
if pattern == "" {
pattern = ">"
}
@@ -299,7 +298,7 @@ func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, er
return nil, err
}
sort.Strings(keys)
var out []buscalls.IndexEntry
var out []ledger.Indexed
for _, k := range keys {
if limit > 0 && len(out) >= limit {
break
@@ -312,7 +311,7 @@ func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, er
return out, err
}
if v != nil {
out = append(out, buscalls.IndexEntry{Key: k, Value: valueOf(v)})
out = append(out, ledger.Indexed{Key: k, Value: v})
}
}
return out, nil
node-tools/internal/ledger/memory.go
+3 -6
View File
@@ -1,12 +1,9 @@
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
@@ -166,7 +163,7 @@ type MemoryIndex struct {
kv map[string][]byte
}
func (x *MemoryIndex) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) {
func (x *MemoryIndex) Entries(pattern string, limit int) ([]Indexed, error) {
x.mu.Lock()
defer x.mu.Unlock()
if pattern == "" {
@@ -179,12 +176,12 @@ func (x *MemoryIndex) Entries(pattern string, limit int) ([]buscalls.IndexEntry,
}
}
sort.Strings(keys)
var out []buscalls.IndexEntry
var out []Indexed
for _, k := range keys {
if limit > 0 && len(out) >= limit {
break
}
out = append(out, buscalls.IndexEntry{Key: k, Value: json.RawMessage(x.kv[k])})
out = append(out, Indexed{Key: k, Value: x.kv[k]})
}
return out, nil
}
node-tools/internal/ledger/read.go
+16 -5
View File
@@ -16,13 +16,19 @@ import (
// 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)
// (every one when 0), each as the bucket holds it.
Entries(pattern string, limit int) ([]Indexed, error)
// Put writes one entry; Delete removes one.
Put(key string, value []byte) error
Delete(key string) error
}
// Indexed is one key of an index's bucket and its value, a buscalls.IndexEntry as JSON.
type Indexed struct {
Key string
Value []byte
}
// 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)
@@ -144,12 +150,17 @@ func (l *Ledger) Query(r buscalls.QueryRequest) (buscalls.QueryAnswer, error) {
if err != nil {
return buscalls.QueryAnswer{}, err
}
entries, err := ix.Entries(r.Key, r.Limit)
held, err := ix.Entries(r.Key, r.Limit)
if err != nil {
return buscalls.QueryAnswer{}, err
}
if entries == nil {
entries = []buscalls.IndexEntry{}
entries := make([]buscalls.IndexEntry, 0, len(held))
for _, h := range held {
var e buscalls.IndexEntry
if err := json.Unmarshal(h.Value, &e); err != nil {
return buscalls.QueryAnswer{}, fmt.Errorf("the index %s's entry %s is not one: %w", r.Index, h.Key, err)
}
entries = append(entries, e)
}
return buscalls.QueryAnswer{Entries: entries}, nil
}
node-tools/internal/ledger/read_test.go
+5 -5
View File
@@ -47,10 +47,10 @@ func TestReadsAndRepairsOfALedger(t *testing.T) {
}
}},
{"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"}`))
_ = f.index.Put("One.decision.1", []byte(`{"instance":"decision/1","sequence":1}`))
_ = f.index.Put("Two.decision.2", []byte(`{"instance":"decision/2","sequence":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"}` {
if err != nil || len(got.Entries) != 1 || got.Entries[0] != (buscalls.IndexEntry{Instance: "decision/1", Sequence: 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"`) {
@@ -82,7 +82,7 @@ func TestReadsAndRepairsOfALedger(t *testing.T) {
}
}},
{"reindex builds a declared index from the states, and lag says an index behind until it does", func(t *testing.T, l *Ledger, f fakes) {
applied(t, l, "decision/1", "title", `{"title":"One ledger"}`, owner)
applied(t, l, "decision/1", "title", `{"title":"One"}`, owner)
applied(t, l, "decision/2", "note", `{"says":"untitled"}`, owner)
_ = f.index.Put("stale.decision.9", []byte(`{}`))
if lag, _ := l.Lag(buscalls.LedgerRequest{}); len(lag.Indexes) != 1 || lag.Indexes[0].Behind != 1 {
@@ -92,7 +92,7 @@ func TestReadsAndRepairsOfALedger(t *testing.T) {
t.Fatalf("reindexed %+v, %v", got, err)
}
entries, _ := f.index.Entries("", 0)
if len(entries) != 1 || entries[0].Key != "One_ledger.decision.1" || string(entries[0].Value) != `{"instance":"decision/1","sequence":1}` {
if len(entries) != 1 || entries[0].Key != buscalls.IndexKey([]string{"One"}, "decision/1") || string(entries[0].Value) != `{"instance":"decision/1","sequence":1}` {
t.Fatalf("the index holds %+v", entries)
}
if lag, _ := l.Lag(buscalls.LedgerRequest{}); len(lag.Indexes) != 0 || len(lag.Instances) != 0 {
node-tools/internal/ledger/repair.go
+13 -24
View File
@@ -3,7 +3,6 @@ package ledger
import (
"encoding/json"
"fmt"
"regexp"
"sort"
"strings"
@@ -130,41 +129,33 @@ func (l *Ledger) Snapshot(r buscalls.InstanceRequest) (buscalls.Repaired, error)
return out, nil
}
// unsafe is what a key part cannot hold.
var unsafe = regexp.MustCompile(`[^-_A-Za-z0-9]`)
// IndexEntryOf is an instance's entry in a declared index, as reindex writes it: its key the values of the fields the
// index is on, in their order, then the instance's key, each part with what a key cannot hold written as `_`; its value
// the instance and the sequence of the state it was taken from. False when the state holds none of those fields.
func IndexEntryOf(on []string, key string, rec Record) (string, []byte, bool) {
// indexEntryOf is an instance's entry in a declared index, as reindex writes it and the owner's watcher keeps it
// (mesh-sdk buscalls.IndexKey, buscalls.IndexEntry): its key the values of the fields the index is on, in their order,
// then the instance; its value the instance and the sequence of the state it was taken from. False when the state holds
// none of those fields. A value that is not text is written as its JSON.
func indexEntryOf(on []string, key string, rec Record) (string, []byte, bool) {
var state map[string]any
if json.Unmarshal(rec.State, &state) != nil || state == nil {
return "", nil, false
}
parts := make([]string, 0, len(on)+1)
values := make([]string, 0, len(on))
found := false
for _, f := range on {
v, has := state[f]
if !has {
parts = append(parts, "_")
continue
}
found = true
found = found || has
s, ok := v.(string)
if !ok {
raw, _ := json.Marshal(v)
s = string(raw)
}
if s = unsafe.ReplaceAllString(s, "_"); s == "" {
s = "_"
}
parts = append(parts, s)
values = append(values, s)
}
if !found {
return "", nil, false
}
value, _ := json.Marshal(map[string]any{"instance": InstanceOf(key), "sequence": rec.Sequence})
return strings.Join(append(parts, key), "."), value, true
instance := InstanceOf(key)
value, _ := json.Marshal(buscalls.IndexEntry{Instance: instance, Sequence: rec.Sequence})
return buscalls.IndexKey(values, instance), value, true
}
// declared is the fields an index is on and the types declaring it.
@@ -205,7 +196,7 @@ func (l *Ledger) wanted(index string) (map[string][]byte, map[string]uint64, err
if err != nil {
return nil, nil, err
}
if k, v, ok := IndexEntryOf(on, key, held); ok {
if k, v, ok := indexEntryOf(on, key, held); ok {
want[k], seqs[k] = v, held.Sequence
}
}
@@ -295,9 +286,7 @@ func (l *Ledger) Lag(buscalls.LedgerRequest) (buscalls.Lagged, error) {
}
at := map[string]uint64{}
for _, e := range held {
var v struct {
Sequence uint64 `json:"sequence"`
}
var v buscalls.IndexEntry
_ = json.Unmarshal(e.Value, &v)
at[e.Key] = v.Sequence
}