diff --git a/go/process/event_hub.go b/go/process/event_hub.go index df8537e..14fe1a4 100644 --- a/go/process/event_hub.go +++ b/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 } diff --git a/go/views/follow.go b/go/views/follow.go index 1d2519f..8e580e9 100644 --- a/go/views/follow.go +++ b/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) diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 9cc1d74..20601f5 100644 --- a/go/views/process_reader.go +++ b/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) diff --git a/go/views/status.go b/go/views/status.go index 43bd66b..962bde8 100644 --- a/go/views/status.go +++ b/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() diff --git a/go/views/views_test.go b/go/views/views_test.go index 4c79227..7972a31 100644 --- a/go/views/views_test.go +++ b/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)