novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0bcc046c3dde0061760d90e0137a84d3fa60d73b
mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go
T

1344 lines
59 KiB
Go
Raw Blame History

//go:build integration
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"maps"
"os"
"path/filepath"
"slices"
"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"
// The build runner the build modules pull and run asks with: mesh-build-go's own package (novox/hq ADR 0593 §1),
// another repository's, by its published version, so the builder is tested against the real pulling code.
runner "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry"
"git.novox.be/novox/mesh-contracts/go/seats/ociartifactregistry"
sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep"
"git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/views/viewstest"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
)
// 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 (runner.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 []*runner.Puller
// hang, while it holds a channel, keeps every build step running until the channel is closed: a step that never ends,
// so the builder's step bound runs out (design 51's round-2 case).
hang atomic.Pointer[chan struct{}]
}
// open lets the nodes pull again and wakes their pullers.
func (f *fakeBuildModules) open() {
f.closed.Store(false)
f.mu.Lock()
ps := append([]*runner.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()
if h := f.hang.Load(); h != nil && name == "build" {
select {
case <-*h:
case <-f.ctx.Done():
}
}
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 := runner.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) runner.StepFunc {
return func(ctx context.Context, args map[string]any) (any, error) { return f.step(module, name, args) }
}
steps := runner.AskSteps{Fetch: call("fetch"), Build: call("build"), Copy: call("copy")}
for _, node := range nodes {
p := &runner.Puller{Source: &runner.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 runner.RunAsk(ctx, steps, a) }}
f.mu.Lock()
f.pullers = append(f.pullers, p)
f.mu.Unlock()
go func() { _ = p.Pull(f.ctx) }()
}
}
// trunkHead is the trunk head of novox/mesh-catalog the bed's source revision names: newer than the module values' commit.
const trunkHead = "3333333333333333333333333333333333333333"
// trunk writes the gate's source.main naming head as novox/mesh-catalog's trunk head, and answers its revision.
func (b *bed) trunk(t *testing.T, head string) uint64 {
t.Helper()
data, err := delivery.EncodeSource(delivery.Source{V: delivery.Version, Group: "g", Heads: map[string]string{"novox/mesh-catalog": head}})
if err != nil {
t.Fatal(err)
}
rev, err := b.source.Put(b.ctx, delivery.SourceMainKey, data)
if err != nil {
t.Fatal(err)
}
return rev
}
type bed struct {
ctx context.Context
js jetstream.JetStream
modules jetstream.KeyValue
candidates jetstream.KeyValue
source jetstream.KeyValue
builds jetstream.KeyValue
tools map[string]jetstream.KeyValue
instances jetstream.KeyValue
tagAsks jetstream.KeyValue
tagBounds jetstream.KeyValue
packages jetstream.KeyValue // the indexer's packages, served by its read tool and views family
out *saidLog // the builder's process log, every instance's
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)
b.source = mk(delivery.SourceBucket, delivery.SourceHistory)
mk(delivery.ManifestsBucket, delivery.ManifestsHistory)
// The indexer and the assigner as the owners of the states the builder follows: their read tools and views families
// (ADR 0615, DLV-M1); n1 and n2 each have their durable consumer.
readers := []string{"n1_" + delivery.BuilderModule, "n2_" + delivery.BuilderModule}
for _, st := range []string{modulesState, manifestsState, candidatesState} {
viewstest.Serve(t, ctx, js, ctl, delivery.IndexerModule, st, delivery.IndexerModule+".indexer_read_"+st, readers...)
}
b.packages = viewstest.Serve(t, ctx, js, ctl, delivery.IndexerModule, delivery.PackagesState, delivery.IndexerModule+".indexer_read_packages", readers...)
viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule+".assigner_read_placement", readers...)
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
b.instances = mk(instancesBucket, 1)
// The tag asks and their bounds (design 51 *Tagged versions*): the bounds declared key-ttl, so an ended bound leaves the
// bus's marker a watch hears.
b.tagAsks = mk(versions.TagAsksBucket, 1)
if b.tagBounds, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: versions.TagAskBoundsBucket, History: 1, LimitMarkerTTL: time.Minute}); err != nil {
t.Fatal(err)
}
b.out = &saidLog{}
// The builder's module log, as the runtime raises it (ADR 0297): its stream on the log's subjects.
logStream := buscompose.LogStreamName(delivery.BuilderModule, delivery.BuilderLog)
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: logStream, Subjects: []string{buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog) + ".>"},
AllowDirect: true}); err != nil {
t.Fatal(err)
}
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
}
// logged is every entry of module's key in the builder's log, read through process.Conn.ReadLog on the controller's user.
func (b *bed) logged(t *testing.T, module string) []delivery.BuildLogEntry {
t.Helper()
conn, err := process.Connect(process.Account{URL: b.srvURL, Module: "mesh-delivery-view", User: "ctl", Password: "pw"})
if err != nil {
t.Fatal(err)
}
defer conn.Close()
entries, err := conn.ReadLog(delivery.BuilderModule, delivery.BuilderLog).Since(time.Unix(0, 0), module)
if err != nil {
t.Fatal(err)
}
var out []delivery.BuildLogEntry
for _, le := range entries {
e, err := delivery.DecodeBuildLogEntry(le.Value)
if err != nil {
t.Fatal(err)
}
out = append(out, e)
}
return out
}
// 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, bothToolchains)
}
// bothToolchains are the toolchains the bed's nodes report, as a module's fingerprint names them.
var bothToolchains = 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()
return b.moduleFP(t, name, fixture, path, toolchains, at, fp, 0)
}
// moduleFP is moduleCommit with the new-form fingerprint fp2 and the source revision src of its merge (0: none).
func (b *bed) moduleFP(t *testing.T, name, fixture, path string, toolchains map[string]delivery.ToolchainUsed, at, fp2 string, src uint64) 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: fp2,
Manifest: digest, Path: path, Toolchains: toolchains, Src: src})
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(process.Account{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] }, b.out) }()
var once sync.Once
b.stop = func() { once.Do(func() { stop(); <-done }) }
t.Cleanup(b.stop)
}
// saidLog is the builder's process log as its instances write it.
type saidLog struct {
mu sync.Mutex
buf strings.Builder
}
func (l *saidLog) Write(p []byte) (int, error) {
l.mu.Lock()
defer l.mu.Unlock()
return l.buf.Write(p)
}
// awaitSaid reads the process log until it holds want, within 20 s, and answers the line.
func (b *bed) awaitSaid(t *testing.T, want string) string {
t.Helper()
deadline := time.Now().Add(20 * time.Second)
for {
b.out.mu.Lock()
all := b.out.buf.String()
b.out.mu.Unlock()
for _, line := range strings.Split(all, "\n") {
if strings.Contains(line, want) {
return line
}
}
if time.Now().After(deadline) {
t.Fatalf("the builder never said %q; it said:\n%s", want, all)
}
select {
case <-b.ctx.Done():
t.Fatal(b.ctx.Err())
case <-time.After(50 * time.Millisecond):
}
}
}
// 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()
return b.awaitFP(t, module, fp, pred)
}
// awaitFP is await of the build key of fingerprint at.
func (b *bed) awaitFP(t *testing.T, module, at string, pred func(delivery.BuilderBuild) bool) delivery.BuilderBuild {
t.Helper()
key, err := delivery.BuilderBuildKey(module, at)
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)
// DLV-B10 (design 51, round 2): the builder's versions, findings and the floor's three sources are on the bus; the seat
// answers 0.1.4, so netcheck's bundles are numbered from 0.1.5. Two merges close together, the first replaced before
// its claim: the second is numbered at once, and the first never.
reg := b.registry(t, "n1", "0.1.4")
const netcheck = "catalogue-a67f0196-netcheck.module.json"
first := "src2:" + strings.Repeat("1", 64)
b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, first, 1)
digest := b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, fp, 2)
// DLV-B6: a tag ask kept and never numbered (no check builds its head) is narrated once its bound ends on the bus, with
// what it waits on, and its bound is written again; no timer runs in the builder.
key, ask, err := versions.NewTagAsk("novox/mesh-sdk", "feat/x", "abc", "a group member's check", "me@shanks", time.Now())
if err != nil {
t.Fatal(err)
}
if _, err := b.tagAsks.Put(b.ctx, key, ask); err != nil {
t.Fatal(err)
}
b.start(t, "n1", map[string]string{"MESH_BUILDER_TAG_ASK_BOUND": "1s", "MESH_BUILDER_STEP_BOUND": "1s"})
v := b.await(t, "netcheck", isState(delivery.BuildDone))
for _, a := range []string{"tools-go", "tools-typescript"} {
es := reg.of(t, "netcheck/"+a)
if len(es) != 1 || es[0].Version != "0.1.5" || es[0].Fingerprint != fp || es[0].State != versions.Numbered || es[0].Digest != digestOf(a) {
t.Errorf("versions of netcheck/%s = %+v, want the second merge numbered 0.1.5 at once with its digest, the first never", a, es)
}
}
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)
}
// ADR 0618 §1: the done key names the reference forms it is written in.
if len(v.Needs) != 1 || v.Needs[0] != "form:artifact-store" {
t.Errorf("needs = %v, want the first form", v.Needs)
}
// 1955/L7 (ADR 0571 §1): the build's start and end are in the module log build-log under the module's key, read as a
// reader with the log's read reads them, the end naming the manifest and what was made.
// DLV-B10: between them the build log holds the build's narrations (build.said), each version numbered, and why.
var ended []delivery.BuildLogEntry
for _, e := range b.logged(t, "netcheck") {
if e.Event != delivery.BuildSaidEvent {
ended = append(ended, e)
} else if e.Why != versions.SaidNumbered || !strings.Contains(e.Line, "numbered 0.1.5") || e.FP != fp {
t.Errorf("said = %+v, want netcheck's versions numbered", e)
}
}
if len(ended) != 2 || ended[0].Event != delivery.BuildStartedEvent || ended[1].Event != delivery.BuildEndedEvent ||
ended[1].Manifest != digest || len(ended[1].Made) != 2 || ended[1].Node != "n1" {
t.Errorf("logged = %+v, want the start and the end of netcheck's build", ended)
}
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)
}
}
said := b.awaitSaid(t, "the tag ask of novox/mesh-sdk feat/x at abc (feat-x) is not answered after")
if !strings.Contains(said, "the check's build of abc is not claimed yet") {
t.Errorf("the narration = %q, want what the ask waits on", said)
}
// And in the build log, under the builder's own key (the ruling of 13:17: build.said).
if !slices.ContainsFunc(b.logged(t, delivery.BuilderModule), func(e delivery.BuildLogEntry) bool {
return e.Event == delivery.BuildSaidEvent && e.Why == versions.SaidUnanswered && e.FP == key && e.Commit == "abc" &&
strings.Contains(e.Line, "the check's build of abc is not claimed yet")
}) {
t.Errorf("the build log of %s = %+v, want the unanswered tag ask said", delivery.BuilderModule, b.logged(t, delivery.BuilderModule))
}
if e, err := b.tagAsks.Get(b.ctx, key); err != nil {
t.Fatal(err)
} else if a, _ := versions.DecodeTagAsk(e.Value()); a.Answered.IsZero() || !a.Unanswered() {
t.Errorf("the tag ask = %+v, want kept and answered, no version yet", a)
}
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(50 * time.Millisecond) {
if _, err := b.tagBounds.Get(b.ctx, key); err == nil {
break
} else if time.Now().After(deadline) {
t.Fatalf("the bound is not written again after its narration: %v", err)
}
}
// Design 32, ADR 0615 §1: a package's name the indexer writes later reaches the builder by its packages' views family
// (never their bucket), and looks at the module again; the done bundles are kept, not numbered twice.
if _, err := b.packages.Put(b.ctx, "netcheck.0.1", []byte(`{"v":1,"name":"netcheck","package":"netcheck/lib","series":"0.1"}`)); err != nil {
t.Fatal(err)
}
b.awaitSaid(t, "names a package of netcheck")
if got := reg.numberedOf("netcheck/tools-go"); len(got) != 1 || got[0] != "0.1.5" {
t.Errorf("version.numbered of netcheck/tools-go = %v, want 0.1.5 once", got)
}
// DLV-B10: builder_publish_again after a published takes a new number, and says it numbered.
for _, a := range []string{"tools-go", "tools-typescript"} {
reg.published(t, "netcheck/"+a, "0.1.5", digestOf(a))
}
reg.awaitOf(t, "netcheck/tools-go", func(es []versions.Entry) bool { return len(es) == 1 && es[0].State == versions.Published })
key, again, err := build.NewPublishAgain("netcheck/tools-go", fp, "the holder lost it", "me@shanks", time.Now())
if err != nil {
t.Fatal(err)
}
if a := reg.answered(t, key, again); a.Refused || a.Do != versions.AgainRenumber || a.Version != "0.1.6" {
t.Errorf("builder_publish_again = %+v, want a new number, 0.1.6", a)
}
// Running-testing note (review round 1): two claims never get one number on the bus's own compare-and-set, as
// the builder's state versions keeps them: eight claims of one name at once, eight versions.
store := versions.Versions{Store: awaitedVersions{process.AwaitBucket(b.ctx, b.js, nil, versions.Bucket, "", t.Logf)}}
numbers := make(chan string, 8)
var claims sync.WaitGroup
for i := range 8 {
claims.Go(func() {
d, _, err := store.Number(b.ctx, versions.Claim{Name: "netcheck/race", Kind: versions.KindBundle, Fingerprint: fmt.Sprintf("f%d", i)})
if err != nil {
t.Error(err)
}
numbers <- d.Version
})
}
claims.Wait()
close(numbers)
got := map[string]bool{}
for v := range numbers {
got[v] = true
}
if len(got) != 8 {
t.Errorf("eight claims of one name at once were numbered %v, want eight versions", slices.Sorted(maps.Keys(got)))
}
if got := reg.numberedOf("netcheck/tools-go"); len(got) != 2 || got[1] != "0.1.6" {
t.Errorf("version.numbered of netcheck/tools-go = %v, want 0.1.6 said after 0.1.5", got)
}
// DLV-B10: a build module's bound running out fails the trunk build, which is the builder's finding build-failed; the
// fingerprint built again keeps its number, said with why, and its done build clears the finding.
release := make(chan struct{})
b.fake.hang.Store(&release)
third := "src2:" + strings.Repeat("4", 64)
b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, third, 3)
if f := b.awaitFP(t, "netcheck", third, isState(delivery.BuildFailed)); !strings.Contains(f.Error, "did not end within") {
t.Errorf("the failed build = %+v, want its bound run out", f)
}
failed := health.FindingKey(health.ScopeModule, "netcheck", BuildFailed)
reg.awaitFinding(t, failed, true)
b.fake.hang.Store(nil)
close(release)
b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, third, 3) // the indexer writes it again
b.awaitFP(t, "netcheck", third, isState(delivery.BuildDone))
b.awaitSaid(t, "kept: fingerprint "+third+" is numbered under it")
if !slices.ContainsFunc(b.logged(t, "netcheck"), func(e delivery.BuildLogEntry) bool {
return e.Event == delivery.BuildSaidEvent && e.Why == versions.SaidKept && e.FP == third && strings.Contains(e.Line, "kept: fingerprint "+third)
}) {
t.Errorf("the build log of netcheck says no version kept at %s", third)
}
reg.awaitFinding(t, failed, false)
// DLV-B4: the seats' refused and removed, heard on the builder's durable consumer, move a numbered version: a retention
// removal turns the entry removed with its reason, a refusal refused with the holder's reason and the bytes it refused.
reg.outcome(t, ociartifactregistry.SubjectRemoved, versions.Outcome{Name: "netcheck/tools-typescript", Version: "0.1.5", Kind: "bundle", Holder: "distribution", Reason: "retention"})
reg.outcome(t, ociartifactregistry.SubjectRefused, versions.Outcome{Name: "netcheck/tools-go", Version: "0.1.6", Digest: digestOf("other"), Kind: "bundle", Holder: "distribution", Reason: "the cap"})
reg.awaitOf(t, "netcheck/tools-typescript", func(es []versions.Entry) bool {
return slices.ContainsFunc(es, func(e versions.Entry) bool {
return e.Version == "0.1.5" && e.State == versions.Removed && e.Reason == "retention"
})
})
reg.awaitOf(t, "netcheck/tools-go", func(es []versions.Entry) bool {
return slices.ContainsFunc(es, func(e versions.Entry) bool {
return e.Version == "0.1.6" && e.State == versions.Refused && e.Reason == "the cap" && e.RefusedDigest == digestOf("other")
})
})
// DLV-B6: a branch's end (the forge bridge's merged) asks the seat to remove each version tagged for it, the reason
// tag-gone, and the seat's removed turns it removed.
tagged := versions.Entry{Name: "mesh-sdk/go", Kind: versions.KindPackage, Series: "0.1", Version: "0.1.15-feat-x.1", Tag: "feat-x", Module: "mesh-sdk",
Fingerprint: "src2:" + strings.Repeat("5", 64), Repository: "novox/mesh-sdk", Seat: gopackageregistry.Seat, State: versions.Published, At: time.Now().UTC()}
raw, _ := json.Marshal(tagged)
if _, err := reg.versions.Put(b.ctx, versions.Key(tagged.Name, tagged.Version), raw); err != nil {
t.Fatal(err)
}
if _, err := b.js.Publish(b.ctx, process.ModuleStreamSubject(delivery.ForgeBridgeModule, "forge"),
[]byte(`{"kind":"`+delivery.ForgeMerged+`","repository":"novox/mesh-sdk","branch":"feat/x"}`)); err != nil {
t.Fatal(err)
}
for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(50 * time.Millisecond) {
reg.mu.Lock()
asked := slices.ContainsFunc(reg.removes, func(a map[string]any) bool {
return a["name"] == tagged.Name && a["version"] == tagged.Version && a["reason"] == "tag-gone"
})
reg.mu.Unlock()
if asked {
break
}
if time.Now().After(deadline) {
t.Fatalf("the seat's remove of %s@%s was not asked at the branch's end: %v", tagged.Name, tagged.Version, reg.removes)
}
}
reg.outcome(t, gopackageregistry.SubjectRemoved, versions.Outcome{Name: tagged.Name, Version: tagged.Version, Kind: versions.KindPackage, Holder: "gitea", Reason: "tag-gone"})
reg.awaitOf(t, tagged.Name, func(es []versions.Entry) bool { return len(es) == 1 && es[0].State == versions.Removed })
}
// TestAnArchiveOnlyModuleAndARootModuleAreBuilt: 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). A module whose path is "."
// is given to the build tools as it is (1615/AC4). Two builders on two nodes start together on the archive's key: the
// second takes over the key the first holds building (ADR 0478 clause 6), and the key ends done, written once (1607/AC7).
// One table: three tests were merged, to pay for the enrolment's and to bring mesh-delivery's count down (DLV-B8).
func TestAnArchiveOnlyModuleAndARootModuleAreBuilt(t *testing.T) {
for _, c := range []struct {
name, path string
two bool
}{{"no toolchain", "modules/ca-trust", false}, {"root", delivery.RootPath, false}, {"two instances together", "modules/ca-trust", true}} {
t.Run(c.name, func(t *testing.T) {
b := startBed(t)
b.archive(t, "ca-trust", c.path, commit)
b.start(t, "n1", nil)
if c.two {
b.start(t, "n2", nil)
}
v := b.await(t, "ca-trust", isState(delivery.BuildDone))
if c.name == "no toolchain" && (len(v.Toolchains) != 0 || len(v.Artifacts) != 1 || strings.HasPrefix(v.Step, "one node (")) {
t.Errorf("build = %+v, want done naming no toolchain", v)
}
if c.two && (len(v.Artifacts) != 1 || v.Artifacts[0].Kind != "archive" || v.Artifacts[0].Digest != digestOf("files")) {
t.Errorf("build = %+v, want the archive's digest", v)
}
if n := b.fake.count("mesh-build-go.build_go_build"); c.two && 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)
}
})
}
}
// 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)
}
}
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 head of the candidate's source
// revision, newer than the module value's commit (issue 1684; the operator 2026-10-25: a value names the commit its
// fingerprint first appeared at).
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", Base: b.trunk(t, trunkHead), 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"] == trunkHead {
found = true
}
}
if !found {
t.Errorf("fetches = %v, want one of the candidate's head with its ref and the trunk head it was previewed onto, not the module value's 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", Base: b.trunk(t, trunkHead), 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
// A build module's tool, never the owners' read tools the builder's first views ask (DLV-M1).
s, err := sub.Subscribe("mesh.mod.*.call.>", func(m *nats.Msg) {
if strings.HasPrefix(m.Subject, "mesh.mod.mesh-build-") {
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())
}
}
// 1945 note 62, on the real bus: the stream read as builder_seat_queue reads it shows the Go ask waiting; the ask can
// still be answered (nothing was taken since); and a message dropped as an ask asked again is gone from the stream.
held := heldOnTheBus{b.js}
goAsk := delivery.StepAsk{ID: v.Asks[delivery.AskGo], Language: delivery.AskGo, Kind: delivery.AskDelivery}
stale := delivery.StepAsk{ID: "mvo53jcm70b5f7", Language: delivery.AskGo, Kind: delivery.AskDelivery}
subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskDelivery)
if _, err := b.js.Publish(b.ctx, subject, []byte("{}"), jetstream.WithMsgID(stale.ID)); err != nil {
t.Fatal(err)
}
queued := func() []build.QueuedState {
q, err := held.Queue(b.ctx, delivery.AskGo, 10)
if err != nil || len(q.Kinds) != 1 || q.Kinds[0].Kind != delivery.AskDelivery {
t.Fatalf("the Go seat's queue = %+v, %v; want its delivery asks", q, err)
}
return q.Kinds[0].Messages
}
if m := queued(); len(m) != 2 || m[0].Ask != goAsk.ID || !m[0].Waits || m[1].Ask != stale.ID {
t.Fatalf("the Go delivery queue = %+v, want the key's ask waiting, then the stale one", m)
}
if why, err := held.Gone(b.ctx, goAsk); why != "" || err != nil {
t.Errorf("Gone(the waiting ask) = %q, %v; want it answerable", why, err)
}
if err := held.Drop(b.ctx, stale); err != nil {
t.Fatal(err)
}
if m := queued(); len(m) != 1 || m[0].Ask != goAsk.ID {
t.Fatalf("after the drop the Go delivery queue = %+v, want the key's ask alone", m)
}
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)
}
}
// registry is the part of the bed design 51's builder cases need: the builder's states versions, findings, floor-waits and
// reasks, EVENTS with the seat oci-artifact-registry's outcomes and the builder's durable consumer of it, the builder's
// own events kept, and the floor's three sources served on the bus as their holders answer through the tool runner
// ({result}): the seat's versions, the forge bridge's bridge_tags and the indexer's indexer_requirements.
type registry struct {
b *bed
versions, findings, reasks jetstream.KeyValue
mu sync.Mutex
numbered []delivery.VersionNumbered
removes []map[string]any
}
func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *registry {
t.Helper()
r := &registry{b: b}
mk := func(name string) jetstream.KeyValue {
kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: name, History: 1})
if err != nil {
t.Fatal(err)
}
return kv
}
r.versions, r.findings, r.reasks = mk(versions.Bucket), mk(findingsBucket), mk(build.ReAsksBucket)
mk(versions.FloorWaitsBucket)
// EVENTS with two seats' outcomes, and the forge bridge's stream (a branch's end): each with the builder's durable consumer.
for stream, subjects := range map[string][]string{
eventsStream: {"mesh.mod.*.event.>", "mesh.seat.*.event.>"},
forgeStream: {process.ModuleStreamSubject(delivery.ForgeBridgeModule, ">")},
} {
if _, err := b.js.CreateOrUpdateStream(b.ctx, jetstream.StreamConfig{Name: stream, Subjects: subjects}); err != nil {
t.Fatal(err)
}
if _, err := b.js.CreateOrUpdateConsumer(b.ctx, stream, jetstream.ConsumerConfig{Durable: node + "_" + delivery.BuilderModule,
AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
t.Fatal(err)
}
}
// The Go seat's remove, as its holder answers it; each call is kept.
rm, err := b.js.Conn().Subscribe("mesh.seat."+gopackageregistry.Seat+".call.remove.>", func(m *nats.Msg) {
var args map[string]any
_ = json.Unmarshal(m.Data, &args)
r.mu.Lock()
r.removes = append(r.removes, args)
r.mu.Unlock()
_ = m.Respond([]byte(`{"result":{},"node":"ctl"}`))
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = rm.Unsubscribe() })
nc := b.js.Conn()
// As a build module hears it: on the contract's subject, read by the contract's decoder (BG2's late publish).
sub, err := nc.Subscribe(delivery.VersionNumberedSubject, func(m *nats.Msg) {
if nb, err := delivery.DecodeVersionNumbered(m.Data); err == nil {
r.mu.Lock()
r.numbered = append(r.numbered, nb)
r.mu.Unlock()
}
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
var listed []map[string]string
for _, v := range seatVersions {
listed = append(listed, map[string]string{"version": v})
}
for subject, answer := range map[string]any{
"mesh.seat." + ociartifactregistry.Seat + ".call.versions.>": map[string]any{"versions": listed},
"mesh.mod.mesh-delivery-forge-bridge.call.bridge_tags.>": map[string]any{"tags": []string{}},
"mesh.mod." + delivery.IndexerModule + ".call.indexer_requirements.>": map[string]any{"requirements": []string{}},
} {
reply, _ := json.Marshal(map[string]any{"result": answer, "node": "ctl"})
s, err := nc.Subscribe(subject, func(m *nats.Msg) { _ = m.Respond(reply) })
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = s.Unsubscribe() })
}
return r
}
// of is every entry versions holds of name.
func (r *registry) of(t *testing.T, name string) []versions.Entry {
t.Helper()
l, err := r.versions.ListKeysFiltered(r.b.ctx, versions.NameFilter(name))
if err != nil {
t.Fatal(err)
}
defer func() { _ = l.Stop() }()
var out []versions.Entry
for k := range l.Keys() {
e, err := r.versions.Get(r.b.ctx, k)
if err != nil {
t.Fatal(err)
}
var v versions.Entry
if err := json.Unmarshal(e.Value(), &v); err != nil {
t.Fatal(err)
}
out = append(out, v)
}
return out
}
// awaitOf reads name's entries until pred holds of them, within 20 s.
func (r *registry) awaitOf(t *testing.T, name string, pred func([]versions.Entry) bool) []versions.Entry {
t.Helper()
var es []versions.Entry
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if es = r.of(t, name); pred(es) {
return es
}
}
t.Fatalf("versions of %s = %+v within 20 s", name, es)
return es
}
// published says the artifact seat's published for name@version, as its holder does.
func (r *registry) published(t *testing.T, name, version, digest string) {
t.Helper()
r.outcome(t, ociartifactregistry.SubjectPublished, versions.Outcome{Name: name, Version: version, Digest: digest, Kind: "bundle", Holder: "distribution"})
}
// outcome says a seat's outcome on subject, as its holder does.
func (r *registry) outcome(t *testing.T, subject string, o versions.Outcome) {
t.Helper()
data, _ := json.Marshal(o)
msg := nats.NewMsg(subject)
msg.Header.Set("x-event-id", subject+"-"+o.Name+"-"+o.Version)
msg.Data = data
if _, err := r.b.js.PublishMsg(r.b.ctx, msg); err != nil {
t.Fatal(err)
}
}
// answered writes a reasks value and reads it until the builder answered it, within 20 s.
func (r *registry) answered(t *testing.T, key string, value []byte) build.ReAsk {
t.Helper()
if _, err := r.reasks.Put(r.b.ctx, key, value); err != nil {
t.Fatal(err)
}
var a build.ReAsk
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err := r.reasks.Get(r.b.ctx, key); err == nil && json.Unmarshal(e.Value(), &a) == nil && !a.Answered.IsZero() {
return a
}
}
t.Fatalf("reasks.%s = %+v, not answered within 20 s", key, a)
return a
}
// awaitFinding reads the builder's finding key until it is there (raised) or gone, within 20 s.
func (r *registry) awaitFinding(t *testing.T, key string, raised bool) {
t.Helper()
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if _, err := r.findings.Get(r.b.ctx, key); (err == nil) == raised {
return
}
}
t.Fatalf("the finding %s raised = %v within 20 s, want %v", key, !raised, raised)
}
// numberedOf are the version.numbered the builder said for name, in order.
func (r *registry) numberedOf(name string) []string {
r.mu.Lock()
defer r.mu.Unlock()
var out []string
for _, nb := range r.numbered {
if nb.Name == name {
out = append(out, nb.Version)
}
}
return out
}
Reference in New Issue View Git Blame Copy Permalink