AS1: a want follows a passed pipeline, by compare-and-set per module and watched branch
The assigner hears pipeline-ended and group-ended, keeps the newest end per module and branch in the state written (tool assigner_written), and writes a want only from an end whose required steps passed. Builds: design 47 G §29, §35, §102, §120, §291, §325, §352, §85 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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 `<node>`, 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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user