Merge pull request 'The indexer never moves a module backwards: a late merged message writes nothing; indexer_heard and indexer_module_history read what it heard and kept (issue 1126)' (#96) from fix/1126-index-order into main

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

@@ -18,6 +18,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -145,25 +146,50 @@ type delivered interface {
 	Ack() error
 	DoubleAck(ctx context.Context) error
 	Term() error
+	Metadata() (*jetstream.MsgMetadata, error)
 }
 
 // handle takes one message of FORGE into the candidates and acknowledges it. A message it can never read is terminated,
 // never redelivered, and the log names its subject and why. A write that failed stays unacknowledged: the server
-// delivers it again (the server's ack wait is the only clock).
+// delivers it again (the server's ack wait is the only clock). Each message handled is written to the heard state
-func handle(ctx context.Context, m delivered, ix *index.Indexer, logger *log.Logger) {
+// (index.Heard; nil, or a state not on the bus yet, writes none), so indexer_heard reads what the indexer heard of FORGE.
+func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Store, logger *log.Logger) {
+	h := index.Heard{Subject: m.Subject()}
+	if md, err := m.Metadata(); err == nil && md != nil {
+		h.Sequence, h.Delivered, h.Published = md.Sequence.Stream, md.NumDelivered, md.Timestamp.UTC()
+	}
+	defer func() {
+		if heard == nil {
+			return
+		}
+		h.Handled = time.Now().UTC()
+		if err := index.WriteHeard(ctx, heard, h); err != nil && !errors.Is(err, process.ErrNotYet) {
+			logger.Printf("%s: sequence %d is not written to the heard state: %v", h.Subject, h.Sequence, err)
+		}
+	}()
 	f, err := delivery.DecodeForge(m.Data())
 	if err != nil {
 		logger.Printf("%s: not a forge message, dropped: %v", m.Subject(), err)
+		h.Outcome, h.Words = index.HeardDropped, err.Error()
 		_ = m.Term()
 		return
 	}
-	switch err := ix.Apply(ctx, f); {
+	h.Kind, h.Repository, h.Number, h.Branch, h.Commit = f.Kind, f.Repository, f.Number, f.Branch, f.Head
+	if f.Merge != "" {
+		h.Commit = f.Merge
+	}
+	applied, err := ix.ApplyMessage(ctx, f)
+	h.Written, h.Late = applied.Written, applied.Late
+	switch {
 	case errors.Is(err, index.ErrUnreadable):
 		logger.Printf("%s: %v; dropped", m.Subject(), err)
+		h.Outcome, h.Words = index.HeardDropped, err.Error()
 		_ = m.Term()
 	case err != nil:
 		logger.Printf("%s: %s#%d left for redelivery: %v", m.Subject(), f.Repository, f.Number, err)
+		h.Outcome, h.Words = index.HeardLeft, err.Error()
 	default:
+		h.Outcome = index.HeardApplied
 		// Waits for the server, so the consumer's info no longer counts the message as unacknowledged when the fill that
 		// follows asks whether the stream is applied (issue 1280); a failed ack is redelivered after the ack wait, which
 		// kicks the fill again.
@@ -407,7 +433,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// 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)
+	// What the indexer heard of FORGE (issue 1126 note 76) is bound when its state appears, as every state a delivery adds
+	// (issue 1790): until then no message is written there, and the rest of the indexer runs.
+	heard := awaitedStore{process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)}
+	cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, heard, logger); ix.KickFill(ctx) }, logger.Printf)
 	if err != nil {
 		return err
 	}

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

@@ -3,9 +3,13 @@ package main
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"log"
 	"strings"
 	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 
@@ -29,6 +33,21 @@ func (f *fakeMsg) DoubleAck(context.Context) error {
 }
 func (f *fakeMsg) Term() error { f.termd++; return nil }
 
+// Metadata is the stream's word on the message: its sequence 41, delivered a second time (issue 1126 note 76).
+func (f *fakeMsg) Metadata() (*jetstream.MsgMetadata, error) {
+	return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: 41, Consumer: 7}, NumDelivered: 2, Timestamp: time.Date(2026, 10, 28, 13, 7, 42, 0, time.UTC)}, nil
+}
+
+// heardOf is the indexer's record of the last FORGE message it handled (index.HeardKey).
+func heardOf(t *testing.T, st *memStore) index.Heard {
+	t.Helper()
+	var h index.Heard
+	if err := json.Unmarshal(st.vals[index.HeardKey], &h); err != nil {
+		t.Fatalf("heard.%s = %q: %v", index.HeardKey, st.vals[index.HeardKey], err)
+	}
+	return h
+}
+
 type memStore struct{ vals map[string][]byte }
 
 func (m *memStore) Get(_ context.Context, k string) ([]byte, uint64, bool, error) {
@@ -57,21 +76,32 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part)
 	data, _ := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head,
 		Action: delivery.PullUpdated, Dirs: []string{"modules/a"}, DirsKnown: true})
 	m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull", data: data}
-	handle(context.Background(), m, ix, log.New(&bytes.Buffer{}, "", 0))
+	heard := &memStore{vals: map[string][]byte{}}
+	handle(context.Background(), m, ix, heard, log.New(&bytes.Buffer{}, "", 0))
 	c, err := delivery.DecodeCandidate(st.vals["feat/x"])
 	if err != nil || len(c.Pulls) != 1 || c.Pulls[0].Modules[0] != "a" || m.acked != 1 || m.doubled != 1 || m.termd != 0 {
 		t.Errorf("candidate %+v err %v acked %d terminated %d", c, err, m.acked, m.termd)
 	}
+	// Issue 1126 note 76: what the indexer heard of FORGE is readable (indexer_heard): subject, sequence, deliveries, time,
+	// repository, commit and what it did.
+	if h := heardOf(t, heard); h.Subject != m.subject || h.Sequence != 41 || h.Delivered != 2 || h.Published.Hour() != 13 || h.Kind != delivery.ForgePull ||
+		h.Repository != "novox/mesh-delivery" || h.Number != 12 || h.Commit != head || h.Outcome != index.HeardApplied || h.Handled.IsZero() {
+		t.Errorf("heard = %+v", h)
+	}
 }
 
 func TestAMessageThatIsNoForgeMessageIsTerminatedAndLogged(t *testing.T) { // 1238 (never redelivered)
 	var logs bytes.Buffer
 	st := &memStore{vals: map[string][]byte{}}
 	m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.x.y.pull", data: []byte(`{"v":1,"kind":"pull"}`)}
-	handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, log.New(&logs, "", 0))
+	heard := &memStore{vals: map[string][]byte{}}
+	handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, heard, log.New(&logs, "", 0))
 	if m.termd != 1 || m.acked != 0 || len(st.vals) != 0 || !strings.Contains(logs.String(), m.subject) {
 		t.Errorf("terminated %d acked %d keys %d log %q", m.termd, m.acked, len(st.vals), logs.String())
 	}
+	if h := heardOf(t, heard); h.Outcome != index.HeardDropped || h.Sequence != 41 || h.Words == "" {
+		t.Errorf("heard = %+v", h)
+	}
 }
 
 func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value)

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

@@ -106,47 +106,62 @@ func (ix *Indexer) logf(format string, args ...any) {
 // by compare-and-set on the revision read, and read again on a conflict. A lone pull request is a group of one. A
 // merged message also writes the fingerprint of each module of its repository (modules.go).
 func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error {
+	_, err := ix.ApplyMessage(ctx, f)
+	return err
+}
+
+// ApplyMessage is Apply, and answers which modules a merged message wrote and kept (for the indexer's heard state). Each
+// applied merged message is said in one line: its repository, commit and pull request, and the modules written.
+func (ix *Indexer) ApplyMessage(ctx context.Context, f delivery.Forge) (Applied, error) {
+	var applied Applied
 	if f.Kind == delivery.ForgeOpen {
-		return ix.applyOpen(ctx, f)
+		return applied, ix.applyOpen(ctx, f)
 	}
 	if f.Kind != delivery.ForgePull && f.Kind != delivery.ForgeMerged {
-		return nil
+		return applied, nil
 	}
 	if f.Kind == delivery.ForgeMerged && ix.Modules != nil {
-		if err := ix.applyMerged(ctx, f); err != nil {
+		w, err := ix.applyMerged(ctx, f)
-			return err
+		applied = Applied{Written: w.written, Late: w.late}
+		if err != nil {
+			return applied, err
 		}
-		if err := ix.writeTaken(ctx, f); err != nil {
+		ix.logf("%s", w.said(f.Number))
-			return err
+		if w.late == "" { // a late message leaves the taken key, which names a later merge
+			if err := ix.writeTakenKnown(ctx, f, w.asked); err != nil {
+				return applied, err
+			}
 		}
 		ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945)
 	}
-	if err := ix.applyPlacement(ctx, f); err != nil {
+	if applied.Late == "" { // a late merged message would write an older placement back
-		return err
+		if err := ix.applyPlacement(ctx, f); err != nil {
+			return applied, err
+		}
 	}
 	if err := ix.applyPlacementCandidates(ctx, f); err != nil {
-		return err
+		return applied, err
 	}
 	var digests map[string]string
 	var held map[string]delivery.Module
 	if f.Kind == delivery.ForgePull {
 		var err error
 		if digests, err = ix.keepManifests(ctx, f); err != nil {
-			return err
+			return applied, err
 		}
 		if held, err = ix.heldModules(ctx); err != nil {
-			return err
+			return applied, err
 		}
 	}
 	if trunkLike[f.Branch] {
 		ix.logf("%s#%d: its branch %q is no delivery group, so no candidate", f.Repository, f.Number, f.Branch)
-		return nil
+		return applied, nil
 	}
 	key, err := delivery.CandidatesKey(f.Branch)
 	if err != nil {
-		return fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err)
+		return applied, fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err)
 	}
-	return write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) {
+	return applied, write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) {
 		var cur *delivery.Candidate
 		if found {
 			c, err := delivery.DecodeCandidate(value)

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

@@ -0,0 +1,60 @@
+package index
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// What the indexer heard of the forge bridge's stream FORGE (issue 1126 note 76): no tool read FORGE's messages, so a merge
+// written back over a later one could not be traced to a late or repeated message. The indexer writes, for each message
+// it handles, one value under the key forge of its state heard: the message's subject, stream sequence, how often the
+// server delivered it, when it was published and handled, what it names and what the indexer did with it. The state keeps
+// the last 64 values (its history); mesh-delivery-view's indexer_heard reads them, newest first, with their times.
+const (
+	HeardState  = "heard"
+	HeardBucket = delivery.IndexerModule + "_" + HeardState
+	HeardKey    = "forge"
+	// What the indexer did with a message.
+	HeardApplied = "applied"
+	HeardLeft    = "left for redelivery"
+	HeardDropped = "dropped"
+)
+
+// Heard is one FORGE message as the indexer handled it.
+type Heard struct {
+	V          int       `json:"v"`
+	Subject    string    `json:"subject"`
+	Sequence   uint64    `json:"sequence"`            // the stream's sequence of the message
+	Delivered  uint64    `json:"delivered,omitempty"` // how often the server delivered it to the indexer, this time included
+	Published  time.Time `json:"published,omitzero"`
+	Handled    time.Time `json:"handled"`
+	Kind       string    `json:"kind,omitempty"`
+	Repository string    `json:"repository,omitempty"`
+	Number     int       `json:"number,omitempty"`
+	Branch     string    `json:"branch,omitempty"`
+	Commit     string    `json:"commit,omitempty"` // the merge commit of a merged message, else the pull request's head
+	Outcome    string    `json:"outcome"`
+	Words      string    `json:"words,omitempty"`   // why it was left or dropped
+	Written    []string  `json:"written,omitempty"` // the modules a merged message wrote
+	Late       string    `json:"late,omitempty"`    // the taken merge a merged message is older than: nothing of it was written (order.go)
+}
+
+// Applied is what Apply did with a merged message's modules: the keys written, and, for a message older than the
+// repository's taken merge, that merge (nothing was written).
+type Applied struct {
+	Written []string
+	Late    string
+}
+
+// WriteHeard writes h as the newest value of heard.forge.
+func WriteHeard(ctx context.Context, st Store, h Heard) error {
+	h.V = 1
+	data, err := json.Marshal(h)
+	if err != nil {
+		return err
+	}
+	return write(ctx, st, HeardKey, func([]byte, bool) ([]byte, bool, error) { return data, false, nil })
+}

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

@@ -24,21 +24,30 @@ import (
 // clause 7). Two directories naming one module write the one the bucket holds, or the first by name, and say the other.
 // A value the module already holds but for its commit is not written again: its commit stays the one its fingerprint first
 // appeared at (writeOverlay), so a merge writes only the modules it changes.
-func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error {
+//
+// A merge never moves a module backwards (order.go): a merged message older than the repository's taken merge writes and
+// deletes nothing, and is said; it is asked once, before anything is written.
+func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) (*mergeWrite, error) {
 	return ix.applyManifests(ctx, f, true)
 }
 
 // applyManifests is applyMerged; drop says whether the modules of the repository the message holds no manifest for are
 // deleted. A message that names only some of a repository's modules (a registered module read on its own, registered.go)
 // drops none.
-func (ix *Indexer) applyManifests(ctx context.Context, f delivery.Forge, drop bool) error {
+func (ix *Indexer) applyManifests(ctx context.Context, f delivery.Forge, drop bool) (*mergeWrite, error) {
+	w := newMergeWrite(f)
+	if drop {
+		if err := ix.lateMerge(ctx, w); err != nil || w.late != "" {
+			return w, err
+		}
+	}
 	digests, err := ix.keepManifests(ctx, f)
 	if err != nil {
-		return err
+		return w, err
 	}
 	held, err := ix.heldModules(ctx)
 	if err != nil {
-		return err
+		return w, err
 	}
 	objects, namedErr := ix.namedObjects(ctx, f)
 	overlay := map[string]delivery.Module{}
@@ -93,11 +102,11 @@ func (ix *Indexer) applyManifests(ctx context.Context, f delivery.Forge, drop bo
 	// (issue 1900). A message that carries no manifests (an older writer) or one that cannot be read whole says nothing.
 	if drop && len(f.Manifests) > 0 && every {
 		if err := ix.dropRemoved(ctx, f.Repository, held, func(key string, _ delivery.Module) bool { return named[key] }); err != nil {
-			return err
+			return w, err
 		}
 	}
 	// What this merge moved reaches the modules standing on it (a base's tree, build-agent's fingerprint), in the graph's order.
-	return ix.recomputeWith(ctx, overlay)
+	return w, ix.recomputeWith(ctx, overlay, w)
 }
 
 // baseFP answers the new-form fingerprint of a base module as modules.<base> holds it, or why it has none.

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

@@ -3,6 +3,8 @@ package index
 import (
 	"context"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"path"
 	"strings"
 	"testing"
@@ -97,10 +99,17 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
 		commit map[string]string
 		// taken is the merge the repository's taken key names after then (issue 1280, ADR 0556 clause 1).
 		taken string
+		// err says then fails (the message is left for redelivery); said is a line the indexer's log (or that error) must hold.
+		err  bool
+		said string
+		// contains, when set, is the forge's order, and asks how often then asks it (one question per merged message).
+		contains func(context.Context, string, string, string) (bool, error)
+		asks     int
 	}{
 		{name: "the same merge again", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge,
 			then: func(ix *Indexer) error { return ix.Apply(context.Background(), mergedMsg(repo, three)) }},
 		{name: "a merge touching one of three modules", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next,
+			said: "novox/mesh-delivery at cccccccccccc (#12): merged message applied; modules written: a",
 			then: func(ix *Indexer) error {
 				m := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"})
 				m.Merge = next
@@ -170,6 +179,78 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
 				m.Merge = next
 				return ix.Apply(context.Background(), m)
 			}},
+		// Issue 1126 note 76: mesh-catalog's merge 9cdac09a was written back over the later 69056948, rolling kernel and
+		// mesh-build-image back on the nodes. A merge never moves a module backwards: the held commit must be in the incoming one.
+		{name: "an earlier merge heard again after a later one moves no module back", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next,
+			said: "the merged message is older than taken.novox/mesh-delivery at cccccccccccc",
+			then: func(ix *Indexer) error {
+				ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) {
+					return commit == in || commit == merge && in == next, nil
+				}
+				later := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"})
+				later.Merge = next
+				if err := ix.Apply(context.Background(), later); err != nil {
+					return err
+				}
+				return ix.Apply(context.Background(), mergedMsg(repo, three))
+			}},
+		{name: "an earlier merge that lacks a later merge's module deletes nothing", writes: 1, commit: map[string]string{"a": merge, "d": next}, taken: next,
+			said: "the merged message is older than taken.novox/mesh-delivery at cccccccccccc",
+			then: func(ix *Indexer) error {
+				ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) {
+					return commit == in || commit == merge && in == next, nil
+				}
+				later := mergedMsg(repo, map[string]string{"modules/a": "treeA", "modules/b": "treeB", "modules/c": "treeC", "modules/d": "treeD"})
+				later.Merge = next
+				if err := ix.Apply(context.Background(), later); err != nil {
+					return err
+				}
+				return ix.Apply(context.Background(), mergedMsg(repo, three))
+			}},
+		// Review round 2: a 502 whose words carry "404" inside a sha is no not-found; the late message is left for redelivery.
+		{name: "a late message whose order the forge fails with a 502 writes nothing and is left for redelivery", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next, err: true,
+			said: "502 Bad Gateway",
+			then: func(ix *Indexer) error {
+				ix.Contains = func(context.Context, string, string, string) (bool, error) { return false, nil }
+				later := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"})
+				later.Merge = next
+				if err := ix.Apply(context.Background(), later); err != nil {
+					return err
+				}
+				ix.Contains = func(context.Context, string, string, string) (bool, error) {
+					return false, errors.New("Gitea API /repos/x/y/compare/aaa404aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa...bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb: 502 Bad Gateway")
+				}
+				return ix.Apply(context.Background(), mergedMsg(repo, three))
+			}},
+		{name: "a forge that cannot say the order writes nothing and leaves the message for redelivery", writes: 0, commit: map[string]string{"a": merge}, taken: merge, err: true,
+			said: "cannot be read",
+			then: func(ix *Indexer) error {
+				ix.Contains = func(context.Context, string, string, string) (bool, error) {
+					return false, errors.New("the forge did not answer")
+				}
+				later := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"})
+				later.Merge = next
+				return ix.Apply(context.Background(), later)
+			}},
+		// Review of 77d738d: only a merge older than the one taken in is kept back. A diverged or rewritten history, or a
+		// taken merge the forge no longer has, is not older, and one question per message is asked.
+		{name: "a merge of a diverged history is written, asking the forge once", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next,
+			contains: func(context.Context, string, string, string) (bool, error) { return false, nil }, asks: 1,
+			then: func(ix *Indexer) error {
+				later := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"})
+				later.Merge = next
+				return ix.Apply(context.Background(), later)
+			}},
+		{name: "a taken merge the forge no longer has stops no later merge", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: next,
+			said: "is not known to be older",
+			then: func(ix *Indexer) error {
+				ix.Contains = func(context.Context, string, string, string) (bool, error) {
+					return false, errors.New("gitea_contains: the forge answered: 404 Not Found")
+				}
+				later := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"})
+				later.Merge = next
+				return ix.Apply(context.Background(), later)
+			}},
 		{name: "a later source revision of another repository's merge", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge,
 			then: func(ix *Indexer) error {
 				return ix.ApplySource(context.Background(), 9, delivery.Source{Group: "g", Heads: map[string]string{repo: merge, "novox/mesh-catalog": otherMerge}})
@@ -187,8 +268,27 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
 				t.Fatal(err)
 			}
 			writes := mods.writes
-			if err := tc.then(ix); err != nil {
+			asks := 0
-				t.Fatal(err)
+			if tc.contains != nil {
+				ix.Contains = func(ctx context.Context, repo, commit, in string) (bool, error) {
+					asks++
+					return tc.contains(ctx, repo, commit, in)
+				}
+			}
+			var said strings.Builder
+			ix.Logf = func(format string, args ...any) { fmt.Fprintf(&said, format+"\n", args...) }
+			err := tc.then(ix)
+			if tc.contains != nil && asks != tc.asks {
+				t.Errorf("the forge was asked %d times, want %d", asks, tc.asks)
+			}
+			if (err != nil) != tc.err {
+				t.Fatalf("then = %v, want an error %v", err, tc.err)
+			}
+			if err != nil {
+				said.WriteString(err.Error()) // what handle logs as left for redelivery
+			}
+			if !strings.Contains(said.String(), tc.said) {
+				t.Errorf("the log %q does not say %q", said.String(), tc.said)
 			}
 			if got := mods.writes - writes; got != tc.writes {
 				t.Errorf("%d values written, want %d", got, tc.writes)

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

@@ -0,0 +1,118 @@
+package index
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
+)
+
+// A merge never moves a module backwards (issue 1126, note 76). The indexer wrote mesh-catalog's merge 9cdac09a back over
+// the later 69056948 for kernel and mesh-build-image: applyManifests and writeOverlay wrote whatever commit a merged
+// message carried, so a late or repeated message rolled the modules, and the nodes that want them, back.
+//
+// The question is whether the incoming commit is OLDER: the forge says the incoming commit is in the later one and the
+// two differ (Contains(incoming, in: later)). Only then is it kept back. A diverged or rewritten history answers no, and
+// is written forward; a commit the forge no longer has (a 404) is not known to be older, is written, and said. A forge
+// that cannot say for another reason (a timeout) writes nothing and is an error: the server delivers the message again.
+//
+//   - A merged message asks once, before it writes anything: is its merge older than the repository's taken merge (the
+//     last merge applied whole, taken.go)? Then the whole message is late: no module is written or deleted, no placement
+//     is written, the taken key stays, and the log says so.
+//   - A take-in of a held module asks once for that module: is the source's head older than the module's record
+//     (takeInHeld)?
+
+// mergeWrite is one merge's write of the modules: its repository and commit, whether it is late (older than the taken
+// merge), the taken merge it was compared with and the answer, and the module keys written, for the line the indexer says
+// per applied merged message.
+type mergeWrite struct {
+	repository, merge string
+	late              string // the taken merge this one is older than, or ""
+	asked             string // the taken merge the order was asked against, or "" when none was asked
+	written           []string
+}
+
+func newMergeWrite(f delivery.Forge) *mergeWrite {
+	return &mergeWrite{repository: f.Repository, merge: f.Merge}
+}
+
+// notFound says the forge answered that a commit it was asked about does not exist: gitea's status ": 404 ", matched as
+// readVia matches it (cmd/mesh-delivery-indexer named.go). A bare "404" is not enough: the error's words carry both shas,
+// and a 502 naming a sha with 404 in it would read as not older and write a late message backwards. The thorough form,
+// a typed not-found answered by gitea_contains, is noted on issue 1126.
+func notFound(err error) bool { return err != nil && strings.Contains(err.Error(), ": 404 ") }
+
+// older says commit is older than later: the forge says commit is in later and the two differ. A commit the forge does not
+// have is not known to be older (false, said); a forge that cannot say otherwise is an error. Without a forge to ask
+// (Contains nil) nothing is older.
+func (ix *Indexer) older(ctx context.Context, repository, commit, later string) (bool, error) {
+	if ix.Contains == nil || commit == "" || later == "" || commit == later {
+		return false, nil
+	}
+	in, err := ix.Contains(ctx, repository, commit, later)
+	if notFound(err) {
+		ix.logf("%s: the forge does not have %.12s or %.12s (a rewritten history?), so %.12s is not known to be older and is written: %v", repository, commit, later, commit, err)
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("%s: whether %.12s is older than %.12s cannot be read, so nothing of it is written (a merge never moves a module backwards, issue 1126): %w", repository, commit, later, err)
+	}
+	return in, nil
+}
+
+// lateMerge asks, before a merged message writes anything, whether its merge is older than the repository's taken merge
+// (one forge call). A taken state not on the bus yet, or no taken key, asks nothing.
+func (ix *Indexer) lateMerge(ctx context.Context, w *mergeWrite) error {
+	if ix.Taken == nil || w.merge == "" {
+		return nil
+	}
+	key, err := delivery.RepositoryKey(w.repository)
+	if err != nil {
+		return nil // writeTaken says it
+	}
+	value, _, found, err := ix.Taken.Get(ctx, key)
+	if errors.Is(err, process.ErrNotYet) || !found {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("taken.%s: %w", key, err)
+	}
+	cur, err := delivery.DecodeTaken(value)
+	if err != nil || cur.Merge == "" || cur.Merge == w.merge {
+		return nil
+	}
+	old, err := ix.older(ctx, w.repository, w.merge, cur.Merge)
+	if err != nil {
+		return err
+	}
+	w.asked = cur.Merge
+	if old {
+		w.late = cur.Merge
+		ix.logf("%s at %.12s: the merged message is older than taken.%s at %.12s, so no module is written or deleted and no placement written (a late or repeated message; a merge never moves a module backwards, issue 1126)",
+			w.repository, w.merge, key, cur.Merge)
+	}
+	return nil
+}
+
+// wrote records a module key the merge wrote.
+func (w *mergeWrite) wrote(key string) {
+	if w != nil && !slices.Contains(w.written, key) {
+		w.written = append(w.written, key)
+	}
+}
+
+// said is the line of an applied merged message: its repository, commit and pull request, and the modules written.
+func (w *mergeWrite) said(number int) string {
+	if w.late != "" {
+		return fmt.Sprintf("%s at %.12s (#%d): merged message applied as late: older than the taken %.12s, nothing written", w.repository, w.merge, number, w.late)
+	}
+	written := "none"
+	if len(w.written) > 0 {
+		written = strings.Join(slices.Sorted(slices.Values(w.written)), ", ")
+	}
+	return fmt.Sprintf("%s at %.12s (#%d): merged message applied; modules written: %s", w.repository, w.merge, number, written)
+}

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

@@ -129,7 +129,7 @@ func (ix *Indexer) indexAt(ctx context.Context, r Registration, force, reread bo
 	if l := read.Locks[dir]; len(l) > 0 {
 		f.Locks = map[string]map[string]string{dir: l}
 	}
-	if err := ix.applyManifests(ctx, f, false); err != nil {
+	if _, err := ix.applyManifests(ctx, f, false); err != nil {
 		return t, err
 	}
 	if _, _, found, err := ix.Modules.Get(ctx, key); err != nil {
@@ -313,20 +313,21 @@ func (ix *Indexer) takeInHeld(ctx context.Context, module string) (TakenIn, bool
 	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
+	// Never backwards (order.go): a source head older than the record (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.
+	// read and written back over it. A diverged or rewritten history, or a record whose commit the forge no longer has, is
-	behind, why := false, "the forge cannot be asked"
+	// not older and is read; where the forge cannot say otherwise, nothing is read. One forge call.
+	old, why := true, "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 {
+		if old, err = ix.older(ctx, cur.Repository, head, cur.Commit); err != nil {
-			why = err.Error()
+			old, why = true, err.Error()
-		} else if !behind {
+		} else if old {
-			why = fmt.Sprintf("its record at %.12s is not in %.12s", cur.Commit, head)
+			why = fmt.Sprintf("%.12s is older than its record at %.12s", head, cur.Commit)
 		}
 	}
-	if !behind {
+	if old {
 		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)
+		out.Answer = fmt.Sprintf("held; the source names another head, %.12s, and not one newer than its record (%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)

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

@@ -123,7 +123,13 @@ func TestARegisteredModuleIsIndexedFromItsRepository(t *testing.T) { // the gran
 	}{
 		{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 source head older than the record is not read", contains: func(_ context.Context, _, commit, in string) (bool, error) { return commit == h2 && in == merge, nil }},
+		// Review of 77d738d: a diverged or rewritten history, or a record whose commit the forge no longer has, is not older.
+		{name: "a source head of a diverged history is read again", moved: true,
+			contains: func(context.Context, string, string, string) (bool, error) { return false, nil }},
+		{name: "a record whose commit the forge no longer has is read again", moved: true, contains: func(context.Context, string, string, string) (bool, error) {
+			return false, errors.New("gitea_contains: 404 Not Found")
+		}},
 		{name: "a forge that cannot say the order reads nothing", contains: func(context.Context, string, string, string) (bool, error) {
 			return false, errors.New("timeout")
 		}},

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

@@ -20,6 +20,12 @@ import (
 
 // writeTaken writes the repository's taken key for the merge f carries, unless it names that merge or a later one.
 func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error {
+	return ix.writeTakenKnown(ctx, f, "")
+}
+
+// writeTakenKnown is writeTaken where the merge is known not to be older than notOlder (lateMerge asked it, issue 1126): a
+// key naming that merge is written over without asking the forge again.
+func (ix *Indexer) writeTakenKnown(ctx context.Context, f delivery.Forge, notOlder string) error {
 	if ix.Taken == nil || f.Merge == "" {
 		return nil
 	}
@@ -34,7 +40,7 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error {
 				if cur.Merge == f.Merge {
 					return nil, false, nil
 				}
-				if ix.Contains != nil {
+				if ix.Contains != nil && cur.Merge != notOlder {
 					later, err := ix.Contains(ctx, f.Repository, f.Merge, cur.Merge)
 					if err != nil {
 						ix.logf("%s: whether %s is in %s cannot be read, so taken.%s stays at %s: %v", f.Repository, f.Merge, cur.Merge, key, cur.Merge, err)

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

@@ -123,28 +123,29 @@ type known struct {
 // the order the bucket lists them in; modules that stand on each other have a cycle as their reason. A module is written
 // only when its fingerprint or its reason changes, and only if its manifest is still the one computed from. A module with
 // no kept inputs (a message that carried no manifest) is left as it is.
-func (ix *Indexer) Recompute(ctx context.Context) error { return ix.recomputeWith(ctx, nil) }
+func (ix *Indexer) Recompute(ctx context.Context) error { return ix.recomputeWith(ctx, nil, nil) }
 
 // recomputeWith is Recompute for a merge as well: the records the merge gives (overlay, by module key) stand in place
 // of what the bucket holds, are computed with the others, and are written once, with their final fingerprint or reason
-// (issue 1444: a merge writes each module once). A record of the overlay that has no kept inputs is written as it is.
+// (issue 1444: a merge writes each module once). A record of the overlay that has no kept inputs is written as it is. w is
-func (ix *Indexer) recomputeWith(ctx context.Context, overlay map[string]delivery.Module) error {
+// the merge's write (order.go), nil for none; it is told each module written.
+func (ix *Indexer) recomputeWith(ctx context.Context, overlay map[string]delivery.Module, w *mergeWrite) error {
 	if err := ix.fillPaths(ctx); err != nil {
 		return err
 	}
-	if err := ix.recomputeModules(ctx, overlay); err != nil {
+	if err := ix.recomputeModules(ctx, overlay, w); err != nil {
 		return err
 	}
 	// What the modules moved to reaches the candidates' fingerprints too (a base, the builder, a toolchain).
 	return ix.RecomputeCandidates(ctx)
 }
 
-func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]delivery.Module) error {
+func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]delivery.Module, w *mergeWrite) error {
 	if ix.Modules == nil {
 		return nil
 	}
 	if ix.Manifests == nil {
-		return ix.writeOverlay(ctx, overlay, nil)
+		return ix.writeOverlay(ctx, overlay, nil, w)
 	}
 	ix.recompute.Lock()
 	defer ix.recompute.Unlock()
@@ -294,6 +295,7 @@ func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]deli
 			continue
 		}
 		digest, inputs, paths := k.cur.Manifest, k.cur.Inputs2, k.cur.ArtifactPaths
+		wrote := false
 		err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
 			if !found {
 				return nil, false, nil
@@ -308,13 +310,17 @@ func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]deli
 			cur.FP2, cur.None2, cur.Toolchains = r.fp, r.none, r.tcs
 			cur.Inputs2, cur.ArtifactPaths = inputs, paths
 			data, err := delivery.EncodeModule(cur)
+			wrote = err == nil
 			return data, false, err
 		})
 		if err != nil {
 			return fmt.Errorf("modules.%s: %w", key, err)
 		}
+		if wrote {
+			w.wrote(key)
+		}
 	}
-	return ix.writeOverlay(ctx, overlay, finals)
+	return ix.writeOverlay(ctx, overlay, finals, w)
 }
 
 // writeOverlay writes the records of a merge, each once and only if it differs from what the bucket holds; finals has
@@ -325,24 +331,32 @@ func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]deli
 // appeared at: when the merge gives a module the value it holds but for the commit (and the source revision stamped on it),
 // the held value stands, its commit unchanged, so a catalogue merge touching one module writes one key, not one per module of
 // the repository, and the controller's take-in hears only that one. The repository's trunk head is the gate's source.main.
-func (ix *Indexer) writeOverlay(ctx context.Context, overlay, finals map[string]delivery.Module) error {
+//
+// A merged message older than the repository's taken merge never reaches here (order.go): it is asked before any write.
+func (ix *Indexer) writeOverlay(ctx context.Context, overlay, finals map[string]delivery.Module, w *mergeWrite) error {
 	for _, key := range slices.Sorted(maps.Keys(overlay)) {
 		next := overlay[key]
 		if f, ok := finals[key]; ok {
 			next = f
 		}
+		wrote := false
 		err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
+			wrote = false
 			if found {
 				if cur, err := delivery.DecodeModule(value); err == nil && sameButWhere(cur, next) {
 					return nil, false, nil
 				}
 			}
 			data, err := delivery.EncodeModule(next)
+			wrote = err == nil
 			return data, false, err
 		})
 		if err != nil {
 			return fmt.Errorf("modules.%s: %w", key, err)
 		}
+		if wrote {
+			w.wrote(key)
+		}
 	}
 	return nil
 }

modules/mesh-delivery-indexer/manifest_test.go

@@ -85,7 +85,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
 	if want := []string{"placement-changed"}; !reflect.DeepEqual(m.Emits, want) {
 		t.Errorf("emits = %v, want %v", m.Emits, want)
 	}
-	if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "placement-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) {
+	// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
+	if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "placement-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) {
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {

modules/mesh-delivery-indexer/module.json

@@ -17,6 +17,7 @@
   ],
   "state": [
     { "name": "candidates", "history": 10 },
+    { "name": "heard", "history": 64 },
     { "name": "manifests", "history": 1 },
     { "name": "modules", "history": 20 },
     { "name": "placement", "history": 10 },

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

@@ -31,6 +31,29 @@ func Tools(r view.Reader, u view.UsageReader) []stdio.Tool {
 				repository, _ := args["repository"].(string)
 				return r.Reach(repository)
 			}},
+		{Name: view.ModuleHistoryTool,
+			Description: "The history of one module's value in the indexer's modules state (modules.<module>), with times: each value it " +
+				"still keeps (up to 20), newest first, with its revision, when it was written, put or del, the repository and commit " +
+				"it was computed at, its source revision (src), its new-form fingerprint (fp2) or why it has none (none2), its manifest " +
+				"digest and path. Shows when a merge wrote a module, and when an older merge wrote it back. Read-only.",
+			Input: map[string]any{"type": "object", "required": []string{"module"}, "properties": map[string]any{
+				"module": map[string]any{"type": "string", "description": "the module's name"}}},
+			Run: func(args map[string]any) (any, error) {
+				name, _ := args["module"].(string)
+				return r.ModuleHistory(name)
+			}},
+		{Name: view.HeardTool,
+			Description: "The forge stream's (FORGE, the forge bridge's) recent messages as the indexer heard them: for each merged, pull, " +
+				"closed or open message it handled, newest first (its last 64), the subject, the stream sequence, how often the server " +
+				"delivered it (a redelivery shows above 1), when it was published and handled, its kind, repository, pull request, " +
+				"branch and commit (the merge commit, or the head), and what the indexer did: applied (with the modules written, or, " +
+				"for a late merged message, the later merge already taken in, when nothing was written), left for redelivery or dropped, with why. Read-only.",
+			Input: map[string]any{"type": "object", "properties": map[string]any{
+				"limit": map[string]any{"type": "integer", "description": "at most this many messages, newest first (default every one kept)"}}},
+			Run: func(args map[string]any) (any, error) {
+				limit, _ := args["limit"].(float64)
+				return r.Heard(int(limit))
+			}},
 		{Name: view.BuildUsageTool,
 			Description: "How the build nodes are used, and whether a node is eligible to build, per node (every node holding a build module or named by a capacity): each build seat it holds (node-build-go, -node, -image) with its capacity " +
 				"(mesh-config's resolved value, or the modules' default) and pause, and whether it is eligible to build and why not; the asks " +
@@ -199,7 +222,14 @@ func backlogOf(path string) func() ([]view.BacklogRow, error) {
 }
 
 func read(state stdio.KeptState) view.StateRead {
-	return view.StateRead{Keys: state.Keys, GetEntry: func(key string) (json.RawMessage, uint64, error) {
+	return view.StateRead{Keys: state.Keys, History: func(key string) ([]view.Revision, error) {
+		revs, err := state.History(key)
+		out := make([]view.Revision, 0, len(revs))
+		for _, rev := range revs {
+			out = append(out, view.Revision{Revision: rev.Revision, At: rev.At, Op: rev.Op, Value: rev.Value})
+		}
+		return out, err
+	}, GetEntry: func(key string) (json.RawMessage, uint64, error) {
 		e, err := state.Get(key)
 		if err != nil || e == nil {
 			return nil, 0, err
@@ -219,7 +249,8 @@ func main() {
 		WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")),
 		ReachState:               read(stdio.State("reach")),
 		PlacementState:           read(stdio.State("mesh-delivery-indexer.placement")),
-		PlacementCandidatesState: read(stdio.State("mesh-delivery-indexer.placement-candidates"))}
+		PlacementCandidatesState: read(stdio.State("mesh-delivery-indexer.placement-candidates")),
+		HeardState:               read(stdio.State("mesh-delivery-indexer.heard"))}
 	u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")),
 		Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())}
 	for _, lang := range delivery.AskLanguages {

modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go

@@ -84,6 +84,36 @@ func TestModuleToolAnswersTheStoredValue(t *testing.T) {
 	if _, err := run(t, r, view.ReachReadTool, map[string]any{"repository": "novox/gone"}); err == nil || !strings.Contains(err.Error(), "no reach of novox/gone") {
 		t.Errorf("a repository with no reach: %v", err)
 	}
+	// Issue 1126 note 76: indexer_module_history answers modules.<module>'s kept values newest first, each with its
+	// revision, time, commit, source revision and fp2; indexer_heard the FORGE messages the indexer handled, newest first.
+	at := time.Date(2026, 10, 28, 13, 7, 46, 0, time.UTC)
+	r.State.History = func(key string) ([]view.Revision, error) {
+		if key != "a" {
+			return nil, nil
+		}
+		return []view.Revision{{Revision: 538, At: at.Add(-time.Minute), Op: "put", Value: json.RawMessage(withNone)},
+			{Revision: 539, At: at, Op: "put", Value: json.RawMessage(withFP2)}, {Revision: 540, At: at.Add(time.Minute), Op: "del"}}, nil
+	}
+	out, err := run(t, r, view.ModuleHistoryTool, map[string]any{"module": "a"})
+	h, _ := out.(view.ModuleHistory)
+	if err != nil || len(h.Values) != 3 || h.Values[0].Revision != 540 || h.Values[0].Op != "del" || h.Values[1].Revision != 539 || !h.Values[1].At.Equal(at) ||
+		h.Values[1].Commit != "c1" || h.Values[1].FP2 != "b2" || h.Values[2].Commit != "c2" || h.Values[2].None2 == "" {
+		t.Errorf("indexer_module_history a = %+v, %v", out, err)
+	}
+	if _, err := run(t, r, view.ModuleHistoryTool, map[string]any{"module": "gone"}); err == nil || !strings.Contains(err.Error(), "no history of gone") {
+		t.Errorf("a module with no history: %v", err)
+	}
+	const heard1, heard2 = `{"v":1,"subject":"s","sequence":41,"repository":"novox/mesh-catalog","commit":"9cdac09a"}`, `{"v":1,"subject":"s","sequence":42,"repository":"novox/mesh-catalog","commit":"69056948"}`
+	r.HeardState = view.StateRead{History: func(key string) ([]view.Revision, error) {
+		return []view.Revision{{Revision: 1, At: at, Op: "put", Value: json.RawMessage(heard1)}, {Revision: 2, At: at, Op: "put", Value: json.RawMessage(heard2)}}, nil
+	}}
+	out, err = run(t, r, view.HeardTool, map[string]any{})
+	if rows, _ := out.(view.HeardAnswer); err != nil || len(rows.Heard) != 2 || rows.Heard[0].Revision != 2 || string(rows.Heard[0].Heard) != heard2 || !rows.Heard[1].At.Equal(at) {
+		t.Errorf("indexer_heard = %+v, %v", out, err)
+	}
+	if out, err = run(t, r, view.HeardTool, map[string]any{"limit": float64(1)}); err != nil || len(out.(view.HeardAnswer).Heard) != 1 {
+		t.Errorf("indexer_heard limit 1 = %+v, %v", out, err)
+	}
 }
 
 // 1562/AC1: indexer_modules answers every key as stored, and with none2 only those that say why they have no FP2.

modules/mesh-delivery-view/internal/view/history.go

@@ -0,0 +1,141 @@
+package view
+
+// Two read tools for tracing a module's value back to the message that wrote it (issue 1126 note 76: mesh-catalog's merge
+// 9cdac09a was written back over the later 69056948, and no tool read a modules key's history with times, nor what the
+// indexer heard of the forge stream FORGE).
+
+import (
+	"encoding/json"
+	"fmt"
+	"slices"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const (
+	// ModuleHistoryTool reads modules.<module>'s kept values with their times.
+	ModuleHistoryTool = "indexer_module_history"
+	// HeardTool reads the FORGE messages the indexer handled (its state heard).
+	HeardTool = "indexer_heard"
+	// heardKey is the indexer's one key of its heard state (mesh-delivery-indexer internal/index heard.go HeardKey).
+	heardKey = "forge"
+)
+
+// Revision is one value a state still keeps for a key (mesh-sdk stdio StateRevision): its revision, when it was written,
+// put or del, and the value (none for a delete).
+type Revision struct {
+	Revision uint64          `json:"revision"`
+	At       time.Time       `json:"at"`
+	Op       string          `json:"op"`
+	Value    json.RawMessage `json:"value,omitempty"`
+}
+
+// ModuleRevision is one kept value of modules.<module>, read: where it was computed and what it computed.
+type ModuleRevision struct {
+	Revision   uint64    `json:"revision"`
+	At         time.Time `json:"at"`
+	Op         string    `json:"op"`
+	Repository string    `json:"repository,omitempty"`
+	Commit     string    `json:"commit,omitempty"`
+	Src        uint64    `json:"src,omitempty"`
+	FP2        string    `json:"fp2,omitempty"`
+	None2      string    `json:"none2,omitempty"`
+	Manifest   string    `json:"manifest,omitempty"`
+	Path       string    `json:"path,omitempty"`
+	Unreadable string    `json:"unreadable,omitempty"`
+}
+
+// ModuleHistory is a module's kept values, newest first.
+type ModuleHistory struct {
+	Module string           `json:"module"`
+	Values []ModuleRevision `json:"values"`
+	Note   string           `json:"note"`
+}
+
+// ModuleHistory reads every value the modules state keeps for module, newest first, each with its revision, time, commit,
+// source revision and fingerprint.
+func (r Reader) ModuleHistory(module string) (ModuleHistory, error) {
+	module = strings.TrimSpace(module)
+	out := ModuleHistory{Module: module, Note: "newest first; the modules state keeps 20 values a key (its history), so an older one is gone"}
+	key, err := delivery.ModulesKey(module)
+	if err != nil {
+		return out, err
+	}
+	if r.State.History == nil {
+		return out, fmt.Errorf("the modules state's history cannot be read here")
+	}
+	revs, err := r.State.History(key)
+	if err != nil {
+		return out, err
+	}
+	if len(revs) == 0 {
+		return out, fmt.Errorf("no history of %s in the modules state", module)
+	}
+	for _, rev := range newestFirst(revs) {
+		row := ModuleRevision{Revision: rev.Revision, At: rev.At, Op: rev.Op}
+		if len(rev.Value) > 0 && string(rev.Value) != "null" {
+			// Read field by field, whatever the value's version, so an older or newer writer's value still shows.
+			var m delivery.Module
+			if err := json.Unmarshal(rev.Value, &m); err != nil {
+				row.Unreadable = err.Error()
+			} else {
+				row.Repository, row.Commit, row.Src, row.FP2, row.None2, row.Manifest, row.Path = m.Repository, m.Commit, m.Src, m.FP2, m.None2, m.Manifest, m.Path
+			}
+		}
+		out.Values = append(out.Values, row)
+	}
+	return out, nil
+}
+
+// HeardRow is one FORGE message as the indexer handled it, with the revision and time of its value.
+type HeardRow struct {
+	Revision uint64          `json:"revision"`
+	At       time.Time       `json:"at"`
+	Heard    json.RawMessage `json:"heard"`
+}
+
+// HeardAnswer is the FORGE messages the indexer handled, newest first.
+type HeardAnswer struct {
+	Heard []HeardRow `json:"heard"`
+	Note  string     `json:"note"`
+}
+
+// Heard reads the indexer's heard state: the last FORGE messages it handled (at most limit, 0 for every one kept), newest
+// first.
+func (r Reader) Heard(limit int) (HeardAnswer, error) {
+	out := HeardAnswer{Heard: []HeardRow{}, Note: "newest first; the indexer keeps its last 64 FORGE messages. A message the stream holds and the indexer never handled is not here; " +
+		"delivered above 1 is a message the server delivered again"}
+	if r.HeardState.History == nil {
+		return out, fmt.Errorf("the indexer's heard state cannot be read here")
+	}
+	revs, err := r.HeardState.History(heardKey)
+	if err != nil {
+		return out, err
+	}
+	for _, rev := range newestFirst(revs) {
+		if limit > 0 && len(out.Heard) == limit {
+			break
+		}
+		if len(rev.Value) == 0 || string(rev.Value) == "null" {
+			continue
+		}
+		out.Heard = append(out.Heard, HeardRow{Revision: rev.Revision, At: rev.At, Heard: rev.Value})
+	}
+	return out, nil
+}
+
+func newestFirst(revs []Revision) []Revision {
+	revs = slices.Clone(revs)
+	slices.SortFunc(revs, func(a, b Revision) int {
+		switch {
+		case a.Revision > b.Revision:
+			return -1
+		case a.Revision < b.Revision:
+			return 1
+		}
+		return 0
+	})
+	return revs
+}

modules/mesh-delivery-view/internal/view/read.go

@@ -31,6 +31,8 @@ type StateRead struct {
 	Get  func(key string) (json.RawMessage, error)
 	// GetEntry, when set, also gives the revision of the value; builder_builds orders by it.
 	GetEntry func(key string) (json.RawMessage, uint64, error)
+	// History, when set, gives every value the state keeps for a key with its time (indexer_module_history, indexer_heard).
+	History func(key string) ([]Revision, error)
 }
 
 // Reader answers the read tools from a modules state.
@@ -46,6 +48,8 @@ type Reader struct {
 	PlacementState   StateRead // the indexer's placement
 	// PlacementCandidatesState is the indexer's placement candidates (issue 1126); unset reads none.
 	PlacementCandidatesState StateRead
+	// HeardState is the FORGE messages the indexer handled (issue 1126 note 76); unset reads none.
+	HeardState StateRead
 }
 
 // ReachReadTool reads the view's reach: per repository, how far each module reached and the reached head (ADR 0556).

modules/mesh-delivery-view/manifest_test.go

@@ -87,7 +87,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 	if m.Module != "mesh-delivery-view" {
 		t.Errorf("module = %q", m.Module)
 	}
-	want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-indexer.placement-candidates", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // indexer_placement reads the placement candidates and every <node>.node (ADR 0559 §7); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
+	want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.heard", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-indexer.placement-candidates", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // indexer_placement reads the placement candidates and every <node>.node (ADR 0559 §7); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
 	got := slices.Clone(m.Reads)
 	slices.Sort(got)
 	if !reflect.DeepEqual(got, want) {
@@ -119,7 +119,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 	}
 	// The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer
 	// itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads.
-	if want := []string{"indexer_module", "indexer_modules", "indexer_placement", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage", "view_reach"}; !reflect.DeepEqual(m.Tools, want) {
+	if want := []string{"indexer_module", "indexer_modules", "indexer_placement", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage", "view_reach", "indexer_module_history", "indexer_heard"}; !reflect.DeepEqual(m.Tools, want) {
 		t.Errorf("tools = %v, want %v", m.Tools, want)
 	}
 	for _, a := range m.Build.Artifacts {

modules/mesh-delivery-view/module.json

@@ -16,7 +16,9 @@
     "assigner_differences",
     "node_haves",
     "build_usage",
-    "view_reach"
+    "view_reach",
+    "indexer_module_history",
+    "indexer_heard"
   ],
   "emits": [
     "delivered",
@@ -47,6 +49,7 @@
     "mesh-delivery-gate.asks",
     "mesh-delivery-gate.source",
     "mesh-delivery-indexer.candidates",
+    "mesh-delivery-indexer.heard",
     "mesh-delivery-indexer.modules",
     "mesh-delivery-indexer.placement",
     "mesh-delivery-indexer.placement-candidates",