AS5-2b: the composition step's machines and their inputs, asked by compose-candidate and kept
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-<id>, or <branch>.<revision>) 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
This commit is contained in:
@@ -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-<id>`, or for a branch
|
||||
// pipeline the branch's catalog graph revision `<branch>.<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
|
||||
}
|
||||
@@ -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-<id>` for a run on request, else `<branch>.<revision>`.
|
||||
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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user