package buildrun import ( "context" "encoding/json" "errors" "strings" "sync" "testing" "time" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" ) 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} var seen []delivery.Step // issue 1890: a waiter that asks to see each value of the step sees it step, err := k.Ask(WithStepSeen(context.Background(), func(s delivery.Step) { seen = append(seen, s) }), goAsk("mvl0frv18d1957")) if err != nil || step.Node != "ace" || len(f.Asks()) != 1 || len(seen) != 1 || seen[0].Node != "ace" { t.Fatalf("step %+v, %v, asks %d, seen %+v", step, err, len(f.Asks()), seen) } 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 round 1 of issue 734: a hearing the bus refuses at start is deaf and says so, never holds the start up, and // tries again on the next ask waited for; heard, it says so and hands the given-up asks on. fol := &refusingFollower{refuse: 1} var mu sync.Mutex var told []Hearing hearing := NewDeadLetterHearing(context.Background(), fol, []string{delivery.AskGo}, dead, nil, func(h Hearing) { mu.Lock(); told = append(told, h); mu.Unlock() }) hearing.Wait() if why := hearing.Deaf(); !strings.Contains(why, "SEAT_NODE_BUILD_GO: permissions violation") { t.Fatalf("a refused hearing says %q", why) } k = Asker{Publisher: g, Waiter: LostAskWaiter{Inner: g, Dead: dead, Hearing: hearing}, Bound: time.Minute} g.Write(delivery.Step{ID: "mvl0frv18d1973", Node: "g14", State: delivery.StepPassed}) if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1973")); err != nil { t.Fatal(err) } hearing.Wait() mu.Lock() if hearing.Deaf() != "" || len(told) != 2 || told[0].Heard || !told[1].Heard || fol.calls() != 2 { t.Fatalf("after an ask: deaf %q, told %+v, followed %d times", hearing.Deaf(), told, fol.calls()) } mu.Unlock() fol.heard(GivenUp{AskID: "mvl0frv18d1974", Deliveries: 5}) step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1974")) if err != nil || !IsLost(step) { t.Fatalf("an ask heard given up after the hearing came back: %+v, %v", step, err) } for _, row := range []struct { h *Hearing want string }{{nil, health.StateUnhealthy}, {&told[0], health.StateUnhealthy}, {&told[1], health.StateHealthy}} { var raw []byte if row.h != nil { raw, _ = json.Marshal(row.h) } if a := HearingHealth(raw); a.State != row.want || a.Why == "" { t.Errorf("health of %+v = %+v, want %s", row.h, a, row.want) } } } // 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) } } // refusingFollower refuses the first `refuse` follows, as the bus does before the asker's grant is composed. type refusingFollower struct { mu sync.Mutex refuse int n int hand func(GivenUp) } func (f *refusingFollower) Follow(_ context.Context, _ string, _ time.Time, heard func(GivenUp)) (func(), error) { f.mu.Lock() defer f.mu.Unlock() f.n++ if f.n <= f.refuse { return nil, errors.New("permissions violation") } f.hand = heard return func() {}, nil } func (f *refusingFollower) calls() int { f.mu.Lock(); defer f.mu.Unlock(); return f.n } func (f *refusingFollower) heard(g GivenUp) { f.mu.Lock(); h := f.hand; f.mu.Unlock(); h(g) }