Say every walk a verb keeps, and clear a stop from a walk that walks again

cancel, clear and rebuild kept a walk's end or restart without saying it, as plans stop, retry and
release-backlog did before them. The controller's run now says every save a verb makes on the bus, so the
next verb cannot lose a mark the same way. A stop is cleared when a walk is resumed, so one that fails again
ends failed and the search for its merge runs (review 21, ADR 0308, issue 532).

cmd/mesh-controller/delivery.go

@@ -10,6 +10,7 @@ import (
 	"slices"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/nats-io/nats.go"
@@ -1253,9 +1254,10 @@ func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error
 			return err
 		}
 		bus := link.OverNATS{Conn: conn, JS: js}
+		was := inventory.PlanSaved
 		checkEvents = bus
-		inventory.PlanSaved = func(p inventory.Plan) { publishPlanMoved(ctx, bus, facts, p) }
-		defer func() { checkEvents, inventory.PlanSaved = nil, nil }()
+		inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { publishPlanMoved(ctx, bus, facts, p) }
+		defer func() { checkEvents, inventory.PlanSaved = nil, was }()
 		ran = true
 		return f()
 	})
@@ -1265,3 +1267,65 @@ func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error
 	}
 	return err
 }
+
+// sayingWhatAVerbKept is PlanSaved for a process that is not the serving controller: each walk the verb keeps is
+// said as `plan-moved` before the save returns, with the facts of the inventory that kept it, on the bus the verb
+// reaches. It reaches the bus on the first save and never for a verb that keeps no walk. When the bus cannot be
+// reached, a save that started or ended a walk says so on the verb's error output: its mark is kept and no later
+// save says it, so the router holds what the walk touched until its bound, or not at all for a start. done lets
+// the bus go and takes the hook back.
+func sayingWhatAVerbKept(ctx context.Context) (said func(*inventory.Inventory, inventory.Plan), done func()) {
+	var (
+		mu      sync.Mutex
+		reached bool
+		bus     link.Bus
+		closing func()
+		failed  error
+	)
+	said = func(inv *inventory.Inventory, p inventory.Plan) {
+		mu.Lock()
+		defer mu.Unlock()
+		if !reached {
+			reached = true
+			if handActConn != nil {
+				bus, failed = overNATS(handActConn)
+			} else if js, err := aBus(); err != nil {
+				failed = err
+			} else if bus, failed = overNATS(js.Conn()); failed == nil {
+				closing = js.Close
+			} else {
+				js.Close()
+			}
+		}
+		if bus == nil {
+			switch p.WalkSaid {
+			case inventory.WalkStarted:
+				fmt.Fprintf(os.Stderr, "%s: its start is kept and could NOT be said as plan-moved (%v): the router "+
+					"does not hold what it touches\n", p.ID, failed)
+			case inventory.WalkEnded:
+				fmt.Fprintf(os.Stderr, "%s: its end is kept and could NOT be said as plan-moved (%v): the router holds "+
+					"what it touched until its bound\n", p.ID, failed)
+			}
+			return
+		}
+		publishPlanMoved(ctx, bus, inv, p)
+	}
+	done = func() {
+		mu.Lock()
+		defer mu.Unlock()
+		inventory.PlanSaved = nil
+		if closing != nil {
+			closing()
+		}
+	}
+	return said, done
+}
+
+// overNATS is the bus a connection carries, with its JetStream.
+func overNATS(conn *nats.Conn) (link.Bus, error) {
+	js, err := conn.JetStream()
+	if err != nil {
+		return nil, err
+	}
+	return link.OverNATS{Conn: conn, JS: js}, nil
+}