From a7aa836ff9ac565d6f2cca1149c7df7b24a70700 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 7 Nov 2026 12:25:26 +0100 Subject: [PATCH] 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) }