Indexer: take directories and outside-built registrations from the builder's log, watched from its start, so no build record of the controller's is read (issue 1955, ADR 0571 §1)
This commit is contained in:
@@ -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.<m> 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)
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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 `<module>.build.<id>`,
|
||||
// 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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user