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)
mesh/delivery merged: source.main moved to novox/mesh-catalog@9bd824791509
mesh/check being carried to 116751130f5c on 9bd824791509 | from c05aa5e6eafd

This commit is contained in:
2026-10-27 21:17:53 +01:00
parent c05aa5e6ea
commit 116751130f
5 changed files with 205 additions and 10 deletions
@@ -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
+128 -10
View File
@@ -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
}
@@ -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
}
@@ -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:]))
}
@@ -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