263 lines
13 KiB
Go
263 lines
13 KiB
Go
package selfcompose
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
|
|
"github.com/novox/mesh-host/internal/agent"
|
|
"github.com/novox/mesh-host/internal/compose"
|
|
)
|
|
|
|
// 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)
|
|
}
|
|
// Issue 774: the input it sends the sandbox says how it composed the node (the builds), so the sandbox composes it
|
|
// through the same path.
|
|
sent := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted, Input: true})
|
|
var in compose.Input
|
|
if err := json.Unmarshal(sent.Input, &in); err != nil || in.ComposedHow == nil || in.ComposedHow.Node != "n1" ||
|
|
in.ComposedHow.Builds["lemurs"] != fpNew {
|
|
t.Errorf("the input sent for the sandbox does not say how the node composes: %+v, %v", in.ComposedHow, err)
|
|
}
|
|
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())
|
|
}
|
|
// Issue 1126, 2026-11-03: off the control-node an ask of compose-candidate was not answered within a minute, and said
|
|
// nothing of why. Each of its two compositions read every manifest in the mesh one by one, on copies of the shadow
|
|
// that kept nothing, so every ask read them all twice again. Each manifest is now read once, before either
|
|
// composition, into the shadow, and a later ask reads none; an ask whose read does not answer is answered unready at
|
|
// its bound, naming what the node waits on; and every answer is said in one line with what it waited on.
|
|
t.Run("each manifest is read once, and the answer comes within its bound", func(t *testing.T) {
|
|
fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
|
|
fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
|
|
lemurs2 := strings.Replace(lemurs, `"version":"1"`, `"version":"2"`, 1)
|
|
other := strings.NewReplacer(`"module":"lemurs"`, `"module":"other"`, "lemurs.service", "other.service").Replace(lemurs)
|
|
digest, digest2, digestOther := delivery.ManifestDigest([]byte(lemurs)), delivery.ManifestDigest([]byte(lemurs2)),
|
|
delivery.ManifestDigest([]byte(other))
|
|
otherKey, _ := delivery.ManifestsKey(digestOther)
|
|
|
|
// shadow is a ready shadow of n1 over b: lemurs placed on it, other placed nowhere (its manifest read for the mesh's
|
|
// catalogue alone), and a candidate laying lemurs at a new build; its clock moves a second at each reading.
|
|
shadow := func(b *fakeBus) (*Shadow, string) {
|
|
for d, body := range map[string]string{digest: lemurs, digest2: lemurs2, digestOther: other} {
|
|
man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "x", JSON: body})
|
|
key, _ := delivery.ManifestsKey(d)
|
|
b.kv[delivery.ManifestsBucket+"/"+key] = man
|
|
}
|
|
head := delivery.Head{Repository: "novox/x", Number: 7, Head: "c0ffee"}
|
|
heads, _ := delivery.HeadsHash([]delivery.Head{head})
|
|
pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"lemurs"},
|
|
Fps: map[string]delivery.CandidateModule{"lemurs": {FP2: fpNew, Manifest: digest2}}}}
|
|
raw, _ := delivery.EncodeCandidate(delivery.Candidate{Group: "fix/1126", HeadsHash: heads, Pulls: pulls,
|
|
FPSet: delivery.FingerprintSet(pulls)})
|
|
ckey, _ := delivery.CandidatesKey("fix/1126")
|
|
b.kv[delivery.CandidatesBucket+"/"+ckey] = raw
|
|
env := b.env()
|
|
var tick atomic.Int64
|
|
env.Now = func() time.Time { return time.Unix(0, 0).Add(time.Duration(tick.Add(1)) * time.Second) }
|
|
s := NewShadow(env, nil)
|
|
s.c.placements["n1"] = delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
|
|
s.c.records["n1"] = delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}
|
|
s.c.modules["lemurs"] = delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}
|
|
s.c.modules["other"] = delivery.Module{FP2: fp, Manifest: digestOther, Repository: "novox/x", Commit: "c"}
|
|
s.ready.Store(true)
|
|
key, _ := delivery.CompositionKey("fix/1126", heads)
|
|
return s, key
|
|
}
|
|
readsOf := func(b *fakeBus, key string) (n int) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
for _, r := range b.reads {
|
|
if r == [2]string{delivery.ManifestsBucket, key} {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
saidLast := func(b *fakeBus) string {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if len(b.said) == 0 {
|
|
return ""
|
|
}
|
|
return b.said[len(b.said)-1]
|
|
}
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
// stuck: the read of other's manifest does not answer until the ask's bound ends.
|
|
stuck bool
|
|
asks int
|
|
verdict string
|
|
reads int // of other's manifest, over every ask
|
|
words string // in the answer's words
|
|
said string // in the line said of the last answer
|
|
}{
|
|
{name: "one ask reads each manifest once", asks: 1, verdict: delivery.ComposeCandidateComposed, reads: 1,
|
|
said: "compose-candidate fix/1126."},
|
|
{name: "a later ask reads none again", asks: 2, verdict: delivery.ComposeCandidateComposed, reads: 1,
|
|
said: "read(s) waited on"},
|
|
{name: "a read that does not answer is named at the bound", stuck: true, asks: 1, verdict: delivery.ComposeCandidateUnready,
|
|
words: "waits on", said: "answered unready"},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
b := newFakeBus()
|
|
s, key := shadow(b)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
if tc.stuck {
|
|
get := s.c.env.Get
|
|
s.c.env.Get = func(rctx context.Context, bucket, k string) ([]byte, error) {
|
|
if bucket == delivery.ManifestsBucket && k == otherKey {
|
|
b.mu.Lock()
|
|
b.reads = append(b.reads, [2]string{bucket, k})
|
|
b.mu.Unlock()
|
|
cancel() // the ask's bound ends while the read waits
|
|
<-rctx.Done()
|
|
return nil, rctx.Err()
|
|
}
|
|
return get(rctx, bucket, k)
|
|
}
|
|
}
|
|
var a delivery.ComposeCandidateAnswer
|
|
for range tc.asks {
|
|
a = s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: key})
|
|
}
|
|
if a.Verdict != tc.verdict || !strings.Contains(a.Refused, tc.words) {
|
|
t.Errorf("answered %s %q; want %s with %q", a.Verdict, a.Refused, tc.verdict, tc.words)
|
|
}
|
|
if tc.stuck && !strings.Contains(a.Refused, "manifest") {
|
|
t.Errorf("answered %q; want the manifest it waits on named", a.Refused)
|
|
}
|
|
if got := readsOf(b, otherKey); tc.reads > 0 && got != tc.reads {
|
|
t.Errorf("other's manifest was read %d time(s); want %d", got, tc.reads)
|
|
}
|
|
if line := saidLast(b); !strings.Contains(line, tc.said) {
|
|
t.Errorf("said %q; want a line with %q", line, tc.said)
|
|
}
|
|
})
|
|
}
|
|
})
|
|
}
|