From aef9957450defff22c2acd7d85038406a7f7f21b Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 16:22:18 +0100 Subject: [PATCH] R1-F-sdk 2: a view event not said is said again before the mark moves past it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- go/views/emit.go | 107 ++++++++++++++++++++++++++++------------- go/views/views_test.go | 52 ++++++++++++++++++++ 2 files changed, 126 insertions(+), 33 deletions(-) diff --git a/go/views/emit.go b/go/views/emit.go index 23a251a..6077f3f 100644 --- a/go/views/emit.go +++ b/go/views/emit.go @@ -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()} diff --git a/go/views/views_test.go b/go/views/views_test.go index e1a990f..3cd0ace 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -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 }