diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 238fa5e1..038a9275 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -1086,13 +1086,44 @@ 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) and every wait +// indexer_pending names the entries by (PendingWaits: a revision by its oldest merge's key, a group by its g. key) not +// already ended by one of those, 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 _, 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..3043d1b1 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)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 { + t.Errorf("said %v, pending %v: want %v once", 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")) @@ -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)}; !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/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 92bc72e8..30457c62 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -105,6 +105,15 @@ func TestTheWatchmansFindings(t *testing.T) { } w.TakeInCaughtUp() }, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}}, + {"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)))