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

@@ -51,7 +51,17 @@ type Follower struct {
 	// during are the events heard while an ask runs: taken again after its answer, so none is lost to a picture read
 	// before it was written.
 	during []Event
-	ready  chan struct{}
+	// heardRev is the highest revision of the owner's state (made at heardCreated) any event handed to Apply carried,
+	// whether it was taken into the picture, passed over or heard while an ask ran: what the picture must reach, so a
+	// 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{}
 }
 
 // Status is what a reader's read tool says of one first view (design 25 § 12.2): the owner, the tool and the revision
@@ -69,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"`
@@ -98,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.
@@ -163,7 +184,14 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error
 		f.st.ByHand++
 	}
 	f.mu.Unlock()
-	defer func() { f.mu.Lock(); f.asking = false; f.mu.Unlock() }()
+	released := false
+	defer func() {
+		if !released {
+			f.mu.Lock()
+			f.asking = false
+			f.mu.Unlock()
+		}
+	}()
 	bound, wait := f.Bound, f.Wait
 	if bound == 0 {
 		bound = Bound
@@ -175,6 +203,23 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error
 	for {
 		err := f.askOnce(ctx)
 		if err == nil {
+			// An event heard that the answer does not hold yet (the owner answered before it wrote it, or the event came
+			// before the first view and was passed over, or it was a gap heard while this ask ran) leaves the picture
+			// behind. No further event is waited for: it is asked again on a goroutine of its own, decided in the same
+			// check that ends this ask, so an event heard after the check finds no ask running and asks itself (ADR 0615
+			// §5). The ask that returns has brought the picture as far as the owner answered; the module need not wait.
+			f.mu.Lock()
+			behind := f.behindLocked()
+			f.asking, released = false, true
+			start := behind != "" && !f.catching
+			if start {
+				f.catching = true
+			}
+			f.mu.Unlock()
+			if start {
+				f.logf("views: %s's %s is behind an event heard (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
+				go f.catchUp(ctx)
+			}
 			return nil
 		}
 		f.mu.Lock()
@@ -190,10 +235,7 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error
 			f.Raise(s)
 		}
 		f.say(s)
-		pause, cancel := context.WithTimeout(ctx, wait) // a context's bound between two asks, no timer of its own
-		<-pause.Done()
-		cancel()
-		if ctx.Err() != nil {
+		if !f.pause(ctx, wait) {
 			return ctx.Err()
 		}
 		if wait *= 2; wait > time.Minute {
@@ -202,6 +244,90 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error
 	}
 }
 
+// catchUp asks the first view once more, each time after a pause (the reader's Wait, doubling to a minute: an owner that
+// answers from before its last write is not asked in a hot loop), until the picture holds every revision heard. It
+// holds no ask while it waits, so a gap, an ask by hand or an event heard meanwhile is not held up by it.
+func (f *Follower) catchUp(ctx context.Context) {
+	wait := f.Wait
+	if wait == 0 {
+		wait = 5 * time.Second
+	}
+	for {
+		if !f.pause(ctx, wait) {
+			f.mu.Lock()
+			f.catching = false
+			f.mu.Unlock()
+			return
+		}
+		f.mu.Lock()
+		behind := f.behindLocked()
+		if behind == "" {
+			f.catching = false
+			f.mu.Unlock()
+			return
+		}
+		f.mu.Unlock()
+		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
+		}
+	}
+}
+
+// pause waits a context's bound between two asks, no timer of its own; false once ctx ended.
+func (f *Follower) pause(ctx context.Context, wait time.Duration) bool {
+	p, cancel := context.WithTimeout(ctx, wait)
+	<-p.Done()
+	cancel()
+	return ctx.Err() == nil
+}
+
+// behindLocked says why the picture is behind an event heard, or "" when it holds every revision heard. The caller
+// 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):
+		return fmt.Sprintf("the state was made again at %s, the picture holds the one made at %s", f.heardCreated.Format(time.RFC3339), f.st.Created.Format(time.RFC3339))
+	case f.heardCreated.Equal(f.st.Created) && f.heardRev > f.st.Revision:
+		return fmt.Sprintf("revision %d heard, the picture holds %d", f.heardRev, f.st.Revision)
+	}
+	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()
+	defer f.mu.Unlock()
+	f.init()
+	return f.behindLocked()
+}
+
 // askOnce asks every page and, once whole, takes it as the picture.
 func (f *Follower) askOnce(ctx context.Context) error {
 	f.mu.Lock()
@@ -282,6 +408,12 @@ func (f *Follower) askOnce(ctx context.Context) error {
 func (f *Follower) Apply(ev Event) (gap bool, why string) {
 	f.mu.Lock()
 	f.init()
+	switch {
+	case ev.Created.After(f.heardCreated):
+		f.heardCreated, f.heardRev = ev.Created, ev.Revision
+	case ev.Created.Equal(f.heardCreated) && ev.Revision > f.heardRev:
+		f.heardRev = ev.Revision
+	}
 	if f.asking {
 		f.during = append(f.during, ev)
 	}
@@ -289,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)
@@ -312,7 +447,9 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) {
 		f.values[ev.Key] = ev.Value
 	}
 	f.st.Revision = ev.Revision
+	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 {
 		f.Changed(ev.Key, ev.Value, ev.Deleted, ev.At)
 	}
@@ -343,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

@@ -6,6 +6,8 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -224,6 +226,101 @@ 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 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.
+			"a gap heard while an ask runs": func(f *Follower, owner *uint64) { *owner = 12; f.Apply(ev(12, "e", `{"n":5}`)) },
+			// An event heard before the first view, passed over, the owner answering from before its write.
+			"an event heard before the first view": func(f *Follower, owner *uint64) { f.Apply(ev(10, "e", `{"n":5}`)) },
+		} {
+			t.Run(name, func(t *testing.T) {
+				owner := uint64(9)
+				var f *Follower
+				asks := 0
+				f, _ = follower(func(context.Context, string) (Answer, error) {
+					asks++
+					a := Answer{V: Version, Owner: "o", State: "modules", Revision: owner, Created: made, Values: map[string]json.RawMessage{"e": json.RawMessage(`{"n":5}`)}}
+					if asks == 1 && name == "a gap heard while an ask runs" {
+						drop(f, &owner) // heard while this ask runs; the answer is the one read before it
+					}
+					if asks == 2 {
+						owner = 12 // the owner has written it by the second ask
+						a.Revision = owner
+					}
+					return a, nil
+				})
+				if name == "an event heard before the first view" {
+					drop(f, &owner)
+				}
+				caught := make(chan struct{})
+				var once sync.Once
+				f.Said = func(s Status) {
+					if s.Revision >= 10 {
+						once.Do(func() { close(caught) })
+					}
+				}
+				if err := f.Start(context.Background()); err != nil {
+					t.Fatal(err)
+				}
+				// No further event follows: the follower asks once more by itself and holds what was heard.
+				select {
+				case <-caught:
+				case <-time.After(10 * time.Second):
+					t.Fatal("the reader stayed behind the event it 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)
+		f.Said = func(s Status) { told = append(told, s.Revision) }
+		_ = f.Start(context.Background())
+		told = nil
+		f.Apply(ev(10, "c", `{"n":3}`))
+		if len(told) != 1 || told[0] != 10 {
+			t.Errorf("said revisions %v, want [10]", told)
+		}
+	})
 	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) {