diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 917f4ebb..54a3106d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -31,7 +31,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/provisions" "mesh-delivery-assigner/internal/judge" - "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-assigner/internal/assign" ) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 21bae134..8e8b47ad 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -29,7 +29,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/provisions" "git.novox.be/novox/mesh-delivery/sdk/graph" - "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-assigner/internal/assign" ) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 96a81df1..008ea532 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -13,7 +13,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" - "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-assigner/internal/assign" ) diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 0f1c8dc7..b6169296 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -9,7 +9,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/views" ) func load(t *testing.T) catalogue.Manifest { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go index c86c82e0..f74065d7 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go @@ -8,7 +8,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-indexer/internal/index" ) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 3a228678..18f8e2d9 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -23,7 +23,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" - "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-indexer/internal/index" ) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go index b642d7ec..c48b5fb4 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -14,7 +14,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-delivery/sdk/graph" - "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/views" ) // findingsBucket is the indexer's own findings, which the condition keeper raises and clears (ADR 0564 §2): a value too diff --git a/modules/mesh-delivery-indexer/internal/index/views.go b/modules/mesh-delivery-indexer/internal/index/views.go index ee2b7f9c..5a3ceaba 100644 --- a/modules/mesh-delivery-indexer/internal/index/views.go +++ b/modules/mesh-delivery-indexer/internal/index/views.go @@ -5,7 +5,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-delivery/sdk/graph" - "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/views" ) // ViewsStream is the indexer's views family on its old streams section (ADR 0615 §5): one event per changed key of diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 478af047..d62d7fc5 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -16,8 +16,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-delivery/sdk/graph" - "git.novox.be/novox/mesh-delivery/sdk/views" "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-indexer/internal/index" ) diff --git a/sdk/go.mod b/sdk/go.mod index 84b39a5d..7e692d40 100644 --- a/sdk/go.mod +++ b/sdk/go.mod @@ -2,15 +2,4 @@ module git.novox.be/novox/mesh-delivery/sdk go 1.26.0 -require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 - github.com/nats-io/nats.go v1.54.0 -) - -require ( - github.com/klauspost/compress v1.20.0 // indirect - github.com/nats-io/nkeys v0.4.16 // indirect - github.com/nats-io/nuid v1.0.1 // indirect - golang.org/x/crypto v0.57.0 // indirect - golang.org/x/sys v0.48.0 // indirect -) +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 diff --git a/sdk/views/emit.go b/sdk/views/emit.go deleted file mode 100644 index 23a251ae..00000000 --- a/sdk/views/emit.go +++ /dev/null @@ -1,196 +0,0 @@ -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/sdk/views/findings.go b/sdk/views/findings.go deleted file mode 100644 index 9795fe10..00000000 --- a/sdk/views/findings.go +++ /dev/null @@ -1,47 +0,0 @@ -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/sdk/views/follow.go b/sdk/views/follow.go deleted file mode 100644 index 2c0a4709..00000000 --- a/sdk/views/follow.go +++ /dev/null @@ -1,358 +0,0 @@ -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/sdk/views/status.go b/sdk/views/status.go deleted file mode 100644 index 19907d31..00000000 --- a/sdk/views/status.go +++ /dev/null @@ -1,145 +0,0 @@ -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/sdk/views/tools.go b/sdk/views/tools.go deleted file mode 100644 index e14f3e3e..00000000 --- a/sdk/views/tools.go +++ /dev/null @@ -1,159 +0,0 @@ -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/sdk/views/views.go b/sdk/views/views.go deleted file mode 100644 index 2be1f32f..00000000 --- a/sdk/views/views.go +++ /dev/null @@ -1,244 +0,0 @@ -// 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/sdk/views/views_test.go b/sdk/views/views_test.go deleted file mode 100644 index da65aada..00000000 --- a/sdk/views/views_test.go +++ /dev/null @@ -1,204 +0,0 @@ -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") - } - }) -}