diff --git a/go/views/process_reader.go b/go/views/process_reader.go index c342bc84..36a3abe6 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -57,7 +57,8 @@ type ProcessReader struct { 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 + // Several followers adopt one event (two branches' views of the catalog half), each hearing it (design 47 G §532). + adopted map[string][]func(context.Context, []byte) error } func (r *ProcessReader) logf(format string, args ...any) { @@ -156,9 +157,10 @@ func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context 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 = map[string][]func(context.Context, []byte) error{} } - r.adopted["mesh.seat."+seat+".event."+event] = hear + subject := "mesh.seat." + seat + ".event." + event + r.adopted[subject] = append(r.adopted[subject], hear) return p, nil } @@ -265,9 +267,11 @@ func (r *ProcessReader) Hear(ctx context.Context) func(jetstream.Msg) { 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) + if hears := adoptedHears(adopted, m.Subject()); hears != nil { + for _, hear := range hears { + if err := hear(ctx, m.Data()); err != nil { + r.logf("views: %s is not taken: %v", m.Subject(), err) + } } _ = m.Ack() return @@ -656,8 +660,9 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred subjects := slices.Sorted(maps.Keys(r.adopted)) r.mu.Unlock() for _, subject := range subjects { + // one subscription per event, with every further part (design 47 G §532) if err == nil { - err = hub.Handle(subject, hear) + err = hub.Handle(subject+".>", hear) } } } @@ -683,3 +688,18 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred } r.Ask(ctx, said) } + +// adoptedHears are the handlers of every follower adopted on the event whose subject is subject or has further parts +// after it (`...`, design 47 G §521, §532), or nil. +func adoptedHears(adopted map[string][]func(context.Context, []byte) error, subject string) []func(context.Context, []byte) error { + if h := adopted[subject]; h != nil { + return h + } + for i := strings.LastIndex(subject, "."); i > 0; i = strings.LastIndex(subject, ".") { + subject = subject[:i] + if h := adopted[subject]; h != nil { + return h + } + } + return nil +} diff --git a/go/views/reader.go b/go/views/reader.go index 05a94ae5..9aa24775 100644 --- a/go/views/reader.go +++ b/go/views/reader.go @@ -55,8 +55,9 @@ type adoption struct { 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 } +// local is an adopted follower's event as the stdio runtime subscribes it: one subscription per event with every +// further part, `..>` (design 47 G §532, §521). +func (a adoption) local() string { return a.seat + "." + a.event + ".>" } func (r *Reader) logf(format string, args ...any) { if r.Logf != nil { @@ -249,8 +250,8 @@ func (r *Reader) Start(ctx context.Context) { 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) + if h := r.hearAdoptedBy(ctx, fs, adopted); h != nil { + handle = h } go func() { for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { @@ -291,6 +292,26 @@ func (r *Reader) hear(ctx context.Context, fs []*Follower) func(stdio.Envelope) } } +// hearAdoptedBy is the handler of an event every adopted follower of fs hears, never the first alone (two branches' views +// of the catalog half share one event, design 47 G §532); nil when none of fs is adopted. +func (r *Reader) hearAdoptedBy(ctx context.Context, fs []*Follower, adopted map[string]adoption) func(stdio.Envelope) error { + var heard []func(stdio.Envelope) error + for _, f := range fs { + if a, ok := adopted[f.Owner+"."+f.State]; ok { + heard = append(heard, r.hearAdopted(ctx, a)) + } + } + if len(heard) == 0 { + return nil + } + return func(e stdio.Envelope) error { + for _, h := range heard { + _ = h(e) + } + return nil + } +} + // 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 { diff --git a/go/views/views_test.go b/go/views/views_test.go index 911ad900..86e3ce65 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -334,7 +334,9 @@ func TestViews(t *testing.T) { t.Errorf("finding = %+v", fd) } none := &Follower{Owner: "o", Tool: "o.revision", State: BranchState("catalog", "new"), Now: now, - Ask: func(context.Context, string) (Answer, error) { return Answer{V: 1, Owner: "o", State: "catalog.new", At: now()}, nil }} + Ask: func(context.Context, string) (Answer, error) { + return Answer{V: 1, Owner: "o", State: "catalog.new", At: now()}, nil + }} if err := none.Start(context.Background()); err != nil { t.Fatal(err) } @@ -709,7 +711,7 @@ func TestViews(t *testing.T) { 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 { + 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") @@ -741,6 +743,40 @@ func TestViews(t *testing.T) { 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) } + // G §521, §532: an event's further parts reach the handler too + pr.Hear(context.Background())(&seatMsg{subject: "mesh.seat.s.event.revision-said.runner.main.abc", data: []byte(`{"revision":5}`)}) + if len(heard) != 3 { + t.Errorf("an event with further parts: heard %v", heard) + } + // G §532: one subscription per event, `.>`, and every follower adopted on the event hears it, two + // branches' views on one reader each + two := &Reader{Module: "reader", Now: now} + var one, other []string + for branch, got := range map[string]*[]string{"main": &one, "next": &other} { + f := &Follower{Owner: "s", State: BranchState("revisions", branch), Ask: answer} + if err := two.Adopt(f, "s", "revision-said", func(_ context.Context, d []byte) error { *got = append(*got, string(d)); return nil }); err != nil { + t.Fatal(err) + } + } + var fol []*Follower + for _, n := range two.order { + fol = append(fol, two.followers[n]) + } + if err := two.hearAdoptedBy(context.Background(), fol, two.adopted)(stdio.Envelope{Key: "s.revision-said.main", Body: []byte(`{"revision":6}`)}); err != nil || len(one) != 1 || len(other) != 1 { + t.Errorf("two adopted views on one reader heard %v and %v (%v)", one, other, err) + } + var a, b []string + prr := &ProcessReader{Module: "reader", Now: now} + for i, got := range []*[]string{&a, &b} { + f := &Follower{Owner: "s", State: BranchState("revisions", []string{"main", "next"}[i]), Ask: answer} + if _, err := prr.Adopt(f, "s", "revision-said", func(_ context.Context, d []byte) error { *got = append(*got, string(d)); return nil }); err != nil { + t.Fatal(err) + } + } + prr.Hear(context.Background())(&seatMsg{subject: "mesh.seat.s.event.revision-said.main", data: []byte(`{"revision":7}`)}) + if len(a) != 1 || len(b) != 1 { + t.Errorf("two adopted views on one process reader heard %v and %v", a, b) + } pf.WaitFor(context.Background(), "4") ended <- true if fd := pfs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" {