Compose a catalogue pull request's own placement in the assigner's check and fail it on any placed node's unintended difference, refusal or silence past ten minutes, so a placement change is proven equal to what runs before it merges (issue 1126, ADR 0559 §5)
This commit is contained in:
@@ -9,6 +9,7 @@ import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -296,6 +297,10 @@ func (b *bed) manifest(t *testing.T, module, body string) string {
|
||||
// 1297/AC1: against the real bus, a fixture change touching one module on one machine is composed and answered within
|
||||
// seconds under the checker's key of the check, naming the machine composed and the mesh-sdk it composed with.
|
||||
func TestAChangeIsComposedOnTheBus(t *testing.T) {
|
||||
// The placement answers' bound (ADR 0559 §5), shortened before the program starts and put back after it stops.
|
||||
bound := assign.AnswerBound
|
||||
t.Cleanup(func() { assign.AnswerBound = bound })
|
||||
assign.AnswerBound = 1500 * time.Millisecond
|
||||
b := startBed(t, "")
|
||||
trunk := b.manifest(t, "web", `{"module":"web","version":"1"}`)
|
||||
changed := b.manifest(t, "web", `{"module":"web","version":"2"}`)
|
||||
@@ -347,6 +352,63 @@ func TestAChangeIsComposedOnTheBus(t *testing.T) {
|
||||
} else if info, err := st.Info(b.ctx); err != nil || info.State.Msgs != 0 {
|
||||
t.Errorf("a group without a mesh-sdk member put %d asks on the Go build seat (%v)", info.State.Msgs, err)
|
||||
}
|
||||
|
||||
// ADR 0559 §5 (issue 1126): a catalogue pull request that places g14 for the first time (the seed's case) is composed
|
||||
// with its candidate placement, touches g14, and waits for g14's answer at its head; g14 answering the same passes it.
|
||||
// A node that does not answer within the bound fails the check, named, at the bound's end with no other event; its later
|
||||
// answer at the same head passes the same heads' composition.
|
||||
record, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "g14"}})
|
||||
b.putJSON(t, delivery.HaveBucket, "g14.node", record)
|
||||
placed := func(group string, number int, head string) (string, uint64) {
|
||||
t.Helper()
|
||||
value, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: group,
|
||||
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: number, Head: head},
|
||||
Placement: &delivery.Placement{Node: "g14", Modules: []delivery.PlacedModule{{Module: "web"}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, _ := delivery.PlacementCandidateKey(group, "g14")
|
||||
rev, err := b.buckets[delivery.PlacementCandidatesBucket].Put(b.ctx, key, value)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pull := []delivery.CandidatePull{{Head: delivery.Head{Repository: delivery.CatalogueRepository, Number: number, Head: head}, Modules: []string{}}}
|
||||
hash, _ := delivery.HeadsHash([]delivery.Head{pull[0].Head})
|
||||
cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: group, HeadsHash: hash, Pulls: pull, FPSet: delivery.FingerprintSet(pull)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ckey, _ := delivery.CandidatesKey(group)
|
||||
b.putJSON(t, delivery.CandidatesBucket, ckey, cand)
|
||||
key, _ = delivery.CompositionKey(group, hash)
|
||||
return key, rev
|
||||
}
|
||||
answer := func(group, head string, rev uint64) {
|
||||
t.Helper()
|
||||
have, err := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "g14"},
|
||||
Candidates: map[string]delivery.CandidateAnswer{group: {Head: head, Revision: rev, Verdict: delivery.CandidateSame, Same: []string{"web"}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.putJSON(t, delivery.HaveBucket, "g14.node", have)
|
||||
}
|
||||
seedHead, lateHead := strings.Repeat("5", 40), strings.Repeat("6", 40)
|
||||
seed, seedRev := placed("feat/seed", 20, seedHead)
|
||||
waiting := b.verdict(t, seed, func(v delivery.CompositionVerdict) bool { return v.State == delivery.CompositionWaits })
|
||||
if !strings.Contains(waiting.Words, "g14") {
|
||||
t.Errorf("the seed's composition waits saying %q, want g14 named", waiting.Words)
|
||||
}
|
||||
answer("feat/seed", seedHead, seedRev)
|
||||
if v := b.verdict(t, seed, func(v delivery.CompositionVerdict) bool { return v.State == delivery.CompositionPass }); !slices.Contains(v.Machines, "g14") {
|
||||
t.Errorf("the seed's composition = %+v, want g14 composed", v)
|
||||
}
|
||||
late, lateRev := placed("feat/late", 21, lateHead)
|
||||
failed := b.verdict(t, late, func(v delivery.CompositionVerdict) bool { return v.State == delivery.CompositionFail })
|
||||
if failed.Machine != "g14" || !strings.Contains(failed.Words, delivery.CompositionUnanswered) {
|
||||
t.Errorf("a node silent past its bound: %+v", failed)
|
||||
}
|
||||
answer("feat/late", lateHead, lateRev)
|
||||
b.verdict(t, late, func(v delivery.CompositionVerdict) bool { return v.State == delivery.CompositionPass })
|
||||
}
|
||||
|
||||
// get waits for a key of one of the bed's buckets and answers its value.
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -60,6 +61,7 @@ var readBuckets = []string{
|
||||
delivery.ManifestsBucket,
|
||||
delivery.ModulesBucket,
|
||||
delivery.PlacementBucket,
|
||||
delivery.PlacementCandidatesBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.StepsBucket(goBuildModule),
|
||||
@@ -177,6 +179,7 @@ var watched = []string{
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ModulesBucket,
|
||||
delivery.PlacementBucket,
|
||||
delivery.PlacementCandidatesBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.AssignerAsksBucket,
|
||||
@@ -231,7 +234,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
value = e.Value()
|
||||
}
|
||||
w.put(name, e.Key(), value)
|
||||
w.putEntry(name, e.Key(), value, e.Revision(), e.Created())
|
||||
wake()
|
||||
}
|
||||
}
|
||||
@@ -273,7 +276,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
started := 0
|
||||
for _, name := range watched {
|
||||
if kvs[name] == nil && name == delivery.PlacementBucket {
|
||||
if kvs[name] == nil && (name == delivery.PlacementBucket || name == delivery.PlacementCandidatesBucket) {
|
||||
continue // not on the bus at this start (open said so)
|
||||
}
|
||||
started++
|
||||
@@ -291,15 +294,24 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
// The indexer's placement state arrives with the same delivery as this assigner (ADR 0549 §2): when it is not on the
|
||||
// bus at this start it is bound once the bus says it was created, and watched from then on, with no restart (issue
|
||||
// 1790's kind). Until then the controller's assignments alone name where modules run (ADR 0549 §5).
|
||||
placementAppeared := make(chan jetstream.KeyValue, 1)
|
||||
type appeared struct {
|
||||
name string
|
||||
kv jetstream.KeyValue
|
||||
}
|
||||
placementAppeared := make(chan appeared, 2)
|
||||
if kvs[delivery.PlacementBucket] == nil && js != nil {
|
||||
process.AwaitBucket(ctx, js, nc, delivery.PlacementBucket, "the controller's assignments alone are read",
|
||||
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- kv })
|
||||
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- appeared{delivery.PlacementBucket, kv} })
|
||||
}
|
||||
takePlacement := func(kv jetstream.KeyValue) error {
|
||||
kvs[delivery.PlacementBucket] = kv
|
||||
logger.Printf("the bucket %s is open: placement is read beside the controller's assignments", delivery.PlacementBucket)
|
||||
return watch(delivery.PlacementBucket)
|
||||
// So are the placement candidates (ADR 0559 §3), with the same delivery as the indexer that writes them.
|
||||
if kvs[delivery.PlacementCandidatesBucket] == nil && js != nil {
|
||||
process.AwaitBucket(ctx, js, nc, delivery.PlacementCandidatesBucket, "no change's placement is composed",
|
||||
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- appeared{delivery.PlacementCandidatesBucket, kv} })
|
||||
}
|
||||
takePlacement := func(a appeared) error {
|
||||
kvs[a.name] = a.kv
|
||||
logger.Printf("the bucket %s is open and watched", a.name)
|
||||
return watch(a.name)
|
||||
}
|
||||
for range started + 1 {
|
||||
select {
|
||||
@@ -311,6 +323,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
}
|
||||
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, said: map[string]bool{}, answeredAt: map[string]uint64{}}
|
||||
p.wakeAt = wakeAtVia(ctx, wake)
|
||||
p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
|
||||
if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
|
||||
p.acts = newPutActs(bus, wake, logger.Printf)
|
||||
@@ -334,6 +347,31 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
}
|
||||
|
||||
// wakeAtVia asks wake once at each time it is given, as a context's deadline (ADR 0437 §8: no timer), each time once.
|
||||
func wakeAtVia(ctx context.Context, wake func()) func(time.Time) {
|
||||
var mu sync.Mutex
|
||||
asked := map[int64]bool{}
|
||||
return func(at time.Time) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if asked[at.UnixNano()] {
|
||||
return
|
||||
}
|
||||
asked[at.UnixNano()] = true
|
||||
go func() {
|
||||
bound, cancel := context.WithDeadline(ctx, at)
|
||||
defer cancel()
|
||||
<-bound.Done()
|
||||
mu.Lock()
|
||||
delete(asked, at.UnixNano())
|
||||
mu.Unlock()
|
||||
if ctx.Err() == nil {
|
||||
wake()
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// PutsReopenEvery is how often, at most, a pass opens the puts' bucket again while it is not open.
|
||||
var PutsReopenEvery = time.Minute
|
||||
|
||||
|
||||
@@ -170,3 +170,5 @@ type kvEntry struct {
|
||||
func (e kvEntry) Key() string { return e.key }
|
||||
func (e kvEntry) Value() []byte { return e.value }
|
||||
func (e kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }
|
||||
func (e kvEntry) Revision() uint64 { return 1 }
|
||||
func (e kvEntry) Created() time.Time { return time.Time{} }
|
||||
|
||||
@@ -43,6 +43,9 @@ type passer struct {
|
||||
// nodeFailed is, by node and module, the words of the condition last raised for a failure that is the node's own (ADR 0557
|
||||
// §1, issue 1496): raised once per words, cleared once when it ends.
|
||||
nodeFailed map[string]string
|
||||
// wakeAt asks a pass at a time: the end of a placement candidate's answer bound (ADR 0559 §5), as a context's deadline
|
||||
// (ADR 0437 §8: no timer). Nil asks none.
|
||||
wakeAt func(at time.Time)
|
||||
}
|
||||
|
||||
func (p *passer) clock() time.Time {
|
||||
@@ -478,6 +481,9 @@ func (p *passer) compositions(ctx context.Context) {
|
||||
} else {
|
||||
v = assign.Compose(*mesh, ch, p.sdk, time.Since(began))
|
||||
}
|
||||
if ch.Placement != nil && v.State != delivery.CompositionWaits {
|
||||
v = p.answers(*mesh, ch, v)
|
||||
}
|
||||
}
|
||||
if before, err := delivery.DecodeCompositionVerdict(written[out]); err == nil {
|
||||
before.Took, before.V = v.Took, 0
|
||||
@@ -489,6 +495,40 @@ func (p *passer) compositions(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// answers judges every placed node's answer to the change's placement into v (ADR 0559 §5). While a node has not answered
|
||||
// within the bound the verdict waits, and the pass at the bound's end is asked (wakeAt), so a silent node is failed then,
|
||||
// named, and a later answer of the same head is judged on its own have's change.
|
||||
func (p *passer) answers(mesh assign.Mesh, ch assign.Change, v delivery.CompositionVerdict) delivery.CompositionVerdict {
|
||||
trunkPlaced := map[string]delivery.Placement{}
|
||||
for key, value := range p.w.read(delivery.PlacementBucket) {
|
||||
node, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if pl, err := delivery.DecodePlacement(value); err == nil && pl.Node == node {
|
||||
trunkPlaced[node] = pl
|
||||
}
|
||||
}
|
||||
fps := map[string]string{}
|
||||
for name, m := range p.w.trunk() {
|
||||
if m.FP2 != "" {
|
||||
fps[name] = m.FP2
|
||||
} else if m.FP != "" {
|
||||
fps[name] = m.FP
|
||||
}
|
||||
}
|
||||
intended, notes := assign.Intended(mesh.Input, trunkPlaced, ch.Placement, fps)
|
||||
a := assign.JudgeAnswers(ch.Placement, assign.PlacedNodes(trunkPlaced, ch.Placement), p.w.candidateAnswers(ch.Group), intended, p.w.online(), p.clock())
|
||||
v = assign.WithAnswers(v, a)
|
||||
if v.State != delivery.CompositionWaits {
|
||||
v.Warnings = append(v.Warnings, notes...)
|
||||
}
|
||||
if a.Waits && p.wakeAt != nil {
|
||||
p.wakeAt(a.Until)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// put writes one key of the assigner's own bucket; a write that fails is said and made again at the next pass.
|
||||
func (p *passer) put(ctx context.Context, bucket, key string, encode func() ([]byte, error)) {
|
||||
value, err := encode()
|
||||
|
||||
@@ -48,6 +48,14 @@ type world struct {
|
||||
healthy map[string][]string
|
||||
// changes counts, by bucket and key, the values this process has seen the key take: a value written again is another.
|
||||
changes map[string]uint64
|
||||
// meta is, by bucket and key, the revision and the time of the value held now, where the watch said them (putEntry).
|
||||
meta map[string]entryMeta
|
||||
}
|
||||
|
||||
// entryMeta is a value's revision in its bucket and when it was written.
|
||||
type entryMeta struct {
|
||||
revision uint64
|
||||
at time.Time
|
||||
}
|
||||
|
||||
// parsed is a manifest read by digest: its module and its manifest, or why it cannot be read.
|
||||
@@ -58,7 +66,27 @@ type parsed struct {
|
||||
}
|
||||
|
||||
func newWorld() *world {
|
||||
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{}}
|
||||
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{},
|
||||
meta: map[string]entryMeta{}}
|
||||
}
|
||||
|
||||
// putEntry is put, keeping the value's revision and the time it was written.
|
||||
func (w *world) putEntry(bucket, key string, value []byte, revision uint64, at time.Time) {
|
||||
w.put(bucket, key, value)
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if value == nil {
|
||||
delete(w.meta, bucket+"|"+key)
|
||||
return
|
||||
}
|
||||
w.meta[bucket+"|"+key] = entryMeta{revision: revision, at: at}
|
||||
}
|
||||
|
||||
// metaOf is the revision and time of the value a bucket holds at key, as the watch said them.
|
||||
func (w *world) metaOf(bucket, key string) entryMeta {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
return w.meta[bucket+"|"+key]
|
||||
}
|
||||
|
||||
// put takes a change of a watched bucket in: a value, or nil for a key deleted or purged.
|
||||
@@ -389,9 +417,92 @@ func (w *world) change(ctx context.Context, manifests getter, c delivery.Candida
|
||||
sort.Strings(ch.Twice[name])
|
||||
}
|
||||
}
|
||||
cp, waits := w.placementOf(ctx, manifests, c)
|
||||
if waits != "" {
|
||||
return ch, waits
|
||||
}
|
||||
ch.Placement = cp
|
||||
return ch, ""
|
||||
}
|
||||
|
||||
// placementOf is the change's own placement (ADR 0559 §5): the indexer's placement candidates of the group's catalogue
|
||||
// pull request, with the input the nodes compose from with them laid over the trunk's placement. None when the group has
|
||||
// no catalogue pull request or it changes no placement file. It waits while a key names another head than the group's.
|
||||
func (w *world) placementOf(ctx context.Context, manifests getter, c delivery.Candidate) (*assign.ChangePlacement, string) {
|
||||
var pull *delivery.CandidatePull
|
||||
for i := range c.Pulls {
|
||||
if delivery.SameRepository(c.Pulls[i].Repository, delivery.CatalogueRepository) {
|
||||
pull = &c.Pulls[i]
|
||||
}
|
||||
}
|
||||
if pull == nil {
|
||||
return nil, ""
|
||||
}
|
||||
cp := &assign.ChangePlacement{Placements: map[string]*delivery.Placement{}, Revisions: map[string]uint64{}, Head: pull.Head.Head}
|
||||
for key, value := range w.read(delivery.PlacementCandidatesBucket) {
|
||||
group, node, err := delivery.ParsePlacementCandidateKey(key)
|
||||
if err != nil || group != c.Group {
|
||||
continue
|
||||
}
|
||||
pc, err := delivery.DecodePlacementCandidate(value)
|
||||
if err != nil || pc.Pull.Number != pull.Number || !delivery.SameRepository(pc.Pull.Repository, pull.Repository) {
|
||||
continue
|
||||
}
|
||||
if pc.Pull.Head != pull.Head.Head {
|
||||
return nil, fmt.Sprintf("the indexer holds the placement candidates of %s#%d at %s, not yet at its head %s", pull.Repository, pull.Number, pc.Pull.Head, pull.Head.Head)
|
||||
}
|
||||
meta := w.metaOf(delivery.PlacementCandidatesBucket, key)
|
||||
cp.Placements[node], cp.Revisions[node] = pc.Placement, meta.revision
|
||||
if meta.at.After(cp.Written) {
|
||||
cp.Written = meta.at
|
||||
}
|
||||
}
|
||||
if len(cp.Placements) == 0 {
|
||||
return nil, ""
|
||||
}
|
||||
in, err := frombus.Input(ctx, worldReader{w: w, manifests: manifests}, frombus.Mesh{}, frombus.Scope{Every: true, Placements: cp.Placements})
|
||||
if err != nil {
|
||||
cp.Unread = err.Error()
|
||||
} else {
|
||||
cp.Input = &in
|
||||
}
|
||||
return cp, ""
|
||||
}
|
||||
|
||||
// candidateAnswers is, by node, each node's answer to the group's placement candidates in its `<node>.node` have.
|
||||
func (w *world) candidateAnswers(group string) map[string]*delivery.CandidateAnswer {
|
||||
out := map[string]*delivery.CandidateAnswer{}
|
||||
for key, value := range w.read(delivery.HaveBucket) {
|
||||
node, _, isNode, err := delivery.ParseHaveKey(key)
|
||||
if err != nil || !isNode {
|
||||
continue
|
||||
}
|
||||
h, err := delivery.DecodeHaveNode(value)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if a, ok := h.Candidates[group]; ok {
|
||||
out[node] = &a
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// online is, by node, whether the watchman holds the node online.
|
||||
func (w *world) online() map[string]bool {
|
||||
out := map[string]bool{}
|
||||
for key, value := range w.read(delivery.PresenceBucket) {
|
||||
node, _, _, err := delivery.ParsePresenceKey(key)
|
||||
if err != nil || node == "" {
|
||||
continue
|
||||
}
|
||||
if p, err := delivery.DecodePresence(value); err == nil {
|
||||
out[node] = p.Online
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// wantsOf is the wants the assigner wrote, by module then node.
|
||||
func (w *world) wantsOf() map[string]map[string]delivery.Want {
|
||||
out := map[string]map[string]delivery.Want{}
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-assigner
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -0,0 +1,242 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// A change's own placement (novox/hq ADR 0559 §5, issue 1126): a catalogue pull request that changes placement files has
|
||||
// the indexer's placement candidates, one key `<group>.<node>` for each node whose file at its head differs from the
|
||||
// trunk's. The check composes the change with those placements laid over the trunk's (frombus, Scope.Placements), counts
|
||||
// every such node among the machines it touches, and then reads each placed node's own answer in its `<node>.node`
|
||||
// have: it fails on what a node composes differently that the change does not intend, on a refusal, and on a node
|
||||
// placed at the head or on the trunk that has not answered at the head within AnswerBound of the keys being written.
|
||||
|
||||
// AnswerBound is how long the check waits for every placed node's answer after the candidate keys were written (ADR 0559
|
||||
// §5): a node asleep or away is named past it, never passed. A variable so a test on the bus can shorten it.
|
||||
var AnswerBound = 10 * time.Minute
|
||||
|
||||
// ChangePlacement is the change's own placement, read from the indexer's placement candidates of its group.
|
||||
type ChangePlacement struct {
|
||||
// Placements is by node the candidate placement; nil when the head removes the node's file.
|
||||
Placements map[string]*delivery.Placement
|
||||
// Revisions is by node the revision of its candidate key: the answer to it names that revision.
|
||||
Revisions map[string]uint64
|
||||
// Head is the catalogue pull request's head the candidates were read at, and Written when the newest key was written.
|
||||
Head string
|
||||
Written time.Time
|
||||
// Input is the trunk's composition input with Placements laid over it (frombus.Input, Scope.Placements), its catalogue
|
||||
// the trunk's; nil with Unread saying why when it could not be read.
|
||||
Input *compose.Input
|
||||
Unread string
|
||||
}
|
||||
|
||||
// Nodes is every node the change's placement names, sorted.
|
||||
func (p *ChangePlacement) Nodes() []string {
|
||||
if p == nil {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, len(p.Placements))
|
||||
for node := range p.Placements {
|
||||
out = append(out, node)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// Intended is, by node the trunk places, the modules whose composition the change's placement changes: those it adds,
|
||||
// removes or whose placed entry differs, and those whose composition (with the trunk's catalogue, the same stand-ins
|
||||
// either side) differs between the trunk's input and the change's (delivery.CompareComposed), the node's own named
|
||||
// "node". A node the trunk does not place yet has none (ADR 0559 §5: at the seed every node must answer the same). fps
|
||||
// gives the build each module is composed at, the trunk's. A node whose compositions cannot be made either side keeps
|
||||
// the modules its placement entries name, and is said in notes.
|
||||
func Intended(trunk *compose.Input, trunkPlaced map[string]delivery.Placement, cp *ChangePlacement, fps map[string]string) (intended map[string]map[string]bool, notes []string) {
|
||||
intended = map[string]map[string]bool{}
|
||||
if cp == nil {
|
||||
return intended, nil
|
||||
}
|
||||
for _, node := range cp.Nodes() {
|
||||
before, placed := trunkPlaced[node]
|
||||
if !placed {
|
||||
continue
|
||||
}
|
||||
set := map[string]bool{}
|
||||
after := cp.Placements[node]
|
||||
var afterModules []delivery.PlacedModule
|
||||
if after != nil {
|
||||
afterModules = after.Modules
|
||||
}
|
||||
entries := map[string]delivery.PlacedModule{}
|
||||
for _, m := range before.Modules {
|
||||
entries[m.Module] = m
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, m := range afterModules {
|
||||
seen[m.Module] = true
|
||||
if e, ok := entries[m.Module]; !ok || !reflect.DeepEqual(e, m) {
|
||||
set[m.Module] = true
|
||||
}
|
||||
}
|
||||
for name := range entries {
|
||||
if !seen[name] {
|
||||
set[name] = true
|
||||
}
|
||||
}
|
||||
if trunk != nil && cp.Input != nil {
|
||||
was, _, _, errB := compose.CompositionsShadow(*trunk, node, fps)
|
||||
now, _, _, errA := compose.CompositionsShadow(*cp.Input, node, fps)
|
||||
if errA == nil && errB == nil {
|
||||
running := make([]delivery.Composed, 0, len(was))
|
||||
for _, c := range was {
|
||||
running = append(running, c)
|
||||
}
|
||||
cmp := delivery.CompareComposed(now, running)
|
||||
for _, d := range cmp.Differs {
|
||||
set[d.Module] = true
|
||||
}
|
||||
for _, d := range cmp.Added {
|
||||
set[d.Module] = true
|
||||
}
|
||||
for _, m := range cmp.Removed {
|
||||
set[m] = true
|
||||
}
|
||||
} else {
|
||||
notes = append(notes, fmt.Sprintf("what the change intends on %s is read from its placement file alone: its compositions cannot be made (trunk: %v; change: %v)", node, errB, errA))
|
||||
}
|
||||
}
|
||||
intended[node] = set
|
||||
}
|
||||
return intended, notes
|
||||
}
|
||||
|
||||
// Answered is what the placed nodes answered: the failures to name, the nodes that have not answered, and the notes.
|
||||
type Answered struct {
|
||||
Fails []string
|
||||
Silent []string
|
||||
Machine string
|
||||
Waits bool
|
||||
Until time.Time
|
||||
Warnings []string
|
||||
}
|
||||
|
||||
// JudgeAnswers reads each placed node's answer to the change's placement (ADR 0559 §5). placed is every node placed at
|
||||
// the head or on the trunk; answers is by node its HaveNode.Candidates entry for the group, nil when it has none; online
|
||||
// says whether the watchman holds the node online. An answer counts when it names the candidates' head and, for a node
|
||||
// with a key of its own, that key's revision. It fails on a refusal, on a module that differs, is added or is removed
|
||||
// that intended does not name, and, from Written plus AnswerBound on, on every node with no answer that counts, named,
|
||||
// asleep or not; before that the check waits for them.
|
||||
func JudgeAnswers(cp *ChangePlacement, placed []string, answers map[string]*delivery.CandidateAnswer, intended map[string]map[string]bool, online map[string]bool, now time.Time) Answered {
|
||||
var out Answered
|
||||
if cp == nil {
|
||||
return out
|
||||
}
|
||||
failAt := func(node, words string) {
|
||||
if out.Machine == "" {
|
||||
out.Machine = node
|
||||
}
|
||||
out.Fails = append(out.Fails, words)
|
||||
}
|
||||
for _, node := range placed {
|
||||
a := answers[node]
|
||||
rev, keyed := cp.Revisions[node]
|
||||
if a == nil || a.Head != cp.Head || (keyed && a.Revision != rev) {
|
||||
out.Silent = append(out.Silent, node)
|
||||
continue
|
||||
}
|
||||
if a.Refused != "" {
|
||||
failAt(node, fmt.Sprintf("%s cannot compose the candidate placement: %s", node, a.Refused))
|
||||
continue
|
||||
}
|
||||
for _, r := range a.Refusals {
|
||||
failAt(node, fmt.Sprintf("%s refuses the candidate placement for %s: %s", node, r.Module, r.Why))
|
||||
}
|
||||
want := intended[node]
|
||||
for _, d := range a.Differs {
|
||||
if !want[d.Module] {
|
||||
failAt(node, fmt.Sprintf("%s composes %s differently with the candidate placement, which the change does not intend: %s", node, d.Module, d.Why))
|
||||
}
|
||||
}
|
||||
for _, d := range a.Added {
|
||||
if !want[d.Module] {
|
||||
failAt(node, fmt.Sprintf("%s would run %s with the candidate placement, which the change does not intend", node, d.Module))
|
||||
}
|
||||
}
|
||||
for _, m := range a.Removed {
|
||||
if !want[m] {
|
||||
failAt(node, fmt.Sprintf("%s would stop running %s with the candidate placement, which the change does not intend", node, m))
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(out.Silent) == 0 {
|
||||
return out
|
||||
}
|
||||
out.Until = cp.Written.Add(AnswerBound)
|
||||
named := make([]string, len(out.Silent))
|
||||
for i, node := range out.Silent {
|
||||
named[i] = node
|
||||
if !online[node] {
|
||||
named[i] = node + " (offline)"
|
||||
}
|
||||
}
|
||||
if now.Before(out.Until) {
|
||||
out.Waits = true
|
||||
out.Warnings = append(out.Warnings, fmt.Sprintf("waits for the answers of %s to the candidate placement at %s, until %s",
|
||||
strings.Join(named, ", "), short(cp.Head), out.Until.UTC().Format(time.RFC3339)))
|
||||
return out
|
||||
}
|
||||
for _, node := range out.Silent {
|
||||
if out.Machine == "" {
|
||||
out.Machine = node
|
||||
}
|
||||
}
|
||||
out.Fails = append(out.Fails, fmt.Sprintf("%s %s (%s after the candidate keys were written at %s)",
|
||||
strings.Join(named, ", "), delivery.CompositionUnanswered, AnswerBound, cp.Written.UTC().Format(time.RFC3339)))
|
||||
return out
|
||||
}
|
||||
|
||||
// PlacedNodes is every node placed at the head or on the trunk: the trunk's placed nodes with the change's placement laid
|
||||
// over them (a node whose file the head removes stays, as the trunk places it), sorted.
|
||||
func PlacedNodes(trunkPlaced map[string]delivery.Placement, cp *ChangePlacement) []string {
|
||||
set := map[string]bool{}
|
||||
for node := range trunkPlaced {
|
||||
set[node] = true
|
||||
}
|
||||
if cp != nil {
|
||||
for node, p := range cp.Placements {
|
||||
if p != nil {
|
||||
set[node] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
out := make([]string, 0, len(set))
|
||||
for node := range set {
|
||||
out = append(out, node)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// WithAnswers is v with the placed nodes' answers judged in: a fail adds its failures (the words of nodes not heard end
|
||||
// with delivery.CompositionUnanswered, so the checker takes a later composition of the same heads in when they answer), a
|
||||
// wait makes a passing verdict wait (a composition that waits composed no machine), and a failing one stands.
|
||||
func WithAnswers(v delivery.CompositionVerdict, a Answered) delivery.CompositionVerdict {
|
||||
if len(a.Fails) > 0 {
|
||||
if v.State == delivery.CompositionFail {
|
||||
v.Words += "; " + strings.Join(a.Fails, "; ")
|
||||
} else {
|
||||
v.State, v.Words, v.Machine = delivery.CompositionFail, strings.Join(a.Fails, "; "), a.Machine
|
||||
}
|
||||
return v
|
||||
}
|
||||
if a.Waits && v.State == delivery.CompositionPass {
|
||||
return delivery.CompositionVerdict{Group: v.Group, Heads: v.Heads, State: delivery.CompositionWaits, SDK: v.SDK, Took: v.Took,
|
||||
Words: strings.Join(a.Warnings, "; ")}
|
||||
}
|
||||
return v
|
||||
}
|
||||
@@ -59,12 +59,15 @@ type Change struct {
|
||||
// Members the head of each member by repository: what a composer at the group's mesh-sdk reads (issues 1622, 1623).
|
||||
Digests map[string]string
|
||||
Members map[string]string
|
||||
// Placement is the change's own placement, read from the indexer's placement candidates of the group (ADR 0559 §5);
|
||||
// nil when the group changes no placement file.
|
||||
Placement *ChangePlacement
|
||||
}
|
||||
|
||||
// Touched is the machines a change touches (ADR 0446 clause 3): those running a module the change moves, adds or
|
||||
// removes, those running a consumer of a provision such a module provides (before or after), and those running a
|
||||
// provider of a provision such a module requires after the change. A module the change adds that is assigned nowhere
|
||||
// touches no machine.
|
||||
// touches no machine. Every node whose placement the change changes is touched too (ADR 0559 §5, changing ADR 0446 §3).
|
||||
func Touched(m Mesh, c Change) []string {
|
||||
moved := map[string]bool{}
|
||||
for name := range c.Manifests {
|
||||
@@ -98,6 +101,11 @@ func Touched(m Mesh, c Change) []string {
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, node := range c.Placement.Nodes() {
|
||||
if !slices.Contains(out, node) {
|
||||
out = append(out, node)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
@@ -170,6 +178,25 @@ func Compose(m Mesh, c Change, sdk string, took time.Duration) delivery.Composit
|
||||
delete(changed, name)
|
||||
}
|
||||
after := input(changed, m.Assigned)
|
||||
if c.Placement != nil {
|
||||
// A change of placement is composed over the input the nodes compose from, its placement laid over the trunk's
|
||||
// (frombus, ADR 0559 §5): the trunk as the nodes compose it where the catalogue places them, the change's
|
||||
// manifests over the trunk's catalogue.
|
||||
if m.Input != nil {
|
||||
trunk = *m.Input
|
||||
trunk.Catalogue = withOverlay(m.Input.Catalogue, m.Manifests)
|
||||
}
|
||||
if c.Placement.Input == nil {
|
||||
failAt("", "the change's placement cannot be composed: the input with it cannot be read from the bus: "+c.Placement.Unread)
|
||||
} else {
|
||||
after = *c.Placement.Input
|
||||
cat := withOverlay(c.Placement.Input.Catalogue, changed)
|
||||
for _, name := range c.Removed {
|
||||
delete(cat, name)
|
||||
}
|
||||
after.Catalogue = cat
|
||||
}
|
||||
}
|
||||
for _, node := range machines {
|
||||
_, before := compose.Resolve(trunk, node)
|
||||
_, now := compose.Resolve(after, node)
|
||||
@@ -279,6 +306,18 @@ func GrantsAlone(m Mesh, c Change, composed []string) []string {
|
||||
return out
|
||||
}
|
||||
|
||||
// withOverlay is base with every manifest of over laid over it, a copy.
|
||||
func withOverlay(base, over map[string]catalogue.Manifest) map[string]catalogue.Manifest {
|
||||
out := make(map[string]catalogue.Manifest, len(base)+len(over))
|
||||
for name, man := range base {
|
||||
out[name] = man
|
||||
}
|
||||
for name, man := range over {
|
||||
out[name] = man
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func sortedKeys[V any](m map[string]V) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
@@ -57,6 +58,79 @@ func TestAMachineTheChangeBreaksFails(t *testing.T) {
|
||||
if v.State != delivery.CompositionFail || v.Machine != "ace" || !strings.Contains(v.Words, "database") {
|
||||
t.Fatalf("composition = %+v, want a fail on ace naming the provision nothing provides", v)
|
||||
}
|
||||
|
||||
// ADR 0559 §5 (issue 1126): a change of placement composes its placement laid over the trunk's input and fails naming
|
||||
// the node it breaks: a module no manifest names placed on g14; ace,
|
||||
// is no longer composed with it; a placement whose input cannot be read fails; a placement that breaks nothing passes, g14 touched.
|
||||
m := meshOf()
|
||||
placed := func(assigned map[string][]string) *ChangePlacement {
|
||||
in := input(m.Manifests, assigned)
|
||||
return &ChangePlacement{Placements: map[string]*delivery.Placement{"g14": {Node: "g14"}}, Input: &in}
|
||||
}
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
cp *ChangePlacement
|
||||
state string
|
||||
machine string
|
||||
words string
|
||||
}{
|
||||
{"a placement that breaks g14", placed(map[string][]string{"ace": {"mail", "store"}, "g14": {"web", "ghost"}}), delivery.CompositionFail, "g14", "ghost"},
|
||||
{"a placement whose input is not read", &ChangePlacement{Placements: map[string]*delivery.Placement{"g14": nil}, Unread: "no record of g14"}, delivery.CompositionFail, "", "no record of g14"},
|
||||
{"a placement that breaks nothing", placed(map[string][]string{"ace": {"mail", "store"}, "g14": {"web"}}), delivery.CompositionPass, "", ""},
|
||||
} {
|
||||
v := Compose(m, Change{Group: "g", Heads: "0123456789ab", Placement: row.cp}, "sdk", 0)
|
||||
if v.State != row.state || v.Machine != row.machine || !strings.Contains(v.Words, row.words) || !slices.Contains(v.Machines, "g14") {
|
||||
t.Errorf("%s: %+v", row.name, v)
|
||||
}
|
||||
}
|
||||
|
||||
// ADR 0559 §5: the placed nodes' answers. ace is placed on the trunk and its candidate moves web there (intended);
|
||||
// g14 is not placed on the trunk (the seed), so nothing is intended there; novox is placed and unchanged.
|
||||
at := time.Date(2026, 10, 27, 2, 0, 0, 0, time.UTC)
|
||||
cp := &ChangePlacement{Head: "h2", Written: at, Revisions: map[string]uint64{"ace": 7, "g14": 8},
|
||||
Placements: map[string]*delivery.Placement{"ace": {Node: "ace"}, "g14": {Node: "g14"}}}
|
||||
intended := map[string]map[string]bool{"ace": {"web": true}}
|
||||
same := func(rev uint64) *delivery.CandidateAnswer {
|
||||
return &delivery.CandidateAnswer{Head: "h2", Revision: rev, Verdict: delivery.CandidateSame, Same: []string{"mail"}}
|
||||
}
|
||||
placedNodes := []string{"ace", "g14", "novox"}
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
answers map[string]*delivery.CandidateAnswer
|
||||
now time.Time
|
||||
state string
|
||||
words string
|
||||
}{
|
||||
{"all the same", map[string]*delivery.CandidateAnswer{"ace": same(7), "g14": same(8), "novox": same(0)}, at, delivery.CompositionPass, ""},
|
||||
{"a module added where intended", map[string]*delivery.CandidateAnswer{"ace": {Head: "h2", Revision: 7, Verdict: delivery.CandidateDiffers,
|
||||
Added: []delivery.ModuleAdded{{Module: "web", Unissued: []string{"its own secret"}}}}, "g14": same(8), "novox": same(0)}, at, delivery.CompositionPass, ""},
|
||||
{"a module differing where not intended", map[string]*delivery.CandidateAnswer{"ace": {Head: "h2", Revision: 7, Verdict: delivery.CandidateDiffers,
|
||||
Differs: []delivery.ModuleDiffers{{Module: "mail", Why: "the port differs"}}}, "g14": same(8), "novox": same(0)}, at, delivery.CompositionFail, "ace composes mail differently"},
|
||||
{"a node's first placement file with a module differing", map[string]*delivery.CandidateAnswer{"ace": same(7), "g14": {Head: "h2", Revision: 8, Verdict: delivery.CandidateDiffers,
|
||||
Differs: []delivery.ModuleDiffers{{Module: "web", Why: "a binding differs"}}}, "novox": same(0)}, at, delivery.CompositionFail, "g14 composes web differently"},
|
||||
{"a refusal", map[string]*delivery.CandidateAnswer{"ace": same(7), "g14": {Head: "h2", Revision: 8, Verdict: delivery.CandidateRefused,
|
||||
Refusals: []delivery.ModuleRefused{{Module: "web", Why: "no own secret"}}}, "novox": same(0)}, at, delivery.CompositionFail, "g14 refuses"},
|
||||
{"a node not answering, within its bound", map[string]*delivery.CandidateAnswer{"ace": same(7), "novox": same(0)}, at.Add(AnswerBound - time.Second), delivery.CompositionWaits, "g14 (offline)"},
|
||||
{"an answer of an earlier revision, past its bound", map[string]*delivery.CandidateAnswer{"ace": same(6), "g14": same(8), "novox": same(0)}, at.Add(AnswerBound), delivery.CompositionFail, "ace " + delivery.CompositionUnanswered},
|
||||
{"its later answer", map[string]*delivery.CandidateAnswer{"ace": same(7), "g14": same(8), "novox": same(0)}, at.Add(time.Hour), delivery.CompositionPass, ""},
|
||||
} {
|
||||
pass := delivery.CompositionVerdict{Group: "g", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "sdk", Machines: []string{"ace", "g14"}}
|
||||
v := WithAnswers(pass, JudgeAnswers(cp, placedNodes, row.answers, intended, map[string]bool{"ace": true, "novox": true}, row.now))
|
||||
if v.State != row.state || !strings.Contains(v.Words, row.words) {
|
||||
t.Errorf("%s: %+v", row.name, v)
|
||||
}
|
||||
if _, err := delivery.EncodeCompositionVerdict(v); err != nil {
|
||||
t.Errorf("%s: the verdict cannot be written: %v", row.name, err)
|
||||
}
|
||||
}
|
||||
trunk := map[string]delivery.Placement{"ace": {Node: "ace", Modules: []delivery.PlacedModule{{Module: "mail"}}}, "novox": {Node: "novox"}}
|
||||
if got := PlacedNodes(trunk, &ChangePlacement{Placements: map[string]*delivery.Placement{"novox": nil, "g14": {Node: "g14"}}}); !slices.Equal(got, []string{"ace", "g14", "novox"}) {
|
||||
t.Errorf("placed at the head or on the trunk = %v", got)
|
||||
}
|
||||
got, _ := Intended(nil, trunk, &ChangePlacement{Placements: map[string]*delivery.Placement{"ace": {Node: "ace", Modules: []delivery.PlacedModule{{Module: "mail", Ports: map[string]int{"25": 2525}}, {Module: "web"}}}, "g14": {Node: "g14", Modules: []delivery.PlacedModule{{Module: "web"}}}}}, nil)
|
||||
if !got["ace"]["mail"] || !got["ace"]["web"] || len(got["g14"]) != 0 {
|
||||
t.Errorf("intended = %v, want ace's mail and web, nothing on g14 the trunk does not place", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B3c: what was already broken on the trunk is said and fails nothing.
|
||||
@@ -109,6 +183,10 @@ func TestTouched(t *testing.T) {
|
||||
if got := Touched(m, Change{Manifests: map[string]catalogue.Manifest{"store": m.Manifests["store"]}}); !slices.Equal(got, []string{"ace", "g14"}) {
|
||||
t.Errorf("touched by store with a consumer on g14 = %v", got)
|
||||
}
|
||||
// ADR 0559 §5: every node whose placement the change changes is touched, its file removed or not.
|
||||
if got := Touched(m, Change{Placement: &ChangePlacement{Placements: map[string]*delivery.Placement{"novox": nil, "shanks": {Node: "shanks"}}}}); !slices.Equal(got, []string{"novox", "shanks"}) {
|
||||
t.Errorf("touched by a placement = %v", got)
|
||||
}
|
||||
c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"new": {Module: "new", Version: "1"}}}
|
||||
if v := Compose(m, c, "sdk", 0); len(v.Machines) != 0 || !hasWarning(v, "touches no machine") || v.State != delivery.CompositionPass {
|
||||
t.Errorf("an added module assigned nowhere: %+v", v)
|
||||
|
||||
@@ -49,9 +49,21 @@ func HeadJob(m Mesh, c Change) (delivery.ComposeJob, error) {
|
||||
in.Catalogue = nil
|
||||
fromClaims = len(in.Holdings) == 0
|
||||
}
|
||||
changeCase := composer.Case{Name: CaseChange, Manifests: change}
|
||||
if c.Placement != nil {
|
||||
// The change's own placement, laid over the input's in the change case (ADR 0559 §5; composer.Case's Placements):
|
||||
// a node the input does not hold yet is added bare, so the composer places it.
|
||||
changeCase.Placements = c.Placement.Placements
|
||||
in.Nodes = slices.Clone(in.Nodes)
|
||||
for _, node := range c.Placement.Nodes() {
|
||||
if !slices.ContainsFunc(in.Nodes, func(n compose.Node) bool { return n.Name == node }) {
|
||||
in.Nodes = append(in.Nodes, compose.Node{Name: node})
|
||||
}
|
||||
}
|
||||
}
|
||||
req := composer.Request{V: composer.Version, Input: in, Depth: composer.DepthResolve,
|
||||
HoldingsFromClaims: fromClaims, Machines: Touched(m, c),
|
||||
Cases: []composer.Case{{Name: CaseTrunk, Manifests: trunk}, {Name: CaseChange, Manifests: change}}}
|
||||
Cases: []composer.Case{{Name: CaseTrunk, Manifests: trunk}, changeCase}}
|
||||
return job(SDKRepository, c.SDKHead, SDKDir, req, "the mesh-sdk head of "+c.Group)
|
||||
}
|
||||
|
||||
|
||||
@@ -120,6 +120,22 @@ func TestAGroupWithAnSDKMemberAsksTheComposerAtItsHead(t *testing.T) {
|
||||
len(placed.Input.Holdings) != 1 || placed.HoldingsFromClaims {
|
||||
t.Errorf("the placed check's request = %+v, holdings from claims %v", placed.Input, placed.HoldingsFromClaims)
|
||||
}
|
||||
|
||||
// ADR 0559 §5: a change of placement composes its own placement in the change case alone (composer.Case's Placements),
|
||||
// a node the input does not hold yet added bare so the composer places it, and the node touched.
|
||||
c.Placement = &ChangePlacement{Placements: map[string]*delivery.Placement{"novox": {Node: "novox"}}}
|
||||
candJob, err := HeadJob(m, c)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cand, err := composer.Decode(candJob.Request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cand.Cases[0].Placements != nil || cand.Cases[1].Placements["novox"] == nil || len(cand.Input.Nodes) != 2 || cand.Input.Nodes[1].Name != "novox" ||
|
||||
!slices.Equal(cand.Machines, []string{"g14", "novox"}) || len(m.Input.Nodes) != 1 {
|
||||
t.Errorf("the placement's request = cases %+v, nodes %+v, machines %v", cand.Cases, cand.Input.Nodes, cand.Machines)
|
||||
}
|
||||
}
|
||||
|
||||
// 1622/AC2, AC3: a catalogue member using a field only the SDK head reads passes with the head's composer, and the
|
||||
|
||||
@@ -34,7 +34,7 @@ func TestManifest(t *testing.T) {
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-build-go.steps", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants",
|
||||
"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-indexer.placement-candidates", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
t.Errorf("reads = %v, want exactly %v", reads, want)
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-indexer.placement",
|
||||
"mesh-delivery-indexer.placement-candidates",
|
||||
"mesh-delivery-teller.answers",
|
||||
"mesh-delivery-watchman.presence",
|
||||
"node-engine.have"
|
||||
|
||||
Reference in New Issue
Block a user