824 lines
35 KiB
Go
824 lines
35 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats-server/v2/server"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
|
|
// controller's user and the builder's own principals <node>.<module> (two instances, for the test of one build).
|
|
const busConf = `
|
|
listen: 127.0.0.1:-1
|
|
jetstream { store_dir: %q }
|
|
accounts {
|
|
MESH {
|
|
jetstream: enabled
|
|
users = [
|
|
{ user: "ctl", password: "pw" }
|
|
{ user: "n1.mesh-delivery-builder", password: "pw" }
|
|
{ user: "n2.mesh-delivery-builder", password: "pw" }
|
|
]
|
|
}
|
|
}
|
|
`
|
|
|
|
const (
|
|
commit = "7a74cee83f88e6585a5511c3cc53bdf0f5303d20"
|
|
fp = "src2:3333333333333333333333333333333333333333333333333333333333333333"
|
|
)
|
|
|
|
// fakeBuildModules are the build modules on the nodes as the real ones run since R4g (ADR 0525; issue 1887 A4): each pulls
|
|
// its build seat's asks on its nodes (buildstep.Puller over the seat's workers) and runs an ask's job with step bodies that
|
|
// answer as the real tools do, ending the step on <node>.<ask id> of its steps bucket.
|
|
type fakeBuildModules struct {
|
|
js jetstream.JetStream
|
|
ctx context.Context
|
|
failBuild atomic.Bool
|
|
// refusePush makes every bundle and archive build fail as a push the artifact store refused (ADR 0498 clause 2).
|
|
refusePush atomic.Bool
|
|
mu sync.Mutex
|
|
calls map[string]int // by `<module>.<tool>`
|
|
fetches []map[string]any
|
|
builds []map[string]any // the arguments of every build call
|
|
// closed, while set, keeps every node's capacity at 0: no ask is pulled until open is called (a test of the asks
|
|
// waiting in their streams).
|
|
closed atomic.Bool
|
|
pullers []*buildstep.Puller
|
|
}
|
|
|
|
// open lets the nodes pull again and wakes their pullers.
|
|
func (f *fakeBuildModules) open() {
|
|
f.closed.Store(false)
|
|
f.mu.Lock()
|
|
ps := append([]*buildstep.Puller(nil), f.pullers...)
|
|
f.mu.Unlock()
|
|
for _, p := range ps {
|
|
p.Wake()
|
|
}
|
|
}
|
|
|
|
func (f *fakeBuildModules) count(tool string) int {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.calls[tool]
|
|
}
|
|
|
|
// digestOf is what every node builds an artifact to.
|
|
func digestOf(artifact string) string {
|
|
sum := sha256.Sum256([]byte(artifact))
|
|
return "sha256:" + hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
var prefixOf = map[string]string{"mesh-build-go": "build_go_", "mesh-build-node": "build_node_", "mesh-build-image": "build_image_"}
|
|
|
|
// step is one tool's answer, as the real module's.
|
|
func (f *fakeBuildModules) step(module, name string, args map[string]any) (any, error) {
|
|
tool := prefixOf[module] + name
|
|
f.mu.Lock()
|
|
f.calls[module+"."+tool]++
|
|
if name == "fetch" {
|
|
f.fetches = append(f.fetches, args)
|
|
} else {
|
|
f.builds = append(f.builds, args)
|
|
}
|
|
f.mu.Unlock()
|
|
switch {
|
|
case name == "fetch":
|
|
merge := ""
|
|
if args["base"] != nil { // a fetch with a base answers the merge commit (mesh-build-go FetchResult)
|
|
merge = `,"merge":"` + strings.Repeat("9", 40) + `"`
|
|
}
|
|
return json.RawMessage(`{"workspace":"w` + buildstep.NewID() + `","commit":"` + commit + `"` + merge + `}`), nil
|
|
case f.failBuild.Load() && tool != "build_image_copy": // the build fails wherever it runs
|
|
return nil, fmt.Errorf("go build of %v: boom", args["artifact"])
|
|
case tool == "build_image_copy": // the upstream copy answers the digest and the registry's reference (ADR 0498 clause 4)
|
|
d := digestOf(fmt.Sprint(args["artifact"]))
|
|
return json.RawMessage(`{"artifact":"` + fmt.Sprint(args["artifact"]) + `","digest":"` + d + `","reference":"registry.example/` + fmt.Sprint(args["module"]) + `/` + fmt.Sprint(args["artifact"]) + `@` + d + `","pushed":true}`), nil
|
|
case tool != "build_image_build" && f.refusePush.Load():
|
|
return nil, fmt.Errorf("%s%v: the artifact store answered 507 Insufficient Storage", delivery.NotStored, args["artifact"])
|
|
case tool != "build_image_build": // a bundle or an archive is pushed and its reference answered (ADR 0498 clause 1)
|
|
d := digestOf(fmt.Sprint(args["artifact"]))
|
|
return json.RawMessage(`{"digest":"` + d + `","reference":"artifact-store://` + filepath.Base(fmt.Sprint(args["module"])) + `/` + fmt.Sprint(args["artifact"]) + `/blobs/` + d + `"}`), nil
|
|
case tool == "build_image_build" && args["publish"] == true: // a trunk image is pushed: the digest and the reference
|
|
d := digestOf(fmt.Sprint(args["artifact"]))
|
|
return json.RawMessage(`{"digest":"` + d + `","reference":"registry.example/imaged/server@` + d + `"}`), nil
|
|
default:
|
|
return json.RawMessage(`{"digest":"` + digestOf(fmt.Sprint(args["artifact"])) + `"}`), nil
|
|
}
|
|
}
|
|
|
|
// serve makes module's steps bucket and its build seat's stream and workers, and pulls its asks on nodes.
|
|
func (f *fakeBuildModules) serve(t *testing.T, module string, nodes ...string) {
|
|
kv, err := f.js.CreateKeyValue(f.ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket(module), History: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
language := strings.TrimPrefix(module, "mesh-build-")
|
|
if err := buildstep.DeclareAskWorkers(f.ctx, f.js, language, 30*time.Second); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
put := func(key string, value any) (uint64, error) {
|
|
raw, _ := value.(json.RawMessage)
|
|
return kv.Put(f.ctx, key, raw)
|
|
}
|
|
call := func(name string) buildstep.StepFunc {
|
|
return func(ctx context.Context, args map[string]any) (any, error) { return f.step(module, name, args) }
|
|
}
|
|
steps := buildstep.AskSteps{Fetch: call("fetch"), Build: call("build"), Copy: call("copy")}
|
|
for _, node := range nodes {
|
|
p := &buildstep.Puller{Source: &buildstep.JetStreamAsks{JS: f.js, Language: language}, Language: language, Node: node, Put: put,
|
|
Capacity: func() int {
|
|
if f.closed.Load() {
|
|
return 0
|
|
}
|
|
return 2
|
|
}, PullWait: 200 * time.Millisecond,
|
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return buildstep.RunAsk(ctx, steps, a) }}
|
|
f.mu.Lock()
|
|
f.pullers = append(f.pullers, p)
|
|
f.mu.Unlock()
|
|
go func() { _ = p.Pull(f.ctx) }()
|
|
}
|
|
}
|
|
|
|
type bed struct {
|
|
ctx context.Context
|
|
js jetstream.JetStream
|
|
modules jetstream.KeyValue
|
|
candidates jetstream.KeyValue
|
|
builds jetstream.KeyValue
|
|
tools map[string]jetstream.KeyValue
|
|
instances jetstream.KeyValue
|
|
stop func() // stops the builder started last
|
|
fake *fakeBuildModules
|
|
srvURL string
|
|
dir string
|
|
}
|
|
|
|
// startBed starts the bus, the buckets the controller makes from the builder's and the build modules' manifests, the
|
|
// build modules' streams and the builder's durable consumers on them, two nodes' toolchains, the indexer's manifests, and
|
|
// the fake build modules. It runs no builder.
|
|
func startBed(t *testing.T) *bed { return startBedWith(t, false) }
|
|
|
|
// startBedWith is startBed with the nodes' capacity 0 from the start when closed: a puller that reads its capacity before a
|
|
// test could close the nodes may be in a pull already and takes the first ask put (issue 1928).
|
|
func startBedWith(t *testing.T, closed bool) *bed {
|
|
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "b")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { os.RemoveAll(dir) })
|
|
conf := filepath.Join(dir, "bus.conf")
|
|
if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
opts, err := server.ProcessConfigFile(conf)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
opts.NoLog, opts.NoSigs = true, true
|
|
srv, err := server.NewServer(opts)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
go srv.Start()
|
|
if !srv.ReadyForConnections(10 * time.Second) {
|
|
t.Fatal("the bus did not come up")
|
|
}
|
|
t.Cleanup(srv.Shutdown)
|
|
ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(ctl.Close)
|
|
js, err := jetstream.New(ctl)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
|
t.Cleanup(cancel)
|
|
b := &bed{ctx: ctx, js: js, tools: map[string]jetstream.KeyValue{}, srvURL: srv.ClientURL(), dir: dir}
|
|
for module, stream := range toolchainStreams {
|
|
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.stream." + module + ".event.toolchain.reported"}}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, node := range []string{"n1", "n2"} {
|
|
if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket(module), History: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.tools[module] = kv
|
|
}
|
|
mk := func(name string, history int) jetstream.KeyValue {
|
|
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return kv
|
|
}
|
|
b.modules = mk(delivery.ModulesBucket, delivery.ModulesHistory)
|
|
b.candidates = mk(delivery.CandidatesBucket, delivery.CandidatesHistory)
|
|
mk(delivery.ManifestsBucket, delivery.ManifestsHistory)
|
|
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
|
|
b.instances = mk(instancesBucket, 1)
|
|
b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}}
|
|
b.fake.closed.Store(closed)
|
|
b.fake.serve(t, "mesh-build-go", "n1", "n2")
|
|
b.fake.serve(t, "mesh-build-node", "n1", "n2")
|
|
b.fake.serve(t, "mesh-build-image", "n1") // the image build module runs on n1 only
|
|
b.report(t, "mesh-build-go", "n1", "go", "go1.27.1", "1")
|
|
b.report(t, "mesh-build-go", "n2", "go", "go1.27.1", "1")
|
|
b.report(t, "mesh-build-node", "n1", "typescript", "tsc5", "1")
|
|
b.report(t, "mesh-build-node", "n2", "typescript", "tsc5", "1")
|
|
return b
|
|
}
|
|
|
|
// report writes a node's toolchains key of a build module and says so on the module's stream, as the module does.
|
|
func (b *bed) report(t *testing.T, module, node, language, version, recipe string) {
|
|
t.Helper()
|
|
data, err := delivery.EncodeToolchain(delivery.Toolchain{Language: language, Version: version, Recipe: recipe, At: time.Now().UTC()})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
key, _ := delivery.ToolchainKey(node, language)
|
|
if _, err := b.tools[module].Put(b.ctx, key, data); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.js.Publish(b.ctx, "mesh.stream."+module+".event.toolchain.reported", []byte(`{}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// module writes modules.<name> as the indexer does for a module with a new-form fingerprint, and its manifest by digest.
|
|
func (b *bed) module(t *testing.T, name, fixture, path string) string {
|
|
return b.moduleWith(t, name, fixture, path, map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.1", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}})
|
|
}
|
|
|
|
// moduleWith is module with the toolchains its fingerprint was computed with.
|
|
func (b *bed) moduleWith(t *testing.T, name, fixture, path string, toolchains map[string]delivery.ToolchainUsed) string {
|
|
return b.moduleCommit(t, name, fixture, path, toolchains, commit)
|
|
}
|
|
|
|
func (b *bed) moduleCommit(t *testing.T, name, fixture, path string, toolchains map[string]delivery.ToolchainUsed, at string) string {
|
|
t.Helper()
|
|
raw, err := os.ReadFile(filepath.Join("testdata", fixture))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
digest := delivery.ManifestDigest(raw)
|
|
manifests, err := b.js.KeyValue(b.ctx, delivery.ManifestsBucket)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
mv, err := delivery.EncodeManifest(delivery.Manifest{Module: name, JSON: string(raw)})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := manifests.Put(b.ctx, digest, mv); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: at, FP: "src1:" + strings.Repeat("f", 64), FP2: fp,
|
|
Manifest: digest, Path: path, Toolchains: toolchains})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.modules.Put(b.ctx, name, data); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return digest
|
|
}
|
|
|
|
// archive writes modules.<name> as the real indexer does for a module of an archive only: its fingerprint names no
|
|
// toolchain (mesh-delivery-indexer internal/index inputs.go languagesOf gives an archive no language), so Toolchains is empty.
|
|
func (b *bed) archive(t *testing.T, name, path, at string) string {
|
|
t.Helper()
|
|
return b.moduleCommit(t, name, "catalogue-a67f0196-ca-trust.module.json", path, nil, at)
|
|
}
|
|
|
|
// start runs the program as the builder of a node.
|
|
func (b *bed) start(t *testing.T, node string, env map[string]string) {
|
|
t.Helper()
|
|
broker := filepath.Join(t.TempDir(), "broker")
|
|
cred, _ := json.Marshal(credential{URL: b.srvURL, Node: node, Module: "mesh-delivery-builder", User: node + ".mesh-delivery-builder", Password: "pw"})
|
|
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
e := map[string]string{"MESH_BROKER_FILE": broker}
|
|
for k, v := range env {
|
|
e[k] = v
|
|
}
|
|
runCtx, stop := context.WithCancel(b.ctx)
|
|
done := make(chan error, 1)
|
|
go func() { done <- run(runCtx, func(k string) string { return e[k] }, io.Discard) }()
|
|
var once sync.Once
|
|
b.stop = func() { once.Do(func() { stop(); <-done }) }
|
|
t.Cleanup(b.stop)
|
|
}
|
|
|
|
// await reads the build key until pred holds of it, within 20 s.
|
|
func (b *bed) await(t *testing.T, module string, pred func(delivery.BuilderBuild) bool) delivery.BuilderBuild {
|
|
t.Helper()
|
|
key, err := delivery.BuilderBuildKey(module, fp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var last delivery.BuilderBuild
|
|
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
e, err := b.builds.Get(b.ctx, key)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if last, err = delivery.DecodeBuilderBuild(e.Value()); err == nil && pred(last) {
|
|
return last
|
|
}
|
|
}
|
|
t.Fatalf("builds.%s = %+v within 20 s", key, last)
|
|
return last
|
|
}
|
|
|
|
func isState(state string) func(delivery.BuilderBuild) bool {
|
|
return func(v delivery.BuilderBuild) bool { return v.State == state }
|
|
}
|
|
|
|
// TestBuilderMakesADoneKeyWithTheDigests: modules.netcheck (a Go and a TypeScript bundle) and its manifest are in the
|
|
// indexer's buckets, and two nodes report each build module's toolchain; the program writes builds.netcheck.<fp> done,
|
|
// the digests of both bundles, both build modules' toolchains, after a fetch and a build on one node (ADR 0478 clause 4).
|
|
func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
|
|
b := startBed(t)
|
|
digest := b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "netcheck", isState(delivery.BuildDone))
|
|
if v.By != "n1.mesh-delivery-builder" || v.Manifest != digest || len(v.Artifacts) != 2 {
|
|
t.Fatalf("build = %+v", v)
|
|
}
|
|
for _, a := range v.Artifacts {
|
|
if a.Digest != digestOf(a.Name) || a.Kind != "bundle" || a.Reference != "artifact-store://netcheck/"+a.Name+"/blobs/"+a.Digest {
|
|
t.Errorf("artifact = %+v, want the digest every node built and the reference its push answered (ADR 0498 clause 1)", a)
|
|
}
|
|
}
|
|
if v.Toolchains["go"].Version != "go1.27.1" || v.Toolchains["typescript"].Version != "tsc5" {
|
|
t.Errorf("toolchains = %+v", v.Toolchains)
|
|
}
|
|
for tool, want := range map[string]int{"mesh-build-go.build_go_fetch": 1, "mesh-build-go.build_go_build": 1, "mesh-build-node.build_node_fetch": 1, "mesh-build-node.build_node_build": 1} {
|
|
if got := b.fake.count(tool); got != want {
|
|
t.Errorf("%s called %d times, want %d (on one node, ADR 0478 clause 4)", tool, got, want)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestTwoBuilderInstancesStartingTogetherEndWithOneDoneKey: two builders on two nodes start together on one module key of an
|
|
// archive only (no toolchain named, as the indexer writes it); the second takes over the key the first holds building
|
|
// (ADR 0478 clause 6), and the key ends done, written once.
|
|
func TestTwoBuilderInstancesStartingTogetherEndWithOneDoneKey(t *testing.T) { // 1607/AC7
|
|
b := startBed(t)
|
|
b.archive(t, "ca-trust", "modules/ca-trust", commit)
|
|
b.start(t, "n1", nil)
|
|
b.start(t, "n2", nil)
|
|
v := b.await(t, "ca-trust", isState(delivery.BuildDone))
|
|
if len(v.Artifacts) != 1 || v.Artifacts[0].Kind != "archive" || v.Artifacts[0].Digest != digestOf("files") {
|
|
t.Fatalf("build = %+v, want the archive's digest", v)
|
|
}
|
|
if n := b.fake.count("mesh-build-go.build_go_build"); n < 1 { // each instance that built, on one node
|
|
t.Errorf("build_go_build was called %d times, want one for each instance that built", n)
|
|
}
|
|
}
|
|
|
|
// TestAnArchiveOnlyModuleNamingNoToolchainIsBuilt: the indexer gives an archive no language, so modules.ca-trust has no
|
|
// toolchains; the builder builds it on the build module's nodes all the same (review round 4).
|
|
func TestAnArchiveOnlyModuleNamingNoToolchainIsBuilt(t *testing.T) {
|
|
b := startBed(t)
|
|
b.archive(t, "ca-trust", "modules/ca-trust", commit)
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "ca-trust", isState(delivery.BuildDone))
|
|
if len(v.Toolchains) != 0 || len(v.Artifacts) != 1 || strings.HasPrefix(v.Step, "one node (") {
|
|
t.Errorf("build = %+v, want done naming no toolchain", v)
|
|
}
|
|
}
|
|
|
|
// TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFingerprint: a build step fails, so the key is
|
|
// failed naming it; the indexer writing modules.<m> again with the same fingerprint wakes the builder, which builds
|
|
// the key again as a new revision of the same key, now done.
|
|
func TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFingerprint(t *testing.T) { // 1607/AC6, AC7
|
|
b := startBed(t)
|
|
b.fake.failBuild.Store(true)
|
|
b.archive(t, "ca-trust", "modules/ca-trust", commit)
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "ca-trust", isState(delivery.BuildFailed))
|
|
if !strings.Contains(v.Error, " on n") || !strings.Contains(v.Error, "boom") {
|
|
t.Errorf("build = %+v, want failed naming the node and the step's reason", v)
|
|
}
|
|
key, _ := delivery.BuilderBuildKey("ca-trust", fp)
|
|
failed, err := b.builds.Get(b.ctx, key)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.fake.failBuild.Store(false)
|
|
// The indexer writes the module again (here at a later commit), with the same fingerprint.
|
|
b.archive(t, "ca-trust", "modules/ca-trust", "89abcdef0123456789abcdef0123456789abcdef")
|
|
b.await(t, "ca-trust", isState(delivery.BuildDone))
|
|
done, err := b.builds.Get(b.ctx, key)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if done.Revision() <= failed.Revision() {
|
|
t.Errorf("revision %d, was %d: the key was not written as a new revision", done.Revision(), failed.Revision())
|
|
}
|
|
}
|
|
|
|
// TestAToolchainChangeDoesNotBuildTheOldFingerprint: the fingerprint names go1.27.1, the nodes then agree on go1.27.2 (a
|
|
// report of both nodes), and the builder writes no key for the old fingerprint although it was woken by the report; a
|
|
// module value naming go1.27.2 (the new fingerprint, computed by the indexer) is built.
|
|
func TestAToolchainChangeDoesNotBuildTheOldFingerprint(t *testing.T) { // 1615/AC4
|
|
b := startBed(t)
|
|
b.report(t, "mesh-build-go", "n1", "go", "go1.27.2", "1")
|
|
b.report(t, "mesh-build-go", "n2", "go", "go1.27.2", "1")
|
|
b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") // names go1.27.1
|
|
b.start(t, "n1", nil)
|
|
// Another module, whose fingerprint names the new toolchain, is built; by then the old one has been looked at.
|
|
b.moduleCommit(t, "netcheck2", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck",
|
|
map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.2", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}}, commit)
|
|
b.await(t, "netcheck2", isState(delivery.BuildDone))
|
|
if _, err := b.builds.Get(b.ctx, mustKey(t, "netcheck")); err == nil {
|
|
t.Error("the old fingerprint, naming go1.27.1, was built with go1.27.2")
|
|
}
|
|
if n := b.fake.count("mesh-build-go.build_go_build"); n != 1 {
|
|
t.Errorf("build_go_build was called %d times, want 1 (the new fingerprint on one node only)", n)
|
|
}
|
|
}
|
|
|
|
// TestARootModuleIsBuilt: a module whose path is "." is given to the build tools as it is.
|
|
func TestARootModuleIsBuilt(t *testing.T) { // 1615/AC4
|
|
b := startBed(t)
|
|
b.archive(t, "ca-trust", delivery.RootPath, commit)
|
|
b.start(t, "n1", nil)
|
|
b.await(t, "ca-trust", isState(delivery.BuildDone))
|
|
}
|
|
|
|
func mustKey(t *testing.T, module string) string {
|
|
k, err := delivery.BuilderBuildKey(module, fp)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return k
|
|
}
|
|
|
|
// moduleAt is module with the go toolchain version the fingerprint names and the commit it was computed at.
|
|
func (b *bed) moduleAt(t *testing.T, name, fixture, path, goVersion, at string) {
|
|
t.Helper()
|
|
b.moduleCommit(t, name, fixture, path, map[string]delivery.ToolchainUsed{"go": {Version: goVersion, Recipe: "1"}}, at)
|
|
}
|
|
|
|
// TestAKeyLeftBuildingForAModuleThatIsGoneIsClosedAtStart: the builds bucket holds a key left building by another node for
|
|
// a module the indexer no longer has; the program, at start, fails it naming why (issue 1660, ADR 0478 clause 6).
|
|
func TestAKeyLeftBuildingForAModuleThatIsGoneIsClosedAtStart(t *testing.T) {
|
|
b := startBed(t)
|
|
data, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "n2.mesh-delivery-builder", Manifest: strings.Repeat("a", 64)})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.builds.Put(b.ctx, mustKey(t, "ghost"), data); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "ghost", isState(delivery.BuildFailed))
|
|
if !strings.Contains(v.Error, "no longer has this fingerprint") || !strings.Contains(v.Error, "n2.mesh-delivery-builder") {
|
|
t.Errorf("build = %+v, want it closed naming who left it", v)
|
|
}
|
|
}
|
|
|
|
// TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk: the trunk's record of netcheck carries a fingerprint of its
|
|
// own; a candidate the indexer wrote gives netcheck the fingerprint fp at a pull request's head; the program writes
|
|
// builds.netcheck.<fp> done, its fetch given the head, the ref that carries it and the trunk commit (issue 1684).
|
|
func TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk(t *testing.T) { // N7/AC1
|
|
b := startBed(t)
|
|
other := "src2:" + strings.Repeat("e", 64)
|
|
digest := b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")
|
|
data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: commit, FP: "src1:" + strings.Repeat("f", 64), FP2: other,
|
|
Manifest: digest, Path: "modules/netcheck", Toolchains: map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.1", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.modules.Put(b.ctx, "netcheck", data); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
const head = "1111111111111111111111111111111111111111"
|
|
pulls := []delivery.CandidatePull{{Head: delivery.Head{Repository: "novox/mesh-catalog", Number: 7, Head: head}, Modules: []string{"netcheck"},
|
|
Fps: map[string]delivery.CandidateModule{"netcheck": {FP2: fp, Manifest: digest}}}}
|
|
hash, _ := delivery.HeadsHash([]delivery.Head{pulls[0].Head})
|
|
cand, err := delivery.EncodeCandidate(delivery.Candidate{V: delivery.Version, Group: "feat/x", HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ckey, err := delivery.CandidatesKey("feat/x")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.candidates.Put(b.ctx, ckey, cand); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "netcheck", isState(delivery.BuildDone))
|
|
if len(v.Artifacts) != 2 {
|
|
t.Fatalf("build = %+v", v)
|
|
}
|
|
b.fake.mu.Lock()
|
|
defer b.fake.mu.Unlock()
|
|
found := false
|
|
for _, f := range b.fake.fetches {
|
|
if f["commit"] == head && f["ref"] == "refs/pull/7/head" && f["base"] == commit {
|
|
found = true
|
|
}
|
|
}
|
|
if !found {
|
|
t.Errorf("fetches = %v, want one of the candidate's head with its ref and the trunk commit", b.fake.fetches)
|
|
}
|
|
}
|
|
|
|
// TestAnImageIsBuiltOnOneNodeAndOnlyATrunkKeyPublishes: modules.imaged (one image artifact) is built on n1, where
|
|
// mesh-build-image runs: a trunk record's key is done with the digest and the reference, its build call carrying publish; a
|
|
// candidate's key for the same module is built first with no publish and no reference; when the pull request is merged and the
|
|
// trunk's record names the candidate's fingerprint, the key is built again with publish and ends with a reference
|
|
// (ADR 0487 clauses 5 and 12).
|
|
func TestAnImageIsBuiltOnOneNodeAndOnlyATrunkKeyPublishes(t *testing.T) { // 1705
|
|
b := startBed(t)
|
|
other := "src2:" + strings.Repeat("e", 64)
|
|
digest := b.moduleCommit(t, "imaged", "catalogue-a67f0196-imaged.module.json", "modules/imaged", nil, commit)
|
|
// The trunk carries `other`; its key is built and published.
|
|
trunk := func(fp2 string) {
|
|
data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: commit, FP: "src1:" + strings.Repeat("f", 64), FP2: fp2,
|
|
Manifest: digest, Path: "modules/imaged"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.modules.Put(b.ctx, "imaged", data); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
trunk(other)
|
|
b.start(t, "n1", nil)
|
|
okey, _ := delivery.BuilderBuildKey("imaged", other)
|
|
awaitKey := func(key string, pred func(delivery.BuilderBuild) bool) delivery.BuilderBuild {
|
|
var last delivery.BuilderBuild
|
|
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
|
if e, err := b.builds.Get(b.ctx, key); err == nil {
|
|
if last, err = delivery.DecodeBuilderBuild(e.Value()); err == nil && pred(last) {
|
|
return last
|
|
}
|
|
}
|
|
}
|
|
t.Fatalf("builds.%s = %+v within 20 s", key, last)
|
|
return last
|
|
}
|
|
v := awaitKey(okey, isState(delivery.BuildDone))
|
|
if len(v.Artifacts) != 1 || v.Artifacts[0].Digest != digestOf("server") || !strings.HasSuffix(v.Artifacts[0].Reference, "@"+digestOf("server")) {
|
|
t.Fatalf("trunk build = %+v, want the digest and the reference", v)
|
|
}
|
|
// The candidate carries fp: built, not published.
|
|
const head = "1111111111111111111111111111111111111111"
|
|
pulls := []delivery.CandidatePull{{Head: delivery.Head{Repository: "novox/mesh-catalog", Number: 7, Head: head}, Modules: []string{"imaged"},
|
|
Fps: map[string]delivery.CandidateModule{"imaged": {FP2: fp, Manifest: digest}}}}
|
|
hash, _ := delivery.HeadsHash([]delivery.Head{pulls[0].Head})
|
|
cand, err := delivery.EncodeCandidate(delivery.Candidate{V: delivery.Version, Group: "feat/x", HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ckey, _ := delivery.CandidatesKey("feat/x")
|
|
if _, err := b.candidates.Put(b.ctx, ckey, cand); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
v = b.await(t, "imaged", isState(delivery.BuildDone))
|
|
if len(v.Artifacts) != 1 || v.Artifacts[0].Reference != "" {
|
|
t.Fatalf("candidate build = %+v, want done with no reference", v)
|
|
}
|
|
b.fake.mu.Lock()
|
|
if len(b.fake.builds) != 2 || b.fake.builds[0]["publish"] != true || b.fake.builds[1]["publish"] != nil {
|
|
t.Errorf("builds = %v, want the trunk's with publish, then the candidate's with none", b.fake.builds)
|
|
}
|
|
b.fake.mu.Unlock()
|
|
// The pull request is merged: the trunk's record names fp, and its key ends published, built once more with publish.
|
|
trunk(fp)
|
|
v = b.await(t, "imaged", func(v delivery.BuilderBuild) bool {
|
|
return v.State == delivery.BuildDone && len(v.Artifacts) == 1 && v.Artifacts[0].Reference != ""
|
|
})
|
|
if v.Artifacts[0].Digest != digestOf("server") {
|
|
t.Errorf("merged build = %+v", v)
|
|
}
|
|
b.fake.mu.Lock()
|
|
defer b.fake.mu.Unlock()
|
|
published := 0
|
|
for _, a := range b.fake.builds {
|
|
if a["publish"] == true {
|
|
published++
|
|
}
|
|
}
|
|
if len(b.fake.builds) != 3 || published != 2 {
|
|
t.Errorf("builds = %v, want three, two of them with publish (the trunk's two fingerprints)", b.fake.builds)
|
|
}
|
|
}
|
|
|
|
// TestEveryDoneKeyNamesItsReferencesAndARefusedPushIsBuiltAgain: with the artifact store refusing pushes, the key of an
|
|
// archive module ends failed opening "not stored:"; once it takes them, a new modules value builds the key again, done, its
|
|
// every artifact naming a reference (issue 1587, B8, ADR 0498 clauses 1, 2 and 8).
|
|
func TestEveryDoneKeyNamesItsReferencesAndARefusedPushIsBuiltAgain(t *testing.T) {
|
|
b := startBed(t)
|
|
b.fake.refusePush.Store(true)
|
|
b.archive(t, "ca-trust", "modules/ca-trust", commit)
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "ca-trust", isState(delivery.BuildFailed))
|
|
if !strings.HasPrefix(v.Error, delivery.NotStored) || !strings.Contains(v.Error, "507") {
|
|
t.Fatalf("build = %+v, want failed opening %q with the store's words", v, delivery.NotStored)
|
|
}
|
|
b.fake.refusePush.Store(false)
|
|
b.archive(t, "ca-trust", "modules/ca-trust", "89abcdef0123456789abcdef0123456789abcdef")
|
|
v = b.await(t, "ca-trust", isState(delivery.BuildDone))
|
|
if ok, missing := v.Referenced(); !ok {
|
|
t.Errorf("build = %+v, artifacts naming no reference: %v", v, missing)
|
|
}
|
|
}
|
|
|
|
// TestAPinnedUpstreamIsCopiedByTheImageBuildModule: hello-web's manifest (captured from mesh-catalog 49caf17f) names an
|
|
// upstream image pinned by its digest and an archive; the upstream is copied on n1, where mesh-build-image runs, with no
|
|
// fetch of the image module, and the key ends done with both references (issue 1587, B8, ADR 0498 clause 4).
|
|
func TestAPinnedUpstreamIsCopiedByTheImageBuildModule(t *testing.T) {
|
|
b := startBed(t)
|
|
b.moduleCommit(t, "hello-web", "catalogue-49caf17f-hello-web.module.json", "modules/hello-web", nil, commit)
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "hello-web", isState(delivery.BuildDone))
|
|
if ok, missing := v.Referenced(); !ok {
|
|
t.Fatalf("build = %+v, artifacts naming no reference: %v", v, missing)
|
|
}
|
|
if n := b.fake.count("mesh-build-image.build_image_copy"); n != 1 {
|
|
t.Errorf("build_image_copy called %d times, want once", n)
|
|
}
|
|
if n := b.fake.count("mesh-build-image.build_image_fetch"); n != 0 {
|
|
t.Errorf("build_image_fetch called %d times, want none for a copy", n)
|
|
}
|
|
}
|
|
|
|
// TestAManifestFallingBackToAPredefinedKeyIsBuilt: claude-code's manifest (mesh-catalog 835bf22b) has a config value falling
|
|
// back to a predefined key; the builder, which holds no published list of them, reads it and builds the key done (issue
|
|
// 1751, mesh-sdk catalogue.ParseManifestForInputs).
|
|
func TestAManifestFallingBackToAPredefinedKeyIsBuilt(t *testing.T) {
|
|
b := startBed(t)
|
|
b.module(t, "claude-code", "catalogue-835bf22b-claude-code.module.json", "modules/claude-code")
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "claude-code", func(v delivery.BuilderBuild) bool {
|
|
return v.State == delivery.BuildDone || v.State == delivery.BuildFailed
|
|
})
|
|
if v.State != delivery.BuildDone {
|
|
t.Errorf("build = %+v, want done: the manifest is read", v)
|
|
}
|
|
}
|
|
|
|
// 1886/AC5: a builder on n1 whose Go, Node and image build modules each answered a reference says under n1, in its bucket
|
|
// of instances, that it writes references, and takes the say away when it stops: the controller delivers from its keys
|
|
// only while it holds.
|
|
func TestABuilderWhoseBuildModulesAnsweredReferencesSaysItAndStoppedSaysNothing(t *testing.T) {
|
|
b := startBed(t)
|
|
b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") // a Go and a TypeScript bundle
|
|
b.moduleCommit(t, "imaged", "catalogue-a67f0196-imaged.module.json", "modules/imaged", nil, commit) // an image, for the trunk
|
|
b.start(t, "n1", nil)
|
|
var said string
|
|
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && said == ""; time.Sleep(50 * time.Millisecond) {
|
|
if e, err := b.instances.Get(b.ctx, "n1"); err == nil {
|
|
said = string(e.Value())
|
|
}
|
|
}
|
|
if !strings.Contains(said, `"writes":["references"]`) {
|
|
t.Fatalf("the builder said %q", said)
|
|
}
|
|
b.stop()
|
|
if _, err := b.instances.Get(b.ctx, "n1"); !errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
t.Fatalf("the stopped builder still says: %v", err)
|
|
}
|
|
}
|
|
|
|
// 1887/AC1, AC2 on the real bus: the builder puts the trunk fingerprint's build as asks on the build seats' delivery
|
|
// streams and calls no node's tool; the build modules that pulled them wrote the steps, and the key is done naming the
|
|
// references.
|
|
func TestTheBuilderAsksTheBuildSeatsAndCallsNoNode(t *testing.T) {
|
|
b := startBed(t)
|
|
b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")
|
|
sub, err := nats.Connect(b.srvURL, nats.UserInfo("ctl", "pw"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer sub.Close()
|
|
var calls atomic.Int64
|
|
s, err := sub.Subscribe("mesh.mod.*.call.>", func(*nats.Msg) { calls.Add(1) })
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer s.Unsubscribe()
|
|
asked := b.asksPut(t, sub)
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "netcheck", isState(delivery.BuildDone))
|
|
for _, a := range v.Artifacts {
|
|
if a.Reference == "" {
|
|
t.Errorf("artifact %s names no reference", a.Name)
|
|
}
|
|
}
|
|
if n := calls.Load(); n != 0 {
|
|
t.Errorf("%d tool calls reached a build module", n)
|
|
}
|
|
// 1887/AC1: one ask per build module the build needs (netcheck: a Go and a TypeScript bundle), each on its seat's
|
|
// delivery stream.
|
|
got := asked()
|
|
if len(got) != 2 {
|
|
t.Fatalf("asks put = %+v, want one Go and one Node delivery ask", got)
|
|
}
|
|
for _, a := range got {
|
|
if a.Kind != delivery.AskDelivery || a.Build == nil || a.Build.FP != fp {
|
|
t.Errorf("ask %+v, want a delivery ask of the fingerprint", a)
|
|
}
|
|
}
|
|
}
|
|
|
|
// asksPut hears every ask put on a build seat's accept subjects (a core subscription beside the stream) and answers the
|
|
// asks heard so far.
|
|
func (b *bed) asksPut(t *testing.T, nc *nats.Conn) func() []delivery.StepAsk {
|
|
t.Helper()
|
|
var mu sync.Mutex
|
|
var asks []delivery.StepAsk
|
|
s, err := nc.Subscribe("mesh.seat.*.accept.>", func(m *nats.Msg) {
|
|
if a, err := delivery.DecodeStepAsk(m.Data); err == nil {
|
|
mu.Lock()
|
|
asks = append(asks, a)
|
|
mu.Unlock()
|
|
}
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = s.Unsubscribe() })
|
|
return func() []delivery.StepAsk {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
return append([]delivery.StepAsk(nil), asks...)
|
|
}
|
|
}
|
|
|
|
// 1887/AC3: a builder that started again while its asks waited in the streams (no node had a free slot) puts no second
|
|
// ask: it waits for the asks its building key names, and the key is done once a node took them.
|
|
func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) {
|
|
b := startBedWith(t, true) // 1928/AC1: no node can take an ask from the start, so no pull in hand takes the first one
|
|
b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")
|
|
sub, err := nats.Connect(b.srvURL, nats.UserInfo("ctl", "pw"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer sub.Close()
|
|
asked := b.asksPut(t, sub)
|
|
b.start(t, "n1", nil)
|
|
v := b.await(t, "netcheck", func(v delivery.BuilderBuild) bool { return v.State == delivery.BuildBuilding && len(v.Asks) == 2 })
|
|
for deadline := time.Now().Add(20 * time.Second); len(asked()) < 2; time.Sleep(20 * time.Millisecond) {
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("asks put = %+v within 20 s, want the two the key names", asked())
|
|
}
|
|
}
|
|
b.stop()
|
|
b.start(t, "n1", nil) // the same builder, started again, while both asks wait
|
|
b.fake.open()
|
|
done := b.await(t, "netcheck", isState(delivery.BuildDone))
|
|
if got := asked(); len(got) != 2 {
|
|
t.Fatalf("asks put = %d (%+v), want the two the key named and none again", len(got), got)
|
|
}
|
|
for lang, id := range v.Asks {
|
|
found := false
|
|
for _, a := range asked() {
|
|
found = found || a.ID == id
|
|
}
|
|
if !found {
|
|
t.Errorf("the %s ask %s the key named was never put", lang, id)
|
|
}
|
|
}
|
|
if len(done.Artifacts) != 2 {
|
|
t.Errorf("done = %+v", done)
|
|
}
|
|
}
|