A module a node wants and runs nothing countable of holds no head back; the reached head's event names its merges as not yet on the machines; pin mesh-sdk e44cbaf (review of issue 1280, ADR 0556 clause 2)
This commit is contained in:
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user