From 06b58bc83bc6b305cb089c7ed806a7286f3cc97c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 23:48:14 +0100 Subject: [PATCH] =?UTF-8?q?Indexer:=20take=20directories=20and=20outside-b?= =?UTF-8?q?uilt=20registrations=20from=20the=20builder's=20log,=20watched?= =?UTF-8?q?=20from=20its=20start,=20so=20no=20build=20record=20of=20the=20?= =?UTF-8?q?controller's=20is=20read=20(issue=201955,=20ADR=200571=20=C2=A7?= =?UTF-8?q?1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../mesh-delivery-indexer/integration_test.go | 95 ++++++++----------- .../cmd/mesh-delivery-indexer/main.go | 67 ++++++------- modules/mesh-delivery-indexer/go.mod | 2 +- modules/mesh-delivery-indexer/go.sum | 4 +- .../internal/index/builds.go | 74 +++++++-------- .../internal/index/index_test.go | 16 +--- .../internal/index/live_builds_test.go | 12 +-- .../internal/index/modules_test.go | 55 ++++++----- .../internal/index/registered_test.go | 4 +- .../internal/index/toolchains_test.go | 13 +-- .../mesh-delivery-indexer/manifest_test.go | 2 +- modules/mesh-delivery-indexer/module.json | 2 +- 12 files changed, 154 insertions(+), 192 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 e27b51f..3262321 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 @@ -19,6 +19,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" ) @@ -44,21 +45,10 @@ accounts { // manifest's state (mesh-delivery-indexer_candidates, history 10) and mesh-controller_builds holding build records, and // no mesh-delivery-gate_source bucket. func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10 - js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds jetstream.KeyValue) { - // Build records in the controller's layout (mesh-sdk go/delivery/testdata/build-record-build-agent.json, captured from - // the controller's encoder), under delivery.BuildKey, with the newest pointer beside them. - for _, b := range []struct{ repo, module, path, id string }{ - {"novox/mesh-delivery", "a", "modules/a", "build-1"}, {"novox/mesh-catalog", "c", "modules/c", "build-2"}, - } { - rec, _ := json.Marshal(map[string]any{"id": b.id, "repository": b.repo, "ref": "main", "module": b.module, - "commit": "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", "path": b.path, "at": "2026-10-19T09:38:00Z"}) - if _, err := builds.Put(ctx, delivery.BuildKey(b.module, b.id), rec); err != nil { - t.Fatal(err) - } - if _, err := builds.Put(ctx, delivery.NewestKey(b.module), []byte(b.id)); err != nil { - t.Fatal(err) - } - } + js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds buildLog) { + // The builder's log entries of two builds' ends (ADR 0571 clause 1), as the builder appends them. + builds.end(t, "novox/mesh-delivery", "a", "modules/a") + builds.end(t, "novox/mesh-catalog", "c", "modules/c") }) h1, h2 := "0123456789abcdef0123456789abcdef01234567", "89abcdef0123456789abcdef0123456789abcdef" for _, f := range []delivery.Forge{ @@ -99,7 +89,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), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) { +func startIndexer(t *testing.T, put func(context.Context, buildLog), 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) @@ -162,11 +152,12 @@ func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue), b if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ManifestsBucket, History: delivery.ManifestsHistory}); err != nil { t.Fatal(err) } - builds, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.BuildsBucket, History: 5}) - if err != nil { + // The builder's module log, as the runtime raises it (ADR 0297), filled by put before the indexer starts. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buscompose.LogStreamName(delivery.BuilderModule, delivery.BuilderLog), + Subjects: []string{buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog) + ".>"}, AllowDirect: true}); err != nil { t.Fatal(err) } - put(ctx, builds) + put(ctx, buildLog{ctx: ctx, js: js}) for _, b := range before { b(ctx, js) } @@ -185,15 +176,32 @@ func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue), b return js, ctx, candidates, modules } +// buildLog appends entries to the builder's log as the builder does. +type buildLog struct { + ctx context.Context + js jetstream.JetStream +} + +// end appends the end of a done trunk build of module from repo's directory path. +func (l buildLog) end(t *testing.T, repo, module, path string) { + t.Helper() + raw, err := delivery.EncodeBuildLogEntry(delivery.BuildLogEntry{Event: delivery.BuildEndedEvent, Module: module, FP: "src2:" + module, + Repository: repo, Directory: path, Commit: "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", Node: "n1", + At: time.Date(2026, 10, 19, 9, 38, 0, 0, time.UTC), Outcome: delivery.BuildDone}) + if err != nil { + t.Fatal(err) + } + if _, err := l.js.Publish(l.ctx, buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog)+"."+module, raw); err != nil { + t.Fatal(err) + } +} + // TestIndexerWritesModulesAtAMerge: a module's record and its newest pointer are in mesh-controller_builds; a merged // message on FORGE makes modules. within 5 s, the fingerprint of the merge's tree and the recorded inputs. func TestIndexerWritesModulesAtAMerge(t *testing.T) { // 1239/AC9; I1, I3: the modules are the merged message's, no build record names them (issue 1851) // The bucket holds only build-agent's record, which no fingerprint reads since R4g (ADR 0530 §1). - js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds jetstream.KeyValue) { - rec := `{"id":"build-1","repository":"novox/mesh-controller","ref":"main","module":"build-agent","commit":"7a74cee83f88e6585a5511c3cc53bdf0f5303d20","path":"modules/build-agent","source-fingerprint":"src1:e846efe73716b78e80860baaac96153de2cb76551325753838ca245d2160a3d7","at":"2026-10-19T09:38:00Z"}` - if _, err := builds.Put(ctx, delivery.BuildKey("build-agent", "build-1"), []byte(rec)); err != nil { - t.Fatal(err) - } + js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds buildLog) { + builds.end(t, "novox/mesh-controller", "build-agent", "modules/build-agent") }) const merge = "89abcdef0123456789abcdef0123456789abcdef" data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, @@ -229,15 +237,8 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4 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) - } + js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds buildLog) { + builds.end(t, "novox/mesh-catalog", "avahi", "modules/avahi") }) data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0", @@ -287,15 +288,8 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes 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) - } + js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds buildLog) { + builds.end(t, "novox/mesh-catalog", "avahi", "modules/avahi") }) data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0", @@ -362,16 +356,9 @@ func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBuc t.Fatal(err) } const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381" - const inputs = `{"Module":"avahi","Tree":"oldtree","Bases":null,"Contexts":null,"Toolchains":{"code":"go sha256:ccc 0123456789abcdef"},"Unpinned":""}` const trunk, head = "5555555555555555555555555555555555555555", "0123456789abcdef0123456789abcdef01234567" - js, ctx, candidates, _ := 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) - } + js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds buildLog) { + builds.end(t, "novox/mesh-catalog", "avahi", "modules/avahi") }, func(ctx context.Context, js jetstream.JetStream) { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: PreviewsStream, Subjects: []string{previewWrittenSubject}}); err != nil { t.Fatal(err) @@ -451,7 +438,7 @@ func TestAMergeOfMeshSDKChangingOnlyGoLeavesItsNamedTreeEqual(t *testing.T) { } objects := map[string]string{"src": "1111111111111111111111111111111111111111", "test": "2222222222222222222222222222222222222222", "package.json": "3333333333333333333333333333333333333333", "package-lock.json": "4444444444444444444444444444444444444444", "tsconfig.json": "5555555555555555555555555555555555555555"} - js, ctx, _, modules := startIndexer(t, func(context.Context, jetstream.KeyValue) {}) + js, ctx, _, modules := startIndexer(t, func(context.Context, buildLog) {}) publish := func(merge, root string, objects map[string]string) { data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-sdk", Branch: "feat/x", Number: 70, Head: "0123456789abcdef0123456789abcdef01234567", Merge: merge, Root: root, Trees: map[string]string{"": root}, @@ -518,7 +505,7 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC } const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381" digest := delivery.ManifestDigest(manifest) - _, ctx, _, modules := startIndexer(t, func(context.Context, jetstream.KeyValue) {}, func(ctx context.Context, js jetstream.JetStream) { + _, ctx, _, modules := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) { mkey, err := delivery.ManifestsKey(digest) if err != nil { t.Fatal(err) @@ -605,7 +592,7 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) { t.Fatal(err) } } - js, ctx, _, modules := startIndexer(t, func(context.Context, jetstream.KeyValue) {}, func(ctx context.Context, js jetstream.JetStream) { + js, ctx, _, modules := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) { kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1}) if err != nil { t.Fatal(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 05e6a27..e02d915 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -10,19 +10,20 @@ package main import ( "context" - "encoding/json" "errors" "fmt" "io" "log" "os" "os/signal" + "sync/atomic" "syscall" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "git.novox.be/novox/mesh-sdk/go/process" @@ -277,28 +278,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The control records (ADR 0565 §2), new with this delivery and bound when they appear: until then nothing is written to them. predefined := process.AwaitBucket(ctx, js, conn, index.PredefinedBucket, "no predefined list is written", logger.Printf) configCandidates := process.AwaitBucket(ctx, js, conn, index.ConfigCandidatesBucket, "no pull request's config declarations are written", logger.Printf) - // The controller's build records only name the directory of a module whose key names none (fillPaths): without them - // such a module waits for its path, every other is indexed (issue 1945). - builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf) + // The builder's log (ADR 0571 clause 1), read on the indexer's own credential (the calls connection) through its declared read: every entry it + // keeps at the start, then each new one as the builder appends it (no timer). It names the directory of a module whose key + // names none (fillPaths) and the places a registered module was built from; no build moves a fingerprint (ADR 0530 §1). + // Where the log cannot be read such a module waits for its path, every other is indexed (issue 1945), and it is said. known := index.NewBuilds() - // The controller's build records name the directory of a module whose key names none (fillPaths); no build of any - // module moves a fingerprint since no fingerprint takes the builder (ADR 0530 §1). - each := func(e jetstream.KeyValueEntry) { - if opOf(e) { - known.Put(e.Key(), e.Value()) - } else { - known.Delete(e.Key()) - } - } errs := make(chan error, 1) - var buildsWatch jetstream.KeyWatcher - if builds != nil { - if w, err := settle(ctx, builds, each); err != nil { - process.NotWatched(logger.Printf, delivery.BuildsBucket, err, "a module whose key names no directory is not given one until the indexer starts again") - } else { - buildsWatch = w - } - } // The indexer asks the forge for the objects of a manifest's named paths as its own account (ADR 0499 clause 3). calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) @@ -310,6 +295,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), ReadOpen: openVia(calls.Call)} + // The log's entries at the start are read before any fingerprint is computed or registered module indexed (below); each + // new one indexes the module it names when that module is registered and the bucket holds none (registered.go). + var started atomic.Bool + if err := calls.ReadLog(delivery.BuilderModule, delivery.BuilderLog).Watch("", stdio.LogFromAll, func(le stdio.LogEntry) error { + if err := known.Put(le.Value); err != nil { + logger.Printf("%s entry %d is not read: %v", delivery.BuildLogRead, le.Sequence, err) + return nil + } + if started.Load() { + if err := ix.IndexRegisteredModule(ctx, le.Key); err != nil { + logger.Printf("the registered module %s is not indexed: %v", le.Key, err) + } + } + return nil + }); err != nil { + logger.Printf("%s is not read (%v): a module whose key names no directory is not given one, and no registered module is "+ + "indexed from it, until the indexer starts again", delivery.BuildLogRead, err) + } + started.Store(true) // The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key // main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does // not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate. @@ -360,28 +364,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := ix.RecomputeAtStart(ctx); err != nil { logger.Printf("computing the new-form fingerprints at start: %v", err) } - // A module registered in the catalogue from its own repository gets its manifest indexed when the controller's records - // name it and the bucket holds none (registered.go); the records that follow are heard below. + // A module registered in the catalogue from its own repository gets its manifest indexed when the builder's log names + // it and the bucket holds none (registered.go); the entries that follow are heard above. if err := ix.IndexRegisteredAll(ctx); err != nil { logger.Printf("indexing the registered modules at start: %v", err) } - if buildsWatch != nil { - go func() { - errs <- follow(ctx, buildsWatch, func(e jetstream.KeyValueEntry) { - each(e) - if !opOf(e) { - return - } - var r delivery.BuildRecord - if json.Unmarshal(e.Value(), &r) != nil || r.Module == "" { - return - } - if err := ix.IndexRegisteredModule(ctx, r.Module); err != nil { - logger.Printf("the registered module %s is not indexed: %v", r.Module, err) - } - }) - }() - } // A person takes a registered module in now with indexer_take_in, which writes the takeins state (issue 1945): watched, // as every other input. A module the forge could not read is also read again by itself on the next source revision and // merged message (index.RetryUnread). diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 52694c1..e98c8bd 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index 685613a..07e0ff7 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd h1:WhqMEXFZ9Zc+fEfLsnf1jhhGkc4IqMwUpXXwZq+V3+E= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029212112-2d2aa7ba9bbd/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641 h1:nP/jrjm2Qt1wdmd1Bc+E6c4+PobAzW5o64SPArE+jaw= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/internal/index/builds.go b/modules/mesh-delivery-indexer/internal/index/builds.go index 7c3eacd..02a7a63 100644 --- a/modules/mesh-delivery-indexer/internal/index/builds.go +++ b/modules/mesh-delivery-indexer/internal/index/builds.go @@ -1,72 +1,63 @@ package index import ( - "encoding/json" + "fmt" "maps" "slices" "strings" "sync" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) -// Builds holds the controller's build records (mesh-controller_builds) for the two things the indexer still reads from -// them until release R4g, when build-agent goes (the lead's decision of R4e, 2026-10-22): -// - the Builder input of the builder's own module, which build-agent builds (TrunkOf; novox/hq ADR 0453 clause 5, -// issue 1727); -// - the directory of a module whose modules key has none (PathOf; issue 1614), filled once and changing no fingerprint. +// Builds holds the builder's log (mesh-delivery-builder.build-log, novox/hq ADR 0571 clause 1) for what the indexer reads +// of past builds: the directory of a module whose modules key has none (PathOf; issue 1614), filled once and changing no +// fingerprint, and the places a module registered from its own repository was built from (RegistrationsOf). It is filled +// from the log's entries at the indexer's start (every one the log keeps, 30 days) and then each new one as the builder +// appends it; no entry is ever taken out, since only age takes one from the log. // // It lists no module and maps no directory to a module: the indexer's modules are those its own messages and bucket name // (ADR 0501 clause 6, issue 1851). Safe for use from a watch and from the message handler at once. type Builds struct { mu sync.Mutex - records map[string]delivery.BuildRecord // by bucket key + records map[string]delivery.BuildRecord // by the entry's event, module, fingerprint and time } // NewBuilds is an empty Builds. func NewBuilds() *Builds { return &Builds{records: map[string]delivery.BuildRecord{}} } -// Put takes a key of the builds bucket and its value. Only a build record of a module (key `.build.`, -// delivery.BuildKey) counts; the newest-pointer keys and requests that knew no module are not records of one. -func (b *Builds) Put(key string, value []byte) { - if !strings.Contains(key, ".build.") { - return +// Put takes one entry of the builder's log. A build's end is an outcome: done registers, failed or queued does not; its +// start is a request, which registers nothing either. An entry that does not read, or names no module or repository, is +// refused and said. +func (b *Builds) Put(value []byte) error { + e, err := delivery.DecodeBuildLogEntry(value) + if err != nil { + return err } - var r delivery.BuildRecord - if err := json.Unmarshal(value, &r); err != nil || r.Module == "" || r.Repository == "" { - return + if e.Module == "" || e.Repository == "" { + return fmt.Errorf("a build log entry naming no module or repository: %s", value) + } + r := delivery.BuildRecord{ID: e.At.UTC().Format(time.RFC3339Nano) + "." + e.FP, Repository: e.Repository, Ref: e.Ref, Module: e.Module, + Path: e.Directory, Commit: e.Commit, At: e.At} + switch { + case e.Event != delivery.BuildEndedEvent: + r.State = e.Event + case e.Outcome == delivery.BuildFailed: + r.Failed = "the build failed" + case e.Outcome != delivery.BuildDone: + r.State = e.Outcome } b.mu.Lock() defer b.mu.Unlock() - b.records[key] = r -} - -// Delete takes a key out. -func (b *Builds) Delete(key string) { - b.mu.Lock() - defer b.mu.Unlock() - delete(b.records, key) + b.records[e.Event+"."+e.Module+"."+r.ID] = r + return nil } func newer(a, b delivery.BuildRecord) bool { return a.At.After(b.At) || (a.At.Equal(b.At) && a.ID > b.ID) } -// TrunkOf is the newest successful trunk build record of a module, in whatever repository it was built from; nil when -// there is none (issue 1727). -func (b *Builds) TrunkOf(module string) *delivery.BuildRecord { - var best *delivery.BuildRecord - b.mu.Lock() - defer b.mu.Unlock() - for _, r := range b.records { - if r.Module == module && r.IsTrunkBuild() && (best == nil || newer(r, *best)) { - c := r - best = &c - } - } - return best -} - // PathOf is the directory of a module of the repository as its newest build record of that repository names it, its // newest successful trunk build's first ("." for the repository's root), and false when no record of the repository // names the module (issue 1614). @@ -79,7 +70,7 @@ func (b *Builds) PathOf(module, repository string) (string, bool) { if r.Module != module || !delivery.SameRepository(r.Repository, repository) { continue } - t := r.IsTrunkBuild() + t := registered(r) && trunkLike[r.Ref] if best == nil || (t && !trunk) || (t == trunk && newer(r, *best)) { c := r best, trunk = &c, t @@ -91,15 +82,14 @@ func (b *Builds) PathOf(module, repository string) (string, bool) { return pathOf(strings.Trim(best.Path, "/")), true } -// Registration is a module registered in the catalogue from its own repository, as the controller's newest outcome record +// Registration is a module registered in the catalogue from its own repository, as the builder log.s newest end of a build // of it names it: the repository, the directory of its manifest there ("" or "." for the root) and the commit it was // built at. type Registration struct { Module, Repository, Path, Commit, Ref string } -// registered is whether the record is an outcome that did not fail: what registers a module. Unlike IsTrunkBuild it -// asks for no recorded source inputs, which no build records since ADR 0530 clause 1, and for no particular branch: a +// registered is whether the record is an outcome that did not fail: what registers a module. It asks for no particular branch: a // repository's trunk is named main, master, or what its owner chose (the registered photos, de-spiegel and link2pay were // built from nox-mesh, and every record of them names that ref). func registered(r delivery.BuildRecord) bool { diff --git a/modules/mesh-delivery-indexer/internal/index/index_test.go b/modules/mesh-delivery-indexer/internal/index/index_test.go index 3dd794d..e6f9da4 100644 --- a/modules/mesh-delivery-indexer/internal/index/index_test.go +++ b/modules/mesh-delivery-indexer/internal/index/index_test.go @@ -2,7 +2,6 @@ package index import ( "context" - "encoding/json" "strings" "testing" @@ -85,14 +84,7 @@ func (m *memStore) Delete(_ context.Context, key string, rev uint64) error { // the controller's encoder; copied here, the vendored SDK does not carry testdata) has it, with the repository, module and path changed. func buildRecord(t *testing.T, repo, module, path, at string) []byte { t.Helper() - raw := []byte(`{"id":"build-1792395447088936520","repository":"novox/mesh-controller","ref":"main","module":"build-agent","commit":"7a74cee83f88e6585a5511c3cc53bdf0f5303d20","path":"modules/build-agent","source-fingerprint":"src1:e846efe73716b78e80860baaac96153de2cb76551325753838ca245d2160a3d7","at":"2026-10-19T09:38:00Z"}`) - var m map[string]any - if err := json.Unmarshal(raw, &m); err != nil { - t.Fatal(err) - } - m["repository"], m["module"], m["path"], m["at"] = repo, module, path, at - b, _ := json.Marshal(m) - return b + return logged(t, "build-"+module, module, repo, path, "", "", at) } func pull(repo string, number int, branch, head string, dirs ...string) delivery.Forge { @@ -104,9 +96,9 @@ func newIndexer(t *testing.T) (*Indexer, *memStore, *strings.Builder) { t.Helper() st, logs := newMem(), &strings.Builder{} b := NewBuilds() - b.Put(delivery.BuildKey("a", "build-1"), buildRecord(t, "novox/mesh-delivery", "a", "modules/a", "2026-10-19T09:00:00Z")) - b.Put(delivery.BuildKey("c", "build-2"), buildRecord(t, "novox/mesh-catalog", "c", "modules/c", "2026-10-19T09:00:00Z")) - b.Put(delivery.BuildKey("odd", "build-3"), buildRecord(t, "novox/mesh-delivery", "odd-name", "modules/odd", "2026-10-19T09:00:00Z")) + _ = b.Put(buildRecord(t, "novox/mesh-delivery", "a", "modules/a", "2026-10-19T09:00:00Z")) + _ = b.Put(buildRecord(t, "novox/mesh-catalog", "c", "modules/c", "2026-10-19T09:00:00Z")) + _ = b.Put(buildRecord(t, "novox/mesh-delivery", "odd-name", "modules/odd", "2026-10-19T09:00:00Z")) return &Indexer{Store: st, Builds: b, Logf: func(f string, a ...any) { logs.WriteString(strings.TrimSpace(strings.ReplaceAll(f, "\n", " ")) + "\n") _ = a diff --git a/modules/mesh-delivery-indexer/internal/index/live_builds_test.go b/modules/mesh-delivery-indexer/internal/index/live_builds_test.go index 7aacda2..fe8c43a 100644 --- a/modules/mesh-delivery-indexer/internal/index/live_builds_test.go +++ b/modules/mesh-delivery-indexer/internal/index/live_builds_test.go @@ -2,6 +2,7 @@ package index import ( "testing" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -23,13 +24,10 @@ const ( ) func liveRecord(id, module, repository, ref, commit, path, at, source string) [2]any { - raw := `{"id":"` + id + `","repository":"` + repository + `",` - if ref != "" { - raw += `"ref":"` + ref + `",` - } - raw += `"module":"` + module + `","commit":"` + commit + `","path":"` + path + `","at":"` + at + `","source-fingerprint":"` + source + - `","source-inputs":{"Module":"` + module + `",` + liveInputsTail + `}` - return [2]any{delivery.BuildKey(module, id), []byte(raw)} + when, _ := time.Parse(time.RFC3339, at) + raw, _ := delivery.EncodeBuildLogEntry(delivery.BuildLogEntry{Event: delivery.BuildEndedEvent, Module: module, FP: source, Repository: repository, + Directory: path, Commit: commit, Ref: ref, Node: "n1", At: when, Outcome: delivery.BuildDone}) + return [2]any{module, raw} } func liveBuilds() [][2]any { diff --git a/modules/mesh-delivery-indexer/internal/index/modules_test.go b/modules/mesh-delivery-indexer/internal/index/modules_test.go index e65bc31..d090f08 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules_test.go +++ b/modules/mesh-delivery-indexer/internal/index/modules_test.go @@ -2,12 +2,12 @@ package index import ( "context" - "encoding/json" "errors" "fmt" "path" "strings" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" @@ -15,29 +15,38 @@ import ( const merge = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" -// record is a build record as the controller writes it (mesh-sdk go/delivery/testdata/build-record-build-agent.json is -// the captured shape): SourceInputs under its Go field names, no json keys. inputs is that object's JSON. -func record(t *testing.T, id, module, repo, path, ref, state, at, inputs string) (string, []byte) { +// logged is one entry of the builder's log (mesh-sdk go/delivery BuildLogEntry, as the builder appends it): a build of +// module from repo's directory path at ref ("" for the trunk), its outcome ("" done, "failed", or "asked" for a build only +// started), at the time at; id makes its fingerprint. +func logged(t *testing.T, id, module, repo, path, ref, outcome, at string) []byte { t.Helper() - // an outcome has no state key (mesh-controller internal/chain/records.go:57-59); state asked is a request - raw := `{"id":"` + id + `",` - switch state { - case "": - case "failed": // a failure is an outcome with failed set and no state (mesh-controller internal/chain/records.go:109) - raw += `"failed":"the build failed",` - default: - raw += `"state":"` + state + `",` - } - raw += `"repository":"` + repo + `","ref":"` + ref + `","module":"` + module + `","commit":"7a74cee83f88e6585a5511c3cc53bdf0f5303d20","path":"` + path + `","at":"` + at + `"` - if inputs != "" { - raw += `,"source-fingerprint":"src1:ignored","source-inputs":` + inputs - } - raw += `}` - var check delivery.BuildRecord - if err := json.Unmarshal([]byte(raw), &check); err != nil { + when, err := time.Parse(time.RFC3339, at) + if err != nil { t.Fatal(err) } - return delivery.BuildKey(module, id), []byte(raw) + if path == "" { + path = delivery.RootPath + } + e := delivery.BuildLogEntry{Event: delivery.BuildEndedEvent, Module: module, FP: "src2:" + id, Repository: repo, Directory: path, + Commit: "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", Ref: ref, Node: "n1", At: when, Outcome: delivery.BuildDone} + switch outcome { + case "": + case "failed": + e.Outcome = delivery.BuildFailed + default: + e.Event, e.Outcome = delivery.BuildStartedEvent, "" + } + raw, err := delivery.EncodeBuildLogEntry(e) + if err != nil { + t.Fatal(err) + } + return raw +} + +// record is logged under the module's log key, as the tests' tables carry it. +func record(t *testing.T, id, module, repo, path, ref, state, at, _ string) (string, []byte) { + t.Helper() + return module, logged(t, id, module, repo, path, ref, state, at) } func modulesIndexer(t *testing.T, recs ...[2]any) (*Indexer, *memStore) { @@ -45,7 +54,9 @@ func modulesIndexer(t *testing.T, recs ...[2]any) (*Indexer, *memStore) { mods := newMem() ix := &Indexer{Store: newMem(), Modules: mods, Builds: NewBuilds()} for _, r := range recs { - ix.Builds.Put(r[0].(string), r[1].([]byte)) + if err := ix.Builds.Put(r[1].([]byte)); err != nil { + t.Fatal(err) + } } return ix, mods } diff --git a/modules/mesh-delivery-indexer/internal/index/registered_test.go b/modules/mesh-delivery-indexer/internal/index/registered_test.go index 78101a4..d5751cd 100644 --- a/modules/mesh-delivery-indexer/internal/index/registered_test.go +++ b/modules/mesh-delivery-indexer/internal/index/registered_test.go @@ -302,14 +302,14 @@ func TestARecordOfAPullRequestDoesNotReplaceTheRegisteredRef(t *testing.T) { // rec(t, "b2", "photos", repo, "", "nox-mesh", "", "2026-10-02T09:00:00Z", ""), rec(t, "b3", "photos", repo, "", "feat/x", "", "2026-10-09T09:00:00Z", ""), } { - b.Put(r[0].(string), r[1].([]byte)) + _ = b.Put(r[1].([]byte)) } got := b.RegistrationsOf("photos") if len(got) != 1 || got[0].Ref != "nox-mesh" { t.Fatalf("registrations = %+v, want the ref most records name, not the newer feat/x", got) } k, v := record(t, "b4", "photos", repo, "", "main", "", "2026-10-03T09:00:00Z", "") - b.Put(k, v) // a trunk's ref beats a branch of its own + _, _ = k, b.Put(v) // a trunk's ref beats a branch of its own if got := b.RegistrationsOf("photos"); len(got) != 1 || got[0].Ref != "main" { t.Fatalf("registrations = %+v", got) } diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go index b9d01be..39b890c 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains_test.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains_test.go @@ -46,13 +46,10 @@ const ( ) func agentBuild(id, src, at string) [2]any { - commit := "cada7e3c" - if id == "build-1792587792086548572" { - commit = "f10fbda1" - } - raw := `{"id":"` + id + `","repository":"` + agentCloneURL + `","ref":"main","module":"build-agent","commit":"` + commit + `",` + - `"path":"modules/build-agent","at":"` + at + `","source-fingerprint":"` + src + `","source-inputs":` + aInputs + `}` - return [2]any{delivery.BuildKey("build-agent", id), []byte(raw)} + when, _ := time.Parse(time.RFC3339, at) + raw, _ := delivery.EncodeBuildLogEntry(delivery.BuildLogEntry{Event: delivery.BuildEndedEvent, Module: "build-agent", FP: src, Repository: agentCloneURL, + Directory: "modules/build-agent", Commit: "cada7e3c", Node: "n1", At: when, Outcome: delivery.BuildDone}) + return [2]any{"build-agent", raw} } type mod struct { @@ -184,7 +181,7 @@ func TestBasesAreComputedFirstAndNoFingerprintTakesTheBuilder(t *testing.T) { // t.Errorf("other: fp2 = %q none2 = %q, want none2 naming the base ts-base", m.FP2, m.None2) } a := agentBuild("build-1792587792086548572", agentSrcY, "2026-10-21T15:03:00Z") - ix.Builds.Put(a[0].(string), a[1].([]byte)) + _ = ix.Builds.Put(a[1].([]byte)) if err := ix.Recompute(context.Background()); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 603cb46..161bf20 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -89,7 +89,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } - if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-assigner.placement", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "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 faeb9f8..6b26c33 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -24,8 +24,8 @@ "reads": [ "mesh-build-go.toolchains", "mesh-build-node.toolchains", - "mesh-controller.builds", "mesh-delivery-assigner.placement", + "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source" ],