jschoubben
3e8bd9b0de
mesh/delivery merged: source.main moved to novox/mesh-delivery@bcb5aafdbe80
mesh/check pass 3e8bd9b0def2 on bcb5aafdbe80 | unit 168 run, 472 kept | integration 178 run, 538 kept | composition passed on shanks
mesh-delivery #51's re-ask never fired for mesh-build-node.4e7b118e: messages the SEAT_NODE_BUILD_GO workers had given up on stay on the work queue, and each counted as waiting ahead, so the ask was never asked again (note 62). A message ahead now waits only when no node wrote its step and no ask of its kind asked after it was taken; the old message of an ask asked again is deleted from its stream; and builder_seat_queue (answered by the builder through its queues state, as only its account may read the stream) shows each message and whether it still waits.
875 lines
37 KiB
Go
875 lines
37 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"
|
|
|
|
"mesh-delivery-builder/internal/build"
|
|
)
|
|
|
|
// 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) }()
|
|
}
|
|
}
|
|
|
|
// 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
|
|
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)
|
|
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 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
|
|
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())
|
|
}
|
|
}
|
|
// 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)
|
|
}
|
|
}
|