Merge pull request 'The delivery view forgets a write or event the bus refused, so 'on the machines' is not left stale (issue 1126)' (#97) from fix/1126-reach-now into main

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} {

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

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)
 				}

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 <node>.<module>
 	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.

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))