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/catalogue/seat_pass.go

@@ -1,14 +1,97 @@
 package catalogue
 
-// An assignment as the passing of a seat reads it: a module on a machine, with its manifest.
+import (
+	"fmt"
+	"sort"
+)
+
+// An assignment as the passing of a seat reads it: a module on a machine, with the manifest of the build that machine
+// runs, else the trunk's (novox/hq ADR 0660 §2).
 type Assignment struct {
 	Node     string
 	Manifest Manifest
+	// Unread is an assignment whose manifest could not be read at all: it is still assigned, so its holdings stand,
+	// and it claims nothing.
+	Unread bool
 }
 
-// SeatPass is one seat whose holder on record is assigned nowhere, and the only assignment that can take it.
+// SeatPass is one seat whose holder on record has left it, and the only assignment that can take it.
 type SeatPass struct {
 	Seat Seat
 	From Held
 	To   Held
 }
+
+// SeatsToPassOn reads the mesh for seats held once whose holder on record has left (novox/hq ADR 0449 §4, ADR 0660).
+//
+// **A holder on record that is assigned on no machine, or whose every assignment's manifest no longer claims the seat
+// (or claims it and cannot hold it, by CanHold), holds nothing.** Of such a seat, exactly one assignment in the mesh
+// able to hold it takes it over: that is a pass, and the caller applies it and says it. An assignment is a module on
+// one node, so one module placed on two nodes is two claimants. With no such assignment the record stays; with two or
+// more nothing is passed and the answer says so, naming two and the `seat --to` that settles it. A holder with an
+// assignment whose manifest still claims the seat changes nothing (ADR 0131), and so does one whose manifest could not
+// be read. Only mesh seats held once are read: a replicated seat's holders are added by an act (ADR 0223), a kinded
+// bench is held by its claims, and a seat with several holders on record is a store that disagrees with the mesh.
+// Nothing here records a pass (ADR 0660 §4).
+func SeatsToPassOn(holdings []Held, assignments []Assignment) ([]SeatPass, []string) {
+	bySeat := map[string][]Held{}
+	var order []string
+	for _, h := range holdings {
+		s, ok := SeatNamed(h.Claim)
+		if !ok || s.Replicated || s.Scope != ScopeMesh || h.Scope != ScopeMesh || KindedBenches[s.Name] {
+			continue
+		}
+		if _, seen := bySeat[s.Name]; !seen {
+			order = append(order, s.Name)
+		}
+		bySeat[s.Name] = append(bySeat[s.Name], h)
+	}
+	sort.Strings(order)
+
+	var passes []SeatPass
+	var said []string
+	for _, name := range order {
+		recs := bySeat[name]
+		if len(recs) != 1 {
+			continue
+		}
+		seat, _ := SeatNamed(name)
+		holds := false
+		for _, a := range assignments {
+			if a.Manifest.Module != recs[0].Module {
+				continue
+			}
+			if a.Unread || CanHold(a.Manifest, seat) == nil {
+				holds = true
+			}
+		}
+		if holds {
+			continue
+		}
+		var claimants []Held
+		for _, a := range assignments {
+			if a.Unread || CanHold(a.Manifest, seat) != nil {
+				continue
+			}
+			claimants = append(claimants, Held{Claim: seat.Name, Scope: seat.Scope, Node: a.Node, Module: a.Manifest.Module})
+		}
+		sort.Slice(claimants, func(i, j int) bool {
+			if claimants[i].Node != claimants[j].Node {
+				return claimants[i].Node < claimants[j].Node
+			}
+			return claimants[i].Module < claimants[j].Module
+		})
+		switch len(claimants) {
+		case 0:
+		case 1:
+			passes = append(passes, SeatPass{Seat: seat, From: recs[0], To: claimants[0]})
+		default:
+			said = append(said, fmt.Sprintf(
+				"%s on %s is on record as the holder of %q and holds it nowhere, and %s on %s and %s on %s both claim it "+
+					"— nothing is passed; `seat %s --to <node>/<module>` records the one that holds it",
+				recs[0].Module, recs[0].Node, seat.Name, claimants[0].Module, claimants[0].Node,
+				claimants[1].Module, claimants[1].Node, seat.Name))
+		}
+	}
+	return passes, said
+}

go/catalogue/seat_pass_test.go

@@ -0,0 +1,96 @@
+package catalogue
+
+import (
+	"strings"
+	"testing"
+)
+
+func passSeat(t *testing.T) {
+	t.Helper()
+	was := Seats()
+	t.Cleanup(func() { UseSeats(was) })
+	UseSeats([]Seat{
+		{Name: "mesh-broker", Scope: ScopeMesh, Delivers: "mesh-bus", Decision: "test"},
+		{Name: "mesh-dns-resolver", Scope: ScopeMesh, Replicated: true, Decision: "test"},
+	})
+}
+
+func broker(module string) Manifest {
+	return Manifest{Module: module, Provides: []Offer{{Name: "mesh-bus", Scope: ScopeMesh}},
+		Claims: []Claim{{Name: "mesh-broker", Scope: ScopeMesh}}}
+}
+
+// A module that still runs but no longer claims the seat.
+func dropped(module string) Manifest { return Manifest{Module: module} }
+
+func recorded(node, module string) []Held {
+	return []Held{{Claim: "mesh-broker", Scope: ScopeMesh, Node: node, Module: module}}
+}
+
+// novox/hq ADR 0660 "How it is checked", and ADR 0449 §4's original case.
+func TestSeatsToPassOn(t *testing.T) {
+	passSeat(t)
+	rec := recorded("novox", "old")
+	cases := []struct {
+		name  string
+		rec   []Held
+		as    []Assignment
+		to    string // "node/module" of the one pass, "" for none
+		saidW []string
+	}{
+		{"holder assigned nowhere, one claimant (ADR 0449 §4)", rec,
+			[]Assignment{{Node: "novox", Manifest: broker("new")}}, "novox/new", nil},
+		{"holder assigned but its manifest dropped the claim", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "novox", Manifest: broker("new")}}, "novox/new", nil},
+		{"holder claims but cannot hold the seat", rec,
+			[]Assignment{{Node: "novox", Manifest: Manifest{Module: "old", Claims: []Claim{{Name: "mesh-broker", Scope: ScopeMesh}}}},
+				{Node: "ace", Manifest: broker("new")}}, "ace/new", nil},
+		{"the running build still claims: the holder keeps the seat", rec,
+			[]Assignment{{Node: "novox", Manifest: broker("old")}, {Node: "novox", Manifest: broker("new")}}, "", nil},
+		{"one of the holder's two nodes still claims it", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "ace", Manifest: broker("old")},
+				{Node: "g14", Manifest: broker("new")}}, "", nil},
+		{"holder's manifest unread: still assigned", rec,
+			[]Assignment{{Node: "novox", Manifest: Manifest{Module: "old"}, Unread: true}, {Node: "novox", Manifest: broker("new")}}, "", nil},
+		{"an unread assignment is no claimant", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "ace", Manifest: Manifest{Module: "new"}, Unread: true}}, "", nil},
+		{"a claimant whose running manifest does not claim is not counted", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "ace", Manifest: dropped("new")}}, "", nil},
+		{"no claimant keeps the record", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "novox", Manifest: dropped("other")}}, "", nil},
+		{"holder gone, no claimant keeps the record", rec, nil, "", nil},
+		{"one module on two nodes is two claimants: none passed, said", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "ace", Manifest: broker("new")}, {Node: "novox", Manifest: broker("new")}},
+			"", []string{"old", "new on ace", "new on novox", "seat mesh-broker --to <node>/<module>"}},
+		{"two claimants, holder gone: said", rec,
+			[]Assignment{{Node: "novox", Manifest: broker("a")}, {Node: "ace", Manifest: broker("b")}},
+			"", []string{"a on novox", "b on ace", "seat mesh-broker --to <node>/<module>"}},
+		{"a seat with two holders on record is not read", append(rec[:1:1], Held{Claim: "mesh-broker", Scope: ScopeMesh, Node: "ace", Module: "x"}),
+			[]Assignment{{Node: "novox", Manifest: broker("new")}}, "", nil},
+		{"a replicated seat is never passed",
+			[]Held{{Claim: "mesh-dns-resolver", Scope: ScopeMesh, Node: "novox", Module: "gone"}},
+			[]Assignment{{Node: "novox", Manifest: Manifest{Module: "r", Claims: []Claim{{Name: "mesh-dns-resolver", Scope: ScopeMesh}}}}}, "", nil},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			passes, said := SeatsToPassOn(c.rec, c.as)
+			if c.to == "" && len(passes) != 0 || c.to != "" && len(passes) != 1 {
+				t.Fatalf("passes=%+v said=%v", passes, said)
+			}
+			if c.to != "" {
+				p := passes[0]
+				if p.Seat.Name != "mesh-broker" || p.From.Module != "old" || p.To.Node+"/"+p.To.Module != c.to {
+					t.Fatalf("the pass is %+v", p)
+				}
+			}
+			if (len(c.saidW) == 0) != (len(said) == 0) {
+				t.Fatalf("said=%v, want %v", said, c.saidW)
+			}
+			for _, w := range c.saidW {
+				if !strings.Contains(said[0], w) {
+					t.Fatalf("the answer does not say %q: %s", w, said[0])
+				}
+			}
+		})
+	}
+}

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) {