jschoubben
c6ae0fc405
mesh-sdk's ComposeCandidateAnswer and ModuleComposed are now the generated nodeengine types (ADR 0613 §3, §5), so the input sent for the checker's sandbox is decoded once into the object the contract carries, its numbers kept as written (json.Number), and its size still counted from the bytes it was marshalled to. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
264 lines
13 KiB
Go
264 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
|
|
raw, _ := json.Marshal(sent.Input)
|
|
if err := json.Unmarshal(raw, &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)
|
|
}
|
|
})
|
|
}
|
|
})
|
|
}
|