diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go index c793d73b..5f611a66 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go @@ -18,7 +18,7 @@ import ( ) // The inputs of a composition step (design 47 G §354, A-243, A-252): one request per machine, each carrying that machine's input, -// asked of the machine's node-engine by the seat's verb compose-candidate and kept by the digest the run ask names it by. +// asked of the machine's node-engine by the seat's verb compose-candidate. // composeCandidateKey is the seat verb's call key, asked of one node. const composeCandidateKey = "seat:" + ne.Seat + "." + ne.VerbComposeCandidate @@ -59,28 +59,28 @@ func (w *world) machinesOf(module, branch string) []string { return machinesFor(w.placedBranches(), module, branch, w.holdsBranch(branch)) } -// nodeInput is one machine's input: its digest and bytes, or why the node gave none. +// nodeInput is one machine's input: its bytes, or why the node gave none, or that it is away. type nodeInput struct { - Node string - Digest string - Body []byte + Node string + Body []byte // Why is set when the node gave no input: its words, or the call's error. Why string + // Away is a machine the watchman said away: it is not asked (§64). + Away bool } -// compositionInputs asks the machines of a step for their inputs and keeps them. +// compositionInputs asks the machines of a step for their inputs. type compositionInputs struct { machines func(module, branch string) []string // call asks one node's seat verb (process.Conn.CallKeptOn). - call func(node, key string, body any) (json.RawMessage, error) - keeper *inputKeeper - bound time.Duration + call func(node, key string, body any) (json.RawMessage, error) + bound time.Duration } -// gather asks each machine of the step's module and branch for its input with the step's candidate key, keeps each input under the -// ask askOf names for that node (the build seat's step-ended of that ask deletes it), and answers every machine in name order. A node -// that is silent, unready or answers no input has Why set and keeps none; no machine at all is an error. -func (c *compositionInputs) gather(ctx context.Context, run meshwants.StepRun, askOf func(node string) string) ([]nodeInput, error) { +// gather asks each machine of the step's module and branch for its input with the step's candidate key, and answers every machine in +// name order: one away (away) is not asked; a node that is silent, unready or answers no input has Why set; no machine at all is an +// error. The run keeps each request built from an input under its run ask (compositionRun.machine). +func (c *compositionInputs) gather(ctx context.Context, run meshwants.StepRun, away func(node string) bool) ([]nodeInput, error) { key, err := candidateKeyOf(run) if err != nil { return nil, err @@ -96,18 +96,22 @@ func (c *compositionInputs) gather(ctx context.Context, run meshwants.StepRun, a out := make([]nodeInput, len(nodes)) var wg sync.WaitGroup for i, node := range nodes { + if away != nil && away(node) { + out[i] = nodeInput{Node: node, Away: true} + continue + } wg.Add(1) go func() { defer wg.Done() - out[i] = c.one(ctx, bound, node, key, askOf(node)) + out[i] = c.one(ctx, bound, node, key) }() } wg.Wait() return out, nil } -// one asks one node and keeps its input. -func (c *compositionInputs) one(ctx context.Context, bound time.Duration, node, key, ask string) nodeInput { +// one asks one node for its input. +func (c *compositionInputs) one(ctx context.Context, bound time.Duration, node, key string) nodeInput { in := nodeInput{Node: node} at, cancel := context.WithTimeout(ctx, bound) defer cancel() @@ -156,11 +160,6 @@ func (c *compositionInputs) one(ctx context.Context, bound time.Duration, node, in.Why = "its composition input cannot be encoded again: " + err.Error() return in } - digest, err := c.keeper.keep(ask, body) - if err != nil { - in.Why = "its input could not be kept: " + err.Error() - return in - } - in.Digest, in.Body = digest, body + in.Body = body return in } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs_test.go index 55c7f67d..9b0f2bd9 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs_test.go @@ -52,9 +52,9 @@ func TestTheMachinesOfAStep(t *testing.T) { } } -// §354, A-243, A-252: each machine is asked compose-candidate with the step's key and input true, and its input is kept under the ask -// named for it; a silent or unready node keeps none and says why. -func TestAStepsInputsAreAskedOfEachMachineAndKept(t *testing.T) { +// §354, A-243, A-252, §64: each machine is asked compose-candidate with the step's key and input true, its numbers as written; a +// silent or unready node says why, and a machine away is not asked. +func TestAStepsInputsAreAskedOfEachMachine(t *testing.T) { asked := map[string]string{} call := func(node, key string, body any) (json.RawMessage, error) { raw, _ := json.Marshal(body) @@ -67,10 +67,9 @@ func TestAStepsInputsAreAskedOfEachMachineAndKept(t *testing.T) { } return json.RawMessage(`{"candidate":"run-r1","node":"` + node + `","verdict":"composed","input":{"n":12345678901234567890,"node":"` + node + `"},"modules":[],"own":{},"took":1,"v":1}`), nil } - keeper := testKeeper(&objectsMem{objs: map[string][]byte{}}, time.Now) - c := &compositionInputs{machines: func(module, branch string) []string { return []string{"ace", "g14", "novox"} }, call: call, keeper: keeper, bound: time.Second} - got, err := c.gather(context.Background(), meshwants.StepRun{Module: "m", Branch: "main", Run: "r1"}, func(node string) string { return "ask-" + node }) - if err != nil || len(got) != 3 { + c := &compositionInputs{machines: func(module, branch string) []string { return []string{"ace", "g14", "novox", "shanks"} }, call: call, bound: time.Second} + got, err := c.gather(context.Background(), meshwants.StepRun{Module: "m", Branch: "main", Run: "r1"}, func(node string) bool { return node == "shanks" }) + if err != nil || len(got) != 4 { t.Fatalf("%+v %v", got, err) } if want := `seat:node-engine.compose-candidate {"candidate":"run-r1","input":true}`; asked["ace"] != want { @@ -79,19 +78,13 @@ func TestAStepsInputsAreAskedOfEachMachineAndKept(t *testing.T) { if got[0].Node != "ace" || got[0].Why != "" || !strings.Contains(string(got[0].Body), "12345678901234567890") { t.Errorf("ace: %+v", got[0]) } - if b, err := keeper.body(got[0].Digest); err != nil || string(b) != string(got[0].Body) { - t.Errorf("the input of ace is not kept: %v", err) - } - if !strings.Contains(got[1].Why, "no responders") || !strings.Contains(got[2].Why, "unready") || got[1].Digest != "" || got[2].Digest != "" { + if !strings.Contains(got[1].Why, "no responders") || !strings.Contains(got[2].Why, "unready") || got[1].Body != nil || got[2].Body != nil { t.Errorf("g14 and novox: %+v %+v", got[1], got[2]) } - if err := keeper.ended("ask-ace"); err != nil { - t.Fatal(err) + if _, was := asked["shanks"]; was || !got[3].Away { + t.Errorf("the machine away is asked or not said away: %+v", got[3]) } - if _, err := keeper.body(got[0].Digest); err == nil { - t.Error("the input is kept past its step-ended") - } - none := &compositionInputs{machines: func(string, string) []string { return nil }, call: call, keeper: keeper} + none := &compositionInputs{machines: func(string, string) []string { return nil }, call: call} if _, err := none.gather(context.Background(), meshwants.StepRun{Module: "m", Run: "r1"}, nil); err == nil { t.Error("a step with no machine is gathered") } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go new file mode 100644 index 00000000..edf581ca --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go @@ -0,0 +1,498 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + meshcatalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// The run of a composition step (design 47 G §354, §63, §64; A-226, A-238, A-240, A-243, A-252, A-254, A-261): one +// composer.Request per machine, depth compose, the one machine and one case `change`, its manifests, verdicts and seats from the +// step's catalog revision read through mesh-catalog's `revision`; one run ask per machine of the seat node-build-go, naming no +// head, its program the node-engine build's catalog entry with the composer's package; the run's end heard on the build seat's +// step-ended of that ask, its answer (a composer.Verdict) read inline from `output`. A machine the watchman said away +// (machine-away) is not composed and passes with a warning (§64). + +const ( + // runAsker is the assigner's name as an asker of a build seat: the last token of the subject its run asks are put on and + // the first of their step keys (§467, §481, §600). + runAsker = "assigner" + // changeCase is the one case each request composes. + changeCase = "change" + // inputVerbKey is the verb a run ask's input is served on: the seat mesh-wants' input (§354). + inputVerbKey = "seat:" + meshwants.Seat + "." + meshwants.VerbInput + // entryVerbKey and seatDefinitionsKey are the mesh-catalog seat verbs the step reads. + entryVerbKey = "seat:" + meshcatalog.Seat + "." + meshcatalog.VerbEntry + seatDefinitionsKey = "seat:" + meshcatalog.Seat + "." + meshcatalog.VerbSeatDefinitions +) + +// runEnds hands each build seat's step-ended of an ask to the composition run that waits for it (A-240), by the ask's id. +type runEnds struct { + mu sync.Mutex + waits map[string]chan buildstep.StepEnded +} + +// expect is the end of the ask id, once heard; called before the ask is put, so no end is missed. +func (r *runEnds) expect(id string) <-chan buildstep.StepEnded { + r.mu.Lock() + defer r.mu.Unlock() + if r.waits == nil { + r.waits = map[string]chan buildstep.StepEnded{} + } + ch := make(chan buildstep.StepEnded, 1) + r.waits[id] = ch + return ch +} + +// forget stops waiting for the ask id. +func (r *runEnds) forget(id string) { + r.mu.Lock() + delete(r.waits, id) + r.mu.Unlock() +} + +// heard hands an end to the run that waits for its ask; one nobody waits for is another asker's, left. +func (r *runEnds) heard(e buildstep.StepEnded) { + r.mu.Lock() + ch := r.waits[e.Ask] + delete(r.waits, e.Ask) + r.mu.Unlock() + if ch != nil { + ch <- e + } +} + +// awayNodes are the machines the watchman said away (machine-away of the seat mesh-watchdogs, §389) and not back since. +type awayNodes struct { + mu sync.Mutex + away map[string]bool +} + +// set keeps machine away (true) or back (false). +func (a *awayNodes) set(machine string, away bool) { + a.mu.Lock() + defer a.mu.Unlock() + if a.away == nil { + a.away = map[string]bool{} + } + if away { + a.away[machine] = true + } else { + delete(a.away, machine) + } +} + +// is says machine is away. +func (a *awayNodes) is(machine string) bool { + if a == nil { + return false + } + a.mu.Lock() + defer a.mu.Unlock() + return a.away[machine] +} + +// compositionRun runs a composition step. +type compositionRun struct { + inputs *compositionInputs + keeper *inputKeeper + // call asks a seat verb held once for the mesh (process.Conn.CallKept). + call func(key string, body any) (json.RawMessage, error) + // trunk is the trunk's catalog half as the assigner holds it. + trunk func() (mcc.Half, bool) + // running is the last `running` heard of a module on a node (world.runningOf), never the node-engine's bucket. + running func(node, module string) (ne.Running, bool) + away *awayNodes + // put puts a run ask on the build seat's check queue under the assigner's name. + put func(ctx context.Context, a buildstep.StepAsk) error + ends *runEnds + // ask asks a seat verb on one node, for the run's answer kept as the build seat's object (buildstep.StepEnded.ReadOutput). + ask buildstep.AskVerb + now func() time.Time + logf func(string, ...any) +} + +// machineEnd is what one machine came to. +type machineEnd struct { + failed, couldNot, warning string +} + +// run runs one composition step and answers what it came to: failed when a machine does not compose, could not run when a machine +// could not be composed, else passed with a warning per machine away or not judged. +func (r *compositionRun) run(ctx context.Context, _ string, run meshwants.StepRun) stepOutcome { + half, err := r.revision(run) + if err != nil { + return couldNotRun("the step's catalog revision is not read: " + err.Error()) + } + seats, err := r.seats(half.Revision.Definitions) + if err != nil { + return couldNotRun(err.Error()) + } + manifests, verdicts, err := caseOf(half) + if err != nil { + return couldNotRun(err.Error()) + } + programOf, err := r.programs(half) + if err != nil { + return couldNotRun(err.Error()) + } + inputs, err := r.inputs.gather(ctx, run, r.away.is) + if err != nil { + return couldNotRun(err.Error()) + } + ends := make([]machineEnd, len(inputs)) + var wg sync.WaitGroup + for i, in := range inputs { + wg.Add(1) + go func() { + defer wg.Done() + ends[i] = r.machine(ctx, in, buildstep.NewID(), programOf, manifests, verdicts, seats) + }() + } + wg.Wait() + var out stepOutcome + var failed, couldNot []string + for _, e := range ends { + failed = appendSome(failed, e.failed) + couldNot = appendSome(couldNot, e.couldNot) + out.Warnings = appendSome(out.Warnings, e.warning) + } + switch { + case len(failed) > 0: + out.Outcome, out.Said = meshwants.StepEndedOutcomeFailed, strings.Join(failed, "; ") + case len(couldNot) > 0: + out.Outcome, out.Said = meshwants.StepEndedOutcomeCouldNotRun, strings.Join(couldNot, "; ") + default: + out.Outcome = meshwants.StepEndedOutcomePassed + } + return out +} + +func appendSome(to []string, s string) []string { + if s == "" { + return to + } + return append(to, s) +} + +func couldNotRun(said string) stepOutcome { + return stepOutcome{Outcome: meshwants.StepEndedOutcomeCouldNotRun, Said: said} +} + +// machine composes one machine: its request kept under its ask, the ask put, its end heard and its verdict read. +func (r *compositionRun) machine(ctx context.Context, in nodeInput, ask string, programOf func(node string) (buildstep.RunProgram, error), + manifests map[string]composer.Manifest, verdicts map[string]json.RawMessage, seats []json.RawMessage) machineEnd { + node := in.Node + switch { + case in.Away: + return machineEnd{warning: node + " is away (the watchman's machine-away), so it is not composed"} + case in.Why != "": + return machineEnd{couldNot: node + " gave no composition input: " + in.Why} + } + program, err := programOf(node) + if err != nil { + return machineEnd{couldNot: node + ": " + err.Error()} + } + req := composer.Request{V: composer.Version, Input: in.Body, Depth: composer.DepthCompose, Machines: []string{node}, + Cases: []composer.Case{{Name: changeCase, Manifests: manifests, Verdicts: verdicts}}, Seats: seats} + raw, err := json.Marshal(req) + if err != nil { + return machineEnd{couldNot: node + ": its request cannot be encoded: " + err.Error()} + } + if _, err := r.keeper.keep(ask, raw); err != nil { + r.logf("the composition request of %s is refused: %v", node, err) + return machineEnd{couldNot: node + ": its request is not kept: " + err.Error()} + } + ref, err := buildstep.RunInputOf(inputVerbKey, raw) + if err != nil { + return machineEnd{couldNot: node + ": " + err.Error()} + } + a := buildstep.StepAsk{ID: ask, Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: runAsker, At: r.now().UTC(), + Run: &buildstep.RunJob{Program: program, Heads: []buildstep.RunHead{}, Input: ref, Reader: "the composition of " + node + " by the node-engine at " + program.Commit}} + ended := r.ends.expect(ask) + defer r.ends.forget(ask) + if err := r.put(ctx, a); err != nil { + return machineEnd{couldNot: node + ": its run ask " + ask + " is not put: " + err.Error()} + } + var e buildstep.StepEnded + select { + case <-ctx.Done(): + return machineEnd{couldNot: node + ": the step stopped before its run ask " + ask + " ended"} + case e = <-ended: + } + if e.Outcome != buildstep.OutcomePassed { + return machineEnd{couldNot: fmt.Sprintf("%s: its run ask %s ended %s on %s: %s", node, ask, e.Outcome, e.Node, e.Said)} + } + answer, err := e.ReadOutput(buildstep.AskGo, r.ask) // the program's last line of standard output (A-254) + if err != nil { + return machineEnd{couldNot: node + ": " + err.Error()} + } + return verdictOf(node, answer) +} + +// verdictOf reads the composer's verdict of one machine: not composing is failed, naming why; not judged passes with a warning. +func verdictOf(node string, answer []byte) machineEnd { + var v composer.Verdict + if err := json.Unmarshal(answer, &v); err != nil { + return machineEnd{couldNot: node + ": the composer answered no verdict: " + err.Error()} + } + if v.V != composer.Version { + return machineEnd{couldNot: fmt.Sprintf("%s: the composer answered a verdict of version %d, the assigner reads %d", node, v.V, composer.Version)} + } + for _, c := range v.Cases { + if c.Name != changeCase { + continue + } + m, ok := c.Machines[node] + switch { + case !ok: + return machineEnd{couldNot: node + ": the composer's verdict names no such machine"} + case m.NotJudged != "": + return machineEnd{warning: node + " was not judged: " + m.NotJudged} + case !m.Composes: + return machineEnd{failed: node + " does not compose: " + strings.Join(appendSome(append(slices.Clone(m.Problems), m.Refusals...), m.Reason), "; ")} + } + return machineEnd{} + } + return machineEnd{couldNot: node + ": the composer's verdict has no case " + changeCase} +} + +// revision reads the step's catalog revision whole, page by page, through mesh-catalog's `revision` (A-252). +func (r *compositionRun) revision(run meshwants.StepRun) (mcc.Half, error) { + if run.Revision <= 0 { + return mcc.Half{}, errors.New("the step names no catalog revision") + } + half := mcc.Half{Entries: map[string]meshcatalog.CatalogEntry{}} + in := meshcatalog.RevisionInput{Branch: keys.EscapeName(run.Branch), Revision: run.Revision} + for { + raw, err := r.call(mcc.RevisionVerb, in) + if err != nil { + return mcc.Half{}, err + } + body, said, ok := buscalls.Unwrap(raw) + if !ok { + return mcc.Half{}, fmt.Errorf("%s answered: %s", mcc.RevisionVerb, said) + } + var a meshcatalog.RevisionAnswer + if err := json.Unmarshal(body, &a); err != nil { + return mcc.Half{}, fmt.Errorf("%s answered no revision: %w", mcc.RevisionVerb, err) + } + if a.NoneYet || a.Revision == nil || a.Revision.Revision != run.Revision { + return mcc.Half{}, fmt.Errorf("%s answered no revision %d of %q", mcc.RevisionVerb, run.Revision, run.Branch) + } + if in.After == "" { + half.Revision = *a.Revision // the first page names what the revision was read from, its definitions included + } + for _, e := range a.Entries { + half.Entries[e.Module] = e + } + if a.Next == "" { + return half, nil + } + in.After = a.Next + } +} + +// seats are the definitions of the revision's (seat, digest) pairs, raw (A-261): the revision's definitions carry no definition +// yet (A-234), so each is mesh-catalog's seat-definitions answer for the seat where the trunk's digest is the revision's; a digest +// neither holds is an error naming the seat and the digest. +func (r *compositionRun) seats(defs []meshcatalog.CatalogRevisionDefinitionsItem) ([]json.RawMessage, error) { + out := make([]json.RawMessage, 0, len(defs)) + for _, d := range defs { + raw, err := r.call(seatDefinitionsKey, meshcatalog.SeatDefinitionsInput{Seat: d.Seat}) + if err != nil { + return nil, fmt.Errorf("the definition of the seat %s is not read: %w", d.Seat, err) + } + body, said, ok := buscalls.Unwrap(raw) + if !ok { + return nil, fmt.Errorf("the definition of the seat %s is not read: %s", d.Seat, said) + } + var a meshcatalog.SeatDefinitionsAnswer + if err := json.Unmarshal(body, &a); err != nil { + return nil, fmt.Errorf("the definition of the seat %s is not read: %w", d.Seat, err) + } + if a.Entry == nil || a.Entry.Digest != d.Digest || a.Entry.Definition == nil { + return nil, fmt.Errorf("no definition of the seat %s with the digest %s is held: neither the revision nor the trunk holds it", d.Seat, d.Digest) + } + def, err := json.Marshal(a.Entry.Definition) + if err != nil { + return nil, err + } + out = append(out, def) + } + return out, nil +} + +// caseOf is the case's manifests and verdicts from the revision's entries: each manifest whole, and the verdict of each one not +// accepted (a module without one is accepted). +func caseOf(half mcc.Half) (map[string]composer.Manifest, map[string]json.RawMessage, error) { + manifests := map[string]composer.Manifest{} + verdicts := map[string]json.RawMessage{} + for name, e := range half.Entries { + if len(e.Manifest) == 0 { + continue + } + raw, err := json.Marshal(e.Manifest) + if err != nil { + return nil, nil, fmt.Errorf("the manifest of %s cannot be encoded: %w", name, err) + } + manifests[name] = composer.Manifest{Raw: raw} + if e.Verdict == "" || e.Verdict == mcc.VerdictAccepted { + continue + } + v := meshcatalog.VerdictsByDigestValue{V: 1, Module: name, State: e.Verdict} + if e.Verdict != mcc.VerdictRefused { + v.State = mcc.VerdictManifestWaits + } + if e.Inputs != nil { + v.Problems, v.Unknown, v.Warnings = e.Inputs.Problems, e.Inputs.Unknown, e.Inputs.Warnings + } + raw, err = json.Marshal(v) + if err != nil { + return nil, nil, err + } + verdicts[name] = raw + } + return manifests, verdicts, nil +} + +// nodeEngineEntry is the entry of the module that claims the seat node-engine, or false. +func nodeEngineEntry(entries map[string]meshcatalog.CatalogEntry) (meshcatalog.CatalogEntry, bool) { + for _, name := range slices.Sorted(mapsKeys(entries)) { + for _, s := range entries[name].Serves { + if strings.HasPrefix(s, ne.Seat+".") { + return entries[name], true + } + } + } + return meshcatalog.CatalogEntry{}, false +} + +func mapsKeys(m map[string]meshcatalog.CatalogEntry) func(func(string) bool) { + return func(yield func(string) bool) { + for k := range m { + if !yield(k) { + return + } + } + } +} + +// programOfEntry is the composer built from a node-engine entry: its repository, commit and directory, the composer's package. +func programOfEntry(e meshcatalog.CatalogEntry) (buildstep.RunProgram, error) { + dir := e.Path + if dir == "" { + dir = "." + } + if e.Repository == "" || e.Commit == "" { + return buildstep.RunProgram{}, fmt.Errorf("the node-engine's entry %s names no repository or commit", e.Fp) + } + return buildstep.RunProgram{Repository: e.Repository, Commit: e.Commit, Dir: dir, Package: composer.Package}, nil +} + +// programs answers, per machine, the composer its run builds (A-226, A-243, A-252): where the revision's node-engine entry differs +// from the trunk's, that build for every machine; else the build the machine's node-engine runs, as its `running` says, its entry +// read through mesh-catalog's `entry`. +func (r *compositionRun) programs(half mcc.Half) (func(node string) (buildstep.RunProgram, error), error) { + engine, ok := nodeEngineEntry(half.Entries) + if !ok { + return nil, fmt.Errorf("the catalog revision %d holds no module claiming the seat %s", half.Revision.Revision, ne.Seat) + } + if trunk, ok := r.trunk(); !ok || func() bool { t, ok := nodeEngineEntry(trunk.Entries); return !ok || t.Fp != engine.Fp }() { + p, err := programOfEntry(engine) + return func(string) (buildstep.RunProgram, error) { return p, err }, nil + } + var mu sync.Mutex + byFp := map[string]buildstep.RunProgram{} + return func(node string) (buildstep.RunProgram, error) { + running, ok := r.running(node, engine.Module) + if !ok || running.Fingerprint == "" { + return buildstep.RunProgram{}, fmt.Errorf("no build of %s is heard running there", engine.Module) + } + mu.Lock() + p, ok := byFp[running.Fingerprint] + mu.Unlock() + if ok { + return p, nil + } + raw, err := r.call(entryVerbKey, meshcatalog.EntryInput{Fingerprint: running.Fingerprint}) + if err != nil { + return buildstep.RunProgram{}, fmt.Errorf("the entry of the node-engine's build %s is not read: %w", running.Fingerprint, err) + } + body, said, ok := buscalls.Unwrap(raw) + if !ok { + return buildstep.RunProgram{}, fmt.Errorf("the entry of the node-engine's build %s is not read: %s", running.Fingerprint, said) + } + var a meshcatalog.EntryAnswer + if err := json.Unmarshal(body, &a); err != nil { + return buildstep.RunProgram{}, fmt.Errorf("the entry of the node-engine's build %s is not read: %w", running.Fingerprint, err) + } + p, err = programOfEntry(a.Entry) + if err != nil { + return p, err + } + mu.Lock() + byFp[running.Fingerprint] = p + mu.Unlock() + return p, nil + }, nil +} + +// heardPresence keeps a machine-away (away true) or machine-back of the seat mesh-watchdogs (§389, §436): the machine it names. +func heardPresence(a *awayNodes, away bool, m jetstream.Msg, logf func(string, ...any)) { + var e struct { + Machine string `json:"machine"` + } + if err := json.Unmarshal(m.Data(), &e); err != nil || e.Machine == "" { + logf("%s holds no machine, left: %v", m.Subject(), err) + } else { + a.set(e.Machine, away) + } + if err := m.Ack(); err != nil { + logf("acknowledging %s: %v", m.Subject(), err) + } +} + +// trunkOf is the trunk's catalog half the assigner's views hold, or false while none is held. +func trunkOf(iv *catalogViews) func() (mcc.Half, bool) { + return func() (mcc.Half, bool) { + if iv == nil || iv.h == nil { + return mcc.Half{}, false + } + h, ok := iv.h.views()[trunkBranch] + return h, ok + } +} + +// runAskPublisher puts a run ask on the Go build seat's check queue under the assigner's name (`accept.check.assigner`, §600), its +// id the message id, so the stream takes it once. +func runAskPublisher(js jetstream.JetStream) func(ctx context.Context, a buildstep.StepAsk) error { + return func(ctx context.Context, a buildstep.StepAsk) error { + raw, err := buildstep.EncodeStepAsk(a) + if err != nil { + return err + } + subject, err := buildstep.AskSubjectBy(a.Language, a.Kind, a.Asker) + if err != nil { + return err + } + _, err = js.Publish(ctx, subject, raw, jetstream.WithMsgID(a.ID)) + return err + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun_test.go new file mode 100644 index 00000000..fa0ca057 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun_test.go @@ -0,0 +1,176 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + meshcatalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer" +) + +const ( + commitE1 = "1111111111111111111111111111111111111111" + commitE2 = "2222222222222222222222222222222222222222" + commitE3 = "3333333333333333333333333333333333333333" +) + +// runStand is a composition run over stand-ins of the mesh-catalog seat, the machines' engines, the build seat and its step-ended. +type runStand struct { + mu sync.Mutex + asks map[string]buildstep.StepAsk // by node + requests map[string]composer.Request // by node + entries []string // the fingerprints `entry` was asked + r *compositionRun +} + +func newRunStand(t *testing.T, revisionEngine, trunkEngine, seatDigest string) *runStand { + s := &runStand{asks: map[string]buildstep.StepAsk{}, requests: map[string]composer.Request{}} + engine := func(fp, commit string) meshcatalog.CatalogEntry { + return meshcatalog.CatalogEntry{Module: "engine", Fp: fp, Repository: "novox/mesh-host", Commit: commit, Path: ".", Serves: []string{ne.Seat + ".composition"}, + Manifest: map[string]any{"module": "engine"}} + } + revisionCommit := commitE1 + if revisionEngine != "src2:e1" { + revisionCommit = commitE3 + } + pages := map[string]meshcatalog.RevisionAnswer{ + "": {Revision: &meshcatalog.CatalogRevision{Revision: 7, Branch: "main", Definitions: []meshcatalog.CatalogRevisionDefinitionsItem{{Seat: "s1", Digest: "d1"}}}, + Entries: []meshcatalog.CatalogEntry{engine(revisionEngine, revisionCommit), {Module: "bad", Verdict: mcc.VerdictRefused, Inputs: &meshcatalog.EntryInputs{Problems: []string{"no seat x"}}, + Manifest: map[string]any{"module": "bad"}}}, Next: "bad"}, + "bad": {Revision: &meshcatalog.CatalogRevision{Revision: 7}, Entries: []meshcatalog.CatalogEntry{{Module: "m", Verdict: mcc.VerdictAccepted, Manifest: map[string]any{"module": "m"}}}}, + } + call := func(key string, body any) (json.RawMessage, error) { + raw, _ := json.Marshal(body) + switch key { + case mcc.RevisionVerb: + var in meshcatalog.RevisionInput + _ = json.Unmarshal(raw, &in) + if in.Revision != 7 || in.Branch != "main" { + return nil, errors.New("asked another revision") + } + return json.Marshal(pages[in.After]) + case seatDefinitionsKey: + return json.Marshal(meshcatalog.SeatDefinitionsAnswer{Found: true, Entry: &meshcatalog.SeatDefinitionsAnswerEntry{Seat: "s1", Digest: seatDigest, Definition: map[string]any{"seat": "s1"}}}) + case entryVerbKey: + var in meshcatalog.EntryInput + _ = json.Unmarshal(raw, &in) + s.mu.Lock() + s.entries = append(s.entries, in.Fingerprint) + s.mu.Unlock() + commit := map[string]string{"src2:e1": commitE1, "src2:e2": commitE2}[in.Fingerprint] + return json.Marshal(meshcatalog.EntryAnswer{Entry: engine(in.Fingerprint, commit)}) + } + return nil, errors.New("no such verb " + key) + } + engines := func(node, _ string, _ any) (json.RawMessage, error) { + return json.RawMessage(`{"candidate":"main.7","node":"` + node + `","verdict":"composed","input":{"node":"` + node + `"},"modules":[],"own":{},"took":1,"v":1}`), nil + } + running := map[string]string{"ace": "src2:e1", "g14": "src2:e2"} + away := &awayNodes{} + away.set("shanks", true) + keeper := testKeeper(&objectsMem{objs: map[string][]byte{}}, time.Now) + ends := &runEnds{} + s.r = &compositionRun{ + inputs: &compositionInputs{machines: func(string, string) []string { return []string{"ace", "g14", "shanks"} }, call: engines, bound: time.Second}, + keeper: keeper, call: call, ends: ends, away: away, now: time.Now, logf: t.Logf, + trunk: func() (mcc.Half, bool) { + return mcc.Half{Entries: map[string]meshcatalog.CatalogEntry{"engine": engine(trunkEngine, commitE1)}}, true + }, + running: func(node, module string) (ne.Running, bool) { + fp, ok := running[node] + return ne.Running{Node: node, Module: module, Fingerprint: fp}, ok && module == "engine" + }, + put: func(_ context.Context, a buildstep.StepAsk) error { + body, err := keeper.body(a.Run.Input.Digest) + if err != nil { + return err + } + req, err := composer.Decode(body) + if err != nil { + return err + } + node := req.Machines[0] + s.mu.Lock() + s.asks[node], s.requests[node] = a, req + s.mu.Unlock() + m := composer.Machine{Composes: true} + if node == "g14" { + m = composer.Machine{Reason: "two modules own one path", Problems: []string{"/etc/x is m's and bad's"}} + } + v, _ := json.Marshal(composer.Verdict{V: composer.Version, Cases: []composer.CaseVerdict{{Name: changeCase, Machines: map[string]composer.Machine{node: m}}}}) + go ends.heard(buildstep.StepEnded{Ask: a.ID, Asker: runAsker, Node: "builder1", Outcome: buildstep.OutcomePassed, Output: string(v)}) + return nil + }, + } + return s +} + +// Design 47 G §354, §63, §64, A-226, A-238, A-240, A-243, A-252, A-254, A-261: one request and one run ask per machine, the request +// depth compose with the one machine and one case `change`, its manifests, verdicts and seats from the step's revision; the +// composer is the node-engine build each machine runs, its entry read through `entry`, or the revision's where it differs from the +// trunk's; the run's verdict is read from its step-ended; a machine away passes with a warning, one not composing fails the step. +func TestTheCompositionStepRunsOneAskPerMachine(t *testing.T) { + step := meshwants.StepRun{Module: "m", Branch: "main", Revision: 7, Step: meshwants.StepRunStepComposition} + s := newRunStand(t, "src2:e1", "src2:e1", "d1") + out := s.r.run(context.Background(), "ask-step", step) + if out.Outcome != meshwants.StepEndedOutcomeFailed || !strings.Contains(out.Said, "g14 does not compose") || !strings.Contains(out.Said, "/etc/x") || + strings.Contains(out.Said, "ace") || len(out.Warnings) != 1 || !strings.Contains(out.Warnings[0], "shanks is away") { + t.Fatalf("the step came to %+v", out) + } + if len(s.asks) != 2 { + t.Fatalf("asked %d runs, want one per machine not away", len(s.asks)) + } + a, req := s.asks["ace"], s.requests["ace"] + if a.Kind != buildstep.AskCheck || a.Language != buildstep.AskGo || a.Asker != runAsker || a.Run == nil || len(a.Run.Heads) != 0 || + a.Run.Input.Verb != "seat:mesh-wants.input" || a.Run.Program != (buildstep.RunProgram{Repository: "novox/mesh-host", Commit: commitE1, Dir: ".", Package: composer.Package}) { + t.Errorf("the run ask of ace %+v %+v", a, a.Run) + } + if s.asks["g14"].Run.Program.Commit != commitE2 { + t.Errorf("g14's composer is built at %s, want the build its node-engine runs", s.asks["g14"].Run.Program.Commit) + } + if req.Depth != composer.DepthCompose || len(req.Machines) != 1 || req.Machines[0] != "ace" || len(req.Cases) != 1 || req.Cases[0].Name != changeCase || + string(req.Input) != `{"node":"ace"}` || len(req.Seats) != 1 || string(req.Seats[0]) != `{"seat":"s1"}` { + t.Errorf("the request of ace %+v", req) + } + c := req.Cases[0] + if string(c.Manifests["m"].Raw) != `{"module":"m"}` || len(c.Manifests) != 3 || len(c.Verdicts) != 1 || !strings.Contains(string(c.Verdicts["bad"]), `"refused"`) || + !strings.Contains(string(c.Verdicts["bad"]), "no seat x") { + t.Errorf("the case %+v %s", c.Manifests, c.Verdicts) + } + if strings.Join(s.entries, " ") == "" { + t.Error("no node-engine entry is read through entry") + } + + // The revision's node-engine differs from the trunk's: that build composes every machine, and no entry is asked. + changed := newRunStand(t, "src2:e3", "src2:e1", "d1") + if out := changed.r.run(context.Background(), "ask-step", step); out.Outcome != meshwants.StepEndedOutcomeFailed { + t.Fatalf("the step came to %+v", out) + } + if changed.asks["ace"].Run.Program.Commit != commitE3 || changed.asks["g14"].Run.Program.Commit != commitE3 || len(changed.entries) != 0 { + t.Errorf("the change's node-engine does not compose every machine: %+v %+v %v", changed.asks["ace"].Run.Program, changed.asks["g14"].Run.Program, changed.entries) + } + + // A seat digest neither the revision nor the trunk holds: could not run, naming the seat and the digest (A-261). + other := newRunStand(t, "src2:e1", "src2:e1", "d2") + if out := other.r.run(context.Background(), "ask-step", step); out.Outcome != meshwants.StepEndedOutcomeCouldNotRun || + !strings.Contains(out.Said, "s1") || !strings.Contains(out.Said, "d1") || len(other.asks) != 0 { + t.Errorf("a seat digest not held came to %+v, %d asks", out, len(other.asks)) + } + + // At the inputs state's cap the ask is refused, naming the store (§354). + full := newRunStand(t, "src2:e1", "src2:e1", "d1") + full.r.keeper.used = func() (int64, error) { return inputsCap, nil } + if out := full.r.run(context.Background(), "ask-step", step); out.Outcome != meshwants.StepEndedOutcomeCouldNotRun || + !strings.Contains(out.Said, "the state inputs is at its cap") || len(full.asks) != 0 { + t.Errorf("an ask at the cap came to %+v, %d asks", out, len(full.asks)) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go index 79c6a2be..b548d835 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go @@ -8,6 +8,7 @@ import ( "github.com/nats-io/nats.go" meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/process" ) // The composition step (design 47 G §22, §354, §63): the assigner takes the seat mesh-wants' accepted kind `step`, the @@ -66,7 +67,7 @@ func stepAskID(m *nats.Msg) string { } // serveCompositionSteps hears the composition step asks on the seat mesh-wants' accept subject. -func serveCompositionSteps(ctx context.Context, conn *nats.Conn, c *compositionSteps) error { +func serveCompositionSteps(ctx context.Context, conn process.Subscriber, c *compositionSteps) error { sub, err := conn.Subscribe(meshwants.AcceptSubjectStep, func(m *nats.Msg) { c.take(ctx, stepAskID(m), m.Data) }) if err != nil { return err diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go index faccb623..211bb25e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go @@ -17,7 +17,7 @@ import ( "github.com/nats-io/nats.go/jetstream" meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -51,11 +51,16 @@ const ( // digestOf is an input's digest as the ask names it: its sha256 in hex. func digestOf(b []byte) string { sum := sha256.Sum256(b); return hex.EncodeToString(sum[:]) } +// inputsCap is the state inputs' cap (design 47 G §354's kept-data row: 64 MiB, the manifest's max-mib): at it, an ask is refused. +const inputsCap = 64 << 20 + // inputKeeper keeps the inputs over the state and its object store. type inputKeeper struct { vals stateobjects.Values now func() time.Time logf func(string, ...any) + // used is how many bytes the state and its object store hold; nil judges no cap. + used func() (int64, error) } // keep keeps the exact bytes of the input of the ask under their write time and sha256, after sweeping the keys more than a day old @@ -67,6 +72,16 @@ func (k *inputKeeper) keep(ask string, body []byte) (string, error) { if err := k.sweep(); err != nil { k.logf("inputs: the keys older than a day are not all deleted: %v", err) } + if k.used != nil { + used, err := k.used() + if err != nil { + return "", fmt.Errorf("the fill of the state %s is not read, so no input is kept: %w", meshwants.StateInputs, err) + } + if used+int64(len(body)) > inputsCap { + return "", fmt.Errorf("the state %s is at its cap of %d MiB (%d bytes kept), so the ask's input of %d bytes is refused", + meshwants.StateInputs, inputsCap>>20, used, len(body)) + } + } digest := digestOf(body) key := meshwants.InputsByTimeDigestKey(k.now().UTC().Format(inputsTime), digest) if err := k.vals.Put(key, base64.StdEncoding.EncodeToString(body)); err != nil && !errors.Is(err, stateobjects.ErrChanged) { @@ -276,8 +291,26 @@ func inputsOver(kv jetstream.KeyValue, js jetstream.JetStream, module string, lo if kv == nil { return nil } - v := stateobjects.Values{KV: stateobjects.BusKV{KV: kv}, Objects: &inputObjects{js: js, module: module}, Store: catalogue.StateObjectsName(meshwants.StateInputs)} - return &inputKeeper{vals: v, now: time.Now, logf: logf} + objects := &inputObjects{js: js, module: module} + v := stateobjects.Values{KV: stateobjects.BusKV{KV: kv}, Objects: objects, Store: catalogue.StateObjectsName(meshwants.StateInputs)} + used := func() (int64, error) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + st, err := kv.Status(ctx) + if err != nil { + return 0, err + } + n := int64(st.Bytes()) + objs, err := objects.List() + if err != nil && !errors.Is(err, process.ErrNotYet) { + return 0, err + } + for _, o := range objs { + n += o.Size + } + return n, nil + } + return &inputKeeper{vals: v, now: time.Now, logf: logf, used: used} } // inputSaid is one kept input as assigner_inputs lists it. @@ -323,10 +356,10 @@ func inputsRead(k *inputKeeper) stdio.Tool { }} } -// heardStepEnded deletes the input kept for the ask a build seat's step-ended names (design 47 G §354); one that cannot be deleted is -// given again by the bus. -func heardStepEnded(k *inputKeeper, m jetstream.Msg, logf func(string, ...any)) { - var e nodebuildgo.StepEnded +// heardStepEnded deletes the input kept for the ask a build seat's step-ended names (design 47 G §354), and hands the end to the +// composition run that waits for that ask (A-240); one whose input cannot be deleted is given again by the bus. +func heardStepEnded(k *inputKeeper, ends *runEnds, m jetstream.Msg, logf func(string, ...any)) { + var e buildstep.StepEnded if err := json.Unmarshal(m.Data(), &e); err != nil { logf("%s holds what is no step-ended, left: %v", m.Subject(), err) _ = m.Ack() @@ -337,5 +370,8 @@ func heardStepEnded(k *inputKeeper, m jetstream.Msg, logf func(string, ...any)) _ = m.Nak() return } + if ends != nil { + ends.heard(e) + } _ = m.Ack() } 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 a1751712..d2599830 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -30,6 +30,7 @@ import ( mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/announce" @@ -491,12 +492,20 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if bus != nil { gate.Follow(seatlist.ProcessBus(bus.Call, hub), wake, logger.Printf) } - // The build seat's step-ended of an ask deletes the input kept for it (design 47 G §354); registered before the hub starts. + // The build seat's step-ended of an ask deletes the input kept for it (design 47 G §354) and ends the composition run that waits + // for it (A-240); the watchman's machine-away and machine-back say which machines a composition step does not compose (§64, + // §389). Registered before the hub starts. + ends, away := &runEnds{}, &awayNodes{} if inputs != nil { - if err := hub.Handle(nodebuildgo.SubjectStepEndedAll, func(m jetstream.Msg) { heardStepEnded(inputs, m, logger.Printf) }); err != nil { + if err := hub.Handle(nodebuildgo.SubjectStepEndedAll, func(m jetstream.Msg) { heardStepEnded(inputs, ends, m, logger.Printf) }); err != nil { return err } } + for subject, isAway := range map[string]bool{meshwatchdogs.SubjectMachineAwayAll: true, meshwatchdogs.SubjectMachineBackAll: false} { + if err := hub.Handle(subject, func(m jetstream.Msg) { heardPresence(away, isAway, m, logger.Printf) }); err != nil { + logger.Printf("%s is not heard, so a composition step asks a machine that sleeps: %v", subject, err) + } + } // The indexer's manifests, modules and module graph, through its first views and views family, never its buckets // (DLV-F2r, ADR 0631): the first pass waits for the three answers, as it waits for every watch to hand what stands. if iv != nil { @@ -628,6 +637,15 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if bus != nil && kvs[buckets.Puts] != nil { p.acts = newPutActs(bus, wake, logger.Printf) } + // The composition step (design 47 G §22, §354): the seat mesh-wants' accepted kind step, run by one run ask per machine. + if bus != nil && js != nil && inputs != nil && p.publishWants != nil { + run := &compositionRun{inputs: &compositionInputs{machines: w.machinesOf, call: bus.CallKeptOn}, keeper: inputs, call: bus.CallKept, + trunk: trunkOf(iv), running: w.runningOf, away: away, put: runAskPublisher(js), ends: ends, ask: bus.Call, now: time.Now, logf: logger.Printf} + steps := &compositionSteps{node: thisNode, say: p.publishWants, run: run.run, now: time.Now, logf: logger.Printf} + if err := serveCompositionSteps(ctx, nc, steps); err != nil { + logger.Printf("the composition step is not heard, so no composition step runs until the assigner starts again: %v", err) + } + } if bus != nil { p.seat = newSeatActs(bus, wake, logger.Printf) } diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 2611d10b..3ac6f5d0 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -130,14 +130,18 @@ 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-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-build-go.step-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) { + if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-build-go.step-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated", + "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back"}) { // design 47 G §64, §389: a composition step passes a machine away t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) } if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "seat:mesh-catalog.entry", // design 47 G A-243, A-252: the node-engine build's entry a composition run builds "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list "seat:mesh-conditions.conditions", // design 47 G §706: the conditions read at each judging of an outside wait "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone "seat:node-backup.backed-up", "seat:node-backup.now", + "seat:node-build-go.output", // A-254: a composition run's answer kept as the build seat's object + "seat:node-engine.compose-candidate", // §354, A-226: each machine's composition input "seat:node-engine.manifest-reads", // ADR 0618 §3 "seat:operator-channel.asks"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0668 §4: the ended asks at bind t.Errorf("invokes = %v, want %v", m.Invokes, want) diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 046f1736..3111f4a2 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -63,11 +63,14 @@ "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", - "mesh-delivery.pins-stated" + "mesh-delivery.pins-stated", + "mesh-watchdogs.machine-away", + "mesh-watchdogs.machine-back" ], "invokes": [ "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "seat:mesh-catalog.entry", "seat:mesh-catalog.registrations", {"verb": "seat:mesh-catalog.revision", "fields": ["none-yet"]}, "seat:mesh-catalog.seat-definitions", @@ -77,6 +80,8 @@ "seat:mesh-hand-acts.warranted", "seat:node-backup.backed-up", "seat:node-backup.now", + "seat:node-build-go.output", + "seat:node-engine.compose-candidate", "seat:node-engine.manifest-reads", "seat:operator-channel.asks" ],