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