@@ -114,7 +114,7 @@ github.com/dmcgowan/go-tar go1.10

 github.com/stevvooe/ttrpc 76e68349ad9ab4d03d764c713826d31216715e4f

 # cluster

-github.com/docker/swarmkit de950a7ed842c7b7e47e9451cde9bf8f96031894

+github.com/docker/swarmkit 4429c763170d9ca96929249353c3270c19e7d39e

 github.com/gogo/protobuf v0.4

 github.com/cloudflare/cfssl 7fb22c8cba7ecaf98e4082d22d65800cf45e042a

 github.com/google/certificate-transparency d90e65c3a07988180c5b1ece71791c0b6506826e

@@ -288,7 +288,9 @@ func Do(ctx context.Context, task *api.Task, ctlr Controller) (*api.TaskStatus,

 		status.PortStatus = portStatus

 	}()

-	if task.DesiredState == api.TaskStateShutdown {

+	// this branch bounds the largest state achievable in the agent as SHUTDOWN, which

+	// is exactly the correct behavior for the agent.

+	if task.DesiredState >= api.TaskStateShutdown {

 		if status.State >= api.TaskStateCompleted {

 			return noop()

 		}

@@ -131,6 +131,8 @@

 		LeaveResponse

 		ProcessRaftMessageRequest

 		ProcessRaftMessageResponse

+		StreamRaftMessageRequest

+		StreamRaftMessageResponse

 		ResolveAddressRequest

 		ResolveAddressResponse

 		InternalRaftRequest

@@ -235,6 +237,7 @@ import (

 import raftselector "github.com/docker/swarmkit/manager/raftselector"

 import codes "google.golang.org/grpc/codes"

+import status "google.golang.org/grpc/status"

 import metadata "google.golang.org/grpc/metadata"

 import transport "google.golang.org/grpc/transport"

 import rafttime "time"

@@ -984,12 +987,12 @@ func NewRaftProxyCAServer(local CAServer, connSelector raftselector.ConnProvider

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -1126,12 +1129,12 @@ func NewRaftProxyNodeCAServer(local NodeCAServer, connSelector raftselector.Conn

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -19,6 +19,7 @@ import (

 import raftselector "github.com/docker/swarmkit/manager/raftselector"

 import codes "google.golang.org/grpc/codes"

+import status "google.golang.org/grpc/status"

 import metadata "google.golang.org/grpc/metadata"

 import transport "google.golang.org/grpc/transport"

 import rafttime "time"

@@ -5859,12 +5860,12 @@ func NewRaftProxyControlServer(local ControlServer, connSelector raftselector.Co

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -24,6 +24,7 @@ import github_com_gogo_protobuf_types "github.com/gogo/protobuf/types"

 import raftselector "github.com/docker/swarmkit/manager/raftselector"

 import codes "google.golang.org/grpc/codes"

+import status "google.golang.org/grpc/status"

 import metadata "google.golang.org/grpc/metadata"

 import transport "google.golang.org/grpc/transport"

 import rafttime "time"

@@ -1602,12 +1603,12 @@ func NewRaftProxyDispatcherServer(local DispatcherServer, connSelector raftselec

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -17,6 +17,7 @@ import (

 import raftselector "github.com/docker/swarmkit/manager/raftselector"

 import codes "google.golang.org/grpc/codes"

+import status "google.golang.org/grpc/status"

 import metadata "google.golang.org/grpc/metadata"

 import transport "google.golang.org/grpc/transport"

 import rafttime "time"

@@ -286,12 +287,12 @@ func NewRaftProxyHealthServer(local HealthServer, connSelector raftselector.Conn

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -20,6 +20,7 @@ import (

 import raftselector "github.com/docker/swarmkit/manager/raftselector"

 import codes "google.golang.org/grpc/codes"

+import status "google.golang.org/grpc/status"

 import metadata "google.golang.org/grpc/metadata"

 import transport "google.golang.org/grpc/transport"

 import rafttime "time"

@@ -1273,12 +1274,12 @@ func NewRaftProxyLogsServer(local LogsServer, connSelector raftselector.ConnProv

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -1396,12 +1397,12 @@ func NewRaftProxyLogBrokerServer(local LogBrokerServer, connSelector raftselecto

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -21,6 +21,7 @@ import (

 import raftselector "github.com/docker/swarmkit/manager/raftselector"

 import codes "google.golang.org/grpc/codes"

+import status "google.golang.org/grpc/status"

 import metadata "google.golang.org/grpc/metadata"

 import transport "google.golang.org/grpc/transport"

 import rafttime "time"

@@ -133,6 +134,23 @@ func (m *ProcessRaftMessageResponse) Reset()                    { *m = ProcessRa

 func (*ProcessRaftMessageResponse) ProtoMessage()               {}

 func (*ProcessRaftMessageResponse) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{6} }

+// Raft message streaming request.

+type StreamRaftMessageRequest struct {

+	Message *raftpb.Message `protobuf:"bytes,1,opt,name=message" json:"message,omitempty"`

+}

+

+func (m *StreamRaftMessageRequest) Reset()                    { *m = StreamRaftMessageRequest{} }

+func (*StreamRaftMessageRequest) ProtoMessage()               {}

+func (*StreamRaftMessageRequest) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{7} }

+

+// Raft message streaming response.

+type StreamRaftMessageResponse struct {

+}

+

+func (m *StreamRaftMessageResponse) Reset()                    { *m = StreamRaftMessageResponse{} }

+func (*StreamRaftMessageResponse) ProtoMessage()               {}

+func (*StreamRaftMessageResponse) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{8} }

+

 type ResolveAddressRequest struct {

 	// raft_id is the ID to resolve to an address.

 	RaftID uint64 `protobuf:"varint,1,opt,name=raft_id,json=raftId,proto3" json:"raft_id,omitempty"`

@@ -140,7 +158,7 @@ type ResolveAddressRequest struct {

 func (m *ResolveAddressRequest) Reset()                    { *m = ResolveAddressRequest{} }

 func (*ResolveAddressRequest) ProtoMessage()               {}

-func (*ResolveAddressRequest) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{7} }

+func (*ResolveAddressRequest) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{9} }

 type ResolveAddressResponse struct {

 	// Addr specifies the address of the member

@@ -149,7 +167,7 @@ type ResolveAddressResponse struct {

 func (m *ResolveAddressResponse) Reset()                    { *m = ResolveAddressResponse{} }

 func (*ResolveAddressResponse) ProtoMessage()               {}

-func (*ResolveAddressResponse) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{8} }

+func (*ResolveAddressResponse) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{10} }

 // Contains one of many protobuf encoded objects to replicate

 // over the raft backend with a request ID to track when the

@@ -161,7 +179,7 @@ type InternalRaftRequest struct {

 func (m *InternalRaftRequest) Reset()                    { *m = InternalRaftRequest{} }

 func (*InternalRaftRequest) ProtoMessage()               {}

-func (*InternalRaftRequest) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{9} }

+func (*InternalRaftRequest) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{11} }

 // StoreAction defines a target and operation to apply on the storage system.

 type StoreAction struct {

@@ -181,7 +199,7 @@ type StoreAction struct {

 func (m *StoreAction) Reset()                    { *m = StoreAction{} }

 func (*StoreAction) ProtoMessage()               {}

-func (*StoreAction) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{10} }

+func (*StoreAction) Descriptor() ([]byte, []int) { return fileDescriptorRaft, []int{12} }

 type isStoreAction_Target interface {

 	isStoreAction_Target()

@@ -512,6 +530,8 @@ func init() {

 	proto.RegisterType((*LeaveResponse)(nil), "docker.swarmkit.v1.LeaveResponse")

 	proto.RegisterType((*ProcessRaftMessageRequest)(nil), "docker.swarmkit.v1.ProcessRaftMessageRequest")

 	proto.RegisterType((*ProcessRaftMessageResponse)(nil), "docker.swarmkit.v1.ProcessRaftMessageResponse")

+	proto.RegisterType((*StreamRaftMessageRequest)(nil), "docker.swarmkit.v1.StreamRaftMessageRequest")

+	proto.RegisterType((*StreamRaftMessageResponse)(nil), "docker.swarmkit.v1.StreamRaftMessageResponse")

 	proto.RegisterType((*ResolveAddressRequest)(nil), "docker.swarmkit.v1.ResolveAddressRequest")

 	proto.RegisterType((*ResolveAddressResponse)(nil), "docker.swarmkit.v1.ResolveAddressResponse")

 	proto.RegisterType((*InternalRaftRequest)(nil), "docker.swarmkit.v1.InternalRaftRequest")

@@ -539,6 +559,14 @@ func (p *authenticatedWrapperRaftServer) ProcessRaftMessage(ctx context.Context,

 	return p.local.ProcessRaftMessage(ctx, r)

 }

+func (p *authenticatedWrapperRaftServer) StreamRaftMessage(stream Raft_StreamRaftMessageServer) error {

+

+	if err := p.authorize(stream.Context(), []string{"swarm-manager"}); err != nil {

+		return err

+	}

+	return p.local.StreamRaftMessage(stream)

+}

+

 func (p *authenticatedWrapperRaftServer) ResolveAddress(ctx context.Context, r *ResolveAddressRequest) (*ResolveAddressResponse, error) {

 	if err := p.authorize(ctx, []string{"swarm-manager"}); err != nil {

@@ -673,6 +701,16 @@ func (m *ProcessRaftMessageResponse) Copy() *ProcessRaftMessageResponse {

 }

 func (m *ProcessRaftMessageResponse) CopyFrom(src interface{}) {}

+func (m *StreamRaftMessageResponse) Copy() *StreamRaftMessageResponse {

+	if m == nil {

+		return nil

+	}

+	o := &StreamRaftMessageResponse{}

+	o.CopyFrom(m)

+	return o

+}

+

+func (m *StreamRaftMessageResponse) CopyFrom(src interface{}) {}

 func (m *ResolveAddressRequest) Copy() *ResolveAddressRequest {

 	if m == nil {

 		return nil

@@ -813,6 +851,12 @@ type RaftClient interface {

 	// ProcessRaftMessage sends a raft message to be processed on a raft member, it is

 	// called from the RaftMember willing to send a message to its destination ('To' field)

 	ProcessRaftMessage(ctx context.Context, in *ProcessRaftMessageRequest, opts ...grpc.CallOption) (*ProcessRaftMessageResponse, error)

+	// StreamRaftMessage accepts a stream of raft messages of type StreamRaftMessageRequest

+	// to be processed on a raft member, returning a StreamRaftMessageResponse

+	// when processing of the streamed messages is complete. A single stream corresponds

+	// to a single raft message, which may be disassembled and streamed as individual messages.

+	// It is called from the Raft leader, which uses it to stream messages to a raft member.

+	StreamRaftMessage(ctx context.Context, opts ...grpc.CallOption) (Raft_StreamRaftMessageClient, error)

 	// ResolveAddress returns the address where the node with the given ID can be reached.

 	ResolveAddress(ctx context.Context, in *ResolveAddressRequest, opts ...grpc.CallOption) (*ResolveAddressResponse, error)

 }

@@ -834,6 +878,40 @@ func (c *raftClient) ProcessRaftMessage(ctx context.Context, in *ProcessRaftMess

 	return out, nil

 }

+func (c *raftClient) StreamRaftMessage(ctx context.Context, opts ...grpc.CallOption) (Raft_StreamRaftMessageClient, error) {

+	stream, err := grpc.NewClientStream(ctx, &_Raft_serviceDesc.Streams[0], c.cc, "/docker.swarmkit.v1.Raft/StreamRaftMessage", opts...)

+	if err != nil {

+		return nil, err

+	}

+	x := &raftStreamRaftMessageClient{stream}

+	return x, nil

+}

+

+type Raft_StreamRaftMessageClient interface {

+	Send(*StreamRaftMessageRequest) error

+	CloseAndRecv() (*StreamRaftMessageResponse, error)

+	grpc.ClientStream

+}

+

+type raftStreamRaftMessageClient struct {

+	grpc.ClientStream

+}

+

+func (x *raftStreamRaftMessageClient) Send(m *StreamRaftMessageRequest) error {

+	return x.ClientStream.SendMsg(m)

+}

+

+func (x *raftStreamRaftMessageClient) CloseAndRecv() (*StreamRaftMessageResponse, error) {

+	if err := x.ClientStream.CloseSend(); err != nil {

+		return nil, err

+	}

+	m := new(StreamRaftMessageResponse)

+	if err := x.ClientStream.RecvMsg(m); err != nil {

+		return nil, err

+	}

+	return m, nil

+}

+

 func (c *raftClient) ResolveAddress(ctx context.Context, in *ResolveAddressRequest, opts ...grpc.CallOption) (*ResolveAddressResponse, error) {

 	out := new(ResolveAddressResponse)

 	err := grpc.Invoke(ctx, "/docker.swarmkit.v1.Raft/ResolveAddress", in, out, c.cc, opts...)

@@ -849,6 +927,12 @@ type RaftServer interface {

 	// ProcessRaftMessage sends a raft message to be processed on a raft member, it is

 	// called from the RaftMember willing to send a message to its destination ('To' field)

 	ProcessRaftMessage(context.Context, *ProcessRaftMessageRequest) (*ProcessRaftMessageResponse, error)

+	// StreamRaftMessage accepts a stream of raft messages of type StreamRaftMessageRequest

+	// to be processed on a raft member, returning a StreamRaftMessageResponse

+	// when processing of the streamed messages is complete. A single stream corresponds

+	// to a single raft message, which may be disassembled and streamed as individual messages.

+	// It is called from the Raft leader, which uses it to stream messages to a raft member.

+	StreamRaftMessage(Raft_StreamRaftMessageServer) error

 	// ResolveAddress returns the address where the node with the given ID can be reached.

 	ResolveAddress(context.Context, *ResolveAddressRequest) (*ResolveAddressResponse, error)

 }

@@ -875,6 +959,32 @@ func _Raft_ProcessRaftMessage_Handler(srv interface{}, ctx context.Context, dec

 	return interceptor(ctx, in, info, handler)

 }

+func _Raft_StreamRaftMessage_Handler(srv interface{}, stream grpc.ServerStream) error {

+	return srv.(RaftServer).StreamRaftMessage(&raftStreamRaftMessageServer{stream})

+}

+

+type Raft_StreamRaftMessageServer interface {

+	SendAndClose(*StreamRaftMessageResponse) error

+	Recv() (*StreamRaftMessageRequest, error)

+	grpc.ServerStream

+}

+

+type raftStreamRaftMessageServer struct {

+	grpc.ServerStream

+}

+

+func (x *raftStreamRaftMessageServer) SendAndClose(m *StreamRaftMessageResponse) error {

+	return x.ServerStream.SendMsg(m)

+}

+

+func (x *raftStreamRaftMessageServer) Recv() (*StreamRaftMessageRequest, error) {

+	m := new(StreamRaftMessageRequest)

+	if err := x.ServerStream.RecvMsg(m); err != nil {

+		return nil, err

+	}

+	return m, nil

+}

+

 func _Raft_ResolveAddress_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {

 	in := new(ResolveAddressRequest)

 	if err := dec(in); err != nil {

@@ -906,7 +1016,13 @@ var _Raft_serviceDesc = grpc.ServiceDesc{

 			Handler:    _Raft_ResolveAddress_Handler,

 		},

 	},

-	Streams:  []grpc.StreamDesc{},

+	Streams: []grpc.StreamDesc{

+		{

+			StreamName:    "StreamRaftMessage",

+			Handler:       _Raft_StreamRaftMessage_Handler,

+			ClientStreams: true,

+		},

+	},

 	Metadata: "github.com/docker/swarmkit/api/raft.proto",

 }

@@ -1212,6 +1328,52 @@ func (m *ProcessRaftMessageResponse) MarshalTo(dAtA []byte) (int, error) {

 	return i, nil

 }

+func (m *StreamRaftMessageRequest) Marshal() (dAtA []byte, err error) {

+	size := m.Size()

+	dAtA = make([]byte, size)

+	n, err := m.MarshalTo(dAtA)

+	if err != nil {

+		return nil, err

+	}

+	return dAtA[:n], nil

+}

+

+func (m *StreamRaftMessageRequest) MarshalTo(dAtA []byte) (int, error) {

+	var i int

+	_ = i

+	var l int

+	_ = l

+	if m.Message != nil {

+		dAtA[i] = 0xa

+		i++

+		i = encodeVarintRaft(dAtA, i, uint64(m.Message.Size()))

+		n4, err := m.Message.MarshalTo(dAtA[i:])

+		if err != nil {

+			return 0, err

+		}

+		i += n4

+	}

+	return i, nil

+}

+

+func (m *StreamRaftMessageResponse) Marshal() (dAtA []byte, err error) {

+	size := m.Size()

+	dAtA = make([]byte, size)

+	n, err := m.MarshalTo(dAtA)

+	if err != nil {

+		return nil, err

+	}

+	return dAtA[:n], nil

+}

+

+func (m *StreamRaftMessageResponse) MarshalTo(dAtA []byte) (int, error) {

+	var i int

+	_ = i

+	var l int

+	_ = l

+	return i, nil

+}

+

 func (m *ResolveAddressRequest) Marshal() (dAtA []byte, err error) {

 	size := m.Size()

 	dAtA = make([]byte, size)

@@ -1315,11 +1477,11 @@ func (m *StoreAction) MarshalTo(dAtA []byte) (int, error) {

 		i = encodeVarintRaft(dAtA, i, uint64(m.Action))

 	}

 	if m.Target != nil {

-		nn4, err := m.Target.MarshalTo(dAtA[i:])

+		nn5, err := m.Target.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += nn4

+		i += nn5

 	}

 	return i, nil

 }

@@ -1330,11 +1492,11 @@ func (m *StoreAction_Node) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x12

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Node.Size()))

-		n5, err := m.Node.MarshalTo(dAtA[i:])

+		n6, err := m.Node.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n5

+		i += n6

 	}

 	return i, nil

 }

@@ -1344,11 +1506,11 @@ func (m *StoreAction_Service) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x1a

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Service.Size()))

-		n6, err := m.Service.MarshalTo(dAtA[i:])

+		n7, err := m.Service.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n6

+		i += n7

 	}

 	return i, nil

 }

@@ -1358,11 +1520,11 @@ func (m *StoreAction_Task) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x22

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Task.Size()))

-		n7, err := m.Task.MarshalTo(dAtA[i:])

+		n8, err := m.Task.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n7

+		i += n8

 	}

 	return i, nil

 }

@@ -1372,11 +1534,11 @@ func (m *StoreAction_Network) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x2a

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Network.Size()))

-		n8, err := m.Network.MarshalTo(dAtA[i:])

+		n9, err := m.Network.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n8

+		i += n9

 	}

 	return i, nil

 }

@@ -1386,11 +1548,11 @@ func (m *StoreAction_Cluster) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x32

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Cluster.Size()))

-		n9, err := m.Cluster.MarshalTo(dAtA[i:])

+		n10, err := m.Cluster.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n9

+		i += n10

 	}

 	return i, nil

 }

@@ -1400,11 +1562,11 @@ func (m *StoreAction_Secret) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x3a

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Secret.Size()))

-		n10, err := m.Secret.MarshalTo(dAtA[i:])

+		n11, err := m.Secret.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n10

+		i += n11

 	}

 	return i, nil

 }

@@ -1414,11 +1576,11 @@ func (m *StoreAction_Resource) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x42

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Resource.Size()))

-		n11, err := m.Resource.MarshalTo(dAtA[i:])

+		n12, err := m.Resource.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n11

+		i += n12

 	}

 	return i, nil

 }

@@ -1428,11 +1590,11 @@ func (m *StoreAction_Extension) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x4a

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Extension.Size()))

-		n12, err := m.Extension.MarshalTo(dAtA[i:])

+		n13, err := m.Extension.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n12

+		i += n13

 	}

 	return i, nil

 }

@@ -1442,11 +1604,11 @@ func (m *StoreAction_Config) MarshalTo(dAtA []byte) (int, error) {

 		dAtA[i] = 0x52

 		i++

 		i = encodeVarintRaft(dAtA, i, uint64(m.Config.Size()))

-		n13, err := m.Config.MarshalTo(dAtA[i:])

+		n14, err := m.Config.MarshalTo(dAtA[i:])

 		if err != nil {

 			return 0, err

 		}

-		i += n13

+		i += n14

 	}

 	return i, nil

 }

@@ -1488,12 +1650,12 @@ func NewRaftProxyRaftServer(local RaftServer, connSelector raftselector.ConnProv

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -1585,6 +1747,63 @@ func (p *raftProxyRaftServer) ProcessRaftMessage(ctx context.Context, r *Process

 	return resp, err

 }

+type Raft_StreamRaftMessageServerWrapper struct {

+	Raft_StreamRaftMessageServer

+	ctx context.Context

+}

+

+func (s Raft_StreamRaftMessageServerWrapper) Context() context.Context {

+	return s.ctx

+}

+

+func (p *raftProxyRaftServer) StreamRaftMessage(stream Raft_StreamRaftMessageServer) error {

+	ctx := stream.Context()

+	conn, err := p.connSelector.LeaderConn(ctx)

+	if err != nil {

+		if err == raftselector.ErrIsLeader {

+			ctx, err = p.runCtxMods(ctx, p.localCtxMods)

+			if err != nil {

+				return err

+			}

+			streamWrapper := Raft_StreamRaftMessageServerWrapper{

+				Raft_StreamRaftMessageServer: stream,

+				ctx: ctx,

+			}

+			return p.local.StreamRaftMessage(streamWrapper)

+		}

+		return err

+	}

+	ctx, err = p.runCtxMods(ctx, p.remoteCtxMods)

+	if err != nil {

+		return err

+	}

+	clientStream, err := NewRaftClient(conn).StreamRaftMessage(ctx)

+

+	if err != nil {

+		return err

+	}

+

+	for {

+		msg, err := stream.Recv()

+		if err == io.EOF {

+			break

+		}

+		if err != nil {

+			return err

+		}

+		if err := clientStream.Send(msg); err != nil {

+			return err

+		}

+	}

+

+	reply, err := clientStream.CloseAndRecv()

+	if err != nil {

+		return err

+	}

+

+	return stream.SendAndClose(reply)

+}

+

 func (p *raftProxyRaftServer) ResolveAddress(ctx context.Context, r *ResolveAddressRequest) (*ResolveAddressResponse, error) {

 	conn, err := p.connSelector.LeaderConn(ctx)

@@ -1630,12 +1849,12 @@ func NewRaftProxyRaftMembershipServer(local RaftMembershipServer, connSelector r

 	redirectChecker := func(ctx context.Context) (context.Context, error) {

 		s, ok := transport.StreamFromContext(ctx)

 		if !ok {

-			return ctx, grpc.Errorf(codes.InvalidArgument, "remote addr is not found in context")

+			return ctx, status.Errorf(codes.InvalidArgument, "remote addr is not found in context")

 		}

 		addr := s.ServerTransport().RemoteAddr().String()

 		md, ok := metadata.FromContext(ctx)

 		if ok && len(md["redirect"]) != 0 {

-			return ctx, grpc.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

+			return ctx, status.Errorf(codes.ResourceExhausted, "more than one redirect to leader from: %s", md["redirect"])

 		}

 		if !ok {

 			md = metadata.New(map[string]string{})

@@ -1843,6 +2062,22 @@ func (m *ProcessRaftMessageResponse) Size() (n int) {

 	return n

 }

+func (m *StreamRaftMessageRequest) Size() (n int) {

+	var l int

+	_ = l

+	if m.Message != nil {

+		l = m.Message.Size()

+		n += 1 + l + sovRaft(uint64(l))

+	}

+	return n

+}

+

+func (m *StreamRaftMessageResponse) Size() (n int) {

+	var l int

+	_ = l

+	return n

+}

+

 func (m *ResolveAddressRequest) Size() (n int) {

 	var l int

 	_ = l

@@ -2057,6 +2292,25 @@ func (this *ProcessRaftMessageResponse) String() string {

 	}, "")

 	return s

 }

+func (this *StreamRaftMessageRequest) String() string {

+	if this == nil {

+		return "nil"

+	}

+	s := strings.Join([]string{`&StreamRaftMessageRequest{`,

+		`Message:` + strings.Replace(fmt.Sprintf("%v", this.Message), "Message", "raftpb.Message", 1) + `,`,

+		`}`,

+	}, "")

+	return s

+}

+func (this *StreamRaftMessageResponse) String() string {

+	if this == nil {

+		return "nil"

+	}

+	s := strings.Join([]string{`&StreamRaftMessageResponse{`,

+		`}`,

+	}, "")

+	return s

+}

 func (this *ResolveAddressRequest) String() string {

 	if this == nil {

 		return "nil"

@@ -2861,6 +3115,139 @@ func (m *ProcessRaftMessageResponse) Unmarshal(dAtA []byte) error {

 	}

 	return nil

 }

+func (m *StreamRaftMessageRequest) Unmarshal(dAtA []byte) error {

+	l := len(dAtA)

+	iNdEx := 0

+	for iNdEx < l {

+		preIndex := iNdEx

+		var wire uint64

+		for shift := uint(0); ; shift += 7 {

+			if shift >= 64 {

+				return ErrIntOverflowRaft

+			}

+			if iNdEx >= l {

+				return io.ErrUnexpectedEOF

+			}

+			b := dAtA[iNdEx]

+			iNdEx++

+			wire |= (uint64(b) & 0x7F) << shift

+			if b < 0x80 {

+				break

+			}

+		}

+		fieldNum := int32(wire >> 3)

+		wireType := int(wire & 0x7)

+		if wireType == 4 {

+			return fmt.Errorf("proto: StreamRaftMessageRequest: wiretype end group for non-group")

+		}

+		if fieldNum <= 0 {

+			return fmt.Errorf("proto: StreamRaftMessageRequest: illegal tag %d (wire type %d)", fieldNum, wire)

+		}

+		switch fieldNum {

+		case 1:

+			if wireType != 2 {

+				return fmt.Errorf("proto: wrong wireType = %d for field Message", wireType)

+			}

+			var msglen int

+			for shift := uint(0); ; shift += 7 {

+				if shift >= 64 {

+					return ErrIntOverflowRaft

+				}

+				if iNdEx >= l {

+					return io.ErrUnexpectedEOF

+				}

+				b := dAtA[iNdEx]

+				iNdEx++

+				msglen |= (int(b) & 0x7F) << shift

+				if b < 0x80 {

+					break

+				}

+			}

+			if msglen < 0 {

+				return ErrInvalidLengthRaft

+			}

+			postIndex := iNdEx + msglen

+			if postIndex > l {

+				return io.ErrUnexpectedEOF

+			}

+			if m.Message == nil {

+				m.Message = &raftpb.Message{}

+			}

+			if err := m.Message.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {

+				return err

+			}

+			iNdEx = postIndex

+		default:

+			iNdEx = preIndex

+			skippy, err := skipRaft(dAtA[iNdEx:])

+			if err != nil {

+				return err

+			}

+			if skippy < 0 {

+				return ErrInvalidLengthRaft

+			}

+			if (iNdEx + skippy) > l {

+				return io.ErrUnexpectedEOF

+			}

+			iNdEx += skippy

+		}

+	}

+

+	if iNdEx > l {

+		return io.ErrUnexpectedEOF

+	}

+	return nil

+}

+func (m *StreamRaftMessageResponse) Unmarshal(dAtA []byte) error {

+	l := len(dAtA)

+	iNdEx := 0

+	for iNdEx < l {

+		preIndex := iNdEx

+		var wire uint64

+		for shift := uint(0); ; shift += 7 {

+			if shift >= 64 {

+				return ErrIntOverflowRaft

+			}

+			if iNdEx >= l {

+				return io.ErrUnexpectedEOF

+			}

+			b := dAtA[iNdEx]

+			iNdEx++

+			wire |= (uint64(b) & 0x7F) << shift

+			if b < 0x80 {

+				break

+			}

+		}

+		fieldNum := int32(wire >> 3)

+		wireType := int(wire & 0x7)

+		if wireType == 4 {

+			return fmt.Errorf("proto: StreamRaftMessageResponse: wiretype end group for non-group")

+		}

+		if fieldNum <= 0 {

+			return fmt.Errorf("proto: StreamRaftMessageResponse: illegal tag %d (wire type %d)", fieldNum, wire)

+		}

+		switch fieldNum {

+		default:

+			iNdEx = preIndex

+			skippy, err := skipRaft(dAtA[iNdEx:])

+			if err != nil {

+				return err

+			}

+			if skippy < 0 {

+				return ErrInvalidLengthRaft

+			}

+			if (iNdEx + skippy) > l {

+				return io.ErrUnexpectedEOF

+			}

+			iNdEx += skippy

+		}

+	}

+

+	if iNdEx > l {

+		return io.ErrUnexpectedEOF

+	}

+	return nil

+}

 func (m *ResolveAddressRequest) Unmarshal(dAtA []byte) error {

 	l := len(dAtA)

 	iNdEx := 0

@@ -3574,66 +3961,69 @@ var (