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