novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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)
mesh/delivery closed: a newer head of the same pull request
Details
mesh/check failed: novox: not composed on novox: mesh-delivery-indexer: frombus: mesh-delivery-indexer is wanted at src… | 0520c126195b on bab05c66b3cb

Browse Source
This commit is contained in:
jschoubben
2026-11-07 12:29:40 +01:00
parent a7aa836ff9
commit 0520c12619
5 changed files with 48 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+3 -3
View File
@@ -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)
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/startwait.go
-18
View File
@@ -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
}
}
modules/mesh-delivery-indexer/internal/index/pending.go
+8 -49
View File
@@ -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.
modules/mesh-delivery-indexer/internal/index/pending_test.go
+37 -23
View File
@@ -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) {
modules/mesh-delivery-indexer/manifest_test.go
-4
View File
@@ -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)
}