jschoubben
02f50bd9d6
Since the build CPU cap (issue 1806) shanks' mesh-build-go kept capacity 3 inside a 12-core slice: three full checks got about 4 cores each and ran past the 30-minute step bound (issue 1890). The puller now judges by min(configured, floor(slice quota cores / cores-per-step)), the quota read live from its cgroup, says it on change and writes capacity.<node> to the pauses state. A run's phases (fetch and merge, digest, tests) are logged with the ask id and written to its running step key; Asker.Wait hands every step value to a WithStepSeen hook.
439 lines
18 KiB
Go
439 lines
18 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. Issue 1806's follow-up: the 2 is the CPU's, a configured 3 bounded by a 12-core slice at 6
|
|
// cores a step, said once in capacity.<node>; and issue 1890: a run's phase is written to its step key and logged with
|
|
// its ask id, and its end keeps the last phase.
|
|
func TestPullHoldsAtMostItsCapacity(t *testing.T) {
|
|
src, st := NewFakeAsks(), newSteps()
|
|
asks := []delivery.StepAsk{}
|
|
for i := 0; i < 3; i++ {
|
|
a := testAsk(delivery.AskDelivery)
|
|
asks = append(asks, a)
|
|
_ = src.PublishAsk(a)
|
|
}
|
|
release := make(chan struct{})
|
|
var mu sync.Mutex
|
|
started := 0
|
|
var said []string
|
|
capacities := map[string][]byte{}
|
|
quota := 1200
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 3 },
|
|
CoresPerStep: func() float64 { return 6 },
|
|
Quota: func() (int, string) { mu.Lock(); defer mu.Unlock(); return quota, "a test's slice" },
|
|
PausePut: func(key string, v any) (uint64, error) {
|
|
mu.Lock()
|
|
capacities[key] = v.(json.RawMessage)
|
|
mu.Unlock()
|
|
return 1, nil
|
|
},
|
|
Logf: func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() },
|
|
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
|
|
mu.Lock()
|
|
started++
|
|
mu.Unlock()
|
|
Phase(ctx, "unit: tests of 3 packages")
|
|
<-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))
|
|
}
|
|
c := p.CapacityDetail()
|
|
if c.Effective != 2 || c.Configured != 3 || c.ByCPU != 2 || c.QuotaCores != 12 || !strings.Contains(c.Why, "bounded by the CPU") {
|
|
t.Fatalf("the capacity judged by = %+v, want 2 of a configured 3 bounded by 12 cores / 6", c)
|
|
}
|
|
mu.Lock()
|
|
raw := capacities["capacity.shanks"]
|
|
capLines := 0
|
|
for _, l := range said {
|
|
if strings.Contains(l, "build capacity of shanks is 2, bounded by the CPU") {
|
|
capLines++
|
|
}
|
|
}
|
|
mu.Unlock()
|
|
if got, err := delivery.DecodeCapacity(raw); err != nil || got.Effective != 2 || capLines != 1 {
|
|
t.Fatalf("capacity.shanks = %s (%v), said %d times; want effective 2, said once", raw, err, capLines)
|
|
}
|
|
waitFor(t, "a phase on the step key", func() bool {
|
|
key, _ := delivery.AskStepKey("shanks", p.Running()[0].ID)
|
|
s, _ := st.get(key)
|
|
return s.State == delivery.StepRunning && s.Phase == "unit: tests of 3 packages" && !s.PhaseAt.IsZero()
|
|
})
|
|
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 })
|
|
for _, a := range asks {
|
|
key, _ := delivery.AskStepKey("shanks", a.ID)
|
|
if s, _ := st.get(key); s.State != delivery.StepPassed || s.Phase != "unit: tests of 3 packages" {
|
|
t.Errorf("the end of %s = %+v, want passed in its last phase", a.ID, s)
|
|
}
|
|
}
|
|
mu.Lock()
|
|
logged := strings.Join(said, "\n")
|
|
quota = 600 // the cap lowered: one step at 6 cores, followed at the next judgement
|
|
mu.Unlock()
|
|
if !strings.Contains(logged, "the delivery ask "+asks[0].ID+" on shanks: unit: tests of 3 packages") {
|
|
t.Errorf("no journal line names the ask's phase: %s", logged)
|
|
}
|
|
if c := p.CapacityDetail(); c.Effective != 1 {
|
|
t.Errorf("after the quota fell to 6 cores, capacity %+v, want 1", c)
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
}
|