diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 238fa5e1..0f6515c6 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -1086,13 +1086,47 @@ func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, ow } ix.forget(claimed...) ix.logf("take-in: %d merges taken in as one revision by %s%s, oldest %s", len(claimed), own, byWords(by), claimed[0].Key) - for _, t := range saidWaits(claimed) { + for _, t := range takenWaits(claimed) { t.By, t.Why = who, why ix.sayTakeIn(ctx, t) } return applied, nil } +// takenWaits are the taken events of a take-in: every wait the entries were said under (saidWaits), every wait +// indexer_pending names the claimed entries by now (WaitsOf: a revision by its oldest merge's key, a group by its g. key) and +// the key of every claimed entry, since indexer_pending named it by that key earlier when it waited behind a group. So a wait +// the watchman learned of without the indexer marking it (seeded from indexer_pending, or said by the own-build-unknown +// path) ends too (design 47, Taken in by the indexer it names). No key is said twice. +func takenWaits(entries []PendingEntry) []takein.TakeIn { + out := saidWaits(entries) + said := map[string]bool{} + for _, t := range out { + said[t.Key] = true + } + w := WaitsOf(entries) + var named []takein.TakeIn + if len(w.Ready) > 0 { + named = append(named, revisionWait(takein.TakeInTaken, w.Ready, w.Ready[len(w.Ready)-1].Value.WaitsFor)) + } + for _, g := range slices.Sorted(maps.Keys(w.Groups)) { + named = append(named, groupWait(takein.TakeInTaken, g, w.Groups[g])) + } + for _, g := range slices.Sorted(maps.Keys(w.Behind)) { + named = append(named, revisionWait(takein.TakeInTaken, w.Behind[g], w.Behind[g][len(w.Behind[g])-1].Value.WaitsFor)) + } + for _, e := range entries { + named = append(named, revisionWait(takein.TakeInTaken, []PendingEntry{e}, e.Value.WaitsFor)) + } + for _, t := range named { + if !said[t.Key] { + said[t.Key] = true + out = append(out, t) + } + } + return out +} + func byWords(by string) string { if by == "" { return "" diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 066f9450..27bff288 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -205,6 +205,33 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) t.Errorf("no pending state: err %v, avahi at %s: want the merge taken in at once", err, b.commit(t, "avahi")) } }) + t.Run("1966/AC195-2 AC195-3 a merge said pending with its build unknown and taken in on redelivery ends that wait under its key; a revision wait is said once", func(t *testing.T) { + b := newTakeInBed(t) + ownBuild := b.ix.OwnBuild + b.ix.ownFP, b.ix.ownFixed = "", false + b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") } + b.ix.FixOwnBuild(ctx) + b.said = nil + if _, _, err := b.ix.Arrive(ctx, 7, b.merge(30, sha('b'), "tree-avahi-2", "")); !errors.Is(err, ErrOwnBuildUnknown) { + t.Fatal(err) + } + b.ix.OwnBuild, b.ix.ownFP, b.ix.ownFixed = ownBuild, "", false + b.ix.FixOwnBuild(ctx) + b.arrive(t, 7, b.merge(30, sha('b'), "tree-avahi-2", "")) // the stream delivers it again + if want := []string{"pending " + takein.PendingKey(7), "taken " + takein.PendingKey(7)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 { + t.Fatalf("said %v, pending %v: want %v and nothing pending", b.events(), b.keys(), want) + } + // Two merges said pending under the oldest's key (waiting for a named indexer), then taken in together: one taken, under + // that key, and no entry's key said twice. + b.said = nil + b.arrive(t, 8, b.merge(31, sha('c'), "tree-avahi-3", "tree-indexer-2")) + b.arrive(t, 9, b.merge(32, sha('d'), "tree-avahi-4", "tree-indexer-2")) + b.said = nil + b.arrive(t, 10, b.merge(33, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+Module])) + if want := []string{"taken " + takein.PendingKey(8), "taken " + takein.PendingKey(9), "taken " + takein.PendingKey(10)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 { + t.Errorf("said %v, pending %v: want %v, the wait's key (m.8, said pending) once and each other entry's own", b.events(), b.keys(), want) + } + }) t.Run("T7 and T3 a merge alone, an indexer change, a merge behind it and a revert", func(t *testing.T) { b := newTakeInBed(t) own, avahi0 := b.fp(t, Module), b.fp(t, "avahi") @@ -225,9 +252,10 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) t.Fatalf("taken in at the bind: %d verdicts (were %d), want the new manifest's", n, judgedBefore) } // A merge with no change to the indexer and nothing pending is taken in alone, as before. - if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || len(b.said) != 0 { - t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in", avahi1, avahi0, b.keys(), b.events()) + if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || !slices.Equal(b.events(), []string{"taken " + takein.PendingKey(1)}) { + t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in, said taken under its key (1966/AC195-1)", avahi1, avahi0, b.keys(), b.events()) } + b.said = nil avahi1, verdicts := b.fp(t, "avahi"), len(b.ix.Verdicts.(*memStore).vals) // T4: a change to the indexer writes only its entry (with its graph key and verdict); the rest is pending. b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) @@ -259,7 +287,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if len(b.keys()) != 0 || b.fp(t, Module) != own || b.commit(t, "avahi") != sha('e') { t.Fatalf("revert: pending %v, indexer %s, avahi at %s: want all taken in", b.keys(), b.fp(t, Module), b.commit(t, "avahi")) } - if want := []string{"pending " + takein.PendingKey(2), "taken " + takein.PendingKey(2)}; !slices.Equal(b.events(), want) { + if want := []string{"pending " + takein.PendingKey(2), "taken " + takein.PendingKey(2), "taken " + takein.PendingKey(3), "taken " + takein.PendingKey(4)}; !slices.Equal(b.events(), want) { t.Errorf("said %v, want %v", b.events(), want) } }) @@ -459,7 +487,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') { t.Fatalf("end: pending %v, avahi at %s: want the group and the merge behind it taken in", b.keys(), b.commit(t, "avahi")) } - if want := []string{"pending g.feat/take-in", "pending g.feat/take-in", "taken g.feat/take-in"}; !slices.Equal(b.events(), want) { + if want := []string{"pending g.feat/take-in", "taken " + takein.PendingKey(2), "pending g.feat/take-in", "taken g.feat/take-in", "taken " + takein.PendingKey(1), "taken " + takein.PendingKey(3), "taken " + takein.PendingKey(4)}; !slices.Equal(b.events(), want) { t.Errorf("said %v, want %v", b.events(), want) } // R1: the end published again after the take-in changes nothing and is not delivered again. @@ -510,7 +538,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if err := b.ix.PendingAgain(ctx); err != nil { t.Fatal(err) } - if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') || !slices.Equal(b.events(), []string{"taken g.feat/take-in"}) { + if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') || !slices.Equal(b.events(), []string{"taken g.feat/take-in", "taken " + takein.PendingKey(2)}) { t.Errorf("all taken: pending %v, avahi at %s, said %v: want the group ended and the merge behind it taken in", b.keys(), b.commit(t, "avahi"), b.events()) } held := func() *takeInBed { diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index 7e8ab4b3..a3f306da 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -13,6 +13,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" + "mesh-delivery-watchman/internal/watch" "github.com/nats-io/nats.go/jetstream" @@ -77,6 +79,34 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") { t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err) } + t.Run("1966/AC195-5: a seed indexer_pending could not answer, or answered unreadably, ends nothing", func(t *testing.T) { + for name, call := range map[string]func(string, any) (json.RawMessage, error){ + "cannot be asked": func(string, any) (json.RawMessage, error) { return nil, errors.New("no answer") }, + "unreadable": func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"waits":`), nil }, + } { + out := &findingsOut{findings: map[string]bool{}} + w := watch.New(out, time.Now) + const key = "m.00000000000000000042" + old := watch.Finding{Scope: watch.ScopeDelivery, ID: key, Kind: watch.KindTakeInWaits, Severity: watch.Warning, Source: "take-in", + Since: time.Now().Add(-3 * time.Hour), Summary: "held", Said: "held", Headline: "held", Explanation: "held", Needs: "a take-in"} + value, err := watch.EncodeFinding(old) + if err != nil { + t.Fatal(err) + } + out.findings[old.Key()] = true + w.SeedFinding(old.Key(), value) + body, err := takein.EncodeTakeIn(takein.TakeIn{Kind: takein.TakeInPending, Key: "m.00000000000000000043", What: takein.TakeInRevision, Since: time.Now(), WaitsFor: "src2:x", + Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}}) + if err != nil { + t.Fatal(err) + } + w.TakeIn(takein.TakeInPendingEvent, body) + seedTakeIn(w, call, func(string, ...any) {}) + if !out.findings[old.Key()] || out.pulseDeletes != 0 { + t.Errorf("%s: finding stands %v, signals ended %d: want the previous run's finding and the held wait kept", name, out.findings[old.Key()], out.pulseDeletes) + } + } + }) t.Run("A2j: a want stood since the assigner's bus wrote it, through the first view", func(t *testing.T) { // The watchman's want wait starts at the entry's time (createdOr): through mesh-sdk's views.ProcessReader it is // the time the owner's bus wrote the key, never the watchman's start. @@ -113,11 +143,14 @@ type refusingBus struct{ err error } func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err } // findingsOut keeps which findings stand. -type findingsOut struct{ findings map[string]bool } +type findingsOut struct { + findings map[string]bool + pulseDeletes int +} func (o *findingsOut) PutPresence(string, bool) error { return nil } func (o *findingsOut) PutPulse(string, []byte, time.Duration) error { return nil } -func (o *findingsOut) DeletePulse(string) error { return nil } +func (o *findingsOut) DeletePulse(string) error { o.pulseDeletes++; return nil } func (o *findingsOut) PutStuck(string, []byte) error { return nil } func (o *findingsOut) PutFinding(key string, _ []byte) error { o.findings[key] = true; return nil } func (o *findingsOut) DeleteFinding(key string) error { delete(o.findings, key); return nil } diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 92bc72e8..2363ea72 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -105,6 +105,36 @@ func TestTheWatchmansFindings(t *testing.T) { } w.TakeInCaughtUp() }, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}}, + {"1966/AC195-5: a seed ends the finding and the held wait of a key indexer_pending no longer names, and keeps one it names with its own since", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + // A previous run's finding for revKey stands in the bucket (read at the start); a wait of staleKey is held from an event. + old := Finding{Scope: ScopeDelivery, ID: revKey, Kind: KindTakeInWaits, Severity: Warning, Source: "take-in", Since: t0.Add(-3 * time.Hour), Observer: Observer, Summary: "held", Said: "held", Headline: "held", Explanation: "held", Needs: "a take-in"} + value, err := EncodeFinding(old) + if err != nil { + t.Fatal(err) + } + o.findings[old.Key()] = old + w.SeedFinding(old.Key(), value) + const staleKey = "m.00000000000000000099" + w.TakeInCaughtUp() + w.TakeIn(takeInEvent(t, takein.TakeInPending, staleKey), takeInBody(t, takein.TakeInPending, staleKey, c.t)) + w.SeedTakeIn(takein.PendingAnswer{Waits: []takein.PendingWait{{Key: groupKey, What: takein.TakeInGroup, Keys: []string{groupKey}, + Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}, Since: c.t.Add(-10 * time.Minute)}}}) + if _, ok := o.findings[old.Key()]; ok || w.signals[signalKey("take-in", staleKey)].Finding.ID != "" || w.signals[signalKey("take-in", revKey)].Finding.ID != "" { + t.Fatalf("a wait the answer does not name stands: findings %v, signals %v", o.findings, w.signals) + } + if o.ttls[signalKey("take-in", groupKey)] != TakeInGroupBound-10*time.Minute { + t.Fatalf("the named wait: ttl %v, want the rest of its bound from its own since", o.ttls[signalKey("take-in", groupKey)]) + } + }, map[string]string{}}, + {"1966/AC195-4: a taken event for a key the watchman holds no wait for changes no finding and no signal", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.TakeInCaughtUp() + w.TakeIn(takeInEvent(t, takein.TakeInPending, groupKey), takeInBody(t, takein.TakeInPending, groupKey, t0)) + ttls, pulses := len(o.ttls), len(o.pulses) + w.TakeIn(takeInEvent(t, takein.TakeInTaken, revKey), takeInBody(t, takein.TakeInTaken, revKey, t0)) + if o.findingWrites != 0 || len(o.ttls) != ttls || len(o.pulses) != pulses || o.ttls[signalKey("take-in", groupKey)] != TakeInGroupBound { + t.Fatalf("a taken for an unheld key: findings written %d, ttls %v, pulses %v", o.findingWrites, o.ttls, o.pulses) + } + }, map[string]string{}}, {"ADR 0633 §5 (1585/T21): a replayed pending followed by its taken never raises", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { w.TakeIn(takeInEvent(t, takein.TakeInPending, revKey), takeInBody(t, takein.TakeInPending, revKey, t0.Add(-2*time.Hour))) w.TakeIn(takeInEvent(t, takein.TakeInTaken, revKey), takeInBody(t, takein.TakeInTaken, revKey, t0.Add(-2*time.Hour))) diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index 171aa403..cc6b6c2e 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/modules/mesh-delivery-watchman/internal/watch/takein.go @@ -52,10 +52,36 @@ func (w *Watchman) TakeIn(event string, body []byte) { // and group the indexer holds now, as indexer_pending answers them (ADR 0633 §5): the consumer is durable, so the // pending events it acknowledged before a restart are not heard again, and a wait within its bound would otherwise be // lost. Each wait's bound runs from its own since, as a pending event's does; the stream's events that follow replace -// or end it. +// or end it. A wait or finding the answer does not name is ended. func (w *Watchman) SeedTakeIn(a takein.PendingAnswer) { w.mu.Lock() defer w.mu.Unlock() + // A wait or finding of a key the answer does not name is ended as a take-in.taken would end it: the finding a previous + // run wrote (read at the start, before the seed), a wait held from an event, a signal standing (design 47, Taken in by + // the indexer it names). Only an answered seed gets here: one that could not ask ends nothing. + named := map[string]bool{} + for _, wait := range a.Waits { + named[wait.Key] = true + } + stale := map[string]bool{} + for _, value := range w.findings { + if f, err := DecodeFinding(value); err == nil && f.Scope == ScopeDelivery && f.Kind == KindTakeInWaits && !named[f.ID] { + stale[f.ID] = true + } + } + for _, due := range w.signals { + if due.Finding.Kind == KindTakeInWaits && due.Finding.Scope == ScopeDelivery && !named[due.Finding.ID] { + stale[due.Finding.ID] = true + } + } + for key := range w.takeInHeld { + if !named[key] { + stale[key] = true + } + } + for _, key := range slices.Sorted(maps.Keys(stale)) { + w.takeIn(takein.TakeInTaken, key, takein.TakeIn{Kind: takein.TakeInTaken, Key: key}) + } for _, wait := range a.Waits { w.takeIn(takein.TakeInPending, wait.Key, takein.TakeIn{Kind: takein.TakeInPending, Key: wait.Key, What: wait.What, Merges: wait.Merges, Group: wait.Group, Members: wait.Members, WaitsFor: wait.WaitsFor, Since: wait.Since})