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 0fcca90..ba9c342 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -75,12 +75,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer conn.Close() model.SetWriter(kvWriter{ctx: ctx, js: js}) - model.SetMotionEvent(func(e view.MotionEvent) { - if err := emitEvent(ctx, js, cred, e.Type, e.Body, e.ID); err != nil { - logger.Printf("the %s of %s on %s could not be said: %v", e.Type, e.Body.Module, e.Body.Node, err) + model.SetMotionEvent(func(e view.MotionEvent) error { + err := emitEvent(ctx, js, cred, e.Type, e.Body, e.ID) + if err != nil { + logger.Printf("the %s of %s on %s could not be said (said again at its next change): %v", e.Type, e.Body.Module, e.Body.Node, err) } + return err }) - model.SetDeliveredEvent(func(e view.DeliveredEvent) { + model.SetDeliveredEvent(func(e view.DeliveredEvent) error { id := fmt.Sprintf("delivered.%d", e.Revision) if e.Repository != "" { // a repository's reached head (ADR 0556 clause 3) key, _ := delivery.RepositoryKey(e.Repository) @@ -90,14 +92,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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) + err := emitEvent(ctx, js, cred, DeliveredEventType, e, id) + if err != nil { + logger.Printf("the delivery of source revision %d could not be said (said again at the next change the view takes in): %v", e.Revision, err) } + return err }) - model.SetReachWriter(func(w view.ReachWrite) { - if err := putReach(ctx, js, w); err != nil { - logger.Printf("the reach of %s could not be written (written again at its next change): %v", w.Key, err) + model.SetReachWriter(func(w view.ReachWrite) error { + err := putReach(ctx, js, w) + if err != nil { + logger.Printf("the reach of %s could not be written (written again at the next change the view takes in): %v", w.Key, err) } + return err }) model.SetBacklog(view.BacklogOver(js, 5*time.Second)) for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket, delivery.TakenBucket} { diff --git a/modules/mesh-delivery-view/internal/view/reach.go b/modules/mesh-delivery-view/internal/view/reach.go index d055f0d..303fb5d 100644 --- a/modules/mesh-delivery-view/internal/view/reach.go +++ b/modules/mesh-delivery-view/internal/view/reach.go @@ -31,8 +31,9 @@ type ReachWrite struct { Value []byte } -// SetReachWriter sets where the view's reach values go; nothing is written until it is set. -func (m *Model) SetReachWriter(put func(ReachWrite)) { +// SetReachWriter sets where the view's reach values go; nothing is written until it is set. A value it answers an error for is +// written again at the next change the view takes in (sendReach). +func (m *Model) SetReachWriter(put func(ReachWrite) error) { m.mu.Lock() defer m.mu.Unlock() m.reachPut = put diff --git a/modules/mesh-delivery-view/internal/view/reach_test.go b/modules/mesh-delivery-view/internal/view/reach_test.go index f459efe..236a169 100644 --- a/modules/mesh-delivery-view/internal/view/reach_test.go +++ b/modules/mesh-delivery-view/internal/view/reach_test.go @@ -1,6 +1,7 @@ package view import ( + "errors" "strings" "testing" @@ -62,6 +63,7 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { 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 + refused bool // the bus refuses every write and event until the watches handed over (a start, issue 1126) }{ {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", @@ -82,16 +84,32 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { then: []Entry{taken(web, commitB)}}, {name: "a change whose source revision is not known yet holds no head the source named", reached: commitC, then: []Entry{mod("web", "src2:"+strings.Repeat("9", 64), commitX, 0, true), mod("web", "src2:"+strings.Repeat("9", 64), commitX, 0, false)}}, + // 1126: the view started while the bus did not answer its writes; its reach and its event are made at the next change it + // takes in, though the derivation is the same, so the state reach never keeps an older have than node_haves says. + {name: "a write and an event refused at the start are made at the next change", reached: commitC, refused: true, + then: []Entry{haveFP(t, "ace", "mail", fpMail, delivery.HealthOK)}}, } { t.Run(tc.name, func(t *testing.T) { m := New() var written []ReachWrite var said []DeliveredEvent - m.SetReachWriter(func(w ReachWrite) { written = append(written, w) }) - m.SetDeliveredEvent(func(e DeliveredEvent) { + refuse := tc.refused + errRefused := errors.New("nats: no responders available for request") + m.SetReachWriter(func(w ReachWrite) error { + if refuse { + return errRefused + } + written = append(written, w) + return nil + }) + m.SetDeliveredEvent(func(e DeliveredEvent) error { + if refuse { + return errRefused + } if e.Repository != "" { said = append(said, e) } + return nil }) fill(m, world(t)...) if len(written) != 0 { @@ -100,6 +118,7 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { for _, b := range reachBuckets { m.Initial(b) } + refuse = false fill(m, tc.then...) key, _ := delivery.RepositoryKey(web) r := m.Reach()[key] @@ -136,6 +155,9 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { last = e } } + if last.Repository == "" { + t.Errorf("the reached head of %s was never said", web) + } if strings.Join(last.NotYet, ",") != strings.Join(tc.notYet, ",") { t.Errorf("not yet = %v, want %v", last.NotYet, tc.notYet) } diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 2c68ba2..ccdeb3b 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -223,8 +223,8 @@ type Model struct { asks map[string]askRecord // the gate's asks, by key wants map[string]wantEntry // the assigner's wants, by . source *sourceEntry // the gate's source key - deliveredEvent func(DeliveredEvent) - motionEvent func(MotionEvent) + deliveredEvent func(DeliveredEvent) error + motionEvent func(MotionEvent) error told map[string]bool // the motion events said, by their ids initial map[string]bool // the buckets whose watch handed over what stood at its start said uint64 // the source revision whose delivery was said @@ -249,7 +249,7 @@ type Model struct { 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) + reachPut func(ReachWrite) error } type presenceEntry struct { @@ -380,24 +380,78 @@ func (m *Model) Apply(e Entry) { say, put := m.deliveredEvent, m.reachPut sayMotion := m.motionEvent m.mu.Unlock() - if event != nil && say != nil { - say(*event) + if event != nil && say != nil && say(*event) != nil { + m.unsay(*event) } + m.sendReach(put, say, writes, reached) + m.sayMotion(sayMotion, motion) +} + +// sayMotion says the motion events after the model's lock is released; one not said is forgotten as told, so the next change of +// its want, have or stuck key says it again. +func (m *Model) sayMotion(say func(MotionEvent) error, events []MotionEvent) { + if say == nil { + return + } + for _, ev := range events { + if say(ev) != nil { + m.mu.Lock() + delete(m.told, ev.ID) + m.mu.Unlock() + } + } +} + +// sendReach writes the reach values and says the reached heads, after the model's lock is released. A value not written or an +// event not said is forgotten as written or said, so the next change of any key the view watches writes or says it again +// (issue 1126: a write refused while the bus was not yet answering left the view's state reach at an older have for good, and +// mesh-delivery.show and knowledge-base.progress said a module not running that node_haves said running). +func (m *Model) sendReach(put func(ReachWrite) error, say func(DeliveredEvent) error, writes []ReachWrite, reached []DeliveredEvent) { + var unwritten []ReachWrite + var unsaid []DeliveredEvent if put != nil { for _, w := range writes { - put(w) + if put(w) != nil { + unwritten = append(unwritten, w) + } } } if say != nil { for _, ev := range reached { - say(ev) + if say(ev) != nil { + unsaid = append(unsaid, ev) + } } } - if sayMotion != nil { - for _, ev := range motion { - sayMotion(ev) + if len(unwritten) == 0 && len(unsaid) == 0 { + return + } + m.mu.Lock() + defer m.mu.Unlock() + for _, w := range unwritten { + r, err := delivery.DecodeReach(w.Value) + if prev, ok := m.reach[w.Key]; ok && err == nil && delivery.SameReach(prev, r) { + delete(m.reach, w.Key) // a newer value derived since stands as it is: its own write says whether it landed } } + for _, ev := range unsaid { + key, err := delivery.RepositoryKey(ev.Repository) + if err == nil && m.reachSaid[key] == ev.Revision && m.reachNotYet[key] == strings.Join(ev.NotYet, ",") { + m.reachNotYet[key] = unsaidMark + } + } +} + +// unsaidMark stands for a reached head's event not said: it equals no list of merges, so the next derivation says it again. +const unsaidMark = "\x00not said" + +// unsay forgets that a source revision's delivered event was said, when it could not be said and nothing newer was said since. +func (m *Model) unsay(ev DeliveredEvent) { + m.mu.Lock() + defer m.mu.Unlock() + if m.said == ev.Revision { + m.said = 0 + } } // DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR @@ -418,7 +472,7 @@ type DeliveredEvent struct { } // SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set. -func (m *Model) SetDeliveredEvent(say func(DeliveredEvent)) { +func (m *Model) SetDeliveredEvent(say func(DeliveredEvent) error) { m.mu.Lock() defer m.mu.Unlock() m.deliveredEvent = say @@ -1191,7 +1245,7 @@ type MotionEvent struct { } // SetMotionEvent sets where the events of modules in motion go; nothing is said until it is set. -func (m *Model) SetMotionEvent(say func(MotionEvent)) { +func (m *Model) SetMotionEvent(say func(MotionEvent) error) { m.mu.Lock() defer m.mu.Unlock() m.motionEvent = say @@ -1265,19 +1319,8 @@ func (m *Model) Initial(bucket string) { writes, reached := m.reachNow() put, sayDelivered := m.reachPut, m.deliveredEvent m.mu.Unlock() - for _, ev := range out { - say(ev) - } - if put != nil { - for _, w := range writes { - put(w) - } - } - if sayDelivered != nil { - for _, ev := range reached { - sayDelivered(ev) - } - } + m.sayMotion(say, out) + m.sendReach(put, sayDelivered, writes, reached) } // motionOf is the events to say now about the node and module of a model key, from their want, have and stuck key. diff --git a/modules/mesh-delivery-view/internal/view/wants_test.go b/modules/mesh-delivery-view/internal/view/wants_test.go index abc4fd4..6019daa 100644 --- a/modules/mesh-delivery-view/internal/view/wants_test.go +++ b/modules/mesh-delivery-view/internal/view/wants_test.go @@ -2,6 +2,7 @@ package view import ( "encoding/json" + "errors" "strings" "testing" @@ -167,7 +168,15 @@ func TestTheJudgementOfTheMoveIsShownBesideTheWant(t *testing.T) { func TestTheViewSaysOnceThatASourceRevisionIsDelivered(t *testing.T) { m := New() var said []DeliveredEvent - m.SetDeliveredEvent(func(e DeliveredEvent) { said = append(said, e) }) + refuse := true // the bus refuses the first try to say it (issue 1126): it is said at the next change, not lost + m.SetDeliveredEvent(func(e DeliveredEvent) error { + if refuse && e.Repository == "" { + refuse = false + return errors.New("nats: no responders available for request") + } + said = append(said, e) + return nil + }) fill(m, sourceE(t, 14355), moduleE(t, "web", fpNew), assign("ace", "web", "b-1", ""), wantE(t, "ace", "web", delivery.Want{FP: fpNew})) if len(said) != 0 { @@ -178,6 +187,10 @@ func TestTheViewSaysOnceThatASourceRevisionIsDelivered(t *testing.T) { t.Fatalf("said %v for a failed have", said) } fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK)) + if len(said) != 0 || refuse { + t.Fatalf("said = %+v though the bus refused it", said) + } + fill(m, wantE(t, "ace", "web", delivery.Want{FP: fpNew})) if len(said) != 1 || said[0].Revision != 14355 || said[0].Group != "g" || said[0].Heads["novox/web"] == "" { t.Fatalf("said = %+v, want revision 14355 once with its group and heads", said) } @@ -234,7 +247,7 @@ func TestADottedModuleNameIsFoundAndKept(t *testing.T) { // is held or failed or the watchman says it stopped moving; each once for a node, a module and a composition. func motionLog(m *Model) *[]MotionEvent { var said []MotionEvent - m.SetMotionEvent(func(e MotionEvent) { said = append(said, e) }) + m.SetMotionEvent(func(e MotionEvent) error { said = append(said, e); return nil }) for _, b := range []string{delivery.AssignerWantsBucket, delivery.HaveBucket, delivery.StuckBucket} { m.Initial(b) } @@ -269,6 +282,20 @@ func TestAModuleInMotionIsSaidMovingThenSettledOnceEach(t *testing.T) { if types(*said) != "moving" { t.Errorf("said %q after the same state again", types(*said)) } + // 1126: the bus refuses the settled once; it is said at the next change of the have, not lost. + refused := false + m.SetMotionEvent(func(e MotionEvent) error { + if !refused { + refused = true + return errors.New("nats: no responders available for request") + } + *said = append(*said, e) + return nil + }) + fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK)) + if types(*said) != "moving" || !refused { + t.Errorf("said %q though the bus refused it", types(*said)) + } fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK)) if types(*said) != "moving,settled" || (*said)[1].Body.Outcome != "running" { t.Errorf("said %q, want settled with its outcome", types(*said)) @@ -330,7 +357,7 @@ func TestANodeThatSettlesWithoutBeingSeenMovingSaysNothing(t *testing.T) { func TestMotionIsSaidOnlyOnceTheBucketsHaveBeenRead(t *testing.T) { m := New() var said []MotionEvent - m.SetMotionEvent(func(e MotionEvent) { said = append(said, e) }) + m.SetMotionEvent(func(e MotionEvent) error { said = append(said, e); return nil }) fill(m, wantE(t, "ace", "web", delivery.Want{FP: fpNew}), wantE(t, "ace", "mail", delivery.Want{FP: fpNew})) m.Initial(delivery.AssignerWantsBucket) fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK), haveFP(t, "ace", "mail", fpOld, delivery.HealthOK))