From c2e466e16c6d13fcd599eef305979bdb9b183323 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 23:41:50 +0100 Subject: [PATCH 1/4] The indexer says take-in.taken for every wait indexer_pending names (issue 1966, note 195) A merge taken in with no take-in.pending said for it (read from indexer_pending at a watchman restart, or said pending by the own-build-unknown path) ended with no taken event, so the watchman raised take-in-waits for a wait nothing held. takeInWholeBy now says taken for each wait WaitsOf names for the claimed entries (a revision by its oldest merge's key, a group by its g. key) as well as for its said waits, without saying a key twice. Drops are unchanged. Builds: design 47 "Taken in by the indexer it names" (A wait is bounded, said and recoverable); 1966/AC195-1, AC195-2, AC195-3, AC195-4 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../internal/index/pending.go | 33 ++++++++++++++++- .../internal/index/pending_test.go | 36 ++++++++++++++++--- .../internal/watch/findings_test.go | 9 +++++ 3 files changed, 73 insertions(+), 5 deletions(-) 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))) From 393f69f76676cc22781290b77286dd5b0067e558 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 23:45:06 +0100 Subject: [PATCH 2/4] The watchman's seed ends the waits and findings indexer_pending no longer names (issue 1966, note 195) SeedTakeIn, run once indexer_pending answered and after the findings of a previous run were read, ends every take-in wait, held wait and take-in-waits finding whose key the answer does not name, as a take-in.taken would. A seed that could not ask is not run, so ends nothing. Builds: design 47 "Taken in by the indexer it names" (A wait is bounded, said and recoverable: "At its start the watchman seeds its open waits from indexer_pending, and ends every take-in wait and take-in-waits finding it holds whose key indexer_pending no longer names"); 1966/AC195-5 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../internal/watch/findings_test.go | 21 ++++++++++++++ .../internal/watch/takein.go | 28 ++++++++++++++++++- 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 30457c62..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,27 @@ 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)) 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}) From 3ed2a3dd331696bd09404e6d8ee33889a5ee124e Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 23:45:46 +0100 Subject: [PATCH 3/4] The indexer says take-in.taken under every claimed entry's own key too (issue 1966, note 195) A merge waiting behind a gathered group is named by indexer_pending under its own key; when the group ends it joins the group's revision under the oldest member's key, so its own key was never said taken. takenWaits now also says every claimed entry's key, no key twice. T8 checks the merge behind the group ends under its own key. Builds: design 47 "Taken in by the indexer it names" (A wait is bounded, said and recoverable); 1966/AC195-1, AC195-3 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../mesh-delivery-indexer/internal/index/pending.go | 13 ++++++++----- .../internal/index/pending_test.go | 8 ++++---- 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 038a9275..0f6515c6 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -1093,11 +1093,11 @@ func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, ow 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. +// 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{} @@ -1115,6 +1115,9 @@ func takenWaits(entries []PendingEntry) []takein.TakeIn { 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 diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 3043d1b1..27bff288 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -228,8 +228,8 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) 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) + 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) { @@ -287,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) } }) @@ -487,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", "taken " + takein.PendingKey(2), "pending g.feat/take-in", "taken g.feat/take-in", "taken " + takein.PendingKey(1)}; !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. From e82da39d299910a442375adbb76778ec01cf3817 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 23:48:45 +0100 Subject: [PATCH 4/4] Test that a seed indexer_pending could not answer ends nothing (issue 1966, note 197) seedTakeIn with a failing call and with an unreadable answer leaves a previous run's take-in-waits finding and a held wait standing; it fails if the error path seeded an empty answer. Builds: design 47 "Taken in by the indexer it names" (A wait is bounded, said and recoverable); 1966/AC195-5 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-watchman/main_test.go | 37 ++++++++++++++++++- 1 file changed, 35 insertions(+), 2 deletions(-) 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 }