Test the builder whole over the real bus with fake build modules, and look again at a value that arrives while its key is built (issue 1607, AC6, AC7)
This commit is contained in:
@@ -0,0 +1,366 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"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/delivery"
|
||||
)
|
||||
|
||||
// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
|
||||
// controller's user and the builder's own principals <node>.<module> (two instances, for the test of one build).
|
||||
const busConf = `
|
||||
listen: 127.0.0.1:-1
|
||||
jetstream { store_dir: %q }
|
||||
accounts {
|
||||
MESH {
|
||||
jetstream: enabled
|
||||
users = [
|
||||
{ user: "ctl", password: "pw" }
|
||||
{ user: "n1.mesh-delivery-builder", password: "pw" }
|
||||
{ user: "n2.mesh-delivery-builder", password: "pw" }
|
||||
]
|
||||
}
|
||||
}
|
||||
`
|
||||
|
||||
const (
|
||||
commit = "7a74cee83f88e6585a5511c3cc53bdf0f5303d20"
|
||||
fp = "src2:3333333333333333333333333333333333333333333333333333333333333333"
|
||||
)
|
||||
|
||||
// fakeBuildModules serve the tools of mesh-build-go and mesh-build-node as the real ones do (ADR 0469 clause 2): a call on
|
||||
// mesh.mod.<module>.call.<tool>.<node>.<caller> answers at once with the step and its key, and the step's end is written to
|
||||
// the key <node>.<id> of the module's steps bucket, some milliseconds later.
|
||||
type fakeBuildModules struct {
|
||||
js jetstream.JetStream
|
||||
ctx context.Context
|
||||
n atomic.Int64
|
||||
failBuild atomic.Bool
|
||||
mu sync.Mutex
|
||||
calls map[string]int // by `<module>.<tool>`
|
||||
}
|
||||
|
||||
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[:])
|
||||
}
|
||||
|
||||
func (f *fakeBuildModules) serve(t *testing.T, nc *nats.Conn, module string) {
|
||||
steps, err := f.js.CreateKeyValue(f.ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket(module), History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sub, err := nc.Subscribe("mesh.mod."+module+".call.>", func(m *nats.Msg) {
|
||||
tokens := strings.Split(m.Subject, ".") // mesh mod <module> call <tool> <node> <caller>
|
||||
tool, node := tokens[4], tokens[5]
|
||||
f.mu.Lock()
|
||||
f.calls[module+"."+tool]++
|
||||
f.mu.Unlock()
|
||||
var args map[string]any
|
||||
_ = json.Unmarshal(m.Data, &args)
|
||||
id := fmt.Sprintf("s%d", f.n.Add(1))
|
||||
key, _ := delivery.StepKey(node, id)
|
||||
put := func(s delivery.Step) {
|
||||
data, err := delivery.EncodeStep(s)
|
||||
if err == nil {
|
||||
_, err = steps.Put(f.ctx, key, data)
|
||||
}
|
||||
if err != nil {
|
||||
t.Errorf("writing step %s: %v", key, err)
|
||||
}
|
||||
}
|
||||
step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: time.Now().UTC()}
|
||||
put(step)
|
||||
answer, _ := json.Marshal(map[string]any{"result": map[string]any{"step": id, "key": key, "state": "running"}, "error": "", "node": node})
|
||||
_ = m.Respond(answer)
|
||||
go func() {
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
step.Ended = time.Now().UTC()
|
||||
switch {
|
||||
case strings.HasSuffix(tool, "_fetch"):
|
||||
step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"workspace":"w`+id+`","commit":"`+commit+`"}`)
|
||||
case f.failBuild.Load() && node == "n1":
|
||||
step.State, step.Reason = delivery.StepFailed, fmt.Sprintf("go build of %v: boom", args["artifact"])
|
||||
default:
|
||||
step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"digest":"`+digestOf(fmt.Sprint(args["artifact"]))+`"}`)
|
||||
}
|
||||
put(step)
|
||||
}()
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
}
|
||||
|
||||
type bed struct {
|
||||
ctx context.Context
|
||||
js jetstream.JetStream
|
||||
modules jetstream.KeyValue
|
||||
builds jetstream.KeyValue
|
||||
tools map[string]jetstream.KeyValue
|
||||
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 {
|
||||
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)
|
||||
mk(delivery.ManifestsBucket, delivery.ManifestsHistory)
|
||||
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
|
||||
b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}}
|
||||
b.fake.serve(t, ctl, "mesh-build-go")
|
||||
b.fake.serve(t, ctl, "mesh-build-node")
|
||||
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 {
|
||||
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: commit, FP: "src1:" + strings.Repeat("f", 64), FP2: fp,
|
||||
Manifest: digest, Path: path})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.modules.Put(b.ctx, name, data); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return digest
|
||||
}
|
||||
|
||||
// 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) }()
|
||||
t.Cleanup(func() { stop(); <-done })
|
||||
}
|
||||
|
||||
// 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 each of the two nodes.
|
||||
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 != "" {
|
||||
t.Errorf("artifact = %+v, want the digest every node built and no reference", 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": 2, "mesh-build-go.build_go_build": 2, "mesh-build-node.build_node_fetch": 2, "mesh-build-node.build_node_build": 2} {
|
||||
if got := b.fake.count(tool); got != want {
|
||||
t.Errorf("%s called %d times, want %d (once on each of two nodes)", tool, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestASecondBuilderInstanceDoesNotBuildTheSameSourceAgain: two builders on two nodes start together on one module key; the
|
||||
// fingerprint is built once.
|
||||
func TestASecondBuilderInstanceDoesNotBuildTheSameSourceAgain(t *testing.T) { // 1607/AC7
|
||||
b := startBed(t)
|
||||
b.module(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust")
|
||||
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 != 2 { // one instance built it, on two nodes
|
||||
t.Errorf("build_go_build was called %d times, want 2", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAFailedKeyIsBuiltAgainAsANewRevisionWhenAToolchainChanges: a build step fails on one node, so the key is failed
|
||||
// naming it; a toolchain report of a node then wakes the builder, which builds the key again as a new revision of the same
|
||||
// key, now done.
|
||||
func TestAFailedKeyIsBuiltAgainAsANewRevisionWhenAToolchainChanges(t *testing.T) { // 1607/AC6, AC7
|
||||
b := startBed(t)
|
||||
b.fake.failBuild.Store(true)
|
||||
b.module(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust")
|
||||
b.start(t, "n1", nil)
|
||||
v := b.await(t, "ca-trust", isState(delivery.BuildFailed))
|
||||
if !strings.Contains(v.Error, "n1") || !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 build module's recipe moved on both nodes: the toolchain's entries changed, twice (between the two the nodes disagree).
|
||||
b.report(t, "mesh-build-go", "n1", "go", "go1.27.1", "2")
|
||||
b.report(t, "mesh-build-go", "n2", "go", "go1.27.1", "2")
|
||||
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())
|
||||
}
|
||||
}
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
{
|
||||
"module": "ca-trust",
|
||||
"version": "1",
|
||||
"slug": "catrust",
|
||||
"capabilities": [
|
||||
"service-manager"
|
||||
],
|
||||
"requires": [
|
||||
"internal-acme-ca"
|
||||
],
|
||||
"seats": [
|
||||
{
|
||||
"name": "the-mesh-trust-anchor",
|
||||
"scope": "node"
|
||||
}
|
||||
],
|
||||
"claims": [
|
||||
{
|
||||
"name": "the-mesh-trust-anchor",
|
||||
"scope": "node"
|
||||
}
|
||||
],
|
||||
"resources": [
|
||||
{
|
||||
"id": "state",
|
||||
"type": "directory",
|
||||
"mode": "0700",
|
||||
"place": ".",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "anchor",
|
||||
"type": "file",
|
||||
"path": "${dir:state}/anchor",
|
||||
"mode": "0755",
|
||||
"template": "files/anchor"
|
||||
},
|
||||
{
|
||||
"id": "unit",
|
||||
"type": "file",
|
||||
"path": "/etc/systemd/system/mesh-ca-trust.service",
|
||||
"mode": "0644",
|
||||
"template": "files/mesh-ca-trust.service"
|
||||
},
|
||||
{
|
||||
"id": "trust",
|
||||
"type": "service",
|
||||
"unit": "mesh-ca-trust.service",
|
||||
"state": "running",
|
||||
"boot": "enabled",
|
||||
"restart-on": [
|
||||
"anchor",
|
||||
"unit"
|
||||
]
|
||||
}
|
||||
],
|
||||
"build": {
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "files",
|
||||
"kind": "archive",
|
||||
"from": "files"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
+57
@@ -0,0 +1,57 @@
|
||||
{
|
||||
"module": "netcheck",
|
||||
"version": "1",
|
||||
"emits": [
|
||||
"tunnel.endpoint-changed",
|
||||
"tunnel.handshake-stale",
|
||||
"tunnel.handshake-fresh"
|
||||
],
|
||||
"tools": [
|
||||
"netcheck_tunnel",
|
||||
"netcheck_tcp",
|
||||
"netcheck_dns",
|
||||
"netcheck_http",
|
||||
"netcheck_listening"
|
||||
],
|
||||
"replaces": {
|
||||
"netcheck_listening": [
|
||||
"ss -lunt",
|
||||
"ss -ltn",
|
||||
"netstat -lnt"
|
||||
],
|
||||
"netcheck_dns": [
|
||||
"dig",
|
||||
"nslookup"
|
||||
],
|
||||
"netcheck_tunnel": [
|
||||
"wg show",
|
||||
"wg show mesh0 dump"
|
||||
]
|
||||
},
|
||||
"build": {
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "tools-go",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "arch",
|
||||
"from": "cmd/netcheck",
|
||||
"binary": "netcheck",
|
||||
"loads": [
|
||||
"netcheck"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "tools-typescript",
|
||||
"kind": "bundle",
|
||||
"language": "typescript",
|
||||
"entrypoints": [
|
||||
"tools/index.js"
|
||||
],
|
||||
"loads": [
|
||||
"tools/index.js"
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -41,6 +41,7 @@ type Builder struct {
|
||||
mu sync.Mutex
|
||||
known map[string]delivery.Module // by modules key
|
||||
inflight map[string]bool // by build key
|
||||
again map[string]delivery.Module // by build key: the newest value that arrived while the key was being built
|
||||
running map[string]int // steps in flight, by `<build module>/<node>`
|
||||
plans map[string]planned // by manifest digest
|
||||
}
|
||||
@@ -57,7 +58,7 @@ func (b *Builder) init() {
|
||||
limit = 4
|
||||
}
|
||||
b.sem = make(chan struct{}, limit)
|
||||
b.known, b.inflight, b.running, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]int{}, map[string]planned{}
|
||||
b.known, b.inflight, b.again, b.running, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]int{}, map[string]planned{}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -147,6 +148,7 @@ func (b *Builder) consider(ctx context.Context, module string, m delivery.Module
|
||||
}
|
||||
b.mu.Lock()
|
||||
if b.inflight[key] {
|
||||
b.again[key] = m // looked at once more when the build ends, so a change is never lost
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
@@ -155,10 +157,21 @@ func (b *Builder) consider(ctx context.Context, module string, m delivery.Module
|
||||
b.wg.Add(1)
|
||||
go func() {
|
||||
defer b.wg.Done()
|
||||
defer func() { b.mu.Lock(); delete(b.inflight, key); b.mu.Unlock() }()
|
||||
b.sem <- struct{}{}
|
||||
defer func() { <-b.sem }()
|
||||
b.try(ctx, module, m)
|
||||
for {
|
||||
b.try(ctx, module, m)
|
||||
b.mu.Lock()
|
||||
next, more := b.again[key]
|
||||
delete(b.again, key)
|
||||
if !more || ctx.Err() != nil {
|
||||
delete(b.inflight, key)
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
b.mu.Unlock()
|
||||
m = next
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
|
||||
@@ -22,6 +22,9 @@ type buildModules struct {
|
||||
steps *fakeSteps
|
||||
digests map[string]string // `<node>/<artifact>` -> digest; the default is `digest`
|
||||
fails map[string]string // `<node>/<tool>` -> the reason the step fails with
|
||||
// hold, when set before the builds start, blocks every call until it is closed; entered is sent to as a call arrives.
|
||||
hold chan struct{}
|
||||
entered chan struct{}
|
||||
}
|
||||
|
||||
func newBuildModules() *buildModules {
|
||||
@@ -29,6 +32,13 @@ func newBuildModules() *buildModules {
|
||||
}
|
||||
|
||||
func (b *buildModules) CallOn(node, key string, body any) (json.RawMessage, error) {
|
||||
if b.hold != nil {
|
||||
select {
|
||||
case b.entered <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
<-b.hold
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.calls = append(b.calls, call{node, key, body})
|
||||
@@ -293,3 +303,20 @@ func TestTwoInstancesBuildOneFingerprintOnce(t *testing.T) { // 1607/AC7
|
||||
t.Errorf("%d builds, want 2: one instance built it", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAModulesValueThatArrivesWhileItsKeyIsBeingBuiltIsLookedAtAgainWhenItEnds(t *testing.T) { // 1607/AC6 (no event is dropped, no timer)
|
||||
r := newRig(t, goBundle)
|
||||
r.bm.fails["ace/build_go_build"] = "go build of tool: boom"
|
||||
r.bm.hold, r.bm.entered = make(chan struct{}), make(chan struct{}, 8)
|
||||
value := moduleValue(t, "modules/m")
|
||||
r.b.Module(context.Background(), "m", value)
|
||||
<-r.bm.entered // the first build is under way
|
||||
r.b.Module(context.Background(), "m", value)
|
||||
r.b.Reconsider(context.Background())
|
||||
delete(r.bm.fails, "ace/build_go_build") // what the event is about: the build can succeed now
|
||||
close(r.bm.hold)
|
||||
r.b.Wait()
|
||||
if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone {
|
||||
t.Errorf("build = %+v, want the failed key built again as a new revision after the first build ended", v)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user