Merge pull request 'Take-in waits end: a group whose members are all taken in ends itself, a failed start is asked again by the next event (issue 1585)' (#163) from fix/1585-waits-end into main

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)
 	}

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

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"))