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..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,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. + // 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 { - return fmt.Errorf("the take-in at start: %w", err) + 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/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index ccc5673f..34c0c766 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -511,6 +511,95 @@ 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 { + // 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 + } + 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 + } + 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. func (ix *Indexer) takenAlready(ctx context.Context, m forgemsg.GroupMerge) (bool, error) { if ix.Taken == nil { @@ -595,6 +684,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..ce3190ab 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -376,6 +376,101 @@ 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 start that fails is asked again by the next event", 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()) + } + 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 + } + // 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 { + if fails > 0 { + fails-- + return errors.New("nats: jetstream not enabled") + } + return nil + } + 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()) + } + 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") } + if _, _, err := b.ix.TakeInAtStart(ctx); err != nil || len(b.keys()) != 2 { + t.Fatalf("a lost start: err %v, pending %v", err, b.keys()) + } + 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) { b := newTakeInBed(t) b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))