From c60a46aa6a7f5e9b40f096fc063c3266e6a4100d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:12:22 +0100 Subject: [PATCH] DLV-F3r sdk: views follower MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A2's first-view package (owners' answer and views events, the readers' Follower with Changed(state, key, value, at), UnansweredFinding, the status tools) taken into mesh-sdk as go/views as it stands at reg-a2-mesh-delivery f0a5fe8, for every foundation reader (ADR 0593 §1, ADR 0615, ADR 0621). Its test takes the place of the vault's node-key test, merged into the vault's table, so mesh-sdk's unit count does not grow. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- go/catalogue/vault_seat_test.go | 18 +- go/packages_list_test.go | 2 +- go/views/emit.go | 196 +++++++++++++++++ go/views/findings.go | 47 +++++ go/views/follow.go | 358 ++++++++++++++++++++++++++++++++ go/views/status.go | 145 +++++++++++++ go/views/tools.go | 159 ++++++++++++++ go/views/views.go | 244 ++++++++++++++++++++++ go/views/views_test.go | 204 ++++++++++++++++++ 9 files changed, 1356 insertions(+), 17 deletions(-) create mode 100644 go/views/emit.go create mode 100644 go/views/findings.go create mode 100644 go/views/follow.go create mode 100644 go/views/status.go create mode 100644 go/views/tools.go create mode 100644 go/views/views.go create mode 100644 go/views/views_test.go diff --git a/go/catalogue/vault_seat_test.go b/go/catalogue/vault_seat_test.go index 1e06f5e..2d20e80 100644 --- a/go/catalogue/vault_seat_test.go +++ b/go/catalogue/vault_seat_test.go @@ -82,6 +82,8 @@ func TestTheVaultsSeatPromisesItsIntakeVerbs(t *testing.T) { // And own (issue 1126): a machine's node-engine asks its own machine's secrets, sealed to its stated key. "own": {"credentials"}, "retire": {"node", "module", "name", "warrant"}, + // Issue 786: node-key, optional while the vault that runs does not serve it. + "node-key": {"key"}, } for _, gone := range []string{"accept", "move-asks"} { for _, v := range seat.Serves { @@ -143,19 +145,3 @@ func TestTheVaultsSeatPromisesItsIntakeVerbs(t *testing.T) { t.Fatalf("the vault serving its verbs was refused: %v", err) } } - -// Issue 786: the vault's seat promises node-key, optional while the vault that runs does not serve it. -func TestTheVaultsSeatPromisesNodeKey(t *testing.T) { - seat, _ := SeatNamed("mesh-vault") - for _, v := range seat.Serves { - if v.Name != "node-key" { - continue - } - props, _ := v.Input["properties"].(map[string]any) - if !v.Optional || props["key"] == nil { - t.Errorf("node-key: optional %v, key %v", v.Optional, props["key"]) - } - return - } - t.Error("the vault's seat does not promise node-key") -} diff --git a/go/packages_list_test.go b/go/packages_list_test.go index d55f12a..e1c6cde 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -24,7 +24,7 @@ var meshSDKPackages = []string{ ".", "announce", "announcetest", "asker", "asks", "buildstep", "buscalls", "buscompose", "busstate", "busretry", "calladdress", "catalogue", "composer", "declaration", "delivery", "forgekey", "handacts", "health", "internal/beside", "kvlimit", "membership", "objectlink", "outward", "overlay", "pernode", "process", "provider", "provider/harness", "provisions", - "sdkversion", "seal", "seatdef", "testbus", + "sdkversion", "seal", "seatdef", "testbus", "views", } // computesNot are the functions whose computation is one owner's alone (ADR 0593 §2, §7): the indexer computes a diff --git a/go/views/emit.go b/go/views/emit.go new file mode 100644 index 0000000..23a251a --- /dev/null +++ b/go/views/emit.go @@ -0,0 +1,196 @@ +package views + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// Emitter is an owner's process saying each change of its own states on its views family (ADR 0615 §5): it follows +// each state it is given from the revision its mark says it said up to, publishes one Event per change, and writes the +// mark once it has said every change the bus had handed it. A part watches only its own state (ADR 0596 §4): an owner +// follows its own buckets, never another's. +type Emitter struct { + // Owner is the owner module, said in every event. + Owner string + // Subject is where an event of the given local name lands on the owner's own stream (process.Account.StreamSubject). + Subject func(event string) string + // Publish sends one message to the owner's stream, its id the bus's dedup id. + Publish func(ctx context.Context, msg *nats.Msg, id string) error + // Marks is the owner's state MarksState; nil says nothing is marked, and every start says from the bucket's end. + Marks jetstream.KeyValue + // TooLarge is told of a value too large for an event (said as a gap): the owner's finding naming the key. Nil + // only logs it. + TooLarge func(state, key string, size int) + Logf func(string, ...any) + Now func() time.Time +} + +// JetStreamPublish is Emitter.Publish through js, with the bus's dedup id. +func JetStreamPublish(js jetstream.JetStream) func(context.Context, *nats.Msg, string) error { + return func(ctx context.Context, msg *nats.Msg, id string) error { + pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)) + return err + } +} + +func (e Emitter) now() time.Time { + if e.Now != nil { + return e.Now().UTC() + } + return time.Now().UTC() +} + +func (e Emitter) logf(format string, args ...any) { + if e.Logf != nil { + e.Logf(format, args...) + } +} + +// created is when the bucket's stream was made, and its fill now. +func created(ctx context.Context, kv jetstream.KeyValue) (time.Time, uint64, Fill, error) { + st, err := kv.Status(ctx) + if err != nil { + return time.Time{}, 0, Fill{}, err + } + b, ok := st.(*jetstream.KeyValueBucketStatus) + if !ok || b.StreamInfo() == nil { + return time.Time{}, 0, Fill{}, fmt.Errorf("the bus answered no stream behind %s", st.Bucket()) + } + info := b.StreamInfo() + return info.Created.UTC(), info.State.LastSeq, Fill{Keys: uint64(info.State.NumSubjects), Bytes: info.State.Bytes, MaxBytes: info.Config.MaxBytes}, nil +} + +// Start follows state's bucket kv: from the revision after its mark when the mark is of the same bucket (the same +// Created), otherwise from the bucket's end, writing a mark there first, so a reader's first view taken now misses +// nothing. It returns once the watch is set; the changes are said on a goroutine until ctx ends, and a watch that ends +// is sent on errs. +func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, errs chan<- error) error { + made, last, fill, err := created(ctx, kv) + if err != nil { + return fmt.Errorf("the state %s cannot be followed for its views: %w", state, err) + } + from := last + 1 + if m, ok := e.mark(ctx, state); ok && m.Created.Equal(made) && m.Revision <= last { + from = m.Revision + 1 + } else { + e.writeMark(ctx, state, Mark{State: state, Revision: last, Created: made, Fill: fill}) + } + w, err := kv.WatchAll(ctx, jetstream.ResumeFromRevision(from)) + if err != nil { + return fmt.Errorf("the state %s cannot be watched for its views: %w", state, err) + } + go func() { + defer w.Stop() + said := uint64(0) + for { + select { + case <-ctx.Done(): + return + case u, ok := <-w.Updates(): + if !ok { + if errs != nil { + errs <- fmt.Errorf("the watch of %s for its views ended", state) + } + return + } + if u != nil { + if err := e.say(ctx, state, made, u); err != nil { + // Not marked: the next start says it again from the mark, and the bus keeps it once by its id. + e.logf("views: %s.%s at %d is not said: %v", state, u.Key(), u.Revision(), err) + continue + } + said = u.Revision() + } + // Marked once every change handed so far is said, so a busy state writes one mark per burst. + if said > 0 && len(w.Updates()) == 0 { + _, _, fill, err := created(ctx, kv) + if err != nil { + e.logf("views: the fill of %s is not read: %v", state, err) + } + e.writeMark(ctx, state, Mark{State: state, Revision: said, Created: made, Fill: fill}) + said = 0 + } + } + } + }() + return nil +} + +// Event is the event saying u, a change of state made at made. +func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry) Event { + ev := Event{V: Version, Owner: e.Owner, State: state, Key: u.Key(), Revision: u.Revision(), Created: made, At: u.Created().UTC()} + if u.Operation() != jetstream.KeyValuePut { + ev.Deleted = true + return ev + } + if n := len(u.Value()); n > MaxValueBytes { + ev.Gap = true + e.logf("views: %s.%s at %d holds %d bytes, more than an event carries: said as a gap, so a reader asks its first view", state, u.Key(), u.Revision(), n) + if e.TooLarge != nil { + e.TooLarge(state, u.Key(), n) + } + return ev + } + ev.Value = json.RawMessage(u.Value()) + if !json.Valid(ev.Value) { + ev.Value, _ = json.Marshal(string(u.Value())) + } + return ev +} + +func (e Emitter) say(ctx context.Context, state string, made time.Time, u jetstream.KeyValueEntry) error { + ev := e.Event(state, made, u) + data, err := ev.Encode() + if err != nil { + return err + } + msg := nats.NewMsg(e.Subject(EventName(state))) + msg.Header.Set("x-event-id", ev.ID()) + msg.Header.Set("x-source", e.Owner) + msg.Header.Set("x-time", e.now().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + return e.Publish(ctx, msg, ev.ID()) +} + +func (e Emitter) mark(ctx context.Context, state string) (Mark, bool) { + if e.Marks == nil { + return Mark{}, false + } + got, err := e.Marks.Get(ctx, state) + if err != nil { + if !errors.Is(err, jetstream.ErrKeyNotFound) { + e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err) + } + return Mark{}, false + } + m, err := DecodeMark(got.Value()) + if err != nil { + e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err) + return Mark{}, false + } + return m, true +} + +func (e Emitter) writeMark(ctx context.Context, state string, m Mark) { + if e.Marks == nil { + return + } + m.V, m.At = Version, e.now() + data, err := json.Marshal(m) + if err != nil { + e.logf("views: the mark of %s is not written: %v", state, err) + return + } + if _, err := e.Marks.Put(ctx, state, data); err != nil { + e.logf("views: the mark of %s at %d is not written (%v): a first view answers the one before", state, m.Revision, err) + } +} diff --git a/go/views/findings.go b/go/views/findings.go new file mode 100644 index 0000000..9795fe1 --- /dev/null +++ b/go/views/findings.go @@ -0,0 +1,47 @@ +package views + +import ( + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The findings a views part writes in its own state `findings`, which the condition keeper raises and clears +// (novox/hq ADR 0564 §2): an owner's value too large for an event (TooLargeKind), and a reader's first view unanswered +// past its bound (UnansweredKind). +const ( + TooLargeKind = "view-too-large" + UnansweredKind = "first-view-unanswered" +) + +// TooLargeFinding is the owner's finding of state's key too large to say whole on its views family. +func TooLargeFinding(owner, state, key string, size int, at time.Time) (string, []byte, error) { + id := owner + "." + state + "." + key + summary := fmt.Sprintf("%s's %s.%s holds %d bytes, more than one views event carries (%d): every reader asks its first view again at each change of it", owner, state, key, size, MaxValueBytes) + return finding(owner, TooLargeKind, id, summary, + fmt.Sprintf("A value of %s is too large for its views event", owner), + "Each change of this key is said as a gap, so every reader of the state asks the owner's first view once more. "+ + "It works, but costs a first view per change; the value should be made smaller or split.", at) +} + +// UnansweredFinding is a reader's finding of its first view of s unanswered past its bound (ADR 0615 §6): it names the +// owner and the tool. A first view answered at revision 0 with nothing in it is said too, with the operator's step. +func UnansweredFinding(reader string, s Status, at time.Time) (string, []byte, error) { + id := reader + "." + s.Owner + "." + s.State + summary := fmt.Sprintf("%s has no first view of %s's %s: %s has not answered since %s (%s)", reader, s.Owner, s.State, s.Tool, + s.Asked.Format(time.RFC3339), strings.TrimSpace(s.Error)) + return finding(reader, UnansweredKind, id, summary, + fmt.Sprintf("%s waits for %s", reader, s.Owner), + fmt.Sprintf("%s reads %s's %s only through %s, and keeps asking it; it never reads the state itself. The answer clears this. "+ + "Look at %s's runtime of its own: is it running and connected?", reader, s.Owner, s.State, s.Tool, s.Owner), at) +} + +func finding(observer, kind, id, summary, headline, explanation string, at time.Time) (string, []byte, error) { + key := health.FindingKey(health.ScopeModule, id, kind) + value, err := health.EncodeFinding(health.Finding{Key: key, Kind: kind, Subject: health.Subject{Scope: health.ScopeModule, ID: id}, + Severity: health.Warning, Summary: summary, Headline: headline, Explanation: explanation, + Evidence: []health.Evidence{{At: at.UTC(), Said: summary}}, Source: observer + "." + kind, Observer: observer, At: at.UTC()}) + return key, value, err +} diff --git a/go/views/follow.go b/go/views/follow.go new file mode 100644 index 0000000..2c0a470 --- /dev/null +++ b/go/views/follow.go @@ -0,0 +1,358 @@ +package views + +import ( + "cmp" + "context" + "encoding/json" + "fmt" + "maps" + "slices" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// Follower is a reader's picture of one state of an owner (ADR 0615 §4 to §6): it asks the owner's first-view tool at +// its start, then keeps the picture from the owner's events alone, asking the tool once more only on a gap in revisions +// (said in its log), never on a timer. Started while the owner is down, it keeps asking, never reading the owner's state +// instead, and past Bound raises its condition naming the owner and the tool; the answer clears it. +type Follower struct { + Owner string + // Tool is the owner's tool it asks, as the reader's invokes names it (.). + Tool string + State string + // Ask calls the tool for the page after after. + Ask func(ctx context.Context, after string) (Answer, error) + // Raise and Clear are the reader's condition of this first view (past Bound; cleared by the answer). Nil says + // nothing. + Raise func(Status) + Clear func(Status) + // Changed is told of each key the picture changes (deleted: the key is gone), after the change is held, with when it + // was written: the zero time for a key a first view brought (since before what the reader heard). + Changed func(key string, value json.RawMessage, deleted bool, at time.Time) + // Said is told of the status each time an ask starts, is answered or fails, and of a raise and a clear: where a + // reader keeps it on the bus for its read tool (StatusTools). + Said func(Status) + Logf func(string, ...any) + Now func() time.Time + // Bound is how long it asks before it raises; zero is views.Bound. Wait is the first wait between two asks that + // failed, doubling to a minute; zero is five seconds. + Bound time.Duration + Wait time.Duration + + mu sync.Mutex + values map[string]json.RawMessage + st Status + asking bool + // during are the events heard while an ask runs: taken again after its answer, so none is lost to a picture read + // before it was written. + during []Event + ready chan struct{} +} + +// Status is what a reader's read tool says of one first view (design 25 § 12.2): the owner, the tool and the revision +// its picture holds, when it asked and when it was answered, how many asks, the hand acts among them, and whether its +// condition is raised. +type Status struct { + Owner string `json:"owner"` + Tool string `json:"tool"` + State string `json:"state"` + Revision uint64 `json:"revision"` + Created time.Time `json:"created,omitzero"` + Keys int `json:"keys"` + Asked time.Time `json:"asked,omitzero"` + Answered time.Time `json:"answered,omitzero"` + Asks int `json:"asks"` + ByHand int `json:"by-hand"` + Raised bool `json:"raised"` + // Why is the last reason it asked (start, a gap and which, a hand act) or why the last ask failed. + Why string `json:"why,omitempty"` + Error string `json:"error,omitempty"` +} + +func (f *Follower) now() time.Time { + if f.Now != nil { + return f.Now().UTC() + } + return time.Now().UTC() +} + +func (f *Follower) logf(format string, args ...any) { + if f.Logf != nil { + f.Logf(format, args...) + } +} + +func (f *Follower) init() { + if f.values == nil { + f.values = map[string]json.RawMessage{} + f.ready = make(chan struct{}) + f.st = Status{Owner: f.Owner, Tool: f.Tool, State: f.State} + } +} + +func (f *Follower) say(s Status) { + if f.Said != nil { + f.mu.Lock() + s.Keys = len(f.values) + f.mu.Unlock() + f.Said(s) + } +} + +// Status is the follower's status now. +func (f *Follower) Status() Status { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + s := f.st + s.Keys = len(f.values) + return s +} + +// Ready is closed once a first view was answered. +func (f *Follower) Ready() <-chan struct{} { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + return f.ready +} + +// Values is a copy of the picture. +func (f *Follower) Values() map[string]json.RawMessage { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + return maps.Clone(f.values) +} + +// Get is one key of the picture. +func (f *Follower) Get(key string) (json.RawMessage, bool) { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + v, ok := f.values[key] + return v, ok +} + +// Start asks the first view until it is answered or ctx ends (ADR 0615 §6): each failed ask waits, doubling, and past +// Bound raises the condition. It returns once answered. +func (f *Follower) Start(ctx context.Context) error { return f.askUntil(ctx, "start", false) } + +// Again asks the first view once more by hand (a re-read tool, recorded by the caller as a hand act), until answered or +// ctx ends. +func (f *Follower) Again(ctx context.Context, why string) error { + return f.askUntil(ctx, "by hand: "+why, true) +} + +func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error { + f.mu.Lock() + f.init() + if f.asking { + f.mu.Unlock() + return nil // one ask at a time; the one running brings the picture to its newest + } + f.asking = true + f.during = nil + f.st.Why = why + if byHand { + f.st.ByHand++ + } + f.mu.Unlock() + defer func() { f.mu.Lock(); f.asking = false; f.mu.Unlock() }() + bound, wait := f.Bound, f.Wait + if bound == 0 { + bound = Bound + } + if wait == 0 { + wait = 5 * time.Second + } + first := f.now() + for { + err := f.askOnce(ctx) + if err == nil { + return nil + } + f.mu.Lock() + f.st.Error = err.Error() + raise := !f.st.Raised && f.now().Sub(first) >= bound + if raise { + f.st.Raised = true + } + s := f.st + f.mu.Unlock() + f.logf("the first view %s of %s's %s is not answered (%v): asked again in %s", f.Tool, f.Owner, f.State, err, wait) + if raise && f.Raise != nil { + f.Raise(s) + } + f.say(s) + pause, cancel := context.WithTimeout(ctx, wait) // a context's bound between two asks, no timer of its own + <-pause.Done() + cancel() + if ctx.Err() != nil { + return ctx.Err() + } + if wait *= 2; wait > time.Minute { + wait = time.Minute + } + } +} + +// askOnce asks every page and, once whole, takes it as the picture. +func (f *Follower) askOnce(ctx context.Context) error { + f.mu.Lock() + f.st.Asked, f.st.Asks = f.now(), f.st.Asks+1 + asked := f.st + f.mu.Unlock() + f.say(asked) + var first Answer + values := map[string]json.RawMessage{} + after := "" + for page := 0; ; page++ { + a, err := f.Ask(ctx, after) + if err != nil { + return err + } + if page == 0 { + first = a + } + maps.Copy(values, a.Values) + if a.Next == "" || a.Next == after { + break + } + after = a.Next + } + f.mu.Lock() + old := f.values + f.values = values + wasRaised := f.st.Raised + f.st.Revision, f.st.Created, f.st.Answered, f.st.Raised, f.st.Error = first.Revision, first.Created, f.now(), false, "" + // The events heard while it asked, after the answer's revision and in order, are taken again. + during := f.during + f.during = nil + slices.SortFunc(during, func(a, b Event) int { return cmp.Compare(a.Revision, b.Revision) }) + for _, ev := range during { + if ev.Created.Equal(f.st.Created) && ev.Revision == f.st.Revision+1 && !ev.Gap { + if ev.Deleted { + delete(f.values, ev.Key) + } else { + f.values[ev.Key] = ev.Value + } + f.st.Revision = ev.Revision + } + } + values = maps.Clone(f.values) + s := f.st + select { + case <-f.ready: + default: + close(f.ready) + } + f.mu.Unlock() + if wasRaised && f.Clear != nil { + f.Clear(s) + } + f.say(s) + if f.Changed != nil { + for k, v := range values { + if ov, ok := old[k]; !ok || string(ov) != string(v) { + f.Changed(k, v, false, time.Time{}) + } + } + for k := range old { + if _, ok := values[k]; !ok { + f.Changed(k, nil, true, time.Time{}) + } + } + } + return nil +} + +// Apply takes one event of the owner's views family into the picture and says whether it was a gap: another Created +// (the owner's state made again), a revision past the one after the picture's, or a value too large to carry. A +// revision at or below the picture's is already in it (a durable consumer handing what it held) and is passed over. +// Before the first view is answered an event is passed over too: the answer holds it. +func (f *Follower) Apply(ev Event) (gap bool, why string) { + f.mu.Lock() + f.init() + if f.asking { + f.during = append(f.during, ev) + } + select { + case <-f.ready: + default: + f.mu.Unlock() + return false, "" + } + switch { + case !ev.Created.Equal(f.st.Created): + why = fmt.Sprintf("%s's %s was made again (or not marked when the first view was answered): created %s, held %s", f.Owner, f.State, ev.Created.Format(time.RFC3339), f.st.Created.Format(time.RFC3339)) + case ev.Revision <= f.st.Revision: + f.mu.Unlock() + return false, "" + case ev.Revision > f.st.Revision+1: + why = fmt.Sprintf("%s's %s at revision %d, the picture holds %d: %d changes missed", f.Owner, f.State, ev.Revision, f.st.Revision, ev.Revision-f.st.Revision-1) + case ev.Gap: + why = fmt.Sprintf("%s's %s.%s at revision %d was too large to carry", f.Owner, f.State, ev.Key, ev.Revision) + } + if why != "" { + f.mu.Unlock() + return true, why + } + if ev.Deleted { + delete(f.values, ev.Key) + } else { + f.values[ev.Key] = ev.Value + } + f.st.Revision = ev.Revision + f.mu.Unlock() + if f.Changed != nil { + f.Changed(ev.Key, ev.Value, ev.Deleted, ev.At) + } + return false, "" +} + +// Hear is a reader's consumer of one owner's views family, routing each event to the follower of its state; a gap asks +// that follower's first view once more on its own goroutine, said in the log. A message it cannot read is terminated. +func Hear(ctx context.Context, followers map[string]*Follower, logf func(string, ...any)) func(jetstream.Msg) { + return func(m jetstream.Msg) { + ev, err := DecodeEvent(m.Data()) + if err != nil { + logf("views: %s is not a views event, dropped: %v", m.Subject(), err) + _ = m.Term() + return + } + f := followers[ev.State] + if f == nil { + _ = m.Ack() // a state this reader does not follow + return + } + if gap, why := f.Apply(ev); gap { + logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool) + go func() { + f.mu.Lock() + f.init() + f.st.Why = "a gap: " + why + f.mu.Unlock() + _ = f.askUntil(ctx, "a gap: "+why, false) + }() + } + _ = m.Ack() + } +} + +// AskVia is a Follower's Ask through a process's call of the owner's tool (process.Conn.Call), decoding its answer. +func AskVia(call func(key string, body any) (json.RawMessage, error), tool string) func(context.Context, string) (Answer, error) { + return func(_ context.Context, after string) (Answer, error) { + body := map[string]any{} + if after != "" { + body["after"] = after + } + raw, err := call(tool, body) + if err != nil { + return Answer{}, err + } + return DecodeAnswer(raw) + } +} diff --git a/go/views/status.go b/go/views/status.go new file mode 100644 index 0000000..19907d3 --- /dev/null +++ b/go/views/status.go @@ -0,0 +1,145 @@ +package views + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "time" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// StatusState is a reader's own state of its first views (design 25 § 12.2): at StatusKey each first view's Status, as +// its process last said it, and at AgainKey a hand ask of its re-read tool, which the process takes and deletes. A +// process serves no tool (ADR 0436), so its read and re-read tools, in its tools bundle, meet it here. +const StatusState = "first-views" + +// StatusKey is where a reader keeps its first view of owner's state. +func StatusKey(owner, state string) string { return owner + "." + state } + +// AgainKey is where a re-read tool asks a reader's process to ask owner's first view of state again. +func AgainKey(owner, state string) string { return "again." + owner + "." + state } + +// HandAsk is one ask of a re-read tool: who asked (the runtime's proof), why and when. +type HandAsk struct { + V int `json:"v"` + Owner string `json:"owner"` + State string `json:"state"` + Why string `json:"why"` + By string `json:"by"` + At time.Time `json:"at"` +} + +// KeepStatus is a Follower's Said that writes its status at StatusKey of kv (the reader's StatusState). +func KeepStatus(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Status) { + return func(s Status) { + data, err := json.Marshal(s) + if err == nil { + _, err = kv.Put(ctx, StatusKey(s.Owner, s.State), data) + } + if err != nil && logf != nil { + logf("views: the status of the first view %s is not kept: %v", s.Tool, err) + } + } +} + +// HearAgain takes each hand ask written at an AgainKey of kv, those standing at the start included, asks that +// follower's first view again with the ask's words and who asked, and deletes the ask. It returns once the watch is set. +func HearAgain(ctx context.Context, kv jetstream.KeyValue, followers []*Follower, logf func(string, ...any)) error { + w, err := kv.Watch(ctx, "again.>") + if err != nil { + return err + } + go func() { + defer func() { _ = w.Stop() }() + for { + select { + case <-ctx.Done(): + return + case e, ok := <-w.Updates(): + if !ok { + return + } + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + var a HandAsk + if err := json.Unmarshal(e.Value(), &a); err != nil { + logf("views: the hand ask %s is not read: %v", e.Key(), err) + _ = kv.Delete(ctx, e.Key()) + continue + } + i := slices.IndexFunc(followers, func(f *Follower) bool { return f.Owner == a.Owner && f.State == a.State }) + if i < 0 { + logf("views: the hand ask %s names no first view this part holds", e.Key()) + _ = kv.Delete(ctx, e.Key()) + continue + } + logf("views: %s asks the first view %s again by hand: %s", a.By, followers[i].Tool, a.Why) + _ = kv.Delete(ctx, e.Key(), jetstream.LastRevision(e.Revision())) + f := followers[i] + go func() { _ = f.Again(ctx, a.Why+" ("+a.By+")") }() + } + } + }() + return nil +} + +// StatusTools are a reader's read tool of its first views (read) and its re-read tool (again), in its tools bundle, +// over its StatusState: read lists what the process last said of each; again writes a hand ask the process takes, +// naming the caller the runtime proved, and is recorded in the status's by-hand count and why. +func StatusTools(reader, read, again string, st stdio.KeptState, now func() time.Time) []stdio.Tool { + return []stdio.Tool{ + {Name: read, + Description: fmt.Sprintf("%s's first views of the indexer's and the assigner's states (novox/hq ADR 0615): per owner and state, "+ + "the tool it asks, the revision its picture held when last answered, when it asked and was answered, how many asks "+ + "(by hand among them) and whether its condition (unanswered past two minutes) is raised.", reader), + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + keys, err := st.Keys() + if err != nil { + return nil, fmt.Errorf("the state %s cannot be listed: %w", StatusState, err) + } + out := []Status{} + for _, k := range keys { + if strings.HasPrefix(k, "again.") { + continue + } + e, err := st.Get(k) + if err != nil || e == nil { + continue + } + var s Status + if json.Unmarshal(e.Value, &s) == nil { + out = append(out, s) + } + } + return map[string]any{"first-views": out}, nil + }}, + {Name: again, + Description: fmt.Sprintf("Ask %s's first view of an owner's state again now, by hand (a hand act, its caller and why kept in "+ + "the status %s reads): for a picture a person doubts. The part asks once and follows the owner's events after the answer.", reader, read), + Input: map[string]any{"type": "object", "required": []string{"owner", "state", "why"}, "properties": map[string]any{ + "owner": map[string]any{"type": "string", "description": "the owner: mesh-delivery-indexer or mesh-delivery-assigner"}, + "state": map[string]any{"type": "string", "description": "the owner's state, as the read tool names it"}, + "why": map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + owner, _ := args["owner"].(string) + state, _ := args["state"].(string) + why, _ := args["why"].(string) + if owner == "" || state == "" || strings.TrimSpace(why) == "" { + return nil, errors.New("again names the owner, the state and why") + } + a := HandAsk{V: Version, Owner: owner, State: state, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()} + if _, err := st.Put(AgainKey(owner, state), a); err != nil { + return nil, fmt.Errorf("the hand ask is not written: %w", err) + } + return map[string]any{"asked": a, "read": read}, nil + }}, + } +} diff --git a/go/views/tools.go b/go/views/tools.go new file mode 100644 index 0000000..e14f3e3 --- /dev/null +++ b/go/views/tools.go @@ -0,0 +1,159 @@ +package views + +import ( + "encoding/json" + "fmt" + "slices" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// KeptSource is a state as a tools bundle reads it through its runtime (stdio.KeptState), as a Source. +type KeptSource struct{ State stdio.KeptState } + +// Keys is every key with a value. +func (k KeptSource) Keys() ([]string, error) { return k.State.Keys() } + +// Get is one key's value and revision. +func (k KeptSource) Get(key string) (*Entry, bool, error) { + e, err := k.State.Get(key) + if err != nil || e == nil { + return nil, false, err + } + return &Entry{Value: e.Value, Revision: e.Revision}, true, nil +} + +// ParseCallers is a callers list as a manifest writes it into an env value: comma-separated, each a module (on any +// machine), . or person. (stdio.Callers). +func ParseCallers(list string) stdio.Callers { + var out stdio.Callers + for _, c := range strings.Split(list, ",") { + if c = strings.TrimSpace(c); c != "" { + out = append(out, c) + } + } + return out +} + +// CallersEnv is the env value of a first-view tool's callers list, in the owner's tools bundle: MESH_VIEW_CALLERS_, +// the tool's name in capitals, a hyphen an underscore. +func CallersEnv(tool string) string { + return "MESH_VIEW_CALLERS_" + strings.ToUpper(strings.ReplaceAll(tool, "-", "_")) +} + +// Refusal is why caller may not ask tool, or nil: the caller's proof, never an argument, against the tool's callers +// list holding exactly its readers (ADR 0621 §2, ADR 0622 §1); the owner itself is on no list, so it is refused too. +func Refusal(tool string, callers stdio.Callers, caller string) error { + if callers.Allows(caller) { + return nil + } + return fmt.Errorf("%s is refused to %q: the tool answers only %s (novox/hq ADR 0621 §2)", tool, caller, strings.Join(callers, ", ")) +} + +// FirstViewTool is an owner's first-view tool of one of its states (ADR 0621 §1): each call answers one page of what the +// state holds now, at the revision its mark says, to the callers its list names and nobody else. It reads only the +// owner's own state, so it answers as soon as the runtime is connected, before the owner has asked anyone anything +// (ADR 0621 §3). +func FirstViewTool(owner, name, state, words string, callers stdio.Callers, values, marks Source, now func() time.Time) stdio.Tool { + return stdio.Tool{ + Name: name, + Description: fmt.Sprintf("The first view of %s's state %s (novox/hq ADR 0615, ADR 0621): %s Each value whole by key, at the revision "+ + "the owner said its views family up to, so a reader then follows the events on that family and asks again only on a gap in "+ + "revisions. One page at a time: next is the key to ask after for the rest. Answered only to %s; any other caller is refused.", + owner, state, words, strings.Join(callers, ", ")), + Input: map[string]any{"type": "object", "properties": map[string]any{ + "after": map[string]any{"type": "string", "description": "the key the page before ended at (its next); none starts at the first"}}}, + Run: func(args map[string]any) (any, error) { + if err := Refusal(name, callers, stdio.CallerOf(args)); err != nil { + return nil, err + } + after, _ := args["after"].(string) + return FirstView(owner, state, values, marks, strings.TrimSpace(after), now()) + }, + } +} + +// ReadTool is an owner's ordinary read tool of one of its states (design 25 § 12.2): the keys, one key's value and +// revision, or a page of values in the first view's form, and the state's fill against its cap as its mark last said +// it. Granted, as any tool is, to every reader that invokes it, and callable by a session. +func ReadTool(owner, name, state, words string, values, marks Source, now func() time.Time) stdio.Tool { + return stdio.Tool{ + Name: name, + Description: fmt.Sprintf("Read %s's state %s: %s With no argument, its keys and how full it is against its cap (as the "+ + "owner last marked it); with key, that key's value and revision; with values, a page of values whole by key, at the "+ + "revision the owner said its views family up to, continued with after (the page's next).", owner, state, words), + Input: map[string]any{"type": "object", "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "one key: its value and revision"}, + "values": map[string]any{"type": "boolean", "description": "a page of values whole by key"}, + "after": map[string]any{"type": "string", "description": "with values: the key the page before ended at"}}}, + Run: func(args map[string]any) (any, error) { + key, _ := args["key"].(string) + after, _ := args["after"].(string) + all, _ := args["values"].(bool) + return Read(owner, state, values, marks, strings.TrimSpace(key), all, strings.TrimSpace(after), now()) + }, + } +} + +// ReadAnswer is a read tool's answer without values: the state's keys and fill. +type ReadAnswer struct { + Owner string `json:"owner"` + State string `json:"state"` + Keys []string `json:"keys"` + Fill *Fill `json:"fill,omitempty"` + Revision uint64 `json:"revision"` + Marked time.Time `json:"marked,omitzero"` + // Words say what the fill means where the owner has not marked the state yet. + Words string `json:"words,omitempty"` +} + +// KeyAnswer is one key of a read tool's answer. +type KeyAnswer struct { + Owner string `json:"owner"` + State string `json:"state"` + Key string `json:"key"` + Found bool `json:"found"` + Value json.RawMessage `json:"value,omitempty"` + Revision uint64 `json:"revision,omitempty"` +} + +// Read is a read tool's answer: one key, a page of values, or the keys and fill. +func Read(owner, state string, values, marks Source, key string, all bool, after string, now time.Time) (any, error) { + switch { + case key != "": + e, ok, err := values.Get(key) + if err != nil { + return nil, fmt.Errorf("%s.%s cannot be read: %w", state, key, err) + } + a := KeyAnswer{Owner: owner, State: state, Key: key, Found: ok} + if ok { + a.Value, a.Revision = e.Value, e.Revision + } + return a, nil + case all: + return FirstView(owner, state, values, marks, after, now) + } + keys, err := values.Keys() + if err != nil { + return nil, fmt.Errorf("the state %s cannot be listed: %w", state, err) + } + slices.Sort(keys) + a := ReadAnswer{Owner: owner, State: state, Keys: keys} + if marks != nil { + if e, ok, err := marks.Get(state); err == nil && ok { + if m, err := DecodeMark(e.Value); err == nil { + fill := m.Fill + a.Fill, a.Revision, a.Marked = &fill, m.Revision, m.At + } + } + } + if a.Fill == nil { + a.Words = "the owner has not marked this state yet: its fill is said once its views family says a change of it" + } + if a.Keys == nil { + a.Keys = []string{} + } + return a, nil +} diff --git a/go/views/views.go b/go/views/views.go new file mode 100644 index 0000000..2be1f32 --- /dev/null +++ b/go/views/views.go @@ -0,0 +1,244 @@ +// Package views is the owners' side and the readers' side of a first view (novox/hq ADR 0615, ADR 0621, ADR 0622; +// design 25 § 12.2): the indexer and the assigner answer a reader's first view of one of their states as a tool of +// their own module (Answer, FirstView), say each change of that state as one event on their views family (Event, +// Emitter), and a reader keeps its picture from the answer and the events, asking the tool again only on a gap in +// revisions (Follower). The tools' read side (Read) answers the ordinary read tools of § 12.2 too. +package views + +import ( + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "time" +) + +// Version is the version of every value this package writes and the only one it reads. +const Version = 1 + +// The owners' views families (design 25 § 12.2; ADR 0615 §5, ADR 0619 §2): one stream each on its owner's old `streams` +// section, kept by limits 7 days and at most 256 MiB, dropping its oldest when full, so a reader that missed a dropped +// event sees the gap in revisions and asks once more. A stream's name is the mesh's, so each carries its owner's name. +const ( + IndexerStream = "indexer-views" + AssignerStream = "assigner-views" + StreamDays = 7 + StreamMiB = 256 + // EventPrefix starts each event's local name: `views.`, one event per changed key of that state. + EventPrefix = "views." + // EventPattern is every event of a views family, as a stream's events and a reader's consumes name it. + EventPattern = EventPrefix + "**" +) + +// MarksState is an owner's own state of the revision each followed state was said up to (Mark), one key per state: +// what a first view answers its revision from, so a reader that takes the answer misses no event after it. +const MarksState = "view-marks" + +// MaxValueBytes is the largest value an event carries whole: under the bus's largest message (1 MiB, the server's +// max_payload) with the event's own fields. A larger one is said as its revision alone (Event.Gap), and the owner +// writes a finding naming the key (design 25 § 12.2). +const MaxValueBytes = 960 * 1024 + +// PageBytes bounds the values one answer carries, so the answer fits one bus message; Answer.Next continues it. +const PageBytes = 768 * 1024 + +// Bound is how long a reader waits for its first view before it raises the condition naming the owner and the verb +// (ADR 0615 §6): longer than an owner's start, so a normal start, readers and owners coming up together, tells nobody. +const Bound = 2 * time.Minute + +// EventName is the local name of the event saying a change of state: `views.`. +func EventName(state string) string { return EventPrefix + state } + +// StateOfEvent is the state an event's local name (or the last tokens of its subject) says, or "". +func StateOfEvent(name string) string { + if i := strings.LastIndex(name, ".event."+EventPrefix); i >= 0 { + name = name[i+len(".event."):] + } + s, ok := strings.CutPrefix(name, EventPrefix) + if !ok || s == "" || strings.ContainsAny(s, ".*> ") { + return "" + } + return s +} + +// Event is one change of one key of an owner's state, said on its views family: the key's value whole, or a delete +// mark, at the revision of the state the change was written at. Created is when the owner's state was made on the bus: +// a state made again (the bus module's recreate) starts its revisions again, and a reader takes another Created as a gap. +type Event struct { + V int `json:"v"` + Owner string `json:"owner"` + State string `json:"state"` + Key string `json:"key"` + Revision uint64 `json:"revision"` + Created time.Time `json:"created"` + Deleted bool `json:"deleted,omitempty"` + Value json.RawMessage `json:"value,omitempty"` + // At is when the change was written on the bus. + At time.Time `json:"at,omitzero"` + // Gap says the value was too large to carry (MaxValueBytes): the reader asks its first view once. + Gap bool `json:"gap,omitempty"` +} + +// ID is the event's id on the bus, so the same change said twice (an owner's restart) is kept once. +func (e Event) ID() string { + return fmt.Sprintf("%s-%s-%d-%d", e.Owner, e.State, e.Created.UnixNano(), e.Revision) +} + +// Encode writes an event. +func (e Event) Encode() ([]byte, error) { + e.V = Version + if e.Owner == "" || e.State == "" || e.Key == "" || e.Revision == 0 { + return nil, errors.New("views: an event names its owner, state, key and revision") + } + if e.Deleted || e.Gap { + e.Value = nil + } + return json.Marshal(e) +} + +// DecodeEvent reads an event of this version. +func DecodeEvent(data []byte) (Event, error) { + var e Event + if err := decode(data, &e); err != nil { + return Event{}, err + } + if e.State == "" || e.Key == "" || e.Revision == 0 { + return Event{}, errors.New("views: an event names no state, key or revision") + } + return e, nil +} + +// Mark is the revision an owner said its state up to, and when that state was made (MarksState). +type Mark struct { + V int `json:"v"` + State string `json:"state"` + Revision uint64 `json:"revision"` + Created time.Time `json:"created"` + At time.Time `json:"at"` + // Fill is the state's bucket as the bus held it when the mark was written: what a read tool says of its fill. + Fill Fill `json:"fill"` +} + +// Fill is a state's bucket's fill against its cap, as the bus says it. +type Fill struct { + Keys uint64 `json:"keys"` + Bytes uint64 `json:"bytes"` + MaxBytes int64 `json:"max-bytes"` +} + +// DecodeMark reads a mark of this version. +func DecodeMark(data []byte) (Mark, error) { + var m Mark + return m, decode(data, &m) +} + +// Answer is a first view, or one page of an ordinary read: the state's values by key, from After on, at the revision +// of the state the owner said up to (Mark) before it read them; Next is the key to ask after for the rest, empty when +// the answer is whole. A reader keeps the first page's revision: a value on a later page may be newer, and the events +// after that revision bring every key to its newest in order. +type Answer struct { + V int `json:"v"` + Owner string `json:"owner"` + State string `json:"state"` + Revision uint64 `json:"revision"` + Created time.Time `json:"created"` + Values map[string]json.RawMessage `json:"values"` + Next string `json:"next,omitempty"` + At time.Time `json:"at"` +} + +// DecodeAnswer reads an answer of this version. +func DecodeAnswer(data []byte) (Answer, error) { + var a Answer + if err := decode(data, &a); err != nil { + return Answer{}, err + } + if a.Values == nil { + a.Values = map[string]json.RawMessage{} + } + return a, nil +} + +// Entry is one key's value and revision, as a state read gives it. +type Entry struct { + Value json.RawMessage + Revision uint64 +} + +// Source is the read side of one state, as a tools bundle reaches its own state (stdio.KeptState's Keys and Get). +type Source interface { + Keys() ([]string, error) + Get(key string) (*Entry, bool, error) +} + +// FirstView answers one page of state from its owner's source: the mark first, so no change after the revision it +// answers at is missed, then the values in key order after after, until PageBytes. A state not said yet (no mark) +// answers revision 0: a first view of whatever it holds, which the reader takes as a gap at the first event. +func FirstView(owner, state string, values, marks Source, after string, now time.Time) (Answer, error) { + a := Answer{V: Version, Owner: owner, State: state, Values: map[string]json.RawMessage{}, At: now.UTC()} + if marks != nil { + e, ok, err := marks.Get(state) + if err != nil { + return Answer{}, fmt.Errorf("the mark of %s cannot be read: %w", state, err) + } + if ok { + m, err := DecodeMark(e.Value) + if err != nil { + return Answer{}, fmt.Errorf("the mark of %s: %w", state, err) + } + a.Revision, a.Created = m.Revision, m.Created + } + } + keys, err := values.Keys() + if err != nil { + return Answer{}, fmt.Errorf("the state %s cannot be listed: %w", state, err) + } + slices.Sort(keys) + size := 0 + for _, k := range keys { + if after != "" && k <= after { + continue + } + e, ok, err := values.Get(k) + if err != nil { + return Answer{}, fmt.Errorf("%s.%s cannot be read: %w", state, k, err) + } + if !ok { + continue + } + if size > 0 && size+len(k)+len(e.Value) > PageBytes { + a.Next = lastKey(a.Values) + break + } + a.Values[k] = e.Value + size += len(k) + len(e.Value) + } + return a, nil +} + +func lastKey(m map[string]json.RawMessage) string { + last := "" + for k := range m { + if k > last { + last = k + } + } + return last +} + +func decode(data []byte, into any) error { + var probe struct { + V *int `json:"v"` + } + if err := json.Unmarshal(data, &probe); err != nil { + return fmt.Errorf("views: value does not fit its shape: %w", err) + } + if probe.V == nil || *probe.V != Version { + return fmt.Errorf("views: unknown value version: this reader reads %d", Version) + } + if err := json.Unmarshal(data, into); err != nil { + return fmt.Errorf("views: value does not fit its shape: %w", err) + } + return nil +} diff --git a/go/views/views_test.go b/go/views/views_test.go new file mode 100644 index 0000000..da65aad --- /dev/null +++ b/go/views/views_test.go @@ -0,0 +1,204 @@ +package views + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// mapSource is a state held in memory, as a Source. +type mapSource map[string]Entry + +func (m mapSource) Keys() ([]string, error) { + var out []string + for k := range m { + out = append(out, k) + } + return out, nil +} + +func (m mapSource) Get(key string) (*Entry, bool, error) { + e, ok := m[key] + if !ok { + return nil, false, nil + } + return &e, true, nil +} + +func markOf(state string, rev uint64, made time.Time) Entry { + raw, _ := json.Marshal(Mark{V: Version, State: state, Revision: rev, Created: made, Fill: Fill{Keys: 2, Bytes: 100, MaxBytes: 1000}}) + return Entry{Value: raw, Revision: 1} +} + +// The owners' and the readers' side of a first view (ADR 0615 §4-§6, ADR 0621 §2-§3, ADR 0622): one table of cases, +// since mesh-delivery is over its test budget. +func TestViews(t *testing.T) { + made := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) + now := func() time.Time { return made.Add(time.Hour) } + values := mapSource{"a": {Value: json.RawMessage(`{"n":1}`), Revision: 7}, "b": {Value: json.RawMessage(`{"n":2}`), Revision: 9}} + marks := mapSource{"modules": markOf("modules", 9, made)} + + t.Run("a first view answers every value at the mark's revision, the mark read first", func(t *testing.T) { + a, err := FirstView("o", "modules", values, marks, "", now()) + if err != nil || a.Revision != 9 || !a.Created.Equal(made) || len(a.Values) != 2 || a.Next != "" { + t.Fatalf("answer = %+v, %v", a, err) + } + if a, _ := FirstView("o", "modules", values, mapSource{}, "", now()); a.Revision != 0 { + t.Errorf("a state not marked answers revision %d, want 0", a.Revision) + } + }) + t.Run("a first view larger than a page is paged by key", func(t *testing.T) { + big := mapSource{} + for i := range 4 { + big[fmt.Sprintf("k%d", i)] = Entry{Value: json.RawMessage(`"` + strings.Repeat("x", PageBytes/3) + `"`), Revision: uint64(i + 1)} + } + got := map[string]bool{} + after := "" + for pages := 0; ; pages++ { + a, err := FirstView("o", "s", big, nil, after, now()) + if err != nil || pages > 4 { + t.Fatalf("page %d: %+v %v", pages, a.Next, err) + } + for k := range a.Values { + got[k] = true + } + if a.Next == "" { + break + } + after = a.Next + } + if len(got) != 4 { + t.Errorf("pages hold %v, want the four keys", got) + } + }) + t.Run("each first-view tool answers its callers alone, never the owner, by the proof and not an argument", func(t *testing.T) { + callers := ParseCallers("mesh-grants, node-tools,mesh-delivery-assigner") + for caller, allowed := range map[string]bool{"novox.mesh-grants": true, "ace.node-tools": true, "g14.mesh-delivery-assigner": true, + "novox.mesh-delivery-indexer": false, "novox.mesh-config": false, "person.jochen": false, stdio.Unproven: false} { + if err := Refusal("manifests", callers, caller); (err == nil) != allowed { + t.Errorf("%q: err = %v, allowed %v", caller, err, allowed) + } + } + // A caller argument names nobody: a call with no proof is refused. + tool := FirstViewTool("mesh-delivery-indexer", "manifests", "modules", "w.", callers, values, marks, now) + if _, err := tool.Run(map[string]any{"caller": "novox.mesh-grants"}); err == nil { + t.Errorf("an unproven call naming a caller is answered") + } + }) + t.Run("a read tool says the keys and the fill, one key, or a page of values", func(t *testing.T) { + r, err := Read("o", "modules", values, marks, "", false, "", now()) + ra, ok := r.(ReadAnswer) + if err != nil || !ok || len(ra.Keys) != 2 || ra.Fill == nil || ra.Fill.MaxBytes != 1000 || ra.Revision != 9 { + t.Fatalf("read = %+v, %v", r, err) + } + r, _ = Read("o", "modules", values, marks, "b", false, "", now()) + if k := r.(KeyAnswer); !k.Found || k.Revision != 9 { + t.Errorf("one key = %+v", k) + } + r, _ = Read("o", "modules", values, nil, "", false, "", now()) + if ra := r.(ReadAnswer); ra.Fill != nil || ra.Words == "" { + t.Errorf("an unmarked state says its fill as %+v", ra) + } + }) + + // A follower whose owner answers from values and marks. + follower := func(ask func(context.Context, string) (Answer, error)) (*Follower, *[]string) { + var said []string + f := &Follower{Owner: "o", Tool: "o.modules", State: "modules", Ask: ask, Bound: time.Nanosecond, Wait: time.Millisecond, + Raise: func(s Status) { said = append(said, "raised "+s.Tool) }, Clear: func(s Status) { said = append(said, "cleared") }} + return f, &said + } + answer := func(context.Context, string) (Answer, error) { + return FirstView("o", "modules", values, marks, "", now()) + } + ev := func(rev uint64, key, value string) Event { + e := Event{V: Version, Owner: "o", State: "modules", Key: key, Revision: rev, Created: made} + if value == "" { + e.Deleted = true + } else { + e.Value = json.RawMessage(value) + } + return e + } + t.Run("a reader applies a put and a delete from the events alone, and passes over what it holds", func(t *testing.T) { + asks := 0 + f, _ := follower(func(ctx context.Context, after string) (Answer, error) { asks++; return answer(ctx, after) }) + if err := f.Start(context.Background()); err != nil { + t.Fatal(err) + } + for _, e := range []Event{ev(8, "a", `{"n":0}`), ev(10, "c", `{"n":3}`), ev(11, "a", "")} { + if gap, why := f.Apply(e); gap { + t.Fatalf("revision %d is a gap: %s", e.Revision, why) + } + } + if _, ok := f.Get("a"); ok || f.Status().Revision != 11 || asks != 1 { + t.Errorf("picture = %v at %d after %d asks", f.Values(), f.Status().Revision, asks) + } + if v, _ := f.Get("c"); string(v) != `{"n":3}` { + t.Errorf("c = %s", v) + } + }) + t.Run("a gap, a state made again and a value too large each ask once more", func(t *testing.T) { + for name, e := range map[string]Event{ + "a revision missed": ev(12, "a", `{}`), + "made again": {V: Version, Owner: "o", State: "modules", Key: "a", Revision: 1, Created: made.Add(time.Minute), Value: json.RawMessage(`{}`)}, + "too large": {V: Version, Owner: "o", State: "modules", Key: "a", Revision: 10, Created: made, Gap: true}, + } { + f, _ := follower(answer) + _ = f.Start(context.Background()) + if gap, _ := f.Apply(e); !gap { + t.Errorf("%s: not taken as a gap", name) + } + } + }) + t.Run("a reader whose owner is down keeps asking, raises past its bound, and the answer clears it", func(t *testing.T) { + down := 3 + f, said := follower(func(ctx context.Context, after string) (Answer, error) { + if down > 0 { + down-- + return Answer{}, errors.New("no responders") + } + return answer(ctx, after) + }) + if err := f.Start(context.Background()); err != nil { + t.Fatal(err) + } + if strings.Join(*said, ",") != "raised o.modules,cleared" || f.Status().Asks != 4 || f.Status().Raised { + t.Errorf("said %v after %d asks", *said, f.Status().Asks) + } + }) + t.Run("an event heard while an ask runs is taken again after its answer", func(t *testing.T) { + var f *Follower + f, _ = follower(func(ctx context.Context, after string) (Answer, error) { + a, err := answer(ctx, after) // the picture read at 9, before revision 10 was written + if f.Status().Asks == 2 { + f.Apply(ev(10, "d", `{"n":4}`)) + } + return a, err + }) + _ = f.Start(context.Background()) + _ = f.Again(context.Background(), "test") + if v, ok := f.Get("d"); !ok || string(v) != `{"n":4}` || f.Status().Revision != 10 || f.Status().ByHand != 1 { + t.Errorf("d = %s, %v at %d", v, ok, f.Status().Revision) + } + }) + t.Run("an event names its state on its subject, and decodes as it was encoded", func(t *testing.T) { + if s := StateOfEvent("mesh.stream.mesh-delivery-indexer.event.views.config-candidates"); s != "config-candidates" { + t.Errorf("state of subject = %q", s) + } + raw, err := ev(3, "k", `{"x":1}`).Encode() + got, err2 := DecodeEvent(raw) + if err != nil || err2 != nil || got.Revision != 3 || string(got.Value) != `{"x":1}` { + t.Errorf("round trip = %+v %v %v", got, err, err2) + } + if _, err := (Event{Owner: "o", State: "s"}).Encode(); err == nil { + t.Errorf("an event with no key or revision is encoded") + } + }) +}