diff --git a/go/views/findings.go b/go/views/findings.go index d80c2cf8..9d02aa51 100644 --- a/go/views/findings.go +++ b/go/views/findings.go @@ -100,6 +100,9 @@ func unansweredFinding(reader string, s Status, at time.Time) viewFinding { func viewWaitsFinding(part string, s Status, at time.Time) viewFinding { summary := fmt.Sprintf("%s waits for the view %s.%s at revision %s since %s and acts on nothing new until it holds it; it holds %s", part, s.Owner, s.State, s.WaitsFor, s.WaitingSince.UTC().Format(time.RFC3339), cmp.Or(s.View, "none")) + if s.WaitsFor == HolderWait { + summary = seatWaitsSummary(part, s) // a seat's view with no holder yet (design 47 G §717) + } return viewFinding{kind: ViewWaitsKind, id: part + "." + s.Owner + "." + s.State, summary: summary, headline: fmt.Sprintf("%s waits for a view of %s", part, s.Owner), at: at.UTC()} } diff --git a/go/views/follow_seat.go b/go/views/follow_seat.go new file mode 100644 index 00000000..974eee90 --- /dev/null +++ b/go/views/follow_seat.go @@ -0,0 +1,290 @@ +package views + +// A view found by its seat (design 47 G §717): a part follows a state of the module holding a seat, the holder read from +// the seat-holders view through the holder function the part is given (A-295, A-298), so no code names the holder and a +// moved holder is followed as it moves. Nothing is asked on a timer: the holder is read again at each change of a view the +// reader follows, the seat-holders view among them. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "sync" + "time" +) + +// HolderWait is what a seat's view waits for while it follows no holder (design 47 G §717): its status's WaitsFor, and +// past WaitsBound the reader's view-waits finding names the seat and the state. +const HolderWait = "a holder of the seat" + +// seatFollow is one seat's view: the picture the part reads (its follower names the seat, never asks) and the holder's +// view it hands on. +type seatFollow struct { + seat, state string + holders func(seat string) []string + pic *Picture + kick chan struct{} + running bool + + mu sync.Mutex + holder string // the holder followed now, "" for none + inner *Picture // its view + cancel context.CancelFunc // ends the wait for its first view + handed *Picture // the view whose state the picture holds +} + +// FollowSeat follows state of the module holding seat (design 47 G §717): it returns at once a picture that reads as empty, +// its follower not ready, and never fails for want of a holder. At each change of a view the reader follows (the +// seat-holders view the holder function answers from among them) it reads the seat's first holder: on a first holder it +// follows as Follow does after Start, the first view asked of `.`, then its views family; on another holder +// it drops the old one's view and follows the new one, the picture handing on the new holder's state whole once its first +// view arrives; with no holder it keeps the picture it has and follows nothing. While no holder's first view is held the +// follower's status names the seat, waiting for HolderWait since the start or since its holder left, and past WaitsBound +// the reader writes view-waits naming the seat and the state, cleared when a holder's first view arrives. The picture's +// bus name is `_` (KeyValue opens it). A seat's state followed already is its picture. +func (r *ProcessReader) FollowSeat(seat, state string, holders func(seat string) []string) (*Picture, error) { + if seat == "" || state == "" || holders == nil { + return nil, errors.New("views: a seat's view names its seat and state, and the holder function that answers its holder") + } + bucket := seat + "_" + state + r.mu.Lock() + if sw := r.seats[bucket]; sw != nil { + r.mu.Unlock() + return sw.pic, nil + } + sw := &seatFollow{seat: seat, state: state, holders: holders, kick: make(chan struct{}, 1)} + p := &Picture{bucket: bucket, ctx: context.Background()} + sf := &Follower{Owner: seat, Tool: "seat:" + seat + "." + state, State: state, Logf: r.Logf, Now: r.Now, Changed: p.changed} + sf.RaiseWaits = func(s Status) { + if r.holdsSeat(sw) { + r.waits(s, true) + } + } + sf.ClearWaits = func(s Status) { r.waits(s, false) } + p.f, sw.pic = sf, p + if r.seats == nil { + r.seats = map[string]*seatFollow{} + } + r.seats[bucket] = sw + asked := r.asked + r.mu.Unlock() + if asked { + r.runSeat(sw) + } + return p, nil +} + +// holdsSeat says whether sw is still a seat's view of the reader. +func (r *ProcessReader) holdsSeat(sw *seatFollow) bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.seats[sw.pic.bucket] == sw +} + +// runSeat starts sw's reading of its holder once the reader asks (Ask): its wait from the start, then a read at once and at +// each change told (seatsChanged). +func (r *ProcessReader) runSeat(sw *seatFollow) { + r.mu.Lock() + if sw.running || !r.asked { + r.mu.Unlock() + return + } + sw.running = true + ctx, said := r.ctx, r.said + r.mu.Unlock() + sw.pic.f.Said = func(s Status) { + if said != nil && sw.waiting() { + said(s) + } + } + sw.pic.f.WaitFor(ctx, HolderWait) + go func() { + for { + r.reread(ctx, sw) + select { + case <-ctx.Done(): + return + case <-sw.kick: + } + } + }() +} + +// seatsChanged tells every seat's view to read its holder again, without blocking (a view's change is told outside its +// follower's lock, but may be told often: one read stands for many). +func (r *ProcessReader) seatsChanged() { + r.mu.Lock() + defer r.mu.Unlock() + for _, sw := range r.seats { + select { + case sw.kick <- struct{}{}: + default: + } + } +} + +// waiting says whether no holder's first view is held: the seat's own status is said only then. +func (sw *seatFollow) waiting() bool { + sw.mu.Lock() + defer sw.mu.Unlock() + return sw.handed == nil +} + +// reread reads the seat's first holder and follows it when it is another than the one followed. +func (r *ProcessReader) reread(ctx context.Context, sw *seatFollow) { + holder := "" + if hs := sw.holders(sw.seat); len(hs) > 0 { + holder = hs[0] + } + sw.mu.Lock() + old, cancel := sw.holder, sw.cancel + if holder == old { + sw.mu.Unlock() + return + } + sw.holder, sw.inner, sw.cancel, sw.handed = "", nil, nil, nil + sw.mu.Unlock() + if old != "" { + cancel() + if err := r.Drop(old, sw.state); err != nil { + r.logf("views: %s's view of %s's %s is not dropped: %v", r.Module, old, sw.state, err) + } + r.logf("views: %s no longer follows %s's %s: the seat %s holder is now %q", r.Module, old, sw.state, sw.seat, holder) + } + if holder == "" { + // the picture is kept; the wait is since the holder left + sw.pic.f.WaitFor(ctx, HolderWait) + return + } + inner, fresh, err := r.follow(holder, sw.state, holder+"."+sw.state) + if err != nil || !fresh { + r.logf("views: %s does not follow %s's %s for the seat %s: %v", r.Module, holder, sw.state, sw.seat, cmpErr(err, "followed already apart from the seat")) + return + } + own := inner.f.Changed + inner.f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) { + own(key, value, deleted, at) + r.apply(sw, inner, key, value, deleted, at) + } + ictx, icancel := context.WithCancel(ctx) + sw.mu.Lock() + sw.holder, sw.inner, sw.cancel = holder, inner, icancel + sw.mu.Unlock() + r.logf("views: %s follows %s's %s, the holder of the seat %s", r.Module, holder, sw.state, sw.seat) + r.begin(inner.bucket) + go func() { + select { + case <-ictx.Done(): + case <-inner.f.Ready(): + r.hand(sw, inner) + } + }() +} + +func cmpErr(err error, words string) error { + if err != nil { + return err + } + return errors.New(words) +} + +// apply takes one change of the holder's view into the seat's picture, once its state was handed on whole. +func (r *ProcessReader) apply(sw *seatFollow, inner *Picture, key string, value json.RawMessage, deleted bool, at time.Time) { + sw.mu.Lock() + if sw.inner != inner { + sw.mu.Unlock() + return + } + if sw.handed != inner { + sw.mu.Unlock() + r.hand(sw, inner) // its first view: the state whole, this change in it + return + } + defer sw.mu.Unlock() + sf := sw.pic.f + sf.mu.Lock() + if deleted { + delete(sf.values, key) + } else { + sf.values[key] = value + } + changed := sf.Changed + sf.mu.Unlock() + if changed != nil { + changed(key, value, deleted, at) + } +} + +// hand takes the holder's state whole into the seat's picture at its first view: each key it changes is told, each key the +// old holder had that the new lacks is told deleted, the picture is ready, and the wait for a holder is cleared. +func (r *ProcessReader) hand(sw *seatFollow, inner *Picture) { + sw.mu.Lock() + if sw.inner != inner || sw.handed == inner { + sw.mu.Unlock() + return + } + select { + case <-inner.f.Ready(): + default: + sw.mu.Unlock() + return + } + sw.handed = inner + holder := sw.holder + values := inner.f.Values() + sf := sw.pic.f + sf.mu.Lock() + sf.init() + old := sf.values + sf.values = maps.Clone(values) + select { + case <-sf.ready: + default: + close(sf.ready) + } + changed := sf.Changed + sf.mu.Unlock() + if changed != nil { + for k, v := range values { + if ov, ok := old[k]; !ok || string(ov) != string(v) { + changed(k, v, false, time.Time{}) + } + } + for k := range old { + if _, ok := values[k]; !ok { + changed(k, nil, true, time.Time{}) + } + } + } + sw.mu.Unlock() + sf.Holds(holder) + r.logf("views: %s holds %s's %s, the holder of the seat %s", r.Module, holder, sw.state, sw.seat) + r.mu.Lock() + status := r.status + r.mu.Unlock() + if key := StatusKey(sw.seat, sw.state); status != nil && key != StatusKey(holder, sw.state) { + if kv := status.KV(); kv != nil { + if err := kv.Delete(context.Background(), key); err != nil { + r.logf("views: the waiting status %s is not removed from %s: %v", key, StatusState, err) + } + } + } +} + +// seatPicture is the picture of a seat's view by its bus name, nil when none. +func (r *ProcessReader) seatPicture(bucket string) *Picture { + r.mu.Lock() + defer r.mu.Unlock() + if sw := r.seats[bucket]; sw != nil { + return sw.pic + } + return nil +} + +// seatWaitsSummary is the view-waits words of a seat's view that follows no holder yet. +func seatWaitsSummary(part string, s Status) string { + return fmt.Sprintf("%s waits for a module holding the seat %s, to follow its view %s, since %s", + part, s.Owner, s.State, s.WaitingSince.UTC().Format(time.RFC3339)) +} diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 1edd9a31..698f07e3 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -59,6 +59,8 @@ type ProcessReader struct { // adopted are the handlers of the adopted followers' seat events (Adopt), by the event's subject. // Several followers adopt one event (two branches' views of the catalog half), each hearing it (design 47 G §532). adopted map[string][]adoptedHear + // seats are the views found by their seat (FollowSeat), by their picture's bus name _ (design 47 G §717). + seats map[string]*seatFollow // After Start and Ask (design 47 G §532): the hub the reader's events come through when it is the module's (Hub) and // its handler, the time the bus made the consumer, the asks' life and Said, and of each first view begun, the cancel @@ -130,7 +132,10 @@ func (r *ProcessReader) follow(owner, state, tool string) (*Picture, bool, error r.finding(s, true) } }, Clear: func(s Status) { r.finding(s, false) }, - Changed: p.changed} + Changed: func(key string, value json.RawMessage, deleted bool, at time.Time) { + p.changed(key, value, deleted, at) + r.seatsChanged() // a seat's view reads its holder again (§717) + }} r.pics[bucket] = p r.order = append(r.order, bucket) return p, true, nil @@ -245,6 +250,7 @@ func (r *ProcessReader) adopt(f *Follower, seat, event string, hear func(context if own != nil { own(key, value, deleted, at) } + r.seatsChanged() // a seat's view reads its holder again (§717) } r.pics[bucket] = p r.order = append(r.order, bucket) @@ -300,6 +306,9 @@ func (r *ProcessReader) Picture(bucket string) *Picture { // 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.seatPicture(bucket); p != nil { + return p, nil + } if p := r.Picture(bucket); p != nil { return p, nil } @@ -410,10 +419,14 @@ func (r *ProcessReader) Ask(ctx context.Context, said func(Status)) { r.mu.Lock() r.ctx, r.said, r.asked = ctx, said, true buckets := slices.Clone(r.order) + seats := slices.Collect(maps.Values(r.seats)) r.mu.Unlock() for _, b := range buckets { r.begin(b) } + for _, sw := range seats { + r.runSeat(sw) + } } // begin asks bucket's first view once Ask ran, on a goroutine of its own, until answered or the view is dropped. diff --git a/go/views/views_test.go b/go/views/views_test.go index 32f945b4..806f67c4 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -960,6 +960,148 @@ func subViews(t *testing.T) { t.Errorf("after the drop: findings %v, followers %d, heard %v, acked %v", pfs.m, len(pr.Followers()), *heard["pnext"], m.acked) } }) + // Design 47 G §717: a view found by its seat starts without a holder, waits naming the seat, follows the holder the + // seat-holders view names at its change, and a moved holder is followed as it moves; nothing is asked between changes. + t.Run("a seat's view starts with no holder, follows the one the seat-holders view names, and moves with it, asking nothing between changes", func(t *testing.T) { + ended := make(chan bool, 2) + keptBound := waitBound.get() + waitBound.set(func(ctx context.Context, _ time.Duration) bool { + select { + case v := <-ended: + return v + case <-ctx.Done(): + return false + } + }) + defer waitBound.set(keptBound) + var mu sync.Mutex + asked := map[string]int{} + states := map[string]map[string]json.RawMessage{ + "assigner.seat-holders": {}, + "h1.asks": {"hold.a": json.RawMessage(`{"n":1}`), "hold.b": json.RawMessage(`{"n":2}`)}, + "h2.asks": {"hold.a": json.RawMessage(`{"n":3}`)}, + } + fs := &signalFindings{put: make(chan seenFinding, 4)} + r := &ProcessReader{Module: "teller", Findings: func() WireWriter { return fs }, Now: now, Wait: time.Millisecond, + Call: func(tool string, _ any) (json.RawMessage, error) { + mu.Lock() + defer mu.Unlock() + asked[tool]++ + vs, ok := states[tool] + if !ok { + return nil, fmt.Errorf("no tool %s", tool) + } + owner, state, _ := strings.Cut(tool, ".") + raw, _ := json.Marshal(Answer{V: Version, Owner: owner, State: state, Revision: 1, Created: made, Values: vs}) + return raw, nil + }} + sh, err := r.Follow("assigner", "seat-holders", "assigner.seat-holders") + if err != nil { + t.Fatal(err) + } + holders := func(seat string) []string { + raw, ok := sh.Follower().Get(seat) + var held struct { + Holders []struct{ Module string } `json:"holders"` + } + if !ok || json.Unmarshal(raw, &held) != nil { + return nil + } + var out []string + for _, h := range held.Holders { + out = append(out, h.Module) + } + return out + } + p, err := r.FollowSeat("mesh-wants", "asks", holders) + if err != nil || p.Bucket() != "mesh-wants_asks" { + t.Fatalf("FollowSeat = %v, %v", p, err) + } + if again, _ := r.FollowSeat("mesh-wants", "asks", holders); again != p { + t.Errorf("a seat's view followed twice is two pictures") + } + if _, err := p.Get(context.Background(), "hold.a"); err == nil { + t.Errorf("a seat's view with no holder reads as held") + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + r.Ask(ctx, nil) + <-sh.Follower().Ready() + // no holder: past the bound the reader writes view-waits naming the seat and the state + ended <- true + if fd := fs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "teller.mesh-wants.asks" || !strings.Contains(fd.Summary, "seat mesh-wants") { + t.Fatalf("view-waits of a seat with no holder: %+v", fd) + } + if s := p.Follower().Status(); s.Owner != "mesh-wants" || s.WaitsFor != HolderWait { + t.Errorf("the waiting status names %q waiting for %q", s.Owner, s.WaitsFor) + } + ready := func(want string) { + t.Helper() + deadline := time.After(2 * time.Second) + for { + if e, err := p.Get(ctx, "hold.a"); err == nil && string(e.Value()) == want { + return + } + select { + case <-deadline: + t.Fatalf("the seat's picture never held hold.a %s", want) + case <-time.After(time.Millisecond): + } + } + } + w, _ := p.Watch(ctx, "hold.*") + // the seat-holders view names h1: its first view is asked of h1.asks, and its family heard + sh.Follower().Apply(Event{V: Version, Owner: "assigner", State: "seat-holders", Key: "mesh-wants", Revision: 2, Created: made, + Value: json.RawMessage(`{"holders":[{"module":"h1"}]}`)}) + ready(`{"n":1}`) + if len(fs.m) != 0 { + t.Errorf("the holder's first view left the findings %v", fs.m) + } + if f := r.Picture("h1_asks"); f == nil { + t.Fatalf("h1's asks is not followed") + } else { + f.Follower().Apply(Event{V: Version, Owner: "h1", State: "asks", Key: "hold.c", Revision: 2, Created: made, Value: json.RawMessage(`{"n":4}`)}) + } + if e, err := p.Get(ctx, "hold.c"); err != nil || string(e.Value()) != `{"n":4}` { + t.Errorf("the holder's event did not reach the seat's picture: %v, %v", e, err) + } + // another holder: h1's view is dropped, h2's taken whole (hold.b and hold.c leave) + sh.Follower().Apply(Event{V: Version, Owner: "assigner", State: "seat-holders", Key: "mesh-wants", Revision: 3, Created: made, + Value: json.RawMessage(`{"holders":[{"module":"h2"}]}`)}) + ready(`{"n":3}`) + if keys, _ := p.Keys(ctx); strings.Join(keys, ",") != "hold.a" || r.Picture("h1_asks") != nil { + t.Errorf("after the move the picture holds %v, h1 followed %v", keys, r.Picture("h1_asks") != nil) + } + var seen []string + for !slices.Contains(seen, "hold.b:KeyValueDeleteOp") || !slices.Contains(seen, "hold.c:KeyValueDeleteOp") { + select { + case e := <-w.Updates(): + if e != nil { + seen = append(seen, e.Key()+":"+e.Operation().String()) + } + case <-time.After(2 * time.Second): + t.Fatalf("the watch saw %v", seen) + } + } + // no holder left: the picture is kept, h2 no longer followed + sh.Follower().Apply(Event{V: Version, Owner: "assigner", State: "seat-holders", Key: "mesh-wants", Revision: 4, Created: made, Deleted: true}) + deadline := time.After(2 * time.Second) + for r.Picture("h2_asks") != nil { + select { + case <-deadline: + t.Fatalf("h2 is still followed with no holder") + case <-time.After(time.Millisecond): + } + } + if e, err := p.Get(ctx, "hold.a"); err != nil || string(e.Value()) != `{"n":3}` { + t.Errorf("with no holder the picture is not kept: %v, %v", e, err) + } + mu.Lock() + defer mu.Unlock() + if asked["h1.asks"] != 1 || asked["h2.asks"] != 1 || asked["assigner.seat-holders"] != 1 { + t.Errorf("asked %v: one first view per holder and none between changes", asked) + } + }) 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")