diff --git a/modules/mesh-delivery/cmd/mesh-delivery/holder.go b/modules/mesh-delivery/cmd/mesh-delivery/holder.go index b75f537e..9223c804 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/holder.go +++ b/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 } diff --git a/modules/mesh-delivery/cmd/mesh-delivery/issue445_test.go b/modules/mesh-delivery/cmd/mesh-delivery/issue445_test.go index 7215ca0f..89ba0119 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/issue445_test.go +++ b/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") + }) +} diff --git a/modules/mesh-delivery/cmd/mesh-delivery/model.go b/modules/mesh-delivery/cmd/mesh-delivery/model.go index 9221ff13..25ec79e1 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/model.go +++ b/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"` diff --git a/modules/mesh-delivery/cmd/mesh-delivery/table.go b/modules/mesh-delivery/cmd/mesh-delivery/table.go index b2740e75..ad55d7ff 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/table.go +++ b/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]