From f98e1594463ede6d28f0c67df66e2e7b9b3d7dfc Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:15:59 +0100 Subject: [PATCH] DLV-F3r sdk: a reader hears a views event by its data, and a gap is asked as a gap (A7's request) views.HearEvent routes one event given as its data, for the tool runner, which takes the owners' events by its own path; Follower.Gap asks once more without counting it by hand. Hear is HearEvent on a consumer's message. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- go/views/follow.go | 44 ++++++++++++++++++++++++++---------------- go/views/views_test.go | 24 +++++++++++++++++++++++ 2 files changed, 51 insertions(+), 17 deletions(-) diff --git a/go/views/follow.go b/go/views/follow.go index 2c0a470..83c7fdc 100644 --- a/go/views/follow.go +++ b/go/views/follow.go @@ -317,31 +317,41 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) { // that follower's first view once more on its own goroutine, said in the log. A message it cannot read is terminated. func Hear(ctx context.Context, followers map[string]*Follower, logf func(string, ...any)) func(jetstream.Msg) { return func(m jetstream.Msg) { - ev, err := DecodeEvent(m.Data()) - if err != nil { + if err := HearEvent(ctx, followers, m.Data(), logf); err != nil { logf("views: %s is not a views event, dropped: %v", m.Subject(), err) _ = m.Term() return } - f := followers[ev.State] - if f == nil { - _ = m.Ack() // a state this reader does not follow - return - } - if gap, why := f.Apply(ev); gap { - logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool) - go func() { - f.mu.Lock() - f.init() - f.st.Why = "a gap: " + why - f.mu.Unlock() - _ = f.askUntil(ctx, "a gap: "+why, false) - }() - } _ = m.Ack() } } +// 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 +// 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) + if err != nil { + return err + } + f := followers[ev.State] + if f == nil { + return nil // a state this reader does not follow + } + if gap, why := f.Apply(ev); gap { + logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool) + go func() { _ = f.Gap(ctx, why) }() + } + return nil +} + +// Gap asks the first view once more for a gap the reader saw, until answered or ctx ends: counted among its asks, +// never among its asks by hand. +func (f *Follower) Gap(ctx context.Context, why string) error { + return f.askUntil(ctx, "a gap: "+why, false) +} + // AskVia is a Follower's Ask through a process's call of the owner's tool (process.Conn.Call), decoding its answer. func AskVia(call func(key string, body any) (json.RawMessage, error), tool string) func(context.Context, string) (Answer, error) { return func(_ context.Context, after string) (Answer, error) { diff --git a/go/views/views_test.go b/go/views/views_test.go index da65aad..b8867b5 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -188,6 +188,30 @@ func TestViews(t *testing.T) { t.Errorf("d = %s, %v at %d", v, ok, f.Status().Revision) } }) + t.Run("an event heard by its data asks a gap once more as a gap, never by hand", func(t *testing.T) { + asked := make(chan struct{}, 4) + f, _ := follower(func(ctx context.Context, after string) (Answer, error) { + asked <- struct{}{} + return answer(ctx, after) + }) + _ = f.Start(context.Background()) + <-asked + raw, _ := ev(12, "a", `{}`).Encode() + if err := HearEvent(context.Background(), map[string]*Follower{"modules": f}, raw, t.Logf); err != nil { + t.Fatal(err) + } + select { + case <-asked: + case <-time.After(3 * time.Second): + t.Fatal("a gap heard by its data asked nothing") + } + if s := f.Status(); s.Asks != 2 || s.ByHand != 0 { + t.Errorf("a gap was counted by hand: %+v", s) + } + if err := HearEvent(context.Background(), nil, []byte("not an event"), t.Logf); err == nil { + t.Error("data that is no views event was taken") + } + }) t.Run("an event names its state on its subject, and decodes as it was encoded", func(t *testing.T) { if s := StateOfEvent("mesh.stream.mesh-delivery-indexer.event.views.config-candidates"); s != "config-candidates" { t.Errorf("state of subject = %q", s)