modules/mesh-delivery-checker/internal/check/case.go

@@ -115,7 +115,7 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case {
 	for _, p := range c.Pulls {
 		switch head, known := trunk[p.Repository]; {
 		case !known || head == "":
-			cs.Left = append(cs.Left, fmt.Sprintf("%s#%d: the trunk head of %s is not known", p.Repository, p.Number, p.Repository))
+			cs.Left = append(cs.Left, fmt.Sprintf("%s#%d: %s", p.Repository, p.Number, UntrunkedWordsOf([]string{p.Repository})))
 		default:
 			cs.Pulls = append(cs.Pulls, Pull{Repository: p.Repository, Number: p.Number, Head: p.Head.Head, Trunk: head, Modules: p.Modules, Fps: p.Fps})
 			cs.Trunk[p.Repository] = head
@@ -133,6 +133,42 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case {
 	return cs
 }
 
+// UntrunkedWords open the words of a verdict, and of a status, of a pull request that cannot be judged because the gate's
+// source.main names no trunk head of its repository (issue 1422): never a failure of the change, and judged whole once the
+// trunk head is known. It is said loudly, never left silently "new".
+const UntrunkedWords = "cannot be judged: the trunk head of "
+
+// UntrunkedWordsOf are those words for the repositories whose trunk head is not known, in order, each named once.
+func UntrunkedWordsOf(repos []string) string {
+	rs := slices.Clone(repos)
+	slices.Sort(rs)
+	rs = slices.Compact(rs)
+	return UntrunkedWords + strings.Join(rs, ", ") + " is not known: the gate's source.main names none, so there is no commit to merge it onto"
+}
+
+// IsUntrunked says whether a verdict says its heads cannot be judged for want of a trunk head (UntrunkedWords).
+func IsUntrunked(v delivery.Verdict) bool {
+	return v.State == delivery.VerdictFail && v.Failed.Machine == "checker" && strings.HasPrefix(v.Failed.Words, UntrunkedWords)
+}
+
+// leftOf are the pull requests of a candidate whose repository has no trunk head known, with Trunk empty, in repository and
+// number order; nil when every repository's is known.
+func leftOf(c delivery.Candidate, trunk map[string]string) []Pull {
+	var out []Pull
+	for _, p := range c.Pulls {
+		if trunk[p.Repository] == "" {
+			out = append(out, Pull{Repository: p.Repository, Number: p.Number, Head: p.Head.Head, Modules: p.Modules, Fps: p.Fps})
+		}
+	}
+	sort.SliceStable(out, func(i, j int) bool {
+		if out[i].Repository != out[j].Repository {
+			return out[i].Repository < out[j].Repository
+		}
+		return out[i].Number < out[j].Number
+	})
+	return out
+}
+
 // GateHeads is the hash the gate matches the members an Approve covers by (mesh-catalog mesh-delivery-gate onestatus.go
 // headsOf): delivery.HeadsHash over every pull request of the candidate, hq's included, as
 // `<owner/name lowercase>#<number>=<full head sha lowercase>`. A lone pull request is a group of one. Empty for a candidate

modules/mesh-delivery-checker/internal/check/checker.go

@@ -295,9 +295,13 @@ func (ch *Checker) heads() map[string]string {
 }
 
 func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidate) {
-	cs := CaseOf(c, ch.heads())
+	heads := ch.heads()
+	cs := CaseOf(c, heads)
+	if left := leftOf(c, heads); len(left) > 0 {
+		ch.untrunked(ctx, c, left, cs == nil)
+	}
 	if cs == nil {
-		return // nothing to judge: no trunk known
+		return // nothing to judge: no trunk known, which untrunked said
 	}
 	cs.Queued = ch.now()
 	ch.mu.Lock()
@@ -351,6 +355,79 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
 	}()
 }
 
+// untrunked says, loudly, that pull requests of a candidate cannot be judged because the gate's source.main names no trunk
+// head of their repository (issue 1422): each gets the status error with UntrunkedWords, and when none of the candidate is
+// judged (whole) its verdict key holds a verdict that says so, which checker_verdicts and checker_status show. It is logged
+// once per head: when a status or the verdict is written, never at a look that finds them already said. A verdict of these
+// heads that says anything else is left. Once the trunk head is known the verdict is stale (it names no trunk) and the heads
+// are judged whole (prepare, claim).
+func (ch *Checker) untrunked(ctx context.Context, c delivery.Candidate, left []Pull, whole bool) {
+	key, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
+	if err != nil {
+		return
+	}
+	repos := make([]string, 0, len(left))
+	for _, p := range left {
+		repos = append(repos, p.Repository)
+	}
+	words := UntrunkedWordsOf(repos)
+	said := false
+	if whole {
+		v := delivery.Verdict{Group: c.Group, Heads: GateHeads(c), Trunk: map[string]string{}, State: delivery.VerdictFail,
+			Failed: delivery.FailedFirst{Machine: "checker", Words: words}}
+		data, err := delivery.EncodeVerdict(v)
+		if err != nil {
+			ch.logf("%s: %s, and the verdict cannot say so: %v", key, words, err)
+			return
+		}
+		wrote, err := putOnce(ctx, ch.Verdicts, key, data, func(cur []byte) bool {
+			held, err := delivery.DecodeVerdict(cur)
+			// The same words stand; a verdict that says anything else (a check that ran, one running) is not this one's to replace.
+			return err != nil || !IsUntrunked(held) || held.Failed.Words == words
+		})
+		if err != nil {
+			ch.logf("%s: %s, and the verdict cannot say so: %v", key, words, err)
+		}
+		said = said || wrote
+	}
+	for _, p := range left {
+		okey, err := delivery.CheckerOwedKey(p.Repository, p.Number, p.Head)
+		if err != nil {
+			ch.logf("%s: %v", key, err)
+			continue
+		}
+		o := delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "error",
+			Target: strings.TrimSuffix(ch.Link, "/") + "/" + key, Text: clipLine(words, MaxLine)}
+		data, err := delivery.EncodeCheckerOwed(o)
+		if err != nil {
+			ch.logf("%s: the status of %s cannot be written: %v", key, okey, err)
+			continue
+		}
+		wrote, err := putOnce(ctx, ch.Owed, okey, data, func(cur []byte) bool {
+			held, err := delivery.DecodeCheckerOwed(cur)
+			o.V = delivery.Version
+			return err == nil && delivery.SameOwed(held, o)
+		})
+		if err != nil {
+			ch.logf("%s: the status of %s cannot be written: %v", key, okey, err)
+			continue
+		}
+		if wrote {
+			said = true
+			if ch.Announce != nil {
+				ch.Announce(ctx, EventOwedWritten, okey)
+			}
+		}
+	}
+	if said {
+		names := make([]string, 0, len(left))
+		for _, p := range left {
+			names = append(names, fmt.Sprintf("%s#%d@%s", p.Repository, p.Number, short12(p.Head)))
+		}
+		ch.logf("%s: %s %s; the status says so (error), and they are judged once the trunk head is known", key, strings.Join(names, ", "), words)
+	}
+}
+
 // The kinds of a check (ADR 0536 §4): a head never judged and a head asked again after a failure of the nodes are asked on
 // the build seats' check stream; a carry, a verdict moved to a newer trunk or sibling commit by its package digests (ADR 0548
 // §5), on the recheck stream, pulled after.
@@ -380,8 +457,8 @@ func (ch *Checker) prepare(ctx context.Context, cs *Case) {
 		return
 	case IsConflictFailure(v):
 		return // a head git refused: its next check is whole (ADR 0548 §7)
-	case IsEnded(v):
-		return // ended unjudged when its heads went (issue 1890): held again, they are checked whole
+	case IsEnded(v), IsUntrunked(v):
+		return // ended unjudged when its heads went (issue 1890), or never judged for want of a trunk head (issue 1422): checked whole
 	}
 	cs.Kind = KindTrunk
 	if v.State == delivery.VerdictPass && ch.stale(v, cs) {

modules/mesh-delivery-checker/internal/check/checker_test.go

@@ -239,14 +239,80 @@ func TestAnUnheardForgeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *t
 	}
 }
 
-func TestARepositoryWithNoTrunkGetsNoVerdict(t *testing.T) { // N6/AC6, ADR 0484 clause 11
+// Issue 1422: a pull request of a repository the gate's source.main names no trunk head of (one never merged through the
+// gate) is never dropped silently. It is not judged, its verdict says it cannot be judged and why, its status is error with
+// those words, checker_status names it, and the log says it once per head. Once its trunk head is known it is judged whole;
+// in a group, the members whose trunk is known are judged and the others' status says why they are not.
+func TestARepositoryWithNoTrunkCannotBeJudgedLoudlyUntilItsTrunkIsKnown(t *testing.T) { // N6/AC6, ADR 0484 clause 11, 1422
 	ctx, j := context.Background(), passJudge()
 	r := newRig(t, j)
-	r.ch.Candidate(ctx, "feat/y", candidateOf(t, "feat/y", pullOf("novox/unknown", 4, head1)))
+	var logged []string
+	r.ch.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
+	const unknown = "novox/unknown"
+	cand := candidateOf(t, "feat/y", pullOf(unknown, 4, head1))
+	key := keyOf(t, "feat/y", cand)
+	r.ch.Candidate(ctx, "feat/y", cand)
 	r.ch.Wait()
-	if j.count() != 0 || len(r.verdicts.vals) != 0 || len(r.owed.vals) != 0 {
-		t.Errorf("judged %d times, %d verdicts, %d statuses, want none", j.count(), len(r.verdicts.vals), len(r.owed.vals))
+	r.ch.Reconsider(ctx) // looked at again: nothing is written or said twice
+	r.ch.NodesChanged(ctx)
+	r.ch.Wait()
+	v := r.verdict(t, key)
+	if j.count() != 0 || !IsUntrunked(v) || !strings.Contains(v.Failed.Words, "the trunk head of "+unknown+" is not known") {
+		t.Fatalf("judged %d times, verdict %+v, want none judged and a verdict that says it cannot be judged", j.count(), v)
 	}
+	if s := owedState(t, r, unknown, 4, head1); s != "error" {
+		t.Errorf("%s#4 owes %q, want error", unknown, s)
+	}
+	if n := countWith(logged, "cannot be judged"); n != 1 {
+		t.Errorf("said %d times in the log, want once for the head: %q", n, logged)
+	}
+	reader := func() Reader {
+		ckey, _ := delivery.CandidatesKey("feat/y")
+		return Reader{CandidateState: fakeState{ckey: cand}.read(), VerdictState: fakeState(r.verdicts.vals).read(),
+			Trunks: func() (map[string]string, error) { return r.trunks.Heads(), nil }}
+	}
+	st, err := reader().Status()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if why := st.CannotBeJudged["feat/y"]; !strings.Contains(why, unknown) || len(st.Waiting[KindNew]) != 0 || len(st.Unjudged) != 0 {
+		t.Errorf("cannot_be_judged %v, waiting %v, unjudged %v: want feat/y named as cannot be judged, not as new", st.CannotBeJudged, st.Waiting, st.Unjudged)
+	}
+	if rows, _, err := reader().Verdicts("feat/y"); err != nil || len(rows) != 1 || !strings.Contains(rows[0].Reports[unknown+"#4"], "cannot be judged") {
+		t.Errorf("checker_verdicts %+v (%v), want the report of %s#4 to say it cannot be judged", rows, err, unknown)
+	}
+
+	// Its trunk head becomes known: the heads are judged whole, the same key rewritten, the status success.
+	r.setTrunks(t, map[string]string{repoA: trunkA, repoSDK: trunkS, unknown: trunkA2})
+	r.ch.TrunkMoved(ctx)
+	r.ch.Wait()
+	if v := r.verdict(t, key); j.count() != 1 || v.State != delivery.VerdictPass || v.Trunk[unknown] != trunkA2 {
+		t.Errorf("judged %d times, verdict %+v, want one pass on %s", j.count(), v, trunkA2)
+	}
+	if s := owedState(t, r, unknown, 4, head1); s != "success" {
+		t.Errorf("%s#4 owes %q once its trunk is known, want success", unknown, s)
+	}
+
+	// A group with one member of a repository whose trunk is not known: the others are judged, its status says why it is not.
+	group := candidateOf(t, "feat/z", pullOf(repoA, 9, head2), pullOf("novox/other", 2, head2))
+	r.ch.Candidate(ctx, "feat/z", group)
+	r.ch.Wait()
+	if s := owedState(t, r, repoA, 9, head2); s != "success" {
+		t.Errorf("%s#9 owes %q, want success", repoA, s)
+	}
+	if s := owedState(t, r, "novox/other", 2, head2); s != "error" {
+		t.Errorf("novox/other#2 owes %q, want error", s)
+	}
+}
+
+func countWith(lines []string, words string) int {
+	n := 0
+	for _, l := range lines {
+		if strings.Contains(l, words) {
+			n++
+		}
+	}
+	return n
 }
 
 // 1887/A10, ADR 0527 §2: hq is judged as any repository, its checks Go tests, and its pull request gets a verdict and a

modules/mesh-delivery-checker/internal/check/read.go

@@ -86,6 +86,11 @@ func (r Reader) read(key string, raw json.RawMessage) (VerdictRead, error) {
 	cs := CaseOf(c, v.Trunk)
 	out.Reports = map[string]string{}
 	if cs == nil {
+		if IsUntrunked(v) {
+			for _, p := range leftOf(c, v.Trunk) {
+				out.Reports[pullKey(p)] = v.Failed.Words
+			}
+		}
 		return out, nil
 	}
 	for _, p := range cs.Pulls {
@@ -323,6 +328,9 @@ type Status struct {
 	// HeldByTurn are the candidates whose check an approved merge's turn holds off (ADR 0548 §3), by group: the repository and
 	// the group of the merge that waits. Their check is claimed when that turn goes.
 	HeldByTurn map[string]string `json:"held_by_turn"`
+	// CannotBeJudged are the candidates no pull request of which can be judged, by group, with why: the gate's source.main
+	// names no trunk head of their repositories (issue 1422). They are judged once it does; until then their status is error.
+	CannotBeJudged map[string]string `json:"cannot_be_judged"`
 	// DeadLetters says whether the checker hears the asks the build seats give up on (issue 734): deaf, a try or check
 	// whose ask is given up waits until its bound.
 	DeadLetters DeadLettersHeard `json:"dead_letters"`
@@ -452,7 +460,7 @@ func (r Reader) Status() (Status, error) {
 	}
 	st.Unjudged = []string{}
 	st.Waiting = map[string][]string{KindNew: {}, KindNodes: {}, KindTrunk: {}}
-	st.HeldByTurn = map[string]string{}
+	st.HeldByTurn, st.CannotBeJudged = map[string]string{}, map[string]string{}
 	turns := r.turns(&st)
 	var heads map[string]string
 	if r.Trunks == nil {
@@ -503,6 +511,15 @@ func (r Reader) Status() (Status, error) {
 				continue
 			}
 			c, raw := row.c, row.raw
+			if heads != nil && CaseOf(c, heads) == nil {
+				// Nothing of it can be judged now, whatever its verdict says (issue 1422): never counted as waiting.
+				var repos []string
+				for _, p := range leftOf(c, heads) {
+					repos = append(repos, p.Repository)
+				}
+				st.CannotBeJudged[c.Group] = UntrunkedWordsOf(repos)
+				continue
+			}
 			if len(raw) == 0 {
 				st.Unjudged = append(st.Unjudged, c.Group)
 				st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group)
@@ -516,6 +533,9 @@ func (r Reader) Status() (Status, error) {
 			case err != nil || v.State == delivery.VerdictRunning:
 			case IsNodesFailure(v):
 				st.Waiting[KindNodes] = append(st.Waiting[KindNodes], c.Group)
+			case IsUntrunked(v):
+				// Its trunk head is known now: it waits for its whole check.
+				st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group)
 			case heads != nil:
 				if cs := CaseOf(c, heads); cs != nil && (!sameTrunk(v.Trunk, cs.Trunk) || besideMoved(v.Beside, besideAt(heads, r.Runs))) {
 					st.Waiting[KindTrunk] = append(st.Waiting[KindTrunk], c.Group)

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go

@@ -72,11 +72,17 @@ accounts {
             "mesh.mod.gitea.tool.gitea_open_pulls"
             "mesh.mod.gitea.tool.gitea_open_pulls.*"
             "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_inputs.>"
             "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_inputs.>"
             "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_inputs.>"
             "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_performed"
+            "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_inputs"
             "$JS.FC.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.FC.KV_mesh-delivery-forge-bridge_inputs.>"
             "$KV.mesh-delivery-forge-bridge_performed.>"
+            "$KV.mesh-delivery-forge-bridge_inputs.>"
             "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-gate_merges.>"
             "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-gate_merges.>"
             "$JS.API.DIRECT.GET.KV_mesh-delivery-gate_merges.>"
@@ -219,7 +225,7 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
 	}
 	// Its other state bucket, and the gate's buckets the bridge reads, made by the controller from the gate's manifest
 	// (mesh-sdk go/delivery MergesBucket, OwedBucket with their histories).
-	for name, history := range map[string]uint8{performedBucket: 1, delivery.MergesBucket: delivery.MergesHistory, delivery.OwedBucket: delivery.OwedHistory} {
+	for name, history := range map[string]uint8{performedBucket: 1, inputsBucket: 1, delivery.MergesBucket: delivery.MergesHistory, delivery.OwedBucket: delivery.OwedHistory} {
 		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
 			t.Fatal(err)
 		}

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/kept.go

@@ -0,0 +1,81 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"regexp"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-forge-bridge/internal/bridge"
+)
+
+// inputsBucket is the module's state bucket inputs (manifest state entry inputs): what the bridge read of a module
+// directory, its manifest and lock-file hashes, by the directory's git tree hash, so a merge reads the forge only for the
+// directories it changed (issue 1126; internal/bridge kept.go). A tree hash names its bytes, so a value is never stale;
+// a merge adds a value only for each directory it changed (a few kilobytes each).
+const inputsBucket = delivery.ForgeBridgeModule + "_inputs"
+
+// keptTimeout bounds one read or write of the bucket: the bucket is a saving, never a reason to fail a merge.
+const keptTimeout = 2 * time.Second
+
+var treeHash = regexp.MustCompile(`^[0-9a-f]{40}$|^[0-9a-f]{64}$`)
+
+// kept is the bridge's Kept: in memory always, and in the bucket inputs when it is on the bus, so the first merge after a
+// start reads only what changed too. A value the bucket cannot give or take is read from the forge, and said.
+type kept struct {
+	kv     jetstream.KeyValue // nil when the bucket is not on the bus
+	logger *log.Logger
+	mu     sync.Mutex
+	mem    map[string]bridge.Inputs
+}
+
+func newKept(kv jetstream.KeyValue, logger *log.Logger) *kept {
+	return &kept{kv: kv, logger: logger, mem: map[string]bridge.Inputs{}}
+}
+
+func (k *kept) Get(tree string) (bridge.Inputs, bool) {
+	k.mu.Lock()
+	in, ok := k.mem[tree]
+	k.mu.Unlock()
+	if ok || k.kv == nil || !treeHash.MatchString(tree) {
+		return in, ok
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), keptTimeout)
+	defer cancel()
+	e, err := k.kv.Get(ctx, tree)
+	if err != nil {
+		return bridge.Inputs{}, false // not kept (or not readable): read from the forge
+	}
+	if err := json.Unmarshal(e.Value(), &in); err != nil || in.Manifest == "" {
+		k.logger.Printf("%s.%s cannot be read, so its directory is read from the forge: %v", inputsBucket, tree, err)
+		return bridge.Inputs{}, false
+	}
+	k.mu.Lock()
+	k.mem[tree] = in
+	k.mu.Unlock()
+	return in, true
+}
+
+func (k *kept) Put(tree string, in bridge.Inputs) {
+	k.mu.Lock()
+	k.mem[tree] = in
+	k.mu.Unlock()
+	if k.kv == nil || !treeHash.MatchString(tree) {
+		return
+	}
+	data, err := json.Marshal(in)
+	if err != nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), keptTimeout)
+	defer cancel()
+	if _, err := k.kv.Put(ctx, tree, data); err != nil {
+		k.logger.Printf("%s.%s is not kept, so the next merge reads its directory from the forge again: %v", inputsBucket, tree, err)
+	}
+}

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go

@@ -67,8 +67,8 @@ type publisher interface {
 // handle turns one event into its message on FORGE and acknowledges it. An event it cannot read is terminated, never
 // redelivered, and the log names its subject and why. A message it could not write stays unacknowledged: the server
 // delivers it again (the server's ack wait is the only clock).
-func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, logger *log.Logger) {
-	msg, err := bridge.ConvertWith(m.Subject(), m.Data(), trees, files, list)
+func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, kept bridge.Kept, logger *log.Logger) {
+	msg, err := bridge.ConvertKept(m.Subject(), m.Data(), trees, files, list, kept)
 	switch {
 	case errors.Is(err, bridge.ErrRetry):
 		logger.Printf("%v; heard again in %s", err, retryDelay)
@@ -96,6 +96,9 @@ func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees,
 		logger.Printf("%s: %s could not be written on FORGE, left for redelivery: %v", m.Subject(), msg.Event, err)
 		return
 	}
+	if msg.Said != "" {
+		logger.Printf("%s: %s", m.Subject(), msg.Said)
+	}
 	_ = m.Ack()
 }
 
@@ -160,6 +163,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	go onReconnect(ctx, again, said, logger)
+	// What a merge read of each module directory, by its tree (issue 1126): a merge reads only the directories it changed.
+	inputs := newKept(process.OptionalBucket(ctx, js, inputsBucket, "a merge's module directories are kept in memory only, so the first merge after a start reads every one", logger.Printf), logger)
 	cons, err := cred.EventConsumer(ctx, js, eventsStream)
 	if err != nil {
 		// Essential: the forge's events are what the bridge carries.
@@ -170,7 +175,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			// Bounded: the turn is held, and no sync or other event goes on without it.
 			hctx, cancel := context.WithTimeout(ctx, syncDeadline)
 			defer cancel()
-			handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), logger)
+			handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), inputs, logger)
 		})
 	}, logger.Printf)
 	if err != nil {

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go

@@ -81,7 +81,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit
 	})
 	var logs bytes.Buffer
 	m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{}
-	handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0))
 	if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull" {
 		t.Fatalf("published on %v", pub.subjects)
 	}
@@ -97,7 +97,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit
 func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5
 	var logs bytes.Buffer
 	m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(map[string]any{"head_sha": "short"})}, &fakePub{}
-	handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0))
 	if len(pub.subjects) != 0 {
 		t.Errorf("a message was written: %v", pub.subjects)
 	}
@@ -112,7 +112,7 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5
 func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is redelivered by the server, not retried here)
 	var logs bytes.Buffer
 	m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")}
-	handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0))
 	if m.acked != 0 || m.termd != 0 {
 		t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd)
 	}
@@ -134,7 +134,7 @@ func TestTreesRefusedLeavesTheEventForTenSecondsAndLogs(t *testing.T) { // 1236/
 	call := func(context.Context, string, any) (json.RawMessage, error) {
 		return nil, errors.New("the tree listing of novox/mesh-delivery was cut")
 	}
-	handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, nil, log.New(&logs, "", 0))
 	if len(pub.subjects) != 0 {
 		t.Errorf("a message was written: %v", pub.subjects)
 	}
@@ -163,7 +163,7 @@ func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/
 		}
 		return json.RawMessage(`{"root":"r0","trees":{"":"t0","modules/a":"t1"}}`), nil
 	}
-	handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, nil, log.New(&logs, "", 0))
 	if len(keys) == 0 || keys[0] != "gitea.gitea_commit_trees" || slices.Index(keys[1:], "gitea.gitea_commit_trees") >= 0 {
 		t.Fatalf("calls %v, want the trees once and first", keys)
 	}

modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go

@@ -54,6 +54,8 @@ var LockFiles = []string{"go.sum", "package-lock.json"}
 type Message struct {
 	Event string
 	Data  []byte
+	// Said is what the bridge did that the log says, beside writing the message; empty for none.
+	Said string
 }
 
 // The subjects of the forge module's events the bridge writes pull messages for (mesh-catalog modules/gitea,
@@ -107,7 +109,7 @@ func unreadable(subject, why string, args ...any) error {
 // the trunk asks trees once and becomes a merged message; a merge into any other branch is a closed pull message, as the
 // pull request left its group. A forge that cannot say is ErrRetry; a merge with no module.json is ErrNoModules.
 func Convert(subject string, data []byte, trees Trees, files Files) (Message, error) {
-	return ConvertWith(subject, data, trees, files, nil)
+	return ConvertKept(subject, data, trees, files, nil, nil)
 }
 
 // Listed asks the forge for the files directly under one directory of a commit, by name to blob id (gitea_commit_trees
@@ -120,6 +122,12 @@ type Listed func(owner, repo, sha, dir string) (map[string]string, error)
 // updated pull request of the catalogue that touches placement/ (touchesPlacement), with every placement file at its
 // head, which the indexer writes as placement candidates (ADR 0549 §4: the nodes it places answer it before it merges).
 func ConvertWith(subject string, data []byte, trees Trees, files Files, list Listed) (Message, error) {
+	return ConvertKept(subject, data, trees, files, list, nil)
+}
+
+// ConvertKept is ConvertWith; with kept, a merge reads the forge only for the module directories whose tree hash kept
+// does not hold, and keeps what it read (kept.go).
+func ConvertKept(subject string, data []byte, trees Trees, files Files, list Listed, kept Kept) (Message, error) {
 	var action string
 	switch subject {
 	case UpdatedSubject:
@@ -146,7 +154,16 @@ func ConvertWith(subject string, data []byte, trees Trees, files Files, list Lis
 		return Message{}, unreadable(subject, "no branch")
 	}
 	if subject == MergedSubject && p.Base == Trunk {
-		return merged(subject, p, trees, files, list)
+		return merged(subject, p, trees, files, list, kept)
+	}
+	said := ""
+	if action == delivery.PullUpdated && p.Base != "" && p.Base != Trunk {
+		// The checker judges pull requests into main only, and source.main names main's head only (novox/hq ADR 0484 §11): a
+		// pull request into another base is no candidate, as it is none at the start (OpenPulls) and at its merge. It is
+		// written closed, so a candidate an older head made is taken out, and said (issue 1422).
+		action = delivery.PullClosed
+		said = fmt.Sprintf("%s/%s#%d at %s is into %s, not %s: written closed, so it is no candidate; the checker judges pull requests into %s only (novox/hq ADR 0484 §11)",
+			p.Owner, p.Repo, p.Number, p.HeadSHA[:12], p.Base, Trunk, Trunk)
 	}
 	f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number,
 		Head: p.HeadSHA, Action: action}
@@ -173,7 +190,7 @@ func ConvertWith(subject string, data []byte, trees Trees, files Files, list Lis
 	if err != nil {
 		return Message{}, unreadable(subject, "%v", err)
 	}
-	return Message{Event: event, Data: body}, nil
+	return Message{Event: event, Data: body, Said: said}, nil
 }
 
 // MaxReadsInFlight is how many files of the forge inputsOf asks for at once. A merge into a repository of many
@@ -186,6 +203,7 @@ const MaxReadsInFlight = 8
 type read struct {
 	dir, path string
 	manifest  bool
+	asked     bool
 	raw       []byte
 	err       error
 }
@@ -199,6 +217,14 @@ type read struct {
 // order the calls finish in, and the error is that of the first failing file in the order of dirs and LockFiles. After
 // a failure no further file is asked for, and inputsOf returns only when every call it started has returned.
 func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, err error) {
+	manifests, locks, _, err = inputsRead(owner, repo, ref, dirs, files)
+	return manifests, locks, err
+}
+
+// inputsRead is inputsOf, and also answers, on a failure too, the directories whose manifest and lock files were all
+// read, with their manifests and lock hashes: a merge keeps them by their tree, so its next try reads only the rest
+// (issue 1126).
+func inputsRead(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, whole []string, err error) {
 	at := func(dir, name string) string {
 		if dir == "" {
 			return name
@@ -227,6 +253,7 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma
 			<-slots
 			break
 		}
+		r.asked = true
 		wg.Add(1)
 		go func(r *read) {
 			defer func() { <-slots; wg.Done() }()
@@ -241,10 +268,16 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma
 	wg.Wait()
 	manifests = map[string]string{}
 	locks = map[string]map[string]string{}
+	broken := map[string]bool{} // a directory with a file not read (failed, or never asked after a failure)
 	for _, r := range reads {
 		switch {
 		case r.err != nil && (r.manifest || !errors.Is(r.err, ErrNotFound)):
-			return nil, nil, fmt.Errorf("%s: %w", r.path, r.err)
+			broken[r.dir] = true
+			if err == nil {
+				err = fmt.Errorf("%s: %w", r.path, r.err)
+			}
+		case r.err == nil && r.raw == nil && !r.asked:
+			broken[r.dir] = true
 		case r.err != nil: // a lock file the forge does not have
 		case r.manifest:
 			manifests[r.dir] = string(r.raw)
@@ -256,17 +289,29 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma
 			locks[r.dir][path.Base(r.path)] = hex.EncodeToString(sum[:])
 		}
 	}
+	for _, dir := range dirs {
+		if !broken[dir] {
+			whole = append(whole, dir)
+		}
+	}
+	if err != nil {
+		for dir := range broken {
+			delete(manifests, dir)
+			delete(locks, dir)
+		}
+		return manifests, locks, whole, err
+	}
 	if len(manifests) == 0 {
 		manifests = nil
 	}
 	if len(locks) == 0 {
 		locks = nil
 	}
-	return manifests, locks, nil
+	return manifests, locks, whole, nil
 }
 
 // merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once.
-func merged(subject string, p pull, trees Trees, files Files, list Listed) (Message, error) {
+func merged(subject string, p pull, trees Trees, files Files, list Listed, kept Kept) (Message, error) {
 	if !sha40.MatchString(p.MergeSHA) {
 		return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA)
 	}
@@ -283,9 +328,20 @@ func merged(subject string, p pull, trees Trees, files Files, list Listed) (Mess
 		names = append(names, dir)
 	}
 	slices.Sort(names)
-	manifests, locks, err := inputsOf(p.Owner, p.Repo, p.MergeSHA, names, files)
+	manifests, locks, toRead := keptInputs(names, dirs, kept)
+	read, readLocks, whole, err := inputsRead(p.Owner, p.Repo, p.MergeSHA, toRead, files)
 	if err != nil {
-		return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err)
+		keep(whole, dirs, read, readLocks, kept) // what was read whole is not asked again at the next try
+		return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s (%d of %d module directories read from the forge, the others kept by their tree): %v", ErrRetry, subject, repository, p.MergeSHA, len(toRead), len(names), err)
+	}
+	keep(toRead, dirs, read, readLocks, kept)
+	maps.Copy(manifests, read)
+	maps.Copy(locks, readLocks)
+	if len(manifests) == 0 {
+		manifests = nil
+	}
+	if len(locks) == 0 {
+		locks = nil
 	}
 	event, err := delivery.ForgeEvent(repository, delivery.ForgeMerged)
 	if err != nil {

modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go

@@ -110,6 +110,21 @@ func TestClosedBecomesAClosedPullMessage(t *testing.T) { // 1235/AC4
 	if !delivery.SameForge(f, want) {
 		t.Errorf("message = %+v, want %+v", f, want)
 	}
+	// Issue 1422, ADR 0484 §11: a head pushed to a pull request into another base is written closed, read nothing for, and said;
+	// one into main is no closed message, and an event naming no base is written as before.
+	msg, err = Convert(updatedSubject, updated(map[string]any{"base": "nox-mesh"}), nil, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if f, err = delivery.DecodeForge(msg.Data); err != nil || !delivery.SameForge(f, want) || !strings.Contains(msg.Said, "is into nox-mesh, not main") {
+		t.Errorf("an update into nox-mesh is %+v (%v), said %q; want it closed and said", f, err, msg.Said)
+	}
+	for _, base := range []any{"main", nil} {
+		msg, err = Convert(updatedSubject, updated(map[string]any{"base": base}), nil, anyFiles)
+		if f, _ = delivery.DecodeForge(msg.Data); err != nil || f.Action != delivery.PullUpdated || msg.Said != "" {
+			t.Errorf("an update with base %v is %+v (%v), said %q; want it updated", base, f, err, msg.Said)
+		}
+	}
 }
 
 func TestUnreadableEventsWriteNothing(t *testing.T) { // 1235/AC5

modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go

@@ -120,6 +120,8 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1
 	if _, err := ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, failing); !errors.Is(err, ErrRetry) {
 		t.Errorf("a listing the forge cannot give: %v", err)
 	}
+	// Issue 1126: a merge reads the forge only for the module directories whose tree changed.
+	t.Run("a merge reads only the directories whose tree changed", mergeReadsOnlyChangedDirs)
 }
 
 func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2
@@ -369,3 +371,73 @@ func TestOneFailingFileIsRetryAndEveryCallHasReturnedBeforeTheReadDoes(t *testin
 		t.Errorf("err = %v, want ErrRetry", err)
 	}
 }
+
+// memKept is a Kept in memory.
+type memKept map[string]Inputs
+
+func (k memKept) Get(tree string) (Inputs, bool) { in, ok := k[tree]; return in, ok }
+func (k memKept) Put(tree string, in Inputs)     { k[tree] = in }
+
+// Issue 1126 (mesh-catalog#535; issue 1458 met the kind first): a merge reads the forge only for the module directories
+// whose tree changed. Its message is the same as one read whole; a directory whose tree is kept costs no read.
+func mergeReadsOnlyChangedDirs(t *testing.T) {
+	manifest, sum := real(t, "module.json"), real(t, "go.sum")
+	forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest, "modules/a/go.sum": sum, "modules/b/module.json": []byte(`{"module":"b"}`)}}
+	kept := memKept{}
+	var calls []call
+	first, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1", "modules/b": "t2"}, nil), forge.get, nil, kept)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(forge.calls) != 6 {
+		t.Fatalf("the first merge read %d files, want 6 (a manifest and two lock files for each directory)", len(forge.calls))
+	}
+	// The next merge changes modules/b alone: only its three files are read, and the message equals one read whole.
+	forge.calls = nil
+	forge.have["modules/b/module.json"] = []byte(`{"module":"b","v":2}`)
+	trees := map[string]string{"modules/a": "t1", "modules/b": "t3"}
+	next, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get, nil, kept)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, c := range forge.calls {
+		if !strings.HasPrefix(c.path, "modules/b/") {
+			t.Errorf("%s was read although its directory's tree is kept", c.path)
+		}
+	}
+	if len(forge.calls) != 3 {
+		t.Errorf("the second merge read %d files, want 3", len(forge.calls))
+	}
+	whole, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(next.Data) != string(whole.Data) {
+		t.Errorf("the message from kept inputs differs from one read whole:\n%s\n%s", next.Data, whole.Data)
+	}
+	if string(first.Data) == string(next.Data) {
+		t.Error("the second merge's message did not change")
+	}
+	// A failed read keeps the directories it read whole and nothing of the one that failed, so the next try reads only the
+	// rest: a merge too large for one try's bound converges over its tries.
+	kept2 := memKept{}
+	failing := &forgeFiles{have: forge.have, errs: map[string]error{"modules/b/module.json": errors.New("context deadline exceeded")}}
+	if _, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, nil, kept2); !errors.Is(err, ErrRetry) {
+		t.Fatalf("err = %v, want ErrRetry", err)
+	}
+	if _, ok := kept2["t1"]; !ok || len(kept2) != 1 { // modules/a's files are asked first, before the failure is seen
+		t.Errorf("a failed read kept %v, want modules/a's tree t1 alone", kept2)
+	}
+	failing.calls = nil
+	failing.errs = nil
+	again, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, nil, kept2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(again.Data) != string(whole.Data) {
+		t.Errorf("the message of the try after a failure differs from one read whole")
+	}
+	if len(failing.calls) != 3 {
+		t.Errorf("the try after a failure read %d files, want only modules/b's 3", len(failing.calls))
+	}
+}

modules/mesh-delivery-forge-bridge/internal/bridge/kept.go

@@ -0,0 +1,57 @@
+package bridge
+
+// A merge of a repository of many modules read three files of the forge for every module, whatever the merge changed:
+// the catalogue's 140 modules are 420 reads, which no longer fit the 30 s bound of a handled event while the forge is
+// busy after a merge, so the merged message of mesh-catalog#535 was never written and the indexer never saw it (issue
+// 1126; issue 1458 met the same kind at 131 modules and raised the reads in flight). A directory's git tree hash names
+// every byte under it, so the manifest and the lock-file hashes read once at a tree hash are those of every later commit
+// whose directory has the same tree: a merge reads the forge only for the directories it changed.
+
+// Inputs is what the bridge read of one module directory at one tree hash: its manifest and its lock files' hex sha256
+// by name (none when it has none).
+type Inputs struct {
+	Manifest string            `json:"manifest"`
+	Locks    map[string]string `json:"locks,omitempty"`
+}
+
+// Kept holds Inputs by a directory's git tree hash (the bridge's state bucket inputs). Get answers false for a tree it
+// does not hold or cannot read; Put is best effort: a value it could not keep is read from the forge next time.
+type Kept interface {
+	Get(tree string) (Inputs, bool)
+	Put(tree string, in Inputs)
+}
+
+// keptInputs splits dirs into the inputs kept for their tree hash and the directories still to read from the forge.
+func keptInputs(dirs []string, trees map[string]string, kept Kept) (manifests map[string]string, locks map[string]map[string]string, read []string) {
+	manifests, locks = map[string]string{}, map[string]map[string]string{}
+	for _, dir := range dirs {
+		tree := trees[dir]
+		if kept == nil || tree == "" {
+			read = append(read, dir)
+			continue
+		}
+		in, ok := kept.Get(tree)
+		if !ok {
+			read = append(read, dir)
+			continue
+		}
+		manifests[dir] = in.Manifest
+		if len(in.Locks) > 0 {
+			locks[dir] = in.Locks
+		}
+	}
+	return manifests, locks, read
+}
+
+// keep puts what was read of each directory under its tree hash.
+func keep(dirs []string, trees map[string]string, manifests map[string]string, locks map[string]map[string]string, kept Kept) {
+	if kept == nil {
+		return
+	}
+	for _, dir := range dirs {
+		raw, ok := manifests[dir]
+		if tree := trees[dir]; ok && tree != "" {
+			kept.Put(tree, Inputs{Manifest: raw, Locks: locks[dir]})
+		}
+	}
+}

modules/mesh-delivery-forge-bridge/manifest_test.go

@@ -103,11 +103,12 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
 	if m.SecretsOwner != "mesh-delivery-forge-bridge" {
 		t.Errorf("secrets-owner = %q", m.SecretsOwner)
 	}
-	// Two buckets: the head last written on FORGE for each pull request, so a sync writes only what differs (ADR 0443
-	// §5), and the revision of each key of the gate's buckets the bridge performed, so a key is performed once per
-	// revision (ADR 0437 §8; issue 1341).
-	if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}}) { // 1279/AC6, 1341
-		t.Errorf("state = %+v, want the buckets heads and performed", m.State)
+	// Three buckets: the head last written on FORGE for each pull request, so a sync writes only what differs (ADR 0443
+	// §5), the revision of each key of the gate's buckets the bridge performed, so a key is performed once per
+	// revision (ADR 0437 §8; issue 1341), and what a merge read of each module directory by its tree, so a merge reads
+	// the forge only for the directories it changed (issue 1126).
+	if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}, {Name: "inputs", History: 1}}) { // 1279/AC6, 1341, 1126
+		t.Errorf("state = %+v, want the buckets heads, performed and inputs", m.State)
 	}
 	// No other state, no bus, no seat, no tool, no event (ADR 0437 §4; ADR 0521: no check.status).
 	for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} {

modules/mesh-delivery-forge-bridge/module.json

@@ -12,7 +12,8 @@
   ],
   "state": [
     { "name": "heads", "history": 1 },
-    { "name": "performed", "history": 1 }
+    { "name": "performed", "history": 1 },
+    { "name": "inputs", "history": 1 }
   ],
   "reads": [
     "mesh-delivery-gate.merges",

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go

@@ -29,7 +29,9 @@ func Tools(ask asker, now func() time.Time) []stdio.Tool {
 			Description: "Take a registered module in now: the indexer reads its manifest from the repository and commit the controller's " +
 				"build records register it from (at the head the gate's source names, when it names one), whatever it tried before, and " +
 				"writes it to its modules state as a merge would. For a module registered from outside the indexed repositories whose " +
-				"repository the forge could not read (a 404 before the forge's account could read it). Answers stored, from which " +
+				"repository the forge could not read (a 404 before the forge's account could read it). A module the indexer holds " +
+				"whose record stands behind the head the gate's source names (a merge whose merged message never reached the " +
+				"indexer) is read again at that head and written when it changed. Answers stored, from which " +
 				"repository, commit and path, or why not (held already, no registration, the forge's error, no manifest there). " +
 				"The indexer reads such a module again by itself on the next source revision and merged message; this asks now. " +
 				"indexer_module (mesh-delivery-view) reads the result.",

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go

@@ -404,15 +404,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		// Essential: the forge's merges are what the indexer indexes.
 		return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the indexer hears no merge without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, forgeStream, err)
 	}
-	// A merge applied while the taken state was missing, and the gate's heads once the forge stream is applied whole, get
-	// their taken key (issue 1280): when the state appears (Ready runs once, now or then), and after each applied message.
-	ix.Drained = func(ctx context.Context) (bool, error) {
-		info, err := cons.Info(ctx)
-		if err != nil {
-			return false, err
-		}
-		return info.NumPending == 0 && info.NumAckPending == 0, nil
-	}
+	// A merge applied while the taken state was missing gets its taken key (issue 1280): when the state appears (Ready runs
+	// once, now or then), and after each applied message. The gate's source heads never move a key (issue 1126).
 	taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) })
 	cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) }, logger.Printf)
 	if err != nil {

modules/mesh-delivery-indexer/internal/index/candidates.go

@@ -79,9 +79,6 @@ type Indexer struct {
 	ReadOpen func(ctx context.Context) ([]delivery.OpenPull, error)
 	// Say says one of the indexer's events (placement-changed); nil says none.
 	Say func(event string, body any) error
-	// Drained says whether the indexer has applied every message of the forge stream (the consumer has none pending and none
-	// unacknowledged); FillTaken writes the source's heads only then. Nil fills none from the source (taken.go).
-	Drained func(ctx context.Context) (bool, error)
 
 	mu         sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
 	source     sourceNow

modules/mesh-delivery-indexer/internal/index/modules_test.go

@@ -139,31 +139,29 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
 		{name: "a source head whose merge was not applied is not a taken key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: "",
 			then: func(ix *Indexer) error { // the forge stream still holds the merge: the consumer is not drained
 				delete(ix.Taken.(*memStore).vals, takenKey(repo))
-				ix.Drained = func(context.Context) (bool, error) { return false, nil }
 				if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
 					return err
 				}
 				return ix.FillTaken(context.Background())
 			}},
-		{name: "a source head of a drained forge stream fills a missing key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: next,
-			then: func(ix *Indexer) error {
-				delete(ix.Taken.(*memStore).vals, takenKey(repo))
-				ix.Drained = func(context.Context) (bool, error) { return true, nil }
-				if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
-					return err
-				}
-				return ix.FillTaken(context.Background())
-			}},
-		{name: "filling from an older source head leaves a later key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: next,
+		// Issue 1126 (mesh-catalog#535): the forge bridge never wrote the merged message of 9fd9569, so the forge stream was
+		// drained with the merge not applied; the source head moved the key all the same, and the view said the merge was
+		// on the machines. A key moves only by a merge applied (ADR 0556 clause 1), never by the source's word.
+		{name: "a source head of a drained forge stream whose merge was never heard leaves the key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge,
 			then: func(ix *Indexer) error {
 				ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == merge && in == next, nil }
-				m := mergedMsg(repo, three)
-				m.Merge = next
-				if err := ix.Apply(context.Background(), m); err != nil {
+				if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
 					return err
 				}
-				ix.Drained = func(context.Context) (bool, error) { return true, nil }
-				return ix.FillTaken(context.Background()) // the source still names merge
+				return ix.FillTaken(context.Background())
+			}},
+		{name: "a source head of a drained forge stream fills no missing key", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: "",
+			then: func(ix *Indexer) error {
+				delete(ix.Taken.(*memStore).vals, takenKey(repo))
+				if err := ix.ApplySource(context.Background(), 8, sourceOf(repo, next)); err != nil {
+					return err
+				}
+				return ix.FillTaken(context.Background())
 			}},
 		{name: "an earlier merge heard again after a later one leaves taken", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge,
 			then: func(ix *Indexer) error {

modules/mesh-delivery-indexer/internal/index/registered.go

@@ -45,6 +45,12 @@ type taken struct {
 }
 
 func (ix *Indexer) indexRegistered(ctx context.Context, r Registration, force bool) (taken, error) {
+	return ix.indexAt(ctx, r, force, false)
+}
+
+// indexAt is indexRegistered; reread also reads a module the modules state holds (a person's take-in of a held module
+// whose record stands behind the head the gate's source names, issue 1126).
+func (ix *Indexer) indexAt(ctx context.Context, r Registration, force, reread bool) (taken, error) {
 	if ix.Modules == nil || ix.Read == nil {
 		return taken{why: "the indexer reads no forge or writes no modules"}, nil
 	}
@@ -55,7 +61,7 @@ func (ix *Indexer) indexRegistered(ctx context.Context, r Registration, force bo
 	}
 	if _, _, found, err := ix.Modules.Get(ctx, key); err != nil {
 		return taken{}, fmt.Errorf("modules.%s: %w", key, err)
-	} else if found {
+	} else if found && !reread {
 		ix.markRead(r.Module)
 		return taken{held: true, why: "the modules state holds it already: a merge to its repository writes it from now on"}, nil
 	}
@@ -204,6 +210,9 @@ type TakenIn struct {
 // controller's records, or whose every registration could not be read or holds no manifest of it, is answered with why.
 func (ix *Indexer) TakeIn(ctx context.Context, module string) TakenIn {
 	out := TakenIn{Module: module}
+	if again, ok := ix.takeInHeld(ctx, module); ok {
+		return again
+	}
 	if ix.Builds == nil {
 		out.Answer = "the indexer reads no controller's build records, so it knows no registration"
 		return out
@@ -273,3 +282,68 @@ func (ix *Indexer) IndexRegisteredAll(ctx context.Context) error {
 	}
 	return first
 }
+
+// takeInHeld reads again, at the head the gate's source names for its repository, a module the modules state holds whose
+// record stands at another commit, and writes it as a merge would (only when it changed). It is the recovery for a merge
+// whose merged message never reached the indexer (issue 1126: mesh-catalog#535's message was given up by the forge bridge
+// after its reads timed out, and mesh-vault's record stayed at the merge before). It answers false where it does not
+// apply: no such module held, or no head named, or the record at that head already.
+func (ix *Indexer) takeInHeld(ctx context.Context, module string) (TakenIn, bool) {
+	out := TakenIn{Module: module}
+	key, err := delivery.ModulesKey(module)
+	if err != nil || ix.Modules == nil {
+		return out, false
+	}
+	value, _, found, err := ix.Modules.Get(ctx, key)
+	if err != nil || !found {
+		return out, false
+	}
+	cur, err := delivery.DecodeModule(value)
+	if err != nil || cur.Repository == "" || cur.Path == "" {
+		return out, false
+	}
+	head := ""
+	ix.mu.Lock()
+	for repo, h := range ix.source.heads {
+		if delivery.SameRepository(repo, cur.Repository) && h != "" {
+			head = h
+		}
+	}
+	ix.mu.Unlock()
+	if head == "" || head == cur.Commit {
+		return out, false
+	}
+	// Only forward: a record at a merge the source does not name yet (a merge applied before the source moved) is never
+	// written back to the source's older head. Where the order cannot be known, nothing is read.
+	behind, why := false, "the forge cannot be asked"
+	if ix.Contains != nil {
+		var err error
+		if behind, err = ix.Contains(ctx, cur.Repository, cur.Commit, head); err != nil {
+			why = err.Error()
+		} else if !behind {
+			why = fmt.Sprintf("its record at %.12s is not in %.12s", cur.Commit, head)
+		}
+	}
+	if !behind {
+		out.Held, out.Repository, out.Commit, out.Path = true, cur.Repository, cur.Commit, cur.Path
+		out.Answer = fmt.Sprintf("held; the source names another head, %.12s, and not one its record is behind (%s), so it is not read again", head, why)
+		return out, true
+	}
+	t, err := ix.indexAt(ctx, Registration{Module: module, Repository: cur.Repository, Path: cur.Path, Commit: head}, true, true)
+	out.Held, out.Repository, out.Commit, out.Path = true, t.repository, t.commit, pathOf(t.dir)
+	switch {
+	case err != nil:
+		out.Answer = "held, and not read again at the source's head: " + err.Error()
+	case !t.stored:
+		out.Answer = "held, and not read again at the source's head: " + t.why
+	default:
+		after, _, _, _ := ix.Modules.Get(ctx, key)
+		out.Stored = true
+		if string(after) == string(value) {
+			out.Answer = fmt.Sprintf("held, and read again from %s at %s: unchanged there, so its record stands", t.repository, t.commit)
+		} else {
+			out.Answer = fmt.Sprintf("held, read again from %s at %s and written to the modules state: it changed since %.12s", t.repository, t.commit, cur.Commit)
+		}
+	}
+	return out, true
+}

modules/mesh-delivery-indexer/internal/index/registered_test.go

@@ -111,6 +111,58 @@ func TestARegisteredModuleIsIndexedFromItsRepository(t *testing.T) { // the gran
 			}
 		})
 	}
+
+	// Issue 1126: mesh-catalog#535 changed mesh-vault, but its merged message never reached the indexer, so the record
+	// stood at the merge before. A take-in of the held module reads it again at the source's head, only when the record is
+	// behind that head: a head older than the record (a merge applied before the source moved), or one whose order the forge
+	// cannot say, is never written back.
+	for _, tc := range []struct {
+		name     string
+		contains func(context.Context, string, string, string) (bool, error)
+		moved    bool
+	}{
+		{name: "a held module behind the source's head is read again", moved: true,
+			contains: func(_ context.Context, _, commit, in string) (bool, error) { return commit == merge && in == h2, nil }},
+		{name: "a source head older than the record is not read", contains: func(context.Context, string, string, string) (bool, error) { return false, nil }},
+		{name: "a forge that cannot say the order reads nothing", contains: func(context.Context, string, string, string) (bool, error) {
+			return false, errors.New("timeout")
+		}},
+		{name: "no forge to ask reads nothing"},
+	} {
+		t.Run("take-in: "+tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			ix, mods := modulesIndexer(t)
+			ix.Manifests = newMem()
+			if err := ix.Apply(ctx, mergedMsg("novox/mesh-catalog", map[string]string{"modules/a": "treeA"})); err != nil {
+				t.Fatal(err)
+			}
+			ix.source = sourceNow{rev: 2, heads: map[string]string{"novox/mesh-catalog": h2}}
+			ix.Contains = tc.contains
+			var asked []string
+			ix.Read = func(_ context.Context, _, sha, dir string) (RepoRead, error) {
+				asked = append(asked, sha+":"+dir)
+				return RepoRead{Root: "r1", Trees: map[string]string{"modules/a": "treeA2"}, Manifests: map[string]string{"modules/a": `{"module":"a","version":"2"}`}}, nil
+			}
+			got := ix.TakeIn(ctx, "a")
+			m, ok := moduleOf(t, mods, "a")
+			if !tc.moved {
+				if got.Stored || !got.Held || len(asked) != 0 || !strings.Contains(got.Answer, "another head") || !ok || m.Commit != merge {
+					t.Errorf("take-in answered %+v, asked %v, modules.a at %.8s", got, asked, m.Commit)
+				}
+				return
+			}
+			if !got.Stored || !got.Held || got.Commit != h2 || got.Path != "modules/a" || !strings.Contains(got.Answer, "written") {
+				t.Errorf("take-in answered %+v", got)
+			}
+			if !ok || m.Commit != h2 || m.Inputs2 == nil || m.Inputs2.Tree != "treeA2" {
+				t.Errorf("modules.a = %+v (%v), want it at %.8s with tree treeA2 (asked %v)", m, ok, h2, asked)
+			}
+			n := len(asked) // at the head now: a take-in answers held and reads nothing
+			if again := ix.TakeIn(ctx, "a"); again.Stored || !again.Held || len(asked) != n {
+				t.Errorf("a take-in of a module at the source's head answered %+v and asked %v", again, asked)
+			}
+		})
+	}
 }
 
 func mustModule(t *testing.T, m delivery.Module) []byte {

modules/mesh-delivery-indexer/internal/index/taken.go

@@ -68,13 +68,13 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error {
 	return err
 }
 
-// FillTaken writes the taken keys that were left unwritten, never a merge the indexer has not applied whole (a key names
-// a merge only after it was applied: ADR 0556 clause 1, and the view reads it as the upper bound of what is reached). Two
-// sources, both through writeTaken as a live merge does (issue 1280):
-//   - the merges applied while the state was not on the bus (kept by writeTaken), whenever the state is there;
-//   - the gate's source heads, only when Drained says the indexer has applied every message of the forge stream that was
-//     there before the heads were read. Otherwise it says so, and the next kick of the fill (after a message is applied)
-//     tries again.
+// FillTaken writes the taken keys of the merges applied while the state was not on the bus (kept by writeTaken), through
+// writeTaken as a live merge does (issue 1280), whenever the state is there. It never writes a merge the indexer has not
+// applied whole (a key names a merge only after it was applied: ADR 0556 clause 1, and the view reads it as the upper bound
+// of what is reached). The gate's source heads are not such merges: a drained forge stream does not say a merge was applied,
+// since the forge bridge may not have written its merged message yet (issue 1126: mesh-catalog#535's merged message was
+// still being retried by the bridge when the indexer started, and the source's head moved the key, so the view said the
+// merge was on the machines while the indexer held mesh-vault at the merge before).
 //
 // It is idempotent: a key naming the merge or a later one is left. A state not on the bus yet writes nothing and says
 // nothing (the fill runs again when it appears). The first error is returned after every repository was tried.
@@ -87,7 +87,6 @@ func (ix *Indexer) FillTaken(ctx context.Context) error {
 	}
 	ix.mu.Lock()
 	pending := maps.Clone(ix.untaken)
-	heads := maps.Clone(ix.source.heads)
 	ix.mu.Unlock()
 	var first error
 	note := func(repository string, err error) {
@@ -106,25 +105,6 @@ func (ix *Indexer) FillTaken(ctx context.Context) error {
 		}
 		note(repository, err)
 	}
-	if ix.Drained == nil {
-		return first
-	}
-	if len(heads) == 0 {
-		ix.logf("the source names no repository head, so no taken key is filled from it")
-		return first
-	}
-	done, err := ix.Drained(ctx)
-	if err != nil {
-		ix.logf("whether the forge stream is applied cannot be read, so no taken key is filled from the source: %v", err)
-		return first
-	}
-	if !done {
-		ix.logf("the forge stream is not applied whole yet, so no taken key is filled from the source; the next applied message tries again")
-		return first
-	}
-	for _, repository := range slices.Sorted(maps.Keys(heads)) {
-		note(repository, ix.writeTaken(ctx, delivery.Forge{Repository: repository, Merge: heads[repository]}))
-	}
 	return first
 }