diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index 03c38f7..b79ba6b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -98,7 +98,7 @@ func askTakeIn(s stdio.KeptState, bound time.Duration) asker { case a := <-answered: return a, nil case <-wait.Done(): - return index.TakeInAsk{}, fmt.Errorf("the indexer did not answer %s.%s within %s: it is not running, or watches no %s (it says so in its journal at start)", index.TakeInsState, key, bound, index.TakeInsBucket) + return index.TakeInAsk{}, fmt.Errorf("the indexer did not answer within %s: the take-in is still running (a forge read can take 30 s); its answer will be in %s.%s", bound, index.TakeInsBucket, key) } } } diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index b637c4a..592b309 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -8,6 +8,7 @@ import ( "slices" "strings" "sync" + "sync/atomic" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -66,6 +67,9 @@ type Indexer struct { toolchains map[string]map[string]delivery.Toolchain // language, then node (toolchains.go) tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go) unread map[string]bool // registered modules whose last read the forge could not answer, read again by RetryUnread + retrying atomic.Bool // a RetryUnread runs in the background (kickRetry) + retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one + retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on recompute sync.Mutex // one Recompute at a time } @@ -90,7 +94,7 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { if err := ix.applyMerged(ctx, f); err != nil { return err } - ix.RetryUnread(ctx) // a registered module the forge could not read is read again on a merge (issue 1945) + ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) } var digests map[string]string var held map[string]delivery.Module diff --git a/modules/mesh-delivery-indexer/internal/index/registered.go b/modules/mesh-delivery-indexer/internal/index/registered.go index ae96305..e83fb4f 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered.go +++ b/modules/mesh-delivery-indexer/internal/index/registered.go @@ -159,6 +159,34 @@ func (ix *Indexer) RetryUnread(ctx context.Context) { } } +// kickRetry runs RetryUnread in one background goroutine, so the message or revision that asked it is acknowledged without +// waiting on the forge (each unread module costs up to two forge calls of 30 s each). A kick while one runs makes it run +// once more when it ends; never two at once. +func (ix *Indexer) kickRetry(ctx context.Context) { + ix.mu.Lock() + none := len(ix.unread) == 0 + ix.mu.Unlock() + if none { + return + } + ix.retryAgain.Store(true) + if !ix.retrying.CompareAndSwap(false, true) { + return + } + ix.retryWG.Add(1) + go func() { + defer ix.retryWG.Done() + for ix.retryAgain.Swap(false) { + ix.RetryUnread(ctx) + } + ix.retrying.Store(false) + // A kick between the last Swap and the Store found it running and returned: run for it. + if ix.retryAgain.Load() { + ix.kickRetry(ctx) + } + }() +} + // TakenIn is the answer of a take-in (the tool indexer_take_in): whether the module is now in the modules state, from // where, and otherwise why not. type TakenIn struct { diff --git a/modules/mesh-delivery-indexer/internal/index/registered_test.go b/modules/mesh-delivery-indexer/internal/index/registered_test.go index 64c554e..13eca05 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered_test.go +++ b/modules/mesh-delivery-indexer/internal/index/registered_test.go @@ -202,10 +202,12 @@ func TestAForgeThatCannotSayLeavesTheModuleUnwrittenAndAsksAgain(t *testing.T) { if err := ix.ApplySource(ctx, 2, delivery.Source{Heads: map[string]string{"jschoubben/snake": h2, "novox/other": h1}}); err != nil { t.Fatal(err) } + ix.retryWG.Wait() // read again in the background case "a merged message of another repository": if err := ix.Apply(ctx, mergedMsg("novox/other", map[string]string{"": "t9"})); err != nil { t.Fatal(err) } + ix.retryWG.Wait() // read again in the background case "a person's take-in": got = ix.TakeIn(ctx, "snake") if !got.Stored || got.Repository != "jschoubben/snake" || got.Commit != h2 || got.Path != "." || got.Answer == "" { @@ -220,6 +222,7 @@ func TestAForgeThatCannotSayLeavesTheModuleUnwrittenAndAsksAgain(t *testing.T) { if err := ix.ApplySource(ctx, 3, delivery.Source{Heads: map[string]string{"jschoubben/snake": h2}}); err != nil { t.Fatal(err) } + ix.retryWG.Wait() if len(asked) != n { t.Errorf("the forge was asked again after the module was taken in: %v", asked) } diff --git a/modules/mesh-delivery-indexer/internal/index/source.go b/modules/mesh-delivery-indexer/internal/index/source.go index d275ad5..9b3e409 100644 --- a/modules/mesh-delivery-indexer/internal/index/source.go +++ b/modules/mesh-delivery-indexer/internal/index/source.go @@ -47,9 +47,10 @@ func (ix *Indexer) ApplySource(ctx context.Context, rev uint64, s delivery.Sourc if ix.Modules == nil { return ix.RecomputeCandidates(ctx) } - // A registered module the forge could not read is read again at each new source revision (issue 1945), before the - // stamping below, so one taken in at the head is stamped too. - ix.RetryUnread(ctx) + // A registered module the forge could not read is read again at each new source revision (issue 1945), in the + // background: a forge that does not answer must not hold this revision. One taken in later is stamped when it is written + // (applyManifests reads srcOf). + ix.kickRetry(ctx) held, err := ix.heldModules(ctx) if err != nil { return err