jschoubben
fd7d93333e
Declining an ask whose probe failed would turn a truly broken toolchain into a wait of up to the decline bound that then fails with an empty "needs a node with" reason. With the node-engine no longer deleting a running build and commands started in a directory that exists, the getwd cause is gone, so the guard's refusal stays as it was. The decline refactor goes with it.
384 lines
15 KiB
Go
384 lines
15 KiB
Go
package buildstep
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// steps records a steps state's puts: the last value of each key.
|
|
type steps struct {
|
|
mu sync.Mutex
|
|
keys map[string]delivery.Step
|
|
}
|
|
|
|
func newSteps() *steps { return &steps{keys: map[string]delivery.Step{}} }
|
|
|
|
func (s *steps) put(key string, value any) (uint64, error) {
|
|
raw, _ := value.(json.RawMessage)
|
|
st, err := delivery.DecodeStep(raw)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
s.mu.Lock()
|
|
s.keys[key] = st
|
|
s.mu.Unlock()
|
|
return 1, nil
|
|
}
|
|
|
|
func (s *steps) get(key string) (delivery.Step, bool) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
st, ok := s.keys[key]
|
|
return st, ok
|
|
}
|
|
|
|
func testAsk(kind string) delivery.StepAsk {
|
|
id := NewID()
|
|
a := delivery.StepAsk{ID: id, Kind: kind, Language: delivery.AskGo, Asker: "mesh-delivery-builder", At: time.Now()}
|
|
fetch := map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}
|
|
if kind == delivery.AskDelivery {
|
|
b := buildJobFixture
|
|
b.Fetch = fetch
|
|
a.Build = &b
|
|
} else {
|
|
a.Check = &delivery.CheckJob{Fetch: fetch, Levels: []string{delivery.LevelUnit}}
|
|
}
|
|
return a
|
|
}
|
|
|
|
var buildJobFixture = delivery.BuildJob{Module: "m", Path: ".", Artifacts: []delivery.AskArtifact{{Artifact: "bundle"}}}
|
|
|
|
func waitFor(t *testing.T, what string, cond func() bool) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for !cond() {
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("waited 2 s for %s", what)
|
|
}
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// 1887/A2a: given a check ask published before a delivery ask and one slot, the delivery ask runs first.
|
|
func TestPullTakesDeliveryBeforeCheck(t *testing.T) {
|
|
src, st := NewFakeAsks(), newSteps()
|
|
check, recheck, deliver := testAsk(delivery.AskCheck), testAsk(delivery.AskRecheck), testAsk(delivery.AskDelivery)
|
|
for _, a := range []delivery.StepAsk{recheck, check, deliver} {
|
|
if err := src.PublishAsk(a); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
var mu sync.Mutex
|
|
var order []string
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
|
|
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
|
|
mu.Lock()
|
|
order = append(order, a.Kind)
|
|
mu.Unlock()
|
|
return map[string]string{"ran": a.ID}, nil
|
|
}}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go p.Pull(ctx)
|
|
waitFor(t, "three asks acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if want := []string{delivery.AskDelivery, delivery.AskCheck, delivery.AskRecheck}; len(order) != 3 || order[0] != want[0] || order[1] != want[1] || order[2] != want[2] {
|
|
t.Fatalf("order = %v, want %v", order, want)
|
|
}
|
|
key, _ := delivery.AskStepKey("shanks", deliver.ID)
|
|
s, ok := st.get(key)
|
|
if !ok || s.State != delivery.StepPassed || string(s.Result) != `{"ran":"`+deliver.ID+`"}` {
|
|
t.Fatalf("the step %s = %+v (%v)", key, s, ok)
|
|
}
|
|
}
|
|
|
|
// 1887/A2a, 1887/AC-pull: given capacity 2 and three asks waiting, two run and the third waits, and it starts when
|
|
// one ends, with no timer.
|
|
func TestPullHoldsAtMostItsCapacity(t *testing.T) {
|
|
src, st := NewFakeAsks(), newSteps()
|
|
for i := 0; i < 3; i++ {
|
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
|
}
|
|
release := make(chan struct{})
|
|
var mu sync.Mutex
|
|
started := 0
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 2 },
|
|
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
|
|
mu.Lock()
|
|
started++
|
|
mu.Unlock()
|
|
<-release
|
|
return nil, nil
|
|
}}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go p.Pull(ctx)
|
|
waitFor(t, "two in hand", func() bool { return p.InHand() == 2 })
|
|
time.Sleep(30 * time.Millisecond) // three pull waits: the third must still wait
|
|
if p.InHand() != 2 || src.Waiting(delivery.AskDelivery) != 1 {
|
|
t.Fatalf("in hand %d, waiting %d; want 2 and 1", p.InHand(), src.Waiting(delivery.AskDelivery))
|
|
}
|
|
release <- struct{}{}
|
|
waitFor(t, "the third taken", func() bool { return src.Waiting(delivery.AskDelivery) == 0 })
|
|
// Taken from the source comes a moment before its run starts: wait for the start, not only the take.
|
|
waitFor(t, "the third started", func() bool { mu.Lock(); defer mu.Unlock(); return started == 3 })
|
|
close(release)
|
|
waitFor(t, "all acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
|
|
}
|
|
|
|
// An ask that is not one, or not this seat's, is dropped and runs nowhere. And issue 890/AC1: a check ask needing
|
|
// node-build-node is declined on a node that lacks it (its step refused as declined, naming what it lacks, the message
|
|
// dropped for the asker to put again), and runs on a node that has it.
|
|
func TestPullDropsWhatIsNoAsk(t *testing.T) {
|
|
src, st := NewFakeAsks(), newSteps()
|
|
src.Publish(delivery.AskDelivery, []byte(`{"v":1}`))
|
|
other := testAsk(delivery.AskDelivery)
|
|
other.Language = delivery.AskNode
|
|
_ = src.PublishAsk(other)
|
|
placed := testAsk(delivery.AskCheck)
|
|
placed.Needs = []string{delivery.NeedNodeBuildNode}
|
|
if err := src.PublishAsk(placed); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ran := false
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "ace", Put: st.put, Capacity: func() int { return 1 },
|
|
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { ran = true; return nil, nil }}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go p.Pull(ctx)
|
|
waitFor(t, "three termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 3 })
|
|
if ran {
|
|
t.Fatal("a wrong or a declined ask ran")
|
|
}
|
|
key, _ := delivery.AskStepKey("ace", placed.ID)
|
|
s, _ := st.get(key)
|
|
if d, ok := delivery.DeclinedOf(s); !ok || len(d.Lacks) != 1 || d.Lacks[0] != delivery.NeedNodeBuildNode || !strings.Contains(s.Reason, "lacks node-build-node") {
|
|
t.Fatalf("the declined step = %+v", s)
|
|
}
|
|
has, hasSteps := NewFakeAsks(), newSteps()
|
|
_ = has.PublishAsk(placed)
|
|
q := &Puller{Source: has, Language: delivery.AskGo, Node: "g14", Put: hasSteps.put, Capacity: func() int { return 1 },
|
|
PullWait: 10 * time.Millisecond, Has: func(n string) bool { return n == delivery.NeedNodeBuildNode },
|
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
|
go q.Pull(ctx)
|
|
waitFor(t, "the placed ask acked where its need is met", func() bool { a, _, _ := has.Fates(); return len(a) == 1 })
|
|
}
|
|
|
|
// load reads the recorded steps state's raw values: the pause key is not a step, so it is kept apart.
|
|
type kv struct {
|
|
mu sync.Mutex
|
|
vals map[string][]byte
|
|
}
|
|
|
|
func (k *kv) put(key string, value any) (uint64, error) {
|
|
raw, _ := value.(json.RawMessage)
|
|
k.mu.Lock()
|
|
defer k.mu.Unlock()
|
|
if k.vals == nil {
|
|
k.vals = map[string][]byte{}
|
|
}
|
|
k.vals[key] = append([]byte(nil), raw...)
|
|
return 1, nil
|
|
}
|
|
|
|
func (k *kv) load(key string) ([]byte, error) {
|
|
k.mu.Lock()
|
|
defer k.mu.Unlock()
|
|
return k.vals[key], nil
|
|
}
|
|
|
|
// 1887/A2b, 1887/AC-zero, 1887/AC-wake: at capacity 0 the puller takes nothing; a capacity raised takes effect at the
|
|
// next pull, or, with the config watch (g14 idle at capacity 2, 2026-10-25), at once when the value moves, with no
|
|
// pause, resume or step end to wake it; a refused value pulls nothing, and a watch that cannot start leaves the
|
|
// settings file's capacity.
|
|
func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
|
|
t.Run("settings file", func(t *testing.T) {
|
|
src, st := NewFakeAsks(), &kv{}
|
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
|
var mu sync.Mutex
|
|
capacity := 0
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "novox", Put: st.put, PullWait: 10 * time.Millisecond,
|
|
Capacity: func() int { mu.Lock(); defer mu.Unlock(); return capacity },
|
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go p.Pull(ctx)
|
|
time.Sleep(50 * time.Millisecond)
|
|
if src.Waiting(delivery.AskDelivery) != 1 {
|
|
t.Fatal("a node at capacity 0 took an ask")
|
|
}
|
|
mu.Lock()
|
|
capacity = 1
|
|
mu.Unlock()
|
|
p.Wake()
|
|
waitFor(t, "the ask taken at capacity 1", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
|
})
|
|
t.Run("config watch wakes", func(t *testing.T) {
|
|
src, st := NewFakeAsks(), newSteps()
|
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
|
w := &configWatch{value: "shanks=3,g14=0"}
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "g14", Put: st.put, PullWait: 10 * time.Millisecond,
|
|
Capacity: func() int { return 0 }, // the settings file, not refilled yet
|
|
WatchConfig: w.watch,
|
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go p.Pull(ctx)
|
|
waitFor(t, "the watch of the capacity", func() bool { w.mu.Lock(); defer w.mu.Unlock(); return w.handle != nil })
|
|
time.Sleep(30 * time.Millisecond)
|
|
if src.Waiting(delivery.AskDelivery) != 1 {
|
|
t.Fatal("a node at capacity 0 took an ask")
|
|
}
|
|
if w.asked != CapacitySetting {
|
|
t.Fatalf("the puller watches %q, want %q", w.asked, CapacitySetting)
|
|
}
|
|
w.set(t, "shanks=3,g14=2")
|
|
waitFor(t, "the ask taken once the capacity moved to 2", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
|
})
|
|
for _, c := range []struct {
|
|
name string
|
|
watch func(string, func(stdio.ProvisionChange) error) error
|
|
room bool
|
|
}{
|
|
{"refused value leaves no room", func(key string, handle func(stdio.ProvisionChange) error) error {
|
|
return handle(stdio.ProvisionChange{Kind: "config", Key: key, Refused: "the module is no longer on g14"})
|
|
}, false},
|
|
{"watch that cannot start keeps the settings file", func(string, func(stdio.ProvisionChange) error) error {
|
|
return errors.New("no runtime")
|
|
}, true},
|
|
} {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
p := &Puller{Source: NewFakeAsks(), Language: delivery.AskGo, Node: "g14", Put: newSteps().put,
|
|
Capacity: func() int { return 1 }, WatchConfig: c.watch,
|
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
|
p.watchCapacity()
|
|
if p.room() != c.room {
|
|
t.Fatalf("room = %v, want %v", p.room(), c.room)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// 1887/A2b: a paused puller takes nothing, its pause is kept on the pause key, and a puller started again on that state
|
|
// stays paused until resumed.
|
|
func TestPullPauseOutlivesARestart(t *testing.T) {
|
|
src, st, steps := NewFakeAsks(), &kv{}, &kv{}
|
|
run := func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load,
|
|
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run}
|
|
if err := p.Pause("the operator"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
key, _ := delivery.PauseKey("shanks")
|
|
raw, _ := st.load(key)
|
|
if v, err := delivery.DecodePause(raw); err != nil || !v.Paused || v.By != "the operator" {
|
|
t.Fatalf("the pause key holds %s (%v)", raw, err)
|
|
}
|
|
if raw, _ := steps.load(key); raw != nil {
|
|
t.Fatalf("the pause was written in the steps state, whose keys age out: %s", raw)
|
|
}
|
|
// The restart: a new puller on the same state.
|
|
again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load,
|
|
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run}
|
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go again.Pull(ctx)
|
|
time.Sleep(50 * time.Millisecond)
|
|
if !again.Paused() || src.Waiting(delivery.AskDelivery) != 1 {
|
|
t.Fatalf("restarted: paused %v, waiting %d; want paused and 1", again.Paused(), src.Waiting(delivery.AskDelivery))
|
|
}
|
|
if err := again.Resume("the operator"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
waitFor(t, "the ask taken after the resume", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
|
|
|
t.Run("a pause that cannot be read is said and pulled as not paused", func(t *testing.T) {
|
|
src := NewFakeAsks()
|
|
var said atomic.Value
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put,
|
|
Load: func(string) ([]byte, error) { return nil, errors.New("no runtime yet") },
|
|
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run,
|
|
Logf: func(f string, a ...any) {
|
|
if m := fmt.Sprintf(f, a...); strings.Contains(m, "cannot be read") {
|
|
said.Store(m)
|
|
}
|
|
}}
|
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go p.Pull(ctx)
|
|
waitFor(t, "the ask taken", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
|
if m, _ := said.Load().(string); !strings.Contains(m, "pulling as not paused") {
|
|
t.Fatalf("the unreadable pause was not said: %q", m)
|
|
}
|
|
})
|
|
}
|
|
|
|
// 1887/A2c: a long step tells the bus it is still being worked; a holder that stops mid-step neither ends nor
|
|
// acknowledges its ask, so the stream gives it to another node.
|
|
func TestPullHeartbeatAndStop(t *testing.T) {
|
|
src, st := NewFakeAsks(), newSteps()
|
|
a := testAsk(delivery.AskDelivery)
|
|
_ = src.PublishAsk(a)
|
|
running := make(chan struct{})
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
|
|
PullWait: 10 * time.Millisecond, Heartbeat: 5 * time.Millisecond,
|
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
|
|
close(running)
|
|
<-ctx.Done()
|
|
return nil, ctx.Err()
|
|
}}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
stopped := make(chan struct{})
|
|
go func() { p.Pull(ctx); close(stopped) }()
|
|
<-running
|
|
waitFor(t, "two heartbeats", func() bool { src.mu.Lock(); defer src.mu.Unlock(); return src.InProgress >= 2 })
|
|
cancel()
|
|
<-stopped
|
|
waitFor(t, "the step's goroutine gone", func() bool { return p.InHand() == 0 })
|
|
acked, termed, naked := src.Fates()
|
|
if len(acked)+len(termed)+len(naked) != 0 {
|
|
t.Fatalf("a stopped holder gave its ask a fate: acked %v termed %v naked %v", acked, termed, naked)
|
|
}
|
|
key, _ := delivery.AskStepKey("shanks", a.ID)
|
|
if s, ok := st.get(key); !ok || s.State != delivery.StepRunning {
|
|
t.Fatalf("the step %s = %+v, want left running", key, s)
|
|
}
|
|
}
|
|
|
|
// 1907: an ask taken past AskRuns (each run took its node down) is run no more: its step ends failed saying why, so the
|
|
// asker's key fails, and it is ended on the stream, not offered to another node.
|
|
func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) {
|
|
src, st := NewFakeAsks(), newSteps()
|
|
src.Delivered = delivery.AskRuns + 1
|
|
a := testAsk(delivery.AskDelivery)
|
|
_ = src.PublishAsk(a)
|
|
ran := false
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
|
|
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { ran = true; return nil, nil }}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go p.Pull(ctx)
|
|
waitFor(t, "the ask termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 1 })
|
|
key, _ := delivery.AskStepKey("shanks", a.ID)
|
|
s, ok := st.get(key)
|
|
if ran || !ok || s.State != delivery.StepFailed || !strings.Contains(s.Reason, "taken 3 times") || !strings.Contains(s.Reason, "poison") {
|
|
t.Fatalf("ran %v, step %+v", ran, s)
|
|
}
|
|
if acked, _, naked := src.Fates(); len(acked)+len(naked) != 0 {
|
|
t.Errorf("acked %v naked %v, want it termed alone", acked, naked)
|
|
}
|
|
}
|