Issue a verified sender again only over the membership that was read, so a newer push is not overwritten (issue 930)

cmd/mesh-controller/reissue_verified_sender.go

@@ -128,10 +128,11 @@ func withVerifiedSenders(issued []byte, composed *broker.SeatTraffic) ([]byte, b
 }
 
 // reissueVerifiedSenders issues again each membership that lacks a verified sender the records now give it:
-// read as last issued, its kinds replaced, published. One never issued is left to its own send. Failures are
+// read as last issued, its kinds replaced, published only if no newer one was issued since (a push runs beside
+// the self-check, and its newer membership must not be overwritten by the older fields read here). One never issued is left to its own send. Failures are
 // said and passed over; the next self-check tries again.
 func reissueVerifiedSenders(r broker.Records, where broker.Placements,
-	last func(node, module string) ([]byte, error), publish func(node, module string, body []byte) error,
+	last func(node, module string) ([]byte, uint64, error), publish func(node, module string, body []byte, seq uint64) error,
 	say func(format string, args ...any)) int {
 	issued := 0
 	for node, declared := range r.Assigned {
@@ -142,7 +143,7 @@ func reissueVerifiedSenders(r broker.Records, where broker.Placements,
 			}) {
 				continue
 			}
-			was, err := last(node, d.Module)
+			was, seq, err := last(node, d.Module)
 			if err != nil {
 				if !errors.Is(err, nats.ErrMsgNotFound) {
 					say("  %s on %s: its membership could not be read to give its verified sender back: %v", d.Module, node, err)
@@ -157,7 +158,13 @@ func reissueVerifiedSenders(r broker.Records, where broker.Placements,
 			if !changed {
 				continue
 			}
-			if err := publish(node, d.Module, body); err != nil {
+			if err := publish(node, d.Module, body, seq); err != nil {
+				// A push issued another since it was read: that one is newer, and the next self-check looks again.
+				if strings.Contains(err.Error(), "wrong last sequence") {
+					say("  %s on %s: a newer membership was issued meanwhile; its verified sender is looked at again next time",
+						d.Module, node)
+					continue
+				}
 				say("  %s on %s: its verified sender could not be issued again: %v", d.Module, node, err)
 				continue
 			}
@@ -203,13 +210,15 @@ func giveVerifiedSendersBack(ctx context.Context, d *doctor) {
 // published to it.
 func reissueVerifiedSendersOn(ctx context.Context, bus link.OverNATS, records broker.Records, where broker.Placements) int {
 	return reissueVerifiedSenders(records, where,
-		func(node, module string) ([]byte, error) {
+		func(node, module string) ([]byte, uint64, error) {
 			m, err := bus.JS.GetLastMsg(broker.AssignmentsStream, broker.MembershipSubject(node, module))
 			if err != nil {
-				return nil, err
+				return nil, 0, err
 			}
-			return m.Data, nil
+			return m.Data, m.Sequence, nil
+		},
+		func(node, module string, body []byte, seq uint64) error {
+			return bus.PublishMembershipAfter(ctx, node, module, body, seq)
 		},
-		func(node, module string, body []byte) error { return bus.PublishMembership(ctx, node, module, body) },
 		func(format string, args ...any) { fmt.Printf(format+"\n", args...) })
 }

cmd/mesh-controller/reissue_verified_sender_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"strings"
 	"testing"
 
@@ -25,17 +26,28 @@ func verifiedSenderRecords(free map[string]bool) (broker.Records, broker.Declare
 type fakeBus struct {
 	last      map[string][]byte
 	published map[string][]byte
+	seq       map[string]uint64
+	// between runs once, after a read and before the write: a push issuing meanwhile.
+	between func()
 }
 
-func (f *fakeBus) get(node, module string) ([]byte, error) {
+func (f *fakeBus) get(node, module string) ([]byte, uint64, error) {
 	if b, ok := f.last[node+"/"+module]; ok {
-		return b, nil
+		return b, f.seq[node+"/"+module], nil
 	}
-	return nil, nats.ErrMsgNotFound
+	return nil, 0, nats.ErrMsgNotFound
 }
-func (f *fakeBus) put(node, module string, body []byte) error {
+func (f *fakeBus) put(node, module string, body []byte, seq uint64) error {
+	if f.between != nil {
+		f.between()
+		f.between = nil
+	}
+	if seq != f.seq[node+"/"+module] {
+		return errors.New("nats: wrong last sequence")
+	}
 	f.published[node+"/"+module] = body
 	f.last[node+"/"+module] = body
+	f.seq[node+"/"+module]++
 	return nil
 }
 
@@ -59,7 +71,7 @@ func hasVerified(t *testing.T, membership []byte) bool {
 func TestAMembershipThatLostItsVerifiedSenderGetsItBackWhenTheJudgementPasses(t *testing.T) {
 	failed, router := verifiedSenderRecords(nil)
 	whereFailed := broker.PlacementsOf(failed, nil)
-	bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}}
+	bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
 	body, _ := json.Marshal(broker.MembershipFor("novox", router, whereFailed))
 	bus.last["novox/messenger"] = body
 	if hasVerified(t, body) {
@@ -100,7 +112,7 @@ func TestADroppedVerifiedSenderIsSaidWithItsReason(t *testing.T) {
 
 func TestAMembershipNeverIssuedIsLeftToItsOwnSend(t *testing.T) {
 	records, _ := verifiedSenderRecords(map[string]bool{"novox": true})
-	bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}}
+	bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
 	var said []string
 	n := reissueVerifiedSenders(records, broker.PlacementsOf(records, nil), bus.get, bus.put,
 		func(f string, a ...any) { said = append(said, f) })
@@ -156,3 +168,20 @@ func TestNatsAMembershipIssuedWithoutItsVerifiedSenderIsIssuedAgainWhenTheJudgem
 		t.Fatalf("a second pass issued %d memberships", n)
 	}
 }
+
+// A push issuing a newer membership between the self-check's read and its write is not overwritten (issue 930).
+func TestANewerMembershipIssuedMeanwhileIsNotOverwritten(t *testing.T) {
+	failed, router := verifiedSenderRecords(nil)
+	bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}}
+	old, _ := json.Marshal(broker.MembershipFor("novox", router, broker.PlacementsOf(failed, nil)))
+	bus.last["novox/messenger"], bus.seq["novox/messenger"] = old, 1
+	newer := []byte(`{"node":"novox","module":"messenger","newer":true}`)
+	bus.between = func() { bus.last["novox/messenger"], bus.seq["novox/messenger"] = newer, 2 }
+	passing, _ := verifiedSenderRecords(map[string]bool{"novox": true})
+	if n := reissueVerifiedSenders(passing, broker.PlacementsOf(passing, nil), bus.get, bus.put, func(string, ...any) {}); n != 0 {
+		t.Fatalf("issued %d over a newer membership", n)
+	}
+	if string(bus.last["novox/messenger"]) != string(newer) || len(bus.published) != 0 {
+		t.Fatalf("the newer membership was overwritten: %s", bus.last["novox/messenger"])
+	}
+}

internal/link/bus.go

@@ -179,6 +179,19 @@ func (b OverNATS) PublishMembership(ctx context.Context, node, module string, bo
 	return nil
 }
 
+// PublishMembershipAfter issues a membership only if the one read at sequence seq is still the newest of its
+// subject: a push that issued another since is not overwritten, and the refusal comes back (novox/hq issue 930).
+func (b OverNATS) PublishMembershipAfter(ctx context.Context, node, module string, body []byte, seq uint64) error {
+	ctx, cancel := context.WithTimeout(ctx, MembershipWait)
+	defer cancel()
+	_, err := b.JS.Publish(broker.MembershipSubject(node, module), body, nats.Context(ctx),
+		nats.ExpectLastSequencePerSubject(seq))
+	if err != nil {
+		return fmt.Errorf("issuing %s on %s its membership after sequence %d: %w", module, node, seq, err)
+	}
+	return nil
+}
+
 func (b OverNATS) PublishDeclaration(ctx context.Context, node string, body []byte) error {
 	_, err := b.JS.Publish(DeclareSubject(node), body, nats.Context(ctx))
 	if err != nil {