A reader asks again when its consumer was made after its first view, says every silent drop, and carries what it heard in its status (issue 1966)

A consumer is made from now (ADR 0443): an event said before it was made is never delivered to it and no later event
shows the loss. The event hub tells a part when the bus made its consumer, and a follower asked before that asks
again. Events passed over, at or below the held revision, or of a state not followed are now said in the log. The
status carries the highest revision heard and why the picture is behind it.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i

go/process/event_hub.go

@@ -23,6 +23,7 @@ type EventHub struct {
 	mu     sync.RWMutex
 	routes []hubRoute
 	cons   jetstream.ConsumeContext
+	bound  []func(created time.Time)
 }
 
 type hubRoute struct {
@@ -56,6 +57,15 @@ func (h *EventHub) Handle(pattern string, handle func(jetstream.Msg)) error {
 	return nil
 }
 
+// OnBound registers f, told each time the hub binds the module's consumer, with when the bus made it. A consumer is made
+// from now (ADR 0443), so an event said before it was made is not delivered to it: a part that took a picture before
+// that must ask its owner again (views.Follower.ConsumerMade). Register before Start.
+func (h *EventHub) OnBound(f func(created time.Time)) {
+	h.mu.Lock()
+	h.bound = append(h.bound, f)
+	h.mu.Unlock()
+}
+
 // Patterns are the registered subject patterns, in order.
 func (h *EventHub) Patterns() []string {
 	h.mu.RLock()
@@ -83,6 +93,14 @@ func (h *EventHub) Start(ctx context.Context, js jetstream.JetStream, cred Accou
 	if err != nil {
 		return fmt.Errorf("the consumer %s on %s is not heard: %w", cred.EventConsumerName(), EventsStream, err)
 	}
+	if info, ierr := cons.Info(ctx); ierr != nil {
+		h.logf("events: the consumer %s on %s is heard, but its information is not read (%v): a first view taken before it was made is not asked again", cred.EventConsumerName(), EventsStream, ierr)
+	} else {
+		h.logf("events: the consumer %s on %s is heard; the bus made it at %s", cred.EventConsumerName(), EventsStream, info.Created.UTC().Format(time.RFC3339))
+		for _, f := range h.bound {
+			f(info.Created.UTC())
+		}
+	}
 	return nil
 }
 

go/views/follow.go

@@ -56,6 +56,9 @@ type Follower struct {
 	// reader that heard it knows it is behind without a further event.
 	heardRev     uint64
 	heardCreated time.Time
+	// consumerMade is when the bus made the reader's consumer of the owner's events (EventHub.OnBound): an event said
+	// before it is not delivered to it, so a picture asked before it is behind.
+	consumerMade time.Time
 	// catching is set while catchUp runs: one catch-up at a time.
 	catching bool
 	ready    chan struct{}
@@ -76,6 +79,11 @@ type Status struct {
 	Asks     int       `json:"asks"`
 	ByHand   int       `json:"by-hand"`
 	Raised   bool      `json:"raised"`
+	// Heard is the highest revision of the owner's state any event heard carried (made at HeardCreated), taken into the
+	// picture or not; Behind says why the picture is short of it, or of its consumer's making, and an ask for it runs.
+	Heard        uint64    `json:"heard,omitempty"`
+	HeardCreated time.Time `json:"heard-created,omitzero"`
+	Behind       string    `json:"behind,omitempty"`
 	// Why is the last reason it asked (start, a gap and which, a hand act) or why the last ask failed.
 	Why   string `json:"why,omitempty"`
 	Error string `json:"error,omitempty"`
@@ -105,20 +113,26 @@ func (f *Follower) init() {
 func (f *Follower) say(s Status) {
 	if f.Said != nil {
 		f.mu.Lock()
-		s.Keys = len(f.values)
+		s = f.statusLocked(s)
 		f.mu.Unlock()
 		f.Said(s)
 	}
 }
 
+// statusLocked is s with what the follower holds beside it: its keys, the highest revision heard and why it is behind.
+// The caller holds mu.
+func (f *Follower) statusLocked(s Status) Status {
+	s.Keys = len(f.values)
+	s.Heard, s.HeardCreated, s.Behind = f.heardRev, f.heardCreated, f.behindLocked()
+	return s
+}
+
 // Status is the follower's status now.
 func (f *Follower) Status() Status {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.init()
-	s := f.st
-	s.Keys = len(f.values)
-	return s
+	return f.statusLocked(f.st)
 }
 
 // Ready is closed once a first view was answered.
@@ -253,7 +267,8 @@ func (f *Follower) catchUp(ctx context.Context) {
 			return
 		}
 		f.mu.Unlock()
-		_ = f.askUntil(ctx, "behind an event heard: "+behind, false)
+		f.logf("views: %s's %s is still behind (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
+		_ = f.askUntil(ctx, "behind: "+behind, false)
 		if wait *= 2; wait > time.Minute {
 			wait = time.Minute
 		}
@@ -272,6 +287,8 @@ func (f *Follower) pause(ctx context.Context, wait time.Duration) bool {
 // holds mu.
 func (f *Follower) behindLocked() string {
 	switch {
+	case !f.consumerMade.IsZero() && f.consumerMade.After(f.st.Asked) && !f.st.Asked.IsZero():
+		return fmt.Sprintf("the consumer of its events was made at %s, after the first view was asked at %s: what was said between is not delivered", f.consumerMade.Format(time.RFC3339), f.st.Asked.Format(time.RFC3339))
 	case f.heardCreated.IsZero():
 		return ""
 	case f.heardCreated.After(f.st.Created):
@@ -282,6 +299,27 @@ func (f *Follower) behindLocked() string {
 	return ""
 }
 
+// ConsumerMade tells the follower when the bus made its consumer of the owner's events. A picture asked before that may
+// lack an event said between its ask and the consumer, which the consumer never delivers and no later event shows: it
+// asks again, by itself (ADR 0443: a consumer is made from now). It does nothing before a first view was asked.
+func (f *Follower) ConsumerMade(ctx context.Context, created time.Time) {
+	f.mu.Lock()
+	f.init()
+	if created.After(f.consumerMade) {
+		f.consumerMade = created
+	}
+	behind := f.behindLocked()
+	start := behind != "" && !f.asking && !f.catching
+	if start {
+		f.catching = true
+	}
+	f.mu.Unlock()
+	if start {
+		f.logf("views: %s's %s is behind (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
+		go f.catchUp(ctx)
+	}
+}
+
 // Behind says whether an event heard is not in the picture yet, and why: for a reader's tool and a test.
 func (f *Follower) Behind() string {
 	f.mu.Lock()
@@ -383,13 +421,16 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) {
 	case <-f.ready:
 	default:
 		f.mu.Unlock()
+		f.logf("views: %s's %s at revision %d was heard before its first view was answered: passed over, the answer holds it or is asked again", f.Owner, f.State, ev.Revision)
 		return false, ""
 	}
 	switch {
 	case !ev.Created.Equal(f.st.Created):
 		why = fmt.Sprintf("%s's %s was made again (or not marked when the first view was answered): created %s, held %s", f.Owner, f.State, ev.Created.Format(time.RFC3339), f.st.Created.Format(time.RFC3339))
 	case ev.Revision <= f.st.Revision:
+		held := f.st.Revision
 		f.mu.Unlock()
+		f.logf("views: %s's %s at revision %d is in the picture already (it holds %d): passed over", f.Owner, f.State, ev.Revision, held)
 		return false, ""
 	case ev.Revision > f.st.Revision+1:
 		why = fmt.Sprintf("%s's %s at revision %d, the picture holds %d: %d changes missed", f.Owner, f.State, ev.Revision, f.st.Revision, ev.Revision-f.st.Revision-1)
@@ -406,7 +447,7 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) {
 		f.values[ev.Key] = ev.Value
 	}
 	f.st.Revision = ev.Revision
-	s := f.st
+	s := f.statusLocked(f.st)
 	f.mu.Unlock()
 	f.say(s) // the revision the picture holds follows the events, not only the asks (a read tool says where it stands)
 	if f.Changed != nil {
@@ -439,7 +480,8 @@ func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte,
 	}
 	f := followers[ev.State]
 	if f == nil {
-		return nil // a state this reader does not follow
+		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
 	}
 	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/views/process_reader.go

@@ -545,6 +545,11 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred
 	if hub == nil {
 		hub, pattern, own = process.NewEventHub(r.logf), ">", true
 	}
+	hub.OnBound(func(created time.Time) {
+		for _, f := range r.Followers() {
+			f.ConsumerMade(ctx, created)
+		}
+	})
 	err := hub.Handle(pattern, r.Hear(ctx))
 	if err == nil && own {
 		err = hub.Start(ctx, js, cred)

go/views/status.go

@@ -99,8 +99,9 @@ func StatusTools(reader, read, again string, st stdio.KeptState, acts HandActs,
 	return []stdio.Tool{
 		{Name: read,
 			Description: fmt.Sprintf("%s's first views of the owners' states it follows (novox/hq ADR 0615): per owner and state, "+
-				"the tool it asks, the revision its picture held when last answered, when it asked and was answered, how many asks "+
-				"(by hand among them) and whether its condition (unanswered past two minutes) is raised.", reader),
+				"the tool it asks, the revision its picture holds now (events taken in included), the highest revision it heard and why "+
+				"it is behind it, if it is, when it asked and was answered, how many asks (by hand among them) and whether its "+
+				"condition (unanswered past two minutes) is raised.", reader),
 			Input: map[string]any{"type": "object", "properties": map[string]any{}},
 			Run: func(map[string]any) (any, error) {
 				keys, err := st.Keys()

go/views/views_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -225,7 +226,7 @@ func TestViews(t *testing.T) {
 			t.Errorf("d = %s, %v at %d", v, ok, f.Status().Revision)
 		}
 	})
-	t.Run("a reader that heard an event its answer does not hold asks again at once, with no further event", func(t *testing.T) {
+	t.Run("a reader that heard an event its answer does not hold asks again by itself on its Wait, with no further event", func(t *testing.T) {
 		// The owner's state: revision 9 at first, 12 once its writer is done. The picture a first view reads lags the events.
 		for name, drop := range map[string]func(f *Follower, owner *uint64){
 			// A gap heard while the ask runs: the running ask swallowed the gap's own ask and the event was not consecutive.
@@ -268,12 +269,47 @@ func TestViews(t *testing.T) {
 				case <-time.After(10 * time.Second):
 					t.Fatal("the reader stayed behind the event it heard")
 				}
-				if s := f.Status(); s.Revision < 10 || f.Behind() != "" || s.Asks != 2 || !strings.HasPrefix(s.Why, "behind an event heard") {
+				if s := f.Status(); s.Revision < 10 || s.Heard < 10 || s.Behind != "" || f.Behind() != "" || s.Asks != 2 || !strings.HasPrefix(s.Why, "behind") {
 					t.Errorf("revision %d after %d asks (%q), behind: %q", s.Revision, s.Asks, s.Why, f.Behind())
 				}
 			})
 		}
 	})
+	t.Run("a consumer made after the first view was asked asks it again by itself, and the status says why until then", func(t *testing.T) {
+		asked := make(chan struct{}, 8)
+		var f *Follower
+		f, _ = follower(func(ctx context.Context, after string) (Answer, error) {
+			asked <- struct{}{}
+			return answer(ctx, after)
+		})
+		var said []Status
+		var mu sync.Mutex
+		var clock atomic.Int64
+		clock.Store(int64(time.Hour))
+		f.Now = func() time.Time { return made.Add(time.Duration(clock.Load())) }
+		f.Said = func(s Status) { mu.Lock(); said = append(said, s); mu.Unlock() }
+		_ = f.Start(context.Background())
+		<-asked
+		// The bus made the consumer after the ask: an event said between is never delivered to it.
+		f.ConsumerMade(context.Background(), made.Add(2*time.Hour))
+		if b := f.Status().Behind; !strings.Contains(b, "consumer") {
+			t.Errorf("behind = %q", b)
+		}
+		clock.Store(int64(3 * time.Hour)) // a later ask is after the consumer's making
+		select {
+		case <-asked:
+		case <-time.After(10 * time.Second):
+			t.Fatal("a consumer made after the first view asked nothing")
+		}
+	})
+	t.Run("an event heard but not in the picture is in the status as heard, with why it is behind", func(t *testing.T) {
+		f, _ := follower(answer)
+		_ = f.Start(context.Background())
+		f.Apply(ev(12, "a", `{}`)) // a gap: heard, not taken
+		if s := f.Status(); s.Heard != 12 || !strings.Contains(s.Behind, "12") {
+			t.Errorf("status = %+v", s)
+		}
+	})
 	t.Run("an event taken into the picture is said in the status, so a read tool shows the revision held", func(t *testing.T) {
 		var told []uint64
 		f, _ := follower(answer)