diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
index 01060f1..9de33a4 100644
--- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+++ b/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).
-func handle(ctx context.Context, m delivered, ix *index.Indexer, logger *log.Logger) {
+// delivers it again (the server's ack wait is the only clock). Each message handled is written to the heard state
+// (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
}
diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go
index e49c636..eb70b62 100644
--- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go
+++ b/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)
diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go
index 2a2b6aa..7f3bbc6 100644
--- a/modules/mesh-delivery-indexer/internal/index/candidates.go
+++ b/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 {
- return err
+ w, err := ix.applyMerged(ctx, f)
+ applied = Applied{Written: w.written, Late: w.late}
+ if err != nil {
+ return applied, err
}
- if err := ix.writeTaken(ctx, f); err != nil {
- return err
+ ix.logf("%s", w.said(f.Number))
+ 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 {
- return err
+ if applied.Late == "" { // a late merged message would write an older placement back
+ 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)
diff --git a/modules/mesh-delivery-indexer/internal/index/heard.go b/modules/mesh-delivery-indexer/internal/index/heard.go
new file mode 100644
index 0000000..0c66225
--- /dev/null
+++ b/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 })
+}
diff --git a/modules/mesh-delivery-indexer/internal/index/modules.go b/modules/mesh-delivery-indexer/internal/index/modules.go
index 5bd2670..199f125 100644
--- a/modules/mesh-delivery-indexer/internal/index/modules.go
+++ b/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. holds it, or why it has none.
diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go
index 1936672..e65bc31 100644
--- a/modules/mesh-delivery-indexer/internal/index/modules_test.go
+++ b/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 {
- t.Fatal(err)
+ asks := 0
+ 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)
diff --git a/modules/mesh-delivery-indexer/internal/index/order.go b/modules/mesh-delivery-indexer/internal/index/order.go
new file mode 100644
index 0000000..c09cc5e
--- /dev/null
+++ b/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)
+}
diff --git a/modules/mesh-delivery-indexer/internal/index/registered.go b/modules/mesh-delivery-indexer/internal/index/registered.go
index 7fe4c28..5666beb 100644
--- a/modules/mesh-delivery-indexer/internal/index/registered.go
+++ b/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
- // 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"
+ // Never backwards (order.go): a source head older than the record (a merge applied before the source moved) is never
+ // read and written back over it. A diverged or rewritten history, or a record whose commit the forge no longer has, is
+ // 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 {
- why = err.Error()
- } else if !behind {
- why = fmt.Sprintf("its record at %.12s is not in %.12s", cur.Commit, head)
+ if old, err = ix.older(ctx, cur.Repository, head, cur.Commit); err != nil {
+ old, why = true, err.Error()
+ } else if old {
+ 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)
diff --git a/modules/mesh-delivery-indexer/internal/index/registered_test.go b/modules/mesh-delivery-indexer/internal/index/registered_test.go
index 879c7e5..78101a4 100644
--- a/modules/mesh-delivery-indexer/internal/index/registered_test.go
+++ b/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")
}},
diff --git a/modules/mesh-delivery-indexer/internal/index/taken.go b/modules/mesh-delivery-indexer/internal/index/taken.go
index a6d97e7..89c0293 100644
--- a/modules/mesh-delivery-indexer/internal/index/taken.go
+++ b/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)
diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go
index c1a64fa..21e022f 100644
--- a/modules/mesh-delivery-indexer/internal/index/toolchains.go
+++ b/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.
-func (ix *Indexer) recomputeWith(ctx context.Context, overlay map[string]delivery.Module) error {
+// (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
+// 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
}
diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go
index 7cf1b86..abd5e9c 100644
--- a/modules/mesh-delivery-indexer/manifest_test.go
+++ b/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) {
diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json
index 2a2437a..6a7fd5e 100644
--- a/modules/mesh-delivery-indexer/module.json
+++ b/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 },
diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go
index 2561c94..9231f3e 100644
--- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go
+++ b/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.), 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 {
diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go
index 9fb8353..e2ec081 100644
--- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go
+++ b/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.'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.
diff --git a/modules/mesh-delivery-view/internal/view/history.go b/modules/mesh-delivery-view/internal/view/history.go
new file mode 100644
index 0000000..8885846
--- /dev/null
+++ b/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.'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., 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
+}
diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go
index b5f66a9..928a37e 100644
--- a/modules/mesh-delivery-view/internal/view/read.go
+++ b/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).
diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go
index 1ad60db..eab7c9c 100644
--- a/modules/mesh-delivery-view/manifest_test.go
+++ b/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 (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 (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 {
diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json
index cbbeaa4..e8ea5a0 100644
--- a/modules/mesh-delivery-view/module.json
+++ b/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",