Send over mesh-sdk's composition: push, the pushes of plans and rollouts, held back, plan and the merge gate's passes (issue 1386)

Every send composes through composeSend (compose over the Input, perform what it named foreseen, read the stores again, compose again, refuse a declaration that still stands in for a value): push.go's two composing sites, composeForPush, declarationFor (the plan verb, which allocates on purpose for one named machine) and the merge gate's two passes and its trial assignment (composeSendValidated, which answers the node-engine's validator beside it). Each machine of a send reads the mesh afresh (freshInput), because the send before it made and recorded things. None calls declarationWith with Allocating any more. The private network is still computed up front, so a send refuses on it before anything is composed. Also here, as they share the helpers: whoResolvesOn, restOfTheMesh and restOfTheMeshOf in compose_sdk.go, and the planFor callers in held_back.go, plan.go and push.go.

cmd/mesh-controller/push.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/compose"
 	"github.com/nats-io/nats.go"
 	"github.com/novox/mesh-controller/internal/broker"
 	"github.com/novox/mesh-controller/internal/inventory"
@@ -521,7 +522,7 @@ func pushCommand(ctx context.Context, args []string) error {
 			return nil
 		}
 	}
-	gens, err := generators(ctx, open)
+	gens, err := generatorsOf(ctx, open)
 	if err != nil {
 		return err
 	}
@@ -652,7 +653,9 @@ func pushCommand(ctx context.Context, args []string) error {
 	// for a machine named, as a count for each of many — the full list is `status`'s.
 	unheld := map[string][]catalogue.Unheld{}
 	sending, refusals := composeEach(asked, allotting(held, inv), func(node string) (sendable, error) {
-		plan, settings, err := planFor(held, open, node)
+		// What this machine's send reads is read afresh: the send before it made and recorded things.
+		held := freshInput(held)
+		plan, _, err := resolveNode(held, open, node)
 		if err != nil {
 			return sendable{}, err
 		}
@@ -665,7 +668,7 @@ func pushCommand(ctx context.Context, args []string) error {
 		// The private network is in here with everything else. It used to be composed separately
 		// and prepended, which meant every machine with an address was on it and no machine could
 		// be kept off. It is a module now, so it arrives the way a module does.
-		declared, err := declarationWith(held, open, node, plan, settings, gens, Allocating)
+		declared, err := composeSend(held, open, node)
 		if err == nil {
 			reportLeftOut(node, declared)
 		}
@@ -1152,7 +1155,7 @@ func brokerListState(ctx context.Context, open *stores, names []string) (busList
 	// refuses rather than sending code whose grants the bus may not hold. It said "not behind" until then: a
 	// send went ahead as if the bus held a list nobody could compose.
 	list, composeErr := "", error(nil)
-	plan, _, err := planFor(ctx, open, h.Node)
+	plan, _, err := resolveNode(ctx, open, h.Node)
 	if err != nil {
 		composeErr = err
 	} else if list, _, err = busUserList(ctx, inv, plan.Modules); err != nil {
@@ -1229,8 +1232,8 @@ func sendToEach(ctx context.Context, open *stores, names []string) ([]string, er
 	}
 	defer ident.Close()
 
-	gens, err := generators(ctx, open)
-	if err != nil {
+	// The private network every declaration of the send is composed with: not computable refuses the send up front.
+	if _, err := generatorsOf(ctx, open); err != nil {
 		return nil, err
 	}
 	holder, behind, err := brokerBehind(ctx, open, names)
@@ -1272,14 +1275,16 @@ func sendToEach(ctx context.Context, open *stores, names []string) ([]string, er
 			refusals = append(refusals, fmt.Sprintf("%s:\n%v", name, err))
 			continue
 		}
-		plan, settings, err := planFor(ctx, open, name)
+		// Read afresh for each machine: the send before it made and recorded things.
+		composing := freshInput(ctx)
+		plan, _, err := resolveNode(composing, open, name)
 		if err != nil {
 			refusals = append(refusals, fmt.Sprintf("%s:\n%v", name, err))
 			continue
 		}
 		reportUnhostable(name, plan)
 		reportKept(ctx, plan)
-		declared, err := declarationWith(ctx, open, name, plan, settings, gens, Allocating)
+		declared, err := composeSend(composing, open, name)
 		if err != nil {
 			refusals = append(refusals, fmt.Sprintf("%s:\n%v", name, err))
 			continue
@@ -1454,20 +1459,21 @@ type planned struct {
 func wouldSendFrom(ctx context.Context, open *stores,
 	nodes []inventory.Node, plans map[string]planned) (map[string]string, error) {
 
-	gens, err := generators(ctx, open)
-	if err != nil {
+	if _, err := generatorsOf(ctx, open); err != nil {
 		return nil, err
 	}
+	var err error
 	out := map[string]string{}
 	for _, n := range nodes {
 		known, have := plans[n.Name]
 		plan, settings := known.plan, known.settings
 		if !have {
-			if plan, settings, err = planFor(ctx, open, n.Name); err != nil {
+			if plan, settings, err = resolveNode(ctx, open, n.Name); err != nil {
 				continue
 			}
 		}
-		declared, err := declarationWith(ctx, open, n.Name, plan, settings, gens, Reading)
+		_, _ = plan, settings
+		declared, _, err := composeQuestion(ctx, open, n.Name, compose.Reading, false)
 		if err != nil {
 			continue
 		}