From 9f2d1df1c9deca9d249dd62f3fe42fa1af1e19cc Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:11:26 +0100 Subject: [PATCH] B1g-1: an adopted event whose handler fails is left unacknowledged, the bus asks again Builds: design 47 G A-222, A-193, A-199 --- go/views/process_reader.go | 9 +++++++-- go/views/views_test.go | 22 ++++++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 3ac45459..2e98cba0 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -181,7 +181,7 @@ func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context // AdoptEvent hears a seat's event on the one subscription the followers adopted on that event share, with no picture and no // first view (design 47 G A-185, A-199): hear is handed each message of the event beside every adopted follower's handler, -// and the message is acknowledged. The subscription is made at once when no follower made it, so a part that follows no +// and the message is acknowledged, unless hear fails: then it is left unacknowledged and the bus asks again (A-222). The subscription is made at once when no follower made it, so a part that follows no // view of the seat still hears the event; adopted before Start, it is made at Start. func (r *ProcessReader) AdoptEvent(seat, event string, hear func(context.Context, []byte) error) error { if seat == "" || event == "" || hear == nil { @@ -365,12 +365,17 @@ func (r *ProcessReader) Hear(ctx context.Context) func(jetstream.Msg) { } r.mu.Unlock() if hears != nil { + failed := false for _, h := range hears { if err := h.hear(ctx, m.Data()); err != nil { r.logf("views: %s is not taken: %v", m.Subject(), err) + // an event adopted with no picture is asked again by the bus (design 47 G A-222) + failed = failed || h.bucket == "" } } - _ = m.Ack() + if !failed { + _ = m.Ack() + } return } if owned != nil { diff --git a/go/views/views_test.go b/go/views/views_test.go index e0fcc69a..1deb04b1 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -820,6 +820,28 @@ func subViews(t *testing.T) { if len(hooked) != 2 || !am.acked { t.Errorf("a reader following no view heard %v, acked %v", hooked, am.acked) } + // A-222: an adopted event whose handler fails is left unacknowledged, so the bus asks again. + failing := &ProcessReader{Module: "reader", Now: now} + var tries int + if err := failing.AdoptEvent("s", "revision-said", func(context.Context, []byte) error { + tries++ + if tries == 1 { + return errors.New("not placed yet") + } + return nil + }); err != nil { + t.Fatal(err) + } + fm := &seatMsg{subject: "mesh.seat.s.event.revision-said.next", data: []byte(`{"revision":10}`)} + failing.Hear(context.Background())(fm) + if fm.acked { + t.Error("a failed adopted event was acknowledged") + } + fm2 := &seatMsg{subject: fm.subject, data: fm.data} + failing.Hear(context.Background())(fm2) + if !fm2.acked || tries != 2 { + t.Errorf("the event asked again: acked %v after %d tries", fm2.acked, tries) + } pf.WaitFor(context.Background(), "4") ended <- true if fd := pfs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" {