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

@@ -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)
 	}
 }

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

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

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
 }

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.