R2-G-sdk-contracts 1: a view change not said past its bound is the owner's finding

A change said again and again without the bus taking it left every reader standing still with only a log
line. Past UnsaidAfter (Bound by default) the emitter writes the owner's finding view-unsaid naming the
state, as it does for a value too large, and clears it once the change is said.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-04 18:16:54 +01:00
parent d420491ca1
commit be1ffdfedb
3 changed files with 95 additions and 2 deletions
+56
View File
@@ -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 {
+15 -2
View File
@@ -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
+24
View File
@@ -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" {