diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go index c0b8e873..70c37dbf 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go @@ -54,6 +54,9 @@ type Holder struct { seatMu sync.Mutex seatWhy string + // catchUp asks the next look to weigh again the newest merge of each repository against source.main (SourceCatchUp). + catchUp bool + mu sync.Mutex deliveries map[string]*Delivery groups map[string]*Group @@ -1845,6 +1848,12 @@ func (h *Holder) adopt(w Walk) { now := h.Now() d := &Delivery{ID: IDOf(w.Repository, w.Commit), Repository: strings.ToLower(w.Repository), Commit: w.Commit, Base: w.Branch, MergedAs: w.Commit, Created: now} + // Dated once, by the walk it is adopted from, which follows the merge: a later walk it follows (linkWalk) never redates + // it, so it never looks newer than a merge heard after it (issue 1726, review). + d.MergedAt = w.Created.UTC() + if w.Created.IsZero() { + d.MergedAt = now + } if old := h.deliveries[d.ID]; old != nil { if old.State.Final() { // Never revived: an ended delivery is not the walk's; the walk, held by nothing, is listed by diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1726_trunk_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1726_trunk_test.go new file mode 100644 index 00000000..c7f89b4d --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1726_trunk_test.go @@ -0,0 +1,161 @@ +package main + +import ( + "errors" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Issue 1726 (2026-10-24, mesh-catalog #471 and #472): the checker takes a repository's trunk head only from source.main +// (mesh-delivery modules/mesh-delivery-checker main.go), and the gate moves source.main only on a merge it hears. While +// gitea's watcher was dead, the operator merged #471; the restarted watcher's poll announced its pull.merged while the gate +// was starting, before it could tell whether it held the seat, and the gate acknowledged it unread. The gate then adopted the +// controller's walk of the merge commit, which has no merge time, so every older merge of the repository looked newer and +// source.main was never moved: the checker judged #472 on the trunk before #471, and the gate refused its merge. + +const ( + i1726Repo = "novox/mesh-catalog" + i1726Head1 = "aaaa000000000000000000000000000000000001" + i1726Merge1 = "dddd000000000000000000000000000000000002" + i1726Merge2 = "bbbb000000000000000000000000000000000003" +) + +// 1726c/AC1: a merge into main whose pull.merged the gate never heard, caught up as the controller's walk of its merge +// commit, moves source.main as the forge's word of it would. +func TestAMergeCaughtUpAsAnAdoptedWalkMovesSourceMain(t *testing.T) { + w := newWorld(t) + w.h.PullUpdated(pr(i1726Repo, 470, i1726Head1, "fix/a", "modules/app/a.go")) + w.h.PullMerged(i888MergedAt(i1726Repo, 470, i1726Head1, i1726Merge1, w.now.Add(-time.Hour))) + w.settleAll() + if w.source.main == nil || w.source.main.Heads[i1726Repo] != i1726Merge1 { + t.Fatalf("the heard merge did not move source.main: %+v", w.source.main) + } + // #471's pull.merged is lost; the controller walks its merge commit and the gate adopts the walk. + walk := aWalk("plan-471", i1726Repo, i1726Merge2, false) + walk.Created = w.now.Add(-time.Minute) + walk.State = walkRolling + w.ctl.put(walk) + if err := w.h.Reconcile(); err != nil { + t.Fatal(err) + } + w.settleAll() + if got := w.source.main.Heads[i1726Repo]; got != i1726Merge2 { + t.Fatalf("source.main holds %s at %s after the walk of %s was adopted; wanted the adopted merge", i1726Repo, got, i1726Merge2) + } +} + +// 1726c/AC2: a merge already marked in the source without being written (as the gate before this fix left #471) is written +// when the gate next holds the seat; one older than the head source.main holds never moves it back. +func TestHoldingTheSeatWritesTheNewestMergeOfEachRepositoryNotYetInSourceMain(t *testing.T) { + w := newWorld(t) + w.h.PullUpdated(pr(i1726Repo, 470, i1726Head1, "fix/a", "modules/app/a.go")) + w.h.PullMerged(i888MergedAt(i1726Repo, 470, i1726Head1, i1726Merge1, w.now.Add(-time.Hour))) + w.settleAll() + walk := aWalk("plan-471", i1726Repo, i1726Merge2, false) + walk.Created = w.now.Add(-time.Minute) + walk.State = walkRolling + w.ctl.put(walk) + if err := w.h.Reconcile(); err != nil { + t.Fatal(err) + } + w.settleAll() + // What the gate before this fix left: the adopted merge marked in the source, source.main still at the merge before. + if d := w.delivery(IDOf(i1726Repo, i1726Merge2)); !d.InSource { + t.Fatalf("the adopted merge is not marked in the source: %+v", d) + } + w.source.main = &sdk.Source{V: sdk.Version, Group: "before", Heads: map[string]string{i1726Repo: i1726Merge1, "novox/other": "b0"}} + w.settleAll() + if got := w.source.main.Heads[i1726Repo]; got != i1726Merge1 { + t.Fatalf("source.main moved to %s before the gate held the seat again", got) + } + w.h.SourceCatchUp() + w.settleAll() + if got := w.source.main.Heads[i1726Repo]; got != i1726Merge2 || w.source.main.Heads["novox/other"] != "b0" { + t.Fatalf("source.main is %+v after the catch-up; wanted %s at %s and the other kept", w.source.main.Heads, i1726Repo, i1726Merge2) + } + n := len(w.source.moves) + w.h.SourceCatchUp() + w.settleAll() + if len(w.source.moves) != n { + t.Fatalf("a second catch-up moved source.main again: %d moves, then %d", n, len(w.source.moves)) + } + + o := newWorld(t) + o.source.main = &sdk.Source{V: sdk.Version, Group: "newer", Heads: map[string]string{i1726Repo: i1726Merge2}} + o.h.PullUpdated(pr(i1726Repo, 471, "cccc000000000000000000000000000000000001", "fix/b", "modules/app/b.go")) + o.h.PullMerged(i888MergedAt(i1726Repo, 471, "cccc000000000000000000000000000000000001", i1726Merge2, o.now.Add(-time.Minute))) + o.h.PullMerged(i888MergedAt(i1726Repo, 470, i1726Head1, i1726Merge1, o.now.Add(-time.Hour))) + o.settleAll() + o.h.SourceCatchUp() + o.settleAll() + if got := o.source.main.Heads[i1726Repo]; got != i1726Merge2 { + t.Fatalf("the catch-up moved source.main back to %s", got) + } +} + +// seatUnknown is a seat whose occupants cannot be read yet. +type seatUnknown struct{} + +func (seatUnknown) Occupants() ([]string, error) { + return nil, errors.New("the watches have not delivered yet") +} + +// seatTaken is a seat today's module still stands on. +type seatTaken struct{} + +func (seatTaken) Occupants() ([]string, error) { return []string{"assigned on elsewhere"}, nil } + +// 1726c/AC3: an event that reaches the gate before it can tell whether it holds the seat is offered again, never +// acknowledged unread; one that reaches a gate whose seat another holds is acknowledged and left, as before. +func TestAnEventBeforeTheGateKnowsWhetherItHoldsTheSeatIsOfferedAgain(t *testing.T) { + w := newWorld(t) + w.h.Seat = seatUnknown{} + _ = w.h.SeatChanged() + ev := stdio.Envelope{Key: "gitea.pull.merged", Body: mustJSON(i888MergedAt(i1726Repo, 471, i1726Head1, i1726Merge2, w.now))} + if err := handle(w.h, ev); err == nil { + t.Fatal("an event heard before the gate could tell whether it holds the seat was taken: acknowledged unread") + } + w.h.Seat = seatTaken{} + _ = w.h.SeatChanged() + if err := handle(w.h, ev); err != nil { + t.Fatalf("an event heard while today's module holds the seat was offered again: %v", err) + } +} + +// 1726c/AC4 (review round 1): an adopted merge that later follows a newer walk keeps the date of the walk it was adopted +// from, so a merge heard from the forge after it stays in source.main through a catch-up. +func TestAnAdoptedMergeFollowingANewerWalkNeverMovesSourceMainBack(t *testing.T) { + w := newWorld(t) + walk := aWalk("plan-471", i1726Repo, i1726Merge2, false) + walk.Created = w.now.Add(-10 * time.Minute) + walk.State = walkRolling + w.ctl.put(walk) + if err := w.h.Reconcile(); err != nil { + t.Fatal(err) + } + w.settleAll() + if got := w.source.main.Heads[i1726Repo]; got != i1726Merge2 { + t.Fatalf("the adopted merge did not move source.main: %s", got) + } + const later = "eeee000000000000000000000000000000000004" + w.h.PullUpdated(pr(i1726Repo, 473, "ffff000000000000000000000000000000000001", "fix/c", "modules/app/c.go")) + w.h.PullMerged(i888MergedAt(i1726Repo, 473, "ffff000000000000000000000000000000000001", later, w.now.Add(-5*time.Minute))) + w.settleAll() + if got := w.source.main.Heads[i1726Repo]; got != later { + t.Fatalf("the later forge merge did not move source.main: %s", got) + } + again := aWalk("plan-471-again", i1726Repo, i1726Merge2, false) + again.Created = w.now.Add(-time.Minute) + again.State = walkRolling + w.h.mu.Lock() + w.h.linkWalk(w.h.deliveries[IDOf(i1726Repo, i1726Merge2)], again, i1726Merge2) + w.h.mu.Unlock() + w.h.SourceCatchUp() + w.settleAll() + if got := w.source.main.Heads[i1726Repo]; got != later { + t.Fatalf("the catch-up moved source.main back to the adopted merge %s after it followed a newer walk", got) + } +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index 2d70ff97..2aa08116 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -147,6 +147,7 @@ func run(h *Holder, l *listening, occupancy *busOccupancy) { func readOnHolding(h *Holder) { h.wantForge() h.hearAsks() + h.SourceCatchUp() // a merge missed or marked unwritten meanwhile reaches source.main, and so the checker (issue 1726) h.Poke() for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := h.Reconcile() @@ -162,6 +163,11 @@ func readOnHolding(h *Holder) { // handle takes one event, then asks for a look at every step the event makes possible (Work, React). An event that cannot be read is // said and acknowledged: read again it would fail again, and the controller's walks are read back by comparison anyway. func handle(h *Holder, e stdio.Envelope) error { + if h.Undecided() { + // Not known yet whether this gate holds the seat (starting, or carrying today's state): offered again in a few + // seconds, so an event heard at a restart is not lost (issue 1726). + return fmt.Errorf("%s: the gate cannot tell yet whether it holds the seat mesh-delivery; offered again", e.Key) + } if !h.Holding() { return nil // not the holder of the seat: the event is acknowledged and taken by no one (seat.go) } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go index 01692ce6..2e2ff9e3 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go @@ -433,10 +433,20 @@ func gateMerges(d *Delivery) bool { // and the checker judges on the trunk source.main names, so source.main moves too. A merge replayed as history, or older than // a merge of its repository already heard, is never written: source.main never moves backwards. Under the lock; the writing // happens outside it. +// +// A merge caught up after a gap (issue 1726) is settled the same way: the controller's walk of a merge commit whose +// pull.merged the gate never heard is adopted with the walk's time as its merge's (adopt, mergedWhen), and once each time the +// gate comes to hold the seat (SourceCatchUp) the newest merge of each repository it holds is weighed again though it is +// marked in the source, so a merge marked there and never written reaches source.main, and with it the checker. func (h *Holder) settleForgeMerges() []func() { var out []func() + again := map[*Delivery]bool{} + if h.catchUp { + h.catchUp = false + again = h.newestInSource() + } for _, d := range h.deliveries { - if d.MergedAs == "" || d.InSource || gateMerges(d) { + if d.MergedAs == "" || (d.InSource && !again[d]) || gateMerges(d) { continue } if d.Base != MergeBranch || d.History != "" || h.newerMergeOf(d) { @@ -444,7 +454,7 @@ func (h *Holder) settleForgeMerges() []func() { h.keep(d) continue } - id, repository, base, commit, at, group := d.ID, d.Repository, d.Base, d.MergedAs, d.MergedAt, firstOf(d.Branch, d.Repository) + id, repository, base, commit, at, group := d.ID, d.Repository, d.Base, d.MergedAs, mergedWhen(d), firstOf(d.Branch, d.Repository) out = append(out, func() { err := h.Source.Move(func(prev *sdk.Source) (sdk.Source, bool) { next := sdk.Source{Group: group, Heads: map[string]string{repository: commit}} @@ -485,7 +495,7 @@ func (h *Holder) settleForgeMerges() []func() { // (Ready.MergeCommit, Ready.Merged), which comes first for a merge the gate handed. Empty when it is not merged. func mergeOf(d *Delivery) (string, time.Time) { if d.MergedAs != "" { - return d.MergedAs, d.MergedAt + return d.MergedAs, mergedWhen(d) } if d.Ready != nil && d.Ready.Merged != nil && d.Ready.MergeCommit != "" { return d.Ready.MergeCommit, *d.Ready.Merged @@ -497,7 +507,7 @@ func mergeOf(d *Delivery) (string, time.Time) { func (h *Holder) newerMergeOf(d *Delivery) bool { for _, o := range h.deliveries { if commit, at := mergeOf(o); o != d && o.Repository == d.Repository && o.Base == d.Base && commit != "" && - !sameCommit(commit, d.MergedAs) && at.After(d.MergedAt) { + !sameCommit(commit, d.MergedAs) && at.After(mergedWhen(d)) { return true } } @@ -518,3 +528,44 @@ func (h *Holder) olderHead(repository, base, head string, at time.Time) bool { } return at.After(h.Now().Add(-keptFinal)) } + +// mergedWhen is when a merged delivery's merge landed: its merge time (the forge's, or for a merge adopted from the +// controller's walk the walk's creation, set once in adopt), else when the gate made the delivery. A zero time made every +// other merge of the repository look newer, so an adopted merge never moved source.main (issue 1726). +func mergedWhen(d *Delivery) time.Time { + if !d.MergedAt.IsZero() { + return d.MergedAt + } + return d.Created +} + +// SourceCatchUp asks the next look to weigh again, once, the newest merge into the trunk of each repository though it is +// marked in the source (issue 1726): what the gate missed while it did not hold the seat, or marked without writing, reaches +// source.main. The write never moves a repository back (olderHead), and one already there is not written again. +func (h *Holder) SourceCatchUp() { + h.mu.Lock() + h.catchUp = true + h.mu.Unlock() + h.Poke() +} + +// newestInSource is, for each repository, its newest merge into the trunk heard from the forge or caught up from a walk, +// marked in the source and not history. Under the lock. +func (h *Holder) newestInSource() map[*Delivery]bool { + newest := map[string]*Delivery{} + for _, d := range h.deliveries { + if d.MergedAs == "" || !d.InSource || gateMerges(d) || d.Base != MergeBranch || d.History != "" { + continue + } + if o := newest[d.Repository]; o == nil || mergedWhen(d).After(mergedWhen(o)) { + newest[d.Repository] = d + } + } + out := map[*Delivery]bool{} + for _, d := range newest { + if !h.newerMergeOf(d) { + out[d] = true + } + } + return out +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go index 3fbff7b0..c0d1d179 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go @@ -105,6 +105,11 @@ func (h *Holder) SeatChanged() error { // Holding is whether the gate acts. A holder with no Seat to ask always holds (the tests'; main sets one). func (h *Holder) Holding() bool { return h.Seat == nil || h.seatHeld.Load() } +// Undecided is whether the gate cannot tell yet whether it holds the seat: it does not hold it, and no reason it waits was +// found (the watches have not delivered, or today's state is being carried). An event heard then is offered again, never +// acknowledged unread (issue 1726: a merge announced at a restart was lost so, and source.main never moved). +func (h *Holder) Undecided() bool { return !h.Holding() && h.SeatWhy() == "" } + // setSeatWhy keeps why the gate waits and says whether it changed. func (h *Holder) setSeatWhy(why string) bool { h.seatMu.Lock()