mesh-delivery: share a carrier's walk only when it delivers the whole superseded member

A turn given on one carrier's waiting walk let the next member go while
another module of the member before it waited on a walk that could still
fail, and the stop then took the shared walk down. The turn is now shared
only when that one walk carries every module not yet carried, each by a
merge the forge said contains the member's; an unknown answer is asked and
waited for (novox/hq issue 445, re-review of #203).

modules/mesh-delivery/cmd/mesh-delivery/holder.go

@@ -1406,8 +1406,14 @@ func (h *Holder) groupTurns(g *Group, members []*Delivery) []func() {
 		if d.State != Published || !d.LetGo.IsZero() {
 			w = carryWait{} // only a member still waiting for its turn waits for a carrier
 		}
-		if d.TurnAwaits != w.said || !d.TurnAwaitsUntil.Equal(w.until) {
-			d.TurnAwaits, d.TurnAwaitsUntil = w.said, w.until
+		var until *time.Time
+		if !w.until.IsZero() {
+			u := w.until
+			until = &u
+		}
+		if d.TurnAwaits != w.said || (d.TurnAwaitsUntil == nil) != (until == nil) ||
+			until != nil && !d.TurnAwaitsUntil.Equal(*until) {
+			d.TurnAwaits, d.TurnAwaitsUntil = w.said, until
 			h.keep(d)
 		}
 	}
@@ -1446,6 +1452,13 @@ func (h *Holder) turnsOf(g *Group, members []*Delivery, waits map[string]carryWa
 			// wait, as for a member still delivering (ADR 0239 decision 4: the next starts when the one before is
 			// delivered). Only when none can carry it are they stopped, saying why (novox/hq issue 445).
 			w := h.carryAwaited(d, h.Now())
+			// A delivered carrier is asked of the forge here too: coverOf asks only once every module has a
+			// delivered carrier, and while another module waits on an open one, this answer is what is awaited.
+			for _, in := range w.ask {
+				if a := h.askContains(d, in, h.Now()); a != nil {
+					asks = append(asks, a)
+				}
+			}
 			if w.covered() {
 				continue // every module delivered by a later merge the forge says contains it: reached
 			}
@@ -1468,7 +1481,12 @@ func (h *Holder) turnsOf(g *Group, members []*Delivery, waits map[string]carryWa
 			// the next member's own walk, which waits for that member's turn. Waiting there would wait for itself
 			// until the carrier's bound. That walk delivers both, in the group's order among its tiers (decision 5),
 			// so the members on it take their turn as members sharing one walk do; a member on another walk waits.
-			if shared := w.waitingWalk(); shared != "" && sharing == "" {
+			// It is shared only when that one walk carries every module of the member not yet carried, each by a
+			// merge the forge said contains the member's: a carrier of some modules on it, others elsewhere, would
+			// let the next member go on a walk that cannot deliver the member before it (review of #203).
+			shared, ask := h.sharedWalk(d, w, h.Now())
+			asks = append(asks, ask...)
+			if shared != "" && sharing == "" {
 				sharing = shared
 				continue
 			}
@@ -1538,22 +1556,65 @@ const readyCarrierBound = 10*time.Minute + 2*time.Hour
 // the moment the wait ends at the latest; or why none can carry it. Neither means every module it moves was
 // delivered by a later merge the forge says contains it: covered.
 type carryWait struct {
-	by    []*Delivery
-	said  string
-	until time.Time
-	why   string
+	by []*Delivery
+	// byModule are the carriers awaited for each module not yet carried.
+	byModule map[string][]*Delivery
+	said     string
+	until    time.Time
+	why      string
+	// ask are the delivered carriers' merge commits the forge has not yet said contain the member's.
+	ask []string
 }
 
 func (w carryWait) covered() bool { return len(w.by) == 0 && w.why == "" }
 
-// waitingWalk is the walk of an awaited carrier that has not started: one waiting for this owner's word.
-func (w carryWait) waitingWalk() string {
+// sharedWalk is the walk, not started and waiting for this owner's word, that carries every module of a superseded
+// member not yet carried, each by a carrier the forge said contains the member's merge: the walk the members after
+// it may take their turn on. Whether a carrier on it contains the merge, when not yet known, is asked of the forge,
+// and the members wait for the answer within the carrier's bound (ADR 0270, novox/hq issue 445).
+func (h *Holder) sharedWalk(d *Delivery, w carryWait, now time.Time) (string, []func()) {
+	var walks []string
 	for _, c := range w.by {
-		if !c.State.Final() && c.Walk != nil && c.Walk.Waits && !c.Walk.Started() {
-			return c.Walk.ID
+		if !c.State.Final() && c.Walk != nil && c.Walk.Waits && !c.Walk.Started() && !containsString(walks, c.Walk.ID) {
+			walks = append(walks, c.Walk.ID)
 		}
 	}
-	return ""
+	sort.Strings(walks)
+	var asks []func()
+	for _, walk := range walks {
+		every, said := true, true
+		var unknown []string
+		for _, carriers := range w.byModule {
+			on, yes := false, false
+			for _, c := range carriers {
+				if c.State.Final() || c.Walk == nil || c.Walk.ID != walk || !c.Walk.Waits || c.Walk.Started() {
+					continue
+				}
+				on = true
+				contained, known := d.ContainedIn[c.MergedAs]
+				switch {
+				case known && contained:
+					yes = true
+				case !known && !containsString(unknown, c.MergedAs):
+					unknown = append(unknown, c.MergedAs)
+				}
+			}
+			every = every && on
+			said = said && yes
+		}
+		if !every {
+			continue // a module is carried elsewhere, or by nothing on this walk: it cannot deliver the member
+		}
+		if said {
+			return walk, asks
+		}
+		for _, in := range unknown {
+			if a := h.askContains(d, in, now); a != nil {
+				asks = append(asks, a)
+			}
+		}
+	}
+	return "", asks
 }
 
 // carryAwaited says which later merges into a superseded member's trunk may still carry every module it moves —
@@ -1593,6 +1654,9 @@ modules:
 					not = append(not, fmt.Sprintf("the forge did not say within %s whether %s contains its merge", containsGrace, c.ID))
 					continue
 				}
+				if !containsString(w.ask, c.MergedAs) {
+					w.ask = append(w.ask, c.MergedAs)
+				}
 			case c.State.Final() || c.State == Held:
 				continue // ended without carrying it, or held for a person: no carrier to wait for
 			case c.State == Ready:
@@ -1612,6 +1676,10 @@ modules:
 				}
 			}
 			found = true
+			if w.byModule == nil {
+				w.byModule = map[string][]*Delivery{}
+			}
+			w.byModule[module] = append(w.byModule[module], c)
 			if until.After(latest) {
 				latest = until
 			}

modules/mesh-delivery/cmd/mesh-delivery/issue445_test.go

@@ -35,6 +35,7 @@ type shape445 struct {
 	a, b, c *Delivery
 	extra   []*Delivery
 	walks   []Walk
+	forge   map[string]bool // what the forge says of containment from the start
 }
 
 func newShape445() *shape445 {
@@ -102,7 +103,11 @@ func (s *shape445) open(t *testing.T) *world {
 	for _, w := range s.walks {
 		ctlr.put(w)
 	}
-	w := newWorldOver(t, store, ctlr, newFakeForge(), at445.Add(10*time.Minute))
+	forge := newFakeForge()
+	for k, v := range s.forge {
+		forge.contains[k] = v
+	}
+	w := newWorldOver(t, store, ctlr, forge, at445.Add(10*time.Minute))
 	if err := w.h.Reconcile(); err != nil {
 		t.Fatal(err)
 	}
@@ -128,7 +133,7 @@ func waits445(t *testing.T, w *world, on string) {
 	if d.State != Published || containsString(w.ctl.delivers, "plan-lab") {
 		t.Fatalf("B is %s, let go %v; want it published and waiting while %s can still carry A", d.State, w.ctl.delivers, on)
 	}
-	if !strings.Contains(d.TurnAwaits, on) || !strings.Contains(d.TurnAwaits, aID) || d.TurnAwaitsUntil.IsZero() {
+	if !strings.Contains(d.TurnAwaits, on) || !strings.Contains(d.TurnAwaits, aID) || d.TurnAwaitsUntil == nil {
 		t.Fatalf("B says it waits for %q until %v; want it to name A and %s", d.TurnAwaits, d.TurnAwaitsUntil, on)
 	}
 	g := w.h.groups[g445]
@@ -227,22 +232,13 @@ func TestAMemberAfterACarriableSupersededOneWaitsForItsCarrier(t *testing.T) {
 // answers C's merge and B's and not A's. The carrier is in B's own walk, which waits for B's turn: B takes its
 // turn there, and the one walk delivers C and B; A is covered once it is done.
 func TestAMemberWhoseWalkCarriesTheSupersededOneBeforeItTakesItsTurn(t *testing.T) {
-	s := newShape445()
-	s.c.State, s.c.Walk = Published, &WalkSeen{ID: "plan-2", Commit: mergedC, State: walkBuilding, Waits: true,
-		WaitedFor: "mesh-delivery", Branch: "main", Created: at445.Add(4 * time.Minute)}
-	s.b.Walk = &WalkSeen{ID: "plan-2", Commit: mergedC, State: walkBuilding, Waits: true, WaitedFor: "mesh-delivery",
-		Branch: "main", Created: at445.Add(4 * time.Minute)}
-	two := ctlWalk("plan-2", mergedC, at445.Add(4*time.Minute), walkBuilding, &WalkDelivery{Awaits: "mesh-delivery",
-		Merges: []WalkMerge{{Repository: ctl445, Commit: mergedC}, {Repository: lab445, Commit: mergedB}}})
-	two.Tiers = [][]string{modules445, {"mesh-lab"}}
-	two.Modules["mesh-lab"] = &WalkModule{}
-	s.walks = []Walk{s.walks[0], two}
+	s := foldShape()
+	s.forge = map[string]bool{mergedA + " in " + mergedC: true}
 	w := s.open(t)
 	letGo445(t, w, "plan-2")
 	if s := w.state(cID); s != Delivering {
 		t.Fatalf("C is %s", s)
 	}
-	w.forge.contains[mergedA+" in "+mergedC] = true
 	end445(w, "plan-2", walkDone)
 	g := w.h.groups[g445]
 	if state, why := GroupState(g, w.h.membersOf(g)); state != "delivered" {
@@ -280,6 +276,8 @@ func TestTheWaitFollowsACarrierSupersededInTurn(t *testing.T) {
 	}
 	w.forge.contains[mergedA+" in "+mergedD] = true
 	end445(w, "plan-3", walkDone)
+	w.later(2 * time.Minute) // the forge, asked while it could not say, is asked again a minute on
+	w.settleAll()
 	if d := w.delivery(aID); d.CoveredBy == "" || !strings.Contains(d.CoveredBy, dID445) {
 		t.Fatalf("A is covered by %q", d.CoveredBy)
 	}
@@ -297,12 +295,23 @@ func TestACarrierInABatchIsAwaitedWithinTheWindowAndOneWalk(t *testing.T) {
 			ClosesAt: &closes, Behind: "plan-other"}, Merges: []WalkMerge{{Repository: ctl445, Commit: mergedC}}}}
 	w := s.open(t)
 	waits445(t, w, cID)
-	if want := s.c.MergedAt.Add(readyCarrierBound); !w.delivery(bID).TurnAwaitsUntil.Equal(want) {
-		t.Fatalf("B waits until %v, want %v", w.delivery(bID).TurnAwaitsUntil, want)
+	// The merge window at its longest, then the one open walk the batch may queue behind (ADR 0276 decisions 1,
+	// 4 and 9).
+	if readyCarrierBound != 10*time.Minute+Bounds[Delivering].For {
+		t.Fatalf("a ready carrier is awaited %s, want the window's 10 minutes and one walk's %s", readyCarrierBound,
+			Bounds[Delivering].For)
 	}
-	w.later(readyCarrierBound)
+	until := s.c.MergedAt.Add(readyCarrierBound)
+	if got := w.delivery(bID).TurnAwaitsUntil; got == nil || !got.Equal(until) {
+		t.Fatalf("B waits until %v, want %v", got, until)
+	}
+	// Within the bound, past every state bound a ready delivery could be read against, B still waits.
+	w.later(until.Add(-time.Minute).Sub(w.now))
 	w.settleAll()
-	stopped445(t, w, "no walk carried it")
+	waits445(t, w, cID)
+	w.later(2 * time.Minute)
+	w.settleAll()
+	stopped445(t, w, "no walk carried it within "+readyCarrierBound.String())
 }
 
 // A member whose every module a later delivered merge carried, the forge saying it contains the member's merge,
@@ -343,3 +352,95 @@ func TestACarriedMemberNotYetMarkedCoveredIsReached(t *testing.T) {
 		t.Fatalf("B is %s, at its turn %v; want its turn taken", b.State, turned)
 	}
 }
+
+// foldShape is the review's shape: B and C on one walk, plan-2, not started, waiting for mesh-delivery's word.
+func foldShape() *shape445 {
+	s := newShape445()
+	s.c.State, s.c.Walk = Published, &WalkSeen{ID: "plan-2", Commit: mergedC, State: walkBuilding, Waits: true,
+		WaitedFor: "mesh-delivery", Branch: "main", Created: at445.Add(4 * time.Minute)}
+	s.b.Walk = &WalkSeen{ID: "plan-2", Commit: mergedC, State: walkBuilding, Waits: true, WaitedFor: "mesh-delivery",
+		Branch: "main", Created: at445.Add(4 * time.Minute)}
+	two := ctlWalk("plan-2", mergedC, at445.Add(4*time.Minute), walkBuilding, &WalkDelivery{Awaits: "mesh-delivery",
+		Merges: []WalkMerge{{Repository: ctl445, Commit: mergedC}, {Repository: lab445, Commit: mergedB}}})
+	two.Tiers = [][]string{modules445, {"mesh-lab"}}
+	two.Modules["mesh-lab"] = &WalkModule{}
+	s.walks = []Walk{s.walks[0], two}
+	return s
+}
+
+// The turn on the carrier's walk is given only on the forge's word that the carrier contains A's merge: not yet
+// said, B waits and the forge is asked; said "no", C is no carrier and B is stopped, never let go on plan-2.
+func TestASharedTurnWaitsForTheForgesWord(t *testing.T) {
+	t.Run("not yet said: B waits and the forge is asked", func(t *testing.T) {
+		w := foldShape().open(t)
+		waits445(t, w, cID)
+		if containsString(w.ctl.delivers, "plan-2") {
+			t.Fatalf("plan-2 was let go %v before the forge said C contains A", w.ctl.delivers)
+		}
+		w.forge.mu.Lock()
+		asked := containsString(w.forge.containsAsks, mergedA+" in "+mergedC)
+		w.forge.mu.Unlock()
+		if !asked {
+			t.Fatal("the forge was not asked whether C's merge contains A's")
+		}
+		w.forge.contains[mergedA+" in "+mergedC] = true
+		w.later(2 * time.Minute)
+		w.settleAll()
+		letGo445(t, w, "plan-2")
+	})
+	t.Run("said no: no shared turn", func(t *testing.T) {
+		s := foldShape()
+		s.forge = map[string]bool{mergedA + " in " + mergedC: false}
+		w := s.open(t)
+		if d := w.delivery(bID); d.State == Delivering || d.LetGo != (time.Time{}) {
+			t.Fatalf("B was given its turn on plan-2 (%s, let go %v)", d.State, d.LetGo)
+		}
+		stopped445(t, w, "does not contain its merge")
+	})
+}
+
+// Two carriers for A: C on B's own walk carries one module, D on a walk of its own the other. B is not given its
+// turn on C's walk, which cannot deliver A whole: it waits for D's walk, goes once D covers its module, and is
+// stopped, never delivering, when D's walk fails (re-review of #203).
+func TestASharedTurnNeedsEveryModuleOnTheOneWalk(t *testing.T) {
+	two := func() *shape445 {
+		s := foldShape()
+		s.a.Plan = planMoving(ctl445, "mesh-controller", "route-proxy")
+		s.c.Plan = planMoving(ctl445, "mesh-controller")
+		s.extra = []*Delivery{{ID: dID445, Repository: ctl445, Commit: headD, Number: 111, Base: "main",
+			Branch: "fix/other", State: Delivering, Since: at445.Add(5 * time.Minute), Created: at445.Add(-time.Hour),
+			MergedAs: mergedD, MergedAt: at445.Add(90 * time.Second), Plan: planMoving(ctl445, "route-proxy"),
+			Check: &Verdict{Gate: "pass"}, Walk: &WalkSeen{ID: "plan-other", Commit: mergedD, State: walkRolling,
+				WaitedFor: "mesh-delivery", GoBy: byOwner, Branch: "main", Created: at445.Add(4 * time.Minute)}}}
+		went := at445.Add(5 * time.Minute)
+		other := ctlWalk("plan-other", mergedD, at445.Add(4*time.Minute), walkRolling, &WalkDelivery{
+			Awaits: "mesh-delivery", Go: &went, By: byOwner, Merges: []WalkMerge{{Repository: ctl445, Commit: mergedD}}})
+		other.Modules = map[string]*WalkModule{"route-proxy": {}}
+		s.walks = append(s.walks, other)
+		s.forge = map[string]bool{mergedA + " in " + mergedC: true, mergedA + " in " + mergedD: true}
+		return s
+	}
+	t.Run("D's walk open: B waits for both", func(t *testing.T) {
+		w := two().open(t)
+		waits445(t, w, dID445)
+		if !strings.Contains(w.delivery(bID).TurnAwaits, cID) || containsString(w.ctl.delivers, "plan-2") {
+			t.Fatalf("B is %s, let go %v, waiting for %q", w.state(bID), w.ctl.delivers, w.delivery(bID).TurnAwaits)
+		}
+	})
+	t.Run("D's walk delivered: B goes on C's walk", func(t *testing.T) {
+		w := two().open(t)
+		end445(w, "plan-other", walkDone)
+		letGo445(t, w, "plan-2")
+	})
+	t.Run("D's walk failed: B is stopped, never delivering", func(t *testing.T) {
+		w := two().open(t)
+		end445(w, "plan-other", walkFailed)
+		d := w.delivery(bID)
+		for _, tr := range d.Transitions {
+			if tr.To == Delivering {
+				t.Fatalf("B was delivering before it was stopped: %+v", d.Transitions)
+			}
+		}
+		stopped445(t, w, "route-proxy")
+	})
+}

modules/mesh-delivery/cmd/mesh-delivery/model.go

@@ -68,8 +68,8 @@ type Delivery struct {
 	// TurnAwaits says, for a group member whose turn waits for a later merge to carry a superseded member before it,
 	// which merges and until when; TurnAwaitsUntil is that moment, which bounds the wait (ADR 0276 decision 9,
 	// novox/hq issue 445). `show` shows both, and `stalled` says them past that moment.
-	TurnAwaits      string    `json:"turn_awaits,omitempty"`
-	TurnAwaitsUntil time.Time `json:"turn_awaits_until,omitempty"`
+	TurnAwaits      string     `json:"turn_awaits,omitempty"`
+	TurnAwaitsUntil *time.Time `json:"turn_awaits_until,omitempty"`
 	// ContainedIn is the forge's answer, by a later merge commit, to whether that commit contains this one's
 	// merge (gitea_contains): asked once per pair, and kept (ADR 0270).
 	ContainedIn map[string]bool `json:"contained_in,omitempty"`

modules/mesh-delivery/cmd/mesh-delivery/table.go

@@ -501,7 +501,7 @@ func boundOf(d *Delivery) (Bound, bool) {
 	if d.waitsForItsOwnWalk() {
 		return rewalkBound, true
 	}
-	if d.State == Published && d.TurnAwaits != "" && !d.TurnAwaitsUntil.IsZero() {
+	if d.State == Published && d.TurnAwaits != "" && d.TurnAwaitsUntil != nil {
 		// Its turn waits for a later merge to carry a member before it: bounded by that carrier's own bound, and
 		// said with it, never the generic "waits for its turn or its word" (novox/hq issue 445).
 		b := Bounds[Published]