From 116751130f5cc741b86c92738933cf8860f6b3a1 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 21:17:53 +0100 Subject: [PATCH] Cover a merged fix from the delivery view's reached head in the sweep, not only from its once-said event, so a fix named after the saying is delivered (issue 1950) --- .../cmd/mesh-issues/delivered_test.go | 70 +++++++++ .../mesh-issues/cmd/mesh-issues/delivery.go | 138 ++++++++++++++++-- modules/mesh-issues/cmd/mesh-issues/learn.go | 3 + modules/mesh-issues/cmd/mesh-issues/main.go | 1 + .../mesh-issues/cmd/mesh-issues/tracker.go | 3 + 5 files changed, 205 insertions(+), 10 deletions(-) diff --git a/modules/mesh-issues/cmd/mesh-issues/delivered_test.go b/modules/mesh-issues/cmd/mesh-issues/delivered_test.go index 67a810be..11255d8f 100644 --- a/modules/mesh-issues/cmd/mesh-issues/delivered_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/delivered_test.go @@ -2,6 +2,8 @@ package main import ( "encoding/json" + "strconv" + "strings" "testing" "time" @@ -157,6 +159,74 @@ func TestANewerRevisionCoversTheOlderMergesOfItsRepositoryAsTheForgeSaysAncestry t.Errorf("per repository: issue %d reads %q, wanted %q", n, v.Delivery, want) } } + reachLevelRows(t) +} + +// The two live answers of mesh-delivery-view.view_reach, captured 2026-10-27 20:30Z (changes shortened to the first two), of the group +// fix/1242-bus-helper whose delivered event the tracker never matched (issue 1950, second meeting): mesh-sdk, a library no node wants, +// and mesh-delivery, whose reached head is a later merge than the fix's. +const ( + liveReachSDK = `{"indexed":"9f92233a28902d5249996914da4972ab65f71b9b","modules":{"mesh-sdk":{"changes":[{"commit":"9f92233a28902d5249996914da4972ab65f71b9b","fp":"src2:cef6f0ef","src":507},{"commit":"a262c7dbe0ff37e7ead50548b6e41fd805f2caed","fp":"src2:1a523588","src":503}],"fp":"src2:cef6f0ef","nodes":0}},"reached":"9f92233a28902d5249996914da4972ab65f71b9b","reached_src":507,"repository":"novox/mesh-sdk","v":1}` + liveReachDelivery = `{"indexed":"da514816204b4bdc6df4c48b1d03ee39ce285159","modules":{"mesh-delivery-assigner":{"changes":[{"commit":"24cb379434f5d44da40276d069429b0bcc0c59ce","fp":"src2:8f305a82","src":507},{"commit":"c38e81c7f5e9b52164127a6610b7e8449dfccd51","fp":"src2:798fe942","src":506}],"floor":"24cb379434f5d44da40276d069429b0bcc0c59ce","fp":"src2:8f305a82","nodes":1}},"reached":"da514816204b4bdc6df4c48b1d03ee39ce285159","reached_src":508,"repository":"novox/mesh-delivery","v":1}` +) + +// Rows of TestANewerRevisionCoversTheOlderMergesOfItsRepositoryAsTheForgeSaysAncestry (issue 1950, second meeting): the view says a repository's reached head once per source revision, so a fix whose record came after +// that saying is matched from what the view says now (view_reach), in the sweep and by learn-fixes. +func reachLevelRows(t *testing.T) { + t.Helper() + h := newHarness(t) + sdk, dlv := fixIssue(t, h, "mesh-sdk#108"), fixIssue(t, h, "mesh-delivery#87") + for pr, as := range map[string][2]string{"novox/mesh-sdk#108": {"9f92233a28902d5249996914da4972ab65f71b9b", "9f92233a28902d5249996914da4972ab65f71b9b"}, + "novox/mesh-delivery#87": {"24cb379434f5d44da40276d069429b0bcc0c59ce", ""}} { + repo, num, _ := strings.Cut(pr, "#") + n, _ := strconv.Atoi(num) + raw, _ := json.Marshal(map[string]any{"repository": repo, "number": n, "to": "merged", "merged_as": as[0], "group": "fix/1242-bus-helper", + "at": time.Date(2026, 10, 27, 19, 49, 22, 0, time.UTC)}) + if err := h.OnTransition(stdio.Envelope{Key: "mesh-delivery-gate.transition", Body: raw}); err != nil { + t.Fatal(err) + } + } + answers := map[string]string{"novox/mesh-sdk": liveReachSDK, "novox/mesh-delivery": liveReachDelivery} + h.Reached = func(repository string) (Reach, error) { return parseReach([]byte(answers[repository])) } + h.Contains = func(_, _, commit, in string) (bool, error) { + return commit == "24cb379434f5d44da40276d069429b0bcc0c59ce" && in == "da514816204b4bdc6df4c48b1d03ee39ce285159", nil + } + if r, err := parseReach([]byte(liveReachDelivery)); err != nil || r.Head != "da514816204b4bdc6df4c48b1d03ee39ce285159" || r.Revision != 508 || len(r.NotYet) != 0 { + t.Fatalf("the live answer reads %+v, %v", r, err) + } + // A module a node wants with no floor holds its changes back: the merge is not yet on the machines. + if r, _ := parseReach([]byte(strings.Replace(liveReachDelivery, `"floor":"24cb379434f5d44da40276d069429b0bcc0c59ce",`, "", 1))); len(r.NotYet) != 2 { + t.Fatalf("a module with no floor says %v not yet", r.NotYet) + } + if _, err := h.Sweep(); err != nil { + t.Fatal(err) + } + for n, pr := range map[uint64]string{sdk: "mesh-sdk#108", dlv: "mesh-delivery#87"} { + if v, _ := h.View(n); v.Delivery != "delivered to every machine it should reach" { + t.Errorf("%s after the sweep reads %q", pr, v.Delivery) + } + } + // A merge the view says is not yet on the machines, or a forge that cannot say, leaves the fix for the next sweep. + h2 := newHarness(t) + n := fixIssue(t, h2, "mesh-delivery#88") + raw, _ := json.Marshal(map[string]any{"repository": "novox/mesh-delivery", "number": 88, "to": "merged", "merged_as": "24cb379434f5d44da40276d069429b0bcc0c59ce", + "at": time.Date(2026, 10, 27, 19, 49, 22, 0, time.UTC)}) + if err := h2.OnTransition(stdio.Envelope{Key: "mesh-delivery-gate.transition", Body: raw}); err != nil { + t.Fatal(err) + } + h2.Reached = func(string) (Reach, error) { + return parseReach([]byte(strings.Replace(liveReachDelivery, `"floor":"24cb379434f5d44da40276d069429b0bcc0c59ce",`, "", 1))) + } + h2.Contains = func(string, string, string, string) (bool, error) { return true, nil } + h2.Sweep() + if v, _ := h2.View(n); v.Delivery != "fix merged, not delivered: mesh-delivery#88" { + t.Fatalf("a merge said not yet: %q", v.Delivery) + } + h2.Reached = func(string) (Reach, error) { return Reach{}, errStuck } + h2.Sweep() + if v, _ := h2.View(n); v.Delivery != "fix merged, not delivered: mesh-delivery#88" { + t.Fatalf("a view that could not say: %q", v.Delivery) + } } // A fix kept with only its merged flag (heard through `checks`, or kept before release 6) is matched to a delivered revision diff --git a/modules/mesh-issues/cmd/mesh-issues/delivery.go b/modules/mesh-issues/cmd/mesh-issues/delivery.go index fd461b2a..d5efd72b 100644 --- a/modules/mesh-issues/cmd/mesh-issues/delivery.go +++ b/modules/mesh-issues/cmd/mesh-issues/delivery.go @@ -477,16 +477,7 @@ func (t *Tracker) OnDelivered(e stdio.Envelope) error { !(deliveredBy(f, body.Group, body.Heads) || t.coveredBy(f, body.Heads)) { continue } - pr := pr - if _, err := t.change(n, Act{Op: "delivery", By: Seat}, func(i *Issue) error { - g := fixOf(i, pr) - if g == nil || contains(i.Delivered, pr) { - return nil - } - g.State, g.Why, g.Since = "delivered", fmt.Sprintf("source revision %d reached the machines", body.Revision), at - i.Delivered = append(i.Delivered, pr) - return nil - }); err != nil { + if err := t.markDelivered(n, pr, fmt.Sprintf("source revision %d reached the machines", body.Revision), at); err != nil { return err } if len(changed) == 0 || changed[len(changed)-1] != n { @@ -761,6 +752,9 @@ func (t *Tracker) Sweep() (int, error) { if err := t.askUnheard(); err != nil { t.Logf("[mesh-issues] the sweep could not ask a fix's delivery (asked again at the next sweep): %v", err) } + if err := t.coverReached(); err != nil { + t.Logf("[mesh-issues] the sweep could not match the fixes to what reached the machines (tried again at the next sweep): %v", err) + } all, err := t.allIssues() if err != nil { return 0, err @@ -836,3 +830,127 @@ func (t *Tracker) Sweep() (int, error) { } return said, nil } + +// markDelivered keeps a fix as delivered, with the reason the delivery gave. +func (t *Tracker) markDelivered(n uint64, pr, why string, at time.Time) error { + _, err := t.change(n, Act{Op: "delivery", By: Seat}, func(i *Issue) error { + g := fixOf(i, pr) + if g == nil || contains(i.Delivered, pr) { + return nil + } + g.State, g.Why, g.Since = "delivered", why, at + i.Delivered = append(i.Delivered, pr) + return nil + }) + return err +} + +// Reach is what the delivery view says of one repository now (mesh-delivery-view.view_reach, ADR 0556): the newest merge on +// the machines, the source revision that named it, and the merges at or before it that are not yet there. +type Reach struct { + Head string + Revision uint64 + NotYet []string +} + +// busReach asks the delivery view for one repository's reached head. +func busReach(repository string) (Reach, error) { + raw, err := stdio.Ask("mesh-delivery-view.view_reach", map[string]any{"repository": repository}) + if err != nil { + return Reach{}, err + } + return parseReach(raw) +} + +// parseReach reads the view's answer, bare or inside an `answer`. +func parseReach(raw []byte) (Reach, error) { + type body struct { + Reached string `json:"reached"` + ReachedSrc uint64 `json:"reached_src"` + Modules map[string]struct { + Changes []struct { + Commit string `json:"commit"` + } `json:"changes"` + Nodes int `json:"nodes"` + Floor string `json:"floor"` + } `json:"modules"` + } + var out struct { + body + Answer *body `json:"answer"` + } + if err := json.Unmarshal(raw, &out); err != nil { + return Reach{}, fmt.Errorf("the view's view_reach is not readable: %.200s", raw) + } + b := out.body + if out.Answer != nil { + b = *out.Answer + } + r := Reach{Head: b.Reached, Revision: b.ReachedSrc} + for _, m := range b.Modules { + if m.Nodes == 0 || m.Floor != "" { + continue // as the view's own notYetOf: a module a node wants and runs nothing countable of + } + for _, c := range m.Changes { + r.NotYet = append(r.NotYet, c.Commit) + } + } + sort.Strings(r.NotYet) + return r, nil +} + +// coverReached is the level-triggered half of delivery: the view says a repository's reached head once per source revision, so a +// fix whose record came after that saying would wait for ever (issue 1950, met twice). Every merged fix not delivered is matched +// against what the view says now of its repository, as OnDelivered matches an event. It answers the issues it changed. +func (t *Tracker) coverReached() error { + if t.Reached == nil { + return nil + } + all, err := t.allIssues() + if err != nil { + return err + } + reach := map[string]*Reach{} + var changed []uint64 + for n, is := range all { + for _, pr := range liveFixes(is) { + f := fixOf(is, pr) + if f != nil && f.Merged && !contains(is.Delivered, pr) && f.MergedAs == "" { + t.fillMerge(n, f, pr) + } + if f == nil || !f.Merged || contains(is.Delivered, pr) || f.MergedAs == "" || f.Repository == "" { + continue + } + key := strings.ToLower(f.Repository) + r, asked := reach[key] + if !asked { + got, err := t.Reached(key) + if err != nil { + t.Logf("[mesh-issues] what reached the machines of %s is not known yet (asked again at the next sweep): %v", key, err) + } else if got.Head != "" { + r = &got + } + reach[key] = r + } + if r == nil || notYet(f, r.NotYet) { + continue + } + if !(sameCommitIn(f.MergedAs, r.Head) || t.coveredBy(f, map[string]string{key: r.Head})) { + continue + } + if err := t.markDelivered(n, pr, fmt.Sprintf("source revision %d reached the machines", r.Revision), t.now()); err != nil { + return err + } + if len(changed) == 0 || changed[len(changed)-1] != n { + changed = append(changed, n) + } + } + } + sort.Slice(changed, func(i, j int) bool { return changed[i] < changed[j] }) + for _, n := range changed { + if err := t.afterFix(n); err != nil { + return err + } + } + return nil +} diff --git a/modules/mesh-issues/cmd/mesh-issues/learn.go b/modules/mesh-issues/cmd/mesh-issues/learn.go index f53f1765..1c39ed04 100644 --- a/modules/mesh-issues/cmd/mesh-issues/learn.go +++ b/modules/mesh-issues/cmd/mesh-issues/learn.go @@ -153,6 +153,9 @@ func (t *Tracker) LearnFixes(n uint64) (map[string]any, error) { if _, _, err := t.learnUnheard(n, is, MostSweepAsks*2); err != nil { return nil, err } + if err := t.coverReached(); err != nil { + return nil, err + } if err := t.afterFix(n); err != nil { return nil, err } diff --git a/modules/mesh-issues/cmd/mesh-issues/main.go b/modules/mesh-issues/cmd/mesh-issues/main.go index d0655a9a..1b6b8e3b 100644 --- a/modules/mesh-issues/cmd/mesh-issues/main.go +++ b/modules/mesh-issues/cmd/mesh-issues/main.go @@ -110,6 +110,7 @@ func main() { } t.Approval = busApproval(forgeOwner) t.Contains = busContains + t.Reached = busReach if len(os.Args) > 1 && os.Args[1] == "dry-run" { os.Exit(dryRunAtTheTerminal(t, os.Args[2:])) } diff --git a/modules/mesh-issues/cmd/mesh-issues/tracker.go b/modules/mesh-issues/cmd/mesh-issues/tracker.go index c125cd3d..7b80cba0 100644 --- a/modules/mesh-issues/cmd/mesh-issues/tracker.go +++ b/modules/mesh-issues/cmd/mesh-issues/tracker.go @@ -45,6 +45,9 @@ type Tracker struct { // Contains asks the forge whether a commit is contained in another (gitea_contains); nil asks nothing, and a delivered // revision then marks only the fixes it names (OnDelivered). Contains func(owner, repo, commit, in string) (bool, error) + // Reached asks the delivery view what reached the machines of one repository now (view_reach); nil asks nothing. A delivered + // event is said once per source revision, so this is the level that a fix named after it is matched to (issue 1950). + Reached func(repository string) (Reach, error) // Notify tells the operator's channel of a stuck fix (ADR 0281) and of a fix over its budget (ADR 0313); nil // says nothing there. Notify Notifier -- 2.54.0