V-2: W1.13 (views adopt): a reader adopts a follower a seat's client made, so its callers write no glue

Six readers of the catalog half each wrote the same glue (dev-G-links' finding): the follower's conditions and view-waits, the seat event's subscription, Keys/Get/Watch over its values and a wrap of the first-view tools. Reader.Adopt and ProcessReader.Adopt take the client's follower with its seat's event and handler: the reader writes its findings (unanswered, view-waits; Follow's followers now write view-waits too), reads and lists it as any first view, keeps its own Changed after the reader's, and hears the seat's event (stdio: <seat>.<event>; process: the durable's mesh.seat.<seat>.event.<event>, also on a shared hub) into the handler. A subtest of TestViews. Builds: design 47 G §262, §260 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-11 10:51:07 +01:00
parent fa16c39057
commit 86ccaae967
3 changed files with 316 additions and 8 deletions
@@ -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 <owner>_<state>
 	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 (`<owner>_<state>`) 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 (`<owner>_<state>`); 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)
 	}
@@ -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 "<owner>.<state>": 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, `<seat>.<event>`.
+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 `<owner>.<state>`.
+// Its events are the seat's event (`<seat>.<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()
@@ -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