A group wait whose members are all taken in ends itself, and the take-in at the start is tried again (issue 1585)

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

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

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

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

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