move grpc message size option to driver level

Author: rekby

config/config.go

@@ -20,18 +20,19 @@ import (
 type Config struct {
 	config.Common
 
-	trace          *trace.Driver
-	dialTimeout    time.Duration
-	connectionTTL  time.Duration
-	balancerConfig *balancerConfig.Config
-	secure         bool
-	endpoint       string
-	database       string
-	metaOptions    []meta.Option
-	grpcOptions    []grpc.DialOption
-	credentials    credentials.Credentials
-	tlsConfig      *tls.Config
-	meta           *meta.Meta
+	trace              *trace.Driver
+	dialTimeout        time.Duration
+	connectionTTL      time.Duration
+	balancerConfig     *balancerConfig.Config
+	secure             bool
+	endpoint           string
+	database           string
+	metaOptions        []meta.Option
+	grpcOptions        []grpc.DialOption
+	grpcMaxMessageSize int
+	credentials        credentials.Credentials
+	tlsConfig          *tls.Config
+	meta               *meta.Meta
 
 	excludeGRPCCodesForPessimization []grpcCodes.Code
 }
@@ -53,6 +54,11 @@ func (c *Config) GrpcDialOptions() []grpc.DialOption {
 	)
 }
 
+// GrpcMaxMessageSize return client settings for max grpc message size
+func (c *Config) GrpcMaxMessageSize() int {
+	return c.grpcMaxMessageSize
+}
+
 // Meta reports meta information about database connection
 func (c *Config) Meta() *meta.Meta {
 	return c.meta
@@ -245,6 +251,19 @@ func WithDialTimeout(timeout time.Duration) Option {
 	}
 }
 
+func WithGrpcMaxMessageSize(sizeBytes int) Option {
+	return func(c *Config) {
+		c.grpcMaxMessageSize = sizeBytes
+		c.grpcOptions = append(c.grpcOptions,
+			// limit size of outgoing and incoming packages
+			grpc.WithDefaultCallOptions(
+				grpc.MaxCallRecvMsgSize(c.grpcMaxMessageSize),
+				grpc.MaxCallSendMsgSize(c.grpcMaxMessageSize),
+			),
+		)
+	}
+}
+
 func WithBalancer(balancer *balancerConfig.Config) Option {
 	return func(c *Config) {
 		c.balancerConfig = balancer

config/defaults.go

@@ -107,9 +107,10 @@ func defaultConfig() (c *Config) {
 		credentials: credentials.NewAnonymousCredentials(
 			credentials.WithSourceInfo(stack.Record(0)),
 		),
-		balancerConfig: balancers.Default(),
-		tlsConfig:      defaultTLSConfig(),
-		dialTimeout:    DefaultDialTimeout,
-		trace:          &trace.Driver{},
+		balancerConfig:     balancers.Default(),
+		tlsConfig:          defaultTLSConfig(),
+		dialTimeout:        DefaultDialTimeout,
+		trace:              &trace.Driver{},
+		grpcMaxMessageSize: DefaultGRPCMsgSize,
 	}
 }

driver.go

@@ -34,6 +34,7 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/stack"
 	internalTable "github.com/ydb-platform/ydb-go-sdk/v3/internal/table"
 	tableConfig "github.com/ydb-platform/ydb-go-sdk/v3/internal/table/config"
+	internalTopic "github.com/ydb-platform/ydb-go-sdk/v3/internal/topic"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic/topicclientinternal"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
@@ -588,6 +589,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 				[]topicoptions.TopicOption{
 					topicoptions.WithOperationTimeout(d.config.OperationTimeout()),
 					topicoptions.WithOperationCancelAfter(d.config.OperationCancelAfter()),
+					internalTopic.WithGrpcMessageSize(d.config.GrpcMaxMessageSize()),
 				},
 				d.topicOptions...,
 			)...,

internal/config/config.go

@@ -7,14 +7,6 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
-// InitDefaultGRPCMessageSize used for init value only
-// don't use it in any place, other than initial get values for addr and config
-var InitDefaultGRPCMessageSize = 64 * 1024 * 1024 // 64MB
-
-// DefaultGRPCMessageSizeAddr pointer to public DefaultGRPCMessageSize value
-// the scheme needs for prevent import public package from internal (and prevent import cycle)
-var DefaultGRPCMessageSizeAddr = &InitDefaultGRPCMessageSize
-
 var defaultRetryBudget = budget.Limited(-1)
 
 type Common struct {

internal/topic/configs.go

@@ -1,11 +1,40 @@
 package topic
 
 import (
+	"time"
+
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
 type Config struct {
 	config.Common
-	Trace *trace.Topic
+	Trace              *trace.Topic
+	MaxGrpcMessageSize int
+}
+
+type Option func(c *Config)
+
+func PublicWithTrace(trace trace.Topic, opts ...trace.TopicComposeOption) Option { //nolint:gocritic
+	return func(c *Config) {
+		c.Trace = c.Trace.Compose(&trace, opts...)
+	}
+}
+
+func PublicWithOperationTimeout(operationTimeout time.Duration) Option {
+	return func(c *Config) {
+		config.SetOperationTimeout(&c.Common, operationTimeout)
+	}
+}
+
+func PublicWithOperationCancelAfter(operationCancelAfter time.Duration) Option {
+	return func(c *Config) {
+		config.SetOperationCancelAfter(&c.Common, operationCancelAfter)
+	}
+}
+
+func WithGrpcMessageSize(sizeBytes int) Option {
+	return func(c *Config) {
+		c.MaxGrpcMessageSize = sizeBytes
+	}
 }

internal/topic/topicclientinternal/client.go

@@ -8,6 +8,7 @@ import (
 	"google.golang.org/grpc"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawydb"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic"
@@ -58,8 +59,10 @@ func New(
 
 func newTopicConfig(opts ...topicoptions.TopicOption) topic.Config {
 	c := topic.Config{
-		Trace: &trace.Topic{},
+		Trace:              &trace.Topic{},
+		MaxGrpcMessageSize: config.DefaultGRPCMsgSize,
 	}
+
 	for _, opt := range opts {
 		if opt != nil {
 			opt(&c)
@@ -322,7 +325,9 @@ func (c *Client) StartReader(
 
 // StartWriter create new topic writer wrapper
 func (c *Client) StartWriter(topicPath string, opts ...topicoptions.WriterOption) (*topicwriter.Writer, error) {
-	cfg := c.createWriterConfig(topicPath, opts)
+	cfg := c.createWriterConfig(topicPath, append(opts, topicwriterinternal.WithMaxGrpcMessageBytes(
+		c.cfg.MaxGrpcMessageSize,
+	)))
 	writer, err := topicwriterinternal.NewWriterReconnector(cfg)
 	if err != nil {
 		return nil, err

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -83,7 +83,7 @@ func NewWriterReconnectorConfig(options ...PublicWriterOption) WriterReconnector
 			clock:              clockwork.NewRealClock(),
 			compressorCount:    runtime.NumCPU(),
 			Tracer:             &trace.Topic{},
-			maxBytesPerMessage: *config.DefaultGRPCMessageSizeAddr,
+			maxBytesPerMessage: config.DefaultGRPCMsgSize,
 		},
 		AutoSetSeqNo:       true,
 		AutoSetCreatedTime: true,

options.go

@@ -354,6 +354,17 @@ func WithDialTimeout(timeout time.Duration) Option {
 	}
 }
 
+// WithGrpcMaxMessageSize set max size of message on grpc level
+// use the option for the driver known custom limit.
+// Driver can't read the limit from direct grpc option
+func WithGrpcMaxMessageSize(sizeBytes int) Option {
+	return func(ctx context.Context, d *Driver) error {
+		d.options = append(d.options, config.WithGrpcMaxMessageSize(sizeBytes))
+
+		return nil
+	}
+}
+
 // With collects additional configuration options.
 //
 // This option does not replace collected option, instead it will append provided options.

tests/integration/topic_read_writer_test.go

@@ -532,11 +532,10 @@ func TestSendMessagesLargerThenGRPCLimit(t *testing.T) {
 	const maxGrpcMsgSize = 10000 // bytes
 	const topicMessageSize = maxGrpcMsgSize / 3
 
-	scope.Driver(ydb.With(config.WithGrpcOptions(grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(maxGrpcMsgSize)))))
+	scope.Driver(ydb.WithGrpcMaxMessageSize(maxGrpcMsgSize))
 	writer, err := scope.Driver().Topic().StartWriter(
 		scope.TopicPath(),
 		topicoptions.WithWriterCodec(topictypes.CodecRaw),
-		topicoptions.WithWriterMaxSizeOfDataGrpcMessageBytes(maxGrpcMsgSize),
 	)
 	scope.Require.NoError(err)
 	defer writer.Close(scope.Ctx)

topic/topicoptions/topicoptions_topic.go

@@ -3,23 +3,20 @@ package topicoptions
 import (
 	"time"
 
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
 // TopicOption
 //
 // Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
-type TopicOption func(c *topic.Config)
+type TopicOption = topic.Option // func(c *topic.Config)
 
 // WithTrace defines trace over persqueue client calls
 //
 // Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 func WithTrace(trace trace.Topic, opts ...trace.TopicComposeOption) TopicOption { //nolint:gocritic
-	return func(c *topic.Config) {
-		c.Trace = c.Trace.Compose(&trace, opts...)
-	}
+	return topic.PublicWithTrace(trace, opts...)
 }
 
 // WithOperationTimeout set the maximum amount of time a YDB server will process
@@ -30,9 +27,7 @@ func WithTrace(trace trace.Topic, opts ...trace.TopicComposeOption) TopicOption
 //
 // Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 func WithOperationTimeout(operationTimeout time.Duration) TopicOption {
-	return func(c *topic.Config) {
-		config.SetOperationTimeout(&c.Common, operationTimeout)
-	}
+	return topic.PublicWithOperationTimeout(operationTimeout)
 }
 
 // WithOperationCancelAfter set the maximum amount of time a YDB server will process an
@@ -43,7 +38,5 @@ func WithOperationTimeout(operationTimeout time.Duration) TopicOption {
 //
 // Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 func WithOperationCancelAfter(operationCancelAfter time.Duration) TopicOption {
-	return func(c *topic.Config) {
-		config.SetOperationCancelAfter(&c.Common, operationCancelAfter)
-	}
+	return topic.PublicWithOperationCancelAfter(operationCancelAfter)
 }