Ask the Go build seat for the composer at a group's mesh-sdk head and at its readers' mesh-sdk, and follow each step (issues 1622, 1623)
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -213,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
|
||||
}
|
||||
|
||||
@@ -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 },
|
||||
|
||||
Reference in New Issue
Block a user