mesh-sdk/go/delivery/buildrun/asks_test.go

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) }