diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 20601f50..edac57bb 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -14,6 +14,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "slices" "strings" "sync" @@ -55,6 +56,8 @@ type ProcessReader struct { mu sync.Mutex pics map[string]*Picture // by bucket name _ order []string + // adopted are the handlers of the adopted followers' seat events (Adopt), by the event's subject. + adopted map[string]func(context.Context, []byte) error } func (r *ProcessReader) logf(format string, args ...any) { @@ -101,6 +104,85 @@ func (r *ProcessReader) Follow(owner, state, tool string) (*Picture, error) { return p, nil } +// Adopt takes a follower a seat's client made (its Ask a seat's verb, such as the catalog half's view) as one of the +// process's first views, so the client's callers write no glue: the reader sets its conditions (the unanswered first +// view and view-waits, in the reader's findings), its log and clock, and its picture (`_`) hands each +// change on before the follower's own Changed, set before. Start keeps its status, hears its hand asks, and asks it with +// the others; its events are the seat's event (seatdef's EventSubject of seat and event, on the module's durable +// consumer, as its `consumes` names it), handed to hear, the client's handler, and acknowledged. Call it before Start; +// a state followed already is refused. +func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) (*Picture, error) { + if f == nil || f.Owner == "" || f.State == "" || seat == "" || event == "" || hear == nil { + return nil, errors.New("views: an adopted follower names its owner and state, and its seat's event and handler") + } + r.mu.Lock() + defer r.mu.Unlock() + if r.pics == nil { + r.pics = map[string]*Picture{} + } + bucket := f.Owner + "_" + f.State + if r.pics[bucket] != nil { + return nil, fmt.Errorf("views: %s's %s is followed already", f.Owner, f.State) + } + p := &Picture{bucket: bucket, ctx: context.Background(), f: f} + if f.Logf == nil { + f.Logf = r.Logf + } + if f.Now == nil { + f.Now = r.Now + } + if f.Bound == 0 { + f.Bound = r.Bound + } + if f.Wait == 0 { + f.Wait = r.Wait + } + f.Raise, f.Clear = func(s Status) { r.finding(s, true) }, func(s Status) { r.finding(s, false) } + f.RaiseWaits, f.ClearWaits = func(s Status) { r.waits(s, true) }, func(s Status) { r.waits(s, false) } + own := f.Changed + f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) { + p.changed(key, value, deleted, at) + if own != nil { + own(key, value, deleted, at) + } + } + r.pics[bucket] = p + r.order = append(r.order, bucket) + if r.adopted == nil { + r.adopted = map[string]func(context.Context, []byte) error{} + } + r.adopted["mesh.seat."+seat+".event."+event] = hear + return p, nil +} + +// waits writes (raised) or clears the reader's view-waits finding of s (design 47 *A part with no view says it waits*). +func (r *ProcessReader) waits(s Status, raised bool) { + if raised { + r.logf("%s has waited for %s's view %s since %s", r.Module, s.Owner, s.WaitsFor, s.WaitingSince.Format(time.RFC3339)) + } + if r.Findings == nil { + return + } + kv := r.Findings() + if kv == nil { + return + } + at := time.Now() + if r.Now != nil { + at = r.Now() + } + f := ViewWaitsFinding(r.Module, s, at) + var err error + if raised { + err = kv.Put(f) + } else { + err = kv.Delete(f) + } + if err != nil { + r.logf("views: the view-waits finding of %s is not written: %v", s.Tool, err) + } +} + // 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 { @@ -172,7 +254,17 @@ func (r *ProcessReader) Hear(ctx context.Context) func(jetstream.Msg) { for owner, m := range fs { views[owner] = Hear(ctx, m, r.logf) } + r.mu.Lock() + adopted := maps.Clone(r.adopted) + r.mu.Unlock() return func(m jetstream.Msg) { + if hear := adopted[m.Subject()]; hear != nil { + if err := hear(ctx, m.Data()); err != nil { + r.logf("views: %s is not taken: %v", m.Subject(), err) + } + _ = m.Ack() + return + } if owner, ok := OwnerOfEvent(m.Subject()); ok { if h := views[owner]; h != nil { h(m) @@ -550,7 +642,18 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred f.ConsumerMade(ctx, created) } }) - err := hub.Handle(pattern, r.Hear(ctx)) + hear := r.Hear(ctx) + err := hub.Handle(pattern, hear) + if !own { + r.mu.Lock() + subjects := slices.Sorted(maps.Keys(r.adopted)) + r.mu.Unlock() + for _, subject := range subjects { + if err == nil { + err = hub.Handle(subject, hear) + } + } + } if err == nil && own { err = hub.Start(ctx, js, cred) } diff --git a/go/views/reader.go b/go/views/reader.go index 8a67ab62..bf4982c4 100644 --- a/go/views/reader.go +++ b/go/views/reader.go @@ -3,7 +3,9 @@ package views import ( "context" "encoding/json" + "errors" "fmt" + "maps" "sort" "strings" "sync" @@ -39,8 +41,20 @@ type Reader struct { order []string byOwner map[string]map[string]*Follower // owner -> state -> follower watchers map[string][]func(key string, value []byte, gone bool) + // adopted are the followers a seat's client made (Adopt), by ".": their events are a seat's, heard by + // the client's own handler, not an owner's views family. + adopted map[string]adoption } +// adoption is how an adopted follower's events are heard: the seat's event, by its local name, and the client's handler. +type adoption struct { + seat, event string + hear func(context.Context, []byte) error +} + +// local is an adopted follower's event as the stdio runtime subscribes it, `.`. +func (a adoption) local() string { return a.seat + "." + a.event } + func (r *Reader) logf(format string, args ...any) { if r.Logf != nil { r.Logf(format, args...) @@ -68,9 +82,57 @@ func (r *Reader) Follow(owner, state, tool string) *Follower { } return DecodeAnswer(raw) } + r.wireLocked(name, f, nil) + return f +} + +// Adopt takes a follower a seat's client made (its Ask a seat's verb, such as the catalog half's view) as one of the +// reader's first views, so the client's callers write no glue: the reader sets its conditions (the unanswered first view +// and view-waits, in the reader's findings), its log and clock, and hands each change to Watch, Changed and the +// follower's own Changed, set before; Keys, Get, Values, Watch, Statuses and the two tools name it `.`. +// Its events are the seat's event (`.`, as the reader's `consumes` names it), which Start subscribes and +// hands to hear, the client's handler. Call it before Start; a state followed already is refused. +func (r *Reader) Adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) error { + if f == nil || f.Owner == "" || f.State == "" || seat == "" || event == "" || hear == nil { + return errors.New("views: an adopted follower names its owner and state, and its seat's event and handler") + } + r.mu.Lock() + defer r.mu.Unlock() + if r.followers == nil { + r.followers, r.byOwner = map[string]*Follower{}, map[string]map[string]*Follower{} + } + name := f.Owner + "." + f.State + if r.followers[name] != nil { + return fmt.Errorf("views: %s is followed already", name) + } + if f.Logf == nil { + f.Logf = r.Logf + } + if f.Now == nil { + f.Now = r.Now + } + if f.Bound == 0 { + f.Bound = r.Bound + } + if f.Wait == 0 { + f.Wait = r.Wait + } + r.wireLocked(name, f, f.Changed) + if r.adopted == nil { + r.adopted = map[string]adoption{} + } + r.adopted[name] = adoption{seat: seat, event: event, hear: hear} + return nil +} + +// wireLocked makes f one of the reader's first views under name: its conditions are the reader's, and each change goes +// to the watchers, Changed and own (the follower's own Changed, or nil). The caller holds mu. +func (r *Reader) wireLocked(name string, f *Follower, own func(string, json.RawMessage, bool, time.Time)) { f.Raise = func(s Status) { r.raise(s) } f.Clear = func(s Status) { r.clear(s) } - f.Changed = func(key string, value json.RawMessage, deleted bool, _ time.Time) { + f.RaiseWaits = func(s Status) { r.waits(s, true) } + f.ClearWaits = func(s Status) { r.waits(s, false) } + f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) { r.mu.Lock() ws := append([]func(string, []byte, bool){}, r.watchers[name]...) r.mu.Unlock() @@ -80,14 +142,36 @@ func (r *Reader) Follow(owner, state, tool string) *Follower { if r.Changed != nil { r.Changed(name, key, value, deleted) } + if own != nil { + own(key, value, deleted, at) + } } r.followers[name] = f r.order = append(r.order, name) - if r.byOwner[owner] == nil { - r.byOwner[owner] = map[string]*Follower{} + if r.byOwner[f.Owner] == nil { + r.byOwner[f.Owner] = map[string]*Follower{} + } + r.byOwner[f.Owner][f.State] = f +} + +// waits writes (raised) or clears the reader's view-waits finding of s (design 47 *A part with no view says it waits*). +func (r *Reader) waits(s Status, raised bool) { + if raised { + r.logf("%s has waited for %s's view %s since %s", r.Module, s.Owner, s.WaitsFor, s.WaitingSince.Format(time.RFC3339)) + } + if r.Findings == nil { + return + } + f := ViewWaitsFinding(r.Module, s, r.nowTime()) + var err error + if raised { + err = r.Findings.Put(f) + } else { + err = r.Findings.Delete(f) + } + if err != nil { + r.logf("the view-waits finding of %s is not written: %v", s.Tool, err) } - r.byOwner[owner][state] = f - return f } func (r *Reader) nowTime() time.Time { @@ -122,16 +206,21 @@ func (r *Reader) clear(s Status) { func (r *Reader) Start(ctx context.Context) { r.mu.Lock() names := append([]string(nil), r.order...) + adopted := maps.Clone(r.adopted) r.mu.Unlock() for _, name := range names { f := r.followers[name] + pattern, handle := f.Owner+"."+EventName(f.State), r.hear(ctx, f) + if a, ok := adopted[name]; ok { + pattern, handle = a.local(), r.hearAdopted(ctx, a) + } go func() { for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - err := stdio.Subscribe(f.Owner+"."+EventName(f.State), r.hear(ctx, f)) + err := stdio.Subscribe(pattern, handle) if err == nil { break } - r.logf("not hearing %s's %s events yet (%v); asking again in %s", f.Owner, f.State, err, wait) + r.logf("not hearing %s yet for %s's %s (%v); asking again in %s", pattern, f.Owner, f.State, err, wait) select { case <-ctx.Done(): return @@ -156,6 +245,17 @@ func (r *Reader) hear(ctx context.Context, f *Follower) func(stdio.Envelope) err } } +// hearAdopted is the handler of an adopted follower's seat event: the client's handler takes it, and what it cannot read is +// said in the log. +func (r *Reader) hearAdopted(ctx context.Context, a adoption) func(stdio.Envelope) error { + return func(e stdio.Envelope) error { + if err := a.hear(ctx, e.Body); err != nil { + r.logf("views: %s is not taken: %v", e.Key, err) + } + return nil + } +} + // Ready is closed once every followed state had its first view. func (r *Reader) Ready() <-chan struct{} { r.mu.Lock() diff --git a/go/views/views_test.go b/go/views/views_test.go index 59cbc2ee..5850ffcf 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -592,6 +592,79 @@ func TestViews(t *testing.T) { } _ = w.Stop() }) + // A seat's client makes the follower (its Ask a seat's verb) and both readers adopt it, so its callers write no glue: + // the reader's findings (unanswered, view-waits), Keys, Get, Watch, statuses and tools, and the seat's event handed to + // the client's handler. + t.Run("a follower a seat's client made is adopted: the reader's findings, reads and tools hold it, and its seat event goes to the client", func(t *testing.T) { + ended := make(chan bool, 1) + kept := waitBound + waitBound = func(context.Context, time.Duration) bool { return <-ended } + defer func() { waitBound = kept }() + fs := &signalFindings{put: make(chan Finding, 4)} + r := &Reader{Module: "reader", Findings: fs, Now: now} + var own []string + f := &Follower{Owner: "s", Tool: "seat:s.revision", State: "revisions", Ask: answer, + Changed: func(k string, _ json.RawMessage, _ bool, _ time.Time) { own = append(own, k) }} + var heard []string + hear := func(_ context.Context, data []byte) error { heard = append(heard, string(data)); return nil } + if err := r.Adopt(f, "s", "revision-said", hear); err != nil { + t.Fatal(err) + } + if err := r.Adopt(f, "s", "revision-said", hear); err == nil { + t.Errorf("a state followed already is adopted twice") + } + if err := r.Adopt(&Follower{Owner: "s", State: "x"}, "s", "", hear); err == nil { + t.Errorf("a follower with no seat event is adopted") + } + if err := f.Start(context.Background()); err != nil { + t.Fatal(err) + } + keys, err := r.Keys("s.revisions") + if v, ok, _ := r.Get("s.revisions", "b"); err != nil || len(keys) != 2 || !ok || string(v) != `{"n":2}` || len(own) != 2 { + t.Errorf("keys %v (%v), b %s, own Changed %v", keys, err, v, own) + } + if st := r.Statuses(); len(st) != 1 || st[0].Tool != "seat:s.revision" { + t.Errorf("statuses = %+v", st) + } + _ = r.hearAdopted(context.Background(), r.adopted["s.revisions"])(stdio.Envelope{Key: "s.revision-said", Body: []byte(`{"revision":3}`)}) + if r.adopted["s.revisions"].local() != "s.revision-said" || len(heard) != 1 { + t.Errorf("the seat event %v reached the client %v", r.adopted, heard) + } + f.WaitFor(context.Background(), "3") + ended <- true + if fd := fs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" { + t.Errorf("view-waits is not the reader's finding: %+v", fd) + } + f.Holds("3") + if len(fs.m) != 0 { + t.Errorf("holding the view left %+v", fs.m) + } + + // The process form: the picture reads the adopted follower, and the seat's event on the durable goes to the client. + pfs := &signalFindings{put: make(chan Finding, 4)} + pr := &ProcessReader{Module: "reader", Findings: func() FindingWriter { return pfs }, Now: now} + pf := &Follower{Owner: "s", Tool: "seat:s.revision", State: "revisions", Ask: answer} + p, err := pr.Adopt(pf, "s", "revision-said", hear) + if err != nil || p.Bucket() != "s_revisions" || len(pr.Followers()) != 1 { + t.Fatalf("adopted %v, %v", p, err) + } + if err := pf.Start(context.Background()); err != nil { + t.Fatal(err) + } + if e, err := p.Get(context.Background(), "a"); err != nil || string(e.Value()) != `{"n":1}` { + t.Errorf("a = %v, %v", e, err) + } + m := &seatMsg{subject: "mesh.seat.s.event.revision-said", data: []byte(`{"revision":4}`)} + pr.Hear(context.Background())(m) + if len(heard) != 2 || heard[1] != `{"revision":4}` || !m.acked { + t.Errorf("the seat event on the durable: heard %v, acked %v", heard, m.acked) + } + pf.WaitFor(context.Background(), "4") + ended <- true + if fd := pfs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" { + t.Errorf("view-waits is not the process reader's finding: %+v", fd) + } + }) t.Run("a reader's again tool records its hand act before it writes the ask, and asks nothing when the act is not taken", func(t *testing.T) { var recorded []HandAct take := errors.New("no responders") @@ -770,6 +843,38 @@ func (m *marksKV) Put(context.Context, string, []byte) (uint64, error) { func (f *findingsMap) Delete(v Finding) error { delete(f.m, v.Kind+"/"+v.ID); f.deletes++; return nil } +// signalFindings is a findingsMap that hands each finding written to put, once written. +type signalFindings struct { + findingsMap + put chan Finding +} + +func (f *signalFindings) Put(v Finding) error { _ = f.findingsMap.Put(v); f.put <- v; return nil } + +// next is the next finding written; none within a second fails the test (a bound, never waited out when it comes). +func (f *signalFindings) next(t *testing.T) Finding { + t.Helper() + select { + case v := <-f.put: + return v + case <-time.After(time.Second): + t.Fatal("no finding was written") + return Finding{} + } +} + +// seatMsg is one message of a seat's event on a module's durable consumer. +type seatMsg struct { + jetstream.Msg + subject string + data []byte + acked bool +} + +func (m *seatMsg) Subject() string { return m.subject } +func (m *seatMsg) Data() []byte { return m.data } +func (m *seatMsg) Ack() error { m.acked = true; return nil } + // actsFunc records a hand act through a test's function. type actsFunc func(HandAct) error