diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index d6d3157..47c923a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -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. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index a3d582b..cc69cca 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index dc9bc87..d146cc7 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -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{} } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 7d25c1b..0413710 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -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() diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 22fb1d4..f0d2949 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -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` 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{} diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 9476f6d..73e5df7 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -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.20261027014310-cadd7b1b2f8f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum index e8660dd..0082dd1 100644 --- a/modules/mesh-delivery-assigner/go.sum +++ b/modules/mesh-delivery-assigner/go.sum @@ -1,5 +1,17 @@ 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= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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= diff --git a/modules/mesh-delivery-assigner/internal/assign/candidates.go b/modules/mesh-delivery-assigner/internal/assign/candidates.go new file mode 100644 index 0000000..67b7739 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/candidates.go @@ -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 `.` 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` +// 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 +} diff --git a/modules/mesh-delivery-assigner/internal/assign/compose.go b/modules/mesh-delivery-assigner/internal/assign/compose.go index cb788b4..6bde172 100644 --- a/modules/mesh-delivery-assigner/internal/assign/compose.go +++ b/modules/mesh-delivery-assigner/internal/assign/compose.go @@ -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 { diff --git a/modules/mesh-delivery-assigner/internal/assign/compose_test.go b/modules/mesh-delivery-assigner/internal/assign/compose_test.go index 96c46ab..4054e91 100644 --- a/modules/mesh-delivery-assigner/internal/assign/compose_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/compose_test.go @@ -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) diff --git a/modules/mesh-delivery-assigner/internal/assign/composer.go b/modules/mesh-delivery-assigner/internal/assign/composer.go index 8c60ce8..03b4d60 100644 --- a/modules/mesh-delivery-assigner/internal/assign/composer.go +++ b/modules/mesh-delivery-assigner/internal/assign/composer.go @@ -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) } diff --git a/modules/mesh-delivery-assigner/internal/assign/composer_test.go b/modules/mesh-delivery-assigner/internal/assign/composer_test.go index 2bc3258..a01cebd 100644 --- a/modules/mesh-delivery-assigner/internal/assign/composer_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/composer_test.go @@ -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 diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 3c6e8a9..9e5584f 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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) } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index a979017..eb171e7 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -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" diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index d202053..08c70f0 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026233910-a2759ec92ec9 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 918a889..6cf00d5 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,5 +1,11 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026233910-a2759ec92ec9 h1:MS7i4NVNQHMAMqRRFMdzbto31BHZY+BuEZY1ql3p4Rs= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026233910-a2759ec92ec9/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= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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= diff --git a/modules/mesh-delivery-checker/internal/check/awaiting.go b/modules/mesh-delivery-checker/internal/check/awaiting.go index 74abfda..c617e31 100644 --- a/modules/mesh-delivery-checker/internal/check/awaiting.go +++ b/modules/mesh-delivery-checker/internal/check/awaiting.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -96,9 +97,19 @@ func notJudged(v delivery.Verdict) bool { return v.Composition != nil && v.Composition.State == delivery.CompositionNotRun && v.Composition.Words == delivery.AssignerNotRunning } +// unanswered says whether a verdict failed only on its composition, and that only because nodes the change places did +// not answer its placement in time (delivery.CompositionUnanswered, ADR 0559 §5): its tests passed (a composition is read +// only into a passing verdict), so a later composition of the same heads is taken in on its event, with no new check +// (ADR 0548 §1: a part of the mesh not heard is judged again when it is heard). +func unanswered(v delivery.Verdict) bool { + return v.State == delivery.VerdictFail && v.Composition != nil && v.Composition.State == delivery.CompositionFail && + strings.Contains(v.Composition.Words, delivery.CompositionUnanswered) && v.Failed.Words == v.Composition.Words +} + // Composed takes in a composition the assigner wrote after the check's verdict said it was not judged because the -// assigner was not running (the assigner started since): the verdict and its status are written again with it. A -// composition of heads no candidate holds now, or of a verdict that judged its composition, is left. +// assigner was not running (the assigner started since), or after it failed only because placed nodes had not answered +// (unanswered): the verdict and its status are written again with it. A composition of heads no candidate holds now, or +// of a verdict that judged its composition otherwise, is left. func (ch *Checker) Composed(ctx context.Context, comp delivery.CompositionVerdict) { ch.init() if comp.State == delivery.CompositionWaits { @@ -123,7 +134,15 @@ func (ch *Checker) Composed(ctx context.Context, comp delivery.CompositionVerdic return } v, err := delivery.DecodeVerdict(cur) - if err != nil || v.State != delivery.VerdictPass || !notJudged(v) { + if err != nil { + return + } + switch { + case v.State == delivery.VerdictPass && notJudged(v): + case unanswered(v) && comp.Words != v.Composition.Words: + // The nodes answered since (or another failure stands now): the verdict is its tests' pass with this composition. + v.State, v.Failed = delivery.VerdictPass, delivery.FailedFirst{} + default: return } withComposition(&v, comp) diff --git a/modules/mesh-delivery-checker/internal/check/awaiting_test.go b/modules/mesh-delivery-checker/internal/check/awaiting_test.go index c775c49..c373d0a 100644 --- a/modules/mesh-delivery-checker/internal/check/awaiting_test.go +++ b/modules/mesh-delivery-checker/internal/check/awaiting_test.go @@ -113,6 +113,27 @@ func TestTheAssignerStartingBringsTheComposition(t *testing.T) { t.Errorf("the status's line says %q", o.Text) } } + + // ADR 0559 §5: a verdict that failed only because a placed node had not answered takes the same heads' later + // composition in on its event, a pass making it pass with no new check; a composition failing the same way is left. + ch, cs := checkerWithNotJudged(t) + silent := delivery.CompositionVerdict{Group: cs.Group, Heads: cs.HeadsHash, State: delivery.CompositionFail, SDK: "x", Machine: "g14", + Words: "g14 (offline) " + delivery.CompositionUnanswered} + ch.Composed(context.Background(), silent) + for _, row := range []struct { + name string + comp delivery.CompositionVerdict + state string + }{ + {"the same failure again", silent, delivery.VerdictFail}, + {"the node's answer", delivery.CompositionVerdict{Group: cs.Group, Heads: cs.HeadsHash, State: delivery.CompositionPass, SDK: "x"}, delivery.VerdictPass}, + } { + ch.Composed(context.Background(), row.comp) + raw, _, _, _ := ch.Verdicts.Get(context.Background(), cs.Key) + if v, err := delivery.DecodeVerdict(raw); err != nil || v.State != row.state || v.Composition == nil || v.Composition.State != row.comp.State { + t.Errorf("%s: the verdict is %+v, %v", row.name, v, err) + } + } } // checkerWithNotJudged is a checker holding a candidate whose passing verdict said its composition was not judged. diff --git a/modules/mesh-delivery-forge-bridge/go.mod b/modules/mesh-delivery-forge-bridge/go.mod index 6d3738c..75272b1 100644 --- a/modules/mesh-delivery-forge-bridge/go.mod +++ b/modules/mesh-delivery-forge-bridge/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge 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.20261027014310-cadd7b1b2f8f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-forge-bridge/go.sum b/modules/mesh-delivery-forge-bridge/go.sum index e8660dd..0082dd1 100644 --- a/modules/mesh-delivery-forge-bridge/go.sum +++ b/modules/mesh-delivery-forge-bridge/go.sum @@ -1,5 +1,17 @@ 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= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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= diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index e65f8fd..45f4334 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -78,6 +78,24 @@ type pull struct { ModuleDirsSaid bool `json:"module_dirs_said"` Base string `json:"base"` MergeSHA string `json:"merge_commit_sha"` + // Paths are the files the pull request changes (a removed or renamed one included), PathsTruncated says the forge + // cut the list; nil when the forge did not say them. + Paths *[]string `json:"paths"` + PathsTruncated bool `json:"paths_truncated"` +} + +// touchesPlacement says whether a pull request may change a placement file: a changed path under +// delivery.PlacementDir, or paths the forge did not say or cut (the safe direction: the files are read). +func (p pull) touchesPlacement() bool { + if p.Paths == nil || p.PathsTruncated { + return true + } + for _, path := range *p.Paths { + if strings.HasPrefix(path, delivery.PlacementDir+"/") { + return true + } + } + return false } func unreadable(subject, why string, args ...any) error { @@ -98,7 +116,9 @@ type Listed func(owner, repo, sha, dir string) (map[string]string, error) // ConvertWith is Convert; with list, a merge of the catalogue repository (delivery.CatalogueRepository) also carries every // placement file of its merge commit (novox/hq ADR 0549 §2, issue 1126), read whole or not at all: a forge that cannot -// say is ErrRetry, as for the manifests, so the indexer never takes a part of placement for all of it. +// say is ErrRetry, as for the manifests, so the indexer never takes a part of placement for all of it. So does an +// updated pull request of the catalogue that touches placement/ (touchesPlacement), with every placement file at its +// head, which the indexer writes as placement candidates (ADR 0549 §4: the nodes it places answer it before it merges). func ConvertWith(subject string, data []byte, trees Trees, files Files, list Listed) (Message, error) { var action string switch subject { @@ -138,6 +158,13 @@ func ConvertWith(subject string, data []byte, trees Trees, files Files, list Lis return Message{}, fmt.Errorf("%w: %s: the manifests of %s/%s at the head %s: %v", ErrRetry, subject, p.Owner, p.Repo, p.HeadSHA, err) } } + if action == delivery.PullUpdated && list != nil && delivery.SameRepository(f.Repository, delivery.CatalogueRepository) && p.touchesPlacement() { + var err error + if f.Placements, err = placementsOf(p.Owner, p.Repo, p.HeadSHA, list, files); err != nil { + return Message{}, fmt.Errorf("%w: %s: the placement files of %s at the head %s: %v", ErrRetry, subject, f.Repository, p.HeadSHA, err) + } + f.PlacementsRead = true + } event, err := delivery.ForgeEvent(f.Repository, delivery.ForgePull) if err != nil { return Message{}, unreadable(subject, "%v", err) diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go index d4afe54..3d07f17 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -146,6 +146,55 @@ func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2 if f.Manifests != nil || f.Locks != nil || len(forge.calls) != 0 { t.Errorf("manifests %v locks %v calls %v", f.Manifests, f.Locks, forge.calls) } + + // Issue 1126 (ADR 0549 §4): an updated pull request of the catalogue that touches placement/ carries every placement + // file at its head, read whole or not at all; one that touches none, or of another repository, carries none and lists + // nothing; paths the forge did not say or cut are read as touching; a closed one carries none. + forge = &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest, + "placement/ace.json": []byte(`{"v":1,"node":"ace","modules":[]}`), "placement/g14.json": []byte(`{"v":1,"node":"g14","modules":[]}`)}} + listing := map[string]string{"ace.json": "b1", "g14.json": "b2", "README.md": "b3"} + var listedAt []string + list := func(owner, repo, sha, dir string) (map[string]string, error) { + listedAt = append(listedAt, owner+"/"+repo+"@"+sha+":"+dir) + return listing, nil + } + all := map[string]string{"ace": `{"v":1,"node":"ace","modules":[]}`, "g14": `{"v":1,"node":"g14","modules":[]}`} + catalogue := map[string]any{"repo": "mesh-catalog", "paths": []string{"placement/g14.json"}} + for _, row := range []struct { + name string + subject string + body map[string]any + want map[string]string + read bool + }{ + {"a catalogue pull touching placement", updatedSubject, catalogue, all, true}, + {"a catalogue pull touching no placement", updatedSubject, map[string]any{"repo": "mesh-catalog"}, nil, false}, + {"a catalogue pull whose paths are cut", updatedSubject, map[string]any{"repo": "mesh-catalog", "paths_truncated": true}, all, true}, + {"a catalogue pull whose paths are not said", updatedSubject, map[string]any{"repo": "mesh-catalog", "paths": nil}, all, true}, + {"another repository's pull touching placement/", updatedSubject, map[string]any{"paths": []string{"placement/g14.json"}}, nil, false}, + {"a closed catalogue pull", closedSubject, catalogue, nil, false}, + } { + listedAt = nil + body := updated(row.body) + msg, err := ConvertWith(row.subject, body, nil, forge.get, list) + if err != nil { + t.Fatalf("%s: %v", row.name, err) + } + f, err := delivery.DecodeForge(msg.Data) + if err != nil { + t.Fatalf("%s: %v", row.name, err) + } + if f.PlacementsRead != row.read || !reflect.DeepEqual(f.Placements, row.want) || (len(listedAt) > 0) != row.read { + t.Errorf("%s: placements %v (read %v), listed %v", row.name, f.Placements, f.PlacementsRead, listedAt) + } + if row.read && listedAt[0] != "novox/mesh-catalog@"+head+":placement" { + t.Errorf("%s: listed at %v, want the head", row.name, listedAt) + } + } + listing = map[string]string{"gone.json": "b5"} + if _, err := ConvertWith(updatedSubject, updated(catalogue), nil, forge.get, list); !errors.Is(err, ErrRetry) || !strings.Contains(err.Error(), "placement/gone.json") { + t.Errorf("a placement file the forge cannot give at the head: %v", err) + } } func TestAForgeErrorOrAMissingManifestIsRetryAndALockNotFoundIsLeftOut(t *testing.T) { // 1448/AC3 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index b79ba6b..2640989 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -47,6 +47,25 @@ func Tools(ask asker, now func() time.Time) []stdio.Tool { } return a.Result, nil }}, + {Name: index.PlacementRereadTool, + Description: "Read the catalogue's placement files again now (novox/hq ADR 0559 §8): the indexer asks the forge for the trunk's head and " + + "every open catalogue pull request's, and writes its states placement (placement.) and placement-candidates (.) " + + "as a merge and a pull request's head would, only where a key differs; a candidate key whose file no longer differs, or whose pull " + + "request is closed, is deleted. The recovery for a forge message the indexer missed. Answers the heads read, the keys written or " + + "deleted in each state, or why nothing was (the forge could not say). indexer_placement (mesh-delivery-view) reads the result.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "why": map[string]any{"type": "string", "description": "why it is read again now, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + key, value, err := index.NewPlacementReread(str(args, "why"), stdio.CallerOf(args), now()) + if err != nil { + return nil, err + } + a, err := ask(key, value) + if err != nil { + return nil, err + } + return a.Reread, nil + }}, } } @@ -64,7 +83,7 @@ func askTakeIn(s stdio.KeptState, bound time.Duration) asker { return false } a, err := index.DecodeTakeIn(v) - if err != nil || a.Answered.IsZero() || a.Result == nil { + if err != nil || !a.Done() { return false } select { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 4c2a403..3d1d1b1 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -317,6 +317,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Its placement state (ADR 0549 §2) is bound when it appears, as every state a delivery adds (issue 1790): until then a // catalogue merge's placement is said and not written, and the rest of the indexer runs. placement := process.AwaitBucket(ctx, js, conn, delivery.PlacementBucket, "no catalogue merge's placement is written", logger.Printf) + // Its placement candidates (issue 1126, ADR 0549 §4) are bound the same way: until the state appears a catalogue pull + // request's placement is said and not written, so no node answers it and its check names every placed node as silent. + placementCandidates := process.AwaitBucket(ctx, js, conn, delivery.PlacementCandidatesBucket, "no catalogue pull request's placement candidate is written", logger.Printf) // The controller's build records only name the directory of a module whose key names none (fillPaths): without them // such a module waits for its path, every other is indexed (issue 1945). builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf) @@ -352,7 +355,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro defer calls.Close() ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), - Placement: awaitedStore{placement}, Say: func(event string, body any) error { return calls.Emit(event, body) }} + Placement: awaitedStore{placement}, PlacementCandidates: awaitedStore{placementCandidates}, ReadPlacement: placementVia(calls.Call), Say: func(event string, body any) error { return calls.Emit(event, body) }} // The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key // main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does // not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go index 796cfba..0dada0f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go @@ -170,3 +170,110 @@ func readVia(call toolCaller) func(ctx context.Context, repository, sha, dir str return out, nil } } + +// The forge's tools the placement reread asks (indexer_placement_reread, novox/hq ADR 0559 §8): a repository's branches +// with their commits, and every open pull request with its branch and head. +const ( + listBranchesTool = "gitea.gitea_list_branches" + openPullsTool = "gitea.gitea_open_pulls" + placementTrunk = "main" +) + +// placementVia reads the catalogue's placement now through call: the trunk's head (gitea_list_branches), every placement +// file at it, and each open pull request of the catalogue into the trunk (gitea_open_pulls) with every placement file at +// its head, each read whole or not at all (gitea_commit_trees with listed, gitea_get_file). +func placementVia(call toolCaller) func(ctx context.Context) (index.PlacementRead, error) { + owner, repo, _ := strings.Cut(delivery.CatalogueRepository, "/") + filesAt := func(sha string) (map[string]string, error) { + raw, err := call(commitTreesTool, map[string]any{"owner": owner, "repo": repo, "sha": sha, "listed": []string{delivery.PlacementDir}}) + if err != nil { + return nil, err + } + var a struct { + Listed map[string]map[string]string `json:"listed"` + } + if err := json.Unmarshal(raw, &a); err != nil || a.Listed == nil { + return nil, fmt.Errorf("the answer of %s lists no directory: %v", commitTreesTool, err) + } + var out map[string]string + for name := range a.Listed[delivery.PlacementDir] { + file := delivery.PlacementDir + "/" + name + node, ok := delivery.PlacementFileNode(file) + if !ok { + continue + } + raw, err := call(getFileTool, map[string]string{"owner": owner, "repo": repo, "path": file, "ref": sha}) + if err != nil { + return nil, fmt.Errorf("%s at %s: %w", file, sha, err) + } + var f struct { + File *struct { + Content *string `json:"content"` + } `json:"file"` + } + if err := json.Unmarshal(raw, &f); err != nil || f.File == nil || f.File.Content == nil { + return nil, fmt.Errorf("the answer of %s for %s is not {file: {content}}: %v", getFileTool, file, err) + } + if out == nil { + out = map[string]string{} + } + out[node] = *f.File.Content + } + return out, nil + } + return func(context.Context) (index.PlacementRead, error) { + raw, err := call(listBranchesTool, map[string]string{"owner": owner, "repo": repo}) + if err != nil { + return index.PlacementRead{}, err + } + var b struct { + Branches []struct { + Name string `json:"name"` + Commit string `json:"commit"` + } `json:"branches"` + } + if err := json.Unmarshal(raw, &b); err != nil { + return index.PlacementRead{}, fmt.Errorf("the answer of %s is not {branches}: %w", listBranchesTool, err) + } + var out index.PlacementRead + for _, br := range b.Branches { + if br.Name == placementTrunk { + out.Trunk = br.Commit + } + } + if out.Trunk == "" { + return index.PlacementRead{}, fmt.Errorf("%s names no branch %s of %s", listBranchesTool, placementTrunk, delivery.CatalogueRepository) + } + if out.Files, err = filesAt(out.Trunk); err != nil { + return index.PlacementRead{}, err + } + raw, err = call(openPullsTool, map[string]any{}) + if err != nil { + return index.PlacementRead{}, err + } + var o struct { + Pulls *[]struct { + Owner string `json:"owner"` + Repo string `json:"repo"` + Number int `json:"number"` + Base string `json:"base"` + Head string `json:"head"` + HeadSHA string `json:"head_sha"` + } `json:"pulls"` + } + if err := json.Unmarshal(raw, &o); err != nil || o.Pulls == nil { + return index.PlacementRead{}, fmt.Errorf("the answer of %s is not {pulls}: %v", openPullsTool, err) + } + for _, p := range *o.Pulls { + if !delivery.SameRepository(p.Owner+"/"+p.Repo, delivery.CatalogueRepository) || p.Base != placementTrunk || p.Number <= 0 || p.Head == "" || p.HeadSHA == "" { + continue + } + files, err := filesAt(p.HeadSHA) + if err != nil { + return index.PlacementRead{}, fmt.Errorf("%s#%d: %w", delivery.CatalogueRepository, p.Number, err) + } + out.Pulls = append(out.Pulls, index.PullPlacement{Branch: p.Head, Number: p.Number, Head: p.HeadSHA, Files: files}) + } + return out, nil + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index f0c7b02..13d4fbb 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -14,6 +14,7 @@ import ( // takerIn is the part of the indexer a takeins value asks through. type takerIn interface { TakeIn(ctx context.Context, module string) index.TakenIn + RereadPlacement(ctx context.Context) index.PlacementReread } // takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the @@ -31,9 +32,15 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream if !a.Answered.IsZero() { return } - r := ix.TakeIn(ctx, a.Module) - a.Result, a.Answered = &r, time.Now().UTC() - logf("%s: a take-in asked by %s (%s): %s", a.Module, a.By, a.Why, r.Answer) + if a.Placement { + r := ix.RereadPlacement(ctx) + a.Reread, a.Answered = &r, time.Now().UTC() + logf("a placement reread asked by %s (%s): %s", a.By, a.Why, r.Answer) + } else { + r := ix.TakeIn(ctx, a.Module) + a.Result, a.Answered = &r, time.Now().UTC() + logf("%s: a take-in asked by %s (%s): %s", a.Module, a.By, a.Why, r.Answer) + } data, err := json.Marshal(a) if err != nil { return diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 285279f..b35bfdf 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index b3dab4f..9259a48 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,17 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/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= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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= diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 80d9abf..b4e8368 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -67,6 +67,12 @@ type Indexer struct { Logf func(format string, args ...any) // Placement is the placement bucket (delivery.PlacementBucket, ADR 0549 §2); nil writes no placement. Placement Store + // PlacementCandidates is the placement candidates bucket (delivery.PlacementCandidatesBucket, issue 1126); nil writes + // no candidate (placement_candidates.go). + PlacementCandidates Store + // ReadPlacement asks the forge for the catalogue's placement now, at the trunk's head and each open pull request's + // (indexer_placement_reread, ADR 0559 §8); nil rereads none. + ReadPlacement func(ctx context.Context) (PlacementRead, error) // Say says one of the indexer's events (placement-changed); nil says none. Say func(event string, body any) error // Drained says whether the indexer has applied every message of the forge stream (the consumer has none pending and none @@ -117,6 +123,9 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { if err := ix.applyPlacement(ctx, f); err != nil { return err } + if err := ix.applyPlacementCandidates(ctx, f); err != nil { + return err + } var digests map[string]string var held map[string]delivery.Module if f.Kind == delivery.ForgePull { @@ -171,6 +180,14 @@ func (ix *Indexer) applyOpen(ctx context.Context, f delivery.Forge) error { for _, p := range f.Open { named[p] = true } + if ix.PlacementCandidates != nil { + // A pull request the forge no longer has open keeps no placement candidate (issue 1126). + if err := ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool { + return !named[delivery.OpenPull{Repository: c.Pull.Repository, Number: c.Pull.Number}] + }, "its pull request is not open"); err != nil { + return err + } + } keys, err := ix.Store.Keys(ctx) if err != nil { return fmt.Errorf("listing the candidates: %w", err) diff --git a/modules/mesh-delivery-indexer/internal/index/placement_candidates.go b/modules/mesh-delivery-indexer/internal/index/placement_candidates.go new file mode 100644 index 0000000..e301880 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/placement_candidates.go @@ -0,0 +1,302 @@ +package index + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// Placement candidates (issue 1126; novox/hq ADR 0549 §4: a placement change passes the composition check before any +// node sees it). An updated pull request of the catalogue that touches placement/ carries every placement file at its +// head (the forge bridge, PlacementsRead). For each node whose file at the head differs from the trunk's key +// `placement.` (SamePlacement, the revision left out), or whose file the head removes, the indexer writes the key +// `.` of its state `placement-candidates`; a node whose file is the trunk's has none, and a key the head no +// longer needs is deleted. The key is written only when its value changes, so its write is the node's event to compose +// in shadow and answer. When the pull request closes or merges, or an open message no longer names it, its keys are +// deleted, and so are they when a head the forge says the changes of (DirsKnown) carries no placement: it touches none. +// A pull message that says nothing of its head (a resend of the open pull requests) leaves them as they are. +// +// A file at the head that does not read as a placement, or names another node, is said in the log and that node gets no +// key (one it had is deleted): the catalogue's own check refuses the pull request (ADR 0549 §1), and a node answers no +// candidate that is not one. + +// applyPlacementCandidates takes one forge message into the placement candidates. +func (ix *Indexer) applyPlacementCandidates(ctx context.Context, f delivery.Forge) error { + if ix.PlacementCandidates == nil || !delivery.SameRepository(f.Repository, delivery.CatalogueRepository) || trunkLike[f.Branch] { + return nil + } + if f.Kind == delivery.ForgeMerged || f.Action == delivery.PullClosed { + return ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool { + return c.Group == f.Branch && delivery.SameRepository(c.Pull.Repository, f.Repository) && c.Pull.Number == f.Number + }, fmt.Sprintf("%s#%d %s", f.Repository, f.Number, map[bool]string{true: "merged", false: "closed"}[f.Kind == delivery.ForgeMerged])) + } + if f.Kind != delivery.ForgePull { + return nil + } + if !f.PlacementsRead { + if f.DirsKnown { + // The forge said what this head changes (its directories and paths) and the bridge read no placement: the head + // changes no placement file, so a key an earlier head left goes (ADR 0559 §3). + return ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool { + return c.Group == f.Branch && delivery.SameRepository(c.Pull.Repository, f.Repository) && c.Pull.Number == f.Number + }, fmt.Sprintf("%s#%d at %s changes no placement file", f.Repository, f.Number, f.Head)) + } + return nil + } + trunk, err := ix.trunkPlacements(ctx) + if err != nil { + ix.logf("%s#%d at %s: its placement candidates are not written: %v", f.Repository, f.Number, f.Head, err) + return nil + } + nodes := map[string]bool{} + for node := range f.Placements { + nodes[node] = true + } + for node := range trunk { + nodes[node] = true + } + pull := delivery.Head{Repository: f.Repository, Number: f.Number, Head: f.Head} + for _, node := range slices.Sorted(maps.Keys(nodes)) { + key, err := delivery.PlacementCandidateKey(f.Branch, node) + if err != nil { + return fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err) + } + var next *delivery.PlacementCandidate // nil: no key for this node + raw, atHead := f.Placements[node] + before, inTrunk := trunk[node] + switch { + case atHead: + p, err := delivery.DecodePlacement([]byte(raw)) + if err == nil && p.Node != node { + err = fmt.Errorf("the file of %s names the node %q", node, p.Node) + } + if err != nil { + ix.logf("%s#%d at %s: the placement file of %s is refused, so %s has no placement candidate (the catalogue's check refuses the pull request): %v", + f.Repository, f.Number, f.Head, node, node, err) + break + } + if inTrunk && delivery.SamePlacement(before, p) { + break + } + next = &delivery.PlacementCandidate{Group: f.Branch, Pull: pull, Placement: &p} + case inTrunk: + next = &delivery.PlacementCandidate{Group: f.Branch, Pull: pull} // the head removes the node's file + } + if err := ix.writePlacementCandidate(ctx, key, next); err != nil { + return err + } + } + // A node the head and the trunk both no longer name keeps no key of this group. + return ix.dropPlacementCandidates(ctx, func(node string, c delivery.PlacementCandidate) bool { + return c.Group == f.Branch && delivery.SameRepository(c.Pull.Repository, f.Repository) && c.Pull.Number == f.Number && !nodes[node] + }, fmt.Sprintf("%s#%d at %s", f.Repository, f.Number, f.Head)) +} + +// trunkPlacements is every node's placement in the trunk's state placement, by node. +func (ix *Indexer) trunkPlacements(ctx context.Context) (map[string]delivery.Placement, error) { + out := map[string]delivery.Placement{} + if ix.Placement == nil { + return nil, errors.New("the indexer reads no placement state") + } + keys, err := ix.Placement.Keys(ctx) + if err != nil { + return nil, fmt.Errorf("reading the placement keys: %w", err) + } + for _, key := range keys { + node, err := delivery.ParsePlacementKey(key) + if err != nil { + continue + } + value, _, found, err := ix.Placement.Get(ctx, key) + if err != nil { + return nil, fmt.Errorf("reading %s: %w", key, err) + } + if !found { + continue + } + p, err := delivery.DecodePlacement(value) + if err != nil { + return nil, fmt.Errorf("%s holds what is not a placement: %w", key, err) + } + out[node] = p + } + return out, nil +} + +// writePlacementCandidate writes one candidate key, or deletes it for next nil, only when it changes. +func (ix *Indexer) writePlacementCandidate(ctx context.Context, key string, next *delivery.PlacementCandidate) error { + var data []byte + if next != nil { + var err error + if data, err = delivery.EncodePlacementCandidate(*next); err != nil { + return fmt.Errorf("%w: %s.%s: %v", ErrUnreadable, delivery.PlacementCandidatesState, key, err) + } + } + err := write(ctx, ix.PlacementCandidates, key, func(value []byte, found bool) ([]byte, bool, error) { + switch { + case next == nil: + return nil, found, nil + case found && string(value) == string(data): + return nil, false, nil + } + return data, false, nil + }) + if errors.Is(err, process.ErrNotYet) { + ix.logf("%s.%s is not written: %v", delivery.PlacementCandidatesState, key, err) + return nil + } + return err +} + +// dropPlacementCandidates deletes every candidate key whose node and value gone says, and says each with why. +func (ix *Indexer) dropPlacementCandidates(ctx context.Context, gone func(node string, c delivery.PlacementCandidate) bool, why string) error { + keys, err := ix.PlacementCandidates.Keys(ctx) + if errors.Is(err, process.ErrNotYet) { + return nil + } + if err != nil { + return fmt.Errorf("reading the placement candidate keys: %w", err) + } + for _, key := range keys { + _, node, err := delivery.ParsePlacementCandidateKey(key) + if err != nil { + continue + } + dropped := false + err = write(ctx, ix.PlacementCandidates, key, func(value []byte, found bool) ([]byte, bool, error) { + if !found { + return nil, false, nil + } + c, err := delivery.DecodePlacementCandidate(value) + if err != nil { + ix.logf("%s.%s holds what is not a placement candidate, deleted: %v", delivery.PlacementCandidatesState, key, err) + return nil, true, nil + } + dropped = gone(node, c) + return nil, dropped, nil + }) + if err != nil { + return err + } + if dropped { + ix.logf("%s.%s is deleted: %s", delivery.PlacementCandidatesState, key, why) + } + } + return nil +} + +// PlacementRead is the catalogue's placement as the forge has it now (indexer_placement_reread, ADR 0559 §8): the trunk's +// head and every placement file at it, by node, and each open pull request into the trunk with its own. +type PlacementRead struct { + Trunk string + Files map[string]string + Pulls []PullPlacement +} + +// PullPlacement is one open catalogue pull request and every placement file at its head, by node. +type PullPlacement struct { + Branch string + Number int + Head string + Files map[string]string +} + +// PlacementReread is the answer of indexer_placement_reread: the heads read, the keys written or deleted in each state, +// and in words what happened. +type PlacementReread struct { + Trunk string `json:"trunk,omitempty"` + Pulls []string `json:"pulls,omitempty"` + Placement []string `json:"placement"` + Candidates []string `json:"candidates"` + Answer string `json:"answer"` +} + +// RereadPlacement reads the catalogue's placement now, at the trunk's head and at the head of each open catalogue pull +// request, and writes placement and placement-candidates as a merged and a pull message would (ADR 0549 §2, ADR 0559 +// §3), only where a key differs; a candidate key of a pull request that is no longer open, or whose file no longer +// differs at its head, is deleted. It is the recovery for a forge message the indexer missed (ADR 0559 §8). +func (ix *Indexer) RereadPlacement(ctx context.Context) PlacementReread { + out := PlacementReread{Placement: []string{}, Candidates: []string{}} + if ix.ReadPlacement == nil || ix.Placement == nil || ix.PlacementCandidates == nil { + out.Answer = "the indexer reads no placement here: it has no forge reader or no placement state" + return out + } + read, err := ix.ReadPlacement(ctx) + if err != nil { + out.Answer = "the forge could not say the catalogue's placement, nothing is written: " + err.Error() + return out + } + out.Trunk = read.Trunk + before := revisions(ctx, ix.Placement) + beforeC := revisions(ctx, ix.PlacementCandidates) + if err := ix.applyPlacement(ctx, delivery.Forge{Kind: delivery.ForgeMerged, Repository: delivery.CatalogueRepository, Merge: read.Trunk, + Placements: read.Files, PlacementsRead: true}); err != nil { + out.Answer = "the trunk's placement could not be written: " + err.Error() + return out + } + open := map[int]bool{} + var failed []string + for _, p := range read.Pulls { + open[p.Number] = true + out.Pulls = append(out.Pulls, fmt.Sprintf("%s#%d at %s", delivery.CatalogueRepository, p.Number, p.Head)) + f := delivery.Forge{Kind: delivery.ForgePull, Repository: delivery.CatalogueRepository, Branch: p.Branch, Number: p.Number, Head: p.Head, + Action: delivery.PullUpdated, Placements: p.Files, PlacementsRead: true} + if err := ix.applyPlacementCandidates(ctx, f); err != nil { + failed = append(failed, fmt.Sprintf("#%d: %v", p.Number, err)) + } + } + if err := ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool { + return delivery.SameRepository(c.Pull.Repository, delivery.CatalogueRepository) && !open[c.Pull.Number] + }, "its pull request is not open (indexer_placement_reread)"); err != nil { + failed = append(failed, err.Error()) + } + out.Placement = changed(before, revisions(ctx, ix.Placement)) + out.Candidates = changed(beforeC, revisions(ctx, ix.PlacementCandidates)) + switch { + case len(failed) > 0: + out.Answer = "read at the trunk's head " + read.Trunk + "; some candidates are not written: " + fmt.Sprint(failed) + case len(out.Placement)+len(out.Candidates) == 0: + out.Answer = "read at the trunk's head " + read.Trunk + " and every open catalogue pull request's: every key was as the forge has it, nothing written" + default: + out.Answer = fmt.Sprintf("read at the trunk's head %s and every open catalogue pull request's: %d placement keys and %d candidate keys written or deleted", + read.Trunk, len(out.Placement), len(out.Candidates)) + } + return out +} + +// revisions is every key of a store with its revision; a state not on the bus yet has none. +func revisions(ctx context.Context, st Store) map[string]uint64 { + out := map[string]uint64{} + keys, err := st.Keys(ctx) + if err != nil { + return out + } + for _, key := range keys { + if _, rev, found, err := st.Get(ctx, key); err == nil && found { + out[key] = rev + } + } + return out +} + +// changed is every key written, created or deleted between two revisions maps, sorted. +func changed(before, after map[string]uint64) []string { + out := []string{} + for key, rev := range after { + if before[key] != rev { + out = append(out, key) + } + } + for key := range before { + if _, ok := after[key]; !ok { + out = append(out, key) + } + } + slices.Sort(out) + return out +} diff --git a/modules/mesh-delivery-indexer/internal/index/placement_test.go b/modules/mesh-delivery-indexer/internal/index/placement_test.go index adb48f1..3e0fd43 100644 --- a/modules/mesh-delivery-indexer/internal/index/placement_test.go +++ b/modules/mesh-delivery-indexer/internal/index/placement_test.go @@ -3,6 +3,7 @@ package index import ( "context" "reflect" + "slices" "strings" "testing" @@ -94,4 +95,107 @@ func placementIsWrittenAtAMergeOnlyWhereItChanged(t *testing.T) { if p, _ := written("ace"); p.Revision != "m1" || len(said) != 0 { t.Errorf("a message that read no placement changed it: %+v, %v", p, said) } + + // Issue 1126 (ADR 0549 §4), placement candidates: the trunk now places ace alone. A catalogue pull request writes + // `.` for each node whose file at its head differs from the trunk's, or that the head removes, and only + // when the value changes; a pull message that read no placement, or a pull request changing none, writes nothing; a + // file that does not read gets no key; a merge, a close or an open message not naming it deletes its keys. + cands := newMem() + ix.PlacementCandidates = cands + said = nil + pullOf := func(branch string, number int, head string, files map[string]string) delivery.Forge { + return delivery.Forge{V: delivery.Version, Kind: delivery.ForgePull, Repository: delivery.CatalogueRepository, Branch: branch, + Number: number, Head: head, Action: delivery.PullUpdated, Placements: files, PlacementsRead: files != nil} + } + candidate := func(key string) (delivery.PlacementCandidate, uint64, bool) { + t.Helper() + v, ok := cands.vals[key] + if !ok { + return delivery.PlacementCandidate{}, 0, false + } + c, err := delivery.DecodePlacementCandidate(v) + if err != nil { + t.Fatal(err) + } + return c, cands.revs[key], true + } + keys := func() string { k, _ := cands.Keys(ctx); slices.Sort(k); return strings.Join(k, ",") } + g14New := `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}` + for _, row := range []struct { + name string + f delivery.Forge + keys string + }{ + {"a pull request adding g14", pullOf("feat/p", 9, h1, map[string]string{"ace": ace, "g14": g14New}), "feat/p.g14"}, + {"the same head again", pullOf("feat/p", 9, h1, map[string]string{"ace": ace, "g14": g14New}), "feat/p.g14"}, + {"a pull message that read no placement", pullOf("feat/p", 9, h2, nil), "feat/p.g14"}, + {"a pull request changing no placement", pullOf("feat/q", 10, h1, map[string]string{"ace": ace}), "feat/p.g14"}, + {"a new head removing ace", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"}, + {"a head whose g14 file does not read", pullOf("feat/p", 9, h3, map[string]string{"g14": `{"v":1,"node":"ace","modules":[]}`}), "feat/p.ace"}, + {"the head adding g14 again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"}, + {"a head the forge says changes no placement file", func() delivery.Forge { f := pullOf("feat/p", 9, h3, nil); f.DirsKnown = true; return f }(), ""}, + {"the head adding g14 and removing ace again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"}, + {"an open message not naming it", delivery.Forge{V: delivery.Version, Kind: delivery.ForgeOpen, Open: []delivery.OpenPull{{Repository: delivery.CatalogueRepository, Number: 10}}}, ""}, + {"the pull request heard again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"}, + {"its merge", func() delivery.Forge { + m := merged("m7", map[string]string{"g14": g14New}) + m.Branch, m.Number = "feat/p", 9 + return m + }(), ""}, + } { + _, revBefore, _ := candidate("feat/p.g14") + if err := ix.Apply(ctx, row.f); err != nil { + t.Fatalf("%s: %v", row.name, err) + } + if got := keys(); got != row.keys { + t.Errorf("%s: keys %q, want %q", row.name, got, row.keys) + } + c, rev, ok := candidate("feat/p.g14") + switch row.name { + case "a pull request adding g14": + if !ok || c.Group != "feat/p" || c.Pull != (delivery.Head{Repository: delivery.CatalogueRepository, Number: 9, Head: h1}) || c.Placement == nil || + !reflect.DeepEqual(c.Placement.Assigned(), []string{"zsh"}) { + t.Errorf("%s: %+v", row.name, c) + } + case "the same head again", "a pull message that read no placement", "a pull request changing no placement": + if rev != revBefore { + t.Errorf("%s: feat/p.g14 written again (revision %d, was %d)", row.name, rev, revBefore) + } + case "a new head removing ace": + if a, _, _ := candidate("feat/p.ace"); a.Placement != nil || a.Pull.Head != h2 || c.Pull.Head != h2 { + t.Errorf("%s: ace %+v, g14 at %s", row.name, a, c.Pull.Head) + } + } + } + if p, _ := written("g14"); len(p.Modules) != 1 || p.Revision != "m7" { + t.Errorf("the merge's trunk placement of g14: %+v", p) + } + + // ADR 0559 §8, indexer_placement_reread: a reread with nothing changed writes nothing; one after a missed merge and a + // missed head writes those keys; one after a missed revert deletes the candidate; a forge that cannot say writes nothing. + g14Other := `{"v":1,"node":"g14","modules":[{"module":"zsh"},{"module":"htop"}]}` + for _, row := range []struct { + name string + read PlacementRead + err error + placement, cands []string + }{ + {"nothing changed", PlacementRead{Trunk: "m7", Files: map[string]string{"g14": g14New}}, nil, []string{}, []string{}}, + {"a missed merge and a missed head", PlacementRead{Trunk: "m8", Files: map[string]string{"g14": g14New, "ace": ace}, + Pulls: []PullPlacement{{Branch: "feat/r", Number: 11, Head: h1, Files: map[string]string{"g14": g14Other, "ace": ace}}}}, nil, + []string{"placement.ace"}, []string{"feat/r.g14"}}, + {"a missed revert", PlacementRead{Trunk: "m8", Files: map[string]string{"g14": g14New, "ace": ace}, + Pulls: []PullPlacement{{Branch: "feat/r", Number: 11, Head: h2, Files: map[string]string{"g14": g14New, "ace": ace}}}}, nil, + []string{}, []string{"feat/r.g14"}}, + {"a forge that cannot say", PlacementRead{}, context.DeadlineExceeded, []string{}, []string{}}, + } { + ix.ReadPlacement = func(context.Context) (PlacementRead, error) { return row.read, row.err } + r := ix.RereadPlacement(ctx) + if !reflect.DeepEqual(r.Placement, row.placement) || !reflect.DeepEqual(r.Candidates, row.cands) || r.Answer == "" { + t.Errorf("%s: %+v", row.name, r) + } + } + if keys() != "" { + t.Errorf("after the reread of a revert, candidates %q", keys()) + } } diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 512bc9a..5ceec19 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -17,6 +17,10 @@ import ( // of the indexer's states stay in mesh-delivery-view. const ( TakeInTool = "indexer_take_in" + // PlacementRereadTool reads the catalogue's placement again now (ADR 0559 §8); it asks through the same state, under + // PlacementRereadKey, a key no module has (a module's name holds no dot). + PlacementRereadTool = "indexer_placement_reread" + PlacementRereadKey = "placement.reread" // TakeInsState is the indexer's state of take-in asks: the bucket mesh-delivery-indexer_takeins, keyed as the modules state. TakeInsState = "takeins" TakeInsBucket = delivery.IndexerModule + "_" + TakeInsState @@ -26,12 +30,24 @@ const ( // indexer read it, its answer. type TakeInAsk struct { V int `json:"v"` - Module string `json:"module"` + Module string `json:"module,omitempty"` Why string `json:"why,omitempty"` By string `json:"by,omitempty"` At time.Time `json:"at"` Result *TakenIn `json:"result,omitempty"` Answered time.Time `json:"answered,omitzero"` + // Placement asks indexer_placement_reread instead of a module's take-in; Reread is its answer. + Placement bool `json:"placement,omitempty"` + Reread *PlacementReread `json:"reread,omitempty"` +} + +// Done says the ask is answered. +func (a TakeInAsk) Done() bool { return !a.Answered.IsZero() && (a.Result != nil || a.Reread != nil) } + +// NewPlacementReread is the value indexer_placement_reread writes: its key and the value. +func NewPlacementReread(why, by string, at time.Time) (string, []byte, error) { + data, err := json.Marshal(TakeInAsk{V: 1, Placement: true, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) + return PlacementRereadKey, data, err } // DecodeTakeIn reads a takeins value. @@ -43,7 +59,7 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) { if a.V != 1 { return a, fmt.Errorf("a takeins value of version %d, not 1", a.V) } - if a.Module == "" { + if a.Module == "" && !a.Placement { return a, errors.New("a takeins value naming no module") } return a, nil diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 0122ca7..7cf1b86 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -85,7 +85,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []string{"placement-changed"}; !reflect.DeepEqual(m.Emits, want) { t.Errorf("emits = %v, want %v", m.Emits, want) } - if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "placement-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { @@ -105,11 +105,12 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths: // ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a // commit is in another (issue 1900). - if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file"}; !reflect.DeepEqual(m.Invokes, want) { + // ADR 0559 §8: the placement reread also asks the catalogue's branches and open pull requests. + if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_list_branches", "gitea.gitea_open_pulls"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } - // 1945: one tool, indexer_take_in, served by its tools bundle through the takeins state; the process serves none. - if want := []string{"indexer_take_in"}; !reflect.DeepEqual(m.Tools, want) { + // 1945: indexer_take_in, and indexer_placement_reread (ADR 0559 §8), served by its tools bundle through the takeins state; the process serves none. + if want := []string{"indexer_take_in", "indexer_placement_reread"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // No streams, no bus, no seat (ADR 0437 §4); its one event is placement-changed (ADR 0549 §2), checked above. diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 04bca8c..2a2437a 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -3,7 +3,8 @@ "version": "1", "slug": "indexer", "tools": [ - "indexer_take_in" + "indexer_take_in", + "indexer_placement_reread" ], "emits": [ "placement-changed" @@ -19,6 +20,7 @@ { "name": "manifests", "history": 1 }, { "name": "modules", "history": 20 }, { "name": "placement", "history": 10 }, + { "name": "placement-candidates", "history": 5 }, { "name": "taken", "history": 1 }, { "name": "takeins", "history": 1 } ], @@ -33,7 +35,9 @@ "invokes": [ "gitea.gitea_commit_trees", "gitea.gitea_contains", - "gitea.gitea_get_file" + "gitea.gitea_get_file", + "gitea.gitea_list_branches", + "gitea.gitea_open_pulls" ], "own-secrets": { "broker": "${dir:state}/broker" diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index 84f83c0..4838b52 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -65,7 +65,9 @@ func Tools(r view.Reader, u view.UsageReader) []stdio.Tool { {Name: view.PlacementReadTool, Description: "Where modules run, as the indexer read it from the catalogue's placement files at the last merge (placement., novox/hq ADR 0549): " + "each node's modules in order with their host ports, bindings and settings placement, its pins and the seats its modules hold, and the catalogue revision. " + - "With node, that node's; with module, every node placing it; with neither, every node's. Keys that do not read are listed as unreadable. Read-only.", + "With node, that node's; with module, every node placing it; with neither, every node's. Keys that do not read are listed as unreadable. " + + "Beside it, candidates: each open catalogue pull request's placement of a node that differs from the trunk's (placement-candidates ., issue 1126), " + + "narrowed the same way, with the node's answer from its .node have (answered: same, differs, refused, stale for an earlier head, none when it has not answered). Read-only.", Input: map[string]any{"type": "object", "properties": nodeModule}, Run: func(args map[string]any) (any, error) { node, module := nodeModuleOf(args) @@ -202,8 +204,9 @@ func read(state stdio.KeptState) view.StateRead { func main() { r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates")), WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")), - ReachState: read(stdio.State("reach")), - PlacementState: read(stdio.State("mesh-delivery-indexer.placement"))} + ReachState: read(stdio.State("reach")), + PlacementState: read(stdio.State("mesh-delivery-indexer.placement")), + PlacementCandidatesState: read(stdio.State("mesh-delivery-indexer.placement-candidates"))} u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")), Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())} for _, lang := range delivery.AskLanguages { diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go index 51ca95a..3e91541 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go @@ -131,6 +131,51 @@ func TestModulesToolAnswersEveryKeyAsStored(t *testing.T) { if _, err := run(t, p, view.PlacementReadTool, map[string]any{"node": "novox"}); err == nil || !strings.Contains(err.Error(), "no placement of novox") { t.Errorf("a node with no key: %v", err) } + if c := out.(map[string]any)["candidates"].([]view.CandidateRow); len(c) != 0 { + t.Errorf("with no candidate open: %v", c) + } + + // ADR 0559 §7 (issue 1126): beside the trunk's, each open candidate key and the node's answer from its .node: + // answered at the key's head and revision, at an earlier head (stale), or not at all (none); a node the trunk does not + // place yet is shown by its candidate alone; ace, placed on the trunk with its file unchanged (no key of its own), is + // shown with its answer, counted by the head alone. + const h1, h2 = "1111111111111111111111111111111111111111", "2222222222222222222222222222222222222222" + cand := func(node, head string, placed bool) string { + pl := `,"placement":{"v":1,"node":"` + node + `","modules":[{"module":"plex"}]}` + if !placed { + pl = "" + } + return `{"v":1,"group":"feat/p","pull":{"repository":"novox/mesh-catalog","number":9,"head":"` + head + `"}` + pl + `}` + } + cands := fake{"feat/p.g14": cand("g14", h2, false), "feat/p.novox": cand("novox", h2, true)} + revs := map[string]uint64{"feat/p.g14": 5, "feat/p.novox": 6} + p.PlacementCandidatesState = view.StateRead{Keys: cands.reader().State.Keys, GetEntry: func(key string) (json.RawMessage, uint64, error) { + raw, _ := cands.reader().State.Get(key) + return raw, revs[key], nil + }} + p.HavesState = fake{ + "ace.node": `{"v":1,"composed":"x","candidates":{"feat/p":{"head":"` + h2 + `","verdict":"same","same":["plex"]}}}`, + "g14.node": `{"v":1,"composed":"x","candidates":{"feat/p":{"head":"` + h1 + `","revision":2,"verdict":"same"}}}`, + }.reader().State + out, err = run(t, p, view.PlacementReadTool, map[string]any{}) + if err != nil { + t.Fatal(err) + } + answered := map[string]string{} + for _, c := range out.(map[string]any)["candidates"].([]view.CandidateRow) { + answered[c.Node] = c.Answered + } + if want := map[string]string{"ace": "same", "g14": view.AnswerStale, "novox": view.AnswerNone}; fmt.Sprint(answered) != fmt.Sprint(want) { + t.Errorf("answers %v, want %v", answered, want) + } + out, err = run(t, p, view.PlacementReadTool, map[string]any{"node": "novox"}) + if c := out.(map[string]any)["candidates"].([]view.CandidateRow); err != nil || len(c) != 1 || c[0].Pull.Head != h2 || len(out.(map[string]any)["placement"].(map[string]json.RawMessage)) != 0 { + t.Errorf("a node placed by its candidate alone: %v, %v", out, err) + } + out, _ = run(t, p, view.PlacementReadTool, map[string]any{"module": "plex"}) + if c := out.(map[string]any)["candidates"].([]view.CandidateRow); len(c) != 2 { + t.Errorf("the candidates placing plex: %v (g14's removes its file)", c) + } } const ( diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index a1dec0c..8609300 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-view go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-view/go.sum b/modules/mesh-delivery-view/go.sum index b3dab4f..9259a48 100644 --- a/modules/mesh-delivery-view/go.sum +++ b/modules/mesh-delivery-view/go.sum @@ -1,5 +1,17 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/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= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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= diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index 0aea653..d7e3ae3 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -7,6 +7,7 @@ package view import ( "encoding/json" "fmt" + "maps" "slices" "sort" "strings" @@ -43,6 +44,8 @@ type Reader struct { HavesState StateRead // the node-engine's haves ReachState StateRead // the view's own reach (ADR 0556) PlacementState StateRead // the indexer's placement + // PlacementCandidatesState is the indexer's placement candidates (issue 1126); unset reads none. + PlacementCandidatesState StateRead } // ReachReadTool reads the view's reach: per repository, how far each module reached and the reached head (ADR 0556). @@ -652,11 +655,19 @@ func (r Reader) Haves(node, module string) (KeptAnswer, error) { // Placement is the indexer's placement as stored (ADR 0549 §2): one node's key with node, every node placing a module // with module (each node's key whole, so its pins, ports and bindings are beside the module), and every node's key with // neither. A key that does not read as a placement is listed as unreadable, never dropped. +// +// Beside it, candidates (issue 1126, ADR 0549 §4): each open catalogue pull request's placement of a node that differs from +// the trunk's (the indexer's placement-candidates `.`), narrowed the same way, each with the node's answer from +// its `.node` have and whether that answer is for the candidate as it stands (its head and key revision). func (r Reader) Placement(node, module string) (map[string]any, error) { node, module = strings.TrimSpace(node), strings.TrimSpace(module) if r.PlacementState.Keys == nil || r.PlacementState.Get == nil { return nil, fmt.Errorf("the placement state cannot be read here") } + candidates, err := r.PlacementCandidates(node, module) + if err != nil { + return nil, err + } if node != "" { key, err := delivery.PlacementKey(node) if err != nil { @@ -666,10 +677,14 @@ func (r Reader) Placement(node, module string) (map[string]any, error) { if err != nil { return nil, err } - if len(raw) == 0 { + if len(raw) == 0 && len(candidates) == 0 { return nil, fmt.Errorf("no placement of %s: the catalogue holds no placement file for it, or the indexer has read none yet", node) } - return map[string]any{"placement": map[string]json.RawMessage{node: raw}}, nil + out := map[string]json.RawMessage{} + if len(raw) > 0 { + out[node] = raw + } + return map[string]any{"placement": out, "candidates": candidates}, nil } keys, err := r.PlacementState.Keys() if err != nil { @@ -698,5 +713,168 @@ func (r Reader) Placement(node, module string) (map[string]any, error) { } out[n] = values[i] } - return map[string]any{"placement": out, "unreadable": unreadable}, nil + return map[string]any{"placement": out, "unreadable": unreadable, "candidates": candidates}, nil +} + +// The states of a node's answer to a placement candidate, as the view and the assigner's check read it. +const ( + // AnswerNone: the node's have holds no answer for the candidate's group (a node asleep, or not composing yet). + AnswerNone = "none" + // AnswerStale: the node answered an earlier head or revision of the candidate. + AnswerStale = "stale" +) + +// CandidateRow is one placement candidate and the node's answer to it. +type CandidateRow struct { + Group string `json:"group"` + Node string `json:"node"` + Pull delivery.Head `json:"pull"` + Revision uint64 `json:"revision"` + Placement json.RawMessage `json:"placement,omitempty"` // the candidate value as stored + Removes bool `json:"removes,omitempty"` // the head removes the node's file + Answered string `json:"answered"` // same, differs, refused, stale or none + Answer *delivery.CandidateAnswer `json:"answer,omitempty"` +} + +// PlacementCandidates is every placement candidate, narrowed by node and by a module its placement names, with each +// node's answer; beside each group's keys, a row for every other node placed at the group's head or on the trunk, whose +// answer counts by the head alone (its file is unchanged, so it has no key, ADR 0559 §5, §7). A key that does not read is +// a row naming only its key's group and node, answered none, never dropped. +func (r Reader) PlacementCandidates(node, module string) ([]CandidateRow, error) { + st := r.PlacementCandidatesState + if st.Keys == nil || st.GetEntry == nil { + return []CandidateRow{}, nil + } + keys, err := st.Keys() + if err != nil { + return nil, fmt.Errorf("reading the placement candidates: %w", err) + } + sort.Strings(keys) + // The trunk's placed nodes, each with its modules. + trunk := map[string][]string{} + if r.PlacementState.Keys != nil && r.PlacementState.Get != nil { + if tkeys, err := r.PlacementState.Keys(); err == nil { + for _, k := range tkeys { + n, err := delivery.ParsePlacementKey(k) + if err != nil { + continue + } + if raw, err := r.PlacementState.Get(k); err == nil && len(raw) > 0 { + if p, err := delivery.DecodePlacement(raw); err == nil { + trunk[n] = p.Assigned() + } + } + } + } + } + type group struct { + pull delivery.Head + keyed map[string]bool + placed map[string]bool + } + groups := map[string]*group{} + answers := map[string]map[string]delivery.CandidateAnswer{} // node, then group + answerOf := func(n, g string) (delivery.CandidateAnswer, bool) { + if _, read := answers[n]; !read { + answers[n] = r.candidateAnswers(n) + } + a, ok := answers[n][g] + return a, ok + } + rows := []CandidateRow{} + for _, key := range keys { + gname, n, err := delivery.ParsePlacementCandidateKey(key) + if err != nil { + continue + } + raw, rev, err := st.GetEntry(key) + if err != nil { + return nil, fmt.Errorf("reading %s: %w", key, err) + } + if len(raw) == 0 { + continue + } + g := groups[gname] + if g == nil { + g = &group{keyed: map[string]bool{}, placed: map[string]bool{}} + groups[gname] = g + } + g.keyed[n] = true + row := CandidateRow{Group: gname, Node: n, Revision: rev, Placement: raw, Answered: AnswerNone} + c, err := delivery.DecodePlacementCandidate(raw) + if err == nil { + g.pull = c.Pull + if c.Placement != nil { + g.placed[n] = true + } + row.Pull, row.Removes = c.Pull, c.Placement == nil + if module != "" && (c.Placement == nil || !slices.Contains(c.Placement.Assigned(), module)) { + continue + } + } else if module != "" { + continue + } + if node != "" && n != node { + continue + } + if a, ok := answerOf(n, gname); ok && err == nil { + row.Answer = &a + row.Answered = AnswerOf(c.Pull.Head, rev, true, a) + } + rows = append(rows, row) + } + for _, gname := range slices.Sorted(maps.Keys(groups)) { + g := groups[gname] + if g.pull.Head == "" { + continue + } + others := map[string]bool{} + for n := range trunk { + others[n] = true + } + for n := range g.placed { + others[n] = true + } + for _, n := range slices.Sorted(maps.Keys(others)) { + if g.keyed[n] || (node != "" && n != node) || (module != "" && !slices.Contains(trunk[n], module)) { + continue + } + row := CandidateRow{Group: gname, Node: n, Pull: g.pull, Answered: AnswerNone} + if a, ok := answerOf(n, gname); ok { + row.Answer = &a + row.Answered = AnswerOf(g.pull.Head, 0, false, a) + } + rows = append(rows, row) + } + } + return rows, nil +} + +// AnswerOf is the state of a node's answer a to its group's candidates at head: its verdict when it answers that head +// and, for a node with a key of its own (keyed), that key's revision rev; AnswerStale otherwise. +func AnswerOf(head string, rev uint64, keyed bool, a delivery.CandidateAnswer) string { + if a.Head != head || (keyed && a.Revision != rev) { + return AnswerStale + } + return a.Verdict +} + +// candidateAnswers is a node's answers to its candidates, by group, from its `.node` have; none when it has none. +func (r Reader) candidateAnswers(node string) map[string]delivery.CandidateAnswer { + if r.HavesState.Get == nil { + return nil + } + key, err := delivery.HaveNodeKey(node) + if err != nil { + return nil + } + raw, err := r.HavesState.Get(key) + if err != nil || len(raw) == 0 { + return nil + } + h, err := delivery.DecodeHaveNode(raw) + if err != nil { + return nil + } + return h.Candidates } diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index ba1fa65..1ad60db 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -87,7 +87,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if m.Module != "mesh-delivery-view" { t.Errorf("module = %q", m.Module) } - want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 + want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-indexer.placement-candidates", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // indexer_placement reads the placement candidates and every .node (ADR 0559 §7); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 got := slices.Clone(m.Reads) slices.Sort(got) if !reflect.DeepEqual(got, want) { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 41f42f7..cbbeaa4 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -49,6 +49,7 @@ "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", + "mesh-delivery-indexer.placement-candidates", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", diff --git a/modules/mesh-grants/go.mod b/modules/mesh-grants/go.mod index 19d6941..b6cf662 100644 --- a/modules/mesh-grants/go.mod +++ b/modules/mesh-grants/go.mod @@ -3,7 +3,7 @@ module mesh-grants 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.20261027014310-cadd7b1b2f8f github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-grants/go.sum b/modules/mesh-grants/go.sum index 8035b76..f93723f 100644 --- a/modules/mesh-grants/go.sum +++ b/modules/mesh-grants/go.sum @@ -1,5 +1,9 @@ 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.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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= diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 3e72e77..4a0aaa6 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -457,6 +457,24 @@ func TestGrants(t *testing.T) { } }) } + // ADR 0559 §3 (issue 1126): the indexer's placement candidates are read by every node agent (mesh-sdk + // NodeComposeReads), the assigner and the status view, each by its manifest's reads, and by no other module. + cands := fixture() + cands.manifestOf("mesh-delivery-indexer", `{"module":"mesh-delivery-indexer","version":"1","own-secrets":{"broker":"/var/lib/i/broker"},"state":[{"name":"placement-candidates","history":5}]}`) + cands.manifestOf("mesh-delivery-assigner", `{"module":"mesh-delivery-assigner","version":"1","own-secrets":{"broker":"/var/lib/as/broker"},"reads":["mesh-delivery-indexer.placement-candidates"]}`) + cands.manifestOf("mesh-delivery-view", `{"module":"mesh-delivery-view","version":"1","own-secrets":{"broker":"/var/lib/v/broker"},"reads":["mesh-delivery-indexer.placement-candidates"]}`) + for _, m := range []string{"mesh-delivery-indexer", "mesh-delivery-assigner", "mesh-delivery-view"} { + cands.assign("a", m) + } + var readers []string + for _, u := range shadowOf(t, cands) { + if slices.Contains(u.Publish, "$JS.API.DIRECT.GET.KV_"+delivery.PlacementCandidatesBucket+".>") { + readers = append(readers, u.User) + } + } + if want := []string{"a.mesh-delivery-assigner", "a.mesh-delivery-indexer", "a.mesh-delivery-view", "node.a", "node.b"}; !slices.Equal(readers, want) { + t.Errorf("the placement candidates are read by %v, want the node agents, the assigner, the view and their writer %v", readers, want) + } b := fixture() users := shadowOf(t, b) y := users[slices.IndexFunc(users, func(u grants.User) bool { return u.User == "b.mod-y" })]