Repository: bacalhau-project/bacalhau (language: python)

Issue:
NATS Jetstream breaks if asked to connect to self
When setting up a node using NATS, and therefore the persistent node store, Jetstream is a little picky.

If you specify a Cluster setting for NATS, but without any peers, it breaks.
If you specify a Cluster settings for NATS, and it includes the local machine as a perr, it breaks. 

NATS also breaks by timing out when trying to create the bucket, when in reality it is timing out trying to create the RAFT connection to itself. 

To fix this, we should process the cluster settings and remove any address that are the local machine.  


Discussion / hints:

Fix patch:
diff --git a/pkg/nats/transport/nats.go b/pkg/nats/transport/nats.go
index 8c8df1cdbb..151ebe8712 100644
--- a/pkg/nats/transport/nats.go
+++ b/pkg/nats/transport/nats.go
@@ -117,14 +117,15 @@ func NewNATSTransport(ctx context.Context,
 			StoreDir:               config.StoreDir,
 		}
 
-		// Only set cluster options if cluster peers are provided. If we don't Jetstream
-		// will fail to start as it is unable to locate the zero routes we provide when
-		// there are no peers.
+		// Only set cluster options if cluster peers are provided. Jetstream doesn't
+		// like the setting to be present with no values, or with values that are
+		// a local address (e.g. it can't RAFT to itself).
+		routes, err := nats_helper.RoutesFromSlice(config.ClusterPeers, false)
+		if err != nil {
+			return nil, err
+		}
+
 		if len(config.ClusterPeers) > 0 {
-			routes, err := nats_helper.RoutesFromSlice(config.ClusterPeers)
-			if err != nil {
-				return nil, err
-			}
 			serverOpts.Routes = routes
 
 			serverOpts.Cluster = server.ClusterOpts{
diff --git a/pkg/nats/util.go b/pkg/nats/util.go
index f2d8e7cb23..6592a3ccbb 100644
--- a/pkg/nats/util.go
+++ b/pkg/nats/util.go
@@ -3,11 +3,16 @@ package nats
 import (
 	"crypto/sha256"
 	"encoding/base64"
+	"net"
 	"net/url"
 	"regexp"
 	"strings"
 
+	"github.com/bacalhau-project/bacalhau/pkg/lib/network"
 	"github.com/bacalhau-project/bacalhau/pkg/system"
+	"github.com/pkg/errors"
+	"github.com/samber/lo"
+	"golang.org/x/exp/slices"
 )
 
 var schemeRegex = regexp.MustCompile(`^[a-zA-Z][a-zA-Z0-9+-.]*://`)
@@ -16,11 +21,13 @@ const defaultScheme = "nats://"
 
 // RoutesFromStr parses route URLs from a string
 // e.g. "nats://localhost:4222,nats://localhost:4223"
-func RoutesFromStr(routesStr string) ([]*url.URL, error) {
+func RoutesFromStr(routesStr string, allowLocal bool) ([]*url.URL, error) {
 	routes := strings.Split(routesStr, ",")
 	if len(routes) == 0 {
 		return nil, nil
 	}
+
+	var err error
 	var routeUrls []*url.URL
 	for _, r := range routes {
 		r = strings.TrimSpace(r)
@@ -33,15 +40,46 @@ func RoutesFromStr(routesStr string) ([]*url.URL, error) {
 		}
 		routeUrls = append(routeUrls, u)
 	}
+
+	if !allowLocal {
+		routeUrls, err = removeLocalAddresses(routeUrls)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to remove local addresses from NATS routes. please ensure settings do not contain a local address.") //nolint:lll
+		}
+	}
+
 	return routeUrls, nil
 }
 
 // RoutesFromSlice parses route URLs from a slice of strings
-func RoutesFromSlice(routes []string) ([]*url.URL, error) {
+func RoutesFromSlice(routes []string, allowLocal bool) ([]*url.URL, error) {
 	if len(routes) == 0 {
 		return []*url.URL{}, nil
 	}
-	return RoutesFromStr(strings.Join(routes, ","))
+	return RoutesFromStr(strings.Join(routes, ","), allowLocal)
+}
+
+// removeLocalAddresses removes local addresses from a list of URLs
+// and returns the result. This allows for accidental inclusion of local
+// addresses in the list of NATS routes, even when we don't want to allow
+// those local addresses (ie Jetstream clusters).
+func removeLocalAddresses(routes []*url.URL) ([]*url.URL, error) {
+	addrs, err := network.AllAddresses()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get local addresses")
+	}
+
+	localAddresses := lo.Map(addrs, func(item net.IP, _ int) string {
+		return item.String()
+	})
+
+	result := make([]*url.URL, 0, len(routes))
+	for _, u := range routes {
+		if !slices.Contains(localAddresses, u.Hostname()) {
+			result = append(result, u)
+		}
+	}
+	return result, nil
 }
 
 // CreateAuthSecret will return a signed hash of the nodeID


Test patch:
diff --git a/pkg/nats/util_test.go b/pkg/nats/util_test.go
new file mode 100644
index 0000000000..d485331386
--- /dev/null
+++ b/pkg/nats/util_test.go
@@ -0,0 +1,44 @@
+//go:build unit || !integration
+
+package nats
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type NATSUtilSuite struct {
+	suite.Suite
+	localAddresses []string
+}
+
+func TestNATSUtilSuite(t *testing.T) {
+	suite.Run(t, new(NATSUtilSuite))
+}
+
+func (s *NATSUtilSuite) Test_RoutesFromStr() {
+	routesStr := "nats://127.0.0.1:4222,nats://127.0.0.1:4223"
+	routes, err := RoutesFromStr(routesStr, true)
+	s.Require().NoError(err)
+	s.Require().Len(routes, 2)
+	s.Equal("nats://127.0.0.1:4222", routes[0].String())
+	s.Equal("nats://127.0.0.1:4223", routes[1].String())
+}
+
+func (s *NATSUtilSuite) Test_RoutesFromStrNoLocal() {
+	routesStr := "nats://127.0.0.1:4222,nats://127.0.0.1:4223"
+	routes, err := RoutesFromStr(routesStr, false)
+	s.Require().NoError(err)
+	s.Require().Len(routes, 0)
+}
+
+func (s *NATSUtilSuite) Test_RoutesFromStrMixed() {
+	// We'll use a multicast address to test here as it won't be
+	// considered local (even though it is a local multicast address).
+	routesStr := "nats://127.0.0.1:4222,nats://224.0.0.1:4223"
+	routes, err := RoutesFromStr(routesStr, false)
+	s.Require().NoError(err)
+	s.Require().Len(routes, 1)
+	s.Equal("nats://224.0.0.1:4223", routes[0].String())
+}