R1-F-sdk 2: a view event not said is said again before the mark moves past it

A failed say was logged and passed over, and the next change's mark moved past it, so a reader missed the change for good (design 25 §12.2). The emitter now says the same change again on a backoff until it is said or its context ends. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 16:22:18 +01:00
parent c4de50e91e
commit aef9957450
2 changed files with 126 additions and 33 deletions
@@ -27,8 +27,11 @@ type Emitter struct {
 	// TooLarge is told of a value too large for an event (said as a gap): the owner's finding naming the key. Nil
 	// only logs it.
 	TooLarge func(state, key string, size int)
-	Logf     func(string, ...any)
-	Now      func() time.Time
+	// Wait pauses before a change not said is said again (its attempt so far); it answers false when ctx ended. Nil is
+	// a backoff from 250 ms doubling to 30 s.
+	Wait func(ctx context.Context, attempt int) bool
+	Logf func(string, ...any)
+	Now  func() time.Time
 }

 // JetStreamPublish is Emitter.Publish through js, with the bus's dedup id.
@@ -87,43 +90,81 @@ func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue,
 	if err != nil {
 		return fmt.Errorf("the state %s cannot be watched for its views: %w", state, err)
 	}
+	fillNow := func() (Fill, error) {
+		_, _, fill, err := created(ctx, kv)
+		return fill, err
+	}
 	go func() {
 		defer w.Stop()
-		said := uint64(0)
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case u, ok := <-w.Updates():
-				if !ok {
-					if errs != nil {
-						errs <- fmt.Errorf("the watch of %s for its views ended", state)
-					}
-					return
-				}
-				if u != nil {
-					if err := e.say(ctx, state, made, u); err != nil {
-						// Not marked: the next start says it again from the mark, and the bus keeps it once by its id.
-						e.logf("views: %s.%s at %d is not said: %v", state, u.Key(), u.Revision(), err)
-						continue
-					}
-					said = u.Revision()
-				}
-				// Marked once every change handed so far is said, so a busy state writes one mark per burst.
-				if said > 0 && len(w.Updates()) == 0 {
-					_, _, fill, err := created(ctx, kv)
-					if err != nil {
-						e.logf("views: the fill of %s is not read: %v", state, err)
-					}
-					e.writeMark(ctx, state, Mark{State: state, Revision: said, Created: made, Fill: fill})
-					said = 0
-				}
-			}
-		}
+		e.follow(ctx, state, made, 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) {
+	said := uint64(0)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case u, ok := <-updates:
+			if !ok {
+				if errs != nil {
+					errs <- fmt.Errorf("the watch of %s for its views ended", state)
+				}
+				return
+			}
+			if u != nil {
+				for attempt := 1; ; attempt++ {
+					err := e.say(ctx, state, made, u)
+					if err == nil {
+						break
+					}
+					e.logf("views: %s.%s at %d is not said (try %d): %v; said again before the next change", state, u.Key(), u.Revision(), attempt, err)
+					if !e.wait(ctx, attempt) {
+						// Not marked: the next start says it again from the mark, and the bus keeps it once by its id.
+						return
+					}
+				}
+				said = 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 {
+				fill, err := fillNow()
+				if err != nil {
+					e.logf("views: the fill of %s is not read: %v", state, err)
+				}
+				e.writeMark(ctx, state, Mark{State: state, Revision: said, Created: made, Fill: fill})
+				said = 0
+			}
+		}
+	}
+}
+
+// wait is the pause before the attempt+1'th try of a change not said: Wait when given, else a backoff from 250 ms
+// doubling to 30 s. It answers false once ctx ends.
+func (e Emitter) wait(ctx context.Context, attempt int) bool {
+	if e.Wait != nil {
+		return e.Wait(ctx, attempt)
+	}
+	d := 250 * time.Millisecond
+	for i := 1; i < attempt && d < 30*time.Second; i++ {
+		d *= 2
+	}
+	d = min(d, 30*time.Second)
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-t.C:
+		return true
+	}
+}
+
 // Event is the event saying u, a change of state made at made.
 func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry) Event {
 	ev := Event{V: Version, Owner: e.Owner, State: state, Key: u.Key(), Revision: u.Revision(), Created: made, At: u.Created().UTC()}
@@ -9,6 +9,9 @@ import (
 	"testing"
 	"time"

+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/handacts"
 )
@@ -396,6 +399,40 @@ func TestViews(t *testing.T) {
 			}
 		}
 	})
+	t.Run("a change not said is said again on a backoff before the next, and the mark never passes it", func(t *testing.T) {
+		var said []string
+		fails := 2
+		em := Emitter{Owner: "o", Subject: func(e string) string { return "s." + e },
+			Publish: func(_ context.Context, msg *nats.Msg, id string) error {
+				if fails > 0 && strings.HasSuffix(id, "-1") {
+					fails--
+					return errors.New("bus away")
+				}
+				said = append(said, id)
+				return nil
+			},
+			Wait: func(_ context.Context, attempt int) bool { return true }}
+		updates := make(chan jetstream.KeyValueEntry, 2)
+		updates <- kvEntry{key: "a", rev: 1, at: made}
+		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)
+		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)
+		}
+		// A context that ends while a change waits to be said again stops there: nothing after it is said.
+		said, fails = nil, 1
+		ctx, cancel := context.WithCancel(context.Background())
+		em.Wait = func(context.Context, int) bool { cancel(); return false }
+		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)
+		if len(said) != 0 {
+			t.Fatalf("said %v after the context ended on a change not said", said)
+		}
+	})
 	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)
@@ -427,3 +464,18 @@ func (f *findingsMap) Put(k string, v []byte) error {
 }

 func (f *findingsMap) Delete(k string) error { delete(f.m, k); f.deletes++; return nil }
+
+// kvEntry is one change of a bucket as a watch hands it.
+type kvEntry struct {
+	key string
+	rev uint64
+	at  time.Time
+}
+
+func (k kvEntry) Bucket() string                  { return "b" }
+func (k kvEntry) Key() string                     { return k.key }
+func (k kvEntry) Value() []byte                   { return []byte(`{"n":1}`) }
+func (k kvEntry) Revision() uint64                { return k.rev }
+func (k kvEntry) Created() time.Time              { return k.at }
+func (k kvEntry) Delta() uint64                   { return 0 }
+func (k kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }