diff --git a/go/views/emit.go b/go/views/emit.go index 5f009b5..2e16fa7 100644 --- a/go/views/emit.go +++ b/go/views/emit.go @@ -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() } diff --git a/go/views/views_test.go b/go/views/views_test.go index 8fed70b..880cea2 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -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" {