jschoubben
71be15f530
With the toolchain agreed by majority, a node on the minority version can pull an ask for the agreed one. It now declines it (the issue 890 path) for the asker to put again, so it never fails a build. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
521 lines
22 KiB
Go
521 lines
22 KiB
Go
package buildstep
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"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() && s.Owner == Owner // 1890/AC5
|
|
})
|
|
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" || s.Owner != Owner {
|
|
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 })
|
|
// A run that finds its node is not on the agreed toolchain declines the ask after it began (ADR 0584 clause 2); a node on it builds.
|
|
sit, sitSteps := NewFakeAsks(), newSteps()
|
|
toolchain := testAsk(delivery.AskDelivery)
|
|
_ = sit.PublishAsk(toolchain)
|
|
s2 := &Puller{Source: sit, Language: delivery.AskGo, Node: "ace", Put: sitSteps.put, Capacity: func() int { return 1 }, PullWait: 10 * time.Millisecond,
|
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
|
|
return nil, &Declining{Text: "the ask is for 26.13.0 and this node builds with 26.12.1", Lacks: []string{"the agreed toolchain 26.13.0 (recipe 3)"}}
|
|
}}
|
|
go s2.Pull(ctx)
|
|
waitFor(t, "the ask termed on the node that sits out", func() bool { _, tm, _ := sit.Fates(); return len(tm) == 1 })
|
|
sk, _ := delivery.AskStepKey("ace", toolchain.ID)
|
|
ss, _ := sitSteps.get(sk)
|
|
if d, ok := delivery.DeclinedOf(ss); !ok || len(d.Lacks) != 1 || !strings.Contains(ss.Reason, "26.12.1") {
|
|
t.Fatalf("the sitting-out step = %+v", ss)
|
|
}
|
|
}
|
|
|
|
// 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: "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, "2")
|
|
waitFor(t, "the ask taken once the capacity moved to 2", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
|
})
|
|
t.Run("settings refill wakes a parked puller and retries a failed follow", func(t *testing.T) {
|
|
// 2026-10-28: a replaced builder held the old "shanks=2,..." in its settings file, its config follow timed out, and
|
|
// the puller parked at 0 with nothing to wake it when the node-engine refilled the file.
|
|
src, st := NewFakeAsks(), newSteps()
|
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
|
file := filepath.Join(t.TempDir(), "settings.json")
|
|
if err := os.WriteFile(file, []byte(`{"capacity":"shanks=2,ace=2"}`), 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var mu sync.Mutex
|
|
attempts := 0
|
|
watch := func(key string, handle func(stdio.ProvisionChange) error) error {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
attempts++
|
|
if attempts == 1 {
|
|
return errors.New("nats: timeout")
|
|
}
|
|
return nil // followed; no value handed yet
|
|
}
|
|
armed := make(chan struct{}, 8)
|
|
var parked atomic.Int32
|
|
parks := func() int32 { return parked.Load() }
|
|
logf := func(format string, args ...any) {
|
|
if strings.Contains(format, "parks") {
|
|
parked.Add(1)
|
|
}
|
|
}
|
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, PullWait: 10 * time.Millisecond,
|
|
SettingsFile: file, WatchConfig: watch, Logf: logf, watching: func() { armed <- struct{}{} },
|
|
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 parked puller to say why", func() bool { why, fault := p.CapacityNote(); return fault && why != "" })
|
|
if src.Waiting(delivery.AskDelivery) != 1 {
|
|
t.Fatal("a node whose capacity cannot be read took an ask")
|
|
}
|
|
<-armed // the file watch is in place
|
|
tmp := file + ".new"
|
|
if err := os.WriteFile(tmp, []byte(`{"capacity":2}`), 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := os.Rename(tmp, file); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
waitFor(t, "the ask taken once the file was refilled", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
|
// and the directory itself replaced (IN_IGNORED): the watch is armed again and a later refill still wakes the puller
|
|
dir := filepath.Dir(file)
|
|
before := parks()
|
|
if err := os.RemoveAll(dir); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := os.Mkdir(dir, 0o700); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
<-armed // armed again
|
|
waitFor(t, "the puller parked with no settings file", func() bool { return parks() > before })
|
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
|
if err := os.WriteFile(file, []byte(`{"capacity":2}`), 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
waitFor(t, "the second ask taken after the directory was replaced", func() bool { a, _, _ := src.Fates(); return len(a) == 2 })
|
|
waitFor(t, "the failed follow tried again", func() bool { mu.Lock(); defer mu.Unlock(); return attempts >= 2 })
|
|
})
|
|
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)
|
|
}
|
|
}
|