305 lines
8.2 KiB
Go
305 lines
8.2 KiB
Go
package ledger
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
|
)
|
|
|
|
// The ledger's repairs (ADR 0597 §8, §10; design 25 § 12.5; issue 1966 T4): rebuild an instance's state from its history
|
|
// or its snapshot, rewrite a snapshot behind its state, build a declared index again, and say what is behind its history.
|
|
// The tool runner serves them as tools of every module owning a ledger, its own ledgers only (the runtime's repairs.go).
|
|
|
|
// instances is every instance the ledger knows, by key: those its state holds and those its history names.
|
|
func (l *Ledger) instances() ([]string, error) {
|
|
seen := map[string]bool{}
|
|
keys, err := l.State.Keys()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, k := range keys {
|
|
seen[k] = true
|
|
}
|
|
kept, err := l.History.Read(">", 1, 0)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ops, err := operationsOf(kept)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, o := range ops {
|
|
if k, err := KeyOf(o.Instance); err == nil {
|
|
seen[k] = true
|
|
}
|
|
}
|
|
out := make([]string, 0, len(seen))
|
|
for k := range seen {
|
|
out = append(out, k)
|
|
}
|
|
sort.Strings(out)
|
|
return out, nil
|
|
}
|
|
|
|
// keysOf is the one instance a request names, or every instance when it names none.
|
|
func (l *Ledger) keysOf(instance string) ([]string, error) {
|
|
if instance == "" {
|
|
return l.instances()
|
|
}
|
|
k, err := KeyOf(instance)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return []string{k}, nil
|
|
}
|
|
|
|
// Rebuild derives the state of an instance (every instance when it names none) again, from its whole history or from its
|
|
// snapshot and the operations after it, and writes it, and its snapshot when that is behind. A derivation further back
|
|
// than the state holds is not written: the state never goes back (ADR 0597 §5).
|
|
func (l *Ledger) Rebuild(r buscalls.RebuildRequest) (buscalls.Repaired, error) {
|
|
from := r.From
|
|
if from == "" {
|
|
from = buscalls.RebuildFromHistory
|
|
}
|
|
if from != buscalls.RebuildFromHistory && from != buscalls.RebuildFromSnapshot {
|
|
return buscalls.Repaired{}, fmt.Errorf("a rebuild is from %q or %q, not %q", buscalls.RebuildFromHistory, buscalls.RebuildFromSnapshot, from)
|
|
}
|
|
keys, err := l.keysOf(r.Instance)
|
|
if err != nil {
|
|
return buscalls.Repaired{}, err
|
|
}
|
|
var out buscalls.Repaired
|
|
for _, key := range keys {
|
|
base := Record{}
|
|
if from == buscalls.RebuildFromSnapshot {
|
|
if base, _, err = l.readSnapshot(key); err != nil {
|
|
return out, err
|
|
}
|
|
}
|
|
derived, err := l.replay(key, base, 0)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
held, rev, err := l.readState(key)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
if derived.Sequence < held.Sequence {
|
|
continue
|
|
}
|
|
now, _, err := l.settle(key, derived, rev)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
if err := l.writeSnapshot(key, now); err != nil {
|
|
return out, err
|
|
}
|
|
out.Written++
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// Snapshot rewrites the snapshot of an instance (every instance when it names none) that is behind its state.
|
|
func (l *Ledger) Snapshot(r buscalls.InstanceRequest) (buscalls.Repaired, error) {
|
|
keys, err := l.keysOf(r.Instance)
|
|
if err != nil {
|
|
return buscalls.Repaired{}, err
|
|
}
|
|
var out buscalls.Repaired
|
|
for _, key := range keys {
|
|
held, _, err := l.readState(key)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
snap, at, err := l.readSnapshot(key)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
if held.Sequence == 0 || (at != 0 && snap.Sequence >= held.Sequence) {
|
|
continue
|
|
}
|
|
if err := l.writeSnapshot(key, held); err != nil {
|
|
return out, err
|
|
}
|
|
out.Written++
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// 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
|
|
}
|
|
values := make([]string, 0, len(on))
|
|
found := false
|
|
for _, f := range on {
|
|
v, has := state[f]
|
|
found = found || has
|
|
s, ok := v.(string)
|
|
if !ok {
|
|
raw, _ := json.Marshal(v)
|
|
s = string(raw)
|
|
}
|
|
values = append(values, s)
|
|
}
|
|
if !found {
|
|
return "", nil, false
|
|
}
|
|
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.
|
|
func (l *Ledger) declared(index string) ([]string, map[string]bool, error) {
|
|
types := map[string]bool{}
|
|
var on []string
|
|
for _, t := range l.Types {
|
|
for _, ix := range t.Indexes {
|
|
if ix.Name == index {
|
|
types[t.Name] = true
|
|
on = ix.On
|
|
}
|
|
}
|
|
}
|
|
if len(types) == 0 {
|
|
return nil, nil, fmt.Errorf("%s's ledger %s declares no index %q", l.Module, l.Name, index)
|
|
}
|
|
return on, types, nil
|
|
}
|
|
|
|
// wanted is every entry an index should hold, from the states of the types declaring it.
|
|
func (l *Ledger) wanted(index string) (map[string][]byte, map[string]uint64, error) {
|
|
on, types, err := l.declared(index)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
keys, err := l.instances()
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
want, seqs := map[string][]byte{}, map[string]uint64{}
|
|
for _, key := range keys {
|
|
name, _, _ := strings.Cut(key, ".")
|
|
if !types[name] {
|
|
continue
|
|
}
|
|
held, _, err := l.readState(key)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if k, v, ok := indexEntryOf(on, key, held); ok {
|
|
want[k], seqs[k] = v, held.Sequence
|
|
}
|
|
}
|
|
return want, seqs, nil
|
|
}
|
|
|
|
// Reindex builds a declared index again from the ledger's states, derived from its history: every entry it should hold
|
|
// written, every other removed.
|
|
func (l *Ledger) Reindex(r buscalls.ReindexRequest) (buscalls.Repaired, error) {
|
|
ix, err := l.index(r.Index)
|
|
if err != nil {
|
|
return buscalls.Repaired{}, err
|
|
}
|
|
want, _, err := l.wanted(r.Index)
|
|
if err != nil {
|
|
return buscalls.Repaired{}, err
|
|
}
|
|
held, err := ix.Entries("", 0)
|
|
if err != nil {
|
|
return buscalls.Repaired{}, err
|
|
}
|
|
var out buscalls.Repaired
|
|
for _, e := range held {
|
|
if _, keep := want[e.Key]; !keep {
|
|
if err := ix.Delete(e.Key); err != nil {
|
|
return out, err
|
|
}
|
|
}
|
|
}
|
|
keys := make([]string, 0, len(want))
|
|
for k := range want {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
for _, k := range keys {
|
|
if err := ix.Put(k, want[k]); err != nil {
|
|
return out, err
|
|
}
|
|
out.Written++
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// Lag is every instance whose state or snapshot is behind its history, and how many entries each declared index lacks
|
|
// or holds from an older state; nothing when nothing is behind.
|
|
func (l *Ledger) Lag(buscalls.LedgerRequest) (buscalls.Lagged, error) {
|
|
keys, err := l.instances()
|
|
if err != nil {
|
|
return buscalls.Lagged{}, err
|
|
}
|
|
var out buscalls.Lagged
|
|
for _, key := range keys {
|
|
last, err := l.History.Last(key)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
held, _, err := l.readState(key)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
snap, _, err := l.readSnapshot(key)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
if held.Sequence < last || snap.Sequence < last {
|
|
out.Instances = append(out.Instances, buscalls.InstanceLag{Instance: InstanceOf(key), History: last,
|
|
State: held.Sequence, Snapshot: snap.Sequence})
|
|
}
|
|
}
|
|
names := make([]string, 0, len(l.Indexes))
|
|
for name := range l.Indexes {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
for _, name := range names {
|
|
if _, _, err := l.declared(name); err != nil {
|
|
// A bucket bound for an index no type declares has nothing it should hold.
|
|
continue
|
|
}
|
|
want, seqs, err := l.wanted(name)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
held, err := l.Indexes[name].Entries("", 0)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
at := map[string]uint64{}
|
|
for _, e := range held {
|
|
var v buscalls.IndexEntry
|
|
_ = json.Unmarshal(e.Value, &v)
|
|
at[e.Key] = v.Sequence
|
|
}
|
|
behind := uint64(0)
|
|
for k := range want {
|
|
if s, has := at[k]; !has || s < seqs[k] {
|
|
behind++
|
|
}
|
|
}
|
|
if behind > 0 {
|
|
out.Indexes = append(out.Indexes, buscalls.IndexLag{Index: name, Behind: behind})
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|