From 0930d5754640dd5bed3f4605c7057abf0f07805f Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:33:52 +0100 Subject: [PATCH] AS5-2b: the composition step's machines and their inputs, asked by compose-candidate and kept MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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); each is asked compose-candidate with the step's candidate key (run-, or .) and input true, and its input is kept under the ask named for it. Not yet called by a run (AS5-3). Builds: design 47 G §354, §22, A-226, A-243, A-252 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../compositioninputs.go | 166 ++++++++++++++++++ .../compositioninputs_test.go | 98 +++++++++++ 2 files changed, 264 insertions(+) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs_test.go 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..c793d73b --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs.go @@ -0,0 +1,166 @@ +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 and kept by the digest the run ask names it by. + +// 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 digest and bytes, or why the node gave none. +type nodeInput struct { + Node string + Digest string + Body []byte + // Why is set when the node gave no input: its words, or the call's error. + Why string +} + +// compositionInputs asks the machines of a step for their inputs and keeps them. +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 +} + +// 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) { + 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 { + wg.Add(1) + go func() { + defer wg.Done() + out[i] = c.one(ctx, bound, node, key, askOf(node)) + }() + } + 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 { + 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 + } + 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 + 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..55c7f67d --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositioninputs_test.go @@ -0,0 +1,98 @@ +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: 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) { + 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 + } + 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 { + 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 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 != "" { + t.Errorf("g14 and novox: %+v %+v", got[1], got[2]) + } + if err := keeper.ended("ask-ace"); err != nil { + t.Fatal(err) + } + 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} + if _, err := none.gather(context.Background(), meshwants.StepRun{Module: "m", Run: "r1"}, nil); err == nil { + t.Error("a step with no machine is gathered") + } +}