diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 3706520..1f257b4 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index a889409..73a8b6b 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652 h1:R+gkczO84KN1nbTG3ubIfAAkbx/jTb+txW1fQVhtmSI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3 h1:kOAR8GaTpOP9j2XNm3grqEfSwSDHnWbHKd+Qftkr/QE= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index 27ba1a0..6cf646c 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-teller go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-teller/go.sum b/modules/mesh-delivery-teller/go.sum index a889409..73a8b6b 100644 --- a/modules/mesh-delivery-teller/go.sum +++ b/modules/mesh-delivery-teller/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652 h1:R+gkczO84KN1nbTG3ubIfAAkbx/jTb+txW1fQVhtmSI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3 h1:kOAR8GaTpOP9j2XNm3grqEfSwSDHnWbHKd+Qftkr/QE= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index d8e64e6..aa41f4f 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -147,7 +147,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro id := fmt.Sprintf("delivered.%d", e.Revision) if e.Repository != "" { // a repository's reached head (ADR 0556 clause 3) key, _ := delivery.RepositoryKey(e.Repository) - id = fmt.Sprintf("reached.%s.%d", key, e.Revision) + id = fmt.Sprintf("reached.%s.%d.%d", key, e.Revision, len(e.NotYet)) + if len(e.NotYet) > 0 { + sum := sha256.Sum256([]byte(strings.Join(e.NotYet, ","))) + id += "." + hex.EncodeToString(sum[:6]) + } } if err := emitEvent(ctx, js, cred, DeliveredEventType, e, id); err != nil { logger.Printf("the delivery of source revision %d could not be said: %v", e.Revision, err) diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index 5c13add..dcc49d1 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-view go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-view/go.sum b/modules/mesh-delivery-view/go.sum index a889409..73a8b6b 100644 --- a/modules/mesh-delivery-view/go.sum +++ b/modules/mesh-delivery-view/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652 h1:R+gkczO84KN1nbTG3ubIfAAkbx/jTb+txW1fQVhtmSI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026195749-33144781d652/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3 h1:kOAR8GaTpOP9j2XNm3grqEfSwSDHnWbHKd+Qftkr/QE= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026201248-e44cbaf57fc3/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-view/internal/view/reach.go b/modules/mesh-delivery-view/internal/view/reach.go index 231bf6e..d055f0d 100644 --- a/modules/mesh-delivery-view/internal/view/reach.go +++ b/modules/mesh-delivery-view/internal/view/reach.go @@ -241,7 +241,11 @@ func (m *Model) moduleReach(name string, mod delivery.Module, heads []headAt, wa case mr.Nodes == 0: return mr, noBound, "" case unset != "": - return mr, 0, unset + // A node that wants the module runs nothing countable of it (no have, not running, or a build older than the kept + // history): the module holds no head of the repository back, which would make every apply anywhere, or one module + // stuck failed, freeze the whole repository (review of issue 1280, ADR 0556 clause 2). Its words name it, and a merge + // among its changes is not yet on the machines (notYet). + return mr, noBound, "" } mr.Floor = mr.Changes[floor].Commit if floor == 0 { @@ -307,15 +311,34 @@ func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) { writes = append(writes, ReachWrite{Key: key, Value: data}) } } - if r.Reached != "" && r.ReachedSrc > m.reachSaid[key] { - m.reachSaid[key] = r.ReachedSrc + notYet := notYetOf(r) + said := strings.Join(notYet, ",") + if r.Reached != "" && (r.ReachedSrc > m.reachSaid[key] || said != m.reachNotYet[key]) { + m.reachSaid[key], m.reachNotYet[key] = r.ReachedSrc, said events = append(events, DeliveredEvent{Revision: r.ReachedSrc, Heads: map[string]string{r.Repository: r.Reached}, - Repository: r.Repository}) + Repository: r.Repository, NotYet: notYet}) } } return writes, events } +// notYetOf are the merges at or before a repository's reached head that are not yet on the machines: the changes of each module +// a node wants with no floor (a node runs nothing countable of it), sorted. A consumer takes every other merge at or before +// the reached head as on the machines. +func notYetOf(r delivery.Reach) []string { + var out []string + for _, mr := range r.Modules { + if mr.Nodes == 0 || mr.Floor != "" { + continue + } + for _, c := range mr.Changes { + out = append(out, c.Commit) + } + } + slices.Sort(out) + return slices.Compact(out) +} + // Reach is the reach of every repository as the view derived it last, by key. func (m *Model) Reach() map[string]delivery.Reach { m.mu.Lock() diff --git a/modules/mesh-delivery-view/internal/view/reach_test.go b/modules/mesh-delivery-view/internal/view/reach_test.go index bc6db3c..f459efe 100644 --- a/modules/mesh-delivery-view/internal/view/reach_test.go +++ b/modules/mesh-delivery-view/internal/view/reach_test.go @@ -58,9 +58,10 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { for _, tc := range []struct { name string then []Entry - reached string // web's reached head; "" none - holds string // words one of web's holds carries; "" none - waiting string // words web's module web waits with; "" none + reached string // web's reached head; "" none + holds string // words one of web's holds carries; "" none + waiting string // words web's module web waits with; "" none + notYet []string // the merges the reached head's event says are not yet on the machines }{ {name: "every node runs the newest build", reached: commitC}, {name: "a node lagging stops the head before the change it lacks", reached: commitB, holds: "web on g14", @@ -69,8 +70,12 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { then: []Entry{wantE(t, "g14", "web", delivery.Want{FP: fpOld, Newest: fpNew, Rule: 2}), haveFP(t, "g14", "web", fpOld, delivery.HealthOK)}}, {name: "a failing health check is said and holds nothing", reached: commitC, waiting: "health check failed", then: []Entry{haveFP(t, "g14", "web", fpNew, delivery.HealthFailed)}}, - {name: "a node with no running have leaves no reached head", reached: "", holds: "not running", waiting: "not running", - then: []Entry{haveIn("g14", "web", fpNew, delivery.PhaseApplying, delivery.HealthUnknown)}}, + {name: "a node with no running have holds no head back and its module's merges are not yet", reached: commitC, waiting: "not running", + notYet: []string{commitA, commitC}, + then: []Entry{haveIn("g14", "web", fpNew, delivery.PhaseApplying, delivery.HealthUnknown)}}, + {name: "an unrelated module not running leaves the head reached (the reviewer's probe)", reached: commitC, + notYet: []string{commitA}, + then: []Entry{haveIn("ace", "mail", fpMail, delivery.PhaseFailed, delivery.HealthFailed)}}, {name: "a module no node wants holds nothing", reached: commitC, then: []Entry{wantE(t, "ace", "mail", delivery.Want{Removed: true})}}, {name: "the indexer not yet having taken the head in", reached: commitB, holds: "has not yet taken in", @@ -124,15 +129,27 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { if !delivery.SameReach(last[key], r) { t.Errorf("the last value written is %+v, not the reach derived %+v", last[key], r) } - n := len(said) - fill(m, tc.then...) - fill(m, world(t)[len(world(t))-1]) - if len(said) != n && tc.reached == commitC { - t.Errorf("said again: %+v", said[n:]) + if tc.reached != "" { + var last DeliveredEvent + for _, e := range said { + if e.Repository == web { + last = e + } + } + if strings.Join(last.NotYet, ",") != strings.Join(tc.notYet, ",") { + t.Errorf("not yet = %v, want %v", last.NotYet, tc.notYet) + } } - for _, e := range said { - if len(e.Heads) != 1 || e.Group != "" || e.Revision == 0 { - t.Errorf("a reached head said as %+v", e) + { + n := len(said) + fill(m, tc.then...) + if len(said) != n && tc.reached == commitC { + t.Errorf("said again: %+v", said[n:]) + } + for _, e := range said { + if len(e.Heads) != 1 || e.Group != "" || e.Revision == 0 { + t.Errorf("a reached head said as %+v", e) + } } } }) diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 73795b8..2c68ba2 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -243,12 +243,13 @@ type Model struct { presence map[string]presenceEntry stuck map[string]stuckEntry // how far each repository reached (reach.go, ADR 0556) - modHist map[string][]delivery.Module // the indexer's values of each module, oldest first - srcHist map[uint64]map[string]string // the source's heads, by revision - taken map[string]delivery.Taken // the indexer's taken, by repository key - reach map[string]delivery.Reach // the reach derived last, by repository key - reachSaid map[string]uint64 // the reached head's source revision said last, by repository key - reachPut func(ReachWrite) + modHist map[string][]delivery.Module // the indexer's values of each module, oldest first + srcHist map[uint64]map[string]string // the source's heads, by revision + taken map[string]delivery.Taken // the indexer's taken, by repository key + reach map[string]delivery.Reach // the reach derived last, by repository key + reachSaid map[string]uint64 // the reached head's source revision said last, by repository key + reachNotYet map[string]string // the merges said not yet on the machines with it, by repository key + reachPut func(ReachWrite) } type presenceEntry struct { @@ -359,7 +360,7 @@ func New() *Model { haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{}, presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}, modHist: map[string][]delivery.Module{}, srcHist: map[uint64]map[string]string{}, taken: map[string]delivery.Taken{}, - reach: map[string]delivery.Reach{}, reachSaid: map[string]uint64{}} + reach: map[string]delivery.Reach{}, reachSaid: map[string]uint64{}, reachNotYet: map[string]string{}} } // Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. When the change makes @@ -411,6 +412,9 @@ type DeliveredEvent struct { Group string `json:"group"` Heads map[string]string `json:"heads"` Repository string `json:"repository,omitempty"` + // NotYet are, in the per-repository form, the merges at or before the reached head that are not yet on the machines: the + // changes of a module a node wants and runs nothing countable of. Said again when they change. + NotYet []string `json:"not_yet,omitempty"` } // SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set.