mesh-delivery-indexer reads unread registered modules again in one background run, so a forge timeout holds no merged message or source revision past its ack wait (issue 1945, review round 1)

The take-in tool's bound now says the take-in is still running and where its answer will be.

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

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

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 {

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

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