jschoubben
d96b9fd660
The checker's try and the builder's lanes waited for ever on an ask given up during a bus reload: no build node writes a step for it. LostAskWaiter hands the asker a failed step on the node bus when DEAD_LETTERS keeps the ask's copy (HearDeadLetters, granted by DeadLetterReads), and under NotTakenBound when no node took the ask within the bound.
101 lines
5.1 KiB
Go
101 lines
5.1 KiB
Go
package buildrun
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
func goAsk(id string) delivery.StepAsk {
|
|
return delivery.StepAsk{ID: id, Kind: delivery.AskDelivery, Language: delivery.AskGo, Asker: "mesh-delivery-builder", At: time.Now(),
|
|
Build: &delivery.BuildJob{Module: "m", Path: ".", Fetch: map[string]any{"repository": "novox/x", "commit": strings.Repeat("a", 40)},
|
|
Artifacts: []delivery.AskArtifact{{Artifact: "program"}}}}
|
|
}
|
|
|
|
// 1887/A4a: an asker puts the ask once and answers the step a build node wrote for it, naming the node; one that did not
|
|
// pass is an error naming the node and why; a wait of an ask already put puts nothing.
|
|
func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
|
|
f := &FakeAsks{Answer: func(a delivery.StepAsk) delivery.Step {
|
|
return delivery.Step{ID: a.ID, Node: "ace", State: delivery.StepPassed, Result: []byte(`{"artifacts":[]}`)}
|
|
}}
|
|
k := Asker{Publisher: f, Waiter: f}
|
|
step, err := k.Ask(context.Background(), goAsk("mvl0frv18d1957"))
|
|
if err != nil || step.Node != "ace" || len(f.Asks()) != 1 {
|
|
t.Fatalf("step %+v, %v, asks %d", step, err, len(f.Asks()))
|
|
}
|
|
f.Answer = func(a delivery.StepAsk) delivery.Step {
|
|
return delivery.Step{ID: a.ID, Node: "shanks", State: delivery.StepRefused, Reason: "go 1.27 installed"}
|
|
}
|
|
if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") {
|
|
t.Fatalf("a refused step: %v", err)
|
|
}
|
|
// review round 2 item 1: an ask that waits in its stream for a free slot is not bounded; once a node writes its step
|
|
// running, the step is bounded from its start, and the error names the node.
|
|
g := &FakeAsks{}
|
|
k = Asker{Publisher: g, Waiter: g, Bound: 20 * time.Millisecond}
|
|
waiting, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
|
defer cancel()
|
|
if _, err := k.Wait(waiting, goAsk("mvl0frv18d1959")); err != context.DeadlineExceeded {
|
|
t.Fatalf("an ask no node took: %v, want it still waiting when ctx ended", err)
|
|
}
|
|
g.Write(delivery.Step{ID: "mvl0frv18d1961", Node: "ace", State: delivery.StepRunning, Started: time.Now()})
|
|
if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1961")); err == nil || !strings.Contains(err.Error(), "on ace did not end within") {
|
|
t.Fatalf("a step that never ends: %v", err)
|
|
}
|
|
if len(g.Asks()) != 0 {
|
|
t.Fatal("a wait put an ask")
|
|
}
|
|
// Issue 734: an ask the build seat gave up on (DEAD_LETTERS keeps its copy with the ask's id) ends its wait as failed
|
|
// with the words, though a node had taken it; under NotTakenBound one no node takes ends at the bound; a step written
|
|
// first is answered as the node wrote it.
|
|
dead := &DeadLetters{}
|
|
lw := LostAskWaiter{Inner: g, Dead: dead}
|
|
k = Asker{Publisher: g, Waiter: lw, Bound: time.Minute}
|
|
g.Write(delivery.Step{ID: "mvl0frv18d1970", Node: "ace", State: delivery.StepRunning, Started: time.Now()})
|
|
given, ok := GivenUpOf(map[string][]string{"Mesh-Dead-Msg-Id": {"mvl0frv18d1970"}, "Mesh-Dead-Deliveries": {"5"},
|
|
"Mesh-Dead-Stream": {"SEAT_NODE_BUILD_GO"}, "Mesh-Dead-Gave-Up": {"2026-10-26T12:26:40Z"}})
|
|
if !ok {
|
|
t.Fatal("a dead letter's headers were not read")
|
|
}
|
|
go dead.Heard(given)
|
|
step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1970"))
|
|
if err != nil || step.State != delivery.StepFailed || !IsLost(step) ||
|
|
!strings.Contains(step.Reason, LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z") {
|
|
t.Fatalf("a given-up ask: %+v, %v", step, err)
|
|
}
|
|
if _, err := k.Passed(step, nil); err == nil || !strings.Contains(err.Error(), "on "+LostNode+" failed: "+LostWords) {
|
|
t.Fatalf("a given-up ask's error: %v", err)
|
|
}
|
|
step, err = k.Wait(NotTakenBound(context.Background(), 20*time.Millisecond), goAsk("mvl0frv18d1971"))
|
|
if err != nil || !IsNotTaken(step) || IsLost(step) {
|
|
t.Fatalf("an ask no node took: %+v, %v", step, err)
|
|
}
|
|
g.Write(delivery.Step{ID: "mvl0frv18d1972", Node: "g14", State: delivery.StepPassed})
|
|
if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" {
|
|
t.Fatalf("a step written: %+v, %v", step, err)
|
|
}
|
|
}
|
|
|
|
// Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the
|
|
// new run's start, so its pass is answered after the first run's bound would have ended the wait.
|
|
func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) {
|
|
f := &FakeAsks{}
|
|
k := Asker{Publisher: f, Waiter: f, Bound: 60 * time.Millisecond}
|
|
const id = "mvl0frv18d1962"
|
|
first := time.Now()
|
|
f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: first})
|
|
go func() {
|
|
time.Sleep(40 * time.Millisecond)
|
|
f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: time.Now()})
|
|
time.Sleep(50 * time.Millisecond) // past the first run's bound, inside the second's
|
|
f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepPassed, Started: time.Now()})
|
|
}()
|
|
step, err := k.Wait(context.Background(), goAsk(id))
|
|
if err != nil || step.State != delivery.StepPassed {
|
|
t.Fatalf("step %+v, %v, want the redelivered run's pass", step, err)
|
|
}
|
|
}
|