From 4c8ebfba6dd59dbd94e8276b94aaa30a86093fe3 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 02:16:50 +0100 Subject: [PATCH 1/2] A reader that heard an event its picture lacks asks again by itself, and says the revision it holds as events arrive (issue 1966) A follower passed over an event heard before its first view and dropped a gap heard while an ask ran, so a quiet owner left the reader behind until a further event. It now keeps the highest revision heard and, when an ask ends short of it, asks again on a pause of its own until it holds it. The status a read tool shows followed only the asks, so it said a reader was behind that was not; it follows the events too. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- go/views/follow.go | 108 ++++++++++++++++++++++++++++++++++++++--- go/views/views_test.go | 61 +++++++++++++++++++++++ 2 files changed, 163 insertions(+), 6 deletions(-) diff --git a/go/views/follow.go b/go/views/follow.go index 11107a4..1d2519f 100644 --- a/go/views/follow.go +++ b/go/views/follow.go @@ -51,7 +51,14 @@ 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 + // 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 @@ -163,7 +170,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 +189,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 +221,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 +230,66 @@ 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.askUntil(ctx, "behind an event heard: "+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.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 "" +} + +// 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 +370,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) } @@ -312,7 +406,9 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) { f.values[ev.Key] = ev.Value } f.st.Revision = ev.Revision + s := 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) } diff --git a/go/views/views_test.go b/go/views/views_test.go index 272afb7..4c79227 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "strings" + "sync" "testing" "time" @@ -224,6 +225,66 @@ 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) { + // 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 || f.Behind() != "" || s.Asks != 2 || !strings.HasPrefix(s.Why, "behind an event heard") { + t.Errorf("revision %d after %d asks (%q), behind: %q", s.Revision, s.Asks, s.Why, f.Behind()) + } + }) + } + }) + 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) { From 38659c3937e74869fd513ce3b4d20fbd751de61c Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 02:26:10 +0100 Subject: [PATCH 2/2] 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 | 18 ++++++++++++ go/views/follow.go | 56 +++++++++++++++++++++++++++++++++----- go/views/process_reader.go | 5 ++++ go/views/status.go | 5 ++-- go/views/views_test.go | 40 +++++++++++++++++++++++++-- 5 files changed, 113 insertions(+), 11 deletions(-) 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)