diff --git a/go/views/emit.go b/go/views/emit.go index 6077f3f..5f009b5 100644 --- a/go/views/emit.go +++ b/go/views/emit.go @@ -27,6 +27,14 @@ 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) + // Findings writes and clears the owner's finding of a change not said past UnsaidAfter (UnsaidKind, its own state + // `findings`); nil says it in the log only. + Findings interface { + Put(key string, value []byte) error + Delete(key string) error + } + // UnsaidAfter is how long a change may stay not said before it is the owner's finding; zero is Bound. + UnsaidAfter time.Duration // 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 @@ -106,6 +114,7 @@ func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, // 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) + raised := false for { select { case <-ctx.Done(): @@ -118,17 +127,27 @@ func (e Emitter) follow(ctx context.Context, state string, made time.Time, updat return } if u != nil { + var first time.Time 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 first.IsZero() { + first = e.now() + } + if !raised && e.now().Sub(first) >= e.unsaidAfter() { + raised = e.raiseUnsaid(state, u, first, 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 } } + if raised { + raised = !e.clearUnsaid(state) + } said = u.Revision() } // Marked once every change handed so far is said, so a busy state writes one mark per burst. @@ -144,6 +163,43 @@ func (e Emitter) follow(ctx context.Context, state string, made time.Time, updat } } +func (e Emitter) unsaidAfter() time.Duration { + if e.UnsaidAfter > 0 { + return e.UnsaidAfter + } + return Bound +} + +// raiseUnsaid writes the owner's finding of u not said since first; it answers whether the finding stands. +func (e Emitter) raiseUnsaid(state string, u jetstream.KeyValueEntry, first time.Time, cause error) bool { + e.logf("views: %s.%s at %d is not said since %s: the owner's finding %s", state, u.Key(), u.Revision(), first.Format(time.RFC3339), UnsaidKind) + if e.Findings == nil { + return false + } + key, value, err := UnsaidFinding(e.Owner, state, u.Key(), u.Revision(), first, cause.Error(), e.now()) + if err == nil { + err = e.Findings.Put(key, value) + } + if err != nil { + e.logf("views: the finding of %s's change not said could not be written: %v", state, err) + return false + } + return true +} + +// clearUnsaid clears the finding once the change is said; it answers whether it is cleared. +func (e Emitter) clearUnsaid(state string) bool { + key, _, err := UnsaidFinding(e.Owner, state, "", 0, time.Time{}, "", e.now()) + if err == nil { + err = e.Findings.Delete(key) + } + if err != nil { + e.logf("views: the finding of %s's change not said could not be cleared: %v", state, err) + return false + } + return true +} + // 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 { diff --git a/go/views/findings.go b/go/views/findings.go index 9795fe1..781186d 100644 --- a/go/views/findings.go +++ b/go/views/findings.go @@ -9,13 +9,26 @@ import ( ) // The findings a views part writes in its own state `findings`, which the condition keeper raises and clears -// (novox/hq ADR 0564 §2): an owner's value too large for an event (TooLargeKind), and a reader's first view unanswered -// past its bound (UnansweredKind). +// (novox/hq ADR 0564 §2): an owner's value too large for an event (TooLargeKind), an owner's change not said past its +// bound (UnsaidKind), and a reader's first view unanswered past its bound (UnansweredKind). const ( TooLargeKind = "view-too-large" + UnsaidKind = "view-unsaid" UnansweredKind = "first-view-unanswered" ) +// UnsaidFinding is the owner's finding of a change of state it has not said on its views family since since: every +// reader of the state stands still at the change before it until it is said, which clears this. +func UnsaidFinding(owner, state, key string, revision uint64, since time.Time, cause string, at time.Time) (string, []byte, error) { + id := owner + "." + state + summary := fmt.Sprintf("%s has not said the change of %s.%s at %d on its views family since %s (%s): its readers stand still before it", + owner, state, key, revision, since.UTC().Format(time.RFC3339), strings.TrimSpace(cause)) + return finding(owner, UnsaidKind, id, summary, + fmt.Sprintf("%s cannot say a change of its %s", owner, state), + "The owner says each change of its state as an event on its own stream, and says a change again until the bus takes it; "+ + "its readers hear nothing newer meanwhile. Look at the owner's stream on the bus: is it full, or is the bus away?", at) +} + // TooLargeFinding is the owner's finding of state's key too large to say whole on its views family. func TooLargeFinding(owner, state, key string, size int, at time.Time) (string, []byte, error) { id := owner + "." + state + "." + key diff --git a/go/views/views_test.go b/go/views/views_test.go index 3cd0ace..8fed70b 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -432,6 +432,30 @@ func TestViews(t *testing.T) { if len(said) != 0 { t.Fatalf("said %v after the context ended on a change not said", said) } + // A change not said past its bound (Bound) is the owner's finding naming the state, cleared once it is said; + // one said within the bound raises none. + for _, c := range []struct { + fails, puts int + }{{1, 0}, {3, 1}} { + clock := made + fs := &findingsMap{} + said, fails = nil, c.fails + em.Now, em.Findings = func() time.Time { return clock }, fs + em.Wait = func(context.Context, int) bool { clock = clock.Add(time.Minute); return true } + 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) + if fs.puts != c.puts || fs.deletes != c.puts || 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) + } + if c.puts > 0 { + key, value, _ := UnsaidFinding("o", "s", "a", 1, made, "bus away", clock) + if !strings.Contains(key, UnsaidKind) || !strings.Contains(key, "o.s") || !strings.Contains(string(value), "o has not said the change of s.a at 1") { + t.Errorf("the finding is %s: %s", key, value) + } + } + } }) 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" {