From a7aa836ff9ac565d6f2cca1149c7df7b24a70700 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 7 Nov 2026 12:25:26 +0100 Subject: [PATCH 1/3] A group wait whose members are all taken in ends itself, and the take-in at the start is tried again (issue 1585) --- .../cmd/mesh-delivery-indexer/main.go | 6 +- .../cmd/mesh-delivery-indexer/startwait.go | 18 +++ .../internal/index/pending.go | 138 ++++++++++++++++++ .../internal/index/pending_test.go | 81 ++++++++++ .../mesh-delivery-indexer/manifest_test.go | 4 + 5 files changed, 245 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 5445dc48..1bc29462 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -510,8 +510,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // stream starts (ADR 0633 §2). When the build this program runs is not known yet, nothing waits for it: the consumer // starts, and a merged message left unacknowledged asks the build again at its redelivery (clause 1). A hand take-in // (indexer_take_in_pending) is answered at any time: the takeins watch runs already. - if _, words, err := ix.TakeInAtStart(ctx); err != nil { - return fmt.Errorf("the take-in at start: %w", err) + // A failed take-in is tried again with a backoff (issue 1585, note 70); one that stays failed does not end the process: a + // merged message arriving, a toolchain report or indexer_take_in_pending asks again. + if _, words, err := ix.TakeInAtStartRetrying(ctx, sleepFor); err != nil { + logger.Printf("the take-in at start failed and is asked again at the next merged message, toolchain report or hand take-in: %v", err) } else if words != "" { logger.Printf("take-in at start: %s", words) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go new file mode 100644 index 00000000..81e515e3 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go @@ -0,0 +1,18 @@ +package main + +import ( + "context" + "time" +) + +// sleepFor is the one wait the indexer holds: the pause between the tries of the take-in at its start (issue 1585, note 70), +// one-shot and bounded by index.StartAttempts. It is no ticker and polls nothing; it is kept in this file alone, which +// TestServesNoRequestSubjectAndHoldsNoTicker names. +func sleepFor(ctx context.Context, d time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(d): + return nil + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index ccc5673f..6cb28bc7 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -511,6 +511,89 @@ func (ix *Indexer) EndGroup(ctx context.Context, end forgemsg.ForgeGroupEnd) (Ap return ix.evaluate(ctx, own, false) } +// endTakenGroups ends every group wait whose members are all taken in (issue 1585, notes 70 and 71): a group's end that +// never came, or a member taken in by hand (indexer_take_in_merge, a m. key), would hold the group and every merge behind +// it for ever. A member held is taken in when its merge is at or behind its repository's taken merge; a member no longer +// held when its head is in that taken merge. The wait is said taken (take-in.taken under the group key), and only then are +// its keys removed: a wait that could not be said stays and is ended at the next evaluation. It answers whether one ended. +func (ix *Indexer) endTakenGroups(ctx context.Context, entries []PendingEntry) (bool, error) { + if ix.Taken == nil { + return false, nil + } + w := WaitsOf(entries) + ended := false + for _, group := range slices.Sorted(maps.Keys(w.Groups)) { + held := w.Groups[group] + all, err := ix.groupTaken(ctx, held) + if err != nil { + return ended, err + } + if !all { + continue + } + if !ix.sayTakeIn(ctx, groupWait(takein.TakeInTaken, group, held)) { + continue + } + for _, e := range held { + if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil { + ix.logf("take-in: %s.%s belongs to a group taken in and is not removed: %v", PendingState, e.Key, err) + } + } + ix.forget(held...) + ix.logf("take-in: the group %q ended: every member is taken in already (%d merges held, removed)", group, len(held)) + ended = true + } + return ended, nil +} + +// groupTaken says whether every member of the group whose entries are held is taken in. +func (ix *Indexer) groupTaken(ctx context.Context, held []PendingEntry) (bool, error) { + isHeld := func(m delivery.Head) bool { + return slices.ContainsFunc(held, func(e PendingEntry) bool { + return delivery.SameRepository(e.Forge.Repository, m.Repository) && e.Forge.Number == m.Number + }) + } + for _, e := range held { + if ok, err := ix.takenAlready(ctx, forgemsg.GroupMerge{Repository: e.Forge.Repository, Number: e.Forge.Number, Merge: e.Forge.Merge}); err != nil || !ok { + return false, err + } + } + for _, m := range held[len(held)-1].Forge.Members { + if isHeld(m) { + continue + } + ok, err := ix.headTaken(ctx, m) + if err != nil || !ok { + return false, err + } + } + return true, nil +} + +// headTaken says whether a group member that is not held was taken in: its head is the taken merge of its repository or +// in it. +func (ix *Indexer) headTaken(ctx context.Context, m delivery.Head) (bool, error) { + key, err := delivery.RepositoryKey(m.Repository) + if err != nil { + return false, err + } + value, _, found, err := ix.Taken.Get(ctx, key) + if errors.Is(err, process.ErrNotYet) || (err == nil && !found) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("taken.%s: %w", key, err) + } + cur, err := delivery.DecodeTaken(value) + if err != nil || cur.Merge == "" || ix.Contains == nil { + return false, nil + } + if cur.Merge == m.Head { + return true, nil + } + return ix.Contains(ctx, m.Repository, m.Head, cur.Merge) +} + // takenAlready says whether a group member's merge is at or behind its repository's taken merge. func (ix *Indexer) takenAlready(ctx context.Context, m forgemsg.GroupMerge) (bool, error) { if ix.Taken == nil { @@ -566,6 +649,53 @@ func (ix *Indexer) TakeInAtStart(ctx context.Context) (Applied, string, error) { return a, words, err } +// The take-in at the start is asked again when it fails or leaves a merge lost (issue 1585, note 70): a bus error, or the +// forge bridge not answering because it was restarted by the same delivery, is transient. Each try is said. +const ( + StartAttempts = 6 + StartBackoff = time.Second +) + +// TakeInAtStartRetrying is TakeInAtStart tried up to StartAttempts times with a doubling backoff (StartBackoff, then +// twice that, ...), waited with wait (it returns when the context ends), each failure and each lost merge said in the +// journal. It answers the last try's result; an error is that of the last try, which the caller says and goes on from: +// a merged message arriving, a toolchain report or a hand take-in asks again. +func (ix *Indexer) TakeInAtStartRetrying(ctx context.Context, wait func(context.Context, time.Duration) error) (Applied, string, error) { + back := StartBackoff + for attempt := 1; ; attempt++ { + a, words, err := ix.TakeInAtStart(ctx) + problem := "" + switch { + case err != nil: + problem = err.Error() + default: + if ix.Pending == nil { + break + } + if entries, rerr := ReadPending(ctx, ix.Pending, ix.logf); rerr == nil { + for _, e := range entries { + if e.Value.Lost != "" { + problem = fmt.Sprintf("%s (%s#%d) is lost: %s", e.Key, e.Forge.Repository, e.Forge.Number, e.Value.Lost) + break + } + } + } + } + if problem == "" || ctx.Err() != nil { + return a, words, err + } + if attempt >= StartAttempts { + ix.logf("take-in: at the start, try %d of %d failed (%s): not tried again now", attempt, StartAttempts, problem) + return a, words, err + } + ix.logf("take-in: at the start, try %d of %d failed (%s): tried again in %s", attempt, StartAttempts, problem, back) + if wait(ctx, back) != nil { + return a, words, err + } + back *= 2 + } +} + // PendingAgain evaluates pending again after something a fingerprint is computed from changed, a toolchain report // (ADR 0633 §1, §3): a revision that now names the running build is taken in, and one that names another build is held // for that one. It does nothing while the pending state is not on the bus or the build this program runs is not known. @@ -595,6 +725,14 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie if err != nil { return Applied{}, "", err } + // A group wait whose members are all taken in ends first, so what waited behind it is evaluated in this same call. + if ended, err := ix.endTakenGroups(ctx, entries); err != nil { + return Applied{}, "", err + } else if ended { + if entries, err = ReadPending(ctx, ix.Pending, ix.logf); err != nil { + return Applied{}, "", err + } + } w := WaitsOf(entries) if len(w.Ready) == 0 { return Applied{}, ix.waitingLine(w), nil diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index f750a58c..ee728306 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -8,6 +8,7 @@ import ( "slices" "strings" "testing" + "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" @@ -376,6 +377,86 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) t.Errorf("a group naming another indexer: said %v, want %v", b.events(), want) } }) + t.Run("1585 a group wait whose members are all taken in ends itself, behind it a merge is taken in at once, and a take-in at the start that fails is tried again", func(t *testing.T) { + b := newTakeInBed(t) + b.ix.Taken = newMem() + order := "a756bc" + b.ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { + return strings.IndexByte(order, commit[0]) <= strings.IndexByte(order, in[0]), nil + } + taken := func(repo, merge string) { + key, _ := delivery.RepositoryKey(repo) + data, err := delivery.EncodeTaken(delivery.Taken{Repository: delivery.RepositoryName(repo), Merge: merge}) + if err != nil { + t.Fatal(err) + } + b.ix.Taken.(*memStore).put(key, data) + } + f := b.merge(40, sha('7'), "", "") + f.Group, f.Head = "feat/take-in", sha('1') + f.Members = []delivery.Head{{Repository: catalogueRepo, Number: 40, Head: sha('1')}, {Repository: "novox/mesh-sdk", Number: 41, Head: sha('2')}} + b.arrive(t, 1, f) // the group's other member was taken in by hand: its key is gone and the group's end never came + b.arrive(t, 2, b.merge(51, sha('6'), "tree-avahi-9", "")) + b.said = nil + // Not every member is taken in (the catalogue's is held; the sdk's is not known taken): the wait stays. + taken(catalogueRepo, sha('7')) + if err := b.ix.PendingAgain(ctx); err != nil || len(b.keys()) != 2 || len(b.said) != 0 { + t.Fatalf("a member not taken in: err %v, pending %v, said %v: want the wait kept", err, b.keys(), b.events()) + } + // Every member is taken in: the next evaluation ends the wait (said under the group key) and takes the merge behind it in. + taken("novox/mesh-sdk", sha('b')) + 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"}) { + 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()) + } + // The start is asked again. A take-in that fails on a bus error once, then succeeds: one wait, said in the journal. + held := func() *takeInBed { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")) + b.ix.ownMu.Lock() + b.ix.ownFP = b.fp(t, delivery.IndexerModule) + b.ix.ownMu.Unlock() + return b + } + var waits []time.Duration + wait := func(context.Context, time.Duration) error { return nil } + count := func(_ context.Context, d time.Duration) error { waits = append(waits, d); return wait(ctx, d) } + b = held() + fails := 1 + b.mods.failWrite = func(string) error { + if fails > 0 { + fails-- + return errors.New("nats: jetstream not enabled") + } + return nil + } + if _, _, err := b.ix.TakeInAtStartRetrying(ctx, count); err != nil || len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || !slices.Equal(waits, []time.Duration{StartBackoff}) || + !strings.Contains(b.logs.String(), "tried again in") { + t.Errorf("a bus error: err %v, pending %v, waits %v: want it taken in at the second try, the first said", err, b.keys(), waits) + } + // A merge lost because the forge bridge, restarted by the same delivery, does not answer: asked again once it does. + b, waits = held(), nil + answers := b.ix.Message + b.ix.held = nil + b.ix.Message = func(context.Context, string, string) ([]byte, error) { return nil, errors.New("timed out") } + wait = func(context.Context, time.Duration) error { + waits = append(waits, 0) + b.ix.Message = answers + return nil + } + if _, _, err := b.ix.TakeInAtStartRetrying(ctx, count); err != nil || len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || len(waits) != 2 || !strings.Contains(b.logs.String(), "tried again in") { + t.Errorf("a lost merge: err %v, pending %v, waits %v: want it taken in after the bridge answers, the loss said", err, b.keys(), waits) + } + // A start that never succeeds gives up after StartAttempts tries with the error, backing off by doubling. + b, waits, wait = held(), nil, func(context.Context, time.Duration) error { return nil } + b.mods.failWrite = func(string) error { return errors.New("nats: jetstream not enabled") } + if _, _, err := b.ix.TakeInAtStartRetrying(ctx, count); err == nil || len(waits) != StartAttempts-1 || waits[2] != 4*StartBackoff { + t.Errorf("never succeeds: err %v, waits %v: want %d tries and a doubling backoff", err, waits, StartAttempts) + } + }) t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) { b := newTakeInBed(t) b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 3e536b66..60dc39f8 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -212,6 +212,10 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1238/AC9 (ADR return err } for _, f := range forbidden { + // The one pause between the tries of the take-in at the start (bounded, one-shot) lives in startwait.go alone. + if f == "time.After(" && filepath.ToSlash(path) == "cmd/mesh-delivery-indexer/startwait.go" { + continue + } if strings.Contains(string(src), f) { t.Errorf("%s holds %s: the indexer serves no request subject and holds no timer", path, f) } From 0520c126195be347b4a32383d4bd891089abcb9b Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 7 Nov 2026 12:29:40 +0100 Subject: [PATCH 2/3] Group-end probe is best effort and a failed start is asked again by the next event, with no timer (issue 1585, review round 1) --- .../cmd/mesh-delivery-indexer/main.go | 6 +- .../cmd/mesh-delivery-indexer/startwait.go | 18 ------ .../internal/index/pending.go | 57 +++--------------- .../internal/index/pending_test.go | 60 ++++++++++++------- .../mesh-delivery-indexer/manifest_test.go | 4 -- 5 files changed, 48 insertions(+), 97 deletions(-) delete mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 1bc29462..e006cfd4 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -510,9 +510,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // stream starts (ADR 0633 §2). When the build this program runs is not known yet, nothing waits for it: the consumer // starts, and a merged message left unacknowledged asks the build again at its redelivery (clause 1). A hand take-in // (indexer_take_in_pending) is answered at any time: the takeins watch runs already. - // A failed take-in is tried again with a backoff (issue 1585, note 70); one that stays failed does not end the process: a - // merged message arriving, a toolchain report or indexer_take_in_pending asks again. - if _, words, err := ix.TakeInAtStartRetrying(ctx, sleepFor); err != nil { + // A take-in that fails (a bus error, or a merge lost because the forge bridge did not answer) does not end the process (issue + // 1585, note 70): what pending holds is evaluated again at the next merged message, toolchain report or hand take-in. + if _, words, err := ix.TakeInAtStart(ctx); err != nil { logger.Printf("the take-in at start failed and is asked again at the next merged message, toolchain report or hand take-in: %v", err) } else if words != "" { logger.Printf("take-in at start: %s", words) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go deleted file mode 100644 index 81e515e3..00000000 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go +++ /dev/null @@ -1,18 +0,0 @@ -package main - -import ( - "context" - "time" -) - -// sleepFor is the one wait the indexer holds: the pause between the tries of the take-in at its start (issue 1585, note 70), -// one-shot and bounded by index.StartAttempts. It is no ticker and polls nothing; it is kept in this file alone, which -// TestServesNoRequestSubjectAndHoldsNoTicker names. -func sleepFor(ctx context.Context, d time.Duration) error { - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(d): - return nil - } -} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 6cb28bc7..34c0c766 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -526,7 +526,9 @@ func (ix *Indexer) endTakenGroups(ctx context.Context, entries []PendingEntry) ( held := w.Groups[group] all, err := ix.groupTaken(ctx, held) if err != nil { - return ended, err + // Best effort: a probe that fails keeps this group's wait and stops nothing else; the next evaluation asks again. + ix.logf("take-in: whether the group %q is taken in cannot be read, so its wait is kept: %v", group, err) + continue } if !all { continue @@ -591,7 +593,11 @@ func (ix *Indexer) headTaken(ctx context.Context, m delivery.Head) (bool, error) if cur.Merge == m.Head { return true, nil } - return ix.Contains(ctx, m.Repository, m.Head, cur.Merge) + in, err := ix.Contains(ctx, m.Repository, m.Head, cur.Merge) + if notFound(err) { + return false, nil // the forge does not have it (a rewritten history?): not known to be taken, as older has it + } + return in, err } // takenAlready says whether a group member's merge is at or behind its repository's taken merge. @@ -649,53 +655,6 @@ func (ix *Indexer) TakeInAtStart(ctx context.Context) (Applied, string, error) { return a, words, err } -// The take-in at the start is asked again when it fails or leaves a merge lost (issue 1585, note 70): a bus error, or the -// forge bridge not answering because it was restarted by the same delivery, is transient. Each try is said. -const ( - StartAttempts = 6 - StartBackoff = time.Second -) - -// TakeInAtStartRetrying is TakeInAtStart tried up to StartAttempts times with a doubling backoff (StartBackoff, then -// twice that, ...), waited with wait (it returns when the context ends), each failure and each lost merge said in the -// journal. It answers the last try's result; an error is that of the last try, which the caller says and goes on from: -// a merged message arriving, a toolchain report or a hand take-in asks again. -func (ix *Indexer) TakeInAtStartRetrying(ctx context.Context, wait func(context.Context, time.Duration) error) (Applied, string, error) { - back := StartBackoff - for attempt := 1; ; attempt++ { - a, words, err := ix.TakeInAtStart(ctx) - problem := "" - switch { - case err != nil: - problem = err.Error() - default: - if ix.Pending == nil { - break - } - if entries, rerr := ReadPending(ctx, ix.Pending, ix.logf); rerr == nil { - for _, e := range entries { - if e.Value.Lost != "" { - problem = fmt.Sprintf("%s (%s#%d) is lost: %s", e.Key, e.Forge.Repository, e.Forge.Number, e.Value.Lost) - break - } - } - } - } - if problem == "" || ctx.Err() != nil { - return a, words, err - } - if attempt >= StartAttempts { - ix.logf("take-in: at the start, try %d of %d failed (%s): not tried again now", attempt, StartAttempts, problem) - return a, words, err - } - ix.logf("take-in: at the start, try %d of %d failed (%s): tried again in %s", attempt, StartAttempts, problem, back) - if wait(ctx, back) != nil { - return a, words, err - } - back *= 2 - } -} - // PendingAgain evaluates pending again after something a fingerprint is computed from changed, a toolchain report // (ADR 0633 §1, §3): a revision that now names the running build is taken in, and one that names another build is held // for that one. It does nothing while the pending state is not on the bus or the build this program runs is not known. diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index ee728306..ce3190ab 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -8,7 +8,6 @@ import ( "slices" "strings" "testing" - "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" @@ -377,7 +376,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) t.Errorf("a group naming another indexer: said %v, want %v", b.events(), want) } }) - t.Run("1585 a group wait whose members are all taken in ends itself, behind it a merge is taken in at once, and a take-in at the start that fails is tried again", func(t *testing.T) { + t.Run("1585 a group wait whose members are all taken in ends itself, behind it a merge is taken in at once, and a start that fails is asked again by the next event", func(t *testing.T) { b := newTakeInBed(t) b.ix.Taken = newMem() order := "a756bc" @@ -411,7 +410,6 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') || !slices.Equal(b.events(), []string{"taken g.feat/take-in"}) { 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()) } - // The start is asked again. A take-in that fails on a bus error once, then succeeds: one wait, said in the journal. held := func() *takeInBed { b := newTakeInBed(t) b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) @@ -421,9 +419,29 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) b.ix.ownMu.Unlock() return b } - var waits []time.Duration - wait := func(context.Context, time.Duration) error { return nil } - count := func(_ context.Context, d time.Duration) error { waits = append(waits, d); return wait(ctx, d) } + // A probe that fails keeps the group's wait and stops nothing else: an unrelated merge is still taken in. + b = newTakeInBed(t) + b.ix.Taken = newMem() + b.ix.Contains = func(context.Context, string, string, string) (bool, error) { + return false, errors.New("the forge does not answer") + } + g := b.merge(40, sha('7'), "", "") + g.Group, g.Head = "feat/take-in", sha('1') + g.Members = []delivery.Head{{Repository: catalogueRepo, Number: 40, Head: sha('1')}, {Repository: "novox/mesh-sdk", Number: 41, Head: sha('2')}} + b.arrive(t, 1, g) + key, _ := delivery.RepositoryKey("novox/mesh-sdk") + data, _ := delivery.EncodeTaken(delivery.Taken{Repository: delivery.RepositoryName("novox/mesh-sdk"), Merge: sha('b')}) + b.ix.Taken.(*memStore).put(key, data) + key, _ = delivery.RepositoryKey(catalogueRepo) + data, _ = delivery.EncodeTaken(delivery.Taken{Repository: delivery.RepositoryName(catalogueRepo), Merge: sha('7')}) + b.ix.Taken.(*memStore).put(key, data) + alone := b.merge(60, sha('5'), "", "") + alone.Repository, alone.Trees, alone.Manifests, alone.Locks = "novox/hq", map[string]string{}, nil, nil + b.arrive(t, 2, alone) + if len(b.keys()) != 1 || b.keys()[0] != takein.PendingKey(1) || !strings.Contains(b.logs.String(), "cannot be read, so its wait is kept") { + t.Errorf("a failing probe: pending %v: want the group kept, said, and the unrelated merge taken in", b.keys()) + } + // A start that fails leaves its merges pending; the next event the indexer hears evaluates them again (no timer). b = held() fails := 1 b.mods.failWrite = func(string) error { @@ -433,28 +451,24 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) } return nil } - if _, _, err := b.ix.TakeInAtStartRetrying(ctx, count); err != nil || len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || !slices.Equal(waits, []time.Duration{StartBackoff}) || - !strings.Contains(b.logs.String(), "tried again in") { - t.Errorf("a bus error: err %v, pending %v, waits %v: want it taken in at the second try, the first said", err, b.keys(), waits) + if _, _, err := b.ix.TakeInAtStart(ctx); err == nil || len(b.keys()) != 2 { + t.Fatalf("a failing start: err %v, pending %v: want the error and both merges pending", err, b.keys()) } - // A merge lost because the forge bridge, restarted by the same delivery, does not answer: asked again once it does. - b, waits = held(), nil + if err := b.ix.PendingAgain(ctx); err != nil || len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') { + t.Errorf("the next event: err %v, pending %v, avahi at %s: want both taken in", err, b.keys(), b.commit(t, "avahi")) + } + // A start that leaves a merge lost (the forge bridge restarted) is asked again by the next merged message. + b = held() answers := b.ix.Message b.ix.held = nil b.ix.Message = func(context.Context, string, string) ([]byte, error) { return nil, errors.New("timed out") } - wait = func(context.Context, time.Duration) error { - waits = append(waits, 0) - b.ix.Message = answers - return nil + if _, _, err := b.ix.TakeInAtStart(ctx); err != nil || len(b.keys()) != 2 { + t.Fatalf("a lost start: err %v, pending %v", err, b.keys()) } - if _, _, err := b.ix.TakeInAtStartRetrying(ctx, count); err != nil || len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || len(waits) != 2 || !strings.Contains(b.logs.String(), "tried again in") { - t.Errorf("a lost merge: err %v, pending %v, waits %v: want it taken in after the bridge answers, the loss said", err, b.keys(), waits) - } - // A start that never succeeds gives up after StartAttempts tries with the error, backing off by doubling. - b, waits, wait = held(), nil, func(context.Context, time.Duration) error { return nil } - b.mods.failWrite = func(string) error { return errors.New("nats: jetstream not enabled") } - if _, _, err := b.ix.TakeInAtStartRetrying(ctx, count); err == nil || len(waits) != StartAttempts-1 || waits[2] != 4*StartBackoff { - t.Errorf("never succeeds: err %v, waits %v: want %d tries and a doubling backoff", err, waits, StartAttempts) + b.ix.Message = answers + b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", "tree-indexer-2")) + if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('e') { + t.Errorf("after a lost start: pending %v, avahi at %s: want all taken in by the next merged message", b.keys(), b.commit(t, "avahi")) } }) t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) { diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 60dc39f8..3e536b66 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -212,10 +212,6 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1238/AC9 (ADR return err } for _, f := range forbidden { - // The one pause between the tries of the take-in at the start (bounded, one-shot) lives in startwait.go alone. - if f == "time.After(" && filepath.ToSlash(path) == "cmd/mesh-delivery-indexer/startwait.go" { - continue - } if strings.Contains(string(src), f) { t.Errorf("%s holds %s: the indexer serves no request subject and holds no timer", path, f) } From 70cd624127bf8382b49e496a12d9a97cb34c2368 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 7 Nov 2026 12:38:40 +0100 Subject: [PATCH 3/3] Ask the check again: its composition ran before the indexer's build b5163570 was done, which is now (issue 1585; the checker's skipped build wait is noted on issue 1921)