Compare commits
27
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2259d98c5b | ||
|
|
4702c6bd77 | ||
|
|
23f4304d70 | ||
|
|
c256d732f4 | ||
|
|
93427c08e7 | ||
|
|
c58342786f | ||
|
|
840e56eddb | ||
|
|
90fd07ac26 | ||
|
|
90a79b1585 | ||
|
|
45ab994777 | ||
|
|
55f486e040 | ||
|
|
7a88eb0a56 | ||
|
|
9261347a8f | ||
|
|
263ff54a4e | ||
|
|
cd3e55e056 | ||
|
|
b303e32a41 | ||
|
|
a31bd7d4a5 | ||
|
|
32542ebc14 | ||
|
|
a697ee86d4 | ||
|
|
0373a3b24e | ||
|
|
fae4d48e09 | ||
|
|
b2c545c40a | ||
|
|
85aca46348 | ||
|
|
db5d88f893 | ||
|
|
24753bfc2c | ||
|
|
1dc9d80cfb | ||
|
|
26bfdbb637 |
@@ -0,0 +1,161 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
// AssignerModule is the asker the assigner's compose asks name.
|
||||
const AssignerModule = "mesh-delivery-assigner"
|
||||
|
||||
// composeAsks are the compose asks the assigner put on the Go build seat's check stream (novox/hq ADR 0482 clause 4,
|
||||
// issues 1622 and 1623), by what each composes: one ask per key, put once and followed by a watch of its step key until
|
||||
// it ends, which wakes a pass. No timer: an ask that waits in the stream waits (its backlog is the view's and the
|
||||
// watchman's to show). A restart forgets them and asks again; the build module keeps a composer built for a commit, so a
|
||||
// second ask builds nothing.
|
||||
type composeAsks struct {
|
||||
ctx context.Context
|
||||
asker buildrun.Asker
|
||||
wake func()
|
||||
logf func(string, ...any)
|
||||
now func() time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
asks map[string]*composeAsk
|
||||
}
|
||||
|
||||
type composeAsk struct {
|
||||
id string
|
||||
cancel context.CancelFunc
|
||||
ran *assign.Ran
|
||||
}
|
||||
|
||||
func newComposeAsks(ctx context.Context, asker buildrun.Asker, wake func(), logf func(string, ...any)) *composeAsks {
|
||||
return &composeAsks{ctx: ctx, asker: asker, wake: wake, logf: logf, now: time.Now, asks: map[string]*composeAsk{}}
|
||||
}
|
||||
|
||||
// get is the end of the ask that composes key, or nil and its id while it runs. The first get of a key puts the ask.
|
||||
// A job its layout refuses ends at once, failed with why: never silent.
|
||||
func (a *composeAsks) get(key string, job delivery.ComposeJob) (*assign.Ran, string) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
if k, ok := a.asks[key]; ok {
|
||||
return k.ran, k.id
|
||||
}
|
||||
ask := delivery.StepAsk{ID: buildstep.NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: AssignerModule,
|
||||
At: a.now(), Compose: &job}
|
||||
ctx, cancel := context.WithCancel(a.ctx)
|
||||
k := &composeAsk{id: ask.ID, cancel: cancel}
|
||||
a.asks[key] = k
|
||||
if err := a.asker.Put(ctx, ask); err != nil {
|
||||
a.logf("the compose ask %s for %s could not be put: %v", ask.ID, key, err)
|
||||
k.ran = &assign.Ran{State: delivery.StepFailed, Reason: "the compose ask could not be put on the Go build seat: " + err.Error()}
|
||||
cancel()
|
||||
return k.ran, k.id
|
||||
}
|
||||
go func() {
|
||||
step, err := a.asker.Wait(ctx, ask)
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
ran := assign.RanOf(step)
|
||||
if err != nil {
|
||||
ran = assign.Ran{State: delivery.StepFailed, Reason: "the compose ask's step could not be followed: " + err.Error()}
|
||||
}
|
||||
a.mu.Lock()
|
||||
k.ran = &ran
|
||||
a.mu.Unlock()
|
||||
a.wake()
|
||||
}()
|
||||
return nil, k.id
|
||||
}
|
||||
|
||||
// sweep forgets, and stops following, every ask whose key no pass asked for this time: a candidate gone or moved on.
|
||||
func (a *composeAsks) sweep(seen map[string]bool) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
for key, k := range a.asks {
|
||||
if !seen[key] {
|
||||
k.cancel()
|
||||
delete(a.asks, key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// composeAtHead is the composition of a group with a mesh-sdk member: the composer at its head and at each reader's
|
||||
// mesh-sdk, asked of the Go build seat, and the verdict once every ask ended; until then the verdict waits, naming the
|
||||
// asks. The keys it used are marked in seen.
|
||||
func (p *passer) composeAtHead(ctx context.Context, m assign.Mesh, ch assign.Change, seen map[string]bool, began time.Time) delivery.CompositionVerdict {
|
||||
waiting := func(words string) delivery.CompositionVerdict {
|
||||
return delivery.CompositionVerdict{Group: ch.Group, Heads: ch.Heads, State: delivery.CompositionWaits, Words: words}
|
||||
}
|
||||
head, err := assign.HeadJob(m, ch)
|
||||
if err != nil {
|
||||
return delivery.CompositionVerdict{Group: ch.Group, Heads: ch.Heads, State: delivery.CompositionFail, SDK: "mesh-sdk " + ch.SDKHead,
|
||||
Words: "the compose ask at the group's mesh-sdk head cannot be made: " + err.Error()}
|
||||
}
|
||||
var pending []string
|
||||
key := "head." + ch.Group + "." + ch.Heads + "." + ch.SDKHead
|
||||
seen[key] = true
|
||||
headRan, id := p.asks.get(key, head)
|
||||
if headRan == nil {
|
||||
pending = append(pending, "mesh-sdk "+ch.SDKHead+" (ask "+id+")")
|
||||
}
|
||||
var readers []assign.Reader
|
||||
var notJudged []string
|
||||
for _, module := range sortedNames(assign.ReaderModules) {
|
||||
repository := assign.ReaderModules[module]
|
||||
var commits []string
|
||||
member := false
|
||||
if h := ch.Members[repository]; h != "" {
|
||||
commits, member = []string{h}, true
|
||||
} else {
|
||||
commits = p.w.runningCommits(ctx, kvGetter(p.kvs[delivery.BuildsBucket]), module)
|
||||
}
|
||||
if len(commits) == 0 {
|
||||
notJudged = append(notJudged, "not judged whether "+module+" reads the group's manifests: no build of it is assigned with a "+
|
||||
"commit on record, and the group holds no member of "+repository)
|
||||
continue
|
||||
}
|
||||
for _, commit := range commits {
|
||||
job, err := assign.ReaderJob(ch, module, repository, commit)
|
||||
if err != nil {
|
||||
readers = append(readers, assign.Reader{Module: module, Commit: commit, Member: member,
|
||||
Ran: assign.Ran{State: delivery.StepFailed, Reason: err.Error()}})
|
||||
continue
|
||||
}
|
||||
rkey := "reader." + ch.Group + "." + ch.Heads + "." + module + "." + commit
|
||||
seen[rkey] = true
|
||||
ran, rid := p.asks.get(rkey, job)
|
||||
if ran == nil {
|
||||
pending = append(pending, module+" at "+commit+" (ask "+rid+")")
|
||||
continue
|
||||
}
|
||||
readers = append(readers, assign.Reader{Module: module, Commit: commit, Member: member, Ran: *ran})
|
||||
}
|
||||
}
|
||||
if len(pending) > 0 {
|
||||
return waiting("composing on the Go build seat with the mesh-sdk of " + strings.Join(pending, ", "))
|
||||
}
|
||||
v := assign.FromComposer(m, ch, *headRan, readers, time.Since(began))
|
||||
v.Warnings = append(v.Warnings, notJudged...)
|
||||
return v
|
||||
}
|
||||
|
||||
func sortedNames(m map[string]string) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
+180
@@ -0,0 +1,180 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/composer"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
|
||||
// holder is a test holder of the Go build seat: it takes every ask from the seat's check worker, as mesh-build-go does,
|
||||
// and answers a compose ask by running mesh-sdk's composer over the request filled from the indexer's manifests (the
|
||||
// linked composer standing for the one the build module builds at the asked commit), writing its step on its node's key.
|
||||
type holder struct {
|
||||
mu sync.Mutex
|
||||
asks []delivery.StepAsk
|
||||
}
|
||||
|
||||
func (h *holder) taken() []delivery.StepAsk {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return append([]delivery.StepAsk(nil), h.asks...)
|
||||
}
|
||||
|
||||
func (b *bed) hold(t *testing.T) *holder {
|
||||
t.Helper()
|
||||
seat, _ := delivery.AskStream(delivery.AskGo)
|
||||
worker, _ := delivery.AskWorker(delivery.AskGo, delivery.AskCheck)
|
||||
subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
|
||||
cons, err := b.js.CreateOrUpdateConsumer(b.ctx, seat, jetstream.ConsumerConfig{Durable: worker, FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
steps := b.buckets[delivery.StepsBucket(goBuildModule)]
|
||||
h := &holder{}
|
||||
cc, err := cons.Consume(func(m jetstream.Msg) {
|
||||
a, err := delivery.DecodeStepAsk(m.Data())
|
||||
if err != nil || a.Compose == nil {
|
||||
t.Errorf("the seat was put an ask that is not a compose: %v %s", err, m.Data())
|
||||
m.Ack()
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.asks = append(h.asks, a)
|
||||
h.mu.Unlock()
|
||||
req, err := composer.Decode(a.Compose.Request)
|
||||
if err == nil {
|
||||
err = composer.Fill(&req, func(digest string) ([]byte, error) {
|
||||
key, _ := delivery.ManifestsKey(digest)
|
||||
e, err := b.buckets[delivery.ManifestsBucket].Get(b.ctx, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
man, err := delivery.DecodeManifest(e.Value())
|
||||
return []byte(man.JSON), err
|
||||
})
|
||||
}
|
||||
step := delivery.Step{ID: a.ID, Tool: "compose", Node: "ace", State: delivery.StepPassed, Started: time.Now().UTC(), Ended: time.Now().UTC()}
|
||||
if err != nil {
|
||||
step.State, step.Reason = delivery.StepFailed, err.Error()
|
||||
} else if v, err := composer.Run(req); err != nil {
|
||||
step.State, step.Reason = delivery.StepFailed, err.Error()
|
||||
} else {
|
||||
vr, _ := json.Marshal(v)
|
||||
step.Result, _ = json.Marshal(delivery.ComposeAnswer{Built: true, Key: "composer-0123456789abcdef", Verdict: vr})
|
||||
}
|
||||
raw, err := delivery.EncodeStep(step)
|
||||
if err != nil {
|
||||
t.Errorf("the step cannot be written: %v", err)
|
||||
}
|
||||
key, _ := delivery.AskStepKey("ace", a.ID)
|
||||
if _, err := steps.Put(b.ctx, key, raw); err != nil {
|
||||
t.Errorf("the step %s: %v", key, err)
|
||||
}
|
||||
m.Ack()
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(cc.Stop)
|
||||
return h
|
||||
}
|
||||
|
||||
// 1622/AC1-AC3 and 1623 on the bus: started whole, the assigner composes a group holding a mesh-sdk pull request by
|
||||
// asking the Go build seat for the composer at the member's head, and for the composer at the mesh-sdk the running
|
||||
// controller pins; it waits naming the asks while they run, writes the verdict from their answers naming the mesh-sdk the
|
||||
// head's composer was built with, and asks nothing more on a later pass of the same heads.
|
||||
func TestAGroupWithAnSDKMemberIsComposedByTheComposerAtItsHead(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
trunk := b.manifest(t, "web", `{"module":"web","version":"1"}`)
|
||||
changed := b.manifest(t, "web", `{"module":"web","version":"2"}`)
|
||||
mod, err := delivery.EncodeModule(delivery.Module{FP2: delivery.SourcePrefix2 + strings.Repeat("a", 64), Manifest: trunk, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.putJSON(t, delivery.ModulesBucket, "web", mod)
|
||||
b.putJSON(t, provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"build-1"}`))
|
||||
running := strings.Repeat("3", 40)
|
||||
b.putJSON(t, provider.AssignmentsBucket, "novox.mesh-controller", []byte(`{"node":"novox","module":"mesh-controller","build":"c-1"}`))
|
||||
b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("mesh-controller", "c-1"), []byte(`{"id":"c-1","repository":"novox/mesh-controller","commit":"`+running+`"}`))
|
||||
sdkHead := strings.Repeat("4", 40)
|
||||
heads := []delivery.Head{{Repository: "novox/mesh-sdk", Number: 80, Head: sdkHead}, {Repository: "novox/web", Number: 7, Head: strings.Repeat("2", 40)}}
|
||||
hash, _ := delivery.HeadsHash(heads)
|
||||
pulls := []delivery.CandidatePull{{Head: heads[0]}, {Head: heads[1],
|
||||
Modules: []string{"web"}, Fps: map[string]delivery.CandidateModule{"web": {FP2: delivery.SourcePrefix2 + strings.Repeat("b", 64), Manifest: changed}}}}
|
||||
cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: "feat/x", HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ckey, _ := delivery.CandidatesKey("feat/x")
|
||||
b.putJSON(t, delivery.CandidatesBucket, ckey, cand)
|
||||
b.start(t)
|
||||
key, _ := delivery.CompositionKey("feat/x", hash)
|
||||
// No holder takes the asks yet: the composition waits, naming them.
|
||||
waits := b.verdict(t, key, func(v delivery.CompositionVerdict) bool { return v.State == delivery.CompositionWaits })
|
||||
if !strings.Contains(waits.Words, "mesh-sdk "+sdkHead) || !strings.Contains(waits.Words, "mesh-controller at "+running) {
|
||||
t.Fatalf("the waiting composition says %q, want the head's and the running controller's asks", waits.Words)
|
||||
}
|
||||
h := b.hold(t)
|
||||
v := b.verdict(t, key, func(v delivery.CompositionVerdict) bool { return v.State != delivery.CompositionWaits })
|
||||
if v.State != delivery.CompositionPass || len(v.Machines) != 1 || v.Machines[0] != "g14" || v.SDK == "" {
|
||||
t.Fatalf("composition = %+v, want a pass on g14", v)
|
||||
}
|
||||
asks := h.taken()
|
||||
if len(asks) != 2 {
|
||||
t.Fatalf("the seat was asked %d composes, want two (the head, the running controller)", len(asks))
|
||||
}
|
||||
byRepo := map[string]delivery.StepAsk{}
|
||||
for _, a := range asks {
|
||||
byRepo[a.Compose.Fetch["repository"].(string)] = a
|
||||
if a.Asker != AssignerModule || a.Kind != delivery.AskCheck || a.Language != delivery.AskGo {
|
||||
t.Errorf("ask = %+v", a)
|
||||
}
|
||||
}
|
||||
if a := byRepo["novox/mesh-sdk"]; a.Compose == nil || a.Compose.Fetch["commit"] != sdkHead || a.Compose.Dir != "go" {
|
||||
t.Errorf("the head's ask = %+v", a.Compose)
|
||||
}
|
||||
if a := byRepo["novox/mesh-controller"]; a.Compose == nil || a.Compose.Fetch["commit"] != running || a.Compose.Dir != delivery.RootPath {
|
||||
t.Errorf("the running controller's ask = %+v", a.Compose)
|
||||
}
|
||||
// A later pass of the same heads (woken by any change) asks nothing more.
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
b.putJSON(t, delivery.PresenceBucket, "ace", on)
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
if n := len(h.taken()); n != 2 {
|
||||
t.Errorf("a later pass of the same heads asked again: %d asks", n)
|
||||
}
|
||||
}
|
||||
|
||||
// verdict waits for the composition under key to satisfy ok, and answers it.
|
||||
func (b *bed) verdict(t *testing.T, key string, ok func(delivery.CompositionVerdict) bool) delivery.CompositionVerdict {
|
||||
t.Helper()
|
||||
var last delivery.CompositionVerdict
|
||||
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
e, err := b.buckets[delivery.AssignerCompositionsBucket].Get(b.ctx, key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if last, err = delivery.DecodeCompositionVerdict(e.Value()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ok(last) {
|
||||
return last
|
||||
}
|
||||
}
|
||||
t.Fatalf("no composition under %s as wanted after 10 s; last %+v", key, last)
|
||||
return last
|
||||
}
|
||||
@@ -73,6 +73,12 @@ func startBed(t *testing.T, leaveOut string) *bed {
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The Go build seat's work queue, as the broker derives it from the seat's accepted kinds (delivery.AskStream).
|
||||
seat, _ := delivery.AskStream(delivery.AskGo)
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: seat, Subjects: []string{"mesh.seat.node-build-go.accept.>"},
|
||||
Retention: jetstream.WorkQueuePolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
@@ -181,8 +187,10 @@ func TestAChangeIsComposedOnTheBus(t *testing.T) {
|
||||
b.putJSON(t, provider.AssignmentsBucket, "ace.other", []byte(`{"node":"ace","module":"other","build":"build-2"}`))
|
||||
heads := []delivery.Head{{Repository: "novox/web", Number: 7, Head: strings.Repeat("2", 40)}}
|
||||
hash, _ := delivery.HeadsHash(heads)
|
||||
cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: "feat/x", HeadsHash: hash, Pulls: []delivery.CandidatePull{{Head: heads[0],
|
||||
Modules: []string{"web"}, Fps: map[string]delivery.CandidateModule{"web": {FP2: delivery.SourcePrefix2 + strings.Repeat("b", 64), Manifest: changed}}}}})
|
||||
// 1927/AC1: the candidate carries the fingerprint set the SDK computes, as the indexer writes it.
|
||||
pulls := []delivery.CandidatePull{{Head: heads[0],
|
||||
Modules: []string{"web"}, Fps: map[string]delivery.CandidateModule{"web": {FP2: delivery.SourcePrefix2 + strings.Repeat("b", 64), Manifest: changed}}}}
|
||||
cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: "feat/x", HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -211,6 +219,13 @@ func TestAChangeIsComposedOnTheBus(t *testing.T) {
|
||||
if took := time.Since(began); took > 5*time.Second {
|
||||
t.Errorf("the composition took %s", took)
|
||||
}
|
||||
// 1622/AC3: a group without a mesh-sdk member builds no composer: no ask is put on the Go build seat.
|
||||
seat, _ := delivery.AskStream(delivery.AskGo)
|
||||
if st, err := b.js.Stream(b.ctx, seat); err != nil {
|
||||
t.Fatal(err)
|
||||
} 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)
|
||||
}
|
||||
}
|
||||
|
||||
// get waits for a key of one of the bed's buckets and answers its value.
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
@@ -47,8 +48,13 @@ var readBuckets = []string{
|
||||
delivery.ModulesBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.StepsBucket(goBuildModule),
|
||||
}
|
||||
|
||||
// goBuildModule is the Go build module, which takes the assigner's compose asks on its seat's check stream and writes
|
||||
// their steps in its steps state (novox/hq ADR 0482 clause 4).
|
||||
const goBuildModule = "mesh-build-go"
|
||||
|
||||
// ownBuckets are the assigner's own states, as its manifest declares them.
|
||||
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerWantsBucket}
|
||||
|
||||
@@ -124,7 +130,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err)
|
||||
}
|
||||
logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
|
||||
return follow(ctx, kvs, logger)
|
||||
return follow(ctx, js, kvs, logger)
|
||||
}
|
||||
|
||||
// watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and
|
||||
@@ -145,7 +151,7 @@ var watched = []string{
|
||||
|
||||
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
|
||||
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
|
||||
func follow(ctx context.Context, kvs buckets, logger *log.Logger) error {
|
||||
func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, logger *log.Logger) error {
|
||||
w := newWorld()
|
||||
dirty := make(chan struct{}, 1)
|
||||
wake := func() {
|
||||
@@ -210,6 +216,7 @@ func follow(ctx context.Context, kvs buckets, logger *log.Logger) error {
|
||||
}
|
||||
}
|
||||
p := &passer{kvs: kvs, w: w, logf: logger.Printf}
|
||||
p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
|
||||
wake()
|
||||
for {
|
||||
select {
|
||||
|
||||
@@ -19,6 +19,7 @@ type passer struct {
|
||||
w *world
|
||||
logf func(string, ...any)
|
||||
sdk string
|
||||
asks *composeAsks
|
||||
}
|
||||
|
||||
func (p *passer) pass(ctx context.Context) {
|
||||
@@ -162,6 +163,10 @@ func (p *passer) compositions(ctx context.Context) {
|
||||
written := p.w.read(delivery.AssignerCompositionsBucket)
|
||||
trunk := p.w.trunk()
|
||||
var mesh *assign.Mesh
|
||||
seen := map[string]bool{}
|
||||
if p.asks != nil {
|
||||
defer p.asks.sweep(seen)
|
||||
}
|
||||
for key, value := range p.w.read(delivery.CandidatesBucket) {
|
||||
c, err := delivery.DecodeCandidate(value)
|
||||
if err != nil {
|
||||
@@ -183,7 +188,12 @@ func (p *passer) compositions(ctx context.Context) {
|
||||
m := p.w.mesh(ctx, kvGetter(p.kvs[delivery.ManifestsBucket]))
|
||||
mesh = &m
|
||||
}
|
||||
v = assign.Compose(*mesh, ch, p.sdk, time.Since(began))
|
||||
if ch.SDKHead != "" && p.asks != nil {
|
||||
// A group with a mesh-sdk member is composed by the composer built at its head (ADR 0482 clauses 3 to 5).
|
||||
v = p.composeAtHead(ctx, *mesh, ch, seen, began)
|
||||
} else {
|
||||
v = assign.Compose(*mesh, ch, p.sdk, time.Since(began))
|
||||
}
|
||||
}
|
||||
if before, err := delivery.DecodeCompositionVerdict(written[out]); err == nil {
|
||||
before.Took, before.V = v.Took, 0
|
||||
|
||||
@@ -193,12 +193,13 @@ func (w *world) trunk() map[string]delivery.Module {
|
||||
|
||||
// mesh is the mesh as the trunk holds it, for the composition check.
|
||||
func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh {
|
||||
m := assign.Mesh{Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}}
|
||||
m := assign.Mesh{Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, Digests: map[string]string{}}
|
||||
m.Assigned, _ = w.assigned()
|
||||
for name, mod := range w.trunk() {
|
||||
if mod.Manifest == "" {
|
||||
continue
|
||||
}
|
||||
m.Digests[name] = mod.Manifest
|
||||
p := w.manifest(ctx, manifests, mod.Manifest)
|
||||
if p.err != "" {
|
||||
m.Unreadable[name] = p.err
|
||||
@@ -211,12 +212,14 @@ func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh {
|
||||
|
||||
// change is a candidate's change for the composition check, or why the composition waits for it.
|
||||
func (w *world) change(ctx context.Context, manifests getter, c delivery.Candidate, trunk map[string]delivery.Module) (assign.Change, string) {
|
||||
ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, Twice: map[string][]string{}}
|
||||
ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{},
|
||||
Twice: map[string][]string{}, Digests: map[string]string{}, Members: map[string]string{}}
|
||||
definedIn := map[string]map[string]bool{}
|
||||
for _, p := range c.Pulls {
|
||||
if p.Repository == assign.SDKRepository {
|
||||
ch.SDKHead = p.Head.Head
|
||||
}
|
||||
ch.Members[p.Repository] = p.Head.Head
|
||||
for _, name := range p.Modules {
|
||||
fp, ok := p.Fps[name]
|
||||
if !ok || fp.Manifest == "" {
|
||||
@@ -233,6 +236,7 @@ func (w *world) change(ctx context.Context, manifests getter, c delivery.Candida
|
||||
if strings.HasPrefix(got.err, "the indexer holds no manifest") {
|
||||
return ch, got.err
|
||||
}
|
||||
ch.Digests[name] = fp.Manifest
|
||||
if got.err != "" {
|
||||
ch.Unreadable[name] = got.err
|
||||
continue
|
||||
@@ -368,3 +372,31 @@ func (w *world) sentFP(ctx context.Context, get getter, module, build string) st
|
||||
w.mu.Unlock()
|
||||
return rec.SourceFingerprint
|
||||
}
|
||||
|
||||
// runningCommits is, of every open assignment of module, the commit of the build it names, read from the controller's
|
||||
// build record (`mesh-controller_builds` `<module>.build.<id>`), distinct and sorted: the commits a reader of the
|
||||
// manifests runs at (novox/hq ADR 0482 clause 5, issue 1623). A build whose record names no full commit is left out.
|
||||
func (w *world) runningCommits(ctx context.Context, get getter, module string) []string {
|
||||
_, all := w.assigned()
|
||||
seen := map[string]bool{}
|
||||
for _, a := range all {
|
||||
if a.Module != module || a.Build == "" {
|
||||
continue
|
||||
}
|
||||
value, err := get(ctx, delivery.BuildKey(module, a.Build))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var rec delivery.BuildRecord
|
||||
if json.Unmarshal(value, &rec) != nil || len(rec.Commit) != 40 {
|
||||
continue
|
||||
}
|
||||
seen[rec.Commit] = true
|
||||
}
|
||||
out := make([]string, 0, len(seen))
|
||||
for c := range seen {
|
||||
out = append(out, c)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-assigner
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024022512-dd0d156b9edc
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,11 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024015135-4784762a70ce h1:9pNnygyhuYIpVs0HPzgheGLNjPXigGGQpnFTp8lw+X4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024015135-4784762a70ce/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024021031-66bf26c0d721 h1:N//dDGmry1F3HKeBTWd+z/SfK+wb8bUUMJF0OlgR0/g=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024021031-66bf26c0d721/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024022512-dd0d156b9edc h1:MSywVviXmT3bR2pQLJLHErUGK8wCme8jEspS6x/j/Vo=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024022512-dd0d156b9edc/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=
|
||||
|
||||
@@ -32,6 +32,9 @@ type Mesh struct {
|
||||
Manifests map[string]catalogue.Manifest
|
||||
Unreadable map[string]string
|
||||
Assigned map[string][]string
|
||||
// Digests is every trunk module's manifest digest in the indexer's manifests bucket, readable or not: what a
|
||||
// composer at another mesh-sdk reads (issue 1622).
|
||||
Digests map[string]string
|
||||
}
|
||||
|
||||
// Change is a group's change as the indexer's candidate holds it: the manifest of every module it moves or adds, those
|
||||
@@ -45,6 +48,10 @@ type Change struct {
|
||||
Twice map[string][]string
|
||||
SDKHead string
|
||||
Rebuilt int
|
||||
// Digests is the manifest digest of every module the change moves or adds, readable by the assigner or not, and
|
||||
// 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
|
||||
}
|
||||
|
||||
// Touched is the machines a change touches (ADR 0446 clause 3): those running a module the change moves, adds or
|
||||
@@ -167,13 +174,19 @@ func Compose(m Mesh, c Change, sdk string, took time.Duration) delivery.Composit
|
||||
failAt(node, fmt.Sprintf("%s composes on the trunk and not with the change: %v", node, now))
|
||||
}
|
||||
}
|
||||
if c.Rebuilt > delivery.CompositionRebuildWarning {
|
||||
v.Warnings = append(v.Warnings, fmt.Sprintf("the merge rebuilds %d modules, more than %d", c.Rebuilt, delivery.CompositionRebuildWarning))
|
||||
}
|
||||
if c.SDKHead != "" {
|
||||
v.Warnings = append(v.Warnings, fmt.Sprintf("composed with the assigner's own mesh-sdk %s, not the group's mesh-sdk head %s, "+
|
||||
"and a field the running controller cannot read is not judged (issues 1621, 1622, 1623)", sdk, c.SDKHead))
|
||||
}
|
||||
return finish(v, m, c, machines, fails)
|
||||
}
|
||||
|
||||
// finish adds what every composition says beside its own judgement (more modules rebuilt than CompositionRebuildWarning,
|
||||
// what is not judged, the machines not composed, a change that touches none) and its state from fails.
|
||||
func finish(v delivery.CompositionVerdict, m Mesh, c Change, machines []string, fails []string) delivery.CompositionVerdict {
|
||||
if c.Rebuilt > delivery.CompositionRebuildWarning {
|
||||
v.Warnings = append(v.Warnings, fmt.Sprintf("the merge rebuilds %d modules, more than %d", c.Rebuilt, delivery.CompositionRebuildWarning))
|
||||
}
|
||||
v.Warnings = append(v.Warnings, NotJudged)
|
||||
if could := CouldRun(m, c, machines); len(could) > 0 {
|
||||
v.Warnings = append(v.Warnings, "not judged until the facts snapshot is on the bus (issue 1461, ADR 0537 clause 1): "+
|
||||
|
||||
@@ -0,0 +1,241 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/composer"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The cases of a group's composition at its mesh-sdk head: the trunk as it is, and the trunk with the change.
|
||||
const (
|
||||
CaseTrunk = "trunk"
|
||||
CaseChange = "change"
|
||||
)
|
||||
|
||||
// SDKDir is the directory of mesh-sdk's Go module in its repository, where the composer is built at a head; a reader's
|
||||
// composer is built at its repository's root, at the mesh-sdk its go.mod pins.
|
||||
const SDKDir = "go"
|
||||
|
||||
// The modules whose mesh-sdk reads the manifests a plan sends (ADR 0482 clause 5): the controller composes and sends
|
||||
// them, the node-engine applies them. Each is judged at its repository.
|
||||
var ReaderModules = map[string]string{"mesh-controller": "novox/mesh-controller", "mesh-host": "novox/mesh-host"}
|
||||
|
||||
// HeadJob is the compose job of a group with a mesh-sdk member (ADR 0482 clauses 3 and 4, issue 1622): the composer
|
||||
// built at the member's head composes the machines the change touches, as the trunk holds them and as the change leaves
|
||||
// them, every manifest named by its digest in the indexer's bucket, the seats held from the claims (the assigner's input).
|
||||
func HeadJob(m Mesh, c Change) (delivery.ComposeJob, error) {
|
||||
trunk, change := map[string]composer.Manifest{}, map[string]composer.Manifest{}
|
||||
for name, digest := range m.Digests {
|
||||
trunk[name] = composer.Manifest{Digest: digest}
|
||||
change[name] = composer.Manifest{Digest: digest}
|
||||
}
|
||||
for name, digest := range c.Digests {
|
||||
change[name] = composer.Manifest{Digest: digest}
|
||||
}
|
||||
for _, name := range c.Removed {
|
||||
delete(change, name)
|
||||
}
|
||||
req := composer.Request{V: composer.Version, Input: plainInput(m.Assigned), Depth: composer.DepthResolve,
|
||||
HoldingsFromClaims: true, Machines: Touched(m, c),
|
||||
Cases: []composer.Case{{Name: CaseTrunk, Manifests: trunk}, {Name: CaseChange, Manifests: change}}}
|
||||
return job(SDKRepository, c.SDKHead, SDKDir, req, "the mesh-sdk head of "+c.Group)
|
||||
}
|
||||
|
||||
// ReaderJob is the compose job that reads the manifests a group changes with the mesh-sdk a reader module pins at
|
||||
// commit (ADR 0482 clause 5, issue 1623): no machine is composed, every changed manifest is only read.
|
||||
func ReaderJob(c Change, module, repository, commit string) (delivery.ComposeJob, error) {
|
||||
changed := map[string]composer.Manifest{}
|
||||
for name, digest := range c.Digests {
|
||||
changed[name] = composer.Manifest{Digest: digest}
|
||||
}
|
||||
req := composer.Request{V: composer.Version, Cases: []composer.Case{{Name: CaseChange, Manifests: changed}}}
|
||||
return job(repository, commit, delivery.RootPath, req, module+" at "+commit)
|
||||
}
|
||||
|
||||
func job(repository, commit, dir string, req composer.Request, reader string) (delivery.ComposeJob, error) {
|
||||
raw, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
return delivery.ComposeJob{}, err
|
||||
}
|
||||
return delivery.ComposeJob{Fetch: map[string]any{"repository": repository, "commit": commit}, Dir: dir,
|
||||
Package: composer.Package, Request: raw, Reader: reader}, nil
|
||||
}
|
||||
|
||||
// plainInput is the composition's input without its catalogue: the nodes by name with what is assigned to them (the
|
||||
// composer derives the seats held from the claims, as input does).
|
||||
func plainInput(assigned map[string][]string) compose.Input {
|
||||
var in compose.Input
|
||||
nodes := make([]string, 0, len(assigned))
|
||||
for n := range assigned {
|
||||
nodes = append(nodes, n)
|
||||
}
|
||||
sort.Strings(nodes)
|
||||
for _, n := range nodes {
|
||||
in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: slices.Clone(assigned[n])})
|
||||
}
|
||||
return in
|
||||
}
|
||||
|
||||
// Ran is one compose ask's end: its step's state and reason, and the composer's verdict when the step passed.
|
||||
type Ran struct {
|
||||
State string
|
||||
Reason string
|
||||
Verdict *composer.Verdict
|
||||
}
|
||||
|
||||
// RanOf reads a compose ask's end from its step value; a step that passed with an answer that cannot be read is a
|
||||
// failure, named.
|
||||
func RanOf(step delivery.Step) Ran {
|
||||
r := Ran{State: step.State, Reason: step.Reason}
|
||||
if step.State != delivery.StepPassed {
|
||||
return r
|
||||
}
|
||||
var a delivery.ComposeAnswer
|
||||
var v composer.Verdict
|
||||
if err := json.Unmarshal(step.Result, &a); err != nil {
|
||||
return Ran{State: delivery.StepFailed, Reason: "the compose step's answer cannot be read: " + err.Error()}
|
||||
}
|
||||
if err := json.Unmarshal(a.Verdict, &v); err != nil || v.V != composer.Version {
|
||||
return Ran{State: delivery.StepFailed, Reason: fmt.Sprintf("the composer's verdict cannot be read (version %d): %v", v.V, err)}
|
||||
}
|
||||
r.Verdict = &v
|
||||
return r
|
||||
}
|
||||
|
||||
func caseOf(v *composer.Verdict, name string) (composer.CaseVerdict, bool) {
|
||||
if v == nil {
|
||||
return composer.CaseVerdict{}, false
|
||||
}
|
||||
for _, c := range v.Cases {
|
||||
if c.Name == name {
|
||||
return c, true
|
||||
}
|
||||
}
|
||||
return composer.CaseVerdict{}, false
|
||||
}
|
||||
|
||||
// Reader is one module whose mesh-sdk reads the group's manifests, at the commit judged: the group's member head when
|
||||
// the group holds one in its repository, else a build its node runs.
|
||||
type Reader struct {
|
||||
Module, Commit string
|
||||
Member bool
|
||||
Ran Ran
|
||||
}
|
||||
|
||||
// FromComposer is the composition check of a group with a mesh-sdk member, judged from the composer built at its head
|
||||
// (ADR 0482 clauses 3 to 5, issues 1622 and 1623): the same failures as Compose, read from the composer's two cases, the
|
||||
// verdict naming the mesh-sdk the composer was built with; a trunk manifest the head's mesh-sdk refuses fails too (the
|
||||
// group would deliver a reader that refuses it); and each reader that refuses a changed manifest the head reads fails
|
||||
// it, naming the module and its words, unless the reader's mesh-sdk has no composer to ask (before issue 1621), which is
|
||||
// said and judged nothing.
|
||||
func FromComposer(m Mesh, c Change, head Ran, readers []Reader, took time.Duration) delivery.CompositionVerdict {
|
||||
v := delivery.CompositionVerdict{Group: c.Group, Heads: c.Heads, SDK: "mesh-sdk " + c.SDKHead, Took: took.Milliseconds()}
|
||||
machines := Touched(m, c)
|
||||
v.Machines = machines
|
||||
var fails []string
|
||||
if head.State != delivery.StepPassed {
|
||||
v.Words = fmt.Sprintf("the composer at mesh-sdk %s did not run: %s %s", c.SDKHead, head.State, head.Reason)
|
||||
v.State = delivery.CompositionFail
|
||||
return v
|
||||
}
|
||||
if head.Verdict.SDK != "" {
|
||||
v.SDK = head.Verdict.SDK
|
||||
}
|
||||
trunk, okT := caseOf(head.Verdict, CaseTrunk)
|
||||
change, okC := caseOf(head.Verdict, CaseChange)
|
||||
if !okT || !okC {
|
||||
v.State, v.Words = delivery.CompositionFail, "the composer's verdict holds no trunk or no change case"
|
||||
return v
|
||||
}
|
||||
failAt := func(machine, words string) {
|
||||
if v.Machine == "" && machine != "" {
|
||||
v.Machine = machine
|
||||
}
|
||||
fails = append(fails, words)
|
||||
}
|
||||
for _, name := range sortedKeys(c.Digests) {
|
||||
if r := change.Manifests[name]; !r.Accepted {
|
||||
failAt("", fmt.Sprintf("the manifest of %s cannot be read by mesh-sdk %s: %s", name, c.SDKHead, r.Refused))
|
||||
}
|
||||
}
|
||||
for _, name := range sortedKeys(trunk.Manifests) {
|
||||
if _, changed := c.Digests[name]; changed || slices.Contains(c.Removed, name) || trunk.Manifests[name].Accepted {
|
||||
continue
|
||||
}
|
||||
if _, before := m.Unreadable[name]; before {
|
||||
continue // already unreadable on the trunk: not the group's doing
|
||||
}
|
||||
failAt("", fmt.Sprintf("mesh-sdk %s refuses the trunk's manifest of %s, which the group does not change: %s", c.SDKHead, name, trunk.Manifests[name].Refused))
|
||||
}
|
||||
for _, name := range sortedKeys(c.Twice) {
|
||||
failAt("", fmt.Sprintf("%s is defined twice, in %s", name, strings.Join(c.Twice[name], " and ")))
|
||||
}
|
||||
for _, node := range machines {
|
||||
for _, name := range c.Removed {
|
||||
if slices.Contains(m.Assigned[node], name) {
|
||||
failAt(node, fmt.Sprintf("%s runs %s, which the change removes from its source", node, name))
|
||||
}
|
||||
}
|
||||
before, now := trunk.Machines[node], change.Machines[node]
|
||||
switch {
|
||||
case now.Composes:
|
||||
case !before.Composes:
|
||||
v.Warnings = append(v.Warnings, fmt.Sprintf("%s does not compose on the trunk either, so the change is not blamed: %s", node, before.Reason))
|
||||
default:
|
||||
failAt(node, fmt.Sprintf("%s composes on the trunk and not with the change: %s", node, now.Reason))
|
||||
}
|
||||
}
|
||||
readerFails, readerWarnings := judgeReaders(c, change, readers)
|
||||
fails = append(fails, readerFails...)
|
||||
v.Warnings = append(v.Warnings, readerWarnings...)
|
||||
return finish(v, m, c, machines, fails)
|
||||
}
|
||||
|
||||
// NoComposer says whether a compose step failed because the mesh-sdk it was built at has no composer command (a mesh-sdk
|
||||
// from before issue 1621): Go's words for a package its module does not hold.
|
||||
func NoComposer(reason string) bool {
|
||||
for _, words := range []string{"cannot find package", "no required module provides package", "is not in std", "cannot find module providing package", "does not contain package"} {
|
||||
if strings.Contains(reason, words) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func judgeReaders(c Change, head composer.CaseVerdict, readers []Reader) (fails, warnings []string) {
|
||||
for _, r := range readers {
|
||||
at := r.Module + " at " + r.Commit
|
||||
if r.Member {
|
||||
at = r.Module + "'s member head " + r.Commit
|
||||
}
|
||||
if r.Ran.State != delivery.StepPassed {
|
||||
if NoComposer(r.Ran.Reason) {
|
||||
warnings = append(warnings, fmt.Sprintf("not judged whether %s reads the group's manifests: the mesh-sdk it pins has no "+
|
||||
"composer command (before issue 1621)", at))
|
||||
continue
|
||||
}
|
||||
fails = append(fails, fmt.Sprintf("the composer of the mesh-sdk %s pins did not run: %s %s", at, r.Ran.State, r.Ran.Reason))
|
||||
continue
|
||||
}
|
||||
read, ok := caseOf(r.Ran.Verdict, CaseChange)
|
||||
if !ok {
|
||||
fails = append(fails, fmt.Sprintf("the composer of the mesh-sdk %s pins answered no change case", at))
|
||||
continue
|
||||
}
|
||||
for _, name := range sortedKeys(c.Digests) {
|
||||
if !head.Manifests[name].Accepted || read.Manifests[name].Accepted {
|
||||
continue
|
||||
}
|
||||
fails = append(fails, fmt.Sprintf("the manifest of %s uses what %s (mesh-sdk %s) cannot read: %s; the group must move %s onto "+
|
||||
"mesh-sdk %s", name, at, r.Ran.Verdict.SDK, read.Manifests[name].Refused, r.Module, c.SDKHead))
|
||||
}
|
||||
}
|
||||
return fails, warnings
|
||||
}
|
||||
@@ -0,0 +1,184 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/composer"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
var testTime = time.Date(2026, 10, 24, 3, 0, 0, 0, time.UTC)
|
||||
|
||||
const sdkHead = "1111111111111111111111111111111111111111"
|
||||
|
||||
// raws are the manifests of meshOf as their repositories hold them, by digest, as the indexer keeps them; `web2` is web
|
||||
// with a field `shares` that the assigner's own mesh-sdk does not read (the shape of issue 1593).
|
||||
var raws = map[string]string{
|
||||
"d-mail": `{"module": "mail", "version": "1", "requires": ["postgres-database"]}`,
|
||||
"d-store": `{"module": "store", "version": "1", "provides": [{"name": "postgres-database"}]}`,
|
||||
"d-web": `{"module": "web", "version": "1"}`,
|
||||
"d-web2": `{"module": "web", "version": "2", "shares": ["photos"]}`,
|
||||
}
|
||||
|
||||
// asTheHeadReads is how mesh-sdk at the group's head reads each manifest: as the assigner's own reads it, but for the
|
||||
// field it adds, which it accepts. The head's composer is built and run by the Go build module (ADR 0482 clause 4); here
|
||||
// its verdict is made by mesh-sdk's composer.Run over the same request, with `web2` read as the head reads it.
|
||||
var asTheHeadReads = map[string]string{"d-web2": `{"module": "web", "version": "2"}`}
|
||||
|
||||
func digestedMesh() Mesh {
|
||||
m := meshOf()
|
||||
m.Digests = map[string]string{"mail": "d-mail", "store": "d-store", "web": "d-web"}
|
||||
return m
|
||||
}
|
||||
|
||||
// webShares is a group of mesh-sdk and mesh-catalog whose catalogue member gives web the field only the SDK head reads:
|
||||
// the assigner's own reader refuses it (Unreadable), as the indexer's manifest is read in world.change.
|
||||
func webShares(members map[string]string) Change {
|
||||
_, err := catalogue.ParseManifest([]byte(raws["d-web2"]))
|
||||
return Change{Group: "g", Heads: "0123456789ab", SDKHead: sdkHead, Manifests: map[string]catalogue.Manifest{},
|
||||
Unreadable: map[string]string{"web": err.Error()}, Digests: map[string]string{"web": "d-web2"}, Members: members}
|
||||
}
|
||||
|
||||
// run is the composer's step for a job, as the build module answers it: the request filled from the indexer's digests
|
||||
// (with read standing in for how that mesh-sdk reads a digest), run, and its verdict put in a passed step.
|
||||
func run(t *testing.T, job delivery.ComposeJob, read map[string]string) Ran {
|
||||
t.Helper()
|
||||
req, err := composer.Decode(job.Request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := composer.Fill(&req, func(d string) ([]byte, error) {
|
||||
if r, ok := read[d]; ok {
|
||||
return []byte(r), nil
|
||||
}
|
||||
return []byte(raws[d]), nil
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
v, err := composer.Run(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
v.SDK = "(devel) " + job.Fetch["commit"].(string)
|
||||
vr, _ := json.Marshal(v)
|
||||
ans, _ := json.Marshal(delivery.ComposeAnswer{Built: true, Key: "composer-0", Verdict: vr})
|
||||
return RanOf(delivery.Step{State: delivery.StepPassed, Result: ans})
|
||||
}
|
||||
|
||||
// 1622/AC1: a group holding a mesh-sdk pull request is composed by the composer built at the member's head: the job
|
||||
// fetches mesh-sdk at that head, builds mesh-sdk's composer command in its Go module, and hands it the trunk and the
|
||||
// change by digest, never the manifests' bytes, composing the machines the change touches.
|
||||
func TestAGroupWithAnSDKMemberAsksTheComposerAtItsHead(t *testing.T) {
|
||||
c := webShares(nil)
|
||||
job, err := HeadJob(digestedMesh(), c)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if job.Fetch["repository"] != SDKRepository || job.Fetch["commit"] != sdkHead || job.Dir != "go" || job.Package != composer.Package {
|
||||
t.Fatalf("job = %+v", job)
|
||||
}
|
||||
ask := delivery.StepAsk{ID: "mh3k2x9qa1b2c5", Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-assigner",
|
||||
At: testTime, Compose: &job}
|
||||
if _, err := delivery.EncodeStepAsk(ask); err != nil {
|
||||
t.Fatalf("the ask is refused by its layout: %v", err)
|
||||
}
|
||||
req, err := composer.Decode(job.Request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(req.Cases) != 2 || req.Cases[0].Name != CaseTrunk || req.Cases[1].Name != CaseChange {
|
||||
t.Fatalf("cases = %+v", req.Cases)
|
||||
}
|
||||
if req.Cases[0].Manifests["web"].Digest != "d-web" || req.Cases[1].Manifests["web"].Digest != "d-web2" || len(req.Cases[1].Manifests["web"].Raw) != 0 {
|
||||
t.Errorf("the cases do not name web by its trunk and its change digest: %+v", req.Cases)
|
||||
}
|
||||
if !slices.Equal(req.Machines, []string{"g14"}) || req.Depth != composer.DepthResolve || !req.HoldingsFromClaims {
|
||||
t.Errorf("request = machines %v, depth %q, holdings from claims %v", req.Machines, req.Depth, req.HoldingsFromClaims)
|
||||
}
|
||||
}
|
||||
|
||||
// 1622/AC2, AC3: a catalogue member using a field only the SDK head reads passes with the head's composer, and the
|
||||
// verdict names the mesh-sdk the composer was built with; with the assigner's own mesh-sdk it fails.
|
||||
func TestAFieldTheGroupAddsPassesWithTheHeadsComposer(t *testing.T) {
|
||||
c := webShares(map[string]string{"novox/mesh-controller": "2222222222222222222222222222222222222222"})
|
||||
job, _ := HeadJob(digestedMesh(), c)
|
||||
head := run(t, job, asTheHeadReads)
|
||||
v := FromComposer(digestedMesh(), c, head, nil, 0)
|
||||
if v.State != delivery.CompositionPass || v.SDK != "(devel) "+sdkHead || !slices.Equal(v.Machines, []string{"g14"}) {
|
||||
t.Fatalf("with the head's composer: %+v", v)
|
||||
}
|
||||
if _, err := delivery.EncodeCompositionVerdict(v); err != nil {
|
||||
t.Errorf("the verdict cannot be written: %v", err)
|
||||
}
|
||||
own := Compose(digestedMesh(), c, "v0.1.14-own", 0)
|
||||
if own.State != delivery.CompositionFail || !strings.Contains(own.Words, "shares") {
|
||||
t.Fatalf("with the assigner's own mesh-sdk: %+v, want a fail naming the field", own)
|
||||
}
|
||||
}
|
||||
|
||||
// 1623/AC1, AC2: the group of mesh-sdk and mesh-catalog adding and using a field, without a controller member, fails
|
||||
// naming the field and mesh-controller, whose running build pins an older mesh-sdk; with a controller member pinned to
|
||||
// the SDK head (its composer reads what the head reads) it passes.
|
||||
func TestAFieldTheRunningControllerCannotReadFailsTheGroup(t *testing.T) {
|
||||
c := webShares(nil)
|
||||
job, _ := HeadJob(digestedMesh(), c)
|
||||
head := run(t, job, asTheHeadReads)
|
||||
running := "3333333333333333333333333333333333333333"
|
||||
rjob, err := ReaderJob(c, "mesh-controller", "novox/mesh-controller", running)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if rjob.Fetch["commit"] != running || rjob.Dir != delivery.RootPath {
|
||||
t.Fatalf("reader job = %+v", rjob)
|
||||
}
|
||||
old := run(t, rjob, nil) // the running controller's mesh-sdk reads web2 as the assigner's own does: refused
|
||||
v := FromComposer(digestedMesh(), c, head, []Reader{{Module: "mesh-controller", Commit: running, Ran: old}}, 0)
|
||||
if v.State != delivery.CompositionFail || !strings.Contains(v.Words, "shares") || !strings.Contains(v.Words, "move mesh-controller onto mesh-sdk "+sdkHead) {
|
||||
t.Fatalf("without a controller member: %+v, want a fail naming the field and mesh-controller", v)
|
||||
}
|
||||
member := "2222222222222222222222222222222222222222"
|
||||
pinned := run(t, rjob, asTheHeadReads)
|
||||
v = FromComposer(digestedMesh(), c, head, []Reader{{Module: "mesh-controller", Commit: member, Member: true, Ran: pinned}}, 0)
|
||||
if v.State != delivery.CompositionPass {
|
||||
t.Fatalf("with a controller member pinned to the head: %+v, want a pass", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 1623: a reader whose mesh-sdk has no composer command (before issue 1621) is said, not failed; a reader's composer that
|
||||
// failed for another cause fails the group, loud.
|
||||
func TestAReaderWithNoComposerIsSaidAndAnotherFailureFails(t *testing.T) {
|
||||
c := webShares(nil)
|
||||
job, _ := HeadJob(digestedMesh(), c)
|
||||
head := run(t, job, asTheHeadReads)
|
||||
none := Ran{State: delivery.StepFailed, Reason: "go build: no required module provides package git.novox.be/novox/mesh-sdk/go/cmd/mesh-compose"}
|
||||
v := FromComposer(digestedMesh(), c, head, []Reader{{Module: "mesh-host", Commit: "abc", Ran: none}}, 0)
|
||||
if v.State != delivery.CompositionPass || !hasWarning(v, "mesh-host at abc") || !hasWarning(v, "before issue 1621") {
|
||||
t.Fatalf("a reader with no composer: %+v", v)
|
||||
}
|
||||
broke := Ran{State: delivery.StepFailed, Reason: "the fetch could not reach the forge"}
|
||||
v = FromComposer(digestedMesh(), c, head, []Reader{{Module: "mesh-host", Commit: "abc", Ran: broke}}, 0)
|
||||
if v.State != delivery.CompositionFail || !strings.Contains(v.Words, "could not reach the forge") {
|
||||
t.Fatalf("a reader whose composer failed: %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 1622: the head's composer that did not run fails the composition naming why, never passes it; a trunk manifest the
|
||||
// head's mesh-sdk refuses fails it too, since the group would deliver a reader that refuses it.
|
||||
func TestTheHeadsComposerFailingOrRefusingTheTrunkFails(t *testing.T) {
|
||||
c := webShares(nil)
|
||||
v := FromComposer(digestedMesh(), c, Ran{State: delivery.StepRefused, Reason: "the guard refuses it"}, nil, 0)
|
||||
if v.State != delivery.CompositionFail || !strings.Contains(v.Words, "the guard refuses it") || !strings.Contains(v.Words, sdkHead) {
|
||||
t.Fatalf("a head composer that did not run: %+v", v)
|
||||
}
|
||||
job, _ := HeadJob(digestedMesh(), c)
|
||||
strict := map[string]string{"d-web2": asTheHeadReads["d-web2"], "d-mail": `{"module": "mail", "version": "1", "gone": true}`}
|
||||
v = FromComposer(digestedMesh(), c, run(t, job, strict), nil, 0)
|
||||
if v.State != delivery.CompositionFail || !strings.Contains(v.Words, "refuses the trunk's manifest of mail") {
|
||||
t.Fatalf("a head that refuses a trunk manifest: %+v", v)
|
||||
}
|
||||
}
|
||||
@@ -32,7 +32,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.grants",
|
||||
want := []string{"mesh-build-go.steps", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.grants",
|
||||
"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
@@ -51,6 +51,10 @@ func TestManifest(t *testing.T) {
|
||||
if m.SecretsOwner != "mesh-delivery-assigner" {
|
||||
t.Errorf("secrets-owner = %q", m.SecretsOwner)
|
||||
}
|
||||
// 1622: the assigner puts its compose asks on the Go build seat, as the checker puts its checks.
|
||||
if !slices.Equal(m.Uses, []string{"node-build-go"}) {
|
||||
t.Errorf("uses = %v, want exactly node-build-go", m.Uses)
|
||||
}
|
||||
if len(m.Claims) != 0 || len(m.Tools) != 0 || len(m.Invokes) != 0 {
|
||||
t.Errorf("the assigner claims, serves or calls something: claims %v, tools %v, invokes %v", m.Claims, m.Tools, m.Invokes)
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
"version": "1",
|
||||
"slug": "assigner",
|
||||
"reads": [
|
||||
"mesh-build-go.steps",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-controller.builds",
|
||||
"mesh-controller.grants",
|
||||
@@ -13,6 +14,9 @@
|
||||
"mesh-delivery-watchman.presence",
|
||||
"node-engine.have"
|
||||
],
|
||||
"uses": [
|
||||
"node-build-go"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "asks", "history": 5 },
|
||||
{ "name": "compositions", "history": 5 },
|
||||
|
||||
@@ -179,7 +179,11 @@ type bed struct {
|
||||
// startBed starts the bus, the buckets the controller makes from the builder's and the build modules' manifests, the
|
||||
// build modules' streams and the builder's durable consumers on them, two nodes' toolchains, the indexer's manifests, and
|
||||
// the fake build modules. It runs no builder.
|
||||
func startBed(t *testing.T) *bed {
|
||||
func startBed(t *testing.T) *bed { return startBedWith(t, false) }
|
||||
|
||||
// startBedWith is startBed with the nodes' capacity 0 from the start when closed: a puller that reads its capacity before a
|
||||
// test could close the nodes may be in a pull already and takes the first ask put (issue 1928).
|
||||
func startBedWith(t *testing.T, closed bool) *bed {
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "b")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -243,6 +247,7 @@ func startBed(t *testing.T) *bed {
|
||||
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
|
||||
b.instances = mk(instancesBucket, 1)
|
||||
b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}}
|
||||
b.fake.closed.Store(closed)
|
||||
b.fake.serve(t, "mesh-build-go", "n1", "n2")
|
||||
b.fake.serve(t, "mesh-build-node", "n1", "n2")
|
||||
b.fake.serve(t, "mesh-build-image", "n1") // the image build module runs on n1 only
|
||||
@@ -781,8 +786,7 @@ func (b *bed) asksPut(t *testing.T, nc *nats.Conn) func() []delivery.StepAsk {
|
||||
// 1887/AC3: a builder that started again while its asks waited in the streams (no node had a free slot) puts no second
|
||||
// ask: it waits for the asks its building key names, and the key is done once a node took them.
|
||||
func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) {
|
||||
b := startBed(t)
|
||||
b.fake.closed.Store(true)
|
||||
b := startBedWith(t, true) // 1928/AC1: no node can take an ask from the start, so no pull in hand takes the first one
|
||||
b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")
|
||||
sub, err := nats.Connect(b.srvURL, nats.UserInfo("ctl", "pw"))
|
||||
if err != nil {
|
||||
@@ -792,11 +796,13 @@ func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) {
|
||||
asked := b.asksPut(t, sub)
|
||||
b.start(t, "n1", nil)
|
||||
v := b.await(t, "netcheck", func(v delivery.BuilderBuild) bool { return v.State == delivery.BuildBuilding && len(v.Asks) == 2 })
|
||||
for deadline := time.Now().Add(5 * time.Second); len(asked()) < 2 && time.Now().Before(deadline); time.Sleep(20 * time.Millisecond) {
|
||||
for deadline := time.Now().Add(20 * time.Second); len(asked()) < 2; time.Sleep(20 * time.Millisecond) {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("asks put = %+v within 20 s, want the two the key names", asked())
|
||||
}
|
||||
}
|
||||
b.stop()
|
||||
b.start(t, "n1", nil) // the same builder, started again, while both asks wait
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
b.fake.open()
|
||||
done := b.await(t, "netcheck", isState(delivery.BuildDone))
|
||||
if got := asked(); len(got) != 2 {
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -81,6 +82,9 @@ type fakeBuildModules struct {
|
||||
mu sync.Mutex
|
||||
calls map[string]int // by `<module>.<tool>`
|
||||
testTool *testing.T
|
||||
// beside is, by workspace, the siblings the fetch placed in it as `<owner/repo> <commit>`, sorted: a digest echoes them
|
||||
// (1929/AC2; mesh-sdk buildstep.besideOf), or the level runs whole and keeps nothing.
|
||||
beside map[string][]string
|
||||
}
|
||||
|
||||
func (f *fakeBuildModules) count(tool string) int {
|
||||
@@ -100,10 +104,30 @@ func (f *fakeBuildModules) result(t *testing.T, tool string, args map[string]any
|
||||
}
|
||||
fetched := readCaptured(t, name)
|
||||
fetched["commit"], fetched["base"] = args["commit"], args["base"]
|
||||
var refs []struct {
|
||||
Repository string `json:"repository"`
|
||||
Commit string `json:"commit"`
|
||||
}
|
||||
raw, _ := json.Marshal(args["beside"])
|
||||
_ = json.Unmarshal(raw, &refs)
|
||||
var lines []string
|
||||
for _, r := range refs {
|
||||
lines = append(lines, r.Repository+" "+r.Commit)
|
||||
}
|
||||
sort.Strings(lines)
|
||||
f.mu.Lock()
|
||||
if f.beside == nil {
|
||||
f.beside = map[string][]string{}
|
||||
}
|
||||
f.beside[fmt.Sprint(fetched["workspace"])] = lines
|
||||
f.mu.Unlock()
|
||||
v = fetched
|
||||
case strings.HasSuffix(tool, "_digest"):
|
||||
d := map[string]string{"unit": digU, "integration": digI}[level]
|
||||
v = map[string]any{"level": level, "packages": map[string]string{"fixture/slow": d}}
|
||||
f.mu.Lock()
|
||||
beside := f.beside[fmt.Sprint(args["workspace"])]
|
||||
f.mu.Unlock()
|
||||
v = map[string]any{"level": level, "packages": map[string]string{"fixture/slow": d}, "beside": beside}
|
||||
case strings.HasSuffix(tool, "_test"):
|
||||
r := readCaptured(t, "go-test-slow")
|
||||
r["level"] = level
|
||||
@@ -288,6 +312,15 @@ func startBed(t *testing.T) *bed {
|
||||
}
|
||||
b.buckets[name] = kv
|
||||
}
|
||||
// 1929/AC1: the indexer's value of the repository's module, with no TypeScript manifest, so the Go build module alone
|
||||
// judges the repository (an unknown repository is judged by both, issue 1887 A12).
|
||||
known, err := delivery.EncodeModule(delivery.Module{Repository: repo, Commit: trunk, Path: "modules/m", None2: "the fingerprint is computed with its chain"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.buckets[delivery.ModulesBucket].Put(ctx, "m", known); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.setOnline(t, "n1", true)
|
||||
for _, module := range []string{"mesh-build-go", "mesh-build-node"} {
|
||||
b.setHealth(t, "n1", module)
|
||||
@@ -436,7 +469,7 @@ func (b *bed) owed(t *testing.T, number int, head, state string) delivery.Owed {
|
||||
// TestCheckerJudgesACandidateAndOwesItsStatus: a candidate the indexer wrote, one pull request of mesh-delivery, is fetched
|
||||
// merged onto the trunk the gate names, digested and tested at both levels on the node, kept, previewed; the verdict passes
|
||||
// with the counts of both levels and the composition not run, and the status mesh/check is owed on the head as success.
|
||||
func TestCheckerJudgesACandidateAndOwesItsStatus(t *testing.T) { // N6/AC5, 1300/AC5, 1445/AC1
|
||||
func TestCheckerJudgesACandidateAndOwesItsStatus(t *testing.T) { // N6/AC5, 1300/AC5, 1445/AC1, 1929/AC1
|
||||
b := startBed(t)
|
||||
key := b.candidate(t, "feat/x", pull(repo, 7, head1))
|
||||
b.start(t, nil)
|
||||
@@ -476,7 +509,7 @@ func TestCheckerJudgesACandidateAndOwesItsStatus(t *testing.T) { // N6/AC5, 1300
|
||||
|
||||
// TestACheckOfEqualInputsRunsNothing: a second candidate (another head, the same digests) is checked with every package
|
||||
// taken from the first's kept verdicts: no test step runs and no second kept key appears.
|
||||
func TestACheckOfEqualInputsRunsNothing(t *testing.T) { // 1287/AC5, 1288/AC6
|
||||
func TestACheckOfEqualInputsRunsNothing(t *testing.T) { // 1287/AC5, 1288/AC6, 1929/AC2
|
||||
b := startBed(t)
|
||||
first := b.candidate(t, "feat/x", pull(repo, 7, head1))
|
||||
b.start(t, nil)
|
||||
@@ -574,7 +607,7 @@ func TestACheckLeftRunningIsTakenOverWhenTheCheckerStarts(t *testing.T) { // N6/
|
||||
|
||||
// TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit: the go.mod the forge answers names a commit that main does not
|
||||
// hold; the verdict fails naming the member and the commit and no test runs (ADR 0484 clause 9).
|
||||
func TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit(t *testing.T) { // 1624/AC2
|
||||
func TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit(t *testing.T) { // 1624/AC2, 1929/AC3
|
||||
b := startBed(t)
|
||||
b.stalePin.Store(true)
|
||||
key := b.candidate(t, "feat/x", pull(repo, 7, head1))
|
||||
@@ -583,9 +616,7 @@ func TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit(t *testing.T
|
||||
if v.State != delivery.VerdictFail || !strings.Contains(v.Failed.Words, "9aaa5c80ac5c") || !strings.Contains(v.Failed.Words, repo+"#7") {
|
||||
t.Fatalf("verdict = %+v", v)
|
||||
}
|
||||
if n := b.fake.count("mesh-build-go.build_go_test"); n != 0 {
|
||||
t.Errorf("%d test steps ran after a stale pin", n)
|
||||
}
|
||||
// 1929/AC3: no assertion that no test ran: one ask runs fetch, digest and tests (ADR 0525 §3).
|
||||
}
|
||||
|
||||
// triesStore is the tries bucket as the tools write it.
|
||||
@@ -671,7 +702,7 @@ func TestTheCheckerSaysOwedReadableOnTheBus(t *testing.T) { // R2-1
|
||||
break
|
||||
}
|
||||
if said.Module != "mesh-delivery-checker" || said.Node != "n1" || !said.Keeps("owed") || !said.Readable("owed") ||
|
||||
!said.Readable("previews") || !said.Readable("verdicts") || said.Readable("kept") || said.Readable("tries") || len(said.States) != 5 {
|
||||
t.Fatalf("the checker said %+v, want its five states with owed, previews and verdicts readable", said)
|
||||
!said.Readable("previews") || !said.Readable("verdicts") || !said.Readable("kept") || said.Readable("tries") || len(said.States) != 5 {
|
||||
t.Fatalf("the checker said %+v, want its five states with kept, owed, previews and verdicts readable (1929/AC4; 1887 A5b)", said)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-checker
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/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=
|
||||
|
||||
@@ -26,7 +26,7 @@ const (
|
||||
// checks/deps). Which commit of each is besideCommit's rule.
|
||||
var besideRules = map[string][]string{
|
||||
controllerRepo: {catalogueRepository, nodeEngineRepo, SDKRepository},
|
||||
nodeEngineRepo: {catalogueRepository, SDKRepository},
|
||||
nodeEngineRepo: {catalogueRepository, SDKRepository, controllerRepo},
|
||||
SDKRepository: {catalogueRepository, nodeEngineRepo},
|
||||
catalogueRepository: {SDKRepository, controllerRepo},
|
||||
"novox/mesh-delivery": {SDKRepository, catalogueRepository},
|
||||
@@ -83,6 +83,10 @@ func (j *Examiner) besideFor(ctx context.Context, c *Case, p Pull) ([]BesideRef,
|
||||
out = append(out, BesideRef{Repository: repo, Commit: member.Head, Ref: refOf(member.Number)})
|
||||
} else if head := heads[repo]; head != "" {
|
||||
out = append(out, BesideRef{Repository: repo, Commit: head})
|
||||
} else if repo == controllerRepo && p.Repository == nodeEngineRepo {
|
||||
// The node-engine's step-nine test judges the controller's own manifest and fails without it (issue 1295): a
|
||||
// checker that cannot place it says so, where placing none left that test to skip.
|
||||
return nil, fmt.Errorf("the trunk head of %s is not known, so the controller the tests of %s#%d read cannot be placed beside them", repo, p.Repository, p.Number)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// 1930/AC3: given a check ask whose stored end keeps the digest's packages as a list of {package, digest} (issue 1930),
|
||||
// or one stored before it with the packages a map keyed by package path, when the same heads are checked a second time,
|
||||
// then the kept verdicts the first check wrote are read from it: every package it passed is kept, none run again.
|
||||
func TestKeptVerdictsAreReadFromTheDigestsListForm(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
old bool
|
||||
}{{"the list form", false}, {"a map stored before issue 1930", true}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
b := newBench(t)
|
||||
var stored []string
|
||||
b.asks.Answer = func(a delivery.StepAsk) delivery.Step {
|
||||
s := b.answer(a)
|
||||
if tc.old && s.State == delivery.StepPassed {
|
||||
s.Result = oldDigestForm(t, s.Result)
|
||||
}
|
||||
stored = append(stored, string(s.Result))
|
||||
return s
|
||||
}
|
||||
first := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
|
||||
u := first.Levels[delivery.LevelUnit]
|
||||
if first.State != delivery.VerdictPass || u.Run == 0 || u.Kept != 0 {
|
||||
t.Fatalf("the first check = %+v", first)
|
||||
}
|
||||
if len(stored) == 0 {
|
||||
t.Fatal("no ask's end was stored")
|
||||
}
|
||||
list := strings.Contains(stored[0], `"packages":[{"package":"fixture/slow"`)
|
||||
if list == tc.old || strings.Contains(stored[0], `"fixture/slow":"`) != tc.old {
|
||||
t.Fatalf("the stored end is not in the form under test: %s", stored[0])
|
||||
}
|
||||
b.steps.calls = nil
|
||||
second := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
|
||||
if s := second.Levels[delivery.LevelUnit]; second.State != delivery.VerdictPass || s.Kept != u.Run || s.Run != 0 {
|
||||
t.Fatalf("the second check's unit level = %+v, want %d kept and none run", s, u.Run)
|
||||
}
|
||||
if n := len(b.steps.tool("test")); n != 0 {
|
||||
t.Errorf("%d test steps ran on the second check", n)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// oldDigestForm rewrites a check answer's level digests to the form stored before issue 1930: packages as a map keyed by
|
||||
// package path.
|
||||
func oldDigestForm(t *testing.T, result json.RawMessage) json.RawMessage {
|
||||
t.Helper()
|
||||
var a struct {
|
||||
Fetch json.RawMessage `json:"fetch"`
|
||||
Levels []map[string]any `json:"levels"`
|
||||
}
|
||||
if err := json.Unmarshal(result, &a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, l := range a.Levels {
|
||||
d, ok := l["digest"].(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
m := map[string]any{}
|
||||
list, _ := d["packages"].([]any)
|
||||
for _, p := range list {
|
||||
pd := p.(map[string]any)
|
||||
m[pd["package"].(string)] = pd["digest"]
|
||||
}
|
||||
d["packages"] = m
|
||||
}
|
||||
raw, err := json.Marshal(a)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -54,3 +55,19 @@ func TestWordsLongerThanTheBoundKeepTheirLastLines(t *testing.T) { // 1901/AC1
|
||||
}
|
||||
_ = delivery.VerdictFail
|
||||
}
|
||||
|
||||
// An older build module sends no "cause": its answer is still accepted and the verdict's words are its output, as before.
|
||||
func TestAnOlderBuildModulesFailureIsAcceptedWithItsWholeOutput(t *testing.T) { // 1901/AC3
|
||||
old := `{"package":"p","test":"TestX","output":"x_test.go:1: logged\nx_test.go:2: the old kept lines"}`
|
||||
var f Failure
|
||||
if err := json.Unmarshal([]byte(old), &f); err != nil || f.Cause != "" {
|
||||
t.Fatalf("1901/AC3: err = %v, cause = %q", err, f.Cause)
|
||||
}
|
||||
v := levelFailure(&target{}, "unit", Tested{Failed: []Failure{f}})
|
||||
if v.Failed.Words != "x_test.go:1: logged\nx_test.go:2: the old kept lines" {
|
||||
t.Errorf("1901/AC3: words = %q", v.Failed.Words)
|
||||
}
|
||||
if got := firstFailed(*v); !strings.Contains(got, "logged") || strings.Contains(got, "\n") {
|
||||
t.Errorf("1901/AC3: status = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -447,13 +447,11 @@ func (st *state) level(ctx context.Context, t *target, la delivery.LevelAnswer)
|
||||
return fail(t.node, fmt.Sprintf("the check of %s#%d answered the level %q, which is no level", t.pull.Repository, t.pull.Number, level)), 0
|
||||
}
|
||||
var digests map[string]string
|
||||
if t.module == buildrun.GoBuildModule && len(la.Digest) > 0 {
|
||||
var d Digested
|
||||
if err := json.Unmarshal(la.Digest, &d); err == nil {
|
||||
digests = d.Packages
|
||||
if t.runner != "" {
|
||||
t.judgedBy(d.JudgedBy) // the installed module judged it: the runner is not used again
|
||||
}
|
||||
if t.module == buildrun.GoBuildModule && la.Digest != nil {
|
||||
// The packages come as a list, never keyed by a package path (issue 1930); an answer stored before reads too.
|
||||
digests = la.Digest.Map()
|
||||
if t.runner != "" {
|
||||
t.judgedBy(la.Digest.JudgedBy) // the installed module judged it: the runner is not used again
|
||||
}
|
||||
}
|
||||
if t.runner != "" && la.Judged != "" {
|
||||
|
||||
@@ -244,6 +244,39 @@ func TestTheSiblingsAreWhereTheTestsStateThem(t *testing.T) { // review D2, ADR
|
||||
}
|
||||
}
|
||||
|
||||
// 1295/R4g: the node-engine's tests read the controller's manifest, so a mesh-host pull request has mesh-controller beside it:
|
||||
// at the group member's head when the group holds one, else at its trunk.
|
||||
func hostCase(t *testing.T, pulls ...delivery.CandidatePull) *Case {
|
||||
t.Helper()
|
||||
c, err := delivery.DecodeCandidate(candidateOf(t, "feat/x", pulls...))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cs := CaseOf(c, map[string]string{"novox/mesh-host": trunkA, controllerRepo: trunkA2})
|
||||
if cs == nil {
|
||||
t.Fatal("no case")
|
||||
}
|
||||
return cs
|
||||
}
|
||||
|
||||
func TestTheNodeEnginesTestsHaveTheControllerBesideThem(t *testing.T) {
|
||||
const host = "novox/mesh-host"
|
||||
b := newBench(t)
|
||||
b.ex.Trunks = trunksOf(t, map[string]string{host: trunkA, controllerRepo: trunkA2, catalogueRepository: trunkS, SDKRepository: trunkS})
|
||||
b.judge(t, hostCase(t, pullOf(host, 5, head1)))
|
||||
if c := besideOf(t, b, host)[controllerRepo]; c.Commit != trunkA2 || c.Ref != "" {
|
||||
t.Errorf("the controller beside the node-engine alone = %+v, want its trunk %s", c, trunkA2)
|
||||
}
|
||||
b = newBench(t)
|
||||
b.ex.Trunks = trunksOf(t, map[string]string{host: trunkA, controllerRepo: trunkA2, catalogueRepository: trunkS, SDKRepository: trunkS})
|
||||
b.ex.Forge = &fakeForge{goMods: map[string]string{controllerRepo + " go.mod " + head2: capturedGoMod(t, "cb54756c3865")},
|
||||
onTrunk: map[string]bool{"cb54756c3865": true}}
|
||||
b.judge(t, hostCase(t, pullOf(host, 5, head1), pullOf(controllerRepo, 9, head2)))
|
||||
if c := besideOf(t, b, host)[controllerRepo]; c.Commit != head2 || c.Ref != "refs/pull/9/head" {
|
||||
t.Errorf("the controller beside the node-engine in a group = %+v, want the member's head %s", c, head2)
|
||||
}
|
||||
}
|
||||
|
||||
func TestThePinOfTheControllerThatCannotBeReadFailsTheCheck(t *testing.T) { // review D2
|
||||
b := newBench(t)
|
||||
b.ex.Trunks = trunksOf(t, map[string]string{"novox/mesh-controller": trunkA})
|
||||
@@ -313,3 +346,15 @@ func TestAMeshLabHeadGetsACheck(t *testing.T) {
|
||||
t.Fatalf("mesh-lab was not fetched: %+v", fetched)
|
||||
}
|
||||
}
|
||||
|
||||
// 1295: the node-engine's step-nine test fails without the controller beside it, so a checker that knows no trunk head of the
|
||||
// controller fails the check at the placing, loudly, and does not place none.
|
||||
func TestTheNodeEnginesCheckFailsWhereTheControllersTrunkIsNotKnown(t *testing.T) {
|
||||
const host = "novox/mesh-host"
|
||||
b := newBench(t)
|
||||
b.ex.Trunks = trunksOf(t, map[string]string{host: trunkA, catalogueRepository: trunkS, SDKRepository: trunkS})
|
||||
v := b.judge(t, hostCase(t, pullOf(host, 5, head1))).Verdict
|
||||
if v.State != delivery.VerdictFail || !strings.Contains(v.Failed.Words, "mesh-controller") || !strings.Contains(v.Failed.Words, "cannot be placed") {
|
||||
t.Fatalf("verdict = %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -478,3 +478,74 @@ func TestAMergeOfMeshSDKChangingOnlyGoLeavesItsNamedTreeEqual(t *testing.T) {
|
||||
t.Errorf("the new form moved: %q/%q, then %q/%q", first.FP2, first.None2, second.FP2, second.None2)
|
||||
}
|
||||
}
|
||||
|
||||
// TestIndexerJudgesAnUnreadableVerdictAgainAtStart: the modules bucket holds avahi as an older SDK wrote it, with the reason
|
||||
// 'the manifest cannot be read', the inputs the merge carried (the tree, the lock hash by file name) and the digest of its
|
||||
// manifest, which manifests.<digest> holds and the running SDK reads; the go toolchain is reported. Within 5 s of the start
|
||||
// and with no repository merge the value carries a fingerprint and no reason (issue 1923).
|
||||
func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC3
|
||||
manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381"
|
||||
digest := delivery.ManifestDigest(manifest)
|
||||
_, ctx, _, modules := startIndexer(t, func(context.Context, jetstream.KeyValue) {}, func(ctx context.Context, js jetstream.JetStream) {
|
||||
mkey, err := delivery.ManifestsKey(digest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mdata, err := delivery.EncodeManifest(delivery.Manifest{Module: "avahi", JSON: string(manifest)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
manifests, err := js.KeyValue(ctx, delivery.ManifestsBucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := manifests.Put(ctx, mkey, mdata); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stale, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: "89abcdef0123456789abcdef0123456789abcdef", Path: "modules/avahi",
|
||||
Manifest: digest, None2: "the manifest cannot be read: claims node-build-go, which is a seat in the mesh's own namespace that it does not define",
|
||||
Inputs2: &delivery.KeptInputs2{Tree: "tree-avahi", Locks: map[string]string{"go.sum": goSum}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
held, err := js.KeyValue(ctx, delivery.ModulesBucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := held.Put(ctx, "avahi", stale); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, err := delivery.ToolchainKey("n1", "go")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: "go1.27.1", Recipe: "1", At: time.Now()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := kv.Put(ctx, key, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
var got delivery.Module
|
||||
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if e, err := modules.Get(ctx, "avahi"); err == nil {
|
||||
if m, err := delivery.DecodeModule(e.Value()); err == nil {
|
||||
if got = m; got.FP2 != "" {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if got.FP2 == "" || got.None2 != "" {
|
||||
t.Errorf("avahi = fp2 %q none2 %q within 5 s of the start, want a fingerprint and no reason", got.FP2, got.None2)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,9 @@
|
||||
// bridge's messages on the stream FORGE and keeps the candidate groups (candidates.<group>): the pull requests that share
|
||||
// a branch, their heads and the modules each touches. It reads the controller's builds to know which module a changed
|
||||
// directory is. It serves no request subject and holds no timer: every reaction is a message or a watch (ADR 0437 §8).
|
||||
//
|
||||
// Its build of 2026-10-24 00:37 named a program whose push never reached the artifact store, and a retry sends that
|
||||
// same build again; this paragraph was added only to give the indexer a new build (issue 1933).
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -288,7 +291,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}, Builds: known, Logf: logger.Printf,
|
||||
Named: namedVia(calls.Call)}
|
||||
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call)}
|
||||
// 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.
|
||||
@@ -325,6 +328,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := ix.RecomputeAtStart(ctx); err != nil {
|
||||
logger.Printf("computing the new-form fingerprints at start: %v", err)
|
||||
}
|
||||
// The modules the trunk no longer holds go from the bucket (issue 1900); a merge does the same from its own message.
|
||||
if err := ix.Reconcile(ctx); err != nil {
|
||||
logger.Printf("dropping the modules the trunk does not hold at start: %v", err)
|
||||
}
|
||||
// A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are
|
||||
// made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again
|
||||
// with the build module, as with the gate's source.
|
||||
|
||||
@@ -46,3 +46,50 @@ func namedVia(call toolCaller) func(ctx context.Context, repository, sha string,
|
||||
return a.Paths, nil
|
||||
}
|
||||
}
|
||||
|
||||
// trunkVia asks the forge for the module directories of a commit (issue 1900): the keys of the `trees` the same tool
|
||||
// answers without paths, one per directory that holds a module.json (the root as ""; the tool lists no other).
|
||||
func trunkVia(call toolCaller) func(ctx context.Context, repository, sha string) ([]string, error) {
|
||||
return func(_ context.Context, repository, sha string) ([]string, error) {
|
||||
owner, repo, ok := strings.Cut(delivery.RepositoryName(repository), "/")
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("%q is not <owner>/<name>", repository)
|
||||
}
|
||||
raw, err := call(commitTreesTool, map[string]any{"owner": owner, "repo": repo, "sha": sha})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var a struct {
|
||||
Trees map[string]string `json:"trees"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &a); err != nil {
|
||||
return nil, fmt.Errorf("the answer of %s is not {root, trees}: %w", commitTreesTool, err)
|
||||
}
|
||||
dirs := make([]string, 0, len(a.Trees))
|
||||
for dir := range a.Trees {
|
||||
dirs = append(dirs, dir)
|
||||
}
|
||||
return dirs, nil
|
||||
}
|
||||
}
|
||||
|
||||
// containsVia asks the forge whether a commit is in another (gitea.gitea_contains: contains is true for the same commit or an ancestor).
|
||||
func containsVia(call toolCaller) func(ctx context.Context, repository, commit, in string) (bool, error) {
|
||||
return func(_ context.Context, repository, commit, in string) (bool, error) {
|
||||
owner, repo, ok := strings.Cut(delivery.RepositoryName(repository), "/")
|
||||
if !ok {
|
||||
return false, fmt.Errorf("%q is not <owner>/<name>", repository)
|
||||
}
|
||||
raw, err := call("gitea.gitea_contains", map[string]any{"owner": owner, "repo": repo, "commit": commit, "in": in})
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
var a struct {
|
||||
Contains *bool `json:"contains"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &a); err != nil || a.Contains == nil {
|
||||
return false, fmt.Errorf("the answer of gitea.gitea_contains names no contains: %v", err)
|
||||
}
|
||||
return *a.Contains, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,8 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
@@ -29,3 +31,22 @@ func TestNamedViaAsksTheTreesToolWithThePathsAndTakesTheirObjects(t *testing.T)
|
||||
t.Error("an answer missing a path asked was taken")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3
|
||||
var gotArgs any
|
||||
call := func(key string, body any) (json.RawMessage, error) {
|
||||
gotArgs = body
|
||||
return json.RawMessage(`{"root":"r","trees":{"modules/a":"x","modules/b":"y"}}`), nil
|
||||
}
|
||||
dirs, err := trunkVia(call)(context.Background(), "novox/mesh-catalog", strings.Repeat("a", 40))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
slices.Sort(dirs)
|
||||
if !slices.Equal(dirs, []string{"modules/a", "modules/b"}) {
|
||||
t.Errorf("dirs = %v", dirs)
|
||||
}
|
||||
if _, has := gotArgs.(map[string]any)["paths"]; has {
|
||||
t.Error("the trunk asked for paths")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -52,7 +52,10 @@ type Indexer struct {
|
||||
// manifest's paths (ADR 0495 clause 1, the lead's decision L8 of R4e). Nil asks nothing, and a module naming paths
|
||||
// has none.
|
||||
Named func(ctx context.Context, repository, sha string, paths map[string][]string) (map[string]map[string]string, error)
|
||||
Logf func(format string, args ...any)
|
||||
Trunk func(ctx context.Context, repository, sha string) ([]string, error)
|
||||
// Contains says whether a commit is the second or an ancestor of it (gitea.gitea_contains).
|
||||
Contains func(ctx context.Context, repository, commit, in string) (bool, error)
|
||||
Logf func(format string, args ...any)
|
||||
|
||||
mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
|
||||
source sourceNow
|
||||
|
||||
@@ -69,7 +69,7 @@ func inputsOf(module, raw, tree string, locks map[string]string, base baseFP) mo
|
||||
func inputsOfNamed(module, raw, tree string, locks, objects map[string]string, base baseFP) moduleInputs {
|
||||
m, err := catalogue.ParseManifestForInputs([]byte(raw))
|
||||
if err != nil {
|
||||
return moduleInputs{None2: "the manifest cannot be read: " + err.Error()}
|
||||
return moduleInputs{None2: None2Unreadable + err.Error()}
|
||||
}
|
||||
if m.Module != module {
|
||||
return moduleInputs{None2: fmt.Sprintf("the manifest is of module %q, not %q", m.Module, module)}
|
||||
|
||||
@@ -111,7 +111,8 @@ func TestABaseEntersByTheFingerprintOfItsModuleOrNamesItsAbsence(t *testing.T) {
|
||||
}
|
||||
// Through the indexer: modules.mesh-tools holds a fingerprint of the new form (a record written by a later release).
|
||||
ix, mods, _ := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs))
|
||||
mods.put("mesh-tools", mustEncode(t, delivery.Module{Repository: catalogueRepo, Commit: merge, None: "x", FP2: fp, Manifest: strings.Repeat("a", 64)}))
|
||||
mods.put("mesh-tools", mustEncode(t, delivery.Module{Repository: "novox/mesh-tools", // a base held of another repository stays when the catalogue merges (issue 1900)
|
||||
Commit: merge, None: "x", FP2: fp, Manifest: strings.Repeat("a", 64)}))
|
||||
f := withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "t"}), map[string]string{"modules/avahi": aModuleWithABase(t)}, nil)
|
||||
if err := ix.Apply(context.Background(), f); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -2,8 +2,10 @@ package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
@@ -33,18 +35,22 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error {
|
||||
objects, namedErr := ix.namedObjects(ctx, f)
|
||||
overlay := map[string]delivery.Module{}
|
||||
dirOf := map[string]string{}
|
||||
named, every := map[string]bool{}, true // every module the message's manifests name, and whether each names one
|
||||
for _, dir := range slices.Sorted(maps.Keys(f.Manifests)) {
|
||||
raw := f.Manifests[dir]
|
||||
name := manifestModule(raw)
|
||||
if name == "" {
|
||||
ix.logf("%s at %s: the manifest in %q names no module, so no module is written for it", f.Repository, f.Merge, dir)
|
||||
every = false
|
||||
continue
|
||||
}
|
||||
key, err := delivery.ModulesKey(name)
|
||||
if err != nil {
|
||||
every = false
|
||||
ix.logf("%s: the module %q has no usable key, skipped: %v", f.Repository, name, err)
|
||||
continue
|
||||
}
|
||||
named[key] = true
|
||||
cur, has := held[key]
|
||||
if has && cur.Repository != "" && !delivery.SameRepository(cur.Repository, f.Repository) {
|
||||
ix.logf("%s at %s: %q names the module %s, which the indexer holds of %s; left as it is (a move between repositories is the operator's)", f.Repository, f.Merge, dir, name, cur.Repository)
|
||||
@@ -64,12 +70,24 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error {
|
||||
} else {
|
||||
in := inputsOfNamed(name, raw, f.Trees[dir], f.Locks[dir], objects[dir], ix.baseFP(ctx))
|
||||
next.Inputs2, next.None2, next.ArtifactPaths = in.Kept, in.None2, in.ArtifactPaths
|
||||
if in.Kept == nil && strings.HasPrefix(in.None2, None2Unreadable) {
|
||||
// The manifest cannot be read by this SDK; the inputs the merge carried are kept all the same, so a later SDK
|
||||
// that reads it judges the module again at start with no merge (issue 1923).
|
||||
next.Inputs2 = unreadInputs(f.Trees[dir], f.Locks[dir], f.Paths[dir])
|
||||
}
|
||||
if in.Kept == nil && namedErr[dir] != "" {
|
||||
next.None2 = in.None2 + ": the forge could not say them: " + namedErr[dir]
|
||||
}
|
||||
}
|
||||
overlay[key] = next
|
||||
}
|
||||
// A module of this repository the merge commit holds no module.json for is gone from the trunk, and so is its key
|
||||
// (issue 1900). A message that carries no manifests (an older writer) or one that cannot be read whole says nothing.
|
||||
if len(f.Manifests) > 0 && every {
|
||||
if err := ix.dropRemoved(ctx, f.Repository, held, func(key string, _ delivery.Module) bool { return named[key] }); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// What this merge moved reaches the modules standing on it (a base's tree, build-agent's fingerprint), in the graph's order.
|
||||
return ix.recomputeWith(ctx, overlay)
|
||||
}
|
||||
@@ -149,3 +167,115 @@ func (ix *Indexer) namedObjects(ctx context.Context, f delivery.Forge) (map[stri
|
||||
}
|
||||
return objects, nil
|
||||
}
|
||||
|
||||
// dropRemoved deletes, from the modules the bucket holds of the repository, each one the trunk no longer holds (issue 1900):
|
||||
// held(key, module) says whether the trunk holds it. A module of another repository, or of none known, is never touched.
|
||||
func (ix *Indexer) dropRemoved(ctx context.Context, repository string, held map[string]delivery.Module, trunkHolds func(key string, m delivery.Module) bool) error {
|
||||
for _, key := range slices.Sorted(maps.Keys(held)) {
|
||||
cur := held[key]
|
||||
if cur.Repository == "" || !delivery.SameRepository(cur.Repository, repository) || trunkHolds(key, cur) {
|
||||
continue
|
||||
}
|
||||
err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
if !found {
|
||||
return nil, false, nil
|
||||
}
|
||||
if m, err := delivery.DecodeModule(value); err != nil || !delivery.SameRepository(m.Repository, repository) {
|
||||
return nil, false, nil
|
||||
}
|
||||
return nil, true, nil
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("modules.%s: %w", key, err)
|
||||
}
|
||||
ix.logf("modules.%s deleted: %s holds no module of that name at its trunk (issue 1900)", key, repository)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Reconcile deletes the modules keys whose module the trunk does not hold, for each repository the latest source names a
|
||||
// head of: the module directories at that head are asked of the forge (Trunk), and a module of the repository whose
|
||||
// directory is not among them goes (issue 1900). It is run at the indexer's start; a merge does the same from its own
|
||||
// message (applyMerged). A repository whose trunk cannot be read, or reads as holding no module at all, is left as it is
|
||||
// and said: a failure never empties the bucket.
|
||||
func (ix *Indexer) Reconcile(ctx context.Context) error {
|
||||
if ix.Modules == nil || ix.Trunk == nil {
|
||||
return nil
|
||||
}
|
||||
ix.mu.Lock()
|
||||
heads := maps.Clone(ix.source.heads)
|
||||
ix.mu.Unlock()
|
||||
held, err := ix.heldModules(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, repository := range slices.Sorted(maps.Keys(heads)) {
|
||||
mine := map[string]delivery.Module{}
|
||||
for key, m := range held {
|
||||
if m.Repository != "" && delivery.SameRepository(m.Repository, repository) {
|
||||
mine[key] = m
|
||||
}
|
||||
}
|
||||
if len(mine) == 0 {
|
||||
continue
|
||||
}
|
||||
dirs, err := ix.Trunk(ctx, repository, heads[repository])
|
||||
if err != nil {
|
||||
ix.logf("%s at %s: the modules of the trunk cannot be read, so none is dropped: %v", repository, heads[repository], err)
|
||||
continue
|
||||
}
|
||||
if len(dirs) == 0 {
|
||||
ix.logf("%s at %s: the trunk reads as holding no module, so none is dropped", repository, heads[repository])
|
||||
continue
|
||||
}
|
||||
there := map[string]bool{}
|
||||
for _, d := range dirs {
|
||||
there[pathOf(strings.Trim(d, "/"))] = true
|
||||
}
|
||||
head := heads[repository]
|
||||
// The source can lag a merge already applied: a key written at a commit the head read does not hold (a merge heard
|
||||
// first) is not read as removed, only one whose commit the head is, or contains.
|
||||
stays := func(_ string, m delivery.Module) bool {
|
||||
if m.Path == "" || there[m.Path] {
|
||||
return true
|
||||
}
|
||||
if m.Commit == head {
|
||||
return false
|
||||
}
|
||||
if m.Commit == "" || ix.Contains == nil {
|
||||
return true
|
||||
}
|
||||
in, err := ix.Contains(ctx, repository, m.Commit, head)
|
||||
if err != nil {
|
||||
ix.logf("%s: whether %s is in %s cannot be read, so the module at %s is kept: %v", repository, m.Commit, head, m.Path, err)
|
||||
return true
|
||||
}
|
||||
return !in
|
||||
}
|
||||
if err := ix.dropRemoved(ctx, repository, mine, stays); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// None2Unreadable begins the reason of a module whose manifest the running SDK cannot read.
|
||||
const None2Unreadable = "the manifest cannot be read: "
|
||||
|
||||
// unreadInputs are the inputs a merge carried for a module whose manifest cannot be read: the directory's tree, the lock
|
||||
// hashes by file name (the artifacts that own them are not known without the manifest) and the objects of the named
|
||||
// paths the message carried; nil when the message carried no tree. Judging the manifest again (rejudged) turns them into
|
||||
// the inputs of the module (issue 1923).
|
||||
func unreadInputs(tree string, locks, paths map[string]string) *delivery.KeptInputs2 {
|
||||
if tree == "" {
|
||||
return nil
|
||||
}
|
||||
in := &delivery.KeptInputs2{Tree: tree}
|
||||
if len(locks) > 0 {
|
||||
in.Locks = maps.Clone(locks)
|
||||
}
|
||||
if len(paths) > 0 {
|
||||
in.Paths = maps.Clone(paths)
|
||||
}
|
||||
return in
|
||||
}
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
func twoModules(t *testing.T) (*Indexer, *memStore) {
|
||||
t.Helper()
|
||||
ix, mods := modulesIndexer(t)
|
||||
if err := ix.Apply(context.Background(), mergedMsg(catalogueRepo, map[string]string{"modules/a": "t", "modules/b": "t"})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, k := range []string{"a", "b"} {
|
||||
if _, ok := mods.vals[k]; !ok {
|
||||
t.Fatalf("setup: modules.%s not written", k)
|
||||
}
|
||||
}
|
||||
return ix, mods
|
||||
}
|
||||
|
||||
func TestAMergeThatRemovesAModuleDeletesItsKey(t *testing.T) { // 1900/AC1
|
||||
ix, mods := twoModules(t)
|
||||
if err := ix.Apply(context.Background(), mergedMsg(catalogueRepo, map[string]string{"modules/a": "t2"})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := mods.vals["b"]; ok {
|
||||
t.Error("modules.b is still held after a merge whose trunk has no module b")
|
||||
}
|
||||
if _, ok := mods.vals["a"]; !ok {
|
||||
t.Error("modules.a, which the trunk still holds, was deleted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAMergeLeavesTheModulesOfAnotherRepository(t *testing.T) { // 1900/AC1 (scope)
|
||||
ix, mods := twoModules(t)
|
||||
if err := ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/c": "t"})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := mods.vals["a"]; !ok {
|
||||
t.Error("a merge of another repository deleted modules.a")
|
||||
}
|
||||
if _, ok := mods.vals["b"]; !ok {
|
||||
t.Error("a merge of another repository deleted modules.b")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAMergeWithoutManifestsDeletesNothing(t *testing.T) { // 1900/AC1: older writers send none, which does not say the trunk is empty
|
||||
ix, mods := twoModules(t)
|
||||
f := mergedMsg(catalogueRepo, map[string]string{"modules/a": "t2"})
|
||||
f.Manifests = nil
|
||||
if err := ix.Apply(context.Background(), f); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := mods.vals["b"]; !ok {
|
||||
t.Error("a merge message with no manifests deleted modules.b")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAPathWithoutAModuleJSONIsNeverAModule(t *testing.T) { // 1900/AC2: testdata such as fixture-build
|
||||
ix, mods := twoModules(t)
|
||||
f := mergedMsg(catalogueRepo, map[string]string{"modules/a": "t2"})
|
||||
f.Trees["testdata/fixture-build"] = "t" // a directory in the tree whose module.json the message does not carry
|
||||
if err := ix.Apply(context.Background(), f); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := mods.vals["fixture-build"]; ok {
|
||||
t.Error("modules.fixture-build written for a path without a module.json")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartingDeletesTheKeysTheTrunkDoesNotHold(t *testing.T) { // 1900/AC3
|
||||
ix, mods := twoModules(t)
|
||||
ctx := context.Background()
|
||||
if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, merge)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var asked string
|
||||
ix.Trunk = func(_ context.Context, repository, sha string) ([]string, error) {
|
||||
asked = repository + "@" + sha
|
||||
return []string{"modules/a"}, nil
|
||||
}
|
||||
ix.Contains = func(context.Context, string, string, string) (bool, error) { return true, nil }
|
||||
if err := ix.Reconcile(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if asked != catalogueRepo+"@"+merge {
|
||||
t.Errorf("the trunk was asked as %q, want the source's head of the repository", asked)
|
||||
}
|
||||
if _, ok := mods.vals["b"]; ok {
|
||||
t.Error("modules.b is still held though the trunk holds no module at modules/b")
|
||||
}
|
||||
if _, ok := mods.vals["a"]; !ok {
|
||||
t.Error("modules.a was deleted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartingDeletesNothingWhenTheTrunkCannotBeRead(t *testing.T) { // 1900/AC3: failure is loud, never a mass delete
|
||||
ix, mods := twoModules(t)
|
||||
ctx := context.Background()
|
||||
if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, merge)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for name, trunk := range map[string]func(context.Context, string, string) ([]string, error){
|
||||
"an error": func(context.Context, string, string) ([]string, error) { return nil, errors.New("forge down") },
|
||||
"no module at all": func(context.Context, string, string) ([]string, error) { return nil, nil },
|
||||
} {
|
||||
ix.Trunk = trunk
|
||||
_ = ix.Reconcile(ctx)
|
||||
if len(mods.vals) != 2 {
|
||||
t.Errorf("%s: %d keys held, want 2 untouched", name, len(mods.vals))
|
||||
}
|
||||
}
|
||||
_ = delivery.RootPath
|
||||
}
|
||||
|
||||
func TestAKeyWrittenAheadOfTheSourceHeadStaysAndOneWithoutAPathIsSkipped(t *testing.T) { // 1900/AC3 (review): the source lags a merge already applied
|
||||
ix, mods := twoModules(t) // written at the merge commit C
|
||||
ctx := context.Background()
|
||||
const older = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||
if err := ix.ApplySource(ctx, 3, sourceOf(catalogueRepo, older)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mods.put("nopath", mustEncode(t, delivery.Module{Repository: catalogueRepo, Commit: older, None2: "x", Manifest: strings.Repeat("a", 64)}))
|
||||
ix.Trunk = func(context.Context, string, string) ([]string, error) { return []string{"modules/a"}, nil }
|
||||
ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == in, nil } // C is not in B
|
||||
if err := ix.Reconcile(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, k := range []string{"a", "b", "nopath"} {
|
||||
if _, ok := mods.vals[k]; !ok {
|
||||
t.Errorf("modules.%s was deleted", k)
|
||||
}
|
||||
}
|
||||
ix.Contains = func(context.Context, string, string, string) (bool, error) { return false, errors.New("forge down") }
|
||||
_ = ix.Reconcile(ctx)
|
||||
if _, ok := mods.vals["b"]; !ok {
|
||||
t.Error("modules.b deleted though whether its commit is in the head cannot be read")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// unreadableAvahi is avahi's manifest with a field the running parser refuses: the shape of a manifest an older SDK could
|
||||
// not read.
|
||||
func unreadableAvahi(t *testing.T) string {
|
||||
t.Helper()
|
||||
raw := strings.Replace(catalogueManifest(t, "avahi"), `"version": "1",`, `"version": "1", "scope": "everything",`, 1)
|
||||
if raw == catalogueManifest(t, "avahi") {
|
||||
t.Fatal("the fixture has no version line")
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
func TestAMergeOfAnUnreadableManifestKeepsTheInputsBesideTheReason(t *testing.T) { // 1923/AC1
|
||||
_, mods := newFormIndexer(t, mod{"avahi", unreadableAvahi(t)})
|
||||
m, ok := moduleOf(t, mods, "avahi")
|
||||
if !ok || !strings.HasPrefix(m.None2, "the manifest cannot be read: ") || m.FP2 != "" {
|
||||
t.Fatalf("module = %+v, want the reason 'the manifest cannot be read' and no fingerprint", m)
|
||||
}
|
||||
if m.Inputs2 == nil || m.Inputs2.Tree != "tree-avahi" || m.Inputs2.Locks["go.sum"] != avahiGoSum {
|
||||
t.Errorf("inputs = %+v, want the directory's tree and its lock hashes kept", m.Inputs2)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAStaleUnreadableVerdictIsJudgedAgainAtStartAndAnotherReasonIsLeft(t *testing.T) { // 1923/AC2
|
||||
ctx := context.Background()
|
||||
// The fingerprint a merge of the readable manifest gives.
|
||||
ixGood, good := newFormIndexer(t, mod{"avahi", catalogueManifest(t, "avahi")})
|
||||
reportToolchain(t, ixGood, "mesh-build-go", "novox", "go", "go1.27.1")
|
||||
reportToolchain(t, ixGood, "mesh-build-go", "ace", "go", "go1.27.1")
|
||||
want, _ := moduleOf(t, good, "avahi")
|
||||
if want.FP2 == "" {
|
||||
t.Fatalf("the readable manifest gives no fingerprint: %+v", want)
|
||||
}
|
||||
|
||||
// The same module as an older SDK wrote it: the reason, the inputs kept at the merge, the manifest stored by digest.
|
||||
ix, mods := newFormIndexer(t, mod{"avahi", unreadableAvahi(t)})
|
||||
reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1")
|
||||
reportToolchain(t, ix, "mesh-build-go", "ace", "go", "go1.27.1")
|
||||
stale, _ := moduleOf(t, mods, "avahi")
|
||||
stale.Manifest = want.Manifest
|
||||
ix.Manifests.(*memStore).put(want.Manifest, good2(t, ixGood, want.Manifest))
|
||||
mods.put("avahi", mustEncode(t, stale))
|
||||
if err := ix.Recompute(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, _ := moduleOf(t, mods, "avahi")
|
||||
if got.FP2 != want.FP2 || got.None2 != "" {
|
||||
t.Errorf("after the start: fp2 %q none2 %q, want %q and no reason", got.FP2, got.None2, want.FP2)
|
||||
}
|
||||
|
||||
// A manifest still unreadable keeps its value as it was.
|
||||
ix2, mods2 := newFormIndexer(t, mod{"avahi", unreadableAvahi(t)})
|
||||
before := string(mods2.vals["avahi"])
|
||||
if err := ix2.Recompute(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if after := string(mods2.vals["avahi"]); after != before {
|
||||
t.Errorf("a manifest still unreadable changed the value:\n%s\n%s", before, after)
|
||||
}
|
||||
}
|
||||
|
||||
// good2 is the stored manifest value of the digest in the indexer that merged the readable manifest.
|
||||
func good2(t *testing.T, ix *Indexer, digest string) []byte {
|
||||
t.Helper()
|
||||
v, _, found, err := ix.Manifests.Get(context.Background(), digest)
|
||||
if err != nil || !found {
|
||||
t.Fatalf("manifests.%s: %v %v", digest, found, err)
|
||||
}
|
||||
return v
|
||||
}
|
||||
@@ -156,6 +156,7 @@ func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]deli
|
||||
}
|
||||
slices.Sort(keys)
|
||||
held := map[string]delivery.Module{}
|
||||
stale := map[string]bool{} // values an SDK that could not read their manifest wrote, judged again (issue 1923)
|
||||
all := map[string]*known{}
|
||||
for _, key := range keys {
|
||||
cur, ok := overlay[key]
|
||||
@@ -176,11 +177,15 @@ func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]deli
|
||||
if cur.Manifest == "" || cur.Inputs2 == nil {
|
||||
continue
|
||||
}
|
||||
m, err := ix.manifestOf(ctx, cur.Manifest)
|
||||
raw, m, err := ix.manifestRaw(ctx, cur.Manifest)
|
||||
if err != nil {
|
||||
ix.logf("modules.%s: its manifest %s cannot be read, left: %v", key, cur.Manifest, err)
|
||||
continue
|
||||
}
|
||||
if again, ok := ix.rejudged(ctx, m, cur, raw); ok {
|
||||
cur, stale[key] = again, true
|
||||
held[key] = cur
|
||||
}
|
||||
all[key] = &known{key: key, cur: cur, languages: languagesOf(m), bases: basesOf(m), contexts: hasContext(m)}
|
||||
}
|
||||
tools := ix.agreed()
|
||||
@@ -275,10 +280,10 @@ func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]deli
|
||||
finals[key] = final
|
||||
continue
|
||||
}
|
||||
if k.cur.FP2 == r.fp && k.cur.None2 == r.none && reflect.DeepEqual(k.cur.Toolchains, r.tcs) {
|
||||
if !stale[key] && k.cur.FP2 == r.fp && k.cur.None2 == r.none && reflect.DeepEqual(k.cur.Toolchains, r.tcs) {
|
||||
continue
|
||||
}
|
||||
digest, inputs := k.cur.Manifest, k.cur.Inputs2
|
||||
digest, inputs, paths := k.cur.Manifest, k.cur.Inputs2, k.cur.ArtifactPaths
|
||||
err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
if !found {
|
||||
return nil, false, nil
|
||||
@@ -291,7 +296,7 @@ func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]deli
|
||||
return nil, false, nil
|
||||
}
|
||||
cur.FP2, cur.None2, cur.Toolchains = r.fp, r.none, r.tcs
|
||||
cur.Inputs2 = inputs
|
||||
cur.Inputs2, cur.ArtifactPaths = inputs, paths
|
||||
data, err := delivery.EncodeModule(cur)
|
||||
return data, false, err
|
||||
})
|
||||
@@ -326,24 +331,41 @@ func (ix *Indexer) writeOverlay(ctx context.Context, overlay, finals map[string]
|
||||
return nil
|
||||
}
|
||||
|
||||
// manifestOf is the parsed manifest manifests.<digest> holds.
|
||||
func (ix *Indexer) manifestOf(ctx context.Context, digest string) (catalogue.Manifest, error) {
|
||||
// manifestRaw is manifests.<digest>: the manifest as stored and parsed by the running SDK.
|
||||
func (ix *Indexer) manifestRaw(ctx context.Context, digest string) (string, catalogue.Manifest, error) {
|
||||
key, err := delivery.ManifestsKey(digest)
|
||||
if err != nil {
|
||||
return catalogue.Manifest{}, err
|
||||
return "", catalogue.Manifest{}, err
|
||||
}
|
||||
value, _, found, err := ix.Manifests.Get(ctx, key)
|
||||
if err != nil {
|
||||
return catalogue.Manifest{}, err
|
||||
return "", catalogue.Manifest{}, err
|
||||
}
|
||||
if !found {
|
||||
return catalogue.Manifest{}, fmt.Errorf("manifests.%s is not there", key)
|
||||
return "", catalogue.Manifest{}, fmt.Errorf("manifests.%s is not there", key)
|
||||
}
|
||||
kept, err := delivery.DecodeManifest(value)
|
||||
if err != nil {
|
||||
return catalogue.Manifest{}, err
|
||||
return "", catalogue.Manifest{}, err
|
||||
}
|
||||
return catalogue.ParseManifestForInputs([]byte(kept.JSON))
|
||||
m, err := catalogue.ParseManifestForInputs([]byte(kept.JSON))
|
||||
return kept.JSON, m, err
|
||||
}
|
||||
|
||||
// rejudged is a value an SDK that could not read its manifest wrote, as the running SDK judges it: its kept inputs (the
|
||||
// tree, the lock hashes by file name, the named paths' objects) taken through the manifest as a merge takes them, and the
|
||||
// reason that gives. It answers false for any other value, and for one whose manifest the running SDK still cannot make
|
||||
// inputs of (issue 1923).
|
||||
func (ix *Indexer) rejudged(ctx context.Context, m catalogue.Manifest, cur delivery.Module, raw string) (delivery.Module, bool) {
|
||||
if cur.FP2 != "" || cur.Inputs2 == nil || !strings.HasPrefix(cur.None2, None2Unreadable) {
|
||||
return cur, false
|
||||
}
|
||||
in := inputsOfNamed(m.Module, raw, cur.Inputs2.Tree, cur.Inputs2.Locks, cur.Inputs2.Paths, ix.baseFP(ctx))
|
||||
if in.Kept == nil {
|
||||
return cur, false
|
||||
}
|
||||
cur.Inputs2, cur.None2, cur.ArtifactPaths = in.Kept, in.None2, in.ArtifactPaths
|
||||
return cur, true
|
||||
}
|
||||
|
||||
// hasContext is whether an artifact of the manifest reads another repository's tree.
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -94,9 +95,10 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if m.SecretsOwner != "mesh-delivery-indexer" {
|
||||
t.Errorf("secrets-owner = %q", m.SecretsOwner)
|
||||
}
|
||||
// One invoke: the forge's trees with 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, the lead's decision L8 of R4e).
|
||||
if want := []string{"gitea.gitea_commit_trees"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
// Two invokes: 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"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
// No emits, no streams, no bus, no seat, no tool (ADR 0437 §4).
|
||||
@@ -169,3 +171,32 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1238/AC9 (ADR
|
||||
t.Fatalf("read %d source files, want the program's", files)
|
||||
}
|
||||
}
|
||||
|
||||
// Every forge tool the program calls is granted in invokes, or the bus refuses the call (issue 1900).
|
||||
func TestEveryGiteaToolTheCodeCallsIsInInvokes(t *testing.T) {
|
||||
m, _ := load(t)
|
||||
files, err := filepath.Glob("cmd/mesh-delivery-indexer/*.go")
|
||||
if err != nil || len(files) == 0 {
|
||||
t.Fatalf("no program files: %v", err)
|
||||
}
|
||||
re := regexp.MustCompile(`gitea\.gitea_[a-z_]+`)
|
||||
seen := 0
|
||||
for _, f := range files {
|
||||
if strings.HasSuffix(f, "_test.go") {
|
||||
continue
|
||||
}
|
||||
src, err := os.ReadFile(f)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, tool := range re.FindAllString(string(src), -1) {
|
||||
seen++
|
||||
if !slices.Contains(m.Invokes, tool) {
|
||||
t.Errorf("%s calls %s, which invokes does not grant", f, tool)
|
||||
}
|
||||
}
|
||||
}
|
||||
if seen < 2 {
|
||||
t.Errorf("found %d tool names in the program, want at least the two it calls", seen)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,7 +21,8 @@
|
||||
"mesh-delivery-gate.source"
|
||||
],
|
||||
"invokes": [
|
||||
"gitea.gitea_commit_trees"
|
||||
"gitea.gitea_commit_trees",
|
||||
"gitea.gitea_contains"
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
Reference in New Issue
Block a user