diff --git a/go/packages_list_test.go b/go/packages_list_test.go index 918ced6..6aa3664 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -26,7 +26,7 @@ var meshSDKPackages = []string{ ".", "announce", "announcetest", "buildstep", "buscalls", "buscompose", "busstate", "busretry", "calladdress", "catalogue", "channel", "composer", "declaration", "delivery", "forgekey", "handacts", "health", "internal/beside", "kvlimit", "membership", "objectlink", "outward", "overlay", "parts", "pernode", "process", "provider", "provider/harness", "provisions", - "sdkversion", "seal", "seatdef", "testbus", "views", + "sdkversion", "seal", "seatdef", "testbus", "views", "views/viewstest", } // clientSeats are the seats whose logic is their client's in mesh-contracts (novox/hq ADR 0627 §1, §4): mesh-sdk names diff --git a/go/state.go b/go/state.go index 90824d8..c420edf 100644 --- a/go/state.go +++ b/go/state.go @@ -16,11 +16,12 @@ import ( // `.`) and names each here as it named it there; the runtime refuses, with the reason, // what it did not declare. -// StateEntry is one key's current value. +// StateEntry is one key's current value, and when the bus wrote it (zero from a runtime that does not say it). type StateEntry struct { Key string `json:"key"` Value json.RawMessage `json:"value"` Revision uint64 `json:"revision"` + At time.Time `json:"at,omitzero"` } // The operations a watch hands over (StateChange.Op). diff --git a/go/views/follow.go b/go/views/follow.go index 6192bf6..11107a4 100644 --- a/go/views/follow.go +++ b/go/views/follow.go @@ -30,8 +30,9 @@ type Follower struct { // 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 is told of each key the picture changes (deleted: the key is gone), after the change is held, with when the + // owner's bus wrote it: the event's time, or for a key a first view brought the time its answer says (Answer.Written; + // zero when the owner does not say it, and for a key the first view no longer holds). 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). @@ -210,6 +211,7 @@ func (f *Follower) askOnce(ctx context.Context) error { f.say(asked) var first Answer values := map[string]json.RawMessage{} + written := map[string]time.Time{} after := "" for page := 0; ; page++ { a, err := f.Ask(ctx, after) @@ -220,6 +222,7 @@ func (f *Follower) askOnce(ctx context.Context) error { first = a } maps.Copy(values, a.Values) + maps.Copy(written, a.Written) if a.Next == "" || a.Next == after { break } @@ -240,6 +243,7 @@ func (f *Follower) askOnce(ctx context.Context) error { delete(f.values, ev.Key) } else { f.values[ev.Key] = ev.Value + written[ev.Key] = ev.At // a change taken again was written when its event says } f.st.Revision = ev.Revision } @@ -259,7 +263,7 @@ func (f *Follower) askOnce(ctx context.Context) error { 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{}) + f.Changed(k, v, false, written[k]) } } for k := range old { diff --git a/go/views/process_reader.go b/go/views/process_reader.go new file mode 100644 index 0000000..c12ce8a --- /dev/null +++ b/go/views/process_reader.go @@ -0,0 +1,546 @@ +package views + +// A process's pictures of the owners' states it used to read (novox/hq ADR 0615 §1, design 25 § 12.2), for the parts +// that run as processes: each state is asked once through the owner's ordinary read tool with `values` (the first +// view), then kept from the owner's views family, asked again only on a gap or by hand (Follower). +// +// A Picture reads as the key-value bucket the process used to bind (jetstream.KeyValue: Get, Keys, ListKeys, Watch, +// WatchAll), so a reader's code is unchanged; every write is refused, for the owner's state is the owner's alone. The +// picture is kept in memory: the owner's state is the truth and a restart asks again (ADR 0615 §4). ProcessReader is +// the process side of Reader (the stdio side). + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// ErrReadOnly is every write to a picture: the state is its owner's. +var ErrReadOnly = errors.New("a first view is read-only: the state is its owner's") + +// OwnerReadTool is the owner's ordinary read tool of state, as a reader's `invokes` names it (design 25 § 12.2): +// `._read_`, short the owner's name past its last dash-separated `mesh-delivery-` prefix +// (indexer, assigner) and the state's dashes as underscores. +func OwnerReadTool(owner, state string) string { + short := strings.TrimPrefix(owner, "mesh-delivery-") + return owner + "." + short + "_read_" + strings.ReplaceAll(state, "-", "_") +} + +// ProcessReader is one process's pictures of its owners' states. +type ProcessReader struct { + // Module is the reading module, named in the finding of a first view unanswered past its bound. + Module string + // Call asks a module's tool (process.Conn.Call): the read tools are asked with `values`. + Call func(key string, body any) (json.RawMessage, error) + // Findings is the reader's own state `findings`, where the condition keeper raises and clears the condition of a + // first view unanswered past its bound (ADR 0615 §6); nil says it in the log only. + Findings func() jetstream.KeyValue + Logf func(string, ...any) + // Now, Bound and Wait are for tests: zero are the clock, Bound and five seconds. + Now func() time.Time + Bound time.Duration + Wait time.Duration + + mu sync.Mutex + pics map[string]*Picture // by bucket name _ + order []string +} + +func (r *ProcessReader) logf(format string, args ...any) { + if r.Logf != nil { + r.Logf(format, args...) + } +} + +// Follow adds the owner's state to follow through its read tool (OwnerReadTool). Call it before Start. +func (r *ProcessReader) Follow(owner, state string) *Picture { + r.mu.Lock() + defer r.mu.Unlock() + if r.pics == nil { + r.pics = map[string]*Picture{} + } + bucket := owner + "_" + state + if p := r.pics[bucket]; p != nil { + return p + } + tool := OwnerReadTool(owner, state) + p := &Picture{bucket: bucket, ctx: context.Background()} + call := r.Call + ask := AskVia(func(key string, body any) (json.RawMessage, error) { + m, _ := body.(map[string]any) + if m == nil { + m = map[string]any{} + } + m["values"] = true + if call == nil { + return nil, errors.New("this reader asks no tool") + } + return call(key, m) + }, tool) + p.f = &Follower{Owner: owner, Tool: tool, State: state, Ask: ask, Logf: r.Logf, Now: r.Now, Bound: r.Bound, Wait: r.Wait, + Raise: func(s Status) { r.finding(s, true) }, Clear: func(s Status) { r.finding(s, false) }, + Changed: p.changed} + r.pics[bucket] = p + r.order = append(r.order, bucket) + return p +} + +// Picture is the picture of the bucket the process used to bind, by its bus name (`_`); nil when not +// followed. +func (r *ProcessReader) Picture(bucket string) *Picture { + r.mu.Lock() + defer r.mu.Unlock() + return r.pics[bucket] +} + +// KeyValue opens a followed state's picture by its bus name, where a process opened the bucket (process.BucketOpener, +// as buildrun.FollowPlacement takes it); any other bucket is refused, for the reader reads its owners' states by their +// first views alone. +func (r *ProcessReader) KeyValue(_ context.Context, bucket string) (jetstream.KeyValue, error) { + if p := r.Picture(bucket); p != nil { + return p, nil + } + return nil, fmt.Errorf("%s reads %s by its owner's first view alone, and does not follow it", r.Module, bucket) +} + +// Followers are the followers, in the order they were added. +func (r *ProcessReader) Followers() []*Follower { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]*Follower, 0, len(r.order)) + for _, b := range r.order { + out = append(out, r.pics[b].f) + } + return out +} + +func (r *ProcessReader) finding(s Status, raised bool) { + if raised { + r.logf("the first view %s of %s's %s is unanswered past its bound: %s", s.Tool, s.Owner, s.State, s.Error) + } + if r.Findings == nil { + return + } + kv := r.Findings() + if kv == nil { + return + } + at := time.Now() + if r.Now != nil { + at = r.Now() + } + key, value, err := UnansweredFinding(r.Module, s, at) + if err == nil && raised { + _, err = kv.Put(context.Background(), key, value) + } else if err == nil { + err = kv.Delete(context.Background(), key) + } + if err != nil { + r.logf("views: the finding of the first view %s is not written: %v", s.Tool, err) + } +} + +// Hear is the handler of the owner's views family on the reader's durable consumer (cred.EventConsumer on the owner's +// views stream): each event to the follower of its state. +func (r *ProcessReader) Hear(ctx context.Context, owner string) func(jetstream.Msg) { + fs := map[string]*Follower{} + for _, f := range r.Followers() { + if f.Owner == owner { + fs[f.State] = f + } + } + return Hear(ctx, fs, r.logf) +} + +// Owners are the owners followed, sorted. +func (r *ProcessReader) Owners() []string { + var out []string + for _, f := range r.Followers() { + if !slices.Contains(out, f.Owner) { + out = append(out, f.Owner) + } + } + slices.Sort(out) + return out +} + +// Ask asks every first view until answered, each on a goroutine of its own, so the process starts at once and waits for +// an owner that is down, saying so (ADR 0615 §6). Said, when set, keeps each status (KeepStatus). +func (r *ProcessReader) Ask(ctx context.Context, said func(Status)) { + for _, f := range r.Followers() { + if said != nil { + f.Said = said + } + go func() { + if err := f.Start(ctx); err != nil { + r.logf("the first view %s was not answered: %v", f.Tool, err) + } + }() + } +} + +// Ready is closed once every followed state had its first view. +func (r *ProcessReader) Ready() <-chan struct{} { + fs := r.Followers() + done := make(chan struct{}) + go func() { + for _, f := range fs { + <-f.Ready() + } + close(done) + }() + return done +} + +// Picture is one owner's state as the process reads it: a read-only jetstream.KeyValue over a Follower. Its +// revisions are the picture's own, one per change it took, so a watcher sees them rise; they are not the owner's. An +// entry's Created is when the owner's bus wrote the key (Follower.Changed), zero when the owner did not say it. +type Picture struct { + jetstream.KeyValue // never set: every method a reader may call is the picture's own + + bucket string + ctx context.Context + f *Follower + + mu sync.Mutex + seq uint64 + revs map[string]uint64 + ats map[string]time.Time // when the owner's bus wrote each key, as the follower handed it (zero unsaid) + watchers []*watcher +} + +// Follower is the picture's follower (its status, a gap or a hand ask). +func (p *Picture) Follower() *Follower { return p.f } + +func (p *Picture) changed(key string, value json.RawMessage, deleted bool, at time.Time) { + p.mu.Lock() + p.seq++ + if p.revs == nil { + p.revs = map[string]uint64{} + } + p.revs[key] = p.seq + if p.ats == nil { + p.ats = map[string]time.Time{} + } + p.ats[key] = at + op := jetstream.KeyValuePut + if deleted { + op, value = jetstream.KeyValueDelete, nil + } + e := entry{bucket: p.bucket, key: key, value: value, rev: p.seq, at: at, op: op} + ws := slices.Clone(p.watchers) + p.mu.Unlock() + for _, w := range ws { + w.send(e) + } +} + +// answered is nil once the first view is answered, else the error a read before it gets (never a wait). +func (p *Picture) answered() error { + select { + case <-p.f.Ready(): + return nil + default: + return fmt.Errorf("no first view of %s yet: %s has not answered", p.bucket, p.f.Tool) + } +} + +func (p *Picture) ready(ctx context.Context) error { + select { + case <-p.f.Ready(): + return nil + case <-ctx.Done(): + return fmt.Errorf("no first view of %s yet (%s has not answered): %w", p.bucket, p.f.Tool, ctx.Err()) + } +} + +// Bucket is the bus name of the owner's state the picture stands for. +func (p *Picture) Bucket() string { return p.bucket } + +// Get is one key; before the first view is answered it is an error, never a wait. +func (p *Picture) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { + if err := p.answered(); err != nil { + return nil, err + } + v, ok := p.f.Get(key) + if !ok { + return nil, jetstream.ErrKeyNotFound + } + p.mu.Lock() + rev, at := p.revs[key], p.ats[key] + p.mu.Unlock() + return entry{bucket: p.bucket, key: key, value: v, rev: rev, at: at, op: jetstream.KeyValuePut}, nil +} + +// Keys is every key; none is jetstream.ErrNoKeysFound, as the bus says it; before the first view is answered it is an +// error, never a wait. +func (p *Picture) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) { + if err := p.answered(); err != nil { + return nil, err + } + var keys []string + for k := range p.f.Values() { + keys = append(keys, k) + } + if len(keys) == 0 { + return nil, jetstream.ErrNoKeysFound + } + slices.Sort(keys) + return keys, nil +} + +// ListKeys is Keys on a channel. +func (p *Picture) ListKeys(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyLister, error) { + keys, err := p.Keys(ctx, opts...) + if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, err + } + ch := make(chan string, len(keys)) + for _, k := range keys { + ch <- k + } + close(ch) + return lister{ch}, nil +} + +// WatchAll is Watch of every key. +func (p *Picture) WatchAll(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return p.Watch(ctx, ">", opts...) +} + +// Watch hands, once the first view is answered, every current value of the keys matching (a subject pattern: `*` one +// token, `>` the rest), then nil, then each change, as the bus's watch does. Options are not read: a picture holds no +// history, and deletes are always handed. +func (p *Picture) Watch(ctx context.Context, keys string, _ ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + w := &watcher{ch: make(chan jetstream.KeyValueEntry, 64), match: keys, wake: make(chan struct{}, 1), stop: make(chan struct{})} + go func() { + defer close(w.ch) + if p.ready(ctx) != nil { + return + } + p.mu.Lock() + p.watchers = append(p.watchers, w) + p.mu.Unlock() + defer p.unwatch(w) + vals := p.f.Values() + names := make([]string, 0, len(vals)) + for k := range vals { + names = append(names, k) + } + slices.Sort(names) + p.mu.Lock() + first := make([]jetstream.KeyValueEntry, 0, len(names)+1) + for _, k := range names { + if matches(keys, k) { + first = append(first, entry{bucket: p.bucket, key: k, value: vals[k], rev: p.revs[k], at: p.ats[k], op: jetstream.KeyValuePut}) + } + } + p.mu.Unlock() + first = append(first, nil) + w.run(ctx, first) + }() + return w, nil +} + +func (p *Picture) unwatch(w *watcher) { + p.mu.Lock() + defer p.mu.Unlock() + p.watchers = slices.DeleteFunc(p.watchers, func(x *watcher) bool { return x == w }) +} + +// The writes and the reads a picture cannot answer are refused, never handed to a bucket. +func (p *Picture) Put(context.Context, string, []byte) (uint64, error) { return 0, ErrReadOnly } +func (p *Picture) PutString(context.Context, string, string) (uint64, error) { return 0, ErrReadOnly } +func (p *Picture) Create(context.Context, string, []byte, ...jetstream.KVCreateOpt) (uint64, error) { + return 0, ErrReadOnly +} +func (p *Picture) Update(context.Context, string, []byte, uint64) (uint64, error) { + return 0, ErrReadOnly +} +func (p *Picture) Delete(context.Context, string, ...jetstream.KVDeleteOpt) error { return ErrReadOnly } +func (p *Picture) Purge(context.Context, string, ...jetstream.KVDeleteOpt) error { return ErrReadOnly } +func (p *Picture) PurgeDeletes(context.Context, ...jetstream.KVPurgeOpt) error { return ErrReadOnly } +func (p *Picture) GetRevision(context.Context, string, uint64) (jetstream.KeyValueEntry, error) { + return nil, errors.New("a first view holds no history") +} +func (p *Picture) History(context.Context, string, ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error) { + return nil, errors.New("a first view holds no history") +} +func (p *Picture) WatchFiltered(ctx context.Context, keys []string, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + if len(keys) != 1 { + return nil, errors.New("a first view watches one pattern") + } + return p.Watch(ctx, keys[0], opts...) +} +func (p *Picture) ListKeysFiltered(ctx context.Context, filters ...string) (jetstream.KeyLister, error) { + keys, err := p.Keys(ctx) + if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, err + } + ch := make(chan string, len(keys)) + for _, k := range keys { + if len(filters) == 0 || slices.ContainsFunc(filters, func(f string) bool { return matches(f, k) }) { + ch <- k + } + } + close(ch) + return lister{ch}, nil +} +func (p *Picture) Status(context.Context) (jetstream.KeyValueStatus, error) { + return nil, errors.New("a first view has no bucket status: the owner's read tool says its fill") +} + +// matches says whether key matches a subject pattern of the bus's watch. +func matches(pattern, key string) bool { + if pattern == "" || pattern == ">" { + return true + } + pt, kt := strings.Split(pattern, "."), strings.Split(key, ".") + for i, t := range pt { + if t == ">" { + return len(kt) > i + } + if i >= len(kt) || (t != "*" && t != kt[i]) { + return false + } + } + return len(pt) == len(kt) +} + +// watcher hands a watch's entries in order: the current values first, then each change heard, never blocking the +// follower (the changes wait in a queue of its own). +type watcher struct { + ch chan jetstream.KeyValueEntry + match string + wake chan struct{} + stop chan struct{} + once sync.Once + + mu sync.Mutex + queue []jetstream.KeyValueEntry +} + +func (w *watcher) send(e entry) { + if !matches(w.match, e.key) { + return + } + w.mu.Lock() + w.queue = append(w.queue, e) + w.mu.Unlock() + select { + case w.wake <- struct{}{}: + default: + } +} + +func (w *watcher) run(ctx context.Context, first []jetstream.KeyValueEntry) { + out := func(e jetstream.KeyValueEntry) bool { + select { + case w.ch <- e: + return true + case <-ctx.Done(): + return false + case <-w.stop: + return false + } + } + for _, e := range first { + if !out(e) { + return + } + } + for { + w.mu.Lock() + q := w.queue + w.queue = nil + w.mu.Unlock() + for _, e := range q { + if !out(e) { + return + } + } + select { + case <-w.wake: + case <-ctx.Done(): + return + case <-w.stop: + return + } + } +} + +func (w *watcher) Updates() <-chan jetstream.KeyValueEntry { return w.ch } + +func (w *watcher) Stop() error { + w.once.Do(func() { close(w.stop) }) + return nil +} + +type lister struct{ ch chan string } + +func (l lister) Keys() <-chan string { return l.ch } +func (l lister) Stop() error { return nil } + +type entry struct { + bucket, key string + value []byte + rev uint64 + at time.Time + op jetstream.KeyValueOp +} + +func (e entry) Bucket() string { return e.bucket } +func (e entry) Key() string { return e.key } +func (e entry) Value() []byte { return e.value } +func (e entry) Revision() uint64 { return e.rev } +func (e entry) Created() time.Time { return e.at } +func (e entry) Delta() uint64 { return 0 } +func (e entry) Operation() jetstream.KeyValueOp { return e.op } + +// StreamOf is the bus name of owner's views family (IndexerStream, AssignerStream): `-views` +// as the owner's old `streams` section names it. +func StreamOf(owner string) string { + return catalogue.StreamBusName(strings.TrimPrefix(owner, "mesh-delivery-") + "-views") +} + +// Start hears each owner's views family on the process's durable consumer of it (its manifest's `consumes` +// `.views.`), keeps each status in the process's state `first-views` and hears the hand asks the tools +// bundle's `_first_view_again` writes there (StatusTools), then asks every first view (Ask). A consumer +// the bus does not have yet is said: the picture is then the first view's alone, asked again on a gap or by hand. +func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred process.Account, status *process.AwaitedBucket) { + for _, owner := range r.Owners() { + stream := StreamOf(owner) + cons, err := cred.EventConsumer(ctx, js, stream) + if err == nil { + _, err = process.HearEvents(cons, r.Hear(ctx, owner), r.logf) + } + if err != nil { + r.logf("the consumer %s on %s is not heard (%v): %s's states are the first view's alone, asked again by hand, until the %s starts again", + cred.EventConsumerName(), stream, err, owner, r.Module) + } + } + var said func(Status) + if status != nil { + status.Ready(func(kv jetstream.KeyValue) { + if err := HearAgain(ctx, kv, r.Followers(), r.logf); err != nil { + r.logf("views: the hand asks of %s's first views are not heard (%v) until it starts again", r.Module, err) + } + }) + said = func(s Status) { + if kv := status.KV(); kv != nil { + KeepStatus(ctx, kv, r.logf)(s) + } + } + } + r.Ask(ctx, said) +} diff --git a/go/views/tools.go b/go/views/tools.go index e14f3e3..20b164e 100644 --- a/go/views/tools.go +++ b/go/views/tools.go @@ -22,7 +22,7 @@ func (k KeptSource) Get(key string) (*Entry, bool, error) { if err != nil || e == nil { return nil, false, err } - return &Entry{Value: e.Value, Revision: e.Revision}, true, nil + return &Entry{Value: e.Value, Revision: e.Revision, Written: e.At}, true, nil } // ParseCallers is a callers list as a manifest writes it into an env value: comma-separated, each a module (on any diff --git a/go/views/views.go b/go/views/views.go index 2be1f32..287f058 100644 --- a/go/views/views.go +++ b/go/views/views.go @@ -146,6 +146,9 @@ type Answer struct { Values map[string]json.RawMessage `json:"values"` Next string `json:"next,omitempty"` At time.Time `json:"at"` + // Written is when the owner's bus wrote each key of Values, as its state keeps it; a key whose time the owner's + // source does not say is absent. + Written map[string]time.Time `json:"written,omitempty"` } // DecodeAnswer reads an answer of this version. @@ -160,10 +163,11 @@ func DecodeAnswer(data []byte) (Answer, error) { return a, nil } -// Entry is one key's value and revision, as a state read gives it. +// Entry is one key's value and revision, as a state read gives it, and when the bus wrote it (zero when unsaid). type Entry struct { Value json.RawMessage Revision uint64 + Written time.Time } // Source is the read side of one state, as a tools bundle reaches its own state (stdio.KeptState's Keys and Get). @@ -212,6 +216,12 @@ func FirstView(owner, state string, values, marks Source, after string, now time break } a.Values[k] = e.Value + if !e.Written.IsZero() { + if a.Written == nil { + a.Written = map[string]time.Time{} + } + a.Written[k] = e.Written.UTC() + } size += len(k) + len(e.Value) } return a, nil diff --git a/go/views/views_test.go b/go/views/views_test.go index 3625195..d4187a3 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -42,14 +42,19 @@ func markOf(state string, rev uint64, made time.Time) Entry { 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}} + written := made.Add(-time.Minute) + values := mapSource{"a": {Value: json.RawMessage(`{"n":1}`), Revision: 7, Written: written}, "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) { + t.Run("a first view answers every value at the mark's revision, the mark read first, with each key's written time", 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) } + raw, _ := json.Marshal(a) + if back, err := DecodeAnswer(raw); err != nil || !back.Written["a"].Equal(written) || len(back.Written) != 1 { + t.Errorf("written times = %v (%v), want a's alone", back.Written, 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) } @@ -260,6 +265,55 @@ func TestViews(t *testing.T) { t.Errorf("asked again by hand: %v, %v", again, err) } }) + t.Run("a process's reader reads its owner's state as a read-only bucket, each key's written time handed on, and opens no other", func(t *testing.T) { + r := &ProcessReader{Module: "reader", Bound: time.Nanosecond, Wait: time.Millisecond, Now: now, + Call: func(tool string, body any) (json.RawMessage, error) { + if m, _ := body.(map[string]any); tool != "mesh-delivery-indexer.indexer_read_modules" || m["values"] != true { + return nil, fmt.Errorf("asked %s with %v", tool, body) + } + a, err := FirstView("o", "modules", values, marks, "", now()) + raw, _ := json.Marshal(map[string]any{"result": a}) + return raw, err + }} + p := r.Follow("mesh-delivery-indexer", "modules") + if _, err := p.Get(context.Background(), "a"); err == nil { + t.Errorf("a picture with no first view is read") + } + if _, err := r.KeyValue(context.Background(), "mesh-delivery-indexer_other"); err == nil { + t.Errorf("a bucket not followed is opened") + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + r.Ask(ctx, nil) + <-r.Ready() + kv, err := r.KeyValue(ctx, "mesh-delivery-indexer_modules") + if err != nil { + t.Fatal(err) + } + e, err := kv.Get(ctx, "a") + if err != nil || string(e.Value()) != `{"n":1}` || !e.Created().Equal(written) { + t.Fatalf("a = %v, %v", e, err) + } + if _, err := kv.Put(ctx, "a", nil); !errors.Is(err, ErrReadOnly) { + t.Errorf("a write to a picture: %v", err) + } + w, _ := kv.Watch(ctx, "*") + var seen []string + for e := range w.Updates() { + if e == nil { + break + } + seen = append(seen, e.Key()) + } + at := made.Add(2 * time.Minute) + ch := ev(10, "c", `{"n":3}`) + ch.At = at + p.Follower().Apply(ch) + if e := <-w.Updates(); strings.Join(seen, ",") != "a,b" || e.Key() != "c" || !e.Created().Equal(at) { + t.Errorf("watch handed %v then %v", seen, e) + } + _ = w.Stop() + }) t.Run("an ask through a tool reads the answer bare or in the tool runner's envelope, and its error", func(t *testing.T) { bare := `{"v":1,"owner":"o","state":"s","revision":3,"values":{"k":"1"}}` for name, c := range map[string]struct { diff --git a/go/views/viewstest/owner.go b/go/views/viewstest/owner.go new file mode 100644 index 0000000..eb5a1f8 --- /dev/null +++ b/go/views/viewstest/owner.go @@ -0,0 +1,107 @@ +// Package viewstest is an owner of a state as a reader's whole-module test needs it (novox/hq ADR 0615): the +// owner's read tool answering the first view from its bucket at its mark, and its views family said by the real +// views emitter onto its views stream, which the reader's durable consumer reads. +package viewstest + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// Serve is owner's state on the bus with the owner's side of it: the bucket `_` (made when absent, history +// 1), the owner's marks, its views stream with a durable consumer per reader (`_`), the emitter saying +// each change, and the read tool (views.OwnerReadTool) answering on nc. It returns the bucket the test writes. +func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state string, readers ...string) jetstream.KeyValue { + t.Helper() + bucket := owner + "_" + state + kv, err := js.KeyValue(ctx, bucket) + if err != nil { + if kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 1}); err != nil { + t.Fatal(err) + } + } + marks, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: owner + "_" + views.MarksState, History: 1}) + if err != nil { + t.Fatal(err) + } + stream := views.StreamOf(owner) + if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{process.ModuleStreamSubject(owner, views.EventPrefix+">")}}); err != nil { + t.Fatal(err) + } + for _, r := range readers { + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: r, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + } + em := views.Emitter{Owner: owner, Subject: func(e string) string { return process.ModuleStreamSubject(owner, e) }, + Publish: views.JetStreamPublish(js), Marks: marks} + if err := em.Start(ctx, state, kv, nil); err != nil { + t.Fatal(err) + } + tool := strings.TrimPrefix(views.OwnerReadTool(owner, state), owner+".") + sub, err := nc.Subscribe("mesh.mod."+owner+".call."+tool+".>", func(m *nats.Msg) { + var in struct { + After string `json:"after"` + } + _ = json.Unmarshal(m.Data, &in) + a, err := views.FirstView(owner, state, Source{ctx, kv}, Source{ctx, marks}, in.After, time.Now()) + out := map[string]any{"result": a, "node": "n0"} + if err != nil { + out = map[string]any{"error": err.Error()} + } + raw, _ := json.Marshal(out) + _ = m.Respond(raw) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + return kv +} + +// Grants are the subjects a reader's account publishes to follow owner's state (its `invokes` and `consumes`): the +// read tool's call subjects, its durable consumer's info, next and acknowledgements on the owner's views stream. +func Grants(owner, state, durable string) []string { + tool := strings.TrimPrefix(views.OwnerReadTool(owner, state), owner+".") + stream := views.StreamOf(owner) + return []string{"mesh.mod." + owner + ".call." + tool + ".>", "mesh.mod." + owner + ".call." + tool, + "$JS.API.CONSUMER.INFO." + stream + "." + durable, "$JS.API.CONSUMER.MSG.NEXT." + stream + "." + durable, + "$JS.ACK." + stream + "." + durable + ".>"} +} + +// Source is a bucket as a views.Source. +type Source struct { + Ctx context.Context + KV jetstream.KeyValue +} + +// Keys is every key; none is no error. +func (s Source) Keys() ([]string, error) { + keys, err := s.KV.Keys(s.Ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + +// Get is one key's value, revision and the time the bus wrote it. +func (s Source) Get(key string) (*views.Entry, bool, error) { + e, err := s.KV.Get(s.Ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + return &views.Entry{Value: e.Value(), Revision: e.Revision(), Written: e.Created()}, true, nil +}