jschoubben
b8158affa7
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: novox: not composed on novox: these assignments cannot be applied: | b8158affa7af on 27a956448c0a
An endpoint's subscription hands its asks over one at a time: two checks asking one node at once queued, and the second answered at about 100 s, past the checker's bound and the bus's one-minute reply window. compose-candidate is now answered in a goroutine of its own, at most four at once, one more refused in words. The new cases are rows of the existing tests (test budget). Issue 1126, review round 1. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
253 lines
13 KiB
Go
253 lines
13 KiB
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())
|
|
}
|
|
// 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)
|
|
}
|
|
})
|
|
}
|
|
})
|
|
}
|