From ad22c5e9fe7c2162d50bda4b16bc0a42682814b7 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 13:40:40 +0100 Subject: [PATCH] Apply a ledger operation: check it against its type, append it with the last sequence expected, derive the state at the revision read, write the snapshot (issue 1966 T2) --- node-tools/internal/bus/ledger.go | 229 ++++++++++ node-tools/internal/ledger/ledger.go | 493 ++++++++++++++++++++++ node-tools/internal/ledger/ledger_test.go | 102 +++++ node-tools/internal/ledger/memory.go | 158 +++++++ node-tools/internal/ledger/types_test.go | 60 +++ 5 files changed, 1042 insertions(+) create mode 100644 node-tools/internal/bus/ledger.go create mode 100644 node-tools/internal/ledger/ledger.go create mode 100644 node-tools/internal/ledger/ledger_test.go create mode 100644 node-tools/internal/ledger/memory.go create mode 100644 node-tools/internal/ledger/types_test.go diff --git a/node-tools/internal/bus/ledger.go b/node-tools/internal/bus/ledger.go new file mode 100644 index 0000000..bf46d53 --- /dev/null +++ b/node-tools/internal/bus/ledger.go @@ -0,0 +1,229 @@ +package bus + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + + "git.novox.be/novox/mesh-sdk/go/membership" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-tools/node-tools/internal/ledger" +) + +// A ledger on the bus (novox/hq ADR 0597, design 25 § 12.4; issue 1966): its history a stream with a subject per +// instance under the binding's subject, its state a key-value bucket, its snapshots a subject per instance under the +// binding's snapshot subject, the latest only (the stream that holds them keeps one per subject, as the bus module +// creates it). **Only the module's ledger connection publishes on its history and its snapshots** (ADR 0597 §9): what +// is read, and the state, go on the runner's own. + +// LedgerStores are the stores of one ledger binding: reads and the state on c, the history's appends and the snapshots' +// writes on writer, the module's ledger connection. +func (c *Conn) LedgerStores(b membership.BusBinding, writer *Conn) (ledger.History, ledger.State, ledger.Snapshots) { + return &busHistory{read: c, write: writer, stream: b.History, prefix: prefixOf(b.Subject)}, + &busState{c: c, bucket: b.State}, + &busSnapshots{read: c, write: writer, prefix: prefixOf(b.Snapshot)} +} + +// prefixOf is a binding's subject as the prefix an instance's key is added to: written with or without its `.>`. +func prefixOf(subject string) string { + return strings.TrimSuffix(strings.TrimSuffix(subject, ">"), ".") +} + +// publishExpecting publishes body on subject on the connection, written only when the subject's last message is still +// at expect (0: it has none); ledger.ErrRaced when it is not. +func publishExpecting(c *Conn, subject string, body []byte, expect uint64) (uint64, error) { + msg := nats.NewMsg(subject) + msg.Data = body + msg.Header.Set(jetstream.ExpectedLastSubjSeqHeader, strconv.FormatUint(expect, 10)) + reply, err := c.nc.RequestMsg(msg, StateTimeout) + if errors.Is(err, nats.ErrNoResponders) { + return 0, fmt.Errorf("no stream on the bus keeps %s", subject) + } + if err != nil { + return 0, err + } + var ack struct { + Seq uint64 `json:"seq"` + Error *struct { + ErrCode uint16 `json:"err_code"` + Description string `json:"description"` + } `json:"error"` + } + if err := json.Unmarshal(reply.Data, &ack); err != nil { + return 0, fmt.Errorf("the bus answered the write to %s with what is not an acknowledgement: %q", subject, reply.Data) + } + if ack.Error != nil { + if code := jetstream.ErrorCode(ack.Error.ErrCode); code == jetstream.JSErrCodeStreamWrongLastSequence || + code == jetstream.JSErrCodeStreamWrongLastSequenceConstant { + return 0, ledger.ErrRaced + } + return 0, fmt.Errorf("the bus refused the write to %s: %s (%d)", subject, ack.Error.Description, ack.Error.ErrCode) + } + return ack.Seq, nil +} + +// lastOn is the last message on a subject of the stream that keeps it: its bytes and sequence, nil and 0 for none. +func lastOn(c *Conn, stream, subject string) ([]byte, uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + js, err := c.jetStream() + if err != nil { + return nil, 0, err + } + if stream == "" { + if stream, err = js.StreamNameBySubject(ctx, subject); err != nil { + return nil, 0, fmt.Errorf("no stream on the bus keeps %s: %w", subject, err) + } + } + s, err := js.Stream(ctx, stream) + if err != nil { + return nil, 0, fmt.Errorf("the stream %s: %w", stream, err) + } + m, err := s.GetLastMsgForSubject(ctx, subject) + if errors.Is(err, jetstream.ErrMsgNotFound) { + return nil, 0, nil + } + if err != nil { + return nil, 0, err + } + return m.Data, m.Sequence, nil +} + +type busHistory struct { + read, write *Conn + stream, prefix string +} + +func (h *busHistory) Last(key string) (uint64, error) { + _, seq, err := lastOn(h.read, h.stream, h.prefix+"."+key) + return seq, err +} + +func (h *busHistory) Append(key string, op []byte, expect uint64) (uint64, error) { + if h.write == nil { + return 0, errors.New("the ledger's history is published only on its module's ledger connection, and none is held") + } + return publishExpecting(h.write, h.prefix+"."+key, op, expect) +} + +func (h *busHistory) Read(keys string, from uint64, limit int) ([]ledger.Kept, error) { + var out []ledger.Kept + for { + want := StreamReadMost + if limit > 0 && limit-len(out) < want { + want = limit - len(out) + } + page, err := h.read.readStream(h.stream, h.prefix+"."+keys, from, want) + if err != nil { + return out, err + } + for _, m := range page.Messages { + out = append(out, ledger.Kept{Sequence: m.Sequence, Data: m.Body}) + } + if page.Next == 0 || (limit > 0 && len(out) >= limit) { + return out, nil + } + from = page.Next + } +} + +type busState struct { + c *Conn + bucket string +} + +func (s *busState) kv(ctx context.Context) (jetstream.KeyValue, error) { + js, err := s.c.jetStream() + if err != nil { + return nil, err + } + kv, err := js.KeyValue(ctx, s.bucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return nil, fmt.Errorf("the ledger's state %s is not on the bus", s.bucket) + } + return kv, err +} + +func (s *busState) Get(key string) ([]byte, uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := s.kv(ctx) + if err != nil { + return nil, 0, err + } + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, 0, nil + } + if err != nil { + return nil, 0, err + } + return e.Value(), e.Revision(), nil +} + +// isFull is a write a bucket or stream refused because it holds as much as it may. +func isFull(err error) bool { + msg := err.Error() + return strings.Contains(msg, "maximum bytes exceeded") || strings.Contains(msg, "maximum messages exceeded") || + strings.Contains(msg, "insufficient resources") +} + +func (s *busState) Write(key string, value []byte, revision uint64) (uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := s.kv(ctx) + if err != nil { + return 0, err + } + var rev uint64 + if revision == 0 { + rev, err = kv.Create(ctx, key, value) + } else { + rev, err = kv.Update(ctx, key, value, revision) + } + switch { + case err == nil: + return rev, nil + case errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) || isWrongSequence(err): + return 0, ledger.ErrChanged + case isFull(err): + return 0, fmt.Errorf("%w: %v", ledger.ErrFull, err) + } + return 0, err +} + +func (s *busState) Keys() ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := s.kv(ctx) + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + +type busSnapshots struct { + read, write *Conn + prefix string +} + +func (s *busSnapshots) Get(key string) ([]byte, uint64, error) { + return lastOn(s.read, "", s.prefix+"."+key) +} + +func (s *busSnapshots) Put(key string, value []byte, expect uint64) error { + if s.write == nil { + return errors.New("a ledger's snapshot is published only on its module's ledger connection, and none is held") + } + _, err := publishExpecting(s.write, s.prefix+"."+key, value, expect) + return err +} diff --git a/node-tools/internal/ledger/ledger.go b/node-tools/internal/ledger/ledger.go new file mode 100644 index 0000000..e3b055b --- /dev/null +++ b/node-tools/internal/ledger/ledger.go @@ -0,0 +1,493 @@ +// Package ledger is the tool runner's side of a ledger (novox/hq ADR 0597, design 25 § 12.4; issue 1966): a storage type +// whose state is derived from its history. `apply` checks an operation against its type, appends it to the instance's +// history with the last sequence expected, derives the state by the operation's effect and writes it as an update of the +// revision read, and writes the instance's snapshot. What the history, the state and the snapshots are kept in is given +// (History, State, Snapshots): the bus in the runtime, fakes in a test. +package ledger + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "regexp" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// The refusals the stores answer, told by errors.Is. +var ( + // ErrRaced is an append or a snapshot written with an expected sequence that is no longer the last: another writer + // wrote first. + ErrRaced = errors.New("another writer wrote the instance's history first") + // ErrChanged is a state written at a revision that is no longer the key's. + ErrChanged = errors.New("the state was written by another writer first") + // ErrFull is a state bucket that refuses a write because it is full. + ErrFull = errors.New("the state bucket is full") +) + +// CodeChanged is the error code a bundle is answered a lost race with, as a lost state update (mesh-sdk stdio.CodeChanged). +const CodeChanged = -32010 + +// raced is an apply refused because another writer appended to the instance first (ADR 0597 §5). +type raced struct{ instance string } + +func (e raced) Error() string { + return "the operation on " + e.instance + " is not taken: another writer appended to its history first; read it and apply again (novox/hq ADR 0597 §5)" +} +func (e raced) Is(target error) bool { return target == ErrRaced } +func (e raced) ErrorCode() int { return CodeChanged } + +// Kept is one message a history keeps: its sequence and its bytes. +type Kept struct { + Sequence uint64 + Data []byte +} + +// History is a ledger's history: a stream with a subject per instance, each instance named by its key. +type History interface { + // Last is the sequence of the instance's last operation, 0 when it has none. + Last(key string) (uint64, error) + // Append writes one operation, expecting the instance's last sequence to be expect (0: it has none); ErrRaced when + // it is not. + Append(key string, op []byte, expect uint64) (uint64, error) + // Read is the operations of the instances keys names (a key, with `*` for one name and `>` for the rest) from a + // sequence on, oldest first, at most limit of them (every one when limit is 0). + Read(keys string, from uint64, limit int) ([]Kept, error) +} + +// State is a ledger's state bucket: each instance's record under its key. +type State interface { + // Get is the key's value and revision; nil and 0 when it has none. + Get(key string) ([]byte, uint64, error) + // Write writes the key's value at the revision read (0: only when it has none); ErrChanged when the key is no longer + // at it, ErrFull when the bucket is full. + Write(key string, value []byte, revision uint64) (uint64, error) + // Keys is every key the bucket holds. + Keys() ([]string, error) +} + +// Snapshots are a ledger's snapshots: the latest record of each instance, on a subject of its own, never expiring. +type Snapshots interface { + // Get is the key's latest snapshot and the sequence it was written at; nil and 0 when it has none. + Get(key string) ([]byte, uint64, error) + // Put writes the key's snapshot, expecting its last to be at expect (0: none); ErrRaced when it is not. + Put(key string, value []byte, expect uint64) error +} + +// Ledger is one ledger of one module: its types and where it is kept. +type Ledger struct { + Module string + Name string + Types []catalogue.LedgerType + History History + State State + Snapshots Snapshots + // 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. + Logf func(format string, args ...any) +} + +func (l *Ledger) say(format string, args ...any) { + if l.Logf != nil { + l.Logf(format, args...) + } +} + +// Record is what the state and a snapshot keep of an instance: its state and the sequence of the last operation in it. +type Record struct { + Sequence uint64 `json:"sequence"` + State json.RawMessage `json:"state"` +} + +// segment is one part of an instance's path. +var segment = regexp.MustCompile(`^[A-Za-z0-9_-]+$`) + +// KeyOf is the key an instance is kept under, in the state and as its history's and snapshot's subject token: its path's +// parts joined by dots (`decision/0290` is `decision.0290`). +func KeyOf(instance string) (string, error) { + parts := strings.Split(instance, "/") + if len(parts) < 2 { + return "", fmt.Errorf("%q is not an instance: an instance is named by a path whose first part is its type's name, "+ + "such as decision/0290 (novox/hq ADR 0597 §1)", instance) + } + for _, p := range parts { + if !segment.MatchString(p) { + return "", fmt.Errorf("%q is not an instance: each part of its path holds letters, digits, - and _ alone", instance) + } + } + return strings.Join(parts, "."), nil +} + +// InstanceOf is the instance a key names. +func InstanceOf(key string) string { return strings.ReplaceAll(key, ".", "/") } + +func (l *Ledger) now() time.Time { + if l.Now != nil { + return l.Now().UTC() + } + return time.Now().UTC() +} + +// typeOf is the instance's type, by its path's first part. +func (l *Ledger) typeOf(instance string) (catalogue.LedgerType, error) { + name, _, _ := strings.Cut(instance, "/") + for _, t := range l.Types { + if t.Name == name { + return t, nil + } + } + return catalogue.LedgerType{}, fmt.Errorf("%s's ledger %s has no type %q, which the instance %s names", l.Module, l.Name, name, instance) +} + +// operationOf is the operation a type allows by its name. +func operationOf(t catalogue.LedgerType, name string) (catalogue.LedgerOperation, error) { + for _, op := range t.Operations { + if op.Name == name { + return op, nil + } + } + return catalogue.LedgerOperation{}, fmt.Errorf("the type %s allows no operation %q", t.Name, name) +} + +// Digest is the digest a type pairs a text field with (ADR 0597 §2): sha256 of the text, as the knowledge base writes it. +func Digest(text string) string { + sum := sha256.Sum256([]byte(text)) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// kinds is the JSON kind each schema type name is held to; a name not here is the judge's to know. +var kinds = map[string]func(any) bool{ + "string": func(v any) bool { _, ok := v.(string); return ok }, + "number": func(v any) bool { _, ok := v.(float64); return ok }, + "integer": func(v any) bool { f, ok := v.(float64); return ok && f == float64(int64(f)) }, + "boolean": func(v any) bool { _, ok := v.(bool); return ok }, + "object": func(v any) bool { _, ok := v.(map[string]any); return ok }, + "array": func(v any) bool { _, ok := v.([]any); return ok }, +} + +// checked is an operation as apply checked it: its instance's key, its operation and its body. +type checked struct { + key string + op catalogue.LedgerOperation + body map[string]any +} + +// check holds an operation to its type (ADR 0597 §2): a type the instance names, an operation it allows, a body that is an +// object whose fields fit the schema, the key its effect names, and every digest matching its text. +func (l *Ledger) check(r buscalls.ApplyRequest) (checked, error) { + key, err := KeyOf(r.Instance) + if err != nil { + return checked{}, err + } + t, err := l.typeOf(r.Instance) + if err != nil { + return checked{}, err + } + op, err := operationOf(t, r.Operation) + if err != nil { + return checked{}, err + } + body := map[string]any{} + if len(r.Body) > 0 && string(r.Body) != "null" { + if err := json.Unmarshal(r.Body, &body); err != nil { + return checked{}, fmt.Errorf("the operation %s's body is a JSON object: %w", r.Operation, err) + } + } + for name, field := range t.Schema { + v, has := body[name] + if !has { + continue + } + if fits, known := kinds[field.Type]; known && !fits(v) { + return checked{}, fmt.Errorf("the operation %s's field %s is a %s in the type %s", r.Operation, name, field.Type, t.Name) + } + } + for name, field := range t.Schema { + if field.DigestOf == "" { + continue + } + digest, hasDigest := body[name] + text, hasText := body[field.DigestOf] + if !hasDigest && !hasText { + continue + } + s, isText := text.(string) + if !hasDigest || !hasText || !isText || digest != Digest(s) { + return checked{}, fmt.Errorf("the operation %s's %s is not the digest of its %s, so it is refused (novox/hq ADR 0597 §2)", + r.Operation, name, field.DigestOf) + } + } + switch op.Effect { + case catalogue.EffectSetFields, catalogue.EffectAppend: + case catalogue.EffectSetEntry, catalogue.EffectRemove: + if op.Effect == catalogue.EffectRemove && op.Into == "" { + break + } + if _, ok := body[op.Key].(string); !ok { + return checked{}, fmt.Errorf("the operation %s names its entry by its field %s, a text, which its body lacks", r.Operation, op.Key) + } + default: + return checked{}, fmt.Errorf("the operation %s's effect %q is not one the mesh offers (novox/hq ADR 0597 §2)", r.Operation, op.Effect) + } + return checked{key: key, op: op, body: body}, nil +} + +// effect is an operation's effect on a state (ADR 0597 §2): it copies only the fields the operation names, and the +// operation's sequence and proven author where it says so; null is a removed instance. +func effect(state json.RawMessage, op catalogue.LedgerOperation, body map[string]any, seq uint64, author buscalls.Author) (json.RawMessage, error) { + s := map[string]any{} + if len(state) > 0 && string(state) != "null" { + if err := json.Unmarshal(state, &s); err != nil { + return nil, fmt.Errorf("the state held is not an object: %w", err) + } + } + copied := map[string]any{} + for _, f := range op.Fields { + if v, ok := body[f]; ok { + copied[f] = v + } + } + if op.Sequence { + copied["sequence"] = seq + } + // Written last, so a body's own `author` never stands for the proven one (ADR 0597 §7). + if op.Author { + copied["author"] = author + } + switch op.Effect { + case catalogue.EffectSetFields: + for k, v := range copied { + s[k] = v + } + case catalogue.EffectSetEntry: + m, _ := s[op.Into].(map[string]any) + if m == nil { + m = map[string]any{} + } + m[body[op.Key].(string)] = copied + s[op.Into] = m + case catalogue.EffectAppend: + list, _ := s[op.Into].([]any) + s[op.Into] = append(list, copied) + case catalogue.EffectRemove: + if op.Into == "" { + return json.RawMessage("null"), nil + } + if m, ok := s[op.Into].(map[string]any); ok { + delete(m, body[op.Key].(string)) + } + } + return json.Marshal(s) +} + +// fold applies one kept operation to a record, as apply derived it. +func (l *Ledger) fold(rec Record, k Kept) (Record, error) { + var o buscalls.Operation + if err := json.Unmarshal(k.Data, &o); err != nil { + return rec, fmt.Errorf("the history's operation %d is not one: %w", k.Sequence, err) + } + t, err := l.typeOf(o.Instance) + if err != nil { + return rec, err + } + op, err := operationOf(t, o.Operation) + if err != nil { + return rec, err + } + body := map[string]any{} + if len(o.Body) > 0 && string(o.Body) != "null" { + if err := json.Unmarshal(o.Body, &body); err != nil { + return rec, err + } + } + if (op.Effect == catalogue.EffectSetEntry || (op.Effect == catalogue.EffectRemove && op.Into != "")) && !isText(body[op.Key]) { + return rec, fmt.Errorf("the history's operation %d names no entry", k.Sequence) + } + state, err := effect(rec.State, op, body, k.Sequence, o.Author) + if err != nil { + return rec, err + } + return Record{Sequence: k.Sequence, State: state}, nil +} + +func isText(v any) bool { _, ok := v.(string); return ok } + +// replay folds the instance's operations after rec's sequence, up to upto (every one when 0), onto rec. +func (l *Ledger) replay(key string, rec Record, upto uint64) (Record, error) { + kept, err := l.History.Read(key, rec.Sequence+1, 0) + if err != nil { + return rec, err + } + for _, k := range kept { + if upto != 0 && k.Sequence > upto { + break + } + if k.Sequence <= rec.Sequence { + continue + } + if rec, err = l.fold(rec, k); err != nil { + return rec, err + } + } + return rec, nil +} + +// readState is the instance's record in the state and its revision; an empty record when it has none. +func (l *Ledger) readState(key string) (Record, uint64, error) { + raw, rev, err := l.State.Get(key) + if err != nil || raw == nil { + return Record{}, rev, err + } + var rec Record + if err := json.Unmarshal(raw, &rec); err != nil { + return Record{}, rev, fmt.Errorf("the state of %s is not a record: %w", key, err) + } + return rec, rev, nil +} + +// readSnapshot is the instance's snapshot and the sequence it was written at; an empty record when it has none. +func (l *Ledger) readSnapshot(key string) (Record, uint64, error) { + raw, at, err := l.Snapshots.Get(key) + if err != nil || raw == nil { + return Record{}, at, err + } + var rec Record + if err := json.Unmarshal(raw, &rec); err != nil { + return Record{}, at, fmt.Errorf("the snapshot of %s is not a record: %w", key, err) + } + return rec, at, nil +} + +// derive is the instance's record caught up with its history up to last: from the state, or from the snapshot when it +// is further on (an instance whose state was lost, or whose operations expired, ADR 0597 §8). +func (l *Ledger) derive(key string, base Record, last uint64) (Record, error) { + if base.Sequence >= last { + return base, nil + } + snap, _, err := l.readSnapshot(key) + if err != nil { + return base, err + } + if snap.Sequence > base.Sequence { + base = snap + } + return l.replay(key, base, last) +} + +// settle writes want into the state at the revision read (ADR 0597 §5): a lost update re-reads, and catches the state +// read up to want's sequence from the history, never appending again; a state already further on is left as it is. It +// answers what the state holds after it, and whether it is behind because the bucket is full (ADR 0597 §6). +func (l *Ledger) settle(key string, want Record, rev uint64) (Record, bool, error) { + for tries := 0; tries < 16; tries++ { + raw, err := json.Marshal(want) + if err != nil { + return want, false, err + } + _, err = l.State.Write(key, raw, rev) + switch { + case err == nil: + return want, false, nil + case errors.Is(err, ErrFull): + return want, true, nil + case !errors.Is(err, ErrChanged): + return want, false, err + } + cur, curRev, err := l.readState(key) + if err != nil { + return want, false, err + } + if cur.Sequence >= want.Sequence { + return cur, false, nil + } + if want, err = l.replay(key, cur, want.Sequence); err != nil { + return want, false, err + } + rev = curRev + } + return want, false, fmt.Errorf("the state of %s kept being written by other writers; the operation is in its history and the next apply catches it up", key) +} + +// writeSnapshot writes rec as the instance's snapshot unless the snapshot is already as far on (ADR 0597 §8). +func (l *Ledger) writeSnapshot(key string, rec Record) error { + for tries := 0; tries < 16; tries++ { + snap, at, err := l.readSnapshot(key) + if err != nil { + return err + } + if at != 0 && snap.Sequence >= rec.Sequence { + return nil + } + raw, err := json.Marshal(rec) + if err != nil { + return err + } + if err = l.Snapshots.Put(key, raw, at); err == nil || !errors.Is(err, ErrRaced) { + return err + } + } + return fmt.Errorf("the snapshot of %s kept being written by other writers", key) +} + +// Apply applies one operation to one instance, stamped with author (ADR 0597 §3, §5, §6, §8; design 25 § 12.4 steps 1-5): +// checked against its type, appended with the instance's last sequence expected, the state derived by its effect and +// written at the revision read, and the snapshot written. A racing writer's append is refused; a full state bucket +// answers taken with the state behind. +func (l *Ledger) Apply(r buscalls.ApplyRequest, author buscalls.Author) (buscalls.Applied, error) { + c, err := l.check(r) + if err != nil { + return buscalls.Applied{}, err + } + last, err := l.History.Last(c.key) + if err != nil { + return buscalls.Applied{}, err + } + held, rev, err := l.readState(c.key) + if err != nil { + return buscalls.Applied{}, err + } + base, err := l.derive(c.key, held, last) + if err != nil { + return buscalls.Applied{}, err + } + body := r.Body + if len(body) == 0 { + body = json.RawMessage("{}") + } + op, err := json.Marshal(buscalls.Operation{Instance: r.Instance, Operation: r.Operation, Body: body, Author: author, At: l.now()}) + if err != nil { + return buscalls.Applied{}, err + } + seq, err := l.History.Append(c.key, op, last) + if errors.Is(err, ErrRaced) { + return buscalls.Applied{}, raced{r.Instance} + } + if err != nil { + return buscalls.Applied{}, err + } + state, err := effect(base.State, c.op, c.body, seq, author) + if err != nil { + return buscalls.Applied{}, err + } + // From here the operation is taken: it is in the history, and a fault below is the state's to catch up, said here + // and answered as the state behind, never as a refusal of what later takes effect (ADR 0597 §6). + now, behind, err := l.settle(c.key, Record{Sequence: seq, State: state}, rev) + if err != nil { + l.say("[mesh-tools] %s's ledger %s took operation %d on %s and its state could not follow: %v; the next apply "+ + "or a rebuild catches it up", l.Module, l.Name, seq, r.Instance, err) + behind, now = true, Record{Sequence: seq, State: state} + } + if behind { + now = Record{Sequence: seq, State: state} + } + if err := l.writeSnapshot(c.key, now); err != nil { + l.say("[mesh-tools] %s's ledger %s took operation %d on %s and its snapshot could not be written: %v; the next "+ + "apply or a snapshot repair writes it", l.Module, l.Name, seq, r.Instance, err) + } + return buscalls.Applied{Taken: true, Sequence: seq, StateBehind: behind}, nil +} diff --git a/node-tools/internal/ledger/ledger_test.go b/node-tools/internal/ledger/ledger_test.go new file mode 100644 index 0000000..01b3017 --- /dev/null +++ b/node-tools/internal/ledger/ledger_test.go @@ -0,0 +1,102 @@ +package ledger + +import ( + "encoding/json" + "fmt" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +var owner = buscalls.Author{Account: "ledger.anchor.m"} + +// 1966/AC13, 1966/AC14 and the checks of an operation against its type (ADR 0597 §2, §5, §6, §8; design 25 § 12.4). +func TestApplyAppendsDerivesAndSnapshots(t *testing.T) { + cases := []struct { + name string + run func(t *testing.T, l *Ledger, f fakes) + }{ + {"1966/AC13 an instance at sequence 4 is taken with 5, its state and snapshot the effect's result", func(t *testing.T, l *Ledger, f fakes) { + for i := 1; i <= 4; i++ { + applied(t, l, "decision/0290", "note", fmt.Sprintf(`{"says":"n%d"}`, i), owner) + } + if last, _ := f.history.Last("decision.0290"); last != 4 { + t.Fatalf("the instance is at %d, want 4", last) + } + got := applied(t, l, "decision/0290", "title", `{"title":"A ledger","ignored":"x"}`, owner) + if !got.Taken || got.Sequence != 5 || got.StateBehind { + t.Fatalf("answered %+v, want taken with 5", got) + } + raw, _, _ := f.state.Get("decision.0290") + state := record(t, raw) + want := `{"author":{"account":"ledger.anchor.m"},"notes":[{"says":"n1"},{"says":"n2"},{"says":"n3"},{"says":"n4"}],"sequence":5,"title":"A ledger"}` + if state.Sequence != 5 || string(state.State) != want { + t.Fatalf("the state is %d %s, want 5 %s", state.Sequence, state.State, want) + } + snap, _, _ := f.snapshots.Get("decision.0290") + if string(snap) != string(raw) { + t.Fatalf("the snapshot is %s, the state %s", snap, raw) + } + }}, + {"1966/AC14 a full state bucket answers taken and state behind, and the next apply after room catches up", func(t *testing.T, l *Ledger, f fakes) { + applied(t, l, "decision/0290", "note", `{"says":"first"}`, owner) + f.state.Full = true + got := applied(t, l, "decision/0290", "note", `{"says":"while full"}`, owner) + if !got.Taken || got.Sequence != 2 || !got.StateBehind { + t.Fatalf("answered %+v, want taken with 2 and state behind", got) + } + if raw, _, _ := f.state.Get("decision.0290"); record(t, raw).Sequence != 1 { + t.Fatalf("the full state moved: %s", raw) + } + f.state.Full = false + got = applied(t, l, "decision/0290", "note", `{"says":"after room"}`, owner) + raw, _, _ := f.state.Get("decision.0290") + state := record(t, raw) + if got.StateBehind || state.Sequence != 3 || string(state.State) != `{"notes":[{"says":"first"},{"says":"while full"},{"says":"after room"}]}` { + t.Fatalf("after room: %+v, the state %d %s", got, state.Sequence, state.State) + } + }}, + {"each effect: set an entry, remove it, remove the instance", func(t *testing.T, l *Ledger, f fakes) { + text := "the section's text" + applied(t, l, "decision/1", "section", fmt.Sprintf(`{"id":"a","text":%q,"digest":%q}`, text, Digest(text)), owner) + applied(t, l, "decision/1", "section", fmt.Sprintf(`{"id":"b","text":"b","digest":%q}`, Digest("b")), owner) + applied(t, l, "decision/1", "drop-section", `{"id":"a"}`, owner) + raw, _, _ := f.state.Get("decision.1") + if got := string(record(t, raw).State); got != fmt.Sprintf(`{"sections":{"b":{"digest":%q}}}`, Digest("b")) { + t.Fatalf("the state is %s", got) + } + applied(t, l, "decision/1", "withdraw", `{}`, owner) + raw, _, _ = f.state.Get("decision.1") + if got := record(t, raw); got.Sequence != 4 || string(got.State) != "null" { + t.Fatalf("the removed instance is %d %s", got.Sequence, got.State) + } + }}, + {"an operation outside its type, a body off its schema and a digest off its text are refused", func(t *testing.T, l *Ledger, f fakes) { + for _, c := range []struct{ instance, operation, body, refusal string }{ + {"listing/a17", "title", `{}`, `no type "listing"`}, + {"decision/0290", "accept", `{}`, `no operation "accept"`}, + {"decision/0290", "title", `{"title":7}`, "field title is a string"}, + {"decision/0290", "section", `{"id":"a","text":"x","digest":"sha256:00"}`, "is not the digest of its text"}, + {"decision/0290", "section", `{"text":"x"}`, "is not the digest"}, + {"decision/0290", "drop-section", `{}`, "names its entry by its field id"}, + {"decision", "title", `{}`, "is not an instance"}, + {"decision/a.b", "title", `{}`, "is not an instance"}, + } { + _, err := l.Apply(buscalls.ApplyRequest{Instance: c.instance, Operation: c.operation, Body: json.RawMessage(c.body)}, owner) + if err == nil || !strings.Contains(err.Error(), c.refusal) { + t.Errorf("%s %s %s: %v, want %q", c.instance, c.operation, c.body, err, c.refusal) + } + } + if last, _ := f.history.Last("decision.0290"); last != 0 { + t.Fatalf("a refused operation was appended: %d", last) + } + }}, + } + 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/memory.go b/node-tools/internal/ledger/memory.go new file mode 100644 index 0000000..c038888 --- /dev/null +++ b/node-tools/internal/ledger/memory.go @@ -0,0 +1,158 @@ +package ledger + +import ( + "sort" + "strings" + "sync" +) + +// 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 +// sequence counts across every instance, a bucket with revisions that may be full, and snapshots written with an expected +// sequence. + +// MemoryHistory is a history in memory. +type MemoryHistory struct { + mu sync.Mutex + seq uint64 + kept []struct { + key string + Kept + } +} + +// Matches is whether a key pattern (`*` one name, `>` the rest) names a key. +func Matches(pattern, key string) bool { + p, k := strings.Split(pattern, "."), strings.Split(key, ".") + for i, part := range p { + if part == ">" { + return len(k) > i + } + if i >= len(k) || (part != "*" && part != k[i]) { + return false + } + } + return len(p) == len(k) +} + +func (h *MemoryHistory) Last(key string) (uint64, error) { + h.mu.Lock() + defer h.mu.Unlock() + last := uint64(0) + for _, k := range h.kept { + if k.key == key { + last = k.Sequence + } + } + return last, nil +} + +func (h *MemoryHistory) Append(key string, op []byte, expect uint64) (uint64, error) { + last, _ := h.Last(key) + h.mu.Lock() + defer h.mu.Unlock() + if last != expect { + return 0, ErrRaced + } + h.seq++ + h.kept = append(h.kept, struct { + key string + Kept + }{key, Kept{Sequence: h.seq, Data: op}}) + return h.seq, nil +} + +func (h *MemoryHistory) Read(keys string, from uint64, limit int) ([]Kept, error) { + h.mu.Lock() + defer h.mu.Unlock() + var out []Kept + for _, k := range h.kept { + if k.Sequence >= from && Matches(keys, k.key) && (limit == 0 || len(out) < limit) { + out = append(out, k.Kept) + } + } + return out, nil +} + +// Expire drops every operation of the history, as a ledger's keeping days end them. +func (h *MemoryHistory) Expire() { + h.mu.Lock() + defer h.mu.Unlock() + h.kept = nil +} + +type memoryValue struct { + value []byte + rev uint64 +} + +// MemoryState is a state bucket in memory. +type MemoryState struct { + mu sync.Mutex + rev uint64 + // Full makes every write refused as a full bucket's. + Full bool + kv map[string]memoryValue +} + +func (s *MemoryState) Get(key string) ([]byte, uint64, error) { + s.mu.Lock() + defer s.mu.Unlock() + v := s.kv[key] + return v.value, v.rev, nil +} + +func (s *MemoryState) Write(key string, value []byte, revision uint64) (uint64, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.Full { + return 0, ErrFull + } + if s.kv[key].rev != revision { + return 0, ErrChanged + } + if s.kv == nil { + s.kv = map[string]memoryValue{} + } + s.rev++ + s.kv[key] = memoryValue{value, s.rev} + return s.rev, nil +} + +func (s *MemoryState) Keys() ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + var out []string + for k := range s.kv { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +// MemorySnapshots are snapshots in memory. +type MemorySnapshots struct { + mu sync.Mutex + at uint64 + kv map[string]memoryValue +} + +func (s *MemorySnapshots) Get(key string) ([]byte, uint64, error) { + s.mu.Lock() + defer s.mu.Unlock() + v := s.kv[key] + return v.value, v.rev, nil +} + +func (s *MemorySnapshots) Put(key string, value []byte, expect uint64) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.kv[key].rev != expect { + return ErrRaced + } + if s.kv == nil { + s.kv = map[string]memoryValue{} + } + s.at++ + s.kv[key] = memoryValue{value, s.at} + return nil +} diff --git a/node-tools/internal/ledger/types_test.go b/node-tools/internal/ledger/types_test.go new file mode 100644 index 0000000..7d8908a --- /dev/null +++ b/node-tools/internal/ledger/types_test.go @@ -0,0 +1,60 @@ +package ledger + +import ( + "encoding/json" + "testing" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// records is a ledger type of each effect: a title set, a section set as an entry, a note appended, a section dropped, +// and the whole instance removed. +var records = catalogue.LedgerType{ + Name: "decision", + Schema: map[string]catalogue.LedgerField{ + "title": {Type: "string"}, "text": {Type: "string"}, "digest": {Type: "string", DigestOf: "text"}, + }, + Operations: []catalogue.LedgerOperation{ + {Name: "title", Effect: catalogue.EffectSetFields, Fields: []string{"title"}, Sequence: true, Author: true}, + {Name: "section", Effect: catalogue.EffectSetEntry, Into: "sections", Key: "id", Fields: []string{"digest"}}, + {Name: "note", Effect: catalogue.EffectAppend, Into: "notes", Fields: []string{"says"}}, + {Name: "drop-section", Effect: catalogue.EffectRemove, Into: "sections", Key: "id"}, + {Name: "withdraw", Effect: catalogue.EffectRemove}, + }, + Indexes: []catalogue.LedgerIndex{{Name: "by-title", On: []string{"title"}}}, +} + +type fakes struct { + history *MemoryHistory + state *MemoryState + snapshots *MemorySnapshots +} + +func newLedger() (*Ledger, fakes) { + f := fakes{&MemoryHistory{}, &MemoryState{}, &MemorySnapshots{}} + return &Ledger{Module: "m", Name: "records", Types: []catalogue.LedgerType{records}, History: f.history, State: f.state, + Snapshots: f.snapshots}, f +} + +// applied applies an operation and fails the test on a refusal. +func applied(t *testing.T, l *Ledger, instance, operation, body string, author buscalls.Author) buscalls.Applied { + t.Helper() + got, err := l.Apply(buscalls.ApplyRequest{Ledger: l.Name, Instance: instance, Operation: operation, Body: json.RawMessage(body)}, author) + if err != nil { + t.Fatalf("%s %s: %v", operation, body, err) + } + return got +} + +// record is the instance's record a store holds, read as a Record. +func record(t *testing.T, raw []byte) Record { + t.Helper() + var r Record + if raw != nil { + if err := json.Unmarshal(raw, &r); err != nil { + t.Fatal(err) + } + } + return r +}