Node shadow: reuse a running seal only at its own slot, lay values as the controller does, answer the newest candidate first and say how it goes (issue 1126, C1 C2, T1 T2 T3)

A box is reused by slot (resource, field) after the composition, from what runs or the
have's slots; a have written before slots is reused within its module alone. Config
values are laid by mesh-sdk's one rule (never a default). The shadow answers the
newest candidate first and hands each answer as it lands, says when an answer starts,
every minute what it waits on, and when it ends, and fits an answer to the node's key.
Needs mesh-sdk fix/1126-seed-same (16b8d8a1).

internal/selfcompose/shadow.go

@@ -276,26 +276,92 @@ func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) {
 	}
 }
 
-// answerAll answers every group and hands the answers when any changed.
+// answerAll answers every group, the newest candidate first, and hands the answers each time one changes (issue 1126,
+// T3: ace answered a closed pull request's group for 22 minutes before the newer one, all handed at the end). Newest
+// is the highest revision of the group's keys: a closed pull request's candidate is the older and waits; the indexer
+// drops a closed one's keys, which ends its answer here (a node does not read the forge). Each answer says when it
+// starts, every progressEvery while it runs what it waits on, and when it ends (T2: the node said nothing for 22
+// minutes).
 func (s *Shadow) answerAll(ctx context.Context) {
 	c := s.c
 	node := c.env.Node
 	s.mu.Lock()
-	groups := make([]string, 0, len(s.candidates))
-	for g := range s.candidates {
-		groups = append(groups, g)
+	type listed struct {
+		group string
+		rev   uint64
+	}
+	var groups []listed
+	for g, keys := range s.candidates {
+		l := listed{group: g}
+		for _, k := range keys {
+			l.rev = max(l.rev, k.rev)
+		}
+		groups = append(groups, l)
+	}
+	// Each group's last answer stands until it is answered again; a group whose keys went is dropped.
+	next := map[string]delivery.CandidateAnswer{}
+	for _, l := range groups {
+		if a, ok := s.answers[l.group]; ok {
+			next[l.group] = a
+		}
 	}
 	s.mu.Unlock()
-	sort.Strings(groups)
-	next := map[string]delivery.CandidateAnswer{}
-	for _, group := range groups {
+	sort.Slice(groups, func(i, j int) bool {
+		if groups[i].rev != groups[j].rev {
+			return groups[i].rev > groups[j].rev
+		}
+		return groups[i].group < groups[j].group
+	})
+	hand := func() {
+		s.mu.Lock()
+		same := len(next) == len(s.answers)
+		for g, a := range next {
+			if old, ok := s.answers[g]; !ok || !sameAnswer(old, a) {
+				same = false
+			}
+		}
+		out := make(map[string]delivery.CandidateAnswer, len(next))
+		for g, a := range next {
+			out[g] = a
+		}
+		if !same {
+			s.answers = out
+		}
+		s.mu.Unlock()
+		if !same && s.answer != nil {
+			given := make(map[string]delivery.CandidateAnswer, len(out))
+			for g, a := range out {
+				given[g] = a
+			}
+			s.answer(ctx, given)
+			c.env.Say(fmt.Sprintf("%s answered %d placement candidate group(s)", node, len(given)))
+		}
+	}
+	for _, l := range groups {
+		if ctx.Err() != nil {
+			return
+		}
+		group := l.group
 		began := time.Now()
-		slow := time.AfterFunc(slowAfter, func() {
-			c.env.Say(fmt.Sprintf("the answer to %s has run over %s, and goes on", group, slowAfter))
-		})
+		c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d)", group, l.rev))
+		stop := make(chan struct{})
+		go func() {
+			tick := time.NewTicker(progressEvery)
+			defer tick.Stop()
+			for {
+				select {
+				case <-stop:
+					return
+				case <-tick.C:
+					c.env.Say(fmt.Sprintf("the answer to %s has run %s and waits on %s", group,
+						time.Since(began).Round(time.Second), c.waitingOn()))
+				}
+			}
+		}()
 		a, ok := s.answerGroup(ctx, group)
-		slow.Stop()
-		if took := time.Since(began); took > slowAfter {
+		close(stop)
+		took := time.Since(began)
+		if took > slowAfter {
 			c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second)))
 			if ok {
 				// Words that do not move with the time it took, so an answer the same as before is not written again.
@@ -303,35 +369,27 @@ func (s *Shadow) answerAll(ctx context.Context) {
 			}
 		}
 		if ok {
+			a = delivery.FitAnswer(a, delivery.MaxAnswerBytes)
 			next[group] = a
+			c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", group, took.Round(time.Millisecond), a.Verdict))
+		} else {
+			delete(next, group)
+			c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group))
 		}
+		hand()
 	}
-	s.mu.Lock()
-	same := len(next) == len(s.answers)
-	for g, a := range next {
-		if old, ok := s.answers[g]; !ok || !sameAnswer(old, a) {
-			same = false
-		}
-	}
-	if !same {
-		s.answers = next
-	}
-	s.mu.Unlock()
-	if !same && s.answer != nil {
-		out := make(map[string]delivery.CandidateAnswer, len(next))
-		for g, a := range next {
-			out[g] = a
-		}
-		s.answer(ctx, out)
-		c.env.Say(fmt.Sprintf("%s answered %d placement candidate group(s)", node, len(out)))
-	}
+	hand() // a group whose keys went, when no group is left to answer
 }
 
+// How often a running answer says what it waits on.
+var progressEvery = time.Minute
+
 // answerGroup is the node's answer to one group, and false when the node has nothing to answer: neither the trunk nor
 // the group places it.
 func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.CandidateAnswer, bool) {
 	c := s.c
 	node := c.env.Node
+	defer c.waiting("composing")()
 	s.mu.Lock()
 	keys := s.candidates[group]
 	given := make(map[string]*delivery.Placement, len(keys))