novox/mesh-sdk
9
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

G §532, §521: one subscription per event, <event>.>, and an event reaches every follower adopted on it, in both readers

Browse Source 
Builds: design 47 G §532, §521

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 14:10:56 +01:00
parent 02bea66055
commit 44063bf430
3 changed files with 90 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/views/process_reader.go
+27 -7
View File
@@ -57,7 +57,8 @@ type ProcessReader struct {
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
// 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 (`<event>.<module>.<branch>.<commit or group>`, 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
}
go/views/reader.go
+25 -4
View File
@@ -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, `<seat>.<event>`.
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, `<seat>.<event>.>` (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 {
go/views/views_test.go
+38 -2
View File
@@ -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, `<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" {