diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go new file mode 100644 index 00000000..5f611a66 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go @@ -0,0 +1,165 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strconv" + "sync" + "time" + + nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// 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. + +// composeCandidateKey is the seat verb's call key, asked of one node. +const composeCandidateKey = "seat:" + ne.Seat + "." + ne.VerbComposeCandidate + +// candidateKeyOf is the key the step's pipeline names its candidate by (A-226): a run on request's `run-`, or for a branch +// pipeline the branch's catalog graph revision `.`, the branch escaped as a key part as the pipeline key does. +func candidateKeyOf(run meshwants.StepRun) (string, error) { + switch { + case run.Run != "": + return "run-" + run.Run, nil + case run.Branch == "" || run.Revision <= 0: + return "", errors.New("the step names neither a run on request nor a branch's revision, so no candidate is named") + } + return keys.EscapeName(run.Branch) + "." + strconv.FormatInt(run.Revision, 10), nil +} + +// machinesFor are the machines of a step (A-252): the nodes whose placement of the module names the step's branch (a placement +// naming none is the trunk's); a branch that is not watched takes the nodes placing the module on the trunk. In name order. +func machinesFor(placed map[string]map[string]string, module, branch string, watched bool) []string { + if branch == "" || !watched { + branch = trunkBranch + } + var out []string + for node, b := range placed[module] { + if b == "" { + b = trunkBranch + } + if b == branch { + out = append(out, node) + } + } + slices.Sort(out) + return out +} + +// machinesOf are the machines of a composition step in the world's assignments. +func (w *world) machinesOf(module, branch string) []string { + return machinesFor(w.placedBranches(), module, branch, w.holdsBranch(branch)) +} + +// nodeInput is one machine's input: its bytes, or why the node gave none, or that it is away. +type nodeInput struct { + 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. +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) + bound time.Duration +} + +// 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 + } + nodes := c.machines(run.Module, run.Branch) + if len(nodes) == 0 { + return nil, fmt.Errorf("no node places %s on the branch %q, so the composition step has no machine to compose", run.Module, run.Branch) + } + bound := c.bound + if bound <= 0 { + bound = nec.AnswerBound + } + 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) + }() + } + wg.Wait() + return out, nil +} + +// 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() + type reply struct { + raw json.RawMessage + err error + } + got := make(chan reply, 1) + go func() { + raw, err := c.call(node, composeCandidateKey, ne.ComposeCandidateInput{Candidate: key, Input: true}) + got <- reply{raw, err} + }() + var raw json.RawMessage + select { + case <-at.Done(): + in.Why = "no answer within its bound: " + at.Err().Error() + return in + case r := <-got: + if r.err != nil { + in.Why = r.err.Error() + return in + } + raw = r.raw + } + var a ne.ComposeCandidateAnswer + // its numbers as written (json.Number), so the input kept is the engine's own bytes' values + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + if err := dec.Decode(&a); err != nil { + in.Why = "its answer is not a compose-candidate answer: " + err.Error() + return in + } + switch { + case a.Verdict == nec.Unready: + in.Why = "unready: " + a.Refused + return in + case len(a.Input) == 0: + in.Why = "the engine answered no composition input" + if a.Refused != "" { + in.Why += ": " + a.Refused + } + return in + } + body, err := json.Marshal(a.Input) + if err != nil { + in.Why = "its composition input cannot be encoded again: " + err.Error() + return in + } + 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 new file mode 100644 index 00000000..9b0f2bd9 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs_test.go @@ -0,0 +1,91 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" +) + +// Design 47 G A-226: the candidate a step asks is `run-` for a run on request, else `.`. +func TestTheCandidateKeyOfAStep(t *testing.T) { + for _, c := range []struct { + run meshwants.StepRun + want string + fail bool + }{ + {run: meshwants.StepRun{Run: "r1", Branch: "main"}, want: "run-r1"}, + {run: meshwants.StepRun{Branch: "feat/x", Revision: 42}, want: "feat/x.42"}, + {run: meshwants.StepRun{Branch: "main"}, fail: true}, + } { + got, err := candidateKeyOf(c.run) + if (err != nil) != c.fail || got != c.want { + t.Errorf("%+v: %q, %v; want %q", c.run, got, err, c.want) + } + } +} + +// A-252: the machines of a step are the nodes whose placement names its branch; none named is the trunk; an unwatched branch takes +// the trunk's nodes. +func TestTheMachinesOfAStep(t *testing.T) { + placed := map[string]map[string]string{"m": {"ace": "", "g14": "feat", "novox": "main", "shanks": "feat"}, "other": {"ace": ""}} + for _, c := range []struct { + branch string + watched bool + want string + }{ + {"main", true, "ace novox"}, + {"feat", true, "g14 shanks"}, + {"gone", false, "ace novox"}, + {"", true, "ace novox"}, + } { + if got := strings.Join(machinesFor(placed, "m", c.branch, c.watched), " "); got != c.want { + t.Errorf("branch %q watched %v: %q, want %q", c.branch, c.watched, got, c.want) + } + } + if got := machinesFor(placed, "none", "main", true); len(got) != 0 { + t.Errorf("a module placed nowhere has machines %v", got) + } +} + +// §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) + asked[node] = key + " " + string(raw) + switch node { + case "g14": + return nil, errors.New("no responders") + case "novox": + return json.RawMessage(`{"candidate":"run-r1","node":"novox","verdict":"unready","refused":"starting","modules":[],"own":{},"took":1,"v":1}`), nil + } + return json.RawMessage(`{"candidate":"run-r1","node":"` + node + `","verdict":"composed","input":{"n":12345678901234567890,"node":"` + node + `"},"modules":[],"own":{},"took":1,"v":1}`), nil + } + 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 { + t.Errorf("asked %q, want %q", asked["ace"], want) + } + if got[0].Node != "ace" || got[0].Why != "" || !strings.Contains(string(got[0].Body), "12345678901234567890") { + t.Errorf("ace: %+v", got[0]) + } + 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 _, was := asked["shanks"]; was || !got[3].Away { + t.Errorf("the machine away is asked or not said away: %+v", got[3]) + } + 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..0aa88e22 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go @@ -0,0 +1,505 @@ +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()) + } + setPhase(ctx, meshwants.StepProgressEventPhaseRun) + 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() + setPhase(ctx, meshwants.StepProgressEventPhaseReport) + 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 away { + logf("%s is away (machine-away): a composition step does not compose it until its machine-back", e.Machine) + } else { + logf("%s is back (machine-back): a composition step composes it again", e.Machine) + } + } + 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 new file mode 100644 index 00000000..70949222 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go @@ -0,0 +1,144 @@ +package main + +import ( + "context" + "encoding/json" + "sync" + "time" + + "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 +// composition step of one module's pipeline that the runner asks, runs it, and says its end as the seat's event +// step-ended., the ask's id as the event's id and the event's `id`, which the runner holds the end to. + +// stepOutcome is what one composition step came to. +type stepOutcome struct { + Outcome string + Said string + Warnings []string +} + +// compositionSteps takes the composition steps and says their ends. +type compositionSteps struct { + node string + // say says one event of the seat mesh-wants (wantsPublisher). + say publisher + // run runs one step and answers what it came to. + run func(ctx context.Context, id string, run meshwants.StepRun) stepOutcome + now func() time.Time + logf func(string, ...any) + // asker is the name the step's progress names as its asker. + asker string + // ticks is a ticker of every d, its channel and its stop (time.NewTicker); nil is the clock's. A test hands its own. + ticks func(d time.Duration) (<-chan time.Time, func()) +} + +// stepAsker is the asker a composition step's progress names: the pipeline runner, the one asker that sends the accepted kind +// step on mesh-wants (design 47 G A-267), whose accept subject names no asker. +const stepAsker = "runner" + +// progressEvery is how often a running composition step says its progress (design 47 G §496, A-262). +const progressEvery = time.Minute + +// stepPhase is the phase a running step is in: fetch, build, run or report (§496), set by the run through its context. +type stepPhase struct { + mu sync.Mutex + phase string +} + +type phaseKey struct{} + +// setPhase sets the phase of the step ctx runs, when it runs one. +func setPhase(ctx context.Context, phase string) { + if p, ok := ctx.Value(phaseKey{}).(*stepPhase); ok { + p.mu.Lock() + p.phase = phase + p.mu.Unlock() + } +} + +func (p *stepPhase) get() string { + p.mu.Lock() + defer p.mu.Unlock() + return p.phase +} + +// take takes one ask of the accepted kind `step` by its id and body: a step that is no composition step, or an ask with no +// id, is left with a log line; one taken is run in its own goroutine and its end said. +func (c *compositionSteps) take(ctx context.Context, id string, data []byte) bool { + var run meshwants.StepRun + if err := json.Unmarshal(data, &run); err != nil || id == "" || run.Step != meshwants.StepRunStepComposition { + c.logf("a step ask that is no composition step with an id is left: %v", err) + return false + } + go c.runOne(ctx, id, run) + return true +} + +// runOne runs one step and says its end. +func (c *compositionSteps) runOne(ctx context.Context, id string, run meshwants.StepRun) { + started := c.now().UTC() + phase := &stepPhase{phase: meshwants.StepProgressEventPhaseFetch} + stop := c.sayProgress(id, phase) + o := c.run(context.WithValue(ctx, phaseKey{}, phase), id, run) + stop() + ended := c.now().UTC() + end := meshwants.StepEnded{V: 1, ID: id, Pipeline: run.Pipeline, Step: meshwants.StepEndedStepComposition, Module: run.Module, + Commit: run.Commit, Fingerprint: run.Fingerprint, Node: c.node, Outcome: o.Outcome, Said: o.Said, Warnings: o.Warnings, + Rerun: run.Rerun, By: run.By, Started: &started, Ended: &ended} + if err := c.say(meshwants.EventStepEnded+"."+c.node, end, id); err != nil { + c.logf("saying the end of the composition step %s: %v", id, err) + } +} + +// sayProgress says the step's progress on the seat's step-progress. once a minute until the returned stop is called (design +// 47 G §496, A-262): a ticker inside the running step, as the build seats' step runner says its own beat. +func (c *compositionSteps) sayProgress(id string, phase *stepPhase) (stop func()) { + ticks := c.ticks + if ticks == nil { + ticks = func(d time.Duration) (<-chan time.Time, func()) { t := time.NewTicker(d); return t.C, t.Stop } + } + tick, stopTicks := ticks(progressEvery) + done := make(chan struct{}) + ended := make(chan struct{}) + go func() { + defer close(ended) + defer stopTicks() + for { + select { + case <-done: + return + case <-tick: + at := c.now().UTC() + p := meshwants.StepProgressEvent{Ask: id, Asker: c.asker, Phase: phase.get(), At: at} + if err := c.say(meshwants.EventStepProgress+"."+c.node, p, id+".progress."+at.Format(time.RFC3339Nano)); err != nil { + c.logf("saying the progress of the composition step %s: %v", id, err) + } + } + } + }() + return func() { close(done); <-ended } +} + +// stepAskID is an ask's id as the asker names it: its event id, else its message id. +func stepAskID(m *nats.Msg) string { + if id := m.Header.Get("x-event-id"); id != "" { + return id + } + return m.Header.Get(nats.MsgIdHdr) +} + +// serveCompositionSteps hears the composition step asks on the seat mesh-wants' accept subject. +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 + } + go func() { <-ctx.Done(); _ = sub.Drain() }() + return nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_integration_test.go new file mode 100644 index 00000000..a1f4cdf2 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_integration_test.go @@ -0,0 +1,235 @@ +//go:build integration + +package main + +import ( + "bytes" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" + "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-contracts/go/seats/nodebuildgo" + "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" + + "mesh-delivery-assigner/internal/buckets" +) + +// Design 47 G §354, §22, §63, §64, §389, A-240, A-252, A-254: the assigner started whole takes the runner's composition step of a +// module placed on three machines, one of them away: it asks the two others' node-engines for their inputs, puts one run ask per +// machine on the Go build seat's check queue under its name, its program the node-engine build each machine runs; the stand-in +// build seat reads each request by the seat mesh-wants' verb input in chunks and says its step-ended with the composer's verdict; +// the step's step-ended fails naming the machine that does not compose, warns of the machine away, and no input stays kept. +func TestTheCompositionStepComposesEachMachineThroughTheBuildSeat(t *testing.T) { + b := startBed(t, "") + nc, err := nats.Connect(b.srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + const engineCommit = "1111111111111111111111111111111111111111" + // What each node runs is the node-engine seat's own stream, kept last per subject (§199): the bed's EVENTS takes every seat's + // events, so here it takes those of the seats this case hears and says, the node-engine's reads changed among them, and not + // what each node runs. + if err := b.js.DeleteStream(b.ctx, "EVENTS"); err != nil { + t.Fatal(err) + } + var seatEvents []string + for _, seat := range []string{"mesh-catalog", "mesh-wants", "mesh-module-assigner", "node-build-go", "mesh-watchdogs", "mesh-delivery", "mesh-pipeline-runner", "mesh-conditions", "operator-channel"} { + seatEvents = append(seatEvents, "mesh.seat."+seat+".event.>") + } + if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: append([]string{"mesh.mod.*.event.>", enginereads.SubjectPrefix + "*"}, seatEvents...)}); err != nil { + t.Fatal(err) + } + // The assigner's durable on EVENTS hears, beside the bed's, the build seat's step-ended and the watchman's machine-away and + // machine-back, as the grants part derives them from its consumes; and what each node runs, kept last per subject. + if _, err := b.js.CreateOrUpdateConsumer(b.ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", + FilterSubjects: []string{enginereads.SubjectPrefix + "*", "mesh.mod." + buckets.IndexerModule + ".event.views.>", meshcatalog.SubjectCatalogRevisionAll, + nodebuildgo.SubjectStepEndedAll, meshwatchdogs.SubjectMachineAwayAll, meshwatchdogs.SubjectMachineBackAll}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: nodeengine.StreamRunning, Subjects: []string{nodeengine.SubjectRunningAll}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + if _, err := b.js.CreateOrUpdateConsumer(b.ctx, nodeengine.StreamRunning, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + for _, node := range []string{"n1", "n2"} { + raw, _ := json.Marshal(nodeengine.Running{V: 1, Node: node, Module: "engine", Fingerprint: "src2:e1", At: time.Now().UTC()}) + if _, err := b.js.Publish(b.ctx, nodeengine.SubjectRunning(node, "engine"), raw); err != nil { + t.Fatal(err) + } + } + away, _ := json.Marshal(meshwatchdogs.MachineAwayEvent{V: 1, Machine: "n3", At: time.Now().UTC(), LastHeard: time.Now().UTC()}) + if _, err := b.js.Publish(b.ctx, meshwatchdogs.SubjectMachineAway("n3"), away); err != nil { + t.Fatal(err) + } + // The trunk's catalog half holds the module and the node-engine, the module claiming the seat node-engine. + engine := meshcatalog.CatalogEntry{Module: "engine", Fp: "src2:e1", Repository: "novox/mesh-host", Commit: engineCommit, Path: ".", + Serves: []string{nodeengine.Seat + ".composition"}, Manifest: map[string]any{"module": "engine", "version": "0.1"}} + b.cat.mu.Lock() + b.cat.entries[trunkBranch] = map[string]meshcatalog.CatalogEntry{"engine": engine, + "m": {Module: "m", Fp: "src2:m1", Repository: "novox/m", Commit: engineCommit, Path: ".", Manifest: map[string]any{"module": "m", "version": "0.1"}}} + b.cat.rev[trunkBranch] = 3 + b.cat.mu.Unlock() + for _, node := range []string{"n1", "n2", "n3"} { + b.place(t, node, "m") + } + // The node-engines of n1 and n2 answer compose-candidate with their inputs; n3 is never asked. + asked := make(chan string, 8) + engines, err := nc.Subscribe("mesh.seat.node-engine.call.compose-candidate.*.*", func(m *nats.Msg) { + node := strings.Split(m.Subject, ".")[5] + asked <- node + raw, _ := json.Marshal(map[string]any{"result": map[string]any{"candidate": "main.3", "node": node, "verdict": "composed", "input": map[string]any{"node": node}, + "modules": []any{}, "own": map[string]any{}, "took": 1, "v": 1}, "node": node}) + _ = m.Respond(raw) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = engines.Unsubscribe() }) + // mesh-catalog's entry answers the node-engine build's entry by its fingerprint. + entries, err := nc.Subscribe("mesh.seat.mesh-catalog.call.entry.*", func(m *nats.Msg) { + raw, _ := json.Marshal(map[string]any{"result": meshcatalog.EntryAnswer{Entry: engine}, "node": "n0"}) + _ = m.Respond(raw) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = entries.Unsubscribe() }) + subscribed(t, nc) + // The runner hears the step's end. + ended := make(chan meshwants.StepEnded, 1) + ends, err := nc.Subscribe("mesh.seat.mesh-wants.event.step-ended.n1", func(m *nats.Msg) { + var e meshwants.StepEnded + _ = json.Unmarshal(m.Data, &e) + ended <- e + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ends.Unsubscribe() }) + subscribed(t, nc) + + from := b.journal.mark() + b.start(t) + b.journal.await(t, from, "the composition step is heard on", "the composition step served", 30*time.Second) + b.journal.await(t, from, "n3 is away (machine-away)", "the machine-away of n3 heard", 30*time.Second) + + step, _ := json.Marshal(meshwants.StepRun{Pipeline: "main.3.m", Step: meshwants.StepRunStepComposition, Module: "m", Branch: trunkBranch, + Revision: 3, Commit: engineCommit, Fingerprint: "src2:m1"}) + msg := nats.NewMsg(meshwants.AcceptSubjectStep) + msg.Header.Set("x-event-id", "step-ask-1") + msg.Data = step + if err := nc.PublishMsg(msg); err != nil { + t.Fatal(err) + } + + // The stand-in Go build seat takes the run asks put under the assigner's name, reads each request by the verb input and says + // its end with the composer's verdict: n1 composes, n2 does not. + queue, _ := buildstep.AskStream(buildstep.AskGo) + subject, _ := buildstep.AskSubjectBy(buildstep.AskGo, buildstep.AskCheck, runAsker) + worker, err := b.js.CreateOrUpdateConsumer(b.ctx, queue, jetstream.ConsumerConfig{Durable: "stand-in", FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy}) + if err != nil { + t.Fatal(err) + } + programs := map[string]string{} + for range 2 { + got, err := worker.Next(jetstream.FetchMaxWait(30 * time.Second)) + if err != nil { + t.Fatalf("no run ask is put: %v", err) + } + _ = got.Ack() + a, err := buildstep.DecodeStepAsk(got.Data()) + if err != nil || a.Run == nil { + t.Fatalf("the run ask %s does not read: %v", got.Data(), err) + } + req, err := composer.Decode(readInput(t, nc, a.Run.Input)) + if err != nil || len(req.Machines) != 1 { + t.Fatalf("the request of %s: %+v %v", a.ID, req, err) + } + node := req.Machines[0] + programs[node] = a.Run.Program.Commit + m := composer.Machine{Composes: true} + if node == "n2" { + m = composer.Machine{Reason: "two modules own one path", Problems: []string{"/etc/x"}} + } + v, _ := json.Marshal(composer.Verdict{V: composer.Version, Cases: []composer.CaseVerdict{{Name: changeCase, Machines: map[string]composer.Machine{node: m}}}}) + end, _ := json.Marshal(buildstep.StepEnded{V: 1, ID: a.ID, Ask: a.ID, Asker: runAsker, Node: "b1", Outcome: buildstep.OutcomePassed, Output: string(v), At: time.Now().UTC()}) + if _, err := b.js.Publish(b.ctx, nodebuildgo.SubjectStepEnded("b1"), end); err != nil { + t.Fatal(err) + } + } + var e meshwants.StepEnded + select { + case e = <-ended: + case <-b.ctx.Done(): + t.Fatal("the composition step said no end") + } + if e.ID != "step-ask-1" || e.Outcome != meshwants.StepEndedOutcomeFailed || !strings.Contains(e.Said, "n2 does not compose") || + strings.Contains(e.Said, "n1") || len(e.Warnings) != 1 || !strings.Contains(e.Warnings[0], "n3 is away") { + t.Fatalf("the step ended %+v", e) + } + if programs["n1"] != engineCommit || programs["n2"] != engineCommit { + t.Errorf("the composers were built at %v, want the node-engine build each machine runs", programs) + } + close(asked) + for node := range asked { + if node == "n3" { + t.Error("the machine away was asked for its input") + } + } + keys, _ := b.buckets[buckets.Inputs].Keys(b.ctx) + if len(keys) != 0 { + t.Errorf("the inputs stay kept past their step-ended: %v", keys) + } +} + +// readInput reads a run's input as a build seat does: the seat mesh-wants' verb input answers its digest and chunk count, then its +// chunks on the call's subjects, each acknowledged. +func readInput(t *testing.T, nc *nats.Conn, in buildstep.RunInput) []byte { + t.Helper() + call := "standin" + strings.Repeat("0", 16) + chunks := buscalls.SeatChunks{Seat: meshwants.Seat, Verb: meshwants.VerbInput, Call: call} + sub, err := nc.SubscribeSync(chunks.Chunks()) + if err != nil { + t.Fatal(err) + } + defer sub.Unsubscribe() + body, _ := json.Marshal(meshwants.InputInput{Digest: in.Digest, Call: call}) + reply, err := nc.Request("mesh.seat.mesh-wants.tool.input", body, 10*time.Second) + if err != nil { + t.Fatal(err) + } + raw, said, ok := buscalls.Unwrap(reply.Data) + var a meshwants.InputAnswer + if !ok || json.Unmarshal(raw, &a) != nil || a.Chunks < 1 { + t.Fatalf("the verb input answered %s: %s", reply.Data, said) + } + var out bytes.Buffer + next := func() (string, []byte, error) { + m, err := sub.NextMsg(buscalls.SeatAckWithin) + if err != nil { + return "", nil, err + } + return m.Subject, m.Data, nil + } + ack := func(subject string) error { _, err := nc.Request(subject, nil, 10*time.Second); return err } + if _, err := chunks.Receive(a.Chunks, next, ack, &out); err != nil { + t.Fatal(err) + } + if err := in.Verify(out.Bytes()); err != nil { + t.Fatal(err) + } + return out.Bytes() +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go new file mode 100644 index 00000000..e5c7f566 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go @@ -0,0 +1,120 @@ +package main + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" +) + +// saidEvents keeps what a publisher says. +type saidEvents struct { + mu sync.Mutex + said map[string]any + ids map[string]string + heard chan string +} + +func newSaidEvents() *saidEvents { + return &saidEvents{said: map[string]any{}, ids: map[string]string{}, heard: make(chan string, 8)} +} + +func (s *saidEvents) say(event string, body any, id string) error { + s.mu.Lock() + s.said[event], s.ids[event] = body, id + s.mu.Unlock() + s.heard <- event + return nil +} + +// AS5 (§22, §354): the assigner takes mesh-wants' accepted kind `step` for the composition step and says the seat's +// step-ended. with the ask's id, what the step was asked at and what it came to; an ask of another step is left. +func TestTheCompositionStepIsTakenAndItsEndSaid(t *testing.T) { + said := newSaidEvents() + at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + var asked meshwants.StepRun + c := &compositionSteps{node: "n1", say: said.say, now: func() time.Time { return at }, logf: t.Logf, + run: func(_ context.Context, id string, run meshwants.StepRun) stepOutcome { + asked = run + return stepOutcome{Outcome: meshwants.StepEndedOutcomeFailed, Said: "a and b own /etc/x", Warnings: []string{"n3 asleep"}} + }} + other, _ := json.Marshal(map[string]any{"pipeline": "main.4.m", "step": "build", "module": "m"}) + if c.take(context.Background(), "ask-0", other) { + t.Fatal("a step ask of another step is taken") + } + body, _ := json.Marshal(meshwants.StepRun{Pipeline: "main.4.m", Step: meshwants.StepRunStepComposition, Module: "m", Branch: "main", + Commit: "c0ffee", Revision: 4, Fingerprint: "fp1", Rerun: true, By: "lead@n1"}) + if c.take(context.Background(), "", body) { + t.Fatal("a step ask with no id is taken") + } + if !c.take(context.Background(), "ask-1", body) { + t.Fatal("the composition step is not taken") + } + select { + case <-said.heard: + case <-time.After(5 * time.Second): + t.Fatal("no step-ended is said") + } + event := meshwants.EventStepEnded + ".n1" + end, ok := said.said[event].(meshwants.StepEnded) + if !ok || said.ids[event] != "ask-1" { + t.Fatalf("said %v under %v, want step-ended.n1 under the ask's id", said.said, said.ids) + } + if asked.Module != "m" || end.ID != "ask-1" || end.Pipeline != "main.4.m" || end.Step != meshwants.StepEndedStepComposition || + end.Module != "m" || end.Commit != "c0ffee" || end.Fingerprint != "fp1" || end.Node != "n1" || + end.Outcome != meshwants.StepEndedOutcomeFailed || end.Said != "a and b own /etc/x" || len(end.Warnings) != 1 || + !end.Rerun || end.By != "lead@n1" || end.Started == nil || end.Ended == nil { + t.Fatalf("step-ended %+v", end) + } +} + +// Design 47 G §496, A-262: a running composition step says step-progress. on every tick of its ticker, with the ask, its +// asker, the phase the run is in and the time, and says none once it has ended. +func TestARunningCompositionStepSaysItsProgress(t *testing.T) { + said := newSaidEvents() + tick := make(chan time.Time) + stopped := make(chan struct{}) + at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + release := make(chan struct{}) + c := &compositionSteps{node: "n1", say: said.say, now: func() time.Time { return at }, logf: t.Logf, asker: "runner", + ticks: func(d time.Duration) (<-chan time.Time, func()) { + if d != time.Minute { + t.Errorf("the progress ticks every %s, want a minute", d) + } + return tick, func() { close(stopped) } + }, + run: func(ctx context.Context, _ string, _ meshwants.StepRun) stepOutcome { + setPhase(ctx, meshwants.StepProgressEventPhaseRun) + <-release + return stepOutcome{Outcome: meshwants.StepEndedOutcomePassed} + }} + body, _ := json.Marshal(meshwants.StepRun{Pipeline: "main.4.m", Step: meshwants.StepRunStepComposition, Module: "m", Branch: "main"}) + if !c.take(context.Background(), "ask-1", body) { + t.Fatal("the composition step is not taken") + } + progress := meshwants.EventStepProgress + ".n1" + for range 2 { + tick <- at + if e := <-said.heard; e != progress { + t.Fatalf("said %s on a tick, want %s", e, progress) + } + } + said.mu.Lock() + p, ok := said.said[progress].(meshwants.StepProgressEvent) + said.mu.Unlock() + if !ok || p.Ask != "ask-1" || p.Asker != "runner" || p.Phase != meshwants.StepProgressEventPhaseRun || !p.At.Equal(at) { + t.Fatalf("step-progress %+v", p) + } + close(release) + if e := <-said.heard; e != meshwants.EventStepEnded+".n1" { + t.Fatalf("said %s at the end, want step-ended", e) + } + select { + case <-stopped: + default: + t.Fatal("the ticker is not stopped when the step has ended") + } +} 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 dd3bb904..211bb25e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go @@ -1,74 +1,316 @@ package main import ( + "context" "crypto/sha256" "encoding/base64" "encoding/hex" "encoding/json" "errors" "fmt" - "sort" + "regexp" + "slices" "strings" + "sync" + "time" + + "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-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" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/stateobjects" ) -// A step's input by its digest (design 47 G §354): the assigner keeps the input of the composition step it asked a build seat to -// run in its state `inputs`, one key `