From 74fef7dd5a8bc1a6f6eb783cda86d901740c0eef Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 20:47:35 +0000 Subject: [PATCH 1/3] Recompute the new-form fingerprints when a build module reports its toolchain (issue 1444), part 1 of 2 --- .../internal/index/candidates.go | 6 +- .../internal/index/modules.go | 6 +- .../internal/index/toolchains.go | 306 ++++++++++++++++++ .../internal/index/toolchains_test.go | 196 +++++++++++ 4 files changed, 510 insertions(+), 4 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/toolchains.go create mode 100644 modules/mesh-delivery-indexer/internal/index/toolchains_test.go diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 747700c..80ab566 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -48,8 +48,10 @@ type Indexer struct { Builds *Builds Logf func(format string, args ...any) - mu sync.Mutex // guards source, which the source watch and the forge messages both reach - source sourceNow + mu sync.Mutex // guards source and toolchains, which the watches and the forge messages all reach + source sourceNow + toolchains map[string]map[string]delivery.Toolchain // language, then node (toolchains.go) + recompute sync.Mutex // one Recompute at a time } func (ix *Indexer) logf(format string, args ...any) { diff --git a/modules/mesh-delivery-indexer/internal/index/modules.go b/modules/mesh-delivery-indexer/internal/index/modules.go index 753df87..1538192 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules.go +++ b/modules/mesh-delivery-indexer/internal/index/modules.go @@ -50,7 +50,8 @@ func fingerprintOf(k Known, trees map[string]string) (fp, none string) { // (ApplySource), and Src 0 when it has not. A module of another repository is not written. A value the module already // holds is not written again. From the manifests and lock hashes the message carries (the forge bridge, issue 1448) it // also writes the module's manifest digest, its new-form inputs and why it has no new-form fingerprint (issues 1444, 1449, -// 1450); this release writes no fingerprint of the new form for any module. +// 1450); the fingerprint itself is computed by Recompute (toolchains.go) once the merge is written, and again when a build +// module reports its toolchain. func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error { digests, err := ix.keepManifests(ctx, f) if err != nil { @@ -89,7 +90,8 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error { return fmt.Errorf("modules.%s: %w", key, err) } } - return nil + // What this merge moved reaches the modules standing on it (a base's tree, build-agent's fingerprint), in the graph's order. + return ix.Recompute(ctx) } // baseFP answers the new-form fingerprint of a base module as modules. holds it, or why it has none. diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go new file mode 100644 index 0000000..fac1331 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -0,0 +1,306 @@ +package index + +import ( + "context" + "fmt" + "maps" + "slices" + "strings" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The new-form fingerprint of a module is written, and written again with no merge, when what it is of changes +// (novox/hq ADR 0465 clause 2; issue 1444): a module's tree and lock hashes come with a merge and are kept in +// modules..Inputs2; its builder's, its bases' and its toolchain's come from here. The indexer holds no graph, so each +// recompute reads the manifest of every module from manifests. and takes the modules in the order their bases +// and the builder come first. + +const ( + // BuilderModule is the module that builds every module (novox/hq ADR 0453 clause 5): each module's Builder input is + // its new-form fingerprint. It does not exist before release R4c-4, and a module then has none for want of it. + BuilderModule = "mesh-delivery-builder" + // BuilderBuiltBy is the module that builds BuilderModule until release R4g: its fingerprint of today's form is the + // builder's own Builder input, so a change of build-agent moves the builder's value (ADR 0453 clause 5). + BuilderBuiltBy = "build-agent" +) + +// ToolchainModules are the build module of each language, whose state toolchains (delivery.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"} + +// 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 ., delivery.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 + } + } + if language == "" { + ix.logf("toolchains of %q are not read: it is no build module the indexer knows", module) + return false + } + node, keyLanguage, err := delivery.ParseToolchainKey(key) + if err != nil || keyLanguage != language { + ix.logf("%s_toolchains.%s is not a %s toolchain's key, left out: %v", module, key, language, err) + return false + } + ix.mu.Lock() + defer ix.mu.Unlock() + if deleted { + if _, had := ix.toolchains[language][node]; !had { + return false + } + delete(ix.toolchains[language], node) + return true + } + tc, err := delivery.DecodeToolchain(value) + if err != nil || tc.Language != language { + ix.logf("%s_toolchains.%s is not a %s toolchain, left out: %v", module, key, language, err) + return false + } + if ix.toolchains == nil { + ix.toolchains = map[string]map[string]delivery.Toolchain{} + } + if ix.toolchains[language] == nil { + ix.toolchains[language] = map[string]delivery.Toolchain{} + } + cur, had := ix.toolchains[language][node] + ix.toolchains[language][node] = tc + return !had || cur.Version != tc.Version || cur.Recipe != tc.Recipe +} + +// ApplyToolchain takes one change of a build module's toolchains in (PutToolchain) and, when the language's toolchain +// changed, writes the modules again that it moves (Recompute). The same version and recipe seen again writes nothing. +func (ix *Indexer) ApplyToolchain(ctx context.Context, module, key string, value []byte, deleted bool) error { + if !ix.PutToolchain(module, key, value, deleted) { + return nil + } + return ix.Recompute(ctx) +} + +// agreed is the toolchain of each language every reporting node agrees on, or why there is none. +func (ix *Indexer) agreed() map[string]toolchainOrNone { + ix.mu.Lock() + defer ix.mu.Unlock() + out := map[string]toolchainOrNone{} + for language := range ToolchainModules { + var entries []delivery.NodeToolchain + for _, node := range slices.Sorted(maps.Keys(ix.toolchains[language])) { + entries = append(entries, delivery.NodeToolchain{Node: node, Toolchain: ix.toolchains[language][node]}) + } + tc, none := delivery.AgreedToolchain(entries) + out[language] = toolchainOrNone{tc, none} + } + return out +} + +type toolchainOrNone struct { + tc delivery.Toolchain2 + none string +} + +// known is a module the indexer holds the inputs of: its record as read, the manifest's shape and the key it is written at. +type known struct { + key string + cur delivery.Module + languages map[string]string + bases []string + contexts bool +} + +// Recompute writes again every module whose new-form fingerprint, or reason for none, is no longer what it holds: from +// the inputs kept at its merge (Inputs2), its manifest (manifests.), the fingerprints of its bases and of the +// builder, and the toolchains the build modules report. A module is computed after its bases and the builder, whatever +// the order the bucket lists them in; modules that stand on each other have a cycle as their reason. A module is written +// only when its fingerprint or its reason changes, and only if its manifest is still the one computed from. A module with +// no kept inputs (a message that carried no manifest) is left as it is. +func (ix *Indexer) Recompute(ctx context.Context) error { + if ix.Modules == nil || ix.Manifests == nil { + return nil + } + ix.recompute.Lock() + defer ix.recompute.Unlock() + keys, err := ix.Modules.Keys(ctx) + if err != nil { + return fmt.Errorf("listing the modules: %w", err) + } + slices.Sort(keys) + held := map[string]delivery.Module{} + all := map[string]*known{} + for _, key := range keys { + value, _, found, err := ix.Modules.Get(ctx, key) + if err != nil { + return fmt.Errorf("modules.%s: %w", key, err) + } + if !found { + continue + } + cur, err := delivery.DecodeModule(value) + if err != nil { + ix.logf("modules.%s holds what is not a module, left: %v", key, err) + continue + } + held[key] = cur + if cur.Manifest == "" || cur.Inputs2 == nil { + continue + } + m, err := ix.manifestOf(ctx, cur.Manifest) + if err != nil { + ix.logf("modules.%s: its manifest %s cannot be read, left: %v", key, cur.Manifest, err) + continue + } + all[key] = &known{key: key, cur: cur, languages: languagesOf(m), bases: basesOf(m), contexts: hasContext(m)} + } + tools := ix.agreed() + type result struct{ fp, none string } + done := map[string]result{} + visiting := map[string]bool{} + var of func(module string, chain []string) result + of = func(module string, chain []string) result { + if r, ok := done[module]; ok { + return r + } + k, ours := all[module] + if !ours { // a module the indexer holds no inputs of: its record, as written + cur, has := held[module] + switch { + case !has: + return result{"", "the indexer holds no record of it"} + case cur.FP2 != "": + return result{cur.FP2, ""} + case cur.None2 != "": + return result{"", cur.None2} + } + return result{"", "its record carries no fingerprint of the new form"} + } + if visiting[module] { + return result{"", "a cycle of bases: " + strings.Join(append(chain, module), " -> ")} + } + visiting[module] = true + defer delete(visiting, module) + chain = append(chain, module) + kept := *k.cur.Inputs2 + kept.Bases, kept.Builder = nil, "" + var missing []string + for _, b := range k.bases { + if r := of(b, chain); r.fp != "" { + kept.Bases = append(kept.Bases, r.fp) + } else { + missing = append(missing, fmt.Sprintf("the base %s has no fingerprint (%s)", b, r.none)) + } + } + slices.Sort(kept.Bases) + slices.Sort(missing) + toolchains := map[string]delivery.Toolchain2{} + var noToolchain string + for _, artifact := range slices.Sorted(maps.Keys(k.languages)) { + t := tools[k.languages[artifact]] + if t.none != "" { + if noToolchain == "" { + noToolchain = t.none + } + continue + } + toolchains[artifact] = t.tc + } + var noBuilder string + if module == BuilderModule { + // Built by build-agent until release R4g: its value of today's form. + if a := held[BuilderBuiltBy]; a.FP != "" { + kept.Builder = a.FP + } else { + noBuilder = fmt.Sprintf("%s, which builds it, has no fingerprint (%s)", BuilderBuiltBy, a.None) + } + } else if r := of(BuilderModule, chain); r.fp != "" { + kept.Builder = r.fp + } else { + noBuilder = None2Builder + if _, has := held[BuilderModule]; has { + noBuilder += ": " + r.none + } + } + var r result + switch { + case k.contexts: + r.none = None2Context + case len(missing) > 0: + r.none = strings.Join(missing, "; ") + case noToolchain != "": + r.none = noToolchain + case noBuilder != "": + r.none = noBuilder + default: + r.fp, r.none = delivery.SourceFingerprint2(kept.With(module, toolchains)) + } + done[module] = r + k.cur.Inputs2 = &kept + return r + } + for _, key := range slices.Sorted(maps.Keys(all)) { + k := all[key] + r := of(key, nil) + if k.cur.FP2 == r.fp && k.cur.None2 == r.none { + continue + } + digest, inputs := k.cur.Manifest, k.cur.Inputs2 + err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) { + if !found { + return nil, false, nil + } + cur, err := delivery.DecodeModule(value) + if err != nil || cur.Manifest != digest || cur.Inputs2 == nil { + return nil, false, nil // a merge changed it meanwhile; its own recompute follows + } + if cur.FP2 == r.fp && cur.None2 == r.none { + return nil, false, nil + } + cur.FP2, cur.None2 = r.fp, r.none + cur.Inputs2 = inputs + data, err := delivery.EncodeModule(cur) + return data, false, err + }) + if err != nil { + return fmt.Errorf("modules.%s: %w", key, err) + } + } + return nil +} + +// manifestOf is the parsed manifest manifests. holds. +func (ix *Indexer) manifestOf(ctx context.Context, digest string) (catalogue.Manifest, error) { + key, err := delivery.ManifestsKey(digest) + if err != nil { + return catalogue.Manifest{}, err + } + value, _, found, err := ix.Manifests.Get(ctx, key) + if err != nil { + return catalogue.Manifest{}, err + } + if !found { + return catalogue.Manifest{}, fmt.Errorf("manifests.%s is not there", key) + } + kept, err := delivery.DecodeManifest(value) + if err != nil { + return catalogue.Manifest{}, err + } + return catalogue.ParseManifest([]byte(kept.JSON)) +} + +// hasContext is whether an artifact of the manifest reads another repository's tree. +func hasContext(m catalogue.Manifest) bool { + if m.Build == nil { + return false + } + for _, a := range m.Build.Artifacts { + if a.Context != nil { + return true + } + } + return false +} diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go new file mode 100644 index 0000000..32de130 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -0,0 +1,196 @@ +package index + +import ( + "context" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The fixtures here are mesh-catalog a67f0196's manifests (testdata, captured from that commit) with the module renamed (and, +// where a test needs one, a build.on entry added as the controller's own test has it, standing_on_test.go:22): the manifest +// parser, not a hand-written manifest, reads them. + +func renamed(t *testing.T, from, to string) string { + t.Helper() + raw := catalogueManifest(t, from) + out := strings.Replace(raw, `"module": "`+from+`"`, `"module": "`+to+`"`, 1) + if out == raw { + t.Fatalf("%s's manifest has no module line", from) + } + return out +} + +func standingOn(t *testing.T, raw, base string) string { + t.Helper() + out := strings.Replace(raw, `"build": {`, `"build": { + "on": [{"arg": "RUNTIME_BASE", "module": "`+base+`", "artifact": "runtime"}],`, 1) + if out == raw { + t.Fatal("the manifest has no build") + } + return out +} + +const buildAgentRepo = "novox/mesh-controller" + +type mod struct { + name, manifest string +} + +// merged applies one merge of the catalogue carrying these modules (every one a Go bundle's go.sum, a TypeScript bundle's +// package-lock.json) and returns the indexer. +func newFormIndexer(t *testing.T, mods ...mod) (*Indexer, *memStore) { + t.Helper() + var recs [][2]any + trees, manifests := map[string]string{}, map[string]string{} + locks := map[string]map[string]string{} + for i, m := range mods { + dir := "modules/" + m.name + recs = append(recs, rec(t, "build-"+m.name, m.name, catalogueRepo, dir, "main", "", "2026-10-19T09:00:00Z", aInputs)) + trees[dir], manifests[dir] = "tree-"+m.name, m.manifest + locks[dir] = map[string]string{"go.sum": avahiGoSum, "package-lock.json": strings.Repeat("c", 63) + string(rune('a'+i))} + } + recs = append(recs, rec(t, "build-agent-1", "build-agent", buildAgentRepo, "modules/build-agent", "main", "", "2026-10-19T09:00:00Z", aInputs)) + ix, store, _ := manifestsIndexer(t, recs...) + f := withManifests(mergedMsg(catalogueRepo, trees), manifests, locks) + if err := ix.Apply(context.Background(), f); err != nil { + t.Fatal(err) + } + // build-agent's value of today's form, from a merge of the controller (modules.build-agent holds its fingerprint). + if err := ix.Apply(context.Background(), mergedMsg(buildAgentRepo, map[string]string{"modules/build-agent": "agent-tree-1"})); err != nil { + t.Fatal(err) + } + return ix, store +} + +// toolchainValue is a toolchains value as the build module's writer encodes it (mesh-sdk go/delivery EncodeToolchain). +func toolchainValue(t *testing.T, language, version, recipe string) []byte { + t.Helper() + b, err := delivery.EncodeToolchain(delivery.Toolchain{Language: language, Version: version, Recipe: recipe, At: time.Now()}) + if err != nil { + t.Fatal(err) + } + return b +} + +func reportToolchain(t *testing.T, ix *Indexer, module, node, language, version string) { + t.Helper() + key, err := delivery.ToolchainKey(node, language) + if err != nil { + t.Fatal(err) + } + if err := ix.ApplyToolchain(context.Background(), module, key, toolchainValue(t, language, version, "1"), false); err != nil { + t.Fatal(err) + } +} + +func TestAgreedToolchainChangesAGoModulesReasonWithNoMerge(t *testing.T) { // 1444/AC5 + ix, mods := newFormIndexer(t, mod{"avahi", catalogueManifest(t, "avahi")}) + if m, _ := moduleOf(t, mods, "avahi"); m.None2 != delivery.ToolchainNoReport { + t.Fatalf("before: none2 = %q", m.None2) + } + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + reportToolchain(t, ix, "mesh-build-go", "ace", "go", "go1.27.1") + m, _ := moduleOf(t, mods, "avahi") + if m.None2 != None2Builder || m.FP2 != "" { + t.Errorf("none2 = %q fp2 = %q, want none2 %q", m.None2, m.FP2, None2Builder) + } +} + +func TestNodesReportingDifferentVersionsLeaveTheReasonNamingBoth(t *testing.T) { // 1444/AC6 + ix, mods := newFormIndexer(t, mod{"avahi", catalogueManifest(t, "avahi")}) + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + reportToolchain(t, ix, "mesh-build-go", "ace", "go", "go1.26.0") + m, _ := moduleOf(t, mods, "avahi") + if !strings.Contains(m.None2, "builders disagree") || !strings.Contains(m.None2, "go1.27.1") || !strings.Contains(m.None2, "go1.26.0") { + t.Errorf("none2 = %q, want 'builders disagree' naming both versions", m.None2) + } +} + +// builderModules are the builder, a base, a module on the base (named so that it is stored before its base) and a module +// whose base is a TypeScript module with no toolchain reported. +func builderModules(t *testing.T) []mod { + return []mod{ + {"mesh-delivery-builder", renamed(t, "avahi", "mesh-delivery-builder")}, + {"mesh-tools", renamed(t, "avahi", "mesh-tools")}, + {"avahi", standingOn(t, catalogueManifest(t, "avahi"), "mesh-tools")}, + {"other", standingOn(t, renamed(t, "avahi", "other"), "ts-base")}, + {"ts-base", renamed(t, "audit-logger", "ts-base")}, + } +} + +func goInputs(t *testing.T, ix *Indexer, mods *memStore, module, builder string, bases ...string) delivery.SourceInputs2 { + t.Helper() + return delivery.SourceInputs2{Module: module, Tree: "tree-" + module, Builder: builder, Bases: bases, + Locks: map[string]string{"tools": avahiGoSum}, + Toolchains: map[string]delivery.Toolchain2{"tools": {Language: "go", Version: "go1.27.1", Recipe: "1"}}} +} + +func fp2(t *testing.T, in delivery.SourceInputs2) string { + t.Helper() + fp, none := delivery.SourceFingerprint2(in) + if none != "" { + t.Fatal(none) + } + return fp +} + +func TestBasesAndTheBuilderAreComputedBeforeTheModulesTakingThem(t *testing.T) { // 1444/AC7, AC8 + ix, mods := newFormIndexer(t, builderModules(t)...) + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + agent, _ := moduleOf(t, mods, "build-agent") + if agent.FP == "" { + t.Fatalf("build-agent has no fingerprint of today's form: %+v", agent) + } + builderFP := fp2(t, goInputs(t, ix, mods, "mesh-delivery-builder", agent.FP)) + toolsFP := fp2(t, goInputs(t, ix, mods, "mesh-tools", builderFP)) + want := map[string]string{"mesh-delivery-builder": builderFP, "mesh-tools": toolsFP, "avahi": fp2(t, goInputs(t, ix, mods, "avahi", builderFP, toolsFP))} + for name, fp := range want { + if m, _ := moduleOf(t, mods, name); m.FP2 != fp || m.None2 != "" { + t.Errorf("%s: fp2 = %q none2 = %q, want %s", name, m.FP2, m.None2, fp) + } + } + if m, _ := moduleOf(t, mods, "other"); m.FP2 != "" || !strings.Contains(m.None2, "ts-base") || !strings.Contains(m.None2, "base") { + t.Errorf("other: fp2 = %q none2 = %q, want none2 naming the base ts-base", m.FP2, m.None2) + } + // 1444/AC8: build-agent changes at a merge of the controller; the builder's value moves with it, and so do the modules built by it. + if err := ix.Apply(context.Background(), mergedMsg(buildAgentRepo, map[string]string{"modules/build-agent": "agent-tree-2"})); err != nil { + t.Fatal(err) + } + moved, _ := moduleOf(t, mods, "build-agent") + if moved.FP == agent.FP { + t.Fatal("build-agent's fingerprint did not change") + } + newBuilder := fp2(t, goInputs(t, ix, mods, "mesh-delivery-builder", moved.FP)) + if m, _ := moduleOf(t, mods, "mesh-delivery-builder"); m.FP2 != newBuilder || newBuilder == builderFP { + t.Errorf("builder: fp2 = %q, want %s (was %s)", m.FP2, newBuilder, builderFP) + } + if m, _ := moduleOf(t, mods, "mesh-tools"); m.FP2 != fp2(t, goInputs(t, ix, mods, "mesh-tools", newBuilder)) { + t.Errorf("mesh-tools did not follow the builder: %+v", m) + } +} + +func TestAToolchainEntryChangedToTheSameValuesWritesNothing(t *testing.T) { // 1444/AC9 + ix, mods := newFormIndexer(t, builderModules(t)...) + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + before := mods.writes + time.Sleep(2 * time.Millisecond) + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") // the same version and recipe, a later time + if mods.writes != before { + t.Errorf("modules written %d times after the same values, want none", mods.writes-before) + } +} + +func TestModulesThatStandOnEachOtherAreNamedAsACycle(t *testing.T) { // 1444 (the indexer has no graph: a cycle is a reason) + ix, mods := newFormIndexer(t, + mod{"aa", standingOn(t, renamed(t, "avahi", "aa"), "bb")}, + mod{"bb", standingOn(t, renamed(t, "avahi", "bb"), "aa")}) + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + for _, name := range []string{"aa", "bb"} { + if m, _ := moduleOf(t, mods, name); m.FP2 != "" || !strings.Contains(m.None2, "cycle") { + t.Errorf("%s: fp2 = %q none2 = %q, want a reason naming the cycle", name, m.FP2, m.None2) + } + } +} From 127ac64ccefa1feea71959ee9290934e10db1409 Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 20:48:30 +0000 Subject: [PATCH 2/3] Leave an artifact without its lock file out of Locks (issue 1459) and read the build modules' toolchains (issue 1444), part 2 of 2 --- .../mesh-delivery-indexer/integration_test.go | 77 +++++++++++++++++- .../cmd/mesh-delivery-indexer/main.go | 49 ++++++++++++ .../internal/index/inputs.go | 78 +++++++++++++------ .../internal/index/inputs_test.go | 35 ++++++--- .../mesh-delivery-indexer/manifest_test.go | 4 +- modules/mesh-delivery-indexer/module.json | 2 + 6 files changed, 209 insertions(+), 36 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index dc88ca3..3962c90 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -95,7 +95,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10 // startIndexer starts the bus, the stream FORGE with the indexer's durable consumer, the buckets, lets put fill // mesh-controller_builds, and runs the program. mesh-delivery-gate_source is never created. -func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) { +func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) { dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i") if err != nil { t.Fatal(err) @@ -154,6 +154,9 @@ func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue)) ( t.Fatal(err) } put(ctx, builds) + for _, b := range before { + b(ctx, js) + } broker := filepath.Join(t.TempDir(), "broker") cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-indexer", User: "n1.mesh-delivery-indexer", Password: "pw"}) @@ -265,3 +268,75 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4 t.Errorf("manifest = %+v err %v, want avahi and the file's bytes", got, err) } } + +// TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain: the bucket mesh-build-go_toolchains exists (the state +// toolchains of the build module, delivery.ToolchainsBucket) and holds no key; a merged message of mesh-catalog carrying +// avahi's real manifest makes modules.avahi with none2 'toolchain not yet reported'; a put of the key n1.go (the build +// module's own layout, delivery.ToolchainKey and EncodeToolchain) makes, within 5 s and with no merge, none2 "builder's +// fingerprint not yet known". +func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *testing.T) { // 1444/AC10 + manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json") + if err != nil { + t.Fatal(err) + } + const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381" + const inputs = `{"Module":"avahi","Tree":"oldtree","Bases":null,"Contexts":null,"Toolchains":{"code":"go sha256:ccc 0123456789abcdef"},"Unpinned":""}` + js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds jetstream.KeyValue) { + rec := `{"id":"build-1","repository":"novox/mesh-catalog","ref":"main","module":"avahi","commit":"7a74cee83f88e6585a5511c3cc53bdf0f5303d20","path":"modules/avahi","at":"2026-10-19T09:38:00Z","source-inputs":` + inputs + `}` + if _, err := builds.Put(ctx, delivery.BuildKey("avahi", "build-1"), []byte(rec)); err != nil { + t.Fatal(err) + } + if _, err := builds.Put(ctx, delivery.NewestKey("avahi"), []byte("build-1")); err != nil { + t.Fatal(err) + } + }, func(ctx context.Context, js jetstream.JetStream) { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1}); err != nil { + t.Fatal(err) + } + }) + data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, + Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0", + Trees: map[string]string{"modules/avahi": "newtree"}, + Manifests: map[string]string{"modules/avahi": string(manifest)}, + Locks: map[string]map[string]string{"modules/avahi": {"go.sum": goSum}}}) + if err != nil { + t.Fatal(err) + } + if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.merged", data); err != nil { + t.Fatal(err) + } + none2 := func(want string) string { + var got string + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := modules.Get(ctx, "avahi"); err == nil { + if m, err := delivery.DecodeModule(e.Value()); err == nil { + if got = m.None2; got == want { + break + } + } + } + } + return got + } + if got := none2(delivery.ToolchainNoReport); got != delivery.ToolchainNoReport { + t.Fatalf("none2 = %q within 5 s, want %q", got, delivery.ToolchainNoReport) + } + key, err := delivery.ToolchainKey("n1", "go") + if err != nil { + t.Fatal(err) + } + value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: "go1.27.1", Recipe: "1", At: time.Now()}) + if err != nil { + t.Fatal(err) + } + kv, err := js.KeyValue(ctx, delivery.ToolchainsBucket("mesh-build-go")) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, value); err != nil { + t.Fatal(err) + } + if got := none2("builder's fingerprint not yet known"); got != "builder's fingerprint not yet known" { + t.Errorf("none2 = %q within 5 s of the toolchain key, want the builder's fingerprint not yet known", got) + } +} 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 ca8db63..4abb8e3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -15,8 +15,10 @@ import ( "fmt" "io" "log" + "maps" "os" "os/signal" + "slices" "strings" "syscall" @@ -296,6 +298,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } go func() { sourceErrs <- follow(ctx, sw, each) }() } + toolErrs, err := watchToolchains(ctx, js, ix, logger) + if err != nil { + return err + } cons, err := js.Consumer(ctx, forgeStream, cred.Node+"_"+cred.Module) if err != nil { return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, forgeStream, err) @@ -313,5 +319,48 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err case err := <-sourceErrs: return err + case err := <-toolErrs: + return err } } + +// watchToolchains reads the toolchains each build module reports (mesh-build-go_toolchains, mesh-build-node_toolchains: the +// module's state toolchains, delivery.ToolchainsBucket; novox/hq ADR 0465 clause 2, issue 1444) and follows them: every +// change of a language's agreed toolchain writes the modules of that language again (index.Indexer.Recompute). A bucket +// that does not exist yet (its build module is not delivered) is said, and not waited for (that needs a timer, ADR 0437 +// §8): the indexer is started again with the module, as with the gate's source. The returned channel gets the error of a +// watch that ended. +func watchToolchains(ctx context.Context, js jetstream.JetStream, ix *index.Indexer, logger *log.Logger) (<-chan error, error) { + errs := make(chan error, len(index.ToolchainModules)) + watched := 0 + for _, module := range slices.Sorted(maps.Values(index.ToolchainModules)) { + name := delivery.ToolchainsBucket(module) + kv, err := js.KeyValue(ctx, name) + if errors.Is(err, jetstream.ErrBucketNotFound) { + logger.Printf("the bucket %s does not exist yet: no toolchain of %s is read until the indexer starts again", name, module) + continue + } + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", name, err) + } + w, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { ix.PutToolchain(module, e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + return nil, err + } + watched++ + go func() { + errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) { + if err := ix.ApplyToolchain(ctx, module, e.Key(), e.Value(), !opOf(e)); err != nil { + logger.Printf("%s.%s could not be taken into the modules: %v", name, e.Key(), err) + } + }) + }() + } + // What the buckets held when the indexer started moves the modules once, not once per key. + if watched > 0 { + if err := ix.Recompute(ctx); err != nil { + logger.Printf("the modules could not be written again from the toolchains: %v", err) + } + } + return errs, nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/inputs.go b/modules/mesh-delivery-indexer/internal/index/inputs.go index 52f189b..250b78b 100644 --- a/modules/mesh-delivery-indexer/internal/index/inputs.go +++ b/modules/mesh-delivery-indexer/internal/index/inputs.go @@ -2,6 +2,8 @@ package index import ( "fmt" + "maps" + "slices" "sort" "strings" @@ -46,9 +48,12 @@ type baseFP func(module string) (fp, none string) // // The rules are the controller's builder (mesh-controller internal/builder/builder.go): a bundle or a package of a // language with a toolchain (go, typescript) is the artifact that has one; a package, and a bundle of those languages, -// resolves packages when it is built (wantsPackages, :1017), so it has an entry in Locks, with the hash of its -// language's lock file or an empty hash when none was carried (the fingerprint is then none, "no lock"); an archive, -// an image and an upstream resolve nothing and have no toolchain. A base is a build.on entry that names a module. +// resolves packages when it is built (wantsPackages, :1017) only when its language's lock file exists: it then has +// an entry in Locks with that file's hash, and an artifact whose lock file the merge did not carry has no entry (a +// dependency-free or vendored bundle resolves nothing, as mesh-sdk SourceInputs2 says of an artifact not in Locks; +// novox/hq ADR 0469 clause 5, issue 1459). A bundle that needs a lock and lacks one is refused where it is built, not +// here, so a fingerprint is never given for a build that could not run. An archive, an image and an upstream resolve +// nothing and have no toolchain. A base is a build.on entry that names a module. func inputsOf(module, raw, tree string, locks map[string]string, base baseFP) moduleInputs { m, err := catalogue.ParseManifest([]byte(raw)) if err != nil { @@ -61,37 +66,29 @@ func inputsOf(module, raw, tree string, locks map[string]string, base baseFP) mo return moduleInputs{None2: "the merge names no tree for the module's directory"} } kept := delivery.KeptInputs2{Tree: tree} - langs := map[string]string{} contexts := false var missing []string + langs := languagesOf(m) if m.Build != nil { for _, a := range m.Build.Artifacts { if a.Context != nil { contexts = true } - if a.Kind != catalogue.ArtifactBundle && a.Kind != catalogue.ArtifactPackage { - continue - } - name, ok := lockFile[a.Language] - if !ok { - continue - } - langs[a.Name] = a.Language - if kept.Locks == nil { - kept.Locks = map[string]string{} - } - kept.Locks[a.Name] = locks[name] } - seen := map[string]bool{} - for _, on := range m.Build.On { - if on.Module == "" || seen[on.Module] { - continue + for _, name := range slices.Sorted(maps.Keys(langs)) { + // A lock file the merge did not carry leaves no entry: the artifact resolves nothing (ADR 0469 clause 5). + if h := locks[lockFile[langs[name]]]; h != "" { + if kept.Locks == nil { + kept.Locks = map[string]string{} + } + kept.Locks[name] = h } - seen[on.Module] = true - if fp, none := base(on.Module); fp != "" { + } + for _, on := range basesOf(m) { + if fp, none := base(on); fp != "" { kept.Bases = append(kept.Bases, fp) } else { - missing = append(missing, fmt.Sprintf("the base %s has no fingerprint (%s)", on.Module, none)) + missing = append(missing, fmt.Sprintf("the base %s has no fingerprint (%s)", on, none)) } } } @@ -113,3 +110,38 @@ func inputsOf(module, raw, tree string, locks map[string]string, base baseFP) mo } return out } + +// languagesOf is the language of each artifact of the manifest that has a toolchain (a bundle or a package of go or +// typescript), by artifact name: the artifacts a build module's toolchain enters the fingerprint of. +func languagesOf(m catalogue.Manifest) map[string]string { + langs := map[string]string{} + if m.Build == nil { + return langs + } + for _, a := range m.Build.Artifacts { + if a.Kind != catalogue.ArtifactBundle && a.Kind != catalogue.ArtifactPackage { + continue + } + if _, ok := lockFile[a.Language]; ok { + langs[a.Name] = a.Language + } + } + return langs +} + +// basesOf is the modules the manifest names as bases (its build.on entries that name a module), each once, sorted. +func basesOf(m catalogue.Manifest) []string { + if m.Build == nil { + return nil + } + seen := map[string]bool{} + var out []string + for _, on := range m.Build.On { + if on.Module != "" && !seen[on.Module] { + seen[on.Module] = true + out = append(out, on.Module) + } + } + sort.Strings(out) + return out +} diff --git a/modules/mesh-delivery-indexer/internal/index/inputs_test.go b/modules/mesh-delivery-indexer/internal/index/inputs_test.go index faee51b..b395d1e 100644 --- a/modules/mesh-delivery-indexer/internal/index/inputs_test.go +++ b/modules/mesh-delivery-indexer/internal/index/inputs_test.go @@ -43,25 +43,40 @@ func TestAGoModuleKeepsItsTreeAndTheHashOfItsLockFile(t *testing.T) { // 1450/AC } } -func TestAnArtifactThatResolvesPackagesAndHasNoLockFileHasAnEmptyHash(t *testing.T) { // 1450/AC2 +func TestAnArtifactWhoseLockFileTheMergeDidNotCarryHasNoLocksEntry(t *testing.T) { // 1459/AC1, AC3 + // A Go bundle without a go.sum (a dependency-free or vendored module): no entry, so a fingerprint is possible (ADR 0469 clause 5). in := inputsOf("avahi", catalogueManifest(t, "avahi"), "avahitree", nil, noBases) - if in.Kept == nil || !reflect.DeepEqual(in.Kept.Locks, map[string]string{"tools": ""}) { - t.Fatalf("kept = %+v, want Locks {tools: \"\"}", in.Kept) + if in.Kept == nil || len(in.Kept.Locks) != 0 { + t.Fatalf("kept = %+v, want no Locks entry", in.Kept) } - // The fingerprint of such inputs is none, "no lock" (mesh-sdk SourceFingerprint2). - if fp, none := delivery.SourceFingerprint2(in.Kept.With("avahi", nil)); fp != "" || none != "no lock" { - t.Errorf("fingerprint = %q none %q, want none 'no lock'", fp, none) + if !reflect.DeepEqual(in.Languages, map[string]string{"tools": "go"}) { + t.Errorf("languages = %v, want tools: go (the toolchain still applies)", in.Languages) } - // A TypeScript bundle's lock file is package-lock.json, not go.sum. - ts := inputsOf("audit-logger", catalogueManifest(t, "audit-logger"), "t", map[string]string{"go.sum": "wrong", "package-lock.json": avahiGoSum}, noBases) - if ts.Kept == nil || !reflect.DeepEqual(ts.Kept.Locks, map[string]string{"code": avahiGoSum}) || !reflect.DeepEqual(ts.Languages, map[string]string{"code": "typescript"}) { + // With the toolchain known, mesh-sdk gives such inputs a src2: value, never 'no lock'. + if fp, none := delivery.SourceFingerprint2(in.Kept.With("avahi", map[string]delivery.Toolchain2{"tools": {Language: "go", Version: "1.25.0", Recipe: "1"}})); fp == "" || none != "" { + t.Errorf("fingerprint = %q none %q, want a value", fp, none) + } + // A TypeScript bundle without package-lock.json (go.sum of another language does not count). + ts := inputsOf("audit-logger", catalogueManifest(t, "audit-logger"), "t", map[string]string{"go.sum": "wrong"}, noBases) + if ts.Kept == nil || len(ts.Kept.Locks) != 0 || !reflect.DeepEqual(ts.Languages, map[string]string{"code": "typescript"}) { t.Errorf("typescript: kept %+v languages %v", ts.Kept, ts.Languages) } - // netcheck has a Go and a TypeScript bundle: each takes its own language's lock file. +} + +func TestAnArtifactWhoseLockFileTheMergeCarriedHasItsHash(t *testing.T) { // 1459/AC2, AC3 + ts := inputsOf("audit-logger", catalogueManifest(t, "audit-logger"), "t", map[string]string{"go.sum": "wrong", "package-lock.json": avahiGoSum}, noBases) + if ts.Kept == nil || !reflect.DeepEqual(ts.Kept.Locks, map[string]string{"code": avahiGoSum}) { + t.Errorf("typescript: kept %+v", ts.Kept) + } + // netcheck has a Go and a TypeScript bundle: each takes its own language's lock file, and one without it has none. both := inputsOf("netcheck", catalogueManifest(t, "netcheck"), "t", map[string]string{"go.sum": "g", "package-lock.json": "p"}, noBases) if both.Kept == nil || !reflect.DeepEqual(both.Kept.Locks, map[string]string{"tools-go": "g", "tools-typescript": "p"}) { t.Errorf("netcheck: kept %+v", both.Kept) } + goOnly := inputsOf("netcheck", catalogueManifest(t, "netcheck"), "t", map[string]string{"go.sum": "g"}, noBases) + if goOnly.Kept == nil || !reflect.DeepEqual(goOnly.Kept.Locks, map[string]string{"tools-go": "g"}) { + t.Errorf("netcheck without package-lock.json: kept %+v", goOnly.Kept) + } } // aModuleWithABase is avahi's manifest (mesh-catalog a67f0196) with the one build.on entry that names a module as the diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 84d4dfc..24a52cc 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -71,7 +71,7 @@ func sorted(s []string) []string { return s } -func TestManifestDeclarations(t *testing.T) { // 1238/AC1 +func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build modules' toolchains) m, top := load(t) if m.Module != "mesh-delivery-indexer" { t.Errorf("module = %q", m.Module) @@ -79,7 +79,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1 if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } - if want := []string{"mesh-controller.builds", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 814948c..3492dc2 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -11,6 +11,8 @@ { "name": "modules", "history": 20 } ], "reads": [ + "mesh-build-go.toolchains", + "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-gate.source" ], From a5da6079de01f4ff5cf94a054128d4585221fd48 Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 21:02:44 +0000 Subject: [PATCH 3/3] Write each module once per merge and watch a build module's toolchains bucket on its toolchain.reported event, with no timer (issue 1444, review round 1) --- .../mesh-delivery-indexer/integration_test.go | 27 +++- .../cmd/mesh-delivery-indexer/main.go | 66 +++----- .../cmd/mesh-delivery-indexer/toolchains.go | 115 ++++++++++++++ .../mesh-delivery-indexer/toolchains_test.go | 141 ++++++++++++++++++ .../internal/index/modules.go | 18 +-- .../internal/index/toolchains.go | 87 +++++++++-- .../internal/index/toolchains_test.go | 62 ++++++-- .../mesh-delivery-indexer/manifest_test.go | 2 +- modules/mesh-delivery-indexer/module.json | 2 + 9 files changed, 428 insertions(+), 92 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 3962c90..c6aadb7 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -137,6 +137,15 @@ func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue), b if _, err := js.CreateOrUpdateConsumer(ctx, "FORGE", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } + // The build modules' streams for toolchain.reported (declared by them, heard by the indexer) and its durable consumers. + for module, stream := range map[string]string{"mesh-build-go": "TOOLCHAINS_GO", "mesh-build-node": "TOOLCHAINS_NODE"} { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.stream." + module + ".event.toolchain.reported"}}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + } candidates, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.CandidatesBucket, History: delivery.CandidatesHistory}) if err != nil { t.Fatal(err) @@ -271,7 +280,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4 // TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain: the bucket mesh-build-go_toolchains exists (the state // toolchains of the build module, delivery.ToolchainsBucket) and holds no key; a merged message of mesh-catalog carrying -// avahi's real manifest makes modules.avahi with none2 'toolchain not yet reported'; a put of the key n1.go (the build +// avahi's real manifest makes modules.avahi with none2 'toolchain not yet reported'; the bucket made after the indexer started, a put of the key n1.go and the build module's event toolchain.reported (the build // module's own layout, delivery.ToolchainKey and EncodeToolchain) makes, within 5 s and with no merge, none2 "builder's // fingerprint not yet known". func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *testing.T) { // 1444/AC10 @@ -289,10 +298,6 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes if _, err := builds.Put(ctx, delivery.NewestKey("avahi"), []byte("build-1")); err != nil { t.Fatal(err) } - }, func(ctx context.Context, js jetstream.JetStream) { - if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1}); err != nil { - t.Fatal(err) - } }) data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0", @@ -321,6 +326,12 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes if got := none2(delivery.ToolchainNoReport); got != delivery.ToolchainNoReport { t.Fatalf("none2 = %q within 5 s, want %q", got, delivery.ToolchainNoReport) } + // The bucket is made only now, as when the build module is assigned after the indexer started (review finding 9): the + // indexer finds it on its own. + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1}) + if err != nil { + t.Fatal(err) + } key, err := delivery.ToolchainKey("n1", "go") if err != nil { t.Fatal(err) @@ -329,11 +340,11 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes if err != nil { t.Fatal(err) } - kv, err := js.KeyValue(ctx, delivery.ToolchainsBucket("mesh-build-go")) - if err != nil { + if _, err := kv.Put(ctx, key, value); err != nil { t.Fatal(err) } - if _, err := kv.Put(ctx, key, value); err != nil { + // The build module says it wrote the key (its event on its stream): the indexer then watches the bucket it did not know. + if _, err := js.Publish(ctx, "mesh.stream.mesh-build-go.event.toolchain.reported", []byte(`{"node":"n1","language":"go"}`)); err != nil { t.Fatal(err) } if got := none2("builder's fingerprint not yet known"); got != "builder's fingerprint not yet known" { 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 4abb8e3..8a5502c 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -15,10 +15,8 @@ import ( "fmt" "io" "log" - "maps" "os" "os/signal" - "slices" "strings" "syscall" @@ -298,10 +296,27 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } go func() { sourceErrs <- follow(ctx, sw, each) }() } - toolErrs, err := watchToolchains(ctx, js, ix, logger) - if err != nil { + tools := newToolchainWatch(ctx, js, toolchainSink{put: ix.PutToolchain, apply: ix.ApplyToolchain, recompute: ix.Recompute}, logger) + if err := tools.start(); err != nil { return err } + // A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are + // made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again + // with the build module, as with the gate's source. + for _, module := range sortedModules() { + stream := index.ToolchainStreams[module] + tcons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module) + if err != nil { + logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the indexer starts again: %v", cred.Node, cred.Module, stream, module, err) + continue + } + tc, err := tcons.Consume(func(m jetstream.Msg) { tools.Reported(m) }, + jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + if err != nil { + return err + } + defer tc.Stop() + } cons, err := js.Consumer(ctx, forgeStream, cred.Node+"_"+cred.Module) if err != nil { return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, forgeStream, err) @@ -319,48 +334,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err case err := <-sourceErrs: return err - case err := <-toolErrs: + case err := <-tools.errs: return err } } - -// watchToolchains reads the toolchains each build module reports (mesh-build-go_toolchains, mesh-build-node_toolchains: the -// module's state toolchains, delivery.ToolchainsBucket; novox/hq ADR 0465 clause 2, issue 1444) and follows them: every -// change of a language's agreed toolchain writes the modules of that language again (index.Indexer.Recompute). A bucket -// that does not exist yet (its build module is not delivered) is said, and not waited for (that needs a timer, ADR 0437 -// §8): the indexer is started again with the module, as with the gate's source. The returned channel gets the error of a -// watch that ended. -func watchToolchains(ctx context.Context, js jetstream.JetStream, ix *index.Indexer, logger *log.Logger) (<-chan error, error) { - errs := make(chan error, len(index.ToolchainModules)) - watched := 0 - for _, module := range slices.Sorted(maps.Values(index.ToolchainModules)) { - name := delivery.ToolchainsBucket(module) - kv, err := js.KeyValue(ctx, name) - if errors.Is(err, jetstream.ErrBucketNotFound) { - logger.Printf("the bucket %s does not exist yet: no toolchain of %s is read until the indexer starts again", name, module) - continue - } - if err != nil { - return nil, fmt.Errorf("the bucket %s: %w", name, err) - } - w, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { ix.PutToolchain(module, e.Key(), e.Value(), !opOf(e)) }) - if err != nil { - return nil, err - } - watched++ - go func() { - errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) { - if err := ix.ApplyToolchain(ctx, module, e.Key(), e.Value(), !opOf(e)); err != nil { - logger.Printf("%s.%s could not be taken into the modules: %v", name, e.Key(), err) - } - }) - }() - } - // What the buckets held when the indexer started moves the modules once, not once per key. - if watched > 0 { - if err := ix.Recompute(ctx); err != nil { - logger.Printf("the modules could not be written again from the toolchains: %v", err) - } - } - return errs, nil -} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go new file mode 100644 index 0000000..30892ae --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go @@ -0,0 +1,115 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "maps" + "slices" + "strings" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-indexer/internal/index" +) + +// toolchainSink is where the toolchains the build modules report go: the indexer's own methods (index.Indexer.PutToolchain, +// ApplyToolchain, Recompute). +type toolchainSink struct { + put func(module, key string, value []byte, deleted bool) bool + apply func(ctx context.Context, module, key string, value []byte, deleted bool) error + recompute func(ctx context.Context) error +} + +// bucketOpener is the part of the bus the toolchain watch uses. +type bucketOpener interface { + KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) +} + +// toolchainWatch reads the toolchains each build module reports (mesh-build-go_toolchains, mesh-build-node_toolchains: the +// module's state toolchains, delivery.ToolchainsBucket; novox/hq ADR 0465 clause 2, issue 1444) and follows them: every +// change of a language's agreed toolchain writes the modules of that language again (index.Indexer.Recompute). A bucket +// is made when its build module is assigned, after the indexer started, and the indexer holds no timer (ADR 0437 §8): +// it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key +// (toolchain.reported, handled by Reported); a bucket not watched yet is then watched, settled as the others. +type toolchainWatch struct { + ctx context.Context + js bucketOpener + sink toolchainSink + logger *log.Logger + errs chan error // the error of a watch that ended + mu sync.Mutex + 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{}} +} + +// 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() { + if err := w.ensure(module); err != nil { + return err + } + } + return nil +} + +// ensure watches the module's toolchains bucket if it is not watched and exists (settling what it holds), then writes the +// modules again from what the watches hold. +func (w *toolchainWatch) ensure(module string) error { + w.mu.Lock() + defer w.mu.Unlock() + name := delivery.ToolchainsBucket(module) + if !w.watched[module] { + kv, err := w.js.KeyValue(w.ctx, name) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module) + return nil + case err != nil: + return fmt.Errorf("the bucket %s: %w", name, err) + } + watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.sink.put(module, e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + return err + } + w.watched[module] = true + go func() { + w.errs <- 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) + } + }) + }() + } + return w.sink.recompute(w.ctx) +} + +// Reported takes one toolchain.reported event of a build module (the message on its stream) in: the module is read off the +// subject, mesh.stream..event.toolchain.reported, its bucket is watched if it is not, and the modules are written +// 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 + } + if !known || !strings.HasPrefix(m.Subject(), "mesh.stream.") { + w.logger.Printf("%s: not a build module's event, dropped", m.Subject()) + _ = m.Term() + return + } + if err := w.ensure(module); err != nil { + w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err) + return + } + _ = m.Ack() +} + +func sortedModules() []string { return slices.Sorted(maps.Values(index.ToolchainModules)) } 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 new file mode 100644 index 0000000..e26bf76 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go @@ -0,0 +1,141 @@ +package main + +import ( + "bytes" + "context" + "log" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// A build module's bucket is made when the module is assigned, after the indexer started (issue 1444, review finding 9). +// The fakes implement only what the watch calls: the bus's KeyValue by bucket name, and a bucket's WatchAll. + +type laterBuckets struct { + mu sync.Mutex + appears map[string]jetstream.KeyValue // buckets that exist now + asked map[string]int +} + +func (l *laterBuckets) KeyValue(_ context.Context, name string) (jetstream.KeyValue, error) { + l.mu.Lock() + defer l.mu.Unlock() + l.asked[name]++ + if kv, ok := l.appears[name]; ok { + return kv, nil + } + return nil, jetstream.ErrBucketNotFound +} + +func (l *laterBuckets) make(name string, kv jetstream.KeyValue) { + l.mu.Lock() + defer l.mu.Unlock() + l.appears[name] = kv +} + +type fakeBucket struct { + jetstream.KeyValue + held []jetstream.KeyValueEntry + w *fakeWatcher +} + +func (b *fakeBucket) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + ch := make(chan jetstream.KeyValueEntry, 8) + for _, e := range b.held { + ch <- e + } + ch <- nil // the mark of the last value held + b.w = &fakeWatcher{ch: ch} + return b.w, nil +} + +type fakeWatcher struct { + jetstream.KeyWatcher + ch chan jetstream.KeyValueEntry +} + +func (w *fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.ch } +func (w *fakeWatcher) Stop() error { return nil } + +type fakeEntry struct { + jetstream.KeyValueEntry + key string + value []byte +} + +func (e fakeEntry) Key() string { return e.key } +func (e fakeEntry) Value() []byte { return e.value } +func (e fakeEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut } + +func TestABuildModulesBucketThatAppearsLaterIsWatchedWhenItsEventArrives(t *testing.T) { // 1444 (review finding 9) + ctx, stop := context.WithCancel(context.Background()) + defer stop() + value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: "go1.27.1", Recipe: "1", At: time.Now()}) + if err != nil { + t.Fatal(err) + } + buckets := &laterBuckets{appears: map[string]jetstream.KeyValue{}, asked: map[string]int{}} + var mu sync.Mutex + var puts, applies []string + recomputes := 0 + sink := toolchainSink{ + put: func(module, key string, _ []byte, _ bool) bool { + mu.Lock() + puts = append(puts, module+" "+key) + mu.Unlock() + return true + }, + apply: func(_ context.Context, module, key string, _ []byte, _ bool) error { + mu.Lock() + applies = append(applies, module+" "+key) + mu.Unlock() + return nil + }, + recompute: func(context.Context) error { mu.Lock(); recomputes++; mu.Unlock(); return nil }, + } + var logs bytes.Buffer + w := newToolchainWatch(ctx, buckets, sink, log.New(&logs, "", 0)) + if err := w.start(); err != nil { // neither bucket exists: nothing is watched, nothing is written, nobody waits + t.Fatal(err) + } + if len(puts) != 0 || recomputes != 0 || !strings.Contains(logs.String(), "does not exist yet") { + t.Fatalf("puts %v recomputes %d log %q after a start with no bucket", puts, recomputes, logs.String()) + } + // The bucket of mesh-build-go is made and already holds a key, and the module says it reported. + bucket := &fakeBucket{held: []jetstream.KeyValueEntry{fakeEntry{key: "novox.go", value: value}}} + buckets.make(delivery.ToolchainsBucket("mesh-build-go"), bucket) + msg := &fakeMsg{subject: "mesh.stream.mesh-build-go.event.toolchain.reported"} + w.Reported(msg) + if msg.acked != 1 || len(puts) != 1 || puts[0] != "mesh-build-go novox.go" || recomputes != 1 { + t.Fatalf("acked %d puts %v recomputes %d, want the settled key, one recompute and an ack", msg.acked, puts, recomputes) + } + // It is followed from now on: a key put later is applied, and a second report only recomputes. + bucket.w.ch <- fakeEntry{key: "ace.go", value: value} + for deadline := time.Now().Add(3 * time.Second); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) { + mu.Lock() + n := len(applies) + mu.Unlock() + if n == 1 { + break + } + } + if len(applies) != 1 || applies[0] != "mesh-build-go ace.go" { + t.Errorf("applies = %v, want the key put later", applies) + } + w.Reported(&fakeMsg{subject: "mesh.stream.mesh-build-go.event.toolchain.reported"}) + if len(puts) != 1 || recomputes != 2 || buckets.asked[delivery.ToolchainsBucket("mesh-build-go")] != 2 { + t.Errorf("puts %v recomputes %d asked %v, want the bucket opened again only at the first report", puts, recomputes, buckets.asked) + } + // An event of no build module is terminated. + other := &fakeMsg{subject: "mesh.stream.avahi.event.toolchain.reported"} + w.Reported(other) + 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) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/modules.go b/modules/mesh-delivery-indexer/internal/index/modules.go index 1538192..6869bd4 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules.go +++ b/modules/mesh-delivery-indexer/internal/index/modules.go @@ -2,8 +2,6 @@ package index import ( "context" - "fmt" - "reflect" "sort" "strings" @@ -57,6 +55,7 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error { if err != nil { return err } + overlay := map[string]delivery.Module{} for _, k := range ix.Builds.Modules(f.Repository) { key, err := delivery.ModulesKey(k.Module) if err != nil { @@ -77,21 +76,10 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error { in := inputsOf(k.Module, raw, f.Trees[k.Path], f.Locks[k.Path], ix.baseFP(ctx)) next.Inputs2, next.None2 = in.Kept, in.None2 } - err = write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) { - if found { - if cur, err := delivery.DecodeModule(value); err == nil && reflect.DeepEqual(cur, withV(next)) { - return nil, false, nil - } - } - data, err := delivery.EncodeModule(next) - return data, false, err - }) - if err != nil { - return fmt.Errorf("modules.%s: %w", key, err) - } + overlay[key] = next } // What this merge moved reaches the modules standing on it (a base's tree, build-agent's fingerprint), in the graph's order. - return ix.Recompute(ctx) + return ix.recomputeWith(ctx, overlay) } // baseFP answers the new-form fingerprint of a base module as modules. holds it, or why it has none. diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go index fac1331..fd5ce2c 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "maps" + "reflect" "slices" "strings" @@ -30,6 +31,11 @@ const ( // 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 ., delivery.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 @@ -120,31 +126,46 @@ type known struct { // the order the bucket lists them in; modules that stand on each other have a cycle as their reason. A module is written // only when its fingerprint or its reason changes, and only if its manifest is still the one computed from. A module with // no kept inputs (a message that carried no manifest) is left as it is. -func (ix *Indexer) Recompute(ctx context.Context) error { - if ix.Modules == nil || ix.Manifests == nil { +func (ix *Indexer) Recompute(ctx context.Context) error { return ix.recomputeWith(ctx, nil) } + +// recomputeWith is Recompute for a merge as well: the records the merge gives (overlay, by module key) stand in place +// of what the bucket holds, are computed with the others, and are written once, with their final fingerprint or reason +// (issue 1444: a merge writes each module once). A record of the overlay that has no kept inputs is written as it is. +func (ix *Indexer) recomputeWith(ctx context.Context, overlay map[string]delivery.Module) error { + if ix.Modules == nil { return nil } + if ix.Manifests == nil { + return ix.writeOverlay(ctx, overlay, nil) + } ix.recompute.Lock() defer ix.recompute.Unlock() keys, err := ix.Modules.Keys(ctx) if err != nil { return fmt.Errorf("listing the modules: %w", err) } + for key := range overlay { + if !slices.Contains(keys, key) { + keys = append(keys, key) + } + } slices.Sort(keys) held := map[string]delivery.Module{} all := map[string]*known{} for _, key := range keys { - value, _, found, err := ix.Modules.Get(ctx, key) - if err != nil { - return fmt.Errorf("modules.%s: %w", key, err) - } - if !found { - continue - } - cur, err := delivery.DecodeModule(value) - if err != nil { - ix.logf("modules.%s holds what is not a module, left: %v", key, err) - continue + cur, ok := overlay[key] + if !ok { + value, _, found, err := ix.Modules.Get(ctx, key) + if err != nil { + return fmt.Errorf("modules.%s: %w", key, err) + } + if !found { + continue + } + if cur, err = delivery.DecodeModule(value); err != nil { + ix.logf("modules.%s holds what is not a module, left: %v", key, err) + continue + } } held[key] = cur if cur.Manifest == "" || cur.Inputs2 == nil { @@ -242,9 +263,25 @@ func (ix *Indexer) Recompute(ctx context.Context) error { k.cur.Inputs2 = &kept return r } + finals := map[string]delivery.Module{} for _, key := range slices.Sorted(maps.Keys(all)) { k := all[key] r := of(key, nil) + if _, merged := overlay[key]; merged { + final := overlay[key] + final.FP2, final.None2, final.Inputs2 = r.fp, r.none, k.cur.Inputs2 + // What the bases and the builder gave is kept as it was while the value it gave does not change, so a merge + // that moves nothing for the module writes nothing (as Recompute does). + if value, _, found, err := ix.Modules.Get(ctx, key); err == nil && found { + if old, err := delivery.DecodeModule(value); err == nil && old.Inputs2 != nil && old.FP2 == r.fp && old.None2 == r.none { + in := *final.Inputs2 + in.Bases, in.Builder = old.Inputs2.Bases, old.Inputs2.Builder + final.Inputs2 = &in + } + } + finals[key] = final + continue + } if k.cur.FP2 == r.fp && k.cur.None2 == r.none { continue } @@ -269,6 +306,30 @@ func (ix *Indexer) Recompute(ctx context.Context) error { return fmt.Errorf("modules.%s: %w", key, err) } } + return ix.writeOverlay(ctx, overlay, finals) +} + +// writeOverlay writes the records of a merge, each once and only if it differs from what the bucket holds; finals has +// the ones that were computed, the others are written as the merge gave them. +func (ix *Indexer) writeOverlay(ctx context.Context, overlay, finals map[string]delivery.Module) error { + for _, key := range slices.Sorted(maps.Keys(overlay)) { + next := overlay[key] + if f, ok := finals[key]; ok { + next = f + } + err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) { + if found { + if cur, err := delivery.DecodeModule(value); err == nil && reflect.DeepEqual(cur, withV(next)) { + return nil, false, nil + } + } + data, err := delivery.EncodeModule(next) + return data, false, err + }) + if err != nil { + return fmt.Errorf("modules.%s: %w", key, err) + } + } return nil } diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go index 32de130..d9fd7f4 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -42,6 +42,20 @@ type mod struct { // merged applies one merge of the catalogue carrying these modules (every one a Go bundle's go.sum, a TypeScript bundle's // package-lock.json) and returns the indexer. func newFormIndexer(t *testing.T, mods ...mod) (*Indexer, *memStore) { + t.Helper() + ix, store, f := newFormIndexerUnmerged(t, mods...) + if err := ix.Apply(context.Background(), f); err != nil { + t.Fatal(err) + } + // build-agent's value of today's form, from a merge of the controller (modules.build-agent holds its fingerprint). + if err := ix.Apply(context.Background(), mergedMsg(buildAgentRepo, map[string]string{"modules/build-agent": "agent-tree-1"})); err != nil { + t.Fatal(err) + } + return ix, store +} + +// newFormIndexerUnmerged is the indexer and the catalogue merge of these modules, not yet applied. +func newFormIndexerUnmerged(t *testing.T, mods ...mod) (*Indexer, *memStore, delivery.Forge) { t.Helper() var recs [][2]any trees, manifests := map[string]string{}, map[string]string{} @@ -54,15 +68,7 @@ func newFormIndexer(t *testing.T, mods ...mod) (*Indexer, *memStore) { } recs = append(recs, rec(t, "build-agent-1", "build-agent", buildAgentRepo, "modules/build-agent", "main", "", "2026-10-19T09:00:00Z", aInputs)) ix, store, _ := manifestsIndexer(t, recs...) - f := withManifests(mergedMsg(catalogueRepo, trees), manifests, locks) - if err := ix.Apply(context.Background(), f); err != nil { - t.Fatal(err) - } - // build-agent's value of today's form, from a merge of the controller (modules.build-agent holds its fingerprint). - if err := ix.Apply(context.Background(), mergedMsg(buildAgentRepo, map[string]string{"modules/build-agent": "agent-tree-1"})); err != nil { - t.Fatal(err) - } - return ix, store + return ix, store, withManifests(mergedMsg(catalogueRepo, trees), manifests, locks) } // toolchainValue is a toolchains value as the build module's writer encodes it (mesh-sdk go/delivery EncodeToolchain). @@ -194,3 +200,41 @@ func TestModulesThatStandOnEachOtherAreNamedAsACycle(t *testing.T) { // 1444 (th } } } + +func TestAMergeWritesEachModuleOnceWithItsFinalValueAndAReplayWritesNothing(t *testing.T) { // 1444 (review finding 8) + ctx := context.Background() + ix, mods, f := newFormIndexerUnmerged(t, mod{"avahi", catalogueManifest(t, "avahi")}) + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") // agreed before the merge is heard + if err := ix.Apply(ctx, f); err != nil { + t.Fatal(err) + } + m, _ := moduleOf(t, mods, "avahi") + if mods.writes != 1 || m.None2 != None2Builder { + t.Errorf("writes = %d none2 = %q after one merge, want one write with %q", mods.writes, m.None2, None2Builder) + } + if err := ix.Apply(ctx, f); err != nil { + t.Fatal(err) + } + if mods.writes != 1 { + t.Errorf("writes = %d after the same merge again, want still 1", mods.writes) + } +} + +func TestAMergeKeepsAFingerprintTheModuleStillHas(t *testing.T) { // 1444 (review finding 8: a merge never blanks FP2) + ctx := context.Background() + ix, mods := newFormIndexer(t, builderModules(t)...) + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + before, _ := moduleOf(t, mods, "mesh-tools") + if before.FP2 == "" { + t.Fatalf("mesh-tools has no fp2: %+v", before) + } + w := mods.writes + // the same catalogue merged again at a later commit: nothing moved + _, _, f := newFormIndexerUnmerged(t, builderModules(t)...) + if err := ix.Apply(ctx, f); err != nil { + t.Fatal(err) + } + if after, _ := moduleOf(t, mods, "mesh-tools"); after.FP2 != before.FP2 || mods.writes != w { + t.Errorf("fp2 = %q (was %q), writes %d more", after.FP2, before.FP2, mods.writes-w) + } +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 24a52cc..90d96c3 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -84,7 +84,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control // internal/broker/state.go:62-93). - if want := []string{"mesh-delivery-forge-bridge.forge.**"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-forge-bridge.forge.**"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 3492dc2..fcfe983 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -3,6 +3,8 @@ "version": "1", "slug": "indexer", "consumes": [ + "mesh-build-go.toolchain.reported", + "mesh-build-node.toolchain.reported", "mesh-delivery-forge-bridge.forge.**" ], "state": [