make the maximum accepted gRPC message size configurable

cmd/gazette/main.go

@@ -85,7 +85,7 @@ func (cmdServe) Execute(args []string) error {
 	}
 
 	// Bind our server listener, grabbing a random available port if Port is zero.
-	srv, err := server.New("", Config.Broker.Host, Config.Broker.Port, serverTLS, peerTLS)
+	srv, err := server.New("", Config.Broker.Host, Config.Broker.Port, serverTLS, peerTLS, Config.Broker.MaxGRPCRecvSize)
 	mbp.Must(err, "building Server instance")
 
 	// If a file:// root was provided, ensure it exists and apply it.

mainboilerplate/runconsumer/run_consumer.go

@@ -149,7 +149,7 @@ func (sc Cmd) Execute(args []string) error {
 	}
 
 	// Bind our server listener, grabbing a random available port if Port is zero.
-	srv, err := server.New("", bc.Consumer.Host, bc.Consumer.Port, serverTLS, peerTLS)
+	srv, err := server.New("", bc.Consumer.Host, bc.Consumer.Port, serverTLS, peerTLS, bc.Consumer.MaxGRPCRecvSize)
 	mbp.Must(err, "building Server instance")
 
 	if bc.Broker.Cache.Size <= 0 {

mainboilerplate/service.go

@@ -23,6 +23,7 @@ type ServiceConfig struct {
 	PeerCertFile      string `long:"peer-cert-file" env:"PEER_CERT_FILE" default:"" description:"Path to the client TLS certificate for peer-to-peer requests"`
 	PeerCertKeyFile   string `long:"peer-cert-key-file" env:"PEER_CERT_KEY_FILE" default:"" description:"Path to the client TLS private key for peer-to-peer requests"`
 	PeerCAFile        string `long:"peer-ca-file" env:"PEER_CA_FILE" default:"" description:"Path to the trusted CA for client verification of peer server certificates. When absent, the system CA pool is used instead."`
+	MaxGRPCRecvSize   uint32 `long:"max-grpc-recv-size" env:"MAX_GRPC_RECV_SIZE" default:"4194304" description:"Maximum size of gRPC messages accepted by this server, in bytes"`
 }
 
 // ProcessSpec of the ServiceConfig.

server/server.go

@@ -51,7 +51,7 @@ type Server struct {
 // and `port` for serving traffic directed at `host`.
 // `port` may be empty, in which case a random free port is assigned.
 // if `tlsConfig` is non-nil, the Server uses TLS (and is otherwise in the clear).
-func New(iface, host, port string, serverTLS, peerTLS *tls.Config) (*Server, error) {
+func New(iface, host, port string, serverTLS, peerTLS *tls.Config, maxSize uint32) (*Server, error) {
 	var network, bind string
 	if port == "" {
 		network, bind = "tcp", fmt.Sprintf("%s:0", iface) // Assign a random free port.
@@ -100,6 +100,7 @@ func New(iface, host, port string, serverTLS, peerTLS *tls.Config) (*Server, err
 		GRPCServer: grpc.NewServer(
 			grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
 			grpc.UnaryInterceptor(grpc_prometheus.UnaryServerInterceptor),
+			grpc.MaxRecvMsgSize(int(maxSize)),
 		),
 	}
 	if serverTLS != nil {
@@ -192,7 +193,7 @@ func BuildTLSConfig(certPath, keyPath, trustedCAPath string) (*tls.Config, error
 // MustLoopback builds and returns a new Server instance bound to a random
 // port on the loopback interface. It panics on error.
 func MustLoopback() *Server {
-	if srv, err := New("127.0.0.1", "127.0.0.1", "", nil, nil); err != nil {
+	if srv, err := New("127.0.0.1", "127.0.0.1", "", nil, nil, 1<<20); err != nil {
 		log.WithField("err", err).Panic("failed to build Server")
 		panic("not reached")
 	} else {