From 371b1a98f6c35550baf26f3860acb8bd3a6cdff1 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 17:32:21 +0100 Subject: [PATCH] B-D14b/W4.8 (indexer half, integration fixtures left): build modules from the build seats' acts and their holders' manifests (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 _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 --- .../cmd/mesh-delivery-indexer/buildmodules.go | 108 ++++++++++++++ .../cmd/mesh-delivery-indexer/main.go | 33 ++++- .../cmd/mesh-delivery-indexer/seatlist.go | 4 +- .../mesh-delivery-indexer/seatlist_test.go | 2 +- .../cmd/mesh-delivery-indexer/toolchains.go | 31 ++-- .../mesh-delivery-indexer/toolchains_test.go | 16 +- .../internal/index/buildmodules.go | 139 ++++++++++++++++++ .../internal/index/candidates.go | 39 ++--- .../internal/index/graph.go | 11 +- .../internal/index/modules_test.go | 9 ++ .../internal/index/names.go | 2 - .../internal/index/pending.go | 1 + .../internal/index/pending_test.go | 3 + .../internal/index/toolchains.go | 26 +--- .../internal/index/toolchains_test.go | 51 +++++++ .../mesh-delivery-indexer/manifest_test.go | 2 +- 16 files changed, 410 insertions(+), 67 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/buildmodules.go create mode 100644 modules/mesh-delivery-indexer/internal/index/buildmodules.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/buildmodules.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/buildmodules.go new file mode 100644 index 00000000..26c70d7c --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/buildmodules.go @@ -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 + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index d5c65a9a..23cb3c21 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go index 252093ee..8a2ebea9 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist.go @@ -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 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go index 2e90b5d9..1773a17b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/seatlist_test.go @@ -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 { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go index a25d19ea..f9621d99 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go @@ -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 +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go index dfbbbcaa..ea81a11c 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go @@ -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) + } } diff --git a/modules/mesh-delivery-indexer/internal/index/buildmodules.go b/modules/mesh-delivery-indexer/internal/index/buildmodules.go new file mode 100644 index 00000000..189baa57 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/buildmodules.go @@ -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: }), 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 "" +} diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 32932c23..1989d64e 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -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) diff --git a/modules/mesh-delivery-indexer/internal/index/graph.go b/modules/mesh-delivery-indexer/internal/index/graph.go index 2d37097f..f3b2f867 100644 --- a/modules/mesh-delivery-indexer/internal/index/graph.go +++ b/modules/mesh-delivery-indexer/internal/index/graph.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 diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go index 8dd4d180..30b398f4 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules_test.go +++ b/modules/mesh-delivery-indexer/internal/index/modules_test.go @@ -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} diff --git a/modules/mesh-delivery-indexer/internal/index/names.go b/modules/mesh-delivery-indexer/internal/index/names.go index 7a53689a..8a905656 100644 --- a/modules/mesh-delivery-indexer/internal/index/names.go +++ b/modules/mesh-delivery-indexer/internal/index/names.go @@ -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 diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 9461dc1b..b40c5445 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -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() diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 0cce565d..a0846e5e 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -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) diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go index 0f8b0f29..b3e9940a 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -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 ., 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 diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go index 4afc5f88..4e1e75aa 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -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. diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 5cedc957..86c7182e 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -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) }