Merge pull request 'The indexer learns its own build right at its start, and says when a start takes nothing in (issue 1921)' (#156) from fix/1921-own-build-at-start into main

This commit was merged in pull request #156.
This commit is contained in:
2026-11-05 19:22:04 +00:00
11 changed files with 147 additions and 5 deletions
@@ -102,7 +102,7 @@ func filesVia(call toolCaller) func(ctx context.Context, repository, sha string,
}
for _, d := range dirs {
if _, ok := a.Files[d]; !ok {
return nil, fmt.Errorf("the answer of %s lists no files of %q", commitTreesTool, d)
return nil, fmt.Errorf("the answer of %s lists no files of %q: %w", commitTreesTool, d, index.ErrNotInTree)
}
}
return a.Files, nil
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"mesh-delivery-indexer/internal/index"
"slices"
"strings"
"testing"
@@ -50,6 +51,15 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3
if _, has := gotArgs.(map[string]any)["paths"]; has {
t.Error("the trunk asked for paths")
}
// A directory the answer lists no files of is a failure of the tree (index.ErrNotInTree); a call that fails is not.
noDir := func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"files":{}}`), nil }
if _, err := filesVia(noDir)(context.Background(), "novox/mesh-sdk", strings.Repeat("a", 40), []string{"go"}); !errors.Is(err, index.ErrNotInTree) {
t.Errorf("a missing directory: %v, want ErrNotInTree", err)
}
down := func(string, any) (json.RawMessage, error) { return nil, errors.New("down") }
if _, err := filesVia(down)(context.Background(), "novox/mesh-sdk", strings.Repeat("a", 40), []string{"go"}); err == nil || errors.Is(err, index.ErrNotInTree) {
t.Errorf("a failed call: %v, want an error that is not ErrNotInTree", err)
}
// The build the indexer runs is the engine's word (Runs.FP of its composition verb), asked of its own machine.
var asked string
own := func(key string, body any) (json.RawMessage, error) {
@@ -59,6 +69,22 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3
if fp, err := ownBuildVia(own, "n1")(context.Background()); err != nil || fp != "src2:abc" || asked != "seat:node-engine.composition@n1" {
t.Errorf("own build = %q %v asked %q", fp, err, asked)
}
// Issue 1921: the start during a replacement. What runs still names the old build, the want (composed by the node itself)
// the successor, which this process is: it fixes the successor. A want that is not self, or is removed, leaves what runs.
for _, row := range []struct{ name, answer, want string }{
{"replacing: runs old, want new and self", `{"runs":{"fp":"src2:old"},"want":{"fp":"src2:new","self":true}}`, "src2:new"},
{"a want the node does not compose", `{"runs":{"fp":"src2:old"},"want":{"fp":"src2:new"}}`, "src2:old"},
{"replacing: the apply of the want failed or is held", `{"runs":{"fp":"src2:old"},"want":{"fp":"src2:new","self":true},"attempt":{"fp":"src2:new","held":true}}`, "src2:old"},
{"replacing: an attempt of another build", `{"runs":{"fp":"src2:old"},"want":{"fp":"src2:new","self":true},"attempt":{"fp":"src2:older"}}`, "src2:new"},
{"the want already runs", `{"runs":{"fp":"src2:new"},"want":{"fp":"src2:new","self":true}}`, "src2:new"},
{"a removed want", `{"runs":{"fp":"src2:old"},"want":{"fp":"src2:new","self":true,"removed":true}}`, "src2:old"},
{"no build runs yet, want self", `{"want":{"fp":"src2:new","self":true}}`, "src2:new"},
} {
ask := func(string, any) (json.RawMessage, error) { return json.RawMessage(row.answer), nil }
if fp, err := ownBuildVia(ask, "n1")(context.Background()); err != nil || fp != row.want {
t.Errorf("%s: own build = %q %v, want %q", row.name, fp, err, row.want)
}
}
none := func(string, any) (json.RawMessage, error) { return json.RawMessage(`{}`), nil }
if _, err := ownBuildVia(none, "n1")(context.Background()); err == nil {
t.Error("an engine that says nothing runs gave a build")
@@ -25,10 +25,32 @@ func ownBuildVia(call toolCaller, node string) func(ctx context.Context) (string
Runs *struct {
FP string `json:"fp"`
} `json:"runs"`
Want *struct {
FP string `json:"fp"`
Self bool `json:"self"`
Removed bool `json:"removed"`
} `json:"want"`
// Attempt is the engine's last apply that did not end running (failed, or held), with the build it was of.
Attempt *struct {
FP string `json:"fp"`
} `json:"attempt"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return "", fmt.Errorf("the answer of %s is not a composition: %w", compositionVerb, err)
}
// While the engine's apply of the want is in progress (what runs still names the old build, the want names another, and no
// attempt names that want as failed or held), this process is the successor and fixes the want's build. self is the node's
// standing way of composing the module, not a mark of this process, so it is not enough alone: an old process started again
// while the want names a successor that does not run (its apply failed or is held) takes what runs. The start fixes the
// build once (quick fix, ADR 0639 §7: nothing is learnt later, nothing waits on a timer); the thorough form is the
// engine passing each process the fingerprint it starts it with.
running := ""
if a.Runs != nil {
running = a.Runs.FP
}
if w := a.Want; w != nil && w.Self && !w.Removed && w.FP != "" && w.FP != running && (a.Attempt == nil || a.Attempt.FP != w.FP) {
return w.FP, nil
}
if a.Runs == nil || a.Runs.FP == "" {
return "", errors.New("the engine says no fingerprint of the indexer runs")
}
@@ -121,6 +121,8 @@ type Indexer struct {
filling atomic.Bool // a FillTaken runs in the background (KickFill)
fillAgain atomic.Bool // a kick arrived while it ran
atHeads map[string]map[string]delivery.PublishedPackages // a member head's published Go modules, by repository@head (candidatePublished)
saidAt map[string]string // the head of a repository whose failed read was said, by repository: said once per head
failedAt map[string]string // the head of a repository whose tree lacks a directory asked, by repository: not asked again until the head changes
// OwnBuild answers the fingerprint of the build this program runs, from the node-engine's seat; nil knows none.
OwnBuild func(ctx context.Context) (string, error)
underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go)
@@ -2,6 +2,7 @@ package index
import (
"context"
"errors"
"fmt"
"maps"
"path"
@@ -297,6 +298,12 @@ func (ix *Indexer) candidatePublished(ctx context.Context, members map[string]st
ix.mu.Lock()
got, ok := ix.atHeads[at]
ix.mu.Unlock()
ix.mu.Lock()
failed := ix.failedAt[repo] == head
ix.mu.Unlock()
if failed {
continue // said once for this head; the candidate takes the published version until the head changes
}
if !ok {
var ms []string
ms = append(ms, modules[repo]...)
@@ -307,10 +314,29 @@ func (ix *Indexer) candidatePublished(ctx context.Context, members map[string]st
}
var err error
if got, err = ix.packagesAt(ctx, repo, head, ms, imports[repo]); err != nil {
ix.logf("%s at the head %s: its packages are not read, so the candidate takes their published version: %v", repo, head, err)
// A failure of the tree itself (the directory is not in the tree at that head) no retry changes: it is asked once per
// head. Any other failure may be transient: it is asked again at the next pass, and said once per head.
ix.mu.Lock()
said := ix.saidAt[repo] == head
if ix.saidAt == nil {
ix.saidAt = map[string]string{}
}
ix.saidAt[repo] = head
if errors.Is(err, ErrNotInTree) {
if ix.failedAt == nil {
ix.failedAt = map[string]string{}
}
ix.failedAt[repo] = head
}
ix.mu.Unlock()
if !said {
ix.logf("%s at the head %s: its packages are not read, so the candidate takes their published version: %v", repo, head, err)
}
continue
}
ix.mu.Lock()
delete(ix.failedAt, repo)
delete(ix.saidAt, repo)
if ix.atHeads == nil || len(ix.atHeads) >= maxHeadsKept {
ix.atHeads = map[string]map[string]delivery.PublishedPackages{}
}
@@ -322,5 +348,9 @@ func (ix *Indexer) candidatePublished(ctx context.Context, members map[string]st
return out
}
// ErrNotInTree is a failure of the tree itself: a directory asked of the forge is not in the tree at that commit. No retry
// changes it, so a head that fails so is asked once (candidatePublished).
var ErrNotInTree = errors.New("the directory is not in the tree at that commit")
// maxHeadsKept bounds the member heads' packages kept between recomputes; past it they are read again.
const maxHeadsKept = 64
@@ -559,6 +559,10 @@ func (ix *Indexer) TakeInAtStart(ctx context.Context) (Applied, string, error) {
if errors.Is(err, process.ErrNotYet) {
return a, "the pending state is not on the bus yet: nothing waits", nil
}
if err == nil && len(a.Written) == 0 {
// A start that takes nothing in says so, with the build it learnt (the wait's own build is in the line holdFor says).
ix.logf("take-in: at the start, this program runs build %s and took nothing in: %s", own, words)
}
return a, words, err
}
@@ -978,6 +982,10 @@ func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own
for _, old := range saidWaits(ended) {
ix.sayTakeIn(ctx, old)
}
if said && !changed {
// Nothing to say on the bus, but the start decided nothing and a reader of the journal sees why: the build learnt, the build the wait names.
ix.logf("take-in: %d merges, oldest %s, still name the indexer build %s, this program runs %s: they wait, as already said", len(ready), t.Key, names, own)
}
if !said || changed {
ix.logf("take-in: %d merges, oldest %s, name the indexer build %s, not %s, which runs: they wait for it", len(ready), t.Key, names, own)
if refused != "" {
@@ -281,6 +281,18 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
n.FixOwnBuild(ctx)
return n
}
// A start that decides nothing says so (issue 1921): the build it learnt, the build the held wait names, also when the
// wait was said before; and indexer_pending names the known build and since when.
oldBuild, _ := b.ix.OwnBuild(ctx)
oldStart := mk(func(context.Context) (string, error) { return oldBuild, nil })
b.logs.Reset()
if _, words, err := oldStart.TakeInAtStart(ctx); err != nil || len(b.keys()) != 2 || !strings.Contains(b.logs.String(), "still name the indexer build") ||
!strings.Contains(b.logs.String(), "at the start, this program runs build") {
t.Fatalf("a start that holds: %q %v, journal %q: want the build learnt and the build the wait names said", words, err, b.logs.String())
}
if answer, err := PendingTools(b.pending, b.ix.OwnBuildState, nil, nil)[0].Run(map[string]any{}); err != nil || answer.(takein.PendingAnswer).OwnBuild != oldBuild || answer.(takein.PendingAnswer).OwnBuildSince.IsZero() {
t.Errorf("indexer_pending = %+v %v, want the known build and since", answer, err)
}
asked := 0
next := mk(func(context.Context) (string, error) { asked++; return "", errors.New("not yet") })
if r := next.TakeInPendingByHand(ctx, "m.nothing", "lead@shanks", "a test"); r.Stored || !strings.Contains(r.Answer, "nothing waits") {
@@ -53,7 +53,7 @@ func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (T
"merges, since when it waits, the modules it holds back (holds), the merges whose message the forge bridge's stream no " +
"longer holds (lost: they are not taken in; indexer_drop_pending drops them) and the last take-in event said of it. Empty waits: " +
"every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " +
"indexer_take_in_pending and indexer_drop_pending steer a wait. own-build-unknown, when set, says the indexer cannot learn the build it " +
"indexer_take_in_pending and indexer_drop_pending steer a wait. own-build says which build the indexer knows it runs and own-build-since since when; own-build-unknown, when set, says the indexer cannot learn the build it " +
"runs (since when): it fixes it once at its start, so every merge waits (a take-in.pending that says so) until the module is started again.",
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) {
@@ -69,8 +69,13 @@ func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (T
if ownBuild != nil {
if v, _, found, err := ownBuild.Get(context.Background(), OwnBuildKey); err == nil && found {
var o OwnBuild
if json.Unmarshal(v, &o) == nil && !o.Known {
answer.OwnBuildUnknown, answer.OwnBuildSince = o.Why, o.Since
if json.Unmarshal(v, &o) == nil {
answer.OwnBuildSince = o.Since
if o.Known {
answer.OwnBuild = o.FP
} else {
answer.OwnBuildUnknown = o.Why
}
}
}
}
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"runtime"
"slices"
@@ -404,6 +405,39 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW
}
})
}
// Issue 1921: a head whose tree lacks the directory asked is asked once, however many passes read it, until the head changes;
// any other failure (the forge does not answer) is asked again at every pass and said once per head.
for _, row := range []struct {
name string
err error
askedAfterTwo int // the same head read twice
askedAfterNew int // then another head
linesAfterNew int
}{
{"the directory is not in the tree", fmt.Errorf("listing: %w", ErrNotInTree), 1, 2, 2},
{"the forge does not answer", errors.New("the forge does not answer"), 2, 3, 2},
} {
t.Run("a failing head: "+row.name, func(t *testing.T) {
ix, _, _ := previewBed(t, false)
var logs strings.Builder
ix.Logf = func(format string, args ...any) { logs.WriteString(fmt.Sprintf(format, args...) + "\n") }
asked := 0
ix.Files = func(context.Context, string, string, []string) (map[string]map[string]string, error) {
asked++
return nil, row.err
}
held := map[string]delivery.Module{"x": {Inputs2: &delivery.KeptInputs2{Packages: []string{"git.novox.be/novox/mesh-sdk/go"}}}}
for pass, head := range []string{h1, h1, h2} {
ix.candidatePublished(context.Background(), map[string]string{"novox/mesh-sdk": head}, held)
if pass == 1 && (asked != row.askedAfterTwo || strings.Count(logs.String(), "its packages are not read") != 1) {
t.Fatalf("after two passes over %s: asked %d (want %d), journal %q: want one line", head, asked, row.askedAfterTwo, logs.String())
}
}
if asked != row.askedAfterNew || strings.Count(logs.String(), "its packages are not read") != row.linesAfterNew {
t.Errorf("asked %d (want %d), journal %q (want %d lines)", asked, row.askedAfterNew, logs.String(), row.linesAfterNew)
}
})
}
}
func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T) {
@@ -396,6 +396,7 @@ func (ix *Indexer) FixOwnBuild(ctx context.Context) {
ix.logf("judge: the build this program runs is not known (%v), so every merge waits and a manifest with a field it does not know waits and is not refused; it is learnt only at the start, so the indexer is started again", err)
} else {
st.Known, st.FP = true, fp
ix.logf("judge: this program runs the indexer build %s, fixed here at its start", fp)
}
ix.ownMu.Lock()
ix.ownFP, ix.ownSince, ix.ownWhy = fp, st.Since, st.Why
@@ -254,6 +254,8 @@ type PendingAnswer struct {
// (a take-in.pending with the same words) and the indexer is started again to learn it. Empty when the build is known.
OwnBuildUnknown string `json:"own-build-unknown,omitempty"`
OwnBuildSince time.Time `json:"own-build-since,omitzero"`
// OwnBuild is the build the running indexer knows it runs, fixed at its start; OwnBuildSince is since when. Empty when unknown.
OwnBuild string `json:"own-build,omitempty"`
}
// DecodePendingAnswer reads indexer_pending's answer, refusing a wait that is not a revision or a group, or names no