mesh-sdk/go/buildstep/pull_test.go

package buildstep

import (
	"context"
	"encoding/json"
	"sync"
	"testing"
	"time"

	"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, Level: 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.
func TestPullHoldsAtMostItsCapacity(t *testing.T) {
	src, st := NewFakeAsks(), newSteps()
	for i := 0; i < 3; i++ {
		_ = src.PublishAsk(testAsk(delivery.AskDelivery))
	}
	release := make(chan struct{})
	var mu sync.Mutex
	started := 0
	p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 2 },
		PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
			mu.Lock()
			started++
			mu.Unlock()
			<-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))
	}
	release <- struct{}{}
	waitFor(t, "the third taken", func() bool { return src.Waiting(delivery.AskDelivery) == 0 })
	mu.Lock()
	if started != 3 {
		t.Errorf("started %d, want 3", started)
	}
	mu.Unlock()
	close(release)
	waitFor(t, "all acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
}

// An ask that is not one, or not this seat's, is dropped and runs nowhere.
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)
	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, "two termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 2 })
	if ran {
		t.Fatal("a wrong ask ran")
	}
}

// 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: at capacity 0 the puller takes nothing; a capacity raised takes effect at the next pull.
func TestPullAtCapacityZeroTakesNothing(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 })
}

// 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 := NewFakeAsks(), &kv{}
	run := func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }
	p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: 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)
	}
	// The restart: a new puller on the same state.
	again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: 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 })
}

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