novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

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

Merged
mesh-tools merged 3 commits from fix/1126-index-order into main 2026-10-28 14:44:34 +00:00
Conversation 1 Commits 3 Files Changed 19
+653 -54
19 changed files with 653 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+33 -4
View File
@@ -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
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go
+32 -2
View File
@@ -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
+29 -14
View File
@@ -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)
modules/mesh-delivery-indexer/internal/index/heard.go
+60
View File
@@ -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
+15 -6
View File
@@ -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
+102 -2
View File
@@ -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)
modules/mesh-delivery-indexer/internal/index/order.go
+118
View File
@@ -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
+11 -10
View File
@@ -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)
modules/mesh-delivery-indexer/internal/index/registered_test.go
+7 -1
View File
@@ -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
+7 -1
View File
@@ -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
+22 -8
View File
@@ -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
}
modules/mesh-delivery-indexer/manifest_test.go
+2 -1
View File
@@ -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
+1
View File
@@ -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
+33 -2
View File
@@ -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
+30
View File
@@ -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
+141
View File
@@ -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
+4
View File
@@ -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
+2 -2
View File
@@ -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
+4 -1
View File
@@ -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",