Merge pull request 'A placement pull request is carried, kept as candidates, and checked against every placed node's answer before it merges (issue 1126, ADR 0559)' (#85) from feat/1126-l2b into main

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