diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 3ec09822..4f040bed 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -79,7 +79,7 @@ var readFamilies = []string{buckets.ComposedRead} var ownBuckets = []string{buckets.Asks, buckets.Differences, buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket, - engineReadsBucket, retiredBucket, copiesBucket} + engineReadsBucket, retiredBucket, copiesBucket, writtenBucket} // engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key ``, as // the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again. @@ -88,6 +88,9 @@ const engineReadsState = "engine-reads" // engineReadsBucket is that state's bucket. const engineReadsBucket = assignment.Module + "_" + engineReadsState +// writtenBucket is the bucket of the state written (pipelines.go): the pipeline ends the wants follow. +const writtenBucket = assignment.Module + "_" + writtenState + // seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants // them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind). var seatBuckets = []string{plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket, @@ -115,6 +118,7 @@ var keptStates = []announce.State{ {Name: engineReadsState, Readable: true}, {Name: retiredState}, // read only by the copy into assignments (migrate.go) {Name: copiesState, Readable: true}, + {Name: writtenState, Readable: true}, } // kvSet is every bucket the assigner opened, by name. @@ -197,7 +201,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any func startsWithout(name string) bool { switch name { case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket, - buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket: + buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket: return true } return slices.Contains(seatBuckets, name) || name == buckets.Grants @@ -297,6 +301,7 @@ var watched = []string{ buckets.Source, buckets.Merges, engineReadsBucket, + writtenBucket, } // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after @@ -485,6 +490,14 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, logger.Printf("%s is not heard, so an ask that ends at the operator channel without an answer is not asked again until the assigner starts again: %v", askEndedSubject, err) } readEndedAtBind(ctx, js, hub, kvs[buckets.Asks], staleOfWorld(w), bus, callWait{heard: func() bool { return w.seatHeard(oc.Seat) }, findings: findings}, logger.Printf) + if kv := kvs[writtenBucket]; kv != nil { + // The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow + // once the first registrations are written (§95, §353). + p.heard = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf) + if err := hearEnds(hub, p.heard); err != nil { + logger.Printf("a pipeline's end is not heard, so no want follows a pipeline until the assigner starts again: %v", err) + } + } hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf) } if bus != nil && kvs[buckets.Puts] != nil { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 477ec9c2..69d369af 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -58,6 +58,10 @@ type passer struct { // neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the // first pass after a start, which clears what a previous process may have raised (startClears). neither map[string]neitherOne + // ends are the pipeline ends the wants follow (pipelines.go); nil is the trunk-only want, kept until the first registrations. + ends *pipelineEnds + // heard is the same follower while the wants do not follow it yet (ends is nil). + heard *pipelineEnds // acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket). acts *putActs // noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is @@ -267,9 +271,16 @@ func (p *passer) wants(ctx context.Context) { moduleGraph := p.moduleGraph() served := p.served(ctx, get, moduleGraph, nodes) offering := p.offering(ctx, get, manifests, nodes) // bus state (ADR 0614 §5) + assignedBranches := p.w.placedBranches() for name := range names { fp := trunk[name].Fp2 name := name + // A want follows a passed pipeline: a module whose fingerprint no pipeline passed at on every branch its placements + // chose, and whose wants do not name it already, gets none (design 47 G §29, §35). + branches := p.ends.branchesOf(name, assignedBranches) + if fp != "" && !p.ends.mayWant(name, fp, branches, wanted[name]) { + continue + } m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Needs: trunk[name].Needs, Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Offered: offered(manifests), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], FormNeeds: p.w.buildNeeds(name, fp), Self: func(node string) bool { return placed[node] }, Composed: func(node, fp string) (assign.ComposedDigest, bool) { @@ -359,6 +370,9 @@ func (p *passer) wants(ctx context.Context) { want := want p.put(ctx, buckets.Wants, key, func() ([]byte, error) { return mwc.EncodeWant(want) }) } + if len(out.Wants) > 0 { + p.ends.wrote(name, fp, branches) + } } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go new file mode 100644 index 00000000..743125a5 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go @@ -0,0 +1,327 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + mprc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient" + mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// The wants follow a passed pipeline (design 47 G §29, §35, §325, §352): the runner says each pipeline's end as +// pipeline-ended, and a group's as group-ended. The assigner keeps, in its old state `written`, one key per module and +// watched branch: the source revision and end count of the newest end it heard for it, and whether the wants were written +// from it. A newer end replaces it by compare-and-set; an older one, or the same one again, writes nothing. + +// writtenState is the assigner's state of the pipeline ends its wants follow; writtenTool reads it. +const ( + writtenState = "written" + writtenTool = "assigner_written" +) + +// pipelineEndedSubject and groupEndedSubject are the runner's events the assigner consumes. +const ( + pipelineEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventPipelineEnded + ".>" + groupEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventGroupEnded + ".>" +) + +// End is one value of the state written: the newest end of a module's pipeline on a watched branch. +type End struct { + Source int64 `json:"source"` + EndCount int64 `json:"end-count"` + Fp string `json:"fp"` + Commit string `json:"commit,omitempty"` + // Passed: the pipeline ended passed, every required step passed (and, for a group, the group ended passed). + Passed bool `json:"passed"` + // Written: the wants were written from this end. + Written bool `json:"written,omitempty"` + Why string `json:"why,omitempty"` +} + +// endKey is the key of a module on a watched branch, each name escaped as a key's name is. +func endKey(module, branch string) string { + return keys.EscapeName(module) + "." + keys.EscapeName(branch) +} + +// newer says whether an end of (source, count) replaces the one kept: a later source revision, or the same one ended again. +func (e End) newer(source, count int64) bool { + return source > e.Source || (source == e.Source && count > e.EndCount) +} + +// requiredPassed says whether no required step of a pipeline stands failed, unasked or unended (design 47 G §30: a failed +// optional step holds nothing). +func requiredPassed(p mpr.PipelineEnded) bool { + for _, s := range p.Steps { + if s.Status != mpr.StepResultStatusRequired { + continue + } + switch s.Outcome { + case mpr.StepResultOutcomeFailed, mpr.StepResultOutcomeCouldNotRun, mpr.StepResultOutcomeDeclined, mpr.StepResultOutcomeNotAsked, + mpr.StepResultOutcomeCancelled, mpr.StepResultOutcomeSuperseded, mpr.StepResultOutcomeDidNotBuild, + mpr.StepResultOutcomeRunning, mpr.StepResultOutcomeWaiting, mpr.StepResultOutcomeWaitingForABuildNode: + return false + } + } + return p.Outcome == mpr.PipelineEndedOutcomePassed +} + +// endStore is the kept state written: the bus bucket, or a test's. +type endStore interface { + // get is a key's value and revision; none for a key not there. + get(key string) (value []byte, revision uint64, err error) + // swap writes value at key if the key stands at revision (0: if it does not stand); errSwapped when it does not. + swap(key string, value []byte, revision uint64) error +} + +var errSwapped = errors.New("the key changed") + +// kvEnds is endStore over a bus bucket. +type kvEnds struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (k kvEnds) get(key string) ([]byte, uint64, error) { + e, err := k.kv.Get(k.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, 0, nil + } + if err != nil { + return nil, 0, err + } + return e.Value(), e.Revision(), nil +} + +func (k kvEnds) swap(key string, value []byte, revision uint64) error { + var err error + if revision == 0 { + _, err = k.kv.Create(k.ctx, key, value) + } else { + _, err = k.kv.Update(k.ctx, key, value, revision) + } + if errors.Is(err, jetstream.ErrKeyExists) || (err != nil && isWrongSequence(err)) { + return errSwapped + } + return err +} + +func isWrongSequence(err error) bool { + var ae *jetstream.APIError + return errors.As(err, &ae) && ae.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence +} + +// pipelineEnds follows the runner's ends. A nil *pipelineEnds is the trunk-only want, kept until the first registrations +// (design 47 G §95, §353). +type pipelineEnds struct { + st endStore + wake func() + logf func(string, ...any) + mu sync.Mutex + // firsts is, by module, the first watched branch of its registration, the branch of a placement that names none (design + // 47 G §14). + firsts map[string]string +} + +func newPipelineEnds(st endStore, wake func(), logf func(string, ...any)) *pipelineEnds { + return &pipelineEnds{st: st, wake: wake, logf: logf, firsts: map[string]string{}} +} + +// take keeps one pipeline's end if it is newer than the one kept for its module and branch, by compare-and-set. +func (e *pipelineEnds) take(p mpr.PipelineEnded, passed bool) error { + key := endKey(p.Module, p.Branch) + for { + raw, rev, err := e.st.get(key) + if err != nil { + return err + } + if raw != nil { + var kept End + if err := json.Unmarshal(raw, &kept); err == nil && !kept.newer(p.Source, p.EndCount) { + return nil // an older end, or this one again + } + } + end := End{Source: p.Source, EndCount: p.EndCount, Fp: p.Fp, Commit: p.Commit, Passed: passed, Why: p.Why} + value, err := json.Marshal(end) + if err != nil { + return err + } + if err := e.st.swap(key, value, rev); errors.Is(err, errSwapped) { + continue + } else if err != nil { + return err + } + e.wake() + return nil + } +} + +// markWritten says the wants of module on branch were written from the end it keeps, if that is still the end of fp. +func (e *pipelineEnds) markWritten(module, branch, fp string) { + key := endKey(module, branch) + for { + raw, rev, err := e.st.get(key) + var kept End + if err != nil || raw == nil || json.Unmarshal(raw, &kept) != nil || kept.Fp != fp || kept.Written { + return + } + kept.Written = true + value, _ := json.Marshal(kept) + if err := e.st.swap(key, value, rev); errors.Is(err, errSwapped) { + continue + } else if err != nil { + e.logf("the end of %s on %s is not marked written: %v", module, branch, err) + } + return + } +} + +// passedAt says whether the newest end kept for module on branch passed at fp. +func (e *pipelineEnds) passedAt(module, branch, fp string) bool { + raw, _, err := e.st.get(endKey(module, branch)) + var kept End + return err == nil && raw != nil && json.Unmarshal(raw, &kept) == nil && kept.Passed && kept.Fp == fp +} + +// branchOf is the watched branch a placement names, else the first of the module's registration, else the trunk's. +func (e *pipelineEnds) branchOf(module, placed string) string { + if placed != "" { + return placed + } + e.mu.Lock() + defer e.mu.Unlock() + if b := e.firsts[module]; b != "" { + return b + } + return trunkBranch +} + +// hearPipeline takes a pipeline-ended. A pipeline of a group writes nothing on its own end, and a run on request delivers +// nothing (design 47 G §29, §64). +func (e *pipelineEnds) hearPipeline(m jetstream.Msg) { + p, err := mprc.PipelineEnded(m.Data()) + switch { + case err != nil: + e.logf("%s is not taken: %v", m.Subject(), err) + case p.Group != "" || p.Run != "": + default: + if err := e.take(p, requiredPassed(p)); err != nil { + e.logf("%s: %v; it is heard again", m.Subject(), err) + _ = m.Nak() + return + } + } + _ = m.Ack() +} + +// hearGroup takes a group-ended: each member's end, passed only if the group passed. +func (e *pipelineEnds) hearGroup(m jetstream.Msg) { + g, err := mprc.GroupEnded(m.Data()) + if err != nil { + e.logf("%s is not taken: %v", m.Subject(), err) + _ = m.Ack() + return + } + for _, p := range g.Pipelines { + if err := e.take(p, g.Outcome == mpr.GroupEndedOutcomePassed && requiredPassed(p)); err != nil { + e.logf("%s: %v; it is heard again", m.Subject(), err) + _ = m.Nak() + return + } + } + _ = m.Ack() +} + +// hearEnds registers the two events on the assigner's hub. +func hearEnds(hub *process.EventHub, e *pipelineEnds) error { + if err := hub.Handle(pipelineEndedSubject, e.hearPipeline); err != nil { + return fmt.Errorf("%s: %w", pipelineEndedSubject, err) + } + if err := hub.Handle(groupEndedSubject, e.hearGroup); err != nil { + return fmt.Errorf("%s: %w", groupEndedSubject, err) + } + return nil +} + +// writtenTools is the tool assigner_written: every key of the state written, with its end. +func writtenTools(st store) []stdio.Tool { + return []stdio.Tool{{Name: writtenTool, Description: "the pipeline ends the assigner's wants follow: per module and watched branch, the source revision and end count of the newest end heard, whether it passed and whether the wants were written from it.", + Run: func(map[string]any) (any, error) { + ks, err := st.keys(writtenState) + if err != nil { + return nil, err + } + sort.Strings(ks) + out := map[string]End{} + for _, k := range ks { + raw, _, ok, err := st.get(writtenState, k) + if err != nil || !ok { + continue + } + var e End + if json.Unmarshal(raw, &e) == nil { + out[k] = e + } + } + return out, nil + }}} +} + +// branchesOf is the watched branches the placements of module chose, each once; the default branch when none names one. +func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]string) []string { + if e == nil { + return nil + } + seen := map[string]bool{} + var out []string + for _, b := range placed[module] { + b = e.branchOf(module, b) + if !seen[b] { + seen[b] = true + out = append(out, b) + } + } + if len(out) == 0 { + out = append(out, e.branchOf(module, "")) + } + sort.Strings(out) + return out +} + +// mayWant says whether module's wants may name fp: a pipeline passed at fp on every branch, or the wants name fp already +// (a node placed later, a module taken away). +func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[string]mwc.Want) bool { + if e == nil { + return true + } + for _, w := range wanted { + if w.Fp == fp { + return true + } + } + for _, b := range branches { + if !e.passedAt(module, b, fp) { + return false + } + } + return true +} + +// wrote marks the end the wants of module were written from. +func (e *pipelineEnds) wrote(module, fp string, branches []string) { + if e == nil { + return + } + for _, b := range branches { + e.markWritten(module, b, fp) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go new file mode 100644 index 00000000..17597d49 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go @@ -0,0 +1,174 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-sdk/go/catalogue" + + "git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + + "mesh-delivery-assigner/internal/buckets" +) + +// memEnds is endStore over a map, with revisions. +type memEnds struct { + vals map[string][]byte + revs map[string]uint64 +} + +func newMemEnds() *memEnds { return &memEnds{vals: map[string][]byte{}, revs: map[string]uint64{}} } + +func (m *memEnds) get(key string) ([]byte, uint64, error) { return m.vals[key], m.revs[key], nil } +func (m *memEnds) swap(key string, v []byte, rev uint64) error { + if m.revs[key] != rev { + return errSwapped + } + m.vals[key], m.revs[key] = v, rev+1 + return nil +} + +func pipe(module, branch string, source, count int64, fp, outcome string, steps ...mpr.StepResult) mpr.PipelineEnded { + return mpr.PipelineEnded{Module: module, Branch: branch, Source: source, EndCount: count, Fp: fp, Outcome: outcome, Steps: steps, + Pipeline: branch + "." + module, Commit: "c"} +} + +func step(status, outcome string) mpr.StepResult { + return mpr.StepResult{Step: "unit", Status: status, Outcome: outcome} +} + +// Design 47 G §29, §35, §325, §352 (tests 66, 67, 89): a want follows a passed pipeline, by compare-and-set per module and branch. +func TestAWantFollowsAPassedPipeline(t *testing.T) { + const fp1, fp2 = "fp1", "fp2" + pass, fail := mpr.PipelineEndedOutcomePassed, mpr.PipelineEndedOutcomeFailed + req, opt := mpr.StepResultStatusRequired, mpr.StepResultStatusOptional + for _, c := range []struct { + name string + ends []mpr.PipelineEnded + group *mpr.GroupEnded + module string + branch string + fp string + may bool + }{ + {"a failed required step writes no want", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, fail, step(req, "failed"))}, nil, "web", "main", fp1, false}, + {"a pipeline passed writes", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"))}, nil, "web", "main", fp1, true}, + {"a failed optional step holds nothing", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"), step(opt, "failed"))}, nil, "web", "main", fp1, true}, + {"a pipeline passed at another fingerprint", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}, nil, "web", "main", fp2, false}, + {"an older revision ending after a newer one writes nothing", []mpr.PipelineEnded{pipe("web", "main", 6, 1, fp2, fail), pipe("web", "main", 5, 1, fp1, pass)}, nil, "web", "main", fp1, false}, + {"another branch writes its own", []mpr.PipelineEnded{pipe("web", "main", 6, 1, fp2, fail), pipe("web", "beta", 5, 1, fp1, pass)}, nil, "web", "beta", fp1, true}, + {"a later end of the same revision writes", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, fail, step(req, "failed")), pipe("web", "main", 5, 2, fp1, pass, step(req, "passed"))}, nil, "web", "main", fp1, true}, + {"the same end again after a later one keeps the later", []mpr.PipelineEnded{pipe("web", "main", 5, 2, fp1, pass), pipe("web", "main", 5, 1, fp1, fail)}, nil, "web", "main", fp1, true}, + {"a member's own end writes nothing", []mpr.PipelineEnded{func() mpr.PipelineEnded { p := pipe("web", "main", 5, 1, fp1, pass); p.Group = "g"; return p }()}, nil, "web", "main", fp1, false}, + {"a group ended passed writes its members", nil, &mpr.GroupEnded{Branch: "main", Group: "g", Outcome: mpr.GroupEndedOutcomePassed, Pipelines: []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}}, "web", "main", fp1, true}, + {"a group ended failed writes none", nil, &mpr.GroupEnded{Branch: "main", Group: "g", Outcome: mpr.GroupEndedOutcomeFailed, Pipelines: []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}}, "web", "main", fp1, false}, + } { + woken := 0 + e := newPipelineEnds(newMemEnds(), func() { woken++ }, t.Logf) + for _, p := range c.ends { + if p.Group != "" { + continue // heard and ignored, as hearPipeline does + } + if err := e.take(p, requiredPassed(p)); err != nil { + t.Fatal(err) + } + } + if g := c.group; g != nil { + for _, p := range g.Pipelines { + if err := e.take(p, g.Outcome == mpr.GroupEndedOutcomePassed && requiredPassed(p)); err != nil { + t.Fatal(err) + } + } + } + if got := e.mayWant(c.module, c.fp, []string{c.branch}, nil); got != c.may { + t.Errorf("%s: may want = %v, want %v", c.name, got, c.may) + } + } + // A want that names the fingerprint already is not held by a later failure (a node placed afterwards gets it). + e := newPipelineEnds(newMemEnds(), func() {}, t.Logf) + if !e.mayWant("web", fp1, []string{"main"}, map[string]mwc.Want{"g14": {Fp: fp1}}) { + t.Error("a want naming the fingerprint was held") + } + // markWritten keeps the end and says it was written from; a newer end replaces it unwritten. + st := newMemEnds() + e = newPipelineEnds(st, func() {}, t.Logf) + _ = e.take(pipe("web", "main", 5, 1, fp1, pass), true) + e.wrote("web", fp1, []string{"main"}) + var end End + _ = json.Unmarshal(st.vals[endKey("web", "main")], &end) + if !end.Written || end.Source != 5 { + t.Errorf("end = %+v, want written at source 5", end) + } + _ = e.take(pipe("web", "main", 6, 1, fp2, pass), true) + end = End{} + _ = json.Unmarshal(st.vals[endKey("web", "main")], &end) + if end.Written || end.Source != 6 { + t.Errorf("a newer end = %+v, want unwritten at source 6", end) + } +} + +// Design 47 G §29: the wants of a pass follow a pipeline: with the follower on, a module whose fingerprint no pipeline passed at +// gets no want and one a pipeline passed at gets it; with it off (nil, the trunk-only want) the pass is as before. +func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) { + fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) + for _, c := range []struct { + name string + follow bool + passed bool + want bool + }{{"the trunk-only want", false, false, true}, {"no pipeline passed", true, false, false}, {"a pipeline passed", true, true, true}} { + wantsKV := &memKV{values: map[string][]byte{}} + p := newPasser(kvSet{buckets.Wants: wantsKV, buckets.Asks: &memKV{values: map[string][]byte{}}}, time.Now()) + p.say = func(oc.Condition) error { return nil } + mdigest := strings.Repeat("e", 64) + mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: mdigest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) + p.w.put(buckets.Modules, "web", mod) + p.w.manifests[mdigest] = parsed{module: "web", manifest: catalogue.Manifest{Module: "web"}} + bkey, _ := builds.BuilderBuildKey("web", fp) + built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest}) + p.w.put(buckets.Builds, bkey, built) + gkey, _ := grantsstate.PartGrantKey("web", fp) + grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{}) + p.w.put(buckets.Grants, gkey, grant) + digest := "sha256:" + strings.Repeat("c", 64) + have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) + p.w.put(buckets.Have, "g14.web", have) + composes(t, p.w, "g14", digest) + placeOn(t, p.w, "g14", "web") + on, _ := presenceValue(true) + p.w.put(buckets.Presence, "g14", on) + if c.follow { + p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf) + if c.passed { + _ = p.ends.take(pipe("web", trunkBranch, 5, 1, fp, mpr.PipelineEndedOutcomePassed), true) + } + } + p.wants(context.Background()) + key, _ := mwc.WantKey("g14", "web") + if _, got := wantsKV.values[key]; got != c.want { + t.Errorf("%s: a want written = %v, want %v", c.name, got, c.want) + } + if c.follow && c.passed { + raw, _, _ := p.ends.st.get(endKey("web", trunkBranch)) + var end End + _ = json.Unmarshal(raw, &end) + if !end.Written { + t.Errorf("%s: the end was not marked written", c.name) + } + } + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index d81e9acf..9b946428 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -404,6 +404,7 @@ func serveTools() error { tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...) // What each node's engine reads, as the assigner uses it (ADR 0618 §3): its own state engine-reads, asked again through it. tools = append(tools, enginereads.Tool(engineReadsTool, "the assigner", enginereads.KeptState(stdio.State(engineReadsState)), false, time.Now)) + tools = append(tools, writtenTools(runtimeStore{})...) tools = append(tools, askAgainTools(runtimeStore{}, stdio.Ask, time.Now)...) return stdio.Serve("", tools) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 893d02f5..a5a1ca9a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -578,3 +578,17 @@ func (w *world) unplacedNodes() []string { // trunkBranch is the branch of the trunk-only want, kept until the first registrations (design 47 G §95, §353). const trunkBranch = "main" + +// placedBranches is, by module and node, the watched branch each placement names ("" when it names none). +func (w *world) placedBranches() map[string]map[string]string { + out := map[string]map[string]string{} + for node, a := range w.assignments() { + for _, m := range a.Modules { + if out[m.Module] == nil { + out[m.Module] = map[string]string{} + } + out[m.Module][node] = m.Branch + } + } + return out +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index c1ceeaea..629dcee0 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -65,7 +65,7 @@ func TestManifest(t *testing.T) { } if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20, "assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1, - "engine-reads": 1}; !reflect.DeepEqual(states, want) { + "engine-reads": 1, "written": 5}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -96,7 +96,7 @@ func TestManifest(t *testing.T) { // which lists and asks again what each node's engine reads. if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assignments", "seat-holders", "wants", "assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies", - "assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again"}) { + "assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written"}) { t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads, the seat list's two (ADR 0636 §7) and assigner_ask_again (ADR 0668 §5)", m.Tools) } if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs { @@ -123,7 +123,7 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "node-engine.manifest-reads-changed"}) { + if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) { t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes) } if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index e594b25c..d81cca86 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -45,6 +45,8 @@ "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "mesh-delivery-teller.ask-ended", + "mesh-pipeline-runner.pipeline-ended", + "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed" ], "invokes": [ @@ -90,7 +92,8 @@ "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", - "assigner_ask_again" + "assigner_ask_again", + "assigner_written" ], "state": [ { "name": "assignments", "history": 64 }, @@ -108,7 +111,8 @@ { "name": "seat-holders", "history": 5 }, { "name": "view-marks", "history": 1 }, { "name": "first-views", "history": 1 }, - { "name": "engine-reads", "history": 1 } + { "name": "engine-reads", "history": 1 }, + { "name": "written", "history": 5 } ], "own-secrets": { "broker": "${dir:state}/broker"