mesh-host/internal/selfcompose/candidate_test.go

package selfcompose

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

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

	"github.com/novox/mesh-host/internal/agent"
)

// novox/hq ADR 0593 §7: the verb compose-candidate composes a check's candidate on this node in shadow. A candidate that
// changes a placed module composes it at the candidate's build, changed; one whose manifest the indexer refused is
// answered refused, naming the module and the judge's words, and the verdict is refused; a candidate the indexer does not
// hold is refused in words. Nothing is applied, nothing is written, no placement answer is handed, and the answer comes
// within the bound; a shadow not ready and an ended bound answer unready at once, which the checker takes as no answer.
func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
	b := newFakeBus()
	fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
	fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
	digest := delivery.ManifestDigest([]byte(lemurs))
	lemurs2 := strings.Replace(lemurs, `"version":"1"`, `"version":"2"`, 1)
	digest2 := delivery.ManifestDigest([]byte(lemurs2))
	for d, body := range map[string]string{digest: lemurs, digest2: lemurs2} {
		man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: body})
		key, _ := delivery.ManifestsKey(d)
		b.kv[delivery.ManifestsBucket+"/"+key] = man
	}
	// The indexer keys a candidate by its group alone (delivery.CandidatesKey, the branch name escaped, a slash
	// allowed), its heads hash inside the value; the checker asks by the composition key, `<group>.<heads hash>`.
	candidateOf := func(group, manifest string) string {
		head := delivery.Head{Repository: "novox/x", Number: 7, Head: "c0ffee" + manifest[len(manifest)-6:]}
		heads, _ := delivery.HeadsHash([]delivery.Head{head})
		pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"lemurs"},
			Fps: map[string]delivery.CandidateModule{"lemurs": {FP2: fpNew, Manifest: manifest}}}}
		raw, err := delivery.EncodeCandidate(delivery.Candidate{Group: group, HeadsHash: heads, Pulls: pulls,
			FPSet: delivery.FingerprintSet(pulls)})
		if err != nil {
			t.Fatal(err)
		}
		ckey, _ := delivery.CandidatesKey(group)
		b.kv[delivery.CandidatesBucket+"/"+ckey] = raw
		key, _ := delivery.CompositionKey(group, heads)
		return key
	}
	accepted, refusedKey := candidateOf("fix/940-no-retire-on-partial-read", digest2), ""
	{
		// The same change, its manifest refused by the indexer's judge.
		refusedMan := strings.Replace(lemurs2, `"version":"2"`, `"version":"3"`, 1)
		d3 := delivery.ManifestDigest([]byte(refusedMan))
		man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: refusedMan})
		key, _ := delivery.ManifestsKey(d3)
		b.kv[delivery.ManifestsBucket+"/"+key] = man
		vkey, _ := delivery.VerdictsKey(d3)
		v, err := delivery.EncodeManifestVerdict(delivery.ManifestVerdict{Module: "lemurs", State: delivery.VerdictRefused,
			Problems: []string{"a field the judge refuses"}, Judge: "src2:j", At: time.Now()})
		if err != nil {
			t.Fatal(err)
		}
		b.kv[delivery.VerdictsBucket+"/"+vkey] = v
		refusedKey = candidateOf("fix/941-refused", d3)
	}

	env := b.env()
	state := &fakeComposed{kv: map[string]ComposedEntry{}}
	env.ComposedState = state
	var applied atomic.Int64
	env.Refused = func(string, map[string]string) { applied.Add(1) }
	var handed atomic.Int64
	s := NewShadow(env, func(context.Context, map[string]delivery.CandidateAnswer) { handed.Add(1) })
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() { _ = s.Run(ctx) }()
	if a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted}); a.Verdict != delivery.ComposeCandidateUnready || a.Refused == "" {
		t.Errorf("a shadow not ready answered %+v", a)
	}
	place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
		Modules: []delivery.PlacedModule{{Module: "lemurs"}}})
	have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
	mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
	deadline := time.Now().Add(5 * time.Second)
	watching := func() int { b.mu.Lock(); defer b.mu.Unlock(); return len(b.watches) }
	for watching() < 6 && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	b.put(t, delivery.PlacementBucket, "placement.n1", place)
	b.put(t, delivery.HaveBucket, "n1.node", have)
	b.put(t, delivery.ModulesBucket, "lemurs", mod)
	for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
		b.send(bucket, agent.Update{Caught: true})
	}
	for !s.ready.Load() && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}

	a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted})
	if _, err := delivery.EncodeComposeCandidateAnswer(a); err != nil {
		t.Fatalf("the answer does not fit the seat's shape: %v (%+v)", err, a)
	}
	if a.Verdict != delivery.ComposeCandidateComposed || len(a.Modules) != 1 || a.Modules[0].Module != "lemurs" ||
		a.Modules[0].State != delivery.ModuleComposedState || a.Modules[0].FP != fpNew || !a.Modules[0].Changed ||
		a.Own.Module != delivery.HaveNodeWord || a.Took > delivery.ComposeAnswerBound.Milliseconds() {
		t.Errorf("an accepted candidate answered %+v", a)
	}
	r := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: refusedKey})
	if r.Verdict != delivery.ComposeCandidateRefused || len(r.Modules) != 1 || r.Modules[0].State != delivery.ModuleRefusedState ||
		!strings.Contains(r.Modules[0].Words, "a field the judge refuses") {
		t.Errorf("a candidate whose manifest the indexer refused answered %+v", r)
	}
	if _, err := delivery.EncodeComposeCandidateAnswer(r); err != nil {
		t.Errorf("the refused answer does not fit the seat's shape: %v", err)
	}
	missing, _ := delivery.CompositionKey("novox/x#8", "0123456789ab")
	if m := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: missing}); !strings.Contains(m.Refused, "holds no candidate") {
		t.Errorf("a candidate the indexer does not hold answered %+v", m)
	}
	// The group's candidate moved to other heads since the ask: the node composes not the one asked, and answers
	// unready (no answer), naming the heads it holds.
	moved, _ := delivery.CompositionKey("fix/940-no-retire-on-partial-read", "0123456789ab")
	if m := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: moved}); m.Verdict != delivery.ComposeCandidateUnready ||
		!strings.Contains(m.Refused, "heads") {
		t.Errorf("a candidate at other heads answered %+v", m)
	}
	ended, end := context.WithCancel(ctx)
	end()
	if e := s.ComposeCandidate(ended, delivery.ComposeCandidateAsk{Candidate: accepted}); e.Verdict != delivery.ComposeCandidateUnready {
		t.Errorf("an ended bound answered %+v", e)
	}
	if len(state.written()) != 0 || applied.Load() != 0 || handed.Load() != 0 {
		t.Errorf("composing candidates wrote %v, marked %d have(s), handed %d answer(s); want nothing", state.written(), applied.Load(), handed.Load())
	}
}