mesh-controller/internal/link/pull_buckets_test.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

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

	"github.com/novox/mesh-controller/internal/broker"
)

// The build agent reads and watches the pull requests' keys in the buckets the check request names (novox/hq ADR 0437
// §3, issue 1212): the controller knows the seat's claimants, the build agent holds no catalogue.

// 1212/AC4: a request naming no bucket reads mesh-delivery's, as before; the field is optional on the wire.
func TestARequestNamingNoPullsBucketUsesMeshDeliverys(t *testing.T) {
	if got := PullBucketsOf(&CheckRequest{}); !reflect.DeepEqual(got, []string{broker.PullsBucket}) {
		t.Errorf("buckets %v", got)
	}
	if got := PullBucketsOf(nil); !reflect.DeepEqual(got, []string{broker.PullsBucket}) {
		t.Errorf("buckets of no check %v", got)
	}
	named := &CheckRequest{Pulls: []string{"mesh-delivery-gate_pulls"}}
	if got := PullBucketsOf(named); !reflect.DeepEqual(got, named.Pulls) {
		t.Errorf("buckets %v", got)
	}
	raw := mustJSON(t, CheckRequest{Owner: "o"})
	if contains(string(raw), "pulls") {
		t.Errorf("an empty field is written: %s", raw)
	}
}

// opened is a set of buckets by name, each holding keys; a bucket not in it does not exist.
func opened(buckets map[string]map[string]string, opens *[]string) func(context.Context, string) (func(context.Context, string) ([]byte, error), error) {
	return func(_ context.Context, bucket string) (func(context.Context, string) ([]byte, error), error) {
		*opens = append(*opens, bucket)
		keys, ok := buckets[bucket]
		if !ok {
			return nil, jetstream.ErrBucketNotFound
		}
		return func(_ context.Context, key string) ([]byte, error) {
			v, ok := keys[key]
			if !ok {
				return nil, jetstream.ErrKeyNotFound
			}
			return []byte(v), nil
		}, nil
	}
}

// 1212/AC3: the buckets are read in turn, the first holding the key answers, a missing bucket is passed over, and
// none holding it is no key.
func TestThePullKeyIsReadFromTheNamedBucketsInTurn(t *testing.T) {
	key := PullKeyOf("novox", "a", 7)
	both := map[string]map[string]string{
		"mesh-delivery_pulls":      {key: `{"head":"old"}`},
		"mesh-delivery-gate_pulls": {key: `{"head":"new"}`},
	}
	var opens []string
	got, found, err := readPullKeyIn(t.Context(), []string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, opened(both, &opens), "novox", "a", 7)
	if err != nil || !found || got.Head != "new" || len(opens) != 1 {
		t.Errorf("first holding answers: %+v %v %v, opened %v", got, found, err, opens)
	}
	opens = nil
	only := map[string]map[string]string{"mesh-delivery_pulls": {key: `{"head":"old"}`}, "mesh-delivery-gate_pulls": {}}
	got, found, err = readPullKeyIn(t.Context(), []string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, opened(only, &opens), "novox", "a", 7)
	if err != nil || !found || got.Head != "old" || len(opens) != 2 {
		t.Errorf("a bucket without the key is passed over: %+v %v %v, opened %v", got, found, err, opens)
	}
	opens = nil
	got, found, err = readPullKeyIn(t.Context(), []string{"gone_pulls", "mesh-delivery_pulls"}, opened(only, &opens), "novox", "a", 7)
	if err != nil || !found || got.Head != "old" {
		t.Errorf("a missing bucket is passed over: %+v %v %v", got, found, err)
	}
	_, found, err = readPullKeyIn(t.Context(), []string{"gone_pulls", "mesh-delivery-gate_pulls"}, opened(only, &opens), "novox", "a", 8)
	if err != nil || found {
		t.Errorf("none holding it: found %v, %v", found, err)
	}
	boom := errors.New("bus away")
	_, _, err = readPullKeyIn(t.Context(), []string{"x"}, func(context.Context, string) (func(context.Context, string) ([]byte, error), error) {
		return nil, boom
	}, "novox", "a", 7)
	if !errors.Is(err, boom) {
		t.Errorf("a real failure is an error: %v", err)
	}
	// The watch takes the first named bucket that exists.
	opens = nil
	bucket, err := firstBucketThatExists(t.Context(), []string{"gone_pulls", "mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, func(_ context.Context, b string) error {
		opens = append(opens, b)
		if b == "gone_pulls" {
			return jetstream.ErrBucketNotFound
		}
		return nil
	})
	if err != nil || bucket != "mesh-delivery-gate_pulls" || len(opens) != 2 {
		t.Errorf("watch bucket %q %v, tried %v", bucket, err, opens)
	}
}

func mustJSON(t *testing.T, v any) string {
	t.Helper()
	raw, err := json.Marshal(v)
	if err != nil {
		t.Fatal(err)
	}
	return string(raw)
}

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

// fakeWatch is one bucket's watch of the key, fed by the test.
type fakeWatch struct {
	updates chan jetstream.KeyValueEntry
	done    chan struct{} // closed when the watch is stopped
	once    sync.Once
}

func newFakeWatch() *fakeWatch {
	return &fakeWatch{updates: make(chan jetstream.KeyValueEntry, 1), done: make(chan struct{})}
}

// handled returns once the watch has taken the entry fed before it and finished with it. The channel holds one entry,
// so the second filler (an entry the watch passes over) can only be sent after the first was taken, which the watch
// does only after it handled the entry before it. It needs a watch that is still listening: after a move it is not.
func (f *fakeWatch) handled(t *testing.T) {
	t.Helper()
	for range 2 {
		select {
		case f.updates <- fakeEntry{value: "not a key"}:
		case <-time.After(2 * time.Second):
			t.Fatal("the watch took no entry")
		}
	}
}

func (f *fakeWatch) Updates() <-chan jetstream.KeyValueEntry { return f.updates }
func (f *fakeWatch) Stop() error                             { f.once.Do(func() { close(f.done) }); return nil }

type fakeEntry struct {
	jetstream.KeyValueEntry
	value string
}

func (e fakeEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }
func (e fakeEntry) Value() []byte                   { return []byte(e.value) }

// 1325/AC2: a running check's watch hears the key move in the gate's bucket when it is there, in today's when only
// it is there, and in either when both are; a bucket that does not exist is passed over, and none is an error.
func TestARunningChecksWatchFindsItsKeyInTheGatesBucketOrTodays(t *testing.T) {
	req := BuildRequest{Ref: "old", Check: &CheckRequest{Owner: "novox", Repo: "a", Number: 7, Trunk: "t",
		Pulls: []string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}}}
	run := func(exist ...string) (map[string]*fakeWatch, chan PullKey, func(), error) {
		watches := map[string]*fakeWatch{}
		for _, b := range exist {
			watches[b] = newFakeWatch()
		}
		moved := make(chan PullKey, 2)
		stop, err := watchPullIn(t.Context(), req.Check.Pulls, func(_ context.Context, bucket string) (pullWatcher, error) {
			if w, ok := watches[bucket]; ok {
				return w, nil
			}
			return nil, jetstream.ErrBucketNotFound
		}, req, func(k PullKey) { moved <- k })
		return watches, moved, stop, err
	}
	wait := func(moved chan PullKey) (PullKey, bool) {
		select {
		case k := <-moved:
			return k, true
		case <-time.After(2 * time.Second):
			return PullKey{}, false
		}
	}
	for name, c := range map[string]struct {
		exist []string
		feed  string
	}{
		"only the gate's":  {[]string{"mesh-delivery-gate_pulls"}, "mesh-delivery-gate_pulls"},
		"only today's":     {[]string{"mesh-delivery_pulls"}, "mesh-delivery_pulls"},
		"both, the gate's": {[]string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, "mesh-delivery-gate_pulls"},
		"both, today's":    {[]string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, "mesh-delivery_pulls"},
	} {
		watches, moved, stop, err := run(c.exist...)
		if err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		watches[c.feed].updates <- fakeEntry{value: `{"head":"newer","trunk":"t"}`}
		if k, ok := wait(moved); !ok || k.Head != "newer" {
			t.Errorf("%s: the move was not heard (%+v, %v)", name, k, ok)
		}
		stop()
	}
	// A key that only gains its verdict is no move, and a move is told once.
	watches, moved, stop, err := run("mesh-delivery-gate_pulls", "mesh-delivery_pulls")
	if err != nil {
		t.Fatal(err)
	}
	watches["mesh-delivery-gate_pulls"].updates <- fakeEntry{value: `{"head":"old","trunk":"t","verdict":"pass"}`}
	watches["mesh-delivery-gate_pulls"].handled(t)
	select {
	case <-moved:
		t.Error("a verdict alone moved the check")
	default:
	}
	// Both buckets hear the move; the second watch is done (stopped) once it has handled its entry, and by then a
	// second telling would be in the channel.
	watches["mesh-delivery-gate_pulls"].updates <- fakeEntry{value: `{"head":"newer","trunk":"t"}`}
	watches["mesh-delivery_pulls"].updates <- fakeEntry{value: `{"head":"newer","trunk":"t"}`}
	wait(moved)
	for b, w := range watches {
		select {
		case <-w.done:
		case <-time.After(2 * time.Second):
			t.Errorf("the watch of %s did not end after the move", b)
		}
	}
	select {
	case <-moved:
		t.Error("a move was told twice")
	default:
	}
	stop()
	if _, _, _, err := run(); err == nil {
		t.Error("no bucket existing was not an error")
	}
}