R3-F-sdk: the first change said after a restart clears a view-unsaid finding left standing

raised started false on every run of follow, so a finding a run before a restart raised was never cleared.
A flag of its own clears it once at the first change said; the raise keeps its own flag, so a change not
said in the new run is still raised past its bound.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-04 18:41:18 +01:00
parent e5d89c5eed
commit f94abb95aa
2 changed files with 18 additions and 2 deletions
+5 -1
View File
@@ -115,6 +115,9 @@ func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue,
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
// A finding a run before this one raised is not known to this one: the first change said clears it once, so a
// restart never leaves it standing (the finding is about the state, its key the same).
leftover := e.Findings != nil
for {
select {
case <-ctx.Done():
@@ -145,8 +148,9 @@ func (e Emitter) follow(ctx context.Context, state string, made time.Time, updat
return
}
}
if raised {
if raised || leftover {
raised = !e.clearUnsaid(state)
leftover = false
}
said = u.Revision()
}
+13 -1
View File
@@ -446,7 +446,7 @@ func TestViews(t *testing.T) {
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 {
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)
}
if c.puts > 0 {
@@ -456,6 +456,18 @@ func TestViews(t *testing.T) {
}
}
}
// A finding a run before a restart raised is cleared by the first change said after it.
key, value, _ := UnsaidFinding("o", "s", "a", 1, made, "bus away", made)
fs := &findingsMap{m: map[string][]byte{key: value}}
said, fails = nil, 0
em.Findings = fs
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)
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("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" {