Indexer ends a pending merge its repository's taken merge covers, without reading its message (issue 1966)

Quick fix: none smaller than this (the bridge fix alone leaves a covered merge needing the bridge).
Thorough fix (chosen): loadNewest asks takenAlready (already used for group members) before loading the newest merge of
a repository; at or behind taken it is marked Covered, said once in the log, and never read. A merge whose ancestry
cannot be asked is read as before. Test: a row pair behind/ahead of taken with a bridge that cannot answer.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i

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

@@ -254,6 +254,14 @@ func (ix *Indexer) loadNewest(ctx context.Context, entries []PendingEntry) (out,
 			continue
 		}
 		newest[repo] = true
+		// A merge at or behind its repository's taken merge writes nothing: it is ended as covered without reading its
+		// message, which a bridge that cannot give it would otherwise hold the whole revision for (issue 1966). A merge whose
+		// ancestry cannot be asked is read as before.
+		if taken, err := ix.takenAlready(ctx, forgemsg.GroupMerge{Repository: out[i].Forge.Repository, Number: out[i].Forge.Number, Merge: out[i].Forge.Merge}); err == nil && taken {
+			ix.logf("take-in: %s (%s#%d) is covered by the taken merge of %s, its message is not read", out[i].Key, out[i].Forge.Repository, out[i].Forge.Number, out[i].Forge.Repository)
+			out[i].Covered = true
+			continue
+		}
 		full, err := ix.load(ctx, out[i])
 		if err != nil {
 			lost = append(lost, out[i])

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

@@ -710,6 +710,9 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 		if len(b.said) != 1 || !slices.Equal(b.said[0].Lost, []string{takein.PendingKey(3)}) || len(waits) != 1 || !slices.Equal(waits[0].Lost, []string{takein.PendingKey(3)}) || b.commit(t, "avahi") != sha('b') {
 			t.Fatalf("lost: said %+v, waits %+v, avahi at %s: want m.3 lost on the event and in indexer_pending, nothing taken in", b.said, waits, b.commit(t, "avahi"))
 		}
+		if _, err := takein.EncodeTakeIn(b.said[0]); err != nil {
+			t.Errorf("the lost hold's event does not pass its check: %v", err)
+		}
 		if v, _, _, _ := b.pending.Get(ctx, takein.PendingKey(3)); !strings.Contains(string(v), "the bridge does not answer") {
 			t.Errorf("m.3 = %s: want the bridge's words", v)
 		}
@@ -753,6 +756,42 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 			t.Errorf("an old value with the message whole: err %v, avahi at %s: want it taken in", err, b.commit(t, "avahi"))
 		}
 	})
+	t.Run("1966 a merge at or behind its repository's taken merge is ended without its message being read, one ahead of it is lost as before", func(t *testing.T) {
+		for _, c := range []struct {
+			name   string
+			behind bool
+		}{{"behind taken", true}, {"ahead of taken", false}} {
+			b := newTakeInBed(t)
+			f := b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")
+			f.Words, f.Branch = strings.Repeat("x", 300*1024), "feat/g"
+			b.arrive(t, 2, f) // held by reference, for an indexer build that is not this one
+			b.ix.Taken = newMem()
+			b.ix.Contains = func(context.Context, string, string, string) (bool, error) { return c.behind, nil }
+			data, _ := takein.EncodeTaken(takein.Taken{Repository: trunk.RepositoryName(catalogueRepo), Merge: sha('d')})
+			key, _ := trunk.RepositoryKey(catalogueRepo)
+			b.ix.Taken.(*memStore).put(key, data)
+			named := b.fp(t, Module)
+			asked := 0
+			next := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
+				Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Taken: b.ix.Taken, Contains: b.ix.Contains,
+				Message: func(context.Context, string, string, int) ([]byte, error) {
+					asked++
+					return nil, errors.New("the bridge does not answer")
+				},
+				SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum, OwnBuild: func(context.Context) (string, error) { return named, nil }}
+			next.toolchains = b.ix.toolchains
+			next.FixOwnBuild(ctx)
+			if _, _, err := next.TakeInAtStart(ctx); err != nil {
+				t.Fatal(err)
+			}
+			if c.behind && (asked != 0 || len(b.keys()) != 0 || !strings.Contains(b.logs.String(), "covered by the taken merge")) {
+				t.Errorf("%s: asked %d, pending %v: want no read and the entry ended", c.name, asked, b.keys())
+			}
+			if !c.behind && (asked == 0 || len(b.keys()) != 1) {
+				t.Errorf("%s: asked %d, pending %v: want the message asked and the entry held", c.name, asked, b.keys())
+			}
+		}
+	})
 	t.Run("the dry run names every field of the indexer", func(t *testing.T) {
 		// dry (pending.go) builds an indexer field by field: a field added to Indexer is named there, or this fails.
 		named := map[string]bool{}