Repository: liftbridge-io/liftbridge
Pull request #77 (2019-07-03 23:19:36)

Fail fast on metadata request propagate
Fail fast if there is no known metadata leader when propagating metadata
requests such as CreateStream.


Patch:
diff --git a/go.mod b/go.mod
index 89502df..6f8e838 100644
--- a/go.mod
+++ b/go.mod
@@ -12,11 +12,13 @@ require (
 	github.com/liftbridge-io/go-liftbridge v0.0.0-20190703015712-9f8cb1ad3118
 	github.com/liftbridge-io/nats-on-a-log v0.0.0-20190703144237-760cefbfc85e
 	github.com/natefinch/atomic v0.0.0-20150920032501-a62ce929ffcc
+	github.com/nats-io/nats-server/v2 v2.0.0
 	github.com/nats-io/nats.go v1.8.1
 	github.com/nats-io/nuid v1.0.1
 	github.com/nsip/gommap v0.0.0-20181229045655-f7881c3a959f
 	github.com/pkg/errors v0.8.1
 	github.com/sirupsen/logrus v1.4.2
+	github.com/stretchr/testify v1.3.0
 	github.com/urfave/cli v1.20.0
 	golang.org/x/net v0.0.0-20190628185345-da137c7871d7
 	google.golang.org/grpc v1.22.0
diff --git a/go.sum b/go.sum
index 66f306a..fe74a7b 100644
--- a/go.sum
+++ b/go.sum
@@ -78,6 +78,7 @@ github.com/nats-io/go-nats v1.7.0 h1:oQOfHcLr8hb43QG8yeVyY2jtarIaTjOv41CGdF3tTvQ
 github.com/nats-io/go-nats v1.7.0/go.mod h1:+t7RHT5ApZebkrQdnn6AhQJmhJJiKAvJUio1PiiCtj0=
 github.com/nats-io/jwt v0.2.6 h1:eAyoYvGgGLXR2EpnsBUvi/FcFrBqN6YKFVbOoEfPN4k=
 github.com/nats-io/jwt v0.2.6/go.mod h1:mQxQ0uHQ9FhEVPIcTSKwx2lqZEpXWWcCgA7R6NrWvvY=
+github.com/nats-io/nats-server v1.4.1 h1:Ul1oSOGNV/L8kjr4v6l2f9Yet6WY+LevH1/7cRZ/qyA=
 github.com/nats-io/nats-server v1.4.1/go.mod h1:c8f/fHd2B6Hgms3LtCaI7y6pC4WD1f4SUxcCud5vhBc=
 github.com/nats-io/nats-server/v2 v2.0.0 h1:rbFV7gfUPErVdKImVMOlW8Qb1V22nlcpqup5cb9rYa8=
 github.com/nats-io/nats-server/v2 v2.0.0/go.mod h1:RyVdsHHvY4B6c9pWG+uRLpZ0h0XsqiuKp2XCTurP5LI=
@@ -110,6 +111,7 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
 github.com/urfave/cli v1.20.0 h1:fDqGv3UG/4jbVl/QkFwEdddtEDjh/5Ov6X+0B/3bPaw=
diff --git a/server/metadata.go b/server/metadata.go
index 5a3d121..7474fc0 100644
--- a/server/metadata.go
+++ b/server/metadata.go
@@ -667,6 +667,11 @@ func (m *metadataAPI) propagateReportLeader(ctx context.Context, req *proto.Repo
 // propagateRequest forwards a metadata request to the metadata leader and
 // returns the response.
 func (m *metadataAPI) propagateRequest(ctx context.Context, req *proto.PropagatedRequest) *status.Status {
+	// Fail fast if there is no known metadata leader currently.
+	if m.raft.Leader() == "" {
+		return status.New(codes.Internal, "No known metadata leader")
+	}
+
 	data, err := req.Marshal()
 	if err != nil {
 		panic(err)


Test patch:
diff --git a/server/api_test.go b/server/api_test.go
index b1e3ffe..28f7dda 100644
--- a/server/api_test.go
+++ b/server/api_test.go
@@ -12,6 +12,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 type message struct {
@@ -88,6 +90,42 @@ func TestCreateStreamPropagate(t *testing.T) {
 	require.Equal(t, lift.ErrStreamExists, err)
 }
 
+// Ensure creating a stream fails when there is no known metadata leader.
+func TestCreateStreamNoMetadataLeader(t *testing.T) {
+	defer cleanupStorage(t)
+
+	// Use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Configure first server.
+	s1Config := getTestConfig("a", true, 0)
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	// Configure second server.
+	s2Config := getTestConfig("b", false, 5050)
+	s2 := runServerWithConfig(t, s2Config)
+	defer s2.Stop()
+
+	// Wait for a leader to be elected to allow the cluster to form, then stop
+	// a server and wait for the leader to step down.
+	getMetadataLeader(t, 10*time.Second, s1, s2)
+	s1.Stop()
+	waitForNoMetadataLeader(t, 10*time.Second, s1, s2)
+
+	// Connect and send the request to the follower.
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+
+	err = client.CreateStream(context.Background(), "foo", "foo")
+	require.Error(t, err)
+	st := status.Convert(err)
+	require.Equal(t, "No known metadata leader", st.Message())
+	require.Equal(t, codes.Internal, st.Code())
+}
+
 // Ensure creating a stream fails when the replication factor is greater than
 // the number of servers in the cluster.
 func TestCreateStreamInsufficientReplicas(t *testing.T) {
diff --git a/server/server_test.go b/server/server_test.go
index 2965d43..bcc9fcc 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -82,6 +82,24 @@ func getMetadataLeader(t *testing.T, timeout time.Duration, servers ...*Server)
 	return leader
 }
 
+func waitForNoMetadataLeader(t *testing.T, timeout time.Duration, servers ...*Server) {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		var leader string
+		for _, s := range servers {
+			if l := string(s.raft.Leader()); l != "" {
+				leader = l
+				break
+			}
+		}
+		if leader == "" {
+			return
+		}
+		time.Sleep(15 * time.Millisecond)
+	}
+	stackFatalf(t, "Metadata leader found")
+}
+
 func getStreamLeader(t *testing.T, timeout time.Duration, subject, name string, servers ...*Server) *Server {
 	var (
 		leader   *Server