Merge pull request 'The consistency cascade sends tolerantly and stops loudly (057 review)' (#32) from fix/cascade-is-tolerant into main

cmd/mesh-controller/push.go

@@ -336,7 +336,13 @@ func pushCommand(ctx context.Context, args []string) error {
 	// flushed send may itself mint, so this converges over a few rounds.
 	if len(args) == 1 {
 		flushed := map[string]bool{args[0]: true}
-		for round := 0; round < 4; round++ {
+		// Bounded by the node count: a node is marked flushed the round it is handled and is
+		// never handled twice, so the loop cannot run more than len(nodes) rounds. The bound is
+		// a guard against a logic error, not a real limit — if it were ever hit, that is a bug
+		// rather than a cascade legitimately still converging, so it is said rather than passed
+		// over in silence, unlike the earlier fixed cap that could stop a real cascade short.
+		rounds := 0
+		for {
 			would, err := wouldSend(ctx, open, nodes)
 			if err != nil {
 				return err
@@ -354,12 +360,49 @@ func pushCommand(ctx context.Context, args []string) error {
 			if len(also) == 0 {
 				break
 			}
+			if rounds++; rounds > len(nodes) {
+				fmt.Printf("\nstopped cascading after %d rounds with %s still behind — this "+
+					"should not happen; run `push --behind` to finish\n",
+					rounds-1, strings.Join(also, ", "))
+				break
+			}
 			sort.Strings(also)
 			fmt.Printf("\nthis push left %s behind — a provision granted from there, or a "+
 				"declaration since changed; sending it too\n", strings.Join(also, ", "))
-			if err := sendTo(ctx, open, also); err != nil {
+			// Tolerantly, exactly as the named send above: a machine that cannot be composed is
-				return err
+			// collected as a refusal and reported at the end, and the others are still sent
+			// (novox/hq ADR 0066). The earlier cut routed these through sendTo, which is
+			// all-or-nothing — so one swept machine's compose error failed the operator's named
+			// push and skipped its --wait, the very intolerance the main path exists to avoid.
+			sending, refused := composeEach(also, func(node string) ([]map[string]any, error) {
+				plan, settings, err := planFor(ctx, open, node)
+				if err != nil {
+					return nil, err
+				}
+				reportUnhostable(node, plan)
+				return declarationWith(ctx, open, node, plan, settings, gens, Allocating)
+			})
+			refusals = append(refusals, refused...)
+			for _, s := range sending {
+				body, err := json.Marshal(map[string]any{"declaration": 1, "resources": s.resources})
+				if err != nil {
+					return err
+				}
+				if err := link.Declare(ctx, server.Channel(), ident, s.node, body, 15*time.Second); err != nil {
+					return err
+				}
+				record, err := inv.NodeByName(ctx, s.node)
+				if err != nil {
+					return err
+				}
+				if err := inv.RecordSent(ctx, record.ID, digestOf(body)); err != nil {
+					return err
+				}
+				fmt.Printf("sent %s %d resource(s)\n", s.node, len(s.resources))
 			}
+			// Every candidate this round is marked handled — the sent ones so they are not
+			// re-listed, and the refused ones so a machine that cannot be composed does not make
+			// the loop spin on it for ever. Its refusal is already in the report.
 			for _, name := range also {
 				flushed[name] = true
 			}