diff --git a/go/views/follow.go b/go/views/follow.go index 7feb26f1..e0a4bf99 100644 --- a/go/views/follow.go +++ b/go/views/follow.go @@ -7,6 +7,7 @@ import ( "fmt" "maps" "slices" + "strings" "sync" "time" @@ -553,7 +554,7 @@ func Hear(ctx context.Context, followers map[string]*Follower, logf func(string, // HearEvent is Hear for one event given as its data, for a reader that takes the owner's events by another path than // a consumer of its own (the tool runner's): routed to the follower of its state, an event of a state it does not -// follow passed over, and a gap asked once more on its own goroutine as a gap, never as a hand ask (Gap). Data that is +// follow passed over (a per-branch state's event goes by the first token of its key, §521), and a gap asked once more on its own goroutine as a gap, never as a hand ask (Gap). Data that is // not a views event is answered as an error, and nothing is asked. func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte, logf func(string, ...any)) error { ev, err := DecodeEvent(data) @@ -561,6 +562,13 @@ func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte, return err } f := followers[ev.State] + if f == nil && ev.Key != "" { + // A per-branch state's event names the state as its owner declares it and the key `.`: it goes to + // the view `.` of the key's first token, the branch escaped as a key's name is (design 47 G §521). + if branch, _, ok := strings.Cut(ev.Key, "."); ok && branch != "" { + f = followers[ev.State+"."+branch] + } + } if f == nil { logf("views: %s's %s is a state this reader does not follow: event at revision %d passed over", ev.Owner, ev.State, ev.Revision) return nil diff --git a/go/views/reader.go b/go/views/reader.go index 47bda3b5..05a94ae5 100644 --- a/go/views/reader.go +++ b/go/views/reader.go @@ -45,6 +45,8 @@ type Reader struct { // 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 + // events is, of each per-branch view (FollowBranch), by "..", the state as its owner's events name it (§521). + events map[string]string } // adoption is how an adopted follower's events are heard: the seat's event, by its local name, and the client's handler. @@ -95,7 +97,14 @@ func BranchState(state, branch string) string { return state + "." + keys.Escape // (BranchState), so a reader follows several branches of one state, each its own view and its own row of // `_first_views` (design 47 G §479). Call it before Start. func (r *Reader) FollowBranch(owner, state, branch, tool string) *Follower { - return r.Follow(owner, BranchState(state, branch), tool) + f := r.Follow(owner, BranchState(state, branch), tool) + r.mu.Lock() + if r.events == nil { + r.events = map[string]string{} + } + r.events[owner+"."+f.State] = state + r.mu.Unlock() + 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 @@ -218,13 +227,30 @@ 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) + adopted, events := maps.Clone(r.adopted), maps.Clone(r.events) r.mu.Unlock() + // One subscription per event pattern: the followers of a per-branch state share the owner's one event, which + // HearEvent routes to the branch's view by the key's first token (§521). + subs := map[string][]*Follower{} + var patterns []string for _, name := range names { f := r.followers[name] - pattern, handle := f.Owner+"."+EventName(f.State), r.hear(ctx, f) + pattern := f.Owner + "." + EventName(f.State) if a, ok := adopted[name]; ok { - pattern, handle = a.local(), r.hearAdopted(ctx, a) + pattern = a.local() + } else if state, ok := events[name]; ok { + pattern = f.Owner + "." + EventName(state) + } + if subs[pattern] == nil { + patterns = append(patterns, pattern) + } + subs[pattern] = append(subs[pattern], f) + } + for _, pattern := range patterns { + fs := subs[pattern] + handle := r.hear(ctx, fs) + if a, ok := adopted[fs[0].Owner+"."+fs[0].State]; ok { + handle = r.hearAdopted(ctx, a) } go func() { for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { @@ -232,15 +258,19 @@ func (r *Reader) Start(ctx context.Context) { if err == nil { break } - r.logf("not hearing %s yet for %s's %s (%v); asking again in %s", pattern, f.Owner, f.State, err, wait) + r.logf("not hearing %s yet for %s's %s (%v); asking again in %s", pattern, fs[0].Owner, fs[0].State, err, wait) select { case <-ctx.Done(): return case <-time.After(wait): } } - if err := f.Start(ctx); err != nil { - r.logf("the first view %s was not answered: %v", f.Tool, err) + for _, f := range fs { + go func() { + if err := f.Start(ctx); err != nil { + r.logf("the first view %s was not answered: %v", f.Tool, err) + } + }() } }() } @@ -248,9 +278,13 @@ func (r *Reader) Start(ctx context.Context) { // hear is the handler of one followed state's events: HearEvent routes it, and a gap asks that first view once more as a // gap, said in the log. -func (r *Reader) hear(ctx context.Context, f *Follower) func(stdio.Envelope) error { +func (r *Reader) hear(ctx context.Context, fs []*Follower) func(stdio.Envelope) error { + byState := map[string]*Follower{} + for _, f := range fs { + byState[f.State] = f + } return func(e stdio.Envelope) error { - if err := HearEvent(ctx, map[string]*Follower{f.State: f}, e.Body, r.logf); err != nil { + if err := HearEvent(ctx, byState, e.Body, r.logf); err != nil { r.logf("views: %s is not a views event, dropped: %v", e.Key, err) } return nil diff --git a/go/views/views_test.go b/go/views/views_test.go index ca313534..911ad900 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -642,6 +642,41 @@ func TestViews(t *testing.T) { } }) + // Design 47 G §521: the owner's event names the state, the key names the branch; one event reaches the branch's own view. + t.Run("an event of a per-branch state goes to the view of its key's first token, one subscription for the branches", func(t *testing.T) { + made := now() + ask := func(context.Context, string) (Answer, error) { + return FirstView("s", "catalog-revisions", mapSource{}, mapSource{"catalog-revisions": markOf("catalog-revisions", 1, made)}, "", now()) + } + r := &Reader{Module: "reader", Now: now} + a, b := r.FollowBranch("s", "catalog-revisions", "main", "s.read"), r.FollowBranch("s", "catalog-revisions", "feat/x", "s.read") + for _, f := range []*Follower{a, b} { + f.Ask = ask + if err := f.Start(context.Background()); err != nil { + t.Fatal(err) + } + } + followers := map[string]*Follower{a.State: a, b.State: b} + say := func(key string, rev uint64) { + data, _ := Event{Owner: "s", State: "catalog-revisions", Key: key, Revision: rev, Created: made, Value: json.RawMessage(`{}`)}.Encode() + if err := HearEvent(context.Background(), followers, data, t.Logf); err != nil { + t.Fatal(err) + } + } + say("main.r1", 2) + say("feat/x.r7", 2) + ka, kb := a.Values(), b.Values() + if _, ok := ka["main.r1"]; !ok || len(ka) != 1 { + t.Errorf("main's view holds %v", ka) + } + if _, ok := kb["feat/x.r7"]; !ok || len(kb) != 1 { + t.Errorf("feat/x's view holds %v", kb) + } + if r.events["s."+a.State] != "catalog-revisions" || r.events["s."+b.State] != "catalog-revisions" { + t.Errorf("events = %v", r.events) + } + }) + 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