The view writes a reach or says an event again when the bus refused it, so a refused write never leaves the reach state at an older have (issue 1126)

The view remembered a reach value, a reached head and a motion event as written or said before the bus took them; a write refused at its start (no responders) was never made again while the derivation stayed the same, and mesh-delivery.show said mesh-delivery-view not running on shanks while node_haves said it running.

Quick fix: forget only the refused reach write. Thorough fix, chosen: every write and event the view makes after its lock (reach, reached head, delivered revision, motion) answers its error, and the model forgets what was refused, so the next change it takes in makes it again; one place, the writer of the state both mesh-delivery.show and knowledge-base.progress read.

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