mesh-sdk/go/buildstep/pull_integration_test.go

//go:build integration

package buildstep

import (
	"context"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/testbus"
)

func askBus(t *testing.T, ackWait time.Duration) jetstream.JetStream {
	t.Helper()
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	if err := DeclareAskWorkers(context.Background(), js, delivery.AskGo, ackWait); err != nil {
		t.Fatal(err)
	}
	return js
}

func publishAsk(t *testing.T, js jetstream.JetStream, a delivery.StepAsk) {
	t.Helper()
	raw, err := delivery.EncodeStepAsk(a)
	if err != nil {
		t.Fatal(err)
	}
	subject, _ := delivery.AskSubject(a.Language, a.Kind)
	if _, err := js.Publish(context.Background(), subject, raw); err != nil {
		t.Fatal(err)
	}
}

// 1887/A2a, 1887/AC-order, 1887/AC-pull on the real server: a check ask published before a delivery ask runs after it;
// with capacity 2 and three asks two run, the third waits in the stream and starts when one ends.
func TestPullOrderAndCapacityOnTheBus(t *testing.T) {
	js := askBus(t, 30*time.Second)
	check, first, second := testAsk(delivery.AskCheck), testAsk(delivery.AskDelivery), testAsk(delivery.AskDelivery)
	publishAsk(t, js, check)
	publishAsk(t, js, first)
	publishAsk(t, js, second)
	release := make(chan struct{})
	var mu sync.Mutex
	var order []string
	st := newSteps()
	p := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "shanks", Put: st.put,
		Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond,
		Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
			mu.Lock()
			order = append(order, a.ID)
			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(500 * time.Millisecond)
	mu.Lock()
	if len(order) != 2 || order[0] != first.ID || order[1] != second.ID {
		t.Fatalf("ran %v, want the two delivery asks %s %s first", order, first.ID, second.ID)
	}
	mu.Unlock()
	release <- struct{}{}
	waitFor(t, "the check taken", func() bool { mu.Lock(); defer mu.Unlock(); return len(order) == 3 })
	close(release)
	waitFor(t, "every step ended", func() bool {
		for _, a := range []delivery.StepAsk{check, first, second} {
			key, _ := delivery.AskStepKey("shanks", a.ID)
			if s, ok := st.get(key); !ok || s.State != delivery.StepPassed {
				return false
			}
		}
		return true
	})
	info, err := js.Stream(ctx, "SEAT_NODE_BUILD_GO")
	if err != nil {
		t.Fatal(err)
	}
	waitFor(t, "the stream empty (every ask acknowledged)", func() bool {
		si, err := info.Info(ctx)
		return err == nil && si.State.Msgs == 0
	})
}

// 1887/A2c, ADR 0525 §6: a killed puller's ask is redelivered by the stream's acknowledgement wait to the other node,
// whose step key names the reference.
func TestPullRedeliversAKilledPullersAsk(t *testing.T) {
	js := askBus(t, 2*time.Second)
	a := testAsk(delivery.AskDelivery)
	publishAsk(t, js, a)
	st := newSteps()
	taken := make(chan struct{})
	dying := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "ace", Put: st.put,
		Capacity: func() int { return 1 }, PullWait: 200 * time.Millisecond, Heartbeat: time.Hour,
		Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
			close(taken)
			<-ctx.Done()
			return nil, ctx.Err()
		}}
	kill, killed := context.WithCancel(context.Background())
	go dying.Pull(kill)
	<-taken
	killed()
	other := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "shanks", Put: st.put,
		Capacity: func() int { return 1 }, PullWait: 200 * time.Millisecond,
		Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
			return delivery.BuildAnswer{Artifacts: []delivery.AskBuilt{{Artifact: "bundle", Digest: "sha256:00", Reference: "artifact-store://x"}}}, nil
		}}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go other.Pull(ctx)
	key, _ := delivery.AskStepKey("shanks", a.ID)
	deadline := time.Now().Add(10 * time.Second)
	for {
		if s, ok := st.get(key); ok && s.State == delivery.StepPassed {
			if want := `artifact-store://x`; !contains(string(s.Result), want) {
				t.Fatalf("the step's result %s names no %s", s.Result, want)
			}
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("the killed puller's ask was not taken by shanks within 10 s")
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func contains(s, sub string) bool { return strings.Contains(s, sub) }