Make add operation ctx timeout configurable (#682)

Co-authored-by: joao.bologna <[email protected]>

Author: joaobologna

config/config.yaml

@@ -82,6 +82,7 @@ operations:
     add:
       limit: 1000
       batchSize: 40
+      operationTimeout: 5s
     remove:
       limit: 50
 

internal/core/operations/rooms/add/executor.go

@@ -41,13 +41,15 @@ import (
 )
 
 const (
-	DefaultAmountLimit = 1000
-	DefaultBatchSize   = 40
+	DefaultAmountLimit      = 1000
+	DefaultBatchSize        = 40
+	DefaultOperationTimeout = 5 * time.Second
 )
 
 type Config struct {
-	AmountLimit int32
-	BatchSize   int32
+	AmountLimit      int32
+	BatchSize        int32
+	OperationTimeout time.Duration
 }
 
 type Executor struct {
@@ -68,6 +70,10 @@ func NewExecutor(roomManager ports.RoomManager, storage ports.SchedulerStorage,
 		zap.L().Sugar().Infof("Add Executor - Batch size wrongly configured with %d, using default value %d", config.BatchSize, DefaultBatchSize)
 		config.BatchSize = DefaultBatchSize
 	}
+	if config.OperationTimeout <= 0 {
+		zap.L().Sugar().Infof("Add Executor - Operation timeout wrongly configured with %v, using default value %v", config.OperationTimeout, DefaultOperationTimeout)
+		config.OperationTimeout = DefaultOperationTimeout
+	}
 	return &Executor{
 		roomManager,
 		storage,
@@ -119,7 +125,7 @@ func (ex *Executor) Execute(ctx context.Context, op *operation.Operation, defini
 			go func() {
 				defer wg.Done()
 
-				ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+				ctx, cancel := context.WithTimeout(ctx, ex.config.OperationTimeout)
 				defer cancel()
 
 				errsCh <- ex.createRoom(ctx, scheduler, executionLogger)

internal/core/operations/rooms/add/executor_test.go

@@ -308,6 +308,89 @@ func TestExecutor_Execute(t *testing.T) {
 
 		require.Nil(t, err)
 	})
+
+	t.Run("use default operation timeout if OperationTimeout not set", func(t *testing.T) {
+		executor, _, _, _ := testSetup(t, Config{})
+		require.NotNil(t, executor)
+		require.Equal(t, executor.config.OperationTimeout, DefaultOperationTimeout)
+	})
+
+	t.Run("use default operation timeout if invalid OperationTimeout value", func(t *testing.T) {
+		executor, _, _, _ := testSetup(t, Config{OperationTimeout: -1 * time.Second})
+		require.NotNil(t, executor)
+		require.Equal(t, executor.config.OperationTimeout, DefaultOperationTimeout)
+	})
+
+	t.Run("use custom operation timeout when valid value provided", func(t *testing.T) {
+		customTimeout := 10 * time.Second
+		executor, _, _, _ := testSetup(t, Config{OperationTimeout: customTimeout})
+		require.NotNil(t, executor)
+		require.Equal(t, executor.config.OperationTimeout, customTimeout)
+	})
+
+	t.Run("should timeout when room creation takes longer than configured timeout", func(t *testing.T) {
+		shortTimeoutDefinition := Definition{Amount: 1}
+		shortTimeoutConfig := Config{
+			AmountLimit:      10,
+			BatchSize:        10,
+			OperationTimeout: 10 * time.Millisecond, // Very short timeout
+		}
+
+		_, roomsManager, schedulerStorage, operationsManager := testSetup(t, shortTimeoutConfig)
+
+		schedulerStorage.EXPECT().GetScheduler(context.Background(), op.SchedulerName).Return(&scheduler, nil)
+
+		// Mock room creation to take longer than the timeout
+		roomsManager.EXPECT().CreateRoom(gomock.Any(), gomock.Any(), false).DoAndReturn(
+			func(ctx context.Context, scheduler entities.Scheduler, checkRoomDeletion bool) (*game_room.GameRoom, *game_room.Instance, error) {
+				select {
+				// Sleep longer than the timeout to trigger context deadline
+				case <-time.After(100 * time.Millisecond):
+					return &gameRoom, &gameRoomInstance, nil
+				// Should return context deadline exceeded
+				case <-ctx.Done():
+					return nil, nil, ctx.Err()
+				}
+			},
+		).Times(1)
+
+		operationsManager.EXPECT().AppendOperationEventToExecutionHistory(gomock.Any(), &op, "more rooms failed than succeeded, errors: 1 and successes: 0 of amount: 1")
+
+		executor := NewExecutor(roomsManager, schedulerStorage, operationsManager, shortTimeoutConfig)
+		err := executor.Execute(context.Background(), &op, &shortTimeoutDefinition)
+
+		require.NotNil(t, err)
+		require.ErrorContains(t, err, "more rooms failed than succeeded")
+	})
+
+	t.Run("should succeed when room creation completes within configured timeout", func(t *testing.T) {
+		fastDefinition := Definition{Amount: 2}
+		reasonableTimeoutConfig := Config{
+			AmountLimit:      10,
+			BatchSize:        10,
+			OperationTimeout: 1 * time.Second, // Reasonable timeout
+		}
+
+		_, roomsManager, schedulerStorage, operationsManager := testSetup(t, reasonableTimeoutConfig)
+
+		schedulerStorage.EXPECT().GetScheduler(context.Background(), op.SchedulerName).Return(&scheduler, nil)
+
+		// Mock room creation to complete quickly within the timeout
+		roomsManager.EXPECT().CreateRoom(gomock.Any(), gomock.Any(), false).DoAndReturn(
+			func(ctx context.Context, scheduler entities.Scheduler, checkRoomDeletion bool) (*game_room.GameRoom, *game_room.Instance, error) {
+				// Complete quickly within timeout
+				time.Sleep(50 * time.Millisecond) // Much less than 1 second
+				return &gameRoom, &gameRoomInstance, nil
+			},
+		).Times(2)
+
+		operationsManager.EXPECT().AppendOperationEventToExecutionHistory(gomock.Any(), &op, fmt.Sprintf("added %d rooms", fastDefinition.Amount))
+
+		executor := NewExecutor(roomsManager, schedulerStorage, operationsManager, reasonableTimeoutConfig)
+		err := executor.Execute(context.Background(), &op, &fastDefinition)
+
+		require.Nil(t, err)
+	})
 }
 
 func TestExecutor_Rollback(t *testing.T) {

internal/service/config.go

@@ -40,17 +40,18 @@ import (
 )
 
 const (
-	healthControllerExecutionIntervalConfigPath = "workers.healthControllerInterval"
-	storagecleanupExecutionIntervalConfigPath   = "workers.storageClenupInterval"
-	roomInitializationTimeoutMillisConfigPath   = "services.roomManager.roomInitializationTimeoutMillis"
-	roomRoomValidationAttemptsConfigPath        = "services.roomManager.roomValidationAttempts"
-	roomPingTimeoutMillisConfigPath             = "services.roomManager.roomPingTimeoutMillis"
-	roomDeletionTimeoutMillisConfigPath         = "services.roomManager.roomDeletionTimeoutMillis"
-	operationLeaseTTLMillisConfigPath           = "services.operationManager.operationLeaseTTLMillis"
-	schedulerCacheTTLMillisConfigPath           = "services.eventsForwarder.schedulerCacheTTLMillis"
-	operationsRoomsAddLimitConfigPath           = "operations.rooms.add.limit"
-	operationsRoomsAddBatchSizePath             = "operations.rooms.add.batchSize"
-	operationsRoomsRemoveLimitConfigPath        = "operations.rooms.remove.limit"
+	healthControllerExecutionIntervalConfigPath  = "workers.healthControllerInterval"
+	storagecleanupExecutionIntervalConfigPath    = "workers.storageClenupInterval"
+	roomInitializationTimeoutMillisConfigPath    = "services.roomManager.roomInitializationTimeoutMillis"
+	roomRoomValidationAttemptsConfigPath         = "services.roomManager.roomValidationAttempts"
+	roomPingTimeoutMillisConfigPath              = "services.roomManager.roomPingTimeoutMillis"
+	roomDeletionTimeoutMillisConfigPath          = "services.roomManager.roomDeletionTimeoutMillis"
+	operationLeaseTTLMillisConfigPath            = "services.operationManager.operationLeaseTTLMillis"
+	schedulerCacheTTLMillisConfigPath            = "services.eventsForwarder.schedulerCacheTTLMillis"
+	operationsRoomsAddLimitConfigPath            = "operations.rooms.add.limit"
+	operationsRoomsAddBatchSizePath              = "operations.rooms.add.batchSize"
+	operationsRoomsAddOperationTimeoutConfigPath = "operations.rooms.add.operationTimeout"
+	operationsRoomsRemoveLimitConfigPath         = "operations.rooms.remove.limit"
 )
 
 // NewCreateSchedulerVersionConfig instantiate a new CreateSchedulerVersionConfig to be used by the NewSchedulerVersion operation to customize its configuration.
@@ -88,10 +89,12 @@ func NewHealthControllerConfig(c config.Config) healthcontroller.Config {
 func NewOperationRoomsAddConfig(c config.Config) add.Config {
 	operationsRoomsAddLimit := int32(c.GetInt(operationsRoomsAddLimitConfigPath))
 	operationsRoomsAddBatchSize := int32(c.GetInt(operationsRoomsAddBatchSizePath))
+	operationsRoomsAddOperationTimeout := c.GetDuration(operationsRoomsAddOperationTimeoutConfigPath)
 
 	config := add.Config{
-		AmountLimit: operationsRoomsAddLimit,
-		BatchSize:   operationsRoomsAddBatchSize,
+		AmountLimit:      operationsRoomsAddLimit,
+		BatchSize:        operationsRoomsAddBatchSize,
+		OperationTimeout: operationsRoomsAddOperationTimeout,
 	}
 
 	return config