B-D14b/W4.8 (indexer half, integration fixtures left): build modules from the build seats' acts and their holders' manifests (issue 1966)

Design 47 *A module's registration*, "No part fixes a name" (b229f5c4 lines 482-485), and *What ADR 0656
leaves to this design*, "The indexer reads it [acts] into the catalog graph" (line 1583): ToolchainModules,
ToolchainStreams, BuildGoModule and BuildNodeModule go. The build module of a language is the module whose
manifest claims the seat whose acting rule is step build on it (index.BuildModulesOf), its stream the one of
its manifest carrying toolchain.reported, its bucket <holder>_toolchains as before. Found again at start, at
every change of the seat list and at every catalog revision; the watch, the gate and the hearing follow.
A language with no build module found reads as no toolchain reported.

Rows: TestAgreedToolchainChangesAGoModulesReasonWithNoMerge/the build module of a language is the holder...
(fails without: no BuildModulesOf, and go-builds' report refused); TestToolchainWatch's later build module.
Left: six integration tests of cmd/mesh-delivery-indexer hold no build seat with acts nor a holder manifest
and fail (TestIndexerWhole, TestIndexerWritesAGoModuleAgain..., TestIndexerGivesACandidateItsFingerprint...,
TestIndexerJudgesAnUnreadableVerdictAgainAtStart, TestIndexerCountsOnlyThePlacedBuildNodesToolchains,
TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage): their fixtures are to be given both.

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

Builds: design 47 G §364, §428, §535
This commit is contained in:
2026-11-11 14:36:48 +01:00
parent 7a7b504836
commit 371b1a98f6
16 changed files with 410 additions and 67 deletions
@@ -0,0 +1,108 @@
package main
import (
"context"
"maps"
"slices"
"sync"
"sync/atomic"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"mesh-delivery-indexer/internal/index"
)
// buildModulesFollow keeps the build modules the indexer takes toolchains of as the seat list and the manifests it holds
// say them (index.BuildModulesOf; design 47 *A module's registration*, "No part fixes a name"): found again at start and
// at every change of the seat list, each new one's toolchains state watched, its stream heard and the gate told.
type buildModulesFollow struct {
ctx context.Context
ix *index.Indexer
defs func() map[string]seatdef.Definition // the seat list reader's definitions
gate interface{ UseModules([]string) bool }
watch *toolchainWatch
logf func(string, ...any)
mu sync.Mutex
hear func(stream, module string) // nil until the indexer's consumers are bound (hearWith)
heard map[string]bool // streams heard
watched map[string]bool // build modules whose toolchains watch was asked
}
// refresh finds the build modules again and acts on what moved.
func (f *buildModulesFollow) refresh() {
by, err := f.ix.BuildModulesNow(f.ctx, f.defs())
if err != nil {
f.logf("the build modules are not found again, the ones found before stand: %v", err)
return
}
modules := map[string]string{} // module -> stream
for _, b := range by {
modules[b.Module] = b.Stream
}
names := slices.Sorted(maps.Keys(modules))
if f.gate != nil && f.gate.UseModules(names) {
if err := f.ix.ToolchainReported(f.ctx); err != nil {
f.logf("writing the modules again after the build modules changed: %v", err)
}
}
f.mu.Lock()
defer f.mu.Unlock()
if f.heard == nil {
f.heard, f.watched = map[string]bool{}, map[string]bool{}
}
for _, module := range names {
if f.watch == nil || f.watched[module] {
continue
}
f.watched[module] = true
if err := f.watch.ensure(module); err != nil {
f.logf("the toolchains of %s are not watched: %v", module, err)
}
}
if f.hear == nil {
return
}
for _, module := range names {
if s := modules[module]; s != "" && !f.heard[s] {
f.heard[s] = true
f.hear(s, module)
}
}
}
// watching is the toolchain watch, started with the build modules found so far: refresh asks it to watch only the ones
// found after.
func (f *buildModulesFollow) watching(w *toolchainWatch) {
f.mu.Lock()
defer f.mu.Unlock()
f.watch = w
if f.watched == nil {
f.heard, f.watched = map[string]bool{}, map[string]bool{}
}
for _, module := range buildModuleNames(f.ix.BuildModules()) {
f.watched[module] = true
}
}
// hearWith hears the stream of every build module found, and of each found later, through hear.
func (f *buildModulesFollow) hearWith(hear func(stream, module string)) {
f.mu.Lock()
f.hear = hear
f.mu.Unlock()
f.refresh()
}
// refreshingBuildModules is say, and then the build modules found again (in the background, so the take-in that said the
// revision does not wait on it), once the follow is made: a revision's manifests may claim a build seat anew.
func refreshingBuildModules(say func(context.Context, meshcatalog.CatalogRevisionEvent) error, follow *atomic.Pointer[buildModulesFollow]) func(context.Context, meshcatalog.CatalogRevisionEvent) error {
return func(ctx context.Context, e meshcatalog.CatalogRevisionEvent) error {
err := say(ctx, e)
if f := follow.Load(); f != nil {
go f.refresh()
}
return err
}
}
@@ -32,6 +32,7 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"mesh-delivery-indexer/internal/index"
)
@@ -404,7 +405,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
seatDefs := process.AwaitBucket(ctx, js, conn, index.SeatDefinitionsBucket, "no seat definition is kept", logger.Printf)
marks := process.AwaitBucket(ctx, js, conn, index.MarksBucket, "no change is said on the views family, so every reader waits on its first view", logger.Printf)
seatBus := &seatBus{defs: awaitedSource{ctx, seatDefs}, marks: awaitedSource{ctx, marks}}
seatReader, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf)
// The build modules are found again at every change of the seat list (buildmodules.go), once the indexer is made.
var buildFollow atomic.Pointer[buildModulesFollow]
seatReader, saySeats := followSeatList(seatBus, saySeatDefinition(js, cred), logger.Printf, func() {
if b := buildFollow.Load(); b != nil {
go b.refresh() // never inside the reader's change: the refresh reads the reader's definitions
}
})
readWhenBound(seatReader, seatDefs, marks)
ix := &index.Indexer{Findings: awaitedStore{b: findings}, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known,
Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred),
@@ -414,7 +421,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, TryRecovery: awaitedStore{b: tryRecovery}, Graph: graph,
SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats,
Message: messageVia(calls.CallKept), Revisions: revisions, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1)
SayCatalogRevision: sayCatalogRevision(conn, cred),
// Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go).
SayCatalogRevision: refreshingBuildModules(sayCatalogRevision(conn, cred), &buildFollow),
SayFill: sayFill(conn, cred),
// The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear;
// until then every merge is taken in at once, as before.
@@ -459,7 +467,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go source.catchUp(ctx)
// Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the
// assignments is settled first, so the toolchains settle through the gate already knowing it.
gate := buildrun.NewToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf)
// The build modules are no names (design 47 *A module's registration*): the holders of the seats whose acting rule is
// step build on a language, among the manifests the indexer holds (index.BuildModulesOf), told to the gate as found.
gate := buildrun.NewToolchainGate(nil, ix.PutToolchain, logger.Printf)
buildModules := &buildModulesFollow{ctx: ctx, ix: ix, gate: gate, logf: logger.Printf, defs: func() map[string]seatdef.Definition {
if r := seatReader.Reader(); r != nil {
return r.Definitions()
}
return nil
}}
buildModules.refresh()
// The assignments is the assigner's: read through its first view and its views family, never its bucket (DLV-F1r).
assignmentsErrs := make(chan error)
// The indexer's one pull of its durable on the shared events stream (process.EventHub): every handler is registered
@@ -480,10 +497,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// the toolchain.
return ix.ToolchainReported(ctx)
},
recompute: ix.Recompute}, logger)
recompute: ix.Recompute}, func() []string { return buildModuleNames(ix.BuildModules()) }, logger)
if err := tools.start(); err != nil {
return err
}
buildModules.watching(tools)
buildFollow.Store(buildModules)
// The checker's merge previews give the candidates their fingerprints: the bucket is watched where it exists, and
// otherwise when the checker says it wrote a preview.
previews := newPreviewWatch(ctx, js, ix, logger)
@@ -590,9 +609,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return nil
})
}
for _, module := range sortedModules() {
hear(index.ToolchainStreams[module], module+"'s toolchain reports are not heard until the indexer starts again", func(m jetstream.Msg) { tools.Reported(m) })
}
buildModules.hearWith(func(stream, module string) {
hear(stream, module+"'s toolchain reports are not heard until the indexer starts again", func(m jetstream.Msg) { tools.Reported(m) })
})
// The consumer on the checker's stream exists before the open tries are listed and is heard after (ADR 0638 §3): an event
// said between the call and the hearing waits in it, and a try that started or ended meanwhile is found either way.
var checksCons jetstream.Consumer
@@ -111,9 +111,9 @@ func (b *seatBus) said(c index.SeatDefinitionChanged) {
// the indexer says is also heard by the reader. Its two tools are not served: the indexer's process serves no request
// subject (ADR 0437 §4), and its list is its own state seat-definitions, which indexer_seat_definition and the seat's verb
// seat-definitions read. It returns the part, whose status the indexer's health can show.
func followSeatList(bus *seatBus, say func(context.Context, index.SeatDefinitionChanged) error, logf func(string, ...any)) (*seatlist.Part, func(context.Context, index.SeatDefinitionChanged) error) {
func followSeatList(bus *seatBus, say func(context.Context, index.SeatDefinitionChanged) error, logf func(string, ...any), changed func()) (*seatlist.Part, func(context.Context, index.SeatDefinitionChanged) error) {
part := seatlist.NewPart(toolPrefix, nil)
part.Follow(bus, nil, logf)
part.Follow(bus, changed, logf)
return part, func(ctx context.Context, c index.SeatDefinitionChanged) error {
if err := say(ctx, c); err != nil {
return err
@@ -44,7 +44,7 @@ func (b *awaitedFake) Ready(f func(jetstream.KeyValue)) { b.ready = append(b.rea
func TestTheSeatListIsReadAgainWhenItsStatesAreBound(t *testing.T) {
src := &unboundSource{}
bus := &seatBus{defs: src, marks: src}
part, _ := followSeatList(bus, nil, t.Logf)
part, _ := followSeatList(bus, nil, t.Logf, nil)
reader := part.Reader()
<-reader.Ready()
if reader.Status().Heard {
@@ -5,7 +5,6 @@ import (
"errors"
"fmt"
"log"
"maps"
"slices"
"strings"
"sync"
@@ -38,6 +37,7 @@ type bucketOpener interface {
// (toolchain.reported, handled by Reported); a bucket not watched yet is then watched, settled as the others.
type toolchainWatch struct {
ctx context.Context
modules func() []string // the build modules (index.Indexer.BuildModules), found from the seat list and the manifests
js bucketOpener
sink toolchainSink
logger *log.Logger
@@ -46,13 +46,13 @@ type toolchainWatch struct {
watched map[string]bool
}
func newToolchainWatch(ctx context.Context, js bucketOpener, sink toolchainSink, logger *log.Logger) *toolchainWatch {
return &toolchainWatch{ctx: ctx, js: js, sink: sink, logger: logger, errs: make(chan error, len(index.ToolchainModules)), watched: map[string]bool{}}
func newToolchainWatch(ctx context.Context, js bucketOpener, sink toolchainSink, modules func() []string, logger *log.Logger) *toolchainWatch {
return &toolchainWatch{ctx: ctx, js: js, sink: sink, modules: modules, logger: logger, errs: make(chan error, 1), watched: map[string]bool{}}
}
// start watches the buckets that exist now; a missing one is said and waits for its build module's event.
func (w *toolchainWatch) start() error {
for _, module := range sortedModules() {
for _, module := range w.modules() {
if err := w.ensure(module); err != nil {
return err
}
@@ -84,11 +84,15 @@ func (w *toolchainWatch) ensure(module string) error {
}
w.watched[module] = true
go func() {
w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) {
err := follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) {
if err := w.sink.apply(w.ctx, module, e.Key(), e.Value(), !opOf(e)); err != nil {
w.logger.Printf("%s.%s could not be taken into the modules: %v", name, e.Key(), err)
}
})
select { // the first watch that ends stops the indexer; a later one has no reader left
case w.errs <- err:
default:
}
}()
}
return w.sink.recompute(w.ctx)
@@ -99,10 +103,7 @@ func (w *toolchainWatch) ensure(module string) error {
// again. A subject of no build module is terminated; a failure stays unacknowledged, and the server delivers it again.
func (w *toolchainWatch) Reported(m delivered) {
module, _, _ := strings.Cut(strings.TrimPrefix(m.Subject(), "mesh.stream."), ".event.")
known := false
for _, k := range index.ToolchainModules {
known = known || k == module
}
known := slices.Contains(w.modules(), module)
if !known || !strings.HasPrefix(m.Subject(), "mesh.stream.") {
w.logger.Printf("%s: not a build module's event, dropped", m.Subject())
_ = m.Term()
@@ -115,4 +116,14 @@ func (w *toolchainWatch) Reported(m delivered) {
_ = m.Ack()
}
func sortedModules() []string { return slices.Sorted(maps.Values(index.ToolchainModules)) }
// buildModuleNames are the modules of the build modules by language, sorted.
func buildModuleNames(by map[string]index.BuildModule) []string {
var out []string
for _, b := range by {
if !slices.Contains(out, b.Module) {
out = append(out, b.Module)
}
}
slices.Sort(out)
return out
}
@@ -109,11 +109,14 @@ func TestABuildModulesBucketThatAppearsLaterIsWatchedWhenItsEventArrives(t *test
var logs bytes.Buffer
// 1945: a bus that refuses both buckets, or does not answer, leaves them unwatched, said, and the indexer running.
refusing := &laterBuckets{appears: map[string]jetstream.KeyValue{}, asked: map[string]int{}, refuse: errors.New("context deadline exceeded")}
if err := newToolchainWatch(ctx, refusing, sink, log.New(&logs, "", 0)).start(); err != nil || !strings.Contains(logs.String(), "cannot be read") {
// The build modules are what the seat list and the manifests say (buildmodules.go): two at first, a third found later.
found := []string{"mesh-build-go", "mesh-build-node"}
modules := func() []string { return found }
if err := newToolchainWatch(ctx, refusing, sink, modules, log.New(&logs, "", 0)).start(); err != nil || !strings.Contains(logs.String(), "cannot be read") {
t.Fatalf("a refused toolchains bucket: start %v, log %q; want it said and the start going on", err, logs.String())
}
logs.Reset()
w := newToolchainWatch(ctx, buckets, sink, log.New(&logs, "", 0))
w := newToolchainWatch(ctx, buckets, sink, modules, log.New(&logs, "", 0))
if err := w.start(); err != nil { // neither bucket exists: nothing is watched, nothing is written, nobody waits
t.Fatal(err)
}
@@ -151,4 +154,13 @@ func TestABuildModulesBucketThatAppearsLaterIsWatchedWhenItsEventArrives(t *test
if other.termd != 1 || other.acked != 0 {
t.Errorf("terminated %d acked %d for a module that is no build module", other.termd, other.acked)
}
// A build module found later, under any name, is heard: its report is taken, not terminated.
later := &fakeMsg{subject: "mesh.stream.go-builds.event.toolchain.reported"}
w.Reported(later)
found = append(found, "go-builds")
again := &fakeMsg{subject: "mesh.stream.go-builds.event.toolchain.reported"}
w.Reported(again)
if later.termd != 1 || again.acked != 1 || again.termd != 0 {
t.Errorf("go-builds before it was found: terminated %d; after: acked %d terminated %d", later.termd, again.acked, again.termd)
}
}
@@ -0,0 +1,139 @@
package index
import (
"context"
"fmt"
"maps"
"slices"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
// The build module of a language is no name in code (design 47 *A module's registration*, "No part fixes a name": every
// module name a part needs is read from the registrations and the catalog graph; *What ADR 0656 leaves to this design*: a
// seat's acting rule `acts`, which "the indexer reads into the catalog graph"). It is the holder of the seat whose acting
// rule builds that language (`acts` {step: build, on: <language>}), found by the claims of the manifests the indexer
// holds; its toolchains are its state toolchains (ToolchainsBucket), and its toolchain reports ride the stream of its
// manifest's `streams` that carries ToolchainReportedEvent.
// ToolchainReportedEvent is the event a build module says each time it wrote its toolchains key, on a stream of its own.
const ToolchainReportedEvent = "toolchain.reported"
// BuildModule is the build module of one language, as the seat list and the manifests say it.
type BuildModule struct {
Language string // the language the seat's acting rule builds
Seat string // the seat whose acting rule it is
Module string // the module whose manifest claims the seat
Stream string // the stream carrying ToolchainReportedEvent, as the bus names it; empty when its manifest declares none
}
// BuildModulesOf are the build modules by language: for each seat whose acting rule is step build on a language, the
// module among manifests that claims it. A language two seats build, a seat two modules claim, a seat nobody claims and a
// holder declaring no stream for its reports are said in the problems; the first two give no build module of that
// language, the last one a build module with no stream.
func BuildModulesOf(defs map[string]seatdef.Definition, manifests []catalogue.Manifest) (map[string]BuildModule, []string) {
var problems []string
seats := map[string][]string{} // language -> the seats building it
for _, name := range slices.Sorted(maps.Keys(defs)) {
for _, r := range defs[name].Acts {
if r.Step != seatdef.StepBuild || r.Acting() == seatdef.OnEvery {
continue
}
seats[r.Acting()] = append(seats[r.Acting()], name)
}
}
out := map[string]BuildModule{}
for _, language := range slices.Sorted(maps.Keys(seats)) {
if len(seats[language]) > 1 {
problems = append(problems, fmt.Sprintf("the seats %v each build %s: no build module of %s is taken", seats[language], language, language))
continue
}
seat := seats[language][0]
var holders []catalogue.Manifest
for _, m := range manifests {
if slices.ContainsFunc(m.Claims, func(c catalogue.Claim) bool { return c.Name == seat }) {
holders = append(holders, m)
}
}
switch len(holders) {
case 0:
problems = append(problems, fmt.Sprintf("no module the indexer holds claims %s, which builds %s", seat, language))
continue
case 1:
default:
var names []string
for _, h := range holders {
names = append(names, h.Module)
}
slices.Sort(names)
problems = append(problems, fmt.Sprintf("the modules %v each claim %s, which builds %s: no build module of %s is taken", names, seat, language, language))
continue
}
b := BuildModule{Language: language, Seat: seat, Module: holders[0].Module}
for _, s := range holders[0].Streams {
if s.Carries(ToolchainReportedEvent) {
b.Stream = catalogue.StreamBusName(s.Name)
break
}
}
if b.Stream == "" {
problems = append(problems, fmt.Sprintf("%s, which builds %s, declares no stream carrying %s: its reports are not heard", b.Module, language, ToolchainReportedEvent))
}
out[language] = b
}
return out, problems
}
// UseBuildModules is what BuildModulesNow found: the build modules whose toolchains the indexer takes in from now on.
func (ix *Indexer) UseBuildModules(by map[string]BuildModule) {
ix.mu.Lock()
defer ix.mu.Unlock()
ix.buildModules = maps.Clone(by)
}
// BuildModules are the build modules by language the indexer takes toolchains of (UseBuildModules).
func (ix *Indexer) BuildModules() map[string]BuildModule {
ix.mu.Lock()
defer ix.mu.Unlock()
return maps.Clone(ix.buildModules)
}
// BuildModulesNow finds the build modules from the seat definitions defs and the manifests of the modules the indexer
// holds (BuildModulesOf), says each problem in the log, uses them (UseBuildModules) and answers them.
func (ix *Indexer) BuildModulesNow(ctx context.Context, defs map[string]seatdef.Definition) (map[string]BuildModule, error) {
held, err := ix.heldModules(ctx)
if err != nil {
return nil, err
}
var manifests []catalogue.Manifest
for _, key := range slices.Sorted(maps.Keys(held)) {
if held[key].Manifest == "" || ix.Manifests == nil {
continue
}
_, m, err := ix.manifestRaw(ctx, held[key].Manifest)
if err != nil {
ix.logf("modules.%s: its manifest %s cannot be read, so it holds no build seat here: %v", key, held[key].Manifest, err)
continue
}
manifests = append(manifests, m)
}
by, problems := BuildModulesOf(defs, manifests)
for _, p := range problems {
ix.logf("%s", p)
}
ix.UseBuildModules(by)
return by, nil
}
// languageBuiltBy is the language module is the build module of, or "".
func (ix *Indexer) languageBuiltBy(module string) string {
ix.mu.Lock()
defer ix.mu.Unlock()
for language, b := range ix.buildModules {
if b.Module == module {
return language
}
}
return ""
}
@@ -138,25 +138,26 @@ type Indexer struct {
// closed message the indexer missed heals at the next one (issue 1126); nil asks none.
ReadOpen func(ctx context.Context) ([]forgemsg.OpenPull, error)
mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
source sourceNow
previews map[string]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue // by previews key (previews.go)
toolchains map[string]map[string]nodebuildgo.Toolchain // language, then node (toolchains.go)
tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go)
unread map[string]bool // registered modules whose last read the forge could not answer, read again by RetryUnread
moving map[string]forgemsg.Forge // by modules key: a merge naming a module another repository holds, waiting for that trunk to drop it (moves.go)
retrying atomic.Bool // a RetryUnread runs in the background (kickRetry)
retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one
retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on
fresh map[string]bool // modules whose import form is computed again by hand (FingerprintsAgain), "*" for every one; under freshMu
freshMu sync.Mutex
recompute sync.Mutex // one Recompute at a time
untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go)
filling atomic.Bool // a FillTaken runs in the background (KickFill)
fillAgain atomic.Bool // a kick arrived while it ran
atHeads map[string]map[string]meshcatalog.PackagesByModuleValue // 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
mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
source sourceNow
previews map[string]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue // by previews key (previews.go)
toolchains map[string]map[string]nodebuildgo.Toolchain // language, then node (toolchains.go)
buildModules map[string]BuildModule // by language: the build modules whose toolchains are taken in (buildmodules.go)
tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go)
unread map[string]bool // registered modules whose last read the forge could not answer, read again by RetryUnread
moving map[string]forgemsg.Forge // by modules key: a merge naming a module another repository holds, waiting for that trunk to drop it (moves.go)
retrying atomic.Bool // a RetryUnread runs in the background (kickRetry)
retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one
retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on
fresh map[string]bool // modules whose import form is computed again by hand (FingerprintsAgain), "*" for every one; under freshMu
freshMu sync.Mutex
recompute sync.Mutex // one Recompute at a time
untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go)
filling atomic.Bool // a FillTaken runs in the background (KickFill)
fillAgain atomic.Bool // a kick arrived while it ran
atHeads map[string]map[string]meshcatalog.PackagesByModuleValue // 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)
@@ -38,7 +38,7 @@ type graphEdges struct {
// graphNode is the graph node of e: its edges, the awaited provisions judged against providers (catalogue.Awaited, as
// the assigner judges them).
func graphNode(e graphEdges, providers func(string) []catalogue.Manifest) graph.Node {
func graphNode(e graphEdges, providers func(string) []catalogue.Manifest, builders map[string]BuildModule) graph.Node {
if e.fp == "" && e.none == "" {
e.none = "its fingerprint was not computed"
}
@@ -50,10 +50,11 @@ func graphNode(e graphEdges, providers func(string) []catalogue.Manifest) graph.
tcs = e.cur.Toolchains // no fingerprint now: the toolchain it was last computed with still builds it
}
for language, t := range tcs {
builder, known := ToolchainModules[language]
b, known := builders[language]
if !known {
continue
}
builder := b.Module
if n.Toolchains == nil {
n.Toolchains = map[string]graph.Toolchain{}
}
@@ -81,7 +82,7 @@ func (ix *Indexer) writeGraph(ctx context.Context, edges map[string]graphEdges)
GraphMost, len(names), names[GraphMost], len(names)-GraphMost-1)
names = names[:GraphMost]
}
whole := wholeGraph(edges)
whole := wholeGraph(edges, ix.BuildModules())
for _, name := range names {
key, err := graph.Key(name)
if err != nil {
@@ -129,7 +130,7 @@ func (ix *Indexer) writeGraph(ctx context.Context, edges map[string]graphEdges)
// wholeGraph is the module graph of every module of edges, the verb edges on a cycle left out over the whole graph
// (ADR 0614 §3), so a module past GraphMost still closes the cycle it lies on.
func wholeGraph(edges map[string]graphEdges) graph.Graph {
func wholeGraph(edges map[string]graphEdges, builders map[string]BuildModule) graph.Graph {
providers := func(provision string) []catalogue.Manifest {
var out []catalogue.Manifest
for _, name := range slices.Sorted(maps.Keys(edges)) {
@@ -141,7 +142,7 @@ func wholeGraph(edges map[string]graphEdges) graph.Graph {
}
whole := graph.Graph{}
for name, e := range edges {
whole[name] = graphNode(e, providers)
whole[name] = graphNode(e, providers, builders)
}
whole.LeaveOut()
return whole
@@ -62,6 +62,7 @@ func modulesIndexer(t *testing.T, recs ...[2]any) (*Indexer, *memStore) {
t.Helper()
mods := newMem()
ix := &Indexer{Store: newMem(), Modules: mods, Builds: NewBuilds()}
ix.UseBuildModules(fixtureBuildModules)
for _, r := range recs {
if err := ix.Builds.Put(r[1].([]byte)); err != nil {
t.Fatal(err)
@@ -70,6 +71,14 @@ func modulesIndexer(t *testing.T, recs ...[2]any) (*Indexer, *memStore) {
return ix, mods
}
// fixtureBuildModules are the build modules BuildModulesOf gives for the seats node-build-go and node-build-node (acts
// {step: build, on: go} and {step: build, on: typescript}) held by the build modules' manifests, as the fixtures' toolchain
// reports name them.
var fixtureBuildModules = map[string]BuildModule{
"go": {Language: "go", Seat: "node-build-go", Module: "mesh-build-go", Stream: "TOOLCHAINS_GO"},
"typescript": {Language: "typescript", Seat: "node-build-node", Module: "mesh-build-node", Stream: "TOOLCHAINS_NODE"},
}
func rec(t *testing.T, id, module, repo, path, ref, state, at, inputs string) [2]any {
k, v := record(t, id, module, repo, path, ref, state, at, inputs)
return [2]any{k, v}
@@ -29,8 +29,6 @@ const (
AssignerModule = "mesh-delivery-assigner"
GateModule = "mesh-delivery-gate"
CheckerModule = "mesh-delivery-checker"
BuildGoModule = "mesh-build-go"
BuildNodeModule = "mesh-build-node"
AssignmentsState = "assignments"
AssignmentsBucket = AssignerModule + "_" + AssignmentsState
AssignmentsHistory = 5
@@ -929,6 +929,7 @@ func (ix *Indexer) dry() *dryRun {
for l, by := range ix.toolchains {
x.toolchains[l] = maps.Clone(by)
}
x.buildModules = maps.Clone(ix.buildModules) // the dry run computes with the build modules the running indexer takes
x.moving = maps.Clone(ix.moving)
x.untaken = maps.Clone(ix.untaken)
ix.mu.Unlock()
@@ -429,6 +429,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
mk := func(ask func(context.Context) (string, error)) *Indexer {
n := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: b.ix.Message, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum}
n.UseBuildModules(b.ix.BuildModules())
n.toolchains = b.ix.toolchains
n.OwnBuild = ask
n.FixOwnBuild(ctx)
@@ -1032,6 +1033,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
next := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: bridgeDown, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf,
GoSum: b.ix.GoSum, OwnBuild: func(context.Context) (string, error) { return named, nil }}
next.UseBuildModules(b.ix.BuildModules())
next.toolchains = b.ix.toolchains
next.FixOwnBuild(ctx)
b.said = nil
@@ -1112,6 +1114,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
},
SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum, OwnBuild: func(context.Context) (string, error) { return named, nil }}
next.toolchains = b.ix.toolchains
next.UseBuildModules(b.ix.BuildModules())
next.FixOwnBuild(ctx)
if _, _, err := next.TakeInAtStart(ctx); err != nil {
t.Fatal(err)
@@ -27,26 +27,12 @@ import (
// records builds and makes no artifact's bytes (ADR 0498 §1), so its Builder input is empty for every module, the builder's
// own included; a change of how an artifact is built is a change of its build module's recipe (ADR 0530 §2).
// ToolchainModules are the build module of each language, whose state toolchains (ToolchainsBucket) the
// indexer reads: the language's agreed toolchain is the one every node's entry in it agrees on.
var ToolchainModules = map[string]string{"go": "mesh-build-go", "typescript": "mesh-build-node"}
// ToolchainStreams are the stream each build module carries its toolchain.reported event on, by module, as the bus names it
// (a stream's local name in capitals, catalogue.StreamBusName): `toolchains-go` and `toolchains-node`, declared by the
// build modules (streams, with the event toolchain.reported) and heard by this module (consumes).
var ToolchainStreams = map[string]string{"mesh-build-go": "TOOLCHAINS_GO", "mesh-build-node": "TOOLCHAINS_NODE"}
// PutToolchain takes one change of a build module's toolchains bucket in, without writing any module: a put of a node's
// toolchain of a language (key <node>.<language>, ToolchainKey), or its delete. It answers whether what the
// language's toolchain is of (version and recipe) changed. A key, or a value, that is not a toolchain of the module's
// language is said in the log and left out.
func (ix *Indexer) PutToolchain(module, key string, value []byte, deleted bool) bool {
language := ""
for l, m := range ToolchainModules {
if m == module {
language = l
}
}
language := ix.languageBuiltBy(module) // the build modules the seat list and the manifests say (buildmodules.go)
if language == "" {
ix.logf("toolchains of %q are not read: it is no build module the indexer knows", module)
return false
@@ -105,7 +91,7 @@ func (ix *Indexer) agreed() map[string]toolchainOrNone {
ix.mu.Lock()
defer ix.mu.Unlock()
out := map[string]toolchainOrNone{}
for language := range ToolchainModules {
for language := range ix.buildModules {
var entries []sdkstep.NodeToolchain
for _, node := range slices.Sorted(maps.Keys(ix.toolchains[language])) {
t := ix.toolchains[language][node]
@@ -280,7 +266,11 @@ func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]meshc
toolchains := map[string]meshcatalogclient.Toolchain2{}
var noToolchain string
for _, artifact := range slices.Sorted(maps.Keys(k.languages)) {
t := tools[k.languages[artifact]]
t, found := tools[k.languages[artifact]]
if !found { // no build module of the language is found: no toolchain of it is reported
_, none := sdkstep.AgreedToolchain(nil)
t = toolchainOrNone{none: none}
}
if t.none != "" {
if noToolchain == "" {
noToolchain = t.none
@@ -375,7 +365,7 @@ func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]meshc
if err := ix.writeGraph(ctx, edges); err != nil {
return err
}
return ix.keepTrunkRevision(ctx, edges, wholeGraph(edges))
return ix.keepTrunkRevision(ctx, edges, wholeGraph(edges, ix.BuildModules()))
}
// writeOverlay writes the records of a merge, each once and only if it differs from what the bucket holds; finals has
@@ -24,6 +24,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"git.novox.be/novox/mesh-sdk/go/views"
"git.novox.be/novox/mesh-delivery/sdk/graph"
@@ -147,6 +148,7 @@ func reportToolchainRecipe(t *testing.T, ix *Indexer, module, node, language, ve
}
func TestAgreedToolchainChangesAGoModulesReasonWithNoMerge(t *testing.T) { // 1444/AC5
t.Run("the build module of a language is the holder of the seat whose acting rule builds it", theBuildModuleOfALanguageIsTheHolderOfItsBuildSeat)
ix, mods := newFormIndexer(t, mod{"avahi", catalogueManifest(t, "avahi")})
if m, _ := moduleOf(t, mods, "avahi"); m.None2 != sdkstep.ToolchainNoReport {
t.Run("a recipe change builds a Go module at once and again once the agreement moves", aRecipeChangeBuildsAGoModuleAtOnceAndAgainOnceTheAgreementMoves)
@@ -195,6 +197,55 @@ func TestAgreedToolchainChangesAGoModulesReasonWithNoMerge(t *testing.T) { // 14
}
}
// Design 47 *A module's registration* ("No part fixes a name") and *What ADR 0656 leaves to this design* (the indexer reads
// `acts` into the catalog graph): the build module of a language is the module claiming the seat whose acting rule is step
// build on it, whatever its name, its stream the one of its manifest carrying toolchain.reported; a language two seats build
// and a seat two modules claim give none, a holder with no such stream one with no stream, each said. The indexer takes a
// toolchain of that module only.
func theBuildModuleOfALanguageIsTheHolderOfItsBuildSeat(t *testing.T) {
defs := map[string]seatdef.Definition{
"node-build-go": {Seat: "node-build-go", Acts: []seatdef.ActingRule{{Step: seatdef.StepBuild, On: "go", Bound: "60m"}}},
"node-build-node": {Seat: "node-build-node", Acts: []seatdef.ActingRule{{Step: seatdef.StepBuild, On: "typescript"}}},
"node-build-image": {Seat: "node-build-image"},
"mesh-catalog": {Seat: "mesh-catalog", Acts: []seatdef.ActingRule{{Step: seatdef.StepTakeIn}, {Step: seatdef.StepBuild}}},
}
claiming := func(module, seat string, events ...string) catalogue.Manifest {
m := catalogue.Manifest{Module: module, Claims: []catalogue.Claim{{Name: seat}}}
if len(events) > 0 {
m.Streams = []catalogue.StreamDeclaration{{Name: "toolchains-" + module, Events: events}}
}
return m
}
by, problems := BuildModulesOf(defs, []catalogue.Manifest{claiming("go-builds", "node-build-go", "toolchain.**"),
claiming("ts-builds", "node-build-node", "step.ended"), claiming("avahi", "node-mdns")})
want := map[string]BuildModule{"go": {Language: "go", Seat: "node-build-go", Module: "go-builds", Stream: "TOOLCHAINS_GO_BUILDS"},
"typescript": {Language: "typescript", Seat: "node-build-node", Module: "ts-builds"}}
if !reflect.DeepEqual(by, want) || len(problems) != 1 || !strings.Contains(problems[0], "ts-builds, which builds typescript, declares no stream") {
t.Errorf("build modules %+v problems %q, want %+v and the stream ts-builds lacks said", by, problems, want)
}
twice := maps.Clone(defs)
twice["node-build-go-2"] = seatdef.Definition{Seat: "node-build-go-2", Acts: []seatdef.ActingRule{{Step: seatdef.StepBuild, On: "go"}}}
by, problems = BuildModulesOf(twice, []catalogue.Manifest{claiming("go-builds", "node-build-go"), claiming("other", "node-build-node"), claiming("third", "node-build-node")})
if len(by) != 0 || len(problems) != 2 || !strings.Contains(problems[0], "each build go") || !strings.Contains(problems[1], "[other third] each claim node-build-node") {
t.Errorf("two seats building go and two holders of node-build-node: %+v %q, want none, both said", by, problems)
}
// The indexer finds them among the manifests it holds, and a report of a module that holds no build seat is left out.
builder := `{"module":"go-builds","version":"1","claims":[{"name":"node-build-go"}],"streams":[{"name":"toolchains-go","events":["toolchain.**"]}]}`
ix, _ := newFormIndexer(t, mod{"avahi", catalogueManifest(t, "avahi")}, mod{"go-builds", builder})
ix.UseBuildModules(nil)
got, err := ix.BuildModulesNow(context.Background(), defs)
if err != nil || !reflect.DeepEqual(got, map[string]BuildModule{"go": {Language: "go", Seat: "node-build-go", Module: "go-builds", Stream: "TOOLCHAINS_GO"}}) {
t.Fatalf("the held manifests give %+v (%v), want go-builds building go", got, err)
}
key, _ := ToolchainKey("novox", "go")
if ix.PutToolchain("mesh-build-go", key, toolchainValue(t, "go", "go1.27.1", "1"), false) {
t.Errorf("a toolchain of mesh-build-go, which holds no build seat here, was taken in")
}
if !ix.PutToolchain("go-builds", key, toolchainValue(t, "go", "go1.27.1", "1"), false) {
t.Errorf("a toolchain of go-builds, the holder of node-build-go, was not taken in")
}
}
// ADR 0603 §3 (D9): a source revision that changes the Go build module's recipe and a Go module builds the module at once, with
// the recipe the nodes run, and again once the agreement moves to the new one: the module never waits and never loses its
// fingerprint between the two, and its toolchain edge names the recipe each was computed with.
@@ -144,7 +144,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
}
}
for bucket, read := range map[string]string{index.PreviewsBucket: "mesh-delivery-checker.previews",
index.ToolchainsBucket(index.BuildGoModule): "mesh-build-go.toolchains", index.ToolchainsBucket(index.BuildNodeModule): "mesh-build-node.toolchains"} {
index.ToolchainsBucket("mesh-build-go"): "mesh-build-go.toolchains", index.ToolchainsBucket("mesh-build-node"): "mesh-build-node.toolchains"} {
if !slices.Contains(m.Reads, read) || strings.Replace(read, ".", "_", 1) != bucket {
t.Errorf("the indexer reads the bucket %s, its manifest's reads are %v (want %s)", bucket, m.Reads, read)
}