feat: allow definition of max message size (#60)

Makefile

@@ -32,7 +32,7 @@ unit-test:
 	make gen-mocks
 	@if ! command -v gotestsum > /dev/null; then echo "Installing gotestsum" && go install -a -v gotest.tools/gotestsum@latest; fi;
 
-	gotestsum --junitfile junit.xml -- -p 6 --parallel 6 -v -short -covermode=atomic -coverprofile coverage.out ./... 2>&1 | tee gotest.out
+	gotestsum --junitfile junit.xml -- -p 6 --parallel 6 -v -short -covermode=atomic -coverprofile coverage.out ./internal/... 2>&1 | tee gotest.out
 	go tool cover -func coverage.out | tee coverage.txt
 
 # Run all tests including integration tests
@@ -41,7 +41,7 @@ test:
 	make gen-mocks
 	@if ! command -v gotestsum > /dev/null; then echo "Installing gotestsum" && go install -a -v gotest.tools/gotestsum@latest; fi;
 
-	gotestsum --junitfile junit.xml -- -v -p 1 -parallel 1 -covermode=atomic -coverprofile coverage.out ./... 2>&1 | tee gotest.out
+	gotestsum --junitfile junit.xml -- -v -p 1 -parallel 1 -covermode=atomic -coverprofile coverage.out ./internal/... 2>&1 | tee gotest.out
 	go tool cover -func coverage.out | tee coverage.txt
 
 # Run only integration tests
@@ -50,7 +50,7 @@ integration-test:
 	make gen-mocks
 	@if ! command -v gotestsum > /dev/null; then echo "Installing gotestsum" && go install -a -v gotest.tools/gotestsum@latest; fi;
 
-	gotestsum --junitfile junit.xml -- -v -p 1 -parallel 1 -run Integration -covermode=atomic -coverprofile coverage.out  ./... 2>&1 | tee gotest.out
+	gotestsum --junitfile junit.xml -- -v -p 1 -parallel 1 -run Integration -covermode=atomic -coverprofile coverage.out ./internal/... 2>&1 | tee gotest.out
 	go tool cover -func coverage.out | tee coverage.txt
 
 gen-proto:

README.md

@@ -122,6 +122,8 @@ All available environment variables are listed below:
 | `DECKARD_GRPC_SERVER_MAX_CONNECTION_IDLE` |  | The maximum duration a connection can remain idle. |
 | `DECKARD_GRPC_SERVER_MAX_CONNECTION_AGE` |  | The maximum duration a connection can exist. |
 | `DECKARD_GRPC_SERVER_MAX_CONNECTION_AGE_GRACE` |  | The additional time the gRPC server allows for a connection to complete its current operations before closing it after reaching the maximum connection age. |
+| `DECKARD_GRPC_SERVER_MAX_RECV_MSG_SIZE` | 4194304 | The maximum message size that can be received. |
+| `DECKARD_GRPC_SERVER_MAX_SEND_MSG_SIZE` | 4194304 | The maximum message size that can be sent. |
 
 > Values should be specified using time units such as `1s` for seconds, `1m` for minutes, `1h` for hours.
 

internal/config/service.go

@@ -30,6 +30,16 @@ var GrpcServerMaxConnectionAgeGrace = Create(&ViperConfigKey{
 	Key: "grpc.server.max_connection_age_grace",
 })
 
+var GrpcServerMaxRecvMsgSize = Create(&ViperConfigKey{
+	Key:     "grpc.server.max_recv_msg_size",
+	Default: 4194304,
+})
+
+var GrpcServerMaxSendMsgSize = Create(&ViperConfigKey{
+	Key:     "grpc.server.max_send_msg_size",
+	Default: 4194304,
+})
+
 var TlsServerCertFilePaths = Create(&ViperConfigKey{
 	Key: "tls.server.cert_file_paths",
 })

internal/service/deckard_service.go

@@ -68,6 +68,8 @@ func NewDeckardService(qpool queue.DeckardQueue, queueConfigurationService queue
 
 func (d *Deckard) ServeGRPCServer(ctx context.Context) (*grpc.Server, error) {
 	port := config.GrpcPort.GetInt()
+	maxMsgRecvSize := config.GrpcServerMaxRecvMsgSize.GetInt()
+	maxMsgSendSize := config.GrpcServerMaxSendMsgSize.GetInt()
 
 	listen, err := net.Listen("tcp", fmt.Sprint(":", port))
 	if err != nil {
@@ -78,6 +80,8 @@ func (d *Deckard) ServeGRPCServer(ctx context.Context) (*grpc.Server, error) {
 		grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor()),
 		grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()),
 		getGrpcKeepaliveParams(),
+		grpc.MaxRecvMsgSize(maxMsgRecvSize),
+		grpc.MaxSendMsgSize(maxMsgSendSize),
 	}
 
 	credentials, err := loadTLSCredentials()

internal/service/deckard_service_test.go

@@ -32,6 +32,56 @@ import (
 
 var ctx = context.Background()
 
+func TestMessageSizeLimitDeckardGRPCServeIntegration(t *testing.T) {
+	if testing.Short() {
+		return
+	}
+
+	config.Configure(true)
+
+	config.GrpcPort.Set("8085")
+	config.GrpcServerMaxRecvMsgSize.Set(1)
+	config.GrpcServerMaxSendMsgSize.Set(1)
+
+	storage := storage.NewMemoryStorage(ctx)
+	cache := cache.NewMemoryCache()
+
+	queueService := queue.NewQueueConfigurationService(ctx, storage)
+
+	queue := queue.NewQueue(&audit.AuditorImpl{}, storage, queueService, cache)
+
+	srv := NewMemoryDeckardService(queue, queueService)
+
+	server, err := srv.ServeGRPCServer(ctx)
+	require.NoError(t, err)
+	defer server.Stop()
+
+	// Set up a connection to the server.
+	ctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
+	defer cancel()
+	conn, err := grpc.DialContext(ctx, fmt.Sprint("localhost:", config.GrpcPort.GetInt()), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
+	if err != nil {
+		log.Fatalf("did not connect: %v", err)
+	}
+	defer conn.Close()
+
+	client := deckard.NewDeckardClient(conn)
+
+	_, err = client.Add(ctx, &deckard.AddRequest{
+		Messages: []*deckard.AddMessage{
+			{
+				Id:       "1",
+				Queue:    "queue",
+				Timeless: true,
+				StringPayload: "This is a message test",
+			},
+		},
+	})
+
+	require.Error(t, err)
+	require.EqualError(t, err, "rpc error: code = ResourceExhausted desc = grpc: received message larger than max (38 vs. 1)")
+}
+
 func TestMemoryDeckardGRPCServeIntegration(t *testing.T) {
 	if testing.Short() {
 		return