From 04b323868fc7c4b2484a01c342d150f1b70c944b Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:15:55 +0100 Subject: [PATCH] Keep a views reader's copy equal to its owner's: no false gaps, a lost last change caught up (issue 1966) The owner's watch of a state keeping one value per key is never handed a revision that a later write of the same key replaced first, so every such skip read as "changes missed" and the reader asked its first view again, over and over in a burst. Each event now names the revision said before it (prev), and a reader takes it when its picture holds that one. A change whose event never reached a reader was only noticed at a later change, and a lost last one never. The owner now says its mark as an event each time it writes it; a reader short of it asks its first view at once and holds what it read, never the event's body. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/views/emit.go | 34 ++++++-- go/views/follow.go | 12 ++- go/views/process_reader_integration_test.go | 38 +++++++++ go/views/views.go | 33 ++++++-- go/views/views_test.go | 91 ++++++++++++++++++++- 5 files changed, 186 insertions(+), 22 deletions(-) diff --git a/go/views/emit.go b/go/views/emit.go index 8119ba3..67ff31d 100644 --- a/go/views/emit.go +++ b/go/views/emit.go @@ -89,6 +89,7 @@ func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, if m, ok := e.mark(ctx, state); ok && m.Created.Equal(made) && m.Revision <= last { from = m.Revision + 1 } else { + // Marked at the end with nothing said between: a reader behind it hears the mark and asks its first view. e.writeMark(ctx, state, Mark{State: state, Revision: last, Created: made, Fill: fill}) } w, err := kv.WatchAll(ctx, jetstream.ResumeFromRevision(from)) @@ -101,15 +102,16 @@ func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, } go func() { defer w.Stop() - e.follow(ctx, state, made, w.Updates(), fillNow, errs) + e.follow(ctx, state, made, from-1, w.Updates(), fillNow, errs) }() return nil } -// follow says each change handed on updates in order, and marks once every change handed so far is said. A change -// that fails to be said is said again, the same change on a backoff (Wait), until it is said or ctx ends: the mark -// never moves past a change not said, so no reader misses one (design 25 §12.2). -func (e Emitter) follow(ctx context.Context, state string, made time.Time, updates <-chan jetstream.KeyValueEntry, fillNow func() (Fill, error), errs chan<- error) { +// follow says each change handed on updates in order, each naming the revision said before it (prev, from the mark it +// starts after), and marks once every change handed so far is said. A change that fails to be said is said again, the +// same change on a backoff (Wait), until it is said or ctx ends: the mark never moves past a change not said, so no +// reader misses one (design 25 §12.2). +func (e Emitter) follow(ctx context.Context, state string, made time.Time, prev uint64, updates <-chan jetstream.KeyValueEntry, fillNow func() (Fill, error), errs chan<- error) { said := uint64(0) raised := false // A finding a run before this one raised is not known to this one: the first change said clears it once, so a @@ -129,7 +131,7 @@ func (e Emitter) follow(ctx context.Context, state string, made time.Time, updat if u != nil { var first time.Time for attempt := 1; ; attempt++ { - err := e.say(ctx, state, made, u) + err := e.say(ctx, state, made, prev, u) if err == nil { break } @@ -149,7 +151,7 @@ func (e Emitter) follow(ctx context.Context, state string, made time.Time, updat raised = !e.clearUnsaid(state) leftover = false } - said = u.Revision() + said, prev = u.Revision(), u.Revision() } // Marked once every change handed so far is said, so a busy state writes one mark per burst. if said > 0 && len(updates) == 0 { @@ -236,8 +238,13 @@ func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry) return ev } -func (e Emitter) say(ctx context.Context, state string, made time.Time, u jetstream.KeyValueEntry) error { +func (e Emitter) say(ctx context.Context, state string, made time.Time, prev uint64, u jetstream.KeyValueEntry) error { ev := e.Event(state, made, u) + ev.Prev = prev + return e.publish(ctx, state, ev) +} + +func (e Emitter) publish(ctx context.Context, state string, ev Event) error { data, err := ev.Encode() if err != nil { return err @@ -282,5 +289,16 @@ func (e Emitter) writeMark(ctx context.Context, state string, m Mark) { } if _, err := e.Marks.Put(ctx, state, data); err != nil { e.logf("views: the mark of %s at %d is not written (%v): a first view answers the one before", state, m.Revision, err) + return + } + // The mark said as an event of its own, after it is written: a reader whose picture is short of it (a change's event + // that never reached it) asks its first view now, which answers at this mark or later. Not said again on a failure: + // the next mark says it. A mark at 0 (a state with no change yet) is no event. + if m.Revision == 0 { + return + } + ev := Event{V: Version, Owner: e.Owner, State: state, Revision: m.Revision, Created: m.Created, At: m.At, Mark: true} + if err := e.publish(ctx, state, ev); err != nil { + e.logf("views: the mark of %s at %d is written but not said (%v): a reader short of it asks at the next change", state, m.Revision, err) } } diff --git a/go/views/follow.go b/go/views/follow.go index 8e580e9..ebdd3df 100644 --- a/go/views/follow.go +++ b/go/views/follow.go @@ -364,7 +364,7 @@ func (f *Follower) askOnce(ctx context.Context) error { f.during = nil slices.SortFunc(during, func(a, b Event) int { return cmp.Compare(a.Revision, b.Revision) }) for _, ev := range during { - if ev.Created.Equal(f.st.Created) && ev.Revision == f.st.Revision+1 && !ev.Gap { + if ev.Created.Equal(f.st.Created) && ev.Revision > f.st.Revision && ev.before() <= f.st.Revision && !ev.Gap && !ev.Mark { if ev.Deleted { delete(f.values, ev.Key) } else { @@ -430,10 +430,14 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) { 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) + if !ev.Mark { + 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) + case ev.Mark: + why = fmt.Sprintf("%s said its %s up to revision %d, the picture holds %d: a change's event was not heard", f.Owner, f.State, ev.Revision, f.st.Revision) + case ev.before() > f.st.Revision: + why = fmt.Sprintf("%s's %s at revision %d was said after %d, the picture holds %d: changes missed", f.Owner, f.State, ev.Revision, ev.before(), f.st.Revision) case ev.Gap: why = fmt.Sprintf("%s's %s.%s at revision %d was too large to carry", f.Owner, f.State, ev.Key, ev.Revision) } diff --git a/go/views/process_reader_integration_test.go b/go/views/process_reader_integration_test.go index 55f0c41..9f73c43 100644 --- a/go/views/process_reader_integration_test.go +++ b/go/views/process_reader_integration_test.go @@ -5,6 +5,7 @@ package views_test import ( "context" "encoding/json" + "fmt" "testing" "time" @@ -77,6 +78,43 @@ func TestAReaderHearsAnOwnersChangesAsOrdinaryEvents(t *testing.T) { if got := <-other; got != "mesh.mod.o.event.something-else" { t.Fatalf("other got %s", got) } + // Across bursts that write the same keys over and over, the reader's copy stays its owner's, by the events alone: a + // revision written over before the owner's watch was handed it is no gap, and the owner's mark after each burst asks + // nothing of a reader that holds it (issue 1966). + all, err := pic.WatchAll(ctx) + if err != nil { + t.Fatal(err) + } + var last uint64 + for round := range 300 { + for range 3 { + if last, err = kv.Put(ctx, fmt.Sprintf("k%d", round%5), fmt.Appendf(nil, `{"round":%d}`, round)); err != nil { + t.Fatal(err) + } + } + } + f := r.Followers()[0] + wait, stop := context.WithTimeout(ctx, 20*time.Second) + defer stop() + for f.Status().Revision < last { + select { + case <-all.Updates(): + case <-wait.Done(): + t.Fatalf("the reader stayed behind its owner at %d: %+v", last, f.Status()) + } + } + for i := range 5 { + e, err := kv.Get(ctx, fmt.Sprintf("k%d", i)) + if err != nil { + t.Fatal(err) + } + if v, _ := f.Get(e.Key()); string(v) != string(e.Value()) { + t.Errorf("%s: the reader holds %s, its owner %s", e.Key(), v, e.Value()) + } + } + if s := f.Status(); s.Asks != 1 { + t.Errorf("the reader asked its first view %d times across the bursts (%s): a revision written over was taken as a gap", s.Asks, s.Why) + } } // A module has one durable consumer on the shared events stream, so the loops that hear events in one module must share diff --git a/go/views/views.go b/go/views/views.go index d5f0c06..d698904 100644 --- a/go/views/views.go +++ b/go/views/views.go @@ -96,20 +96,41 @@ type Event struct { At time.Time `json:"at,omitzero"` // Gap says the value was too large to carry (MaxValueBytes): the reader asks its first view once. Gap bool `json:"gap,omitempty"` + // Prev is the revision of the change the owner said before this one, or of its mark when this is the first said + // since its start. A revision between the two was written over by a later change of the same key before the bus + // handed it to the owner's watch (a state keeps one value per key), so nothing of it is missed: a reader takes the + // event when its picture holds Prev or more. Zero says nothing, and a reader takes Revision-1 as the one before. + Prev uint64 `json:"prev,omitempty"` + // Mark says the event is the owner's mark (MarksState), said each time the mark is written: no key and no value, + // Revision the change the owner said its state up to. A reader whose picture is short of it asks its first view, so a + // change whose event never reached it is caught up without waiting for a later change. + Mark bool `json:"mark,omitempty"` +} + +// before is the revision of the change said before e: Prev, or Revision-1 when the owner said none. +func (e Event) before() uint64 { + if e.Prev != 0 { + return e.Prev + } + return e.Revision - 1 } // ID is the event's id on the bus, so the same change said twice (an owner's restart) is kept once. func (e Event) ID() string { - return fmt.Sprintf("%s-%s-%d-%d", e.Owner, e.State, e.Created.UnixNano(), e.Revision) + id := fmt.Sprintf("%s-%s-%d-%d", e.Owner, e.State, e.Created.UnixNano(), e.Revision) + if e.Mark { + id += "-mark" + } + return id } // Encode writes an event. func (e Event) Encode() ([]byte, error) { e.V = Version - if e.Owner == "" || e.State == "" || e.Key == "" || e.Revision == 0 { - return nil, errors.New("views: an event names its owner, state, key and revision") + if e.Owner == "" || e.State == "" || (e.Key == "" && !e.Mark) || e.Revision == 0 { + return nil, errors.New("views: an event names its owner, state, key (or is a mark) and revision") } - if e.Deleted || e.Gap { + if e.Deleted || e.Gap || e.Mark { e.Value = nil } return json.Marshal(e) @@ -121,8 +142,8 @@ func DecodeEvent(data []byte) (Event, error) { if err := decode(data, &e); err != nil { return Event{}, err } - if e.State == "" || e.Key == "" || e.Revision == 0 { - return Event{}, errors.New("views: an event names no state, key or revision") + if e.State == "" || (e.Key == "" && !e.Mark) || e.Revision == 0 { + return Event{}, errors.New("views: an event names no state, key (nor is a mark) or revision") } return e, nil } diff --git a/go/views/views_test.go b/go/views/views_test.go index f496f51..fabddf1 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -233,6 +233,56 @@ func TestViews(t *testing.T) { } } }) + t.Run("a change said after revisions written over is no gap; one said after a revision the picture lacks, or a mark past the picture, asks once more and changes nothing", func(t *testing.T) { + f, _ := follower(answer) + _ = f.Start(context.Background()) + over := ev(12, "a", `{"n":5}`) + over.Prev = 9 // 10 and 11 were written over before the owner's watch was handed them + if gap, why := f.Apply(over); gap || f.Status().Revision != 12 { + t.Fatalf("a change said after 9 on a picture at 9: gap %v (%s), at %d", gap, why, f.Status().Revision) + } + missed := ev(14, "b", `{"n":6}`) + missed.Prev = 13 + mark := func(rev uint64) Event { + return Event{V: Version, Owner: "o", State: "modules", Revision: rev, Created: made, Mark: true} + } + for _, c := range []struct { + e Event + gap bool + }{{missed, true}, {mark(12), false}, {mark(11), false}, {mark(13), true}} { + if gap, why := f.Apply(c.e); gap != c.gap { + t.Errorf("revision %d (mark %v) on a picture at 12: gap %v, want %v (%s)", c.e.Revision, c.e.Mark, gap, c.gap, why) + } + } + if v, _ := f.Get("a"); string(v) != `{"n":5}` || f.Status().Revision != 12 || len(f.Values()) != 2 { + t.Errorf("a gap or a mark changed the picture: %v at %d", f.Values(), f.Status().Revision) + } + }) + t.Run("a reader that never heard the owner's last change asks its first view on the owner's mark alone, and holds what it read", func(t *testing.T) { + later := mapSource{"a": values["a"], "b": values["b"], "c": {Value: json.RawMessage(`{"n":3}`), Revision: 10}} + asks := 0 + f, _ := follower(func(ctx context.Context, after string) (Answer, error) { + asks++ + if asks == 1 { + return answer(ctx, after) + } + return FirstView("o", "modules", later, mapSource{"modules": markOf("modules", 10, made)}, after, now()) + }) + _ = f.Start(context.Background()) + got := make(chan string, 4) + f.Changed = func(key string, _ json.RawMessage, _ bool, _ time.Time) { got <- key } + // The change at 10 (c) never reached this reader; only the mark the owner said after it does. + data, _ := Event{V: Version, Owner: "o", State: "modules", Revision: 10, Created: made, Mark: true}.Encode() + if err := HearEvent(context.Background(), map[string]*Follower{"modules": f}, data, t.Logf); err != nil { + t.Fatal(err) + } + if k := <-got; k != "c" { + t.Fatalf("changed %s", k) + } + if v, _ := f.Get("c"); string(v) != `{"n":3}` || asks != 2 { + t.Errorf("c = %s after %d asks", v, asks) + } + }) t.Run("a reader whose owner is down keeps asking, raises past its bound, and the answer clears it", func(t *testing.T) { down := 3 f, said := follower(func(ctx context.Context, after string) (Answer, error) { @@ -541,7 +591,7 @@ func TestViews(t *testing.T) { updates <- kvEntry{key: "b", rev: 2, at: made} close(updates) errs := make(chan error, 1) - em.follow(context.Background(), "s", made, updates, func() (Fill, error) { return Fill{}, nil }, errs) + em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, errs) if len(said) != 2 || !strings.HasSuffix(said[0], "-1") || !strings.HasSuffix(said[1], "-2") || fails != 0 { t.Fatalf("said %v (fails left %d): the change at 1 must be said, again, before the one at 2", said, fails) } @@ -552,7 +602,7 @@ func TestViews(t *testing.T) { updates = make(chan jetstream.KeyValueEntry, 2) updates <- kvEntry{key: "a", rev: 1, at: made} updates <- kvEntry{key: "b", rev: 2, at: made} - em.follow(ctx, "s", made, updates, func() (Fill, error) { return Fill{}, nil }, errs) + em.follow(ctx, "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, errs) if len(said) != 0 { t.Fatalf("said %v after the context ended on a change not said", said) } @@ -569,7 +619,7 @@ func TestViews(t *testing.T) { updates = make(chan jetstream.KeyValueEntry, 1) updates <- kvEntry{key: "a", rev: 1, at: made} close(updates) - em.follow(context.Background(), "s", made, updates, func() (Fill, error) { return Fill{}, nil }, nil) + em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, nil) if fs.puts != c.puts || fs.deletes != 1 || len(fs.m) != 0 || len(said) != 1 { t.Errorf("%d failures: %d findings written, %d cleared, %d left, said %v", c.fails, fs.puts, fs.deletes, len(fs.m), said) } @@ -589,11 +639,33 @@ func TestViews(t *testing.T) { updates = make(chan jetstream.KeyValueEntry, 1) updates <- kvEntry{key: "b", rev: 2, at: made} close(updates) - em.follow(context.Background(), "s", made, updates, func() (Fill, error) { return Fill{}, nil }, nil) + em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, nil) if _, left := fs.m[key]; left || len(said) != 1 { t.Errorf("after a restart the leftover finding stands (%v), said %v", left, said) } }) + t.Run("each change names the revision said before it, and the mark is said as an event once written", func(t *testing.T) { + var said []Event + marks := &marksKV{} + em := Emitter{Owner: "o", Subject: func(e string) string { return "s." + e }, Marks: marks, + Publish: func(_ context.Context, msg *nats.Msg, id string) error { + e, err := DecodeEvent(msg.Data) + if err != nil || e.ID() != id { + t.Errorf("said %s as %s: %v", msg.Data, id, err) + } + said = append(said, e) + return nil + }} + // 4 to 6 were written over before the watch was handed them: the watch hands 3, then 7. + updates := make(chan jetstream.KeyValueEntry, 2) + updates <- kvEntry{key: "a", rev: 3, at: made} + updates <- kvEntry{key: "a", rev: 7, at: made} + close(updates) + em.follow(context.Background(), "s", made, 2, updates, func() (Fill, error) { return Fill{}, nil }, nil) + if len(said) != 3 || said[0].Prev != 2 || said[1].Prev != 3 || !said[2].Mark || said[2].Revision != 7 || marks.puts != 1 { + t.Fatalf("said %+v with %d marks written: 3 after 2, 7 after 3, then the mark at 7", said, marks.puts) + } + }) t.Run("an event names its state on its subject, and decodes as it was encoded", func(t *testing.T) { if s := StateOfEvent("mesh.stream.mesh-delivery-indexer.event.views.config-candidates"); s != "config-candidates" { t.Errorf("state of subject = %q", s) @@ -630,6 +702,17 @@ func (f *findingsMap) Put(v Finding) error { return nil } +// marksKV is an owner's marks state that takes puts and nothing else. +type marksKV struct { + jetstream.KeyValue + puts int +} + +func (m *marksKV) Put(context.Context, string, []byte) (uint64, error) { + m.puts++ + return uint64(m.puts), nil +} + func (f *findingsMap) Delete(v Finding) error { delete(f.m, v.Kind+"/"+v.ID); f.deletes++; return nil } // actsFunc records a hand act through a test's function.