diff --git a/.run/RunEnvForLocalDesigner.run.xml b/.run/RunEnvForLocalDesigner.run.xml
index 8cb48532e92..22cf383bd77 100644
--- a/.run/RunEnvForLocalDesigner.run.xml
+++ b/.run/RunEnvForLocalDesigner.run.xml
@@ -12,4 +12,4 @@
       <option name="Make" enabled="true" />
     </method>
   </configuration>
-</component>
\ No newline at end of file
+</component>
diff --git a/designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentManager.scala b/designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentManager.scala
index 0dd52caaf54..f5fa9c7ddfc 100644
--- a/designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentManager.scala
+++ b/designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentManager.scala
@@ -1,6 +1,6 @@
 package pl.touk.nussknacker.engine.api.deployment
 
-import cats.effect.{IO, Resource, SyncIO}
+import cats.effect.{Resource, SyncIO}
 import com.typesafe.config.Config
 import pl.touk.nussknacker.engine.api.definition.EngineScenarioCompilationDependencies
 import pl.touk.nussknacker.engine.api.deployment.scheduler.services._
@@ -35,7 +35,6 @@ trait DeploymentManager extends AutoCloseable {
 
   protected final def notImplemented: Future[Nothing] =
     Future.failed(new NotImplementedError())
-
 }
 
 trait ManagerSpecificScenarioActivitiesStoredByManager { self: DeploymentManager =>
@@ -90,3 +89,8 @@ trait SchedulingSupported extends SchedulingSupport {
 }
 
 case object NoSchedulingSupport extends SchedulingSupport
+
+class ActiveScenariosLimitExceeded(activeScenariosCount: Int)
+    extends IllegalArgumentException(
+      s"Cannot deploy scenario. Active scenarios count is $activeScenariosCount"
+    )
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/api/DeploymentApiHttpService.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/api/DeploymentApiHttpService.scala
index 11bf71667e7..aba8e6627c7 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/api/DeploymentApiHttpService.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/api/DeploymentApiHttpService.scala
@@ -57,6 +57,10 @@ class DeploymentApiHttpService(
                     ScenarioGraphValidationError(errors)
                   case DeploymentService.DeployValidationError(message) =>
                     DeployValidationError(message)
+                  case DeploymentService.ActiveScenariosLimitExceededError(limit) =>
+                    DeployValidationError(
+                      s"The limit of active scenarios has been reached. You can have a maximum of $limit active scenarios."
+                    )
                 }
               case ActivityService.CommentValidationError(message) => CommentValidationError(message)
             })
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/factory/DomainServices.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/factory/DomainServices.scala
index cd7a409aaeb..a25d4dbcdb2 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/factory/DomainServices.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/factory/DomainServices.scala
@@ -7,7 +7,6 @@ import pl.touk.nussknacker.engine.api.component.{AdditionalUIConfigProvider, Des
 import pl.touk.nussknacker.engine.api.process.ProcessingType
 import pl.touk.nussknacker.engine.classloader.{DeploymentManagersClassLoader, DeploymentManagersClassLoaderFactory}
 import pl.touk.nussknacker.engine.definition.component.Components.ComponentDefinitionExtractionMode
-import pl.touk.nussknacker.engine.util.ExecutionContextWithIORuntime
 import pl.touk.nussknacker.engine.util.Implicits.RichScalaMap
 import pl.touk.nussknacker.processCounts.CountsReporter
 import pl.touk.nussknacker.ui.api.{AuthorizeProcess, ScenarioStatusPresenter}
@@ -172,7 +171,7 @@ object DomainServices extends LazyLogging {
         new EngineSideDeploymentStatusesProvider(
           dmDispatcher,
           alreadyLoadedConfig.scenarioStateTimeout
-        )(actorSystem)
+        )
       processRepository = DBFetchingProcessRepository.create(dbRef, actionRepository, scenarioLabelsRepository)
       scenarioStatusProvider = new ScenarioStatusProvider(
         deploymentsStatusesProvider,
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/DeploymentService.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/DeploymentService.scala
index 61fefec7d21..938587592d0 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/DeploymentService.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/DeploymentService.scala
@@ -2,17 +2,16 @@ package pl.touk.nussknacker.ui.process.deployment
 
 import cats.data.Validated
 import com.typesafe.scalalogging.LazyLogging
-import pl.touk.nussknacker.engine.api.component.{
-  ComponentAdditionalConfig,
-  DesignerWideComponentId,
-  NodesDeploymentData
-}
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.ActiveScenariosLimit
+import pl.touk.nussknacker.engine.api.component.{ComponentAdditionalConfig, DesignerWideComponentId}
 import pl.touk.nussknacker.engine.api.deployment._
 import pl.touk.nussknacker.engine.api.process._
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.deployment._
 import pl.touk.nussknacker.engine.util.AdditionalComponentConfigsForRuntimeExtractor
+import pl.touk.nussknacker.ui.process.deployment.DeploymentService.ActiveScenariosLimitExceededError
 import pl.touk.nussknacker.ui.process.deployment.LoggedUserConversions.LoggedUserOps
+import pl.touk.nussknacker.ui.process.deployment.scenariostatus.ScenarioStatusProvider
 import pl.touk.nussknacker.ui.process.exception.DeployingInvalidScenarioError
 import pl.touk.nussknacker.ui.process.processingtype.provider.ProcessingTypeDataProvider
 import pl.touk.nussknacker.ui.process.repository._
@@ -33,6 +32,8 @@ class DeploymentService(
       Map[DesignerWideComponentId, ComponentAdditionalConfig],
       _
     ],
+    scenarioStatusProvider: ScenarioStatusProvider,
+    activeScenariosLimitProvider: ProcessingTypeDataProvider[Option[ActiveScenariosLimit], _],
 )(implicit ec: ExecutionContext)
     extends LazyLogging {
 
@@ -83,7 +84,6 @@ class DeploymentService(
 
   private def runDeployment(command: RunDeploymentCommand): Future[Future[Option[ExternalDeploymentId]]] = {
     import command.commonData._
-
     actionService
       .actionProcessorForLatestVersion[CanonicalProcess]
       .processActionWithCustomFinalization[RunDeploymentCommand, Future[Option[ExternalDeploymentId]]](
@@ -136,7 +136,9 @@ class DeploymentService(
           Future.failed(DeployingInvalidScenarioError(validationResult.errors))
         case _ => Future.successful(())
       }
-      // 2. deployment managers specific checks
+      // 2. limits check
+      _ <- checkActiveScenariosLimits(processDetails.processingType, user)
+      // 3. deployment managers specific checks
       // TODO: scenario was already resolved during validation - use it here
       _ <- dispatcher
         .deploymentManagerUnsafe(processDetails.processingType)
@@ -151,6 +153,22 @@ class DeploymentService(
     } yield ()
   }
 
+  private def checkActiveScenariosLimits(processingType: ProcessingType, deployer: LoggedUser) = {
+    implicit val loggedUser: LoggedUser = deployer
+    activeScenariosLimitProvider.forProcessingType(processingType).flatten match {
+      case Some(ActiveScenariosLimit(activeScenariosLimit)) =>
+        scenarioStatusProvider
+          .getActiveScenariosCountFor(processingType)
+          .map { activeScenariosCount =>
+            if (activeScenariosCount >= activeScenariosLimit) {
+              throw ActiveScenariosLimitExceededError(activeScenariosLimit)
+            }
+          }
+      case None =>
+        Future.unit
+    }
+  }
+
   private def prepareDMRunDeploymentCommand(
       processDetails: ScenarioWithDetailsEntity[CanonicalProcess],
       actionId: ProcessActionId,
@@ -192,3 +210,12 @@ class DeploymentService(
   }
 
 }
+
+object DeploymentService {
+
+  final case class ActiveScenariosLimitExceededError(activeScenariosLimit: Int)
+      extends IllegalArgumentException(
+        s"The limit of active scenarios has been reached. You can have a maximum of $activeScenariosLimit active scenarios."
+      )
+
+}
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/deploymentstatus/DeploymentManagerReliableStatusesWrapper.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/deploymentstatus/DeploymentManagerReliableStatusesWrapper.scala
index c9b38425f44..a070885475b 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/deploymentstatus/DeploymentManagerReliableStatusesWrapper.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/deploymentstatus/DeploymentManagerReliableStatusesWrapper.scala
@@ -1,12 +1,12 @@
 package pl.touk.nussknacker.ui.process.deployment.deploymentstatus
 
-import org.apache.pekko.actor.ActorSystem
+import cats.effect.IO
 import pl.touk.nussknacker.engine.api.deployment.{DataFreshnessPolicy, DeploymentStatusDetails, WithDataFreshnessStatus}
 import pl.touk.nussknacker.engine.api.process.{ProcessingType, ProcessName}
+import pl.touk.nussknacker.engine.util.ExecutionContextWithIORuntime
 import pl.touk.nussknacker.ui.process.deployment.DeploymentManagerDispatcher
 import pl.touk.nussknacker.ui.process.repository.ScenarioIdData
 import pl.touk.nussknacker.ui.security.api.LoggedUser
-import pl.touk.nussknacker.ui.util.FutureUtils.FutureOps
 
 import scala.concurrent.Future
 import scala.concurrent.duration.FiniteDuration
@@ -22,11 +22,10 @@ object DeploymentManagerReliableStatusesWrapper {
     )(
         implicit user: LoggedUser,
         freshnessPolicy: DataFreshnessPolicy,
-        actorSystem: ActorSystem
+        executionContextWithIORuntime: ExecutionContextWithIORuntime
     ): Future[Either[GetDeploymentsStatusesError, WithDataFreshnessStatus[List[DeploymentStatusDetails]]]] = {
-      import actorSystem._
-      val deploymentStatusesOptFuture
-          : Future[Either[GetDeploymentsStatusesError, WithDataFreshnessStatus[List[DeploymentStatusDetails]]]] =
+      import executionContextWithIORuntime.ioRuntime
+      val deploymentStatusesOpt = IO.fromFuture(IO {
         dmDispatcher
           .deploymentManager(scenarioIdData.processingType)
           .map(
@@ -37,13 +36,14 @@ object DeploymentManagerReliableStatusesWrapper {
           .getOrElse(
             Future.successful(Left(ProcessingTypeIsNotConfigured(scenarioIdData.name, scenarioIdData.processingType)))
           )
+      })
 
       timeoutOpt
         .map { timeout =>
-          deploymentStatusesOptFuture
-            .withTimeout(timeout, timeoutResult = Left(GetDeploymentsStatusTimeout(scenarioIdData.name)))
+          deploymentStatusesOpt.timeoutTo(timeout, IO.delay(Left(GetDeploymentsStatusTimeout(scenarioIdData.name))))
         }
-        .getOrElse(deploymentStatusesOptFuture)
+        .getOrElse(deploymentStatusesOpt)
+        .unsafeToFuture()
     }
 
   }
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/deploymentstatus/EngineSideDeploymentStatusesProvider.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/deploymentstatus/EngineSideDeploymentStatusesProvider.scala
index ae67af6166a..5aaec3acd3e 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/deploymentstatus/EngineSideDeploymentStatusesProvider.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/deploymentstatus/EngineSideDeploymentStatusesProvider.scala
@@ -1,17 +1,17 @@
 package pl.touk.nussknacker.ui.process.deployment.deploymentstatus
 
 import com.typesafe.scalalogging.LazyLogging
-import org.apache.pekko.actor.ActorSystem
 import pl.touk.nussknacker.engine.api.deployment._
 import pl.touk.nussknacker.engine.api.deployment.simple.SimpleStateStatus.ProblemStateStatus
 import pl.touk.nussknacker.engine.api.process.{ProcessingType, ProcessName}
+import pl.touk.nussknacker.engine.util.ExecutionContextWithIORuntime
 import pl.touk.nussknacker.engine.util.WithDataFreshnessStatusUtils.WithDataFreshnessStatusMapOps
 import pl.touk.nussknacker.ui.process.deployment.DeploymentManagerDispatcher
 import pl.touk.nussknacker.ui.process.deployment.deploymentstatus.DeploymentManagerReliableStatusesWrapper.Ops
 import pl.touk.nussknacker.ui.process.repository._
 import pl.touk.nussknacker.ui.security.api.LoggedUser
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.Future
 import scala.concurrent.duration.FiniteDuration
 import scala.util.control.NonFatal
 
@@ -30,11 +30,9 @@ class EngineSideDeploymentStatusesProvider(
     dispatcher: DeploymentManagerDispatcher,
     scenarioStateTimeout: Option[FiniteDuration]
 )(
-    implicit system: ActorSystem
+    implicit executionContextWithIORuntime: ExecutionContextWithIORuntime
 ) extends LazyLogging {
 
-  private implicit val ec: ExecutionContext = system.dispatcher
-
   // DeploymentManager's may support fetching state of all scenarios at once
   // State is prefetched only when:
   //  - DM has capability DeploymentsStatusesQueryForAllScenariosSupport
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/scenariostatus/ScenarioStatusProvider.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/scenariostatus/ScenarioStatusProvider.scala
index 22249cda1b4..89771071f1a 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/scenariostatus/ScenarioStatusProvider.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/deployment/scenariostatus/ScenarioStatusProvider.scala
@@ -78,6 +78,20 @@ class ScenarioStatusProvider(
     )
   }
 
+  def getActiveScenariosCountFor(processingType: ProcessingType)(implicit user: LoggedUser): Future[Int] = {
+    implicit val freshnessStatus: DataFreshnessPolicy = DataFreshnessPolicy.Fresh
+    deploymentStatusesProvider
+      .getBulkQueriedDeploymentStatusesForSupportedManagers(processingType :: Nil)
+      .map { statuses =>
+        statuses.getAllDeploymentStatuses
+          .count { status =>
+            status.status == SimpleStateStatus.DuringDeploy ||
+            status.status == SimpleStateStatus.Running ||
+            status.status == SimpleStateStatus.Restarting
+          }
+      }
+  }
+
   private def getNonFragmentScenarioStatus[ScenarioShape, F[_]: Traverse](
       actionsInProgress: Map[ProcessId, Set[ScenarioActionName]],
       prefetchedDeploymentStatuses: BulkQueriedDeploymentStatuses,
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/newdeployment/DeploymentService.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/newdeployment/DeploymentService.scala
index 2450f40574e..dbd1cde4c0e 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/newdeployment/DeploymentService.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/newdeployment/DeploymentService.scala
@@ -4,6 +4,7 @@ import cats.Applicative
 import cats.data.{EitherT, NonEmptyList}
 import com.typesafe.scalalogging.LazyLogging
 import db.util.DBIOActionInstances._
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.ActiveScenariosLimit
 import pl.touk.nussknacker.engine.api.{ProcessVersion => RuntimeVersionData}
 import pl.touk.nussknacker.engine.api.component.{
   ComponentAdditionalConfig,
@@ -26,6 +27,7 @@ import pl.touk.nussknacker.ui.db.entity.ProcessVersionEntityData
 import pl.touk.nussknacker.ui.process.ScenarioMetadata
 import pl.touk.nussknacker.ui.process.deployment.DeploymentManagerDispatcher
 import pl.touk.nussknacker.ui.process.deployment.LoggedUserConversions.LoggedUserOps
+import pl.touk.nussknacker.ui.process.deployment.scenariostatus.ScenarioStatusProvider
 import pl.touk.nussknacker.ui.process.newdeployment.DeploymentEntityFactory.{DeploymentEntityData, WithModifiedAt}
 import pl.touk.nussknacker.ui.process.newdeployment.DeploymentService._
 import pl.touk.nussknacker.ui.process.processingtype.provider.ProcessingTypeDataProvider
@@ -55,6 +57,8 @@ class DeploymentService(
       Map[DesignerWideComponentId, ComponentAdditionalConfig],
       _
     ],
+    scenarioStatusProvider: ScenarioStatusProvider,
+    activeScenariosLimitProvider: ProcessingTypeDataProvider[Option[ActiveScenariosLimit], _],
 )(implicit ec: ExecutionContext)
     extends LazyLogging {
 
@@ -78,12 +82,8 @@ class DeploymentService(
         category = scenarioMetadata.processCategory,
         permission = Permission.Deploy
       )
-      _ <- EitherT.cond[Future](!scenarioMetadata.isFragment, (), DeploymentOfFragmentError)
-      _ <- EitherT.cond[Future](!scenarioMetadata.isArchived, (), DeploymentOfArchivedScenarioError)
-      scenarioGraphVersion <- EitherT(
-        scenarioGraphVersionService.getValidResolvedLatestScenarioGraphVersion(scenarioMetadata, command.user)
-      ).leftMap[RunDeploymentError](error => ScenarioGraphValidationError(error.errors))
-      _ <- validateUsingDeploymentManager(scenarioMetadata, scenarioGraphVersion, command.user)
+      scenarioGraphVersion <- getScenarioGraphVersion(scenarioMetadata, command.user)
+      _                    <- validateDeployment(scenarioMetadata, scenarioGraphVersion, command.user)
       // We keep deployments metrics (used by counts mechanism) keyed by scenario name.
       // Because of that we can't run more than one deployment for scenario in a time.
       // TODO: We should key metrics by deployment id and remove this limitation
@@ -92,6 +92,25 @@ class DeploymentService(
       _ <- runDeploymentUsingDeploymentManagerAsync(scenarioMetadata, scenarioGraphVersion, command)
     } yield DeploymentForeignKeys(scenarioMetadata.id, scenarioGraphVersion.id)).value
 
+  private def validateDeployment(
+      scenarioMetadata: ScenarioMetadata,
+      scenarioGraphVersion: ProcessVersionEntityData,
+      deployer: LoggedUser
+  ) = {
+    for {
+      _ <- EitherT.cond[Future](!scenarioMetadata.isFragment, (), DeploymentOfFragmentError)
+      _ <- EitherT.cond[Future](!scenarioMetadata.isArchived, (), DeploymentOfArchivedScenarioError)
+      _ <- checkActiveScenariosLimits(scenarioMetadata, deployer)
+      _ <- runDeploymentManagerSpecificValidations(scenarioMetadata, scenarioGraphVersion, deployer)
+    } yield ()
+  }
+
+  private def getScenarioGraphVersion(scenarioMetadata: ScenarioMetadata, deployer: LoggedUser) = {
+    EitherT(
+      scenarioGraphVersionService.getValidResolvedLatestScenarioGraphVersion(scenarioMetadata, deployer)
+    ).leftMap[RunDeploymentError](error => ScenarioGraphValidationError(error.errors))
+  }
+
   private def getScenarioMetadata(
       command: RunDeploymentCommand
   ): EitherT[Future, RunDeploymentError, ScenarioMetadata] =
@@ -107,9 +126,7 @@ class DeploymentService(
     EitherT(dbioRunner.runInSerializableTransactionWithRetry((for {
       nonFinishedDeployments <- getConcurrentlyPerformedDeploymentsForScenario(scenarioMetadata)
       _                      <- checkNoConcurrentDeploymentsForScenario(nonFinishedDeployments, scenarioMetadata.name)
-      _ = {
-        logger.debug(s"Saving deployment: ${command.id}")
-      }
+      _ = logger.debug(s"Saving deployment: ${command.id}")
       _ <- saveDeployment(command, scenarioMetadata)
     } yield ()).value))
   }
@@ -157,7 +174,30 @@ class DeploymentService(
     ).leftMap(e => ConflictingDeploymentIdError(e.id))
   }
 
-  private def validateUsingDeploymentManager(
+  private def checkActiveScenariosLimits(
+      scenarioMetadata: ScenarioMetadata,
+      deployer: LoggedUser
+  ): EitherT[Future, RunDeploymentError, Unit] = {
+    implicit val loggedUser: LoggedUser = deployer
+    activeScenariosLimitProvider.forProcessingType(scenarioMetadata.processingType).flatten match {
+      case Some(ActiveScenariosLimit(activeScenariosLimit)) =>
+        EitherT {
+          scenarioStatusProvider
+            .getActiveScenariosCountFor(scenarioMetadata.processingType)
+            .map { activeScenariosCount =>
+              Either.cond(
+                test = activeScenariosCount < activeScenariosLimit,
+                right = (),
+                left = ActiveScenariosLimitExceededError(activeScenariosLimit)
+              )
+            }
+        }
+      case None =>
+        EitherT.pure(())
+    }
+  }
+
+  private def runDeploymentManagerSpecificValidations(
       scenarioMetadata: ScenarioMetadata,
       scenarioGraphVersion: ProcessVersionEntityData,
       user: LoggedUser
@@ -312,6 +352,8 @@ object DeploymentService {
 
   case object DeploymentOfArchivedScenarioError extends RunDeploymentError
 
+  final case class ActiveScenariosLimitExceededError(activeScenariosLimit: Int) extends RunDeploymentError
+
   final case class DeploymentNotFoundError(id: DeploymentId) extends GetDeploymentStatusError
 
   case object NoPermissionError extends RunDeploymentError with GetDeploymentStatusError
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeData.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeData.scala
index 86087a2c959..b77b8b53279 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeData.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeData.scala
@@ -2,6 +2,7 @@ package pl.touk.nussknacker.ui.process.processingtype
 
 import cats.effect.SyncIO
 import pl.touk.nussknacker.engine._
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.ActiveScenariosLimit
 import pl.touk.nussknacker.engine.api.{JobData, ProcessVersion}
 import pl.touk.nussknacker.engine.api.component.ScenarioPropertyConfig
 import pl.touk.nussknacker.engine.api.definition.EngineScenarioCompilationDependencies
@@ -22,6 +23,7 @@ final class ProcessingTypeData private (
     //       to fully split deployment managers from model
     val deploymentData: DeploymentData,
     val category: String,
+    val activeScenariosLimit: Option[ActiveScenariosLimit]
 ) {
 
   // TODO: We should allow to have >1 processing mode configured inside one model and return a List here
@@ -49,6 +51,7 @@ object ProcessingTypeData {
       modelData: ModelData,
       deploymentData: DeploymentData,
       category: String,
+      activeScenariosLimit: Option[ActiveScenariosLimit],
       componentDefinitionExtractionMode: ComponentDefinitionExtractionMode
   ): ProcessingTypeData = {
     val designerModelData =
@@ -62,7 +65,8 @@ object ProcessingTypeData {
       processingType,
       designerModelData,
       deploymentData,
-      category
+      category,
+      activeScenariosLimit
     )
   }
 
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeServices.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeServices.scala
index 7c6019444ac..ef3b55d6f6c 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeServices.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeServices.scala
@@ -1,5 +1,6 @@
 package pl.touk.nussknacker.ui.process.processingtype
 
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.ActiveScenariosLimit
 import pl.touk.nussknacker.engine.api.component.{
   AdditionalUIConfigProvider,
   ComponentAdditionalConfig,
@@ -52,6 +53,8 @@ final class ProcessingTypeServices private (
 
   def category: String = processingTypeData.category
 
+  def activeScenariosLimit: Option[ActiveScenariosLimit] = processingTypeData.activeScenariosLimit
+
   def designerModelData: DesignerModelData = processingTypeData.designerModelData
 
   // TODO: We should replace all usages of this method with access to DeploymentData which is created separately from model
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/loader/ModelDataLoader.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/loader/ModelDataLoader.scala
index a8e8f0f5629..6b0dbc561d3 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/loader/ModelDataLoader.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/loader/ModelDataLoader.scala
@@ -62,6 +62,7 @@ object ModelDataLoader extends LazyLogging {
     val modelDataWithInput = new ModelDataWithProcessingTypeDataInput(
       modelData,
       processingTypeConfig.category,
+      processingTypeConfig.activeScenariosLimit,
       modelDependencies.componentDefinitionExtractionMode
     )
     modelDataWithInput
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/loader/ProcessingTypeDataStateFactory.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/loader/ProcessingTypeDataStateFactory.scala
index 8b2981a93ba..1e43c7c9c6f 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/loader/ProcessingTypeDataStateFactory.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/process/processingtype/loader/ProcessingTypeDataStateFactory.scala
@@ -3,6 +3,7 @@ package pl.touk.nussknacker.ui.process.processingtype.loader
 import cats.data.Validated
 import com.typesafe.scalalogging.LazyLogging
 import pl.touk.nussknacker.engine._
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.ActiveScenariosLimit
 import pl.touk.nussknacker.engine.api.process.ProcessingType
 import pl.touk.nussknacker.engine.definition.component.Components.ComponentDefinitionExtractionMode
 import pl.touk.nussknacker.engine.util.Implicits.RichScalaMap
@@ -33,6 +34,7 @@ object ProcessingTypeDataStateFactory extends LazyLogging {
             modelDataWithInput.modelData,
             deploymentData,
             modelDataWithInput.category,
+            modelDataWithInput.activeScenariosLimit,
             modelDataWithInput.componentDefinitionExtractionMode,
           )
           processingTypeData
@@ -64,6 +66,7 @@ object ProcessingTypeDataStateFactory extends LazyLogging {
   final class ModelDataWithProcessingTypeDataInput(
       val modelData: ModelData,
       val category: String,
+      val activeScenariosLimit: Option[ActiveScenariosLimit],
       val componentDefinitionExtractionMode: ComponentDefinitionExtractionMode
   ) extends AutoCloseable {
 
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/server/PekkoRoutesFactory.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/server/PekkoRoutesFactory.scala
index caa571c717d..e488d606aa8 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/server/PekkoRoutesFactory.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/server/PekkoRoutesFactory.scala
@@ -74,7 +74,9 @@ object PekkoRoutesFactory {
         processingTypeServicesProvider.mapValues(_.scenarioValidator),
         processingTypeServicesProvider.mapValues(_.scenarioResolver),
         actionService,
-        processingTypeServicesProvider.mapValues(_.additionalComponentConfigs)
+        processingTypeServicesProvider.mapValues(_.additionalComponentConfigs),
+        scenarioStatusProvider,
+        processingTypeServicesProvider.mapValues(_.activeScenariosLimit)
       )
       new ManagementResources(
         processAuthorizer,
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/server/TapirHttpServiceFactory.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/server/TapirHttpServiceFactory.scala
index 42453e612ec..70e0c5a0d83 100644
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/server/TapirHttpServiceFactory.scala
+++ b/designer/server/src/main/scala/pl/touk/nussknacker/ui/server/TapirHttpServiceFactory.scala
@@ -171,7 +171,9 @@ object TapirHttpServiceFactory {
           dmDispatcher,
           dbioRunner,
           clock,
-          processingTypeServicesProvider.mapValues(_.additionalComponentConfigs)
+          processingTypeServicesProvider.mapValues(_.additionalComponentConfigs),
+          scenarioStatusProvider,
+          processingTypeServicesProvider.mapValues(_.activeScenariosLimit),
         )
       val activityService =
         new ActivityService(
diff --git a/designer/server/src/main/scala/pl/touk/nussknacker/ui/util/FutureUtils.scala b/designer/server/src/main/scala/pl/touk/nussknacker/ui/util/FutureUtils.scala
deleted file mode 100644
index 5a1b186a7fb..00000000000
--- a/designer/server/src/main/scala/pl/touk/nussknacker/ui/util/FutureUtils.scala
+++ /dev/null
@@ -1,28 +0,0 @@
-package pl.touk.nussknacker.ui.util
-
-import org.apache.pekko.actor.ActorSystem
-
-import scala.concurrent.Future
-import scala.concurrent.duration.FiniteDuration
-
-object FutureUtils {
-
-  implicit class FutureOps[T](future: Future[T]) {
-
-    // Limits Future completion to some timeout, after which timeoutResult suppress result of given future.
-    // This solution is based on: https://stackoverflow.com/a/42468372/1370301
-    def withTimeout(duration: FiniteDuration, timeoutResult: => T)(
-        implicit actorSystem: ActorSystem
-    ): Future[T] = {
-      import actorSystem._
-      Future.firstCompletedOf(
-        Seq(
-          org.apache.pekko.pattern.after(duration)(Future.successful(timeoutResult)),
-          future
-        )
-      )
-    }
-
-  }
-
-}
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/test/base/it/NuResourcesTest.scala b/designer/server/src/test/scala/pl/touk/nussknacker/test/base/it/NuResourcesTest.scala
index 4d80874e65a..cdb7a9071d0 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/test/base/it/NuResourcesTest.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/test/base/it/NuResourcesTest.scala
@@ -27,6 +27,7 @@ import pl.touk.nussknacker.engine.api.process._
 import pl.touk.nussknacker.engine.api.process.VersionId.initialVersionId
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.definition.test.{ModelDataTestInfoProvider, TestInfoProvider}
+import pl.touk.nussknacker.engine.util.ExecutionContextWithIORuntimeAdapter
 import pl.touk.nussknacker.restmodel.{CancelRequest, DeployRequest}
 import pl.touk.nussknacker.restmodel.scenariodetails.ScenarioWithDetails
 import pl.touk.nussknacker.security.Permission
@@ -84,6 +85,9 @@ trait NuResourcesTest
     with LazyLogging {
   self: ScalatestRouteTest with Suite with Matchers with ScalaFutures =>
 
+  private val executionContextWithIORuntime: ExecutionContextWithIORuntimeAdapter =
+    ExecutionContextWithIORuntimeAdapter.unsafeCreateFrom(executor)
+
   protected val adminUser: LoggedUser = TestFactory.adminUser("user")
 
   private implicit val implicitAdminUser: LoggedUser = adminUser
@@ -124,7 +128,7 @@ trait NuResourcesTest
   )
 
   protected val deploymentsStatusesProvider =
-    new EngineSideDeploymentStatusesProvider(dmDispatcher, None)
+    new EngineSideDeploymentStatusesProvider(dmDispatcher, None)(executionContextWithIORuntime)
 
   protected val scenarioStatusProvider: ScenarioStatusProvider = new ScenarioStatusProvider(
     deploymentsStatusesProvider,
@@ -148,11 +152,13 @@ trait NuResourcesTest
 
   protected val deploymentService: DeploymentService =
     new DeploymentService(
-      dmDispatcher,
-      processValidatorByProcessingType,
-      scenarioResolverByProcessingType,
-      actionService,
-      mapProcessingTypeDataProvider(),
+      dispatcher = dmDispatcher,
+      processValidator = processValidatorByProcessingType,
+      scenarioResolver = scenarioResolverByProcessingType,
+      actionService = actionService,
+      additionalComponentConfigs = mapProcessingTypeDataProvider(),
+      scenarioStatusProvider = scenarioStatusProvider,
+      activeScenariosLimitProvider = TestFactory.mapProcessingTypeDataProvider(Streaming.stringify -> None)
     )
 
   protected val processingTypeConfig: ProcessingTypeConfig =
@@ -186,11 +192,12 @@ trait NuResourcesTest
   protected val testProcessingTypeDataProvider: ProcessingTypeDataProvider[ProcessingTypeData, _] =
     mapProcessingTypeDataProvider(
       Streaming.stringify -> ProcessingTypeData.createProcessingTypeData(
-        Streaming.stringify,
-        modelData,
-        deploymentData,
-        processingTypeConfig.category,
-        modelDependencies.componentDefinitionExtractionMode
+        processingType = Streaming.stringify,
+        modelData = modelData,
+        deploymentData = deploymentData,
+        category = processingTypeConfig.category,
+        activeScenariosLimit = None,
+        componentDefinitionExtractionMode = modelDependencies.componentDefinitionExtractionMode
       )
     )
 
@@ -293,6 +300,7 @@ trait NuResourcesTest
   override protected def afterAll(): Unit = {
     super.afterAll()
     closeDeploymentManagers.unsafeRunSync()
+    executionContextWithIORuntime.close()
   }
 
   protected def saveCanonicalProcessAndAssertSuccess(
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/test/mock/MockDeploymentManager.scala b/designer/server/src/test/scala/pl/touk/nussknacker/test/mock/MockDeploymentManager.scala
index 445b09eba73..5eafb21d014 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/test/mock/MockDeploymentManager.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/test/mock/MockDeploymentManager.scala
@@ -29,7 +29,7 @@ import java.util.concurrent.{ConcurrentHashMap, ConcurrentLinkedQueue}
 import scala.concurrent.{ExecutionContext, Future, Promise}
 import scala.concurrent.duration._
 import scala.jdk.CollectionConverters._
-import scala.util.{Failure, Success}
+import scala.util.Success
 import scala.util.control.NonFatal
 
 // DEPRECATED!!! Use `WithMockableDeploymentManager` trait and `MockableDeploymentManager` instead
@@ -104,7 +104,6 @@ class MockDeploymentManager private (
 
   override protected def runDeployment(command: DMRunDeploymentCommand): Future[Option[ExternalDeploymentId]] = {
     import command._
-
     deployResult
       .getOrDefault(
         processVersion.processName,
@@ -184,10 +183,12 @@ object MockDeploymentManager {
       actorSystem,
       SttpBackendStub.asynchronousFuture
     )
+
     def closeCreatedDeps(): Unit = {
       closeDeploymentManagerClassLoader.unsafeRunSync()(IORuntime.global)
       actorSystem.terminate()
     }
+
     new MockDeploymentManager(
       modelData,
       deploymentManagerDependencies,
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/component/DefaultComponentServiceSpec.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/component/DefaultComponentServiceSpec.scala
index 5b79a7c535c..024c985f292 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/component/DefaultComponentServiceSpec.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/component/DefaultComponentServiceSpec.scala
@@ -888,6 +888,7 @@ class DefaultComponentServiceSpec
           modelData,
           deploymentData,
           category = category,
+          activeScenariosLimit = None,
           ComponentDefinitionExtractionMode.FinalDefinition
         )
     }
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/notifications/NotificationServiceTest.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/notifications/NotificationServiceTest.scala
index b1614d213ab..8bacf4345a0 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/notifications/NotificationServiceTest.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/notifications/NotificationServiceTest.scala
@@ -16,6 +16,7 @@ import pl.touk.nussknacker.engine.api.process.{ProcessIdWithName, ProcessName}
 import pl.touk.nussknacker.engine.build.ScenarioBuilder
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.deployment.{DeploymentId, ExternalDeploymentId}
+import pl.touk.nussknacker.engine.util.ExecutionContextWithIORuntimeAdapter
 import pl.touk.nussknacker.test.{EitherValuesDetailedMessage, PatientScalaFutures}
 import pl.touk.nussknacker.test.base.db.WithHsqlDbTesting
 import pl.touk.nussknacker.test.config.WithSimplifiedDesignerConfig.TestProcessingType.Streaming
@@ -41,7 +42,6 @@ import pl.touk.nussknacker.ui.util.InMemoryTimeseriesRepository
 
 import java.time.{Clock, Duration, Instant, ZoneId}
 import java.time.temporal.ChronoUnit
-import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import scala.concurrent.duration.DurationInt
 import scala.util.{Failure, Success, Try}
@@ -57,7 +57,10 @@ class NotificationServiceTest
     with DBIOActionValues
     with BeforeAndAfterAll {
 
-  private implicit val system: ActorSystem            = ActorSystem(getClass.getSimpleName)
+  private implicit val system: ActorSystem = ActorSystem(getClass.getSimpleName)
+  private implicit val executionContextWithIORuntime: ExecutionContextWithIORuntimeAdapter =
+    ExecutionContextWithIORuntimeAdapter.unsafeCreateFrom(system.dispatcher)
+
   override protected val dbioRunner: DBIOActionRunner = DBIOActionRunner(testDbRef)
 
   private var currentInstant: Instant    = Instant.ofEpochMilli(0)
@@ -88,6 +91,7 @@ class NotificationServiceTest
   override protected def afterAll(): Unit = {
     super.afterAll()
     system.terminate().futureValue
+    executionContextWithIORuntime.close()
   }
 
   test("Should return only events for user in given time") {
@@ -100,7 +104,7 @@ class NotificationServiceTest
 
     def notificationsFor(user: LoggedUser): List[Notification] =
       notificationService
-        .notifications(NotificationsScope.NotificationsForLoggedUser(user))(global)
+        .notifications(NotificationsScope.NotificationsForLoggedUser(user))
         .futureValue
 
     def deployProcess(
@@ -151,12 +155,12 @@ class NotificationServiceTest
     val id                = saveSampleProcess(processName)
     val processIdWithName = ProcessIdWithName(id, processName)
 
-    val deploymentManager                                       = mock[DeploymentManager]
-    val (deploymentService, actionService, notificationService) = createServices(deploymentManager)
+    val deploymentManager                           = mock[DeploymentManager]
+    val (deploymentService, _, notificationService) = createServices(deploymentManager)
 
     def notificationsFor(user: LoggedUser): List[Notification] =
       notificationService
-        .notifications(NotificationsScope.NotificationsForLoggedUserAndScenario(user, processName))(global)
+        .notifications(NotificationsScope.NotificationsForLoggedUserAndScenario(user, processName))
         .futureValue
 
     def deployProcess(
@@ -188,7 +192,7 @@ class NotificationServiceTest
     val notifications = notificationsFor(secondUser)
     notifications shouldBe List(
       Notification(
-        notifications(0).id,
+        notifications.head.id,
         Some(processName),
         "Deployment finished",
         None,
@@ -254,7 +258,7 @@ class NotificationServiceTest
     deployProcess(Success(None), user)
     val notificationsAfterDeploy =
       notificationService
-        .notifications(NotificationsScope.NotificationsForLoggedUser(user))(global)
+        .notifications(NotificationsScope.NotificationsForLoggedUser(user))
         .futureValue
 
     notificationsAfterDeploy should have length 1
@@ -265,7 +269,7 @@ class NotificationServiceTest
       .futureValue
     val notificationAfterExecutionFinished =
       notificationService
-        .notifications(NotificationsScope.NotificationsForLoggedUser(user))(global)
+        .notifications(NotificationsScope.NotificationsForLoggedUser(user))
         .futureValue
     // old notification about deployment is replaced by notification about deployment execution finished which has other id
     notificationAfterExecutionFinished should have length 1
@@ -312,6 +316,8 @@ class NotificationServiceTest
       TestFactory.scenarioResolverByProcessingType,
       actionService,
       TestFactory.additionalComponentConfigsByProcessingType,
+      scenarioStatusProvider,
+      TestFactory.mapProcessingTypeDataProvider(Streaming.stringify -> None)
     ) {
       override protected def validateBeforeDeploy(
           processDetails: ScenarioWithDetailsEntity[CanonicalProcess],
@@ -342,11 +348,9 @@ class NotificationServiceTest
 
   private def clockForInstant(currentInstant: () => Instant): Clock = {
     new Clock {
-      override def getZone: ZoneId = ZoneId.systemDefault()
-
+      override def getZone: ZoneId               = ZoneId.systemDefault()
       override def withZone(zone: ZoneId): Clock = ???
-
-      override def instant(): Instant = currentInstant()
+      override def instant(): Instant            = currentInstant()
     }
   }
 
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/ProcessStateDefinitionServiceSpec.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/ProcessStateDefinitionServiceSpec.scala
index 2574767af58..4252a6ce1fc 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/ProcessStateDefinitionServiceSpec.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/ProcessStateDefinitionServiceSpec.scala
@@ -207,7 +207,8 @@ class ProcessStateDefinitionServiceSpec extends AnyFunSuite with Matchers {
       ),
       deploymentData,
       category = category,
-      componentDefinitionExtractionMode = modelDependencies.componentDefinitionExtractionMode
+      activeScenariosLimit = None,
+      componentDefinitionExtractionMode = modelDependencies.componentDefinitionExtractionMode,
     )
   }
 
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/DeploymentServiceSpec.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/DeploymentServiceSpec.scala
index e1336836756..e5d4366af09 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/DeploymentServiceSpec.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/DeploymentServiceSpec.scala
@@ -5,8 +5,9 @@ import cats.instances.list._
 import db.util.DBIOActionInstances.DB
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, OptionValues}
 import org.scalatest.LoneElement._
-import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.ActiveScenariosLimit
 import pl.touk.nussknacker.engine.api.Comment
 import pl.touk.nussknacker.engine.api.component.NodesDeploymentData
 import pl.touk.nussknacker.engine.api.deployment._
@@ -29,6 +30,7 @@ import pl.touk.nussknacker.test.utils.scalas.DBIOActionValues
 import pl.touk.nussknacker.ui.api.DeploymentCommentSettings
 import pl.touk.nussknacker.ui.listener.ProcessChangeEvent.{OnActionExecutionFinished, OnActionSuccess}
 import pl.touk.nussknacker.ui.process.ScenarioQuery
+import pl.touk.nussknacker.ui.process.deployment.DeploymentService.ActiveScenariosLimitExceededError
 import pl.touk.nussknacker.ui.process.deployment.scenariostatus.FragmentStateException
 import pl.touk.nussknacker.ui.process.periodic.flink.FlinkClientStub
 import pl.touk.nussknacker.ui.process.repository.{CommentValidationError, DBIOActionRunner}
@@ -40,7 +42,7 @@ import java.util.UUID
 import scala.concurrent.duration._
 
 class DeploymentServiceSpec
-    extends AnyFunSuite
+    extends AnyWordSpec
     with Matchers
     with PatientScalaFutures
     with DBIOActionValues
@@ -69,7 +71,7 @@ class DeploymentServiceSpec
   private val deploymentManager: MockDeploymentManager = MockDeploymentManager.create()
 
   val TestDeploymentServiceServices(scenarioStatusProvider, actionService, deploymentService, reconciler) =
-    deploymentServiceFactory.create(deploymentManager)
+    deploymentServiceFactory.create(deploymentManager, activeScenariosLimit = Some(ActiveScenariosLimit(2)))
 
   // TODO: temporary step - we would like to extract the validation and the comment validation tests to external validators
   private def createDeploymentServiceWithCommentSettings = {
@@ -79,7 +81,7 @@ class DeploymentServiceSpec
       .deploymentService
   }
 
-  test("should return error when trying to deploy without comment when comment is required") {
+  "should return error when trying to deploy without comment when comment is required" in {
     val deploymentServiceWithCommentSettings = createDeploymentServiceWithCommentSettings
 
     val processName: ProcessName = generateProcessName
@@ -107,7 +109,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("should not deploy without comment when comment is required") {
+  "should not deploy without comment when comment is required" in {
     val deploymentServiceWithCommentSettings = createDeploymentServiceWithCommentSettings
 
     val processName: ProcessName = generateProcessName
@@ -137,7 +139,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("should pass when having an ok comment") {
+  "should pass when having an ok comment" in {
     val deploymentServiceWithCommentSettings = createDeploymentServiceWithCommentSettings
 
     val processName: ProcessName = generateProcessName
@@ -158,7 +160,7 @@ class DeploymentServiceSpec
       .futureValue
   }
 
-  test("should not cancel a deployed process without cancel comment when comment is required") {
+  "should not cancel a deployed process without cancel comment when comment is required" in {
     val deploymentServiceWithCommentSettings = createDeploymentServiceWithCommentSettings
 
     val processName: ProcessName = generateProcessName
@@ -185,7 +187,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("should return state correctly when state is deployed") {
+  "should return state correctly when state is deployed" in {
     val processName: ProcessName = generateProcessName
     val processIdWithName        = prepareProcess(processName).dbioActionValues
 
@@ -209,7 +211,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("should return state correctly when state is cancelled") {
+  "should return state correctly when state is cancelled" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
 
@@ -221,7 +223,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("should mark Action ExecutionFinished and publish an event as finished") {
+  "should mark Action ExecutionFinished and publish an event as finished" in {
     val processName: ProcessName = generateProcessName
     val (processId, actionId)    = prepareDeployedProcess(processName).dbioActionValues
 
@@ -235,7 +237,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should mark finished process as finished") {
+  "Should mark finished process as finished" in {
     val processName: ProcessName    = generateProcessName
     val (processId, deployActionId) = prepareDeployedProcess(processName).dbioActionValues
 
@@ -274,7 +276,7 @@ class DeploymentServiceSpec
     scenarioStatusProvider.getScenarioStatus(processId).futureValue shouldBe SimpleStateStatus.Finished
   }
 
-  test("Should finish deployment only after DeploymentManager finishes") {
+  "Should finish deployment only after DeploymentManager finishes" in {
     val processName: ProcessName = generateProcessName
     val processIdWithName        = prepareProcess(processName).dbioActionValues
 
@@ -324,7 +326,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should skip notifications and deployment on validation errors") {
+  "Should skip notifications and deployment on validation errors" in {
     val processName: ProcessName = generateProcessName
     val requestedParallelism     = FlinkClientStub.maxParallelism + 1
     val processIdWithName =
@@ -357,7 +359,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return properly state when state is canceled and process is canceled") {
+  "Should return properly state when state is canceled and process is canceled" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareCanceledProcess(processName).dbioActionValues
 
@@ -366,7 +368,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return canceled status for canceled process with empty state - cleaned state") {
+  "Should return canceled status for canceled process with empty state - cleaned state" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareCanceledProcess(processName).dbioActionValues
 
@@ -385,7 +387,7 @@ class DeploymentServiceSpec
     processDetails.lastStateAction.exists(_.actionName == ScenarioActionName.Cancel) shouldBe true
   }
 
-  test("Should return canceled status for canceled process with not founded state - cleaned state") {
+  "Should return canceled status for canceled process with not founded state - cleaned state" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareCanceledProcess(processName).dbioActionValues
 
@@ -404,7 +406,7 @@ class DeploymentServiceSpec
     processDetails.lastStateAction.exists(_.actionName == ScenarioActionName.Cancel) shouldBe true
   }
 
-  test("Should return state with warning when state is running and process is canceled") {
+  "Should return state with warning when state is running and process is canceled" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareCanceledProcess(processName).dbioActionValues
 
@@ -417,7 +419,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return not deployed when engine returns any state and process hasn't action") {
+  "Should return not deployed when engine returns any state and process hasn't action" in {
     val processName: ProcessName = generateProcessName
     val processId                = prepareProcess(processName).dbioActionValues
 
@@ -427,7 +429,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return DuringCancel state when is during canceled and process has CANCEL action") {
+  "Should return DuringCancel state when is during canceled and process has CANCEL action" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareCanceledProcess(processName).dbioActionValues
 
@@ -438,7 +440,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return state with status Restarting when process has been deployed and is restarting") {
+  "Should return state with status Restarting when process has been deployed and is restarting" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
 
@@ -457,7 +459,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return state with error when state is not running and process is deployed") {
+  "Should return state with error when state is not running and process is deployed" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
 
@@ -470,7 +472,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return state with error when state is null and process is deployed") {
+  "Should return state with error when state is null and process is deployed" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
 
@@ -483,7 +485,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return error state when state is running and process is deployed with mismatch versions") {
+  "Should return error state when state is running and process is deployed with mismatch versions" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
     val version                  = Some(VersionId(2))
@@ -497,7 +499,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should always return process manager failure, even if some other verifications return invalid") {
+  "Should always return process manager failure, even if some other verifications return invalid" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
     val version                  = Some(VersionId(2))
@@ -511,7 +513,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return warning state when state is running with empty version and process is deployed") {
+  "Should return warning state when state is running with empty version and process is deployed" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
 
@@ -524,7 +526,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return error state when failed to get state") {
+  "Should return error state when failed to get state" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
 
@@ -538,7 +540,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return not deployed status for process with empty state - not deployed state") {
+  "Should return not deployed status for process with empty state - not deployed state" in {
     val processName: ProcessName = generateProcessName
     val processId                = prepareProcess(processName).dbioActionValues
     fetchingScenarioDBIORepository
@@ -559,7 +561,7 @@ class DeploymentServiceSpec
     processDetails.lastAction shouldBe None
   }
 
-  test("Should return not deployed status for process with not found state - not deployed state") {
+  "Should return not deployed status for process with not found state - not deployed state" in {
     val processName: ProcessName = generateProcessName
     val processId                = prepareProcess(processName).dbioActionValues
     fetchingScenarioDBIORepository
@@ -578,7 +580,7 @@ class DeploymentServiceSpec
     processDetails.lastAction shouldBe None
   }
 
-  test("Should return not deployed status for process without actions and with state (it should never happen) ") {
+  "Should return not deployed status for process without actions and with state (it should never happen)" in {
     val processName: ProcessName = generateProcessName
     val processId                = prepareProcess(processName).dbioActionValues
     fetchingScenarioDBIORepository
@@ -599,7 +601,7 @@ class DeploymentServiceSpec
     processDetails.lastAction shouldBe None
   }
 
-  test("Should return not deployed state for archived never deployed process") {
+  "Should return not deployed state for archived never deployed process" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareArchivedProcess(processName, None).dbioActionValues
 
@@ -607,9 +609,7 @@ class DeploymentServiceSpec
     state shouldBe SimpleStateStatus.NotDeployed
   }
 
-  test(
-    "Should return not deployed state for archived never deployed process with running state (it should never happen)"
-  ) {
+  "Should return not deployed state for archived never deployed process with running state (it should never happen)" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareArchivedProcess(processName, None).dbioActionValues
 
@@ -619,7 +619,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return canceled status for archived canceled process") {
+  "Should return canceled status for archived canceled process" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareArchivedProcess(processName, Some(Cancel)).dbioActionValues
 
@@ -627,7 +627,7 @@ class DeploymentServiceSpec
     state shouldBe SimpleStateStatus.Canceled
   }
 
-  test("Should return canceled status for archived canceled process with running state (it should never happen)") {
+  "Should return canceled status for archived canceled process with running state (it should never happen)" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareArchivedProcess(processName, Some(Cancel)).dbioActionValues
 
@@ -637,7 +637,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return not deployed state for unarchived never deployed process") {
+  "Should return not deployed state for unarchived never deployed process" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = preparedUnArchivedProcess(processName, None).dbioActionValues
 
@@ -645,7 +645,7 @@ class DeploymentServiceSpec
     state shouldBe SimpleStateStatus.NotDeployed
   }
 
-  test("Should return during deploy for process in deploy in progress") {
+  "Should return during deploy for process in deploy in progress" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = preparedUnArchivedProcess(processName, None).dbioActionValues
     val _ = actionRepository
@@ -656,7 +656,7 @@ class DeploymentServiceSpec
     state shouldBe SimpleStateStatus.DuringDeploy
   }
 
-  test("Should getScenariosStatuses bulk with the same result as for single scenario") {
+  "Should getScenariosStatuses bulk with the same result as for single scenario" in {
     val (_, _, runningScenarioId) = prepareScenariosInVariousStates
 
     val processesDetails = fetchingScenarioDBIORepository
@@ -691,9 +691,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test(
-    "Should return not deployed status for archived never deployed process with running state (it should never happen)"
-  ) {
+  "Should return not deployed status for archived never deployed process with running state (it should never happen)" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareArchivedProcess(processName, None).dbioActionValues
 
@@ -703,7 +701,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return problem status for archived deployed process (last action deployed instead of cancel)") {
+  "Should return problem status for archived deployed process (last action deployed instead of cancel)" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareArchivedProcess(processName, Some(Deploy)).dbioActionValues
 
@@ -711,7 +709,7 @@ class DeploymentServiceSpec
     state shouldBe ProblemStateStatus.ArchivedShouldBeCanceled
   }
 
-  test("Should return canceled status for unarchived process") {
+  "Should return canceled status for unarchived process" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareArchivedProcess(processName, Some(Cancel)).dbioActionValues
 
@@ -721,7 +719,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("Should return problem status for unarchived process with running state (it should never happen)") {
+  "Should return problem status for unarchived process with running state (it should never happen)" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = preparedUnArchivedProcess(processName, Some(Cancel)).dbioActionValues
 
@@ -733,7 +731,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("should invalidate in progress processes") {
+  "should invalidate in progress processes" in {
     val processName: ProcessName = generateProcessName
     val processIdWithName        = prepareProcess(processName).dbioActionValues
 
@@ -758,7 +756,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("should return problem after occurring timeout during waiting on DM response") {
+  "should return problem after occurring timeout during waiting on DM response" in {
     val processName: ProcessName = generateProcessName
     val (processId, _)           = prepareDeployedProcess(processName).dbioActionValues
 
@@ -775,7 +773,7 @@ class DeploymentServiceSpec
     }
   }
 
-  test("should fail when trying to get state for fragment") {
+  "should fail when trying to get state for fragment" in {
     val processName: ProcessName = generateProcessName
     val id                       = prepareFragment(processName).dbioActionValues
 
@@ -785,7 +783,7 @@ class DeploymentServiceSpec
   }
 
   // TODO: add tests for more advanced things such as changes in model api
-  test("should recover jobs") {
+  "should recover jobs" in {
     val processName = generateProcessName
     prepareDeployedProcess(processName).dbioActionValues
 
@@ -798,6 +796,97 @@ class DeploymentServiceSpec
     }
   }
 
+  "should allow to deploy scenario when active scenarios count is less than the limit" when {
+    "1st scenario is running, and the 2nd scenario is not deployed" in {
+      deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+        deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.NotDeployed) {
+          deployExampleScenario()
+        }
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is cancelled" in {
+      deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+        deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Canceled) {
+          deployExampleScenario()
+        }
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is during cancel" in {
+      deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+        deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.DuringCancel) {
+          deployExampleScenario()
+        }
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is finished" in {
+      deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+        deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Finished) {
+          deployExampleScenario()
+        }
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is problem" in {
+      deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+        deploymentManager.withProcessStateStatus(generateProcessName, StateStatus("PROBLEM")) {
+          deployExampleScenario()
+        }
+      }
+    }
+  }
+
+  "should not allow more scenarios than active scenario limits to be used" when {
+    "1st scenario is running, and the 2nd scenario is running, and the 3rd scenario is not deployed" in {
+      deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+        deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+          deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.NotDeployed) {
+            assertThrowsWithParent[ActiveScenariosLimitExceededError] {
+              deployExampleScenario()
+            }
+          }
+        }
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is during deploy, and the 3rd scenario is not deployed" in {
+      deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+        deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.DuringDeploy) {
+          deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.NotDeployed) {
+            assertThrowsWithParent[ActiveScenariosLimitExceededError] {
+              deployExampleScenario()
+            }
+          }
+        }
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is restarting, and the 3rd scenario is not deployed" in {
+      deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Running) {
+        deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.Restarting) {
+          deploymentManager.withProcessStateStatus(generateProcessName, SimpleStateStatus.NotDeployed) {
+            assertThrowsWithParent[ActiveScenariosLimitExceededError] {
+              deployExampleScenario()
+            }
+          }
+        }
+      }
+    }
+
+  }
+
+  private def deployExampleScenario() = {
+    val processName       = generateProcessName
+    val processIdWithName = prepareProcess(processName).dbioActionValues
+    deploymentManager.withWaitForDeployFinish(processName) {
+      deploymentService
+        .processCommand(
+          RunDeploymentCommand(
+            CommonCommandData(processIdWithName, None, user),
+            StateRestoringStrategy.RestoreStateFromReplacedJobSavepoint,
+            NodesDeploymentData.empty
+          )
+        )
+        .futureValue
+    }
+  }
+
   override def beforeEach(): Unit = {
     super.beforeEach()
     listener.clear()
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/TestDeploymentServiceFactory.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/TestDeploymentServiceFactory.scala
index 2f7afe33b26..6d04765ab5f 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/TestDeploymentServiceFactory.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/deployment/TestDeploymentServiceFactory.scala
@@ -1,12 +1,13 @@
 package pl.touk.nussknacker.ui.process.deployment
 
-import cats.effect.unsafe.IORuntime
 import db.util.DBIOActionInstances.DB
 import org.apache.pekko.actor.ActorSystem
 import pl.touk.nussknacker.engine.{DeploymentManagerDependencies, JobsRecoverySettings, ModelData}
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.ActiveScenariosLimit
 import pl.touk.nussknacker.engine.api.deployment.DeploymentManager
 import pl.touk.nussknacker.engine.compile.ProcessValidator
 import pl.touk.nussknacker.engine.deployment.EngineSetupName
+import pl.touk.nussknacker.engine.util.ExecutionContextWithIORuntimeAdapter
 import pl.touk.nussknacker.test.config.WithSimplifiedDesignerConfig.TestProcessingType
 import pl.touk.nussknacker.test.config.WithSimplifiedDesignerConfig.TestProcessingType.Streaming
 import pl.touk.nussknacker.test.mock.{StubModelDataWithModelDefinition, TestProcessChangeListener}
@@ -15,7 +16,12 @@ import pl.touk.nussknacker.test.utils.domain.ProcessTestData.modelDefinition
 import pl.touk.nussknacker.test.utils.domain.TestFactory._
 import pl.touk.nussknacker.ui.api.DeploymentCommentSettings
 import pl.touk.nussknacker.ui.db.DbRef
-import pl.touk.nussknacker.ui.process.deployment.TestDeploymentServiceFactory.{actorSystem, clock, ec, processingType}
+import pl.touk.nussknacker.ui.process.deployment.TestDeploymentServiceFactory.{
+  actorSystem,
+  clock,
+  executionContextWithIORuntime,
+  processingType
+}
 import pl.touk.nussknacker.ui.process.deployment.deploymentstatus.EngineSideDeploymentStatusesProvider
 import pl.touk.nussknacker.ui.process.deployment.reconciliation.ScenarioDeploymentReconciler
 import pl.touk.nussknacker.ui.process.deployment.scenariostatus.ScenarioStatusProvider
@@ -25,7 +31,7 @@ import pl.touk.nussknacker.ui.process.repository.activities.ScenarioActivityRepo
 import sttp.client3.testing.SttpBackendStub
 
 import java.time.Clock
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.Future
 import scala.concurrent.duration.FiniteDuration
 
 class TestDeploymentServiceFactory(dbRef: DbRef) {
@@ -38,8 +44,8 @@ class TestDeploymentServiceFactory(dbRef: DbRef) {
   val listener                                       = new TestProcessChangeListener
 
   val deploymentManagerDependencies: DeploymentManagerDependencies = new DeploymentManagerDependencies(
-    ec,
-    IORuntime.global,
+    executionContextWithIORuntime,
+    executionContextWithIORuntime.ioRuntime,
     actorSystem,
     SttpBackendStub.asynchronousFuture
   )
@@ -48,7 +54,8 @@ class TestDeploymentServiceFactory(dbRef: DbRef) {
       deploymentManager: DeploymentManager,
       modelData: ModelData = new StubModelDataWithModelDefinition(modelDefinition()),
       scenarioStateTimeout: Option[FiniteDuration] = None,
-      deploymentCommentSettings: Option[DeploymentCommentSettings] = None
+      deploymentCommentSettings: Option[DeploymentCommentSettings] = None,
+      activeScenariosLimit: Option[ActiveScenariosLimit] = None
   ): TestDeploymentServiceServices = {
     val deploymentManagerProvider = TestProcessingTypeDataProviderFactory.createWithEmptyCombinedData(
       Map(processingType.stringify -> ValueWithRestriction.anyUser(deploymentManager))
@@ -112,6 +119,8 @@ class TestDeploymentServiceFactory(dbRef: DbRef) {
       TestFactory.scenarioResolverByProcessingType,
       actionService,
       additionalComponentConfigsByProcessingType,
+      scenarioStatusProvider,
+      TestFactory.mapProcessingTypeDataProvider(Streaming.stringify -> activeScenariosLimit)
     )
     TestDeploymentServiceServices(scenarioStatusProvider, actionService, deploymentService, deploymentsReconciler)
   }
@@ -128,8 +137,10 @@ case class TestDeploymentServiceServices(
 object TestDeploymentServiceFactory {
 
   implicit val actorSystem: ActorSystem = ActorSystem("TestDeploymentServiceFactory")
-  implicit val ec: ExecutionContext     = actorSystem.dispatcher
-  val clock: Clock                      = Clock.systemUTC()
+  implicit val executionContextWithIORuntime: ExecutionContextWithIORuntimeAdapter =
+    ExecutionContextWithIORuntimeAdapter.unsafeCreateFrom(actorSystem.dispatcher)
+
+  val clock: Clock = Clock.systemUTC()
 
   val processingType: TestProcessingType = TestProcessingType.Streaming
 
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/newdeployment/DeploymentServiceTest.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/newdeployment/DeploymentServiceTest.scala
index 0b0d2f3dd45..b08ed0d4bd4 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/newdeployment/DeploymentServiceTest.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/newdeployment/DeploymentServiceTest.scala
@@ -1,22 +1,36 @@
 package pl.touk.nussknacker.ui.process.newdeployment
 
-import cats.effect.unsafe.IORuntime
-import org.scalatest.BeforeAndAfterEach
-import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.{BeforeAndAfterEach, Inside}
 import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+import pl.touk.nussknacker.development.manager.BasicStatusDetails
 import pl.touk.nussknacker.development.manager.MockableDeploymentManagerProvider.MockableDeploymentManager
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.ActiveScenariosLimit
 import pl.touk.nussknacker.engine.api.component.NodesDeploymentData
-import pl.touk.nussknacker.engine.api.deployment.{DeploymentStatus, ProblemDeploymentStatus}
-import pl.touk.nussknacker.engine.api.process.ProcessName
+import pl.touk.nussknacker.engine.api.deployment.{DeploymentStatus, ProblemDeploymentStatus, StateStatus}
+import pl.touk.nussknacker.engine.api.deployment.simple.SimpleStateStatus
+import pl.touk.nussknacker.engine.api.process.{ProcessName, VersionId}
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.newdeployment.DeploymentId
+import pl.touk.nussknacker.engine.util.ExecutionContextWithIORuntimeAdapter
 import pl.touk.nussknacker.test.{EitherValuesDetailedMessage, PatientScalaFutures}
 import pl.touk.nussknacker.test.base.db.WithHsqlDbTesting
 import pl.touk.nussknacker.test.base.it.WithClock
 import pl.touk.nussknacker.test.config.WithSimplifiedDesignerConfig.TestProcessingType.Streaming
 import pl.touk.nussknacker.test.utils.domain.{ProcessTestData, TestFactory, TestProcessingTypeDataProviderFactory}
+import pl.touk.nussknacker.test.utils.domain.TestFactory.{
+  adminUser,
+  newActionProcessRepository,
+  newFetchingProcessRepository
+}
 import pl.touk.nussknacker.test.utils.scalas.DBIOActionValues
 import pl.touk.nussknacker.ui.process.deployment.DeploymentManagerDispatcher
+import pl.touk.nussknacker.ui.process.deployment.deploymentstatus.EngineSideDeploymentStatusesProvider
+import pl.touk.nussknacker.ui.process.deployment.scenariostatus.ScenarioStatusProvider
+import pl.touk.nussknacker.ui.process.newdeployment.DeploymentService.{
+  ActiveScenariosLimitExceededError,
+  DeploymentForeignKeys
+}
 import pl.touk.nussknacker.ui.process.processingtype.ValueWithRestriction
 import pl.touk.nussknacker.ui.process.repository.DBIOActionRunner
 import pl.touk.nussknacker.ui.process.repository.ProcessRepository.CreateProcessAction
@@ -26,8 +40,9 @@ import scala.concurrent.ExecutionContext
 import scala.util.Failure
 
 class DeploymentServiceTest
-    extends AnyFunSuite
+    extends AnyWordSpec
     with Matchers
+    with Inside
     with PatientScalaFutures
     with WithHsqlDbTesting
     with WithClock
@@ -35,8 +50,9 @@ class DeploymentServiceTest
     with EitherValuesDetailedMessage
     with BeforeAndAfterEach {
 
-  private implicit val ec: ExecutionContext = ExecutionContext.global
-  private implicit val ioRuntime: IORuntime = IORuntime.global
+  private implicit val executionContextWithIORuntime: ExecutionContextWithIORuntimeAdapter =
+    ExecutionContextWithIORuntimeAdapter.unsafeCreateFrom(ExecutionContext.global)
+  import executionContextWithIORuntime.ioRuntime
 
   override protected val dbioRunner: DBIOActionRunner = DBIOActionRunner(testDbRef)
 
@@ -45,33 +61,44 @@ class DeploymentServiceTest
   private val service = {
     val clock                      = Clock.fixed(Instant.ofEpochMilli(0), ZoneOffset.UTC)
     val scenarioMetadataRepository = TestFactory.newScenarioMetadataRepository(testDbRef)
+    val deploymentManager          = new MockableDeploymentManager(modelDataProviderOpt = None)
+    val deploymentManagerDispatcher = new DeploymentManagerDispatcher(
+      TestProcessingTypeDataProviderFactory.createWithEmptyCombinedData(
+        Map(Streaming.stringify -> ValueWithRestriction.anyUser(deploymentManager))
+      ),
+      TestFactory.newFutureFetchingScenarioRepository(testDbRef)
+    )
+
+    val scenarioStatusProvider = {
+      new ScenarioStatusProvider(
+        new EngineSideDeploymentStatusesProvider(deploymentManagerDispatcher, scenarioStateTimeout = None),
+        deploymentManagerDispatcher,
+        newFetchingProcessRepository(testDbRef),
+        newActionProcessRepository(testDbRef),
+        dbioRunner
+      )
+    }
+
     new DeploymentService(
       scenarioMetadataRepository,
       TestFactory.newScenarioGraphVersionService(testDbRef),
       TestFactory.newDeploymentRepository(testDbRef, clock),
-      new DeploymentManagerDispatcher(
-        TestProcessingTypeDataProviderFactory.createWithEmptyCombinedData(
-          Map(
-            Streaming.stringify -> ValueWithRestriction.anyUser(
-              new MockableDeploymentManager(modelDataProviderOpt = None)
-            )
-          )
-        ),
-        TestFactory.newFutureFetchingScenarioRepository(testDbRef)
-      ),
+      deploymentManagerDispatcher,
       dbioRunner,
       clock,
-      TestFactory.additionalComponentConfigsByProcessingType
+      TestFactory.additionalComponentConfigsByProcessingType,
+      scenarioStatusProvider,
+      TestFactory.mapProcessingTypeDataProvider(Streaming.stringify -> Some(ActiveScenariosLimit(2)))
     )
   }
 
-  test("request deployment and provide status for it") {
+  "request deployment and provide status for it" in {
     val scenarioName = ProcessName("validScenario")
     val scenario     = ProcessTestData.validProcessWithName(scenarioName)
     saveSampleScenario(scenario)
 
     val deploymentId = DeploymentId.generate
-    val user         = TestFactory.adminUser()
+    val user         = adminUser()
     service
       .runDeployment(
         RunDeploymentCommand(deploymentId, scenarioName, NodesDeploymentData.empty, user)
@@ -83,7 +110,7 @@ class DeploymentServiceTest
     status.value shouldEqual DeploymentStatus.DuringDeploy
   }
 
-  test("deployment which ended up with failure during request should has problem status") {
+  "deployment which ended up with failure during request should has problem status" in {
     val scenarioName = ProcessName("scenarioCausingFailure")
     val scenario     = ProcessTestData.validProcessWithName(scenarioName)
     saveSampleScenario(scenario)
@@ -92,7 +119,7 @@ class DeploymentServiceTest
       Map(deploymentId -> Failure(new Exception("Some failure during deployment")))
     )
 
-    val user = TestFactory.adminUser()
+    val user = adminUser()
     service
       .runDeployment(
         RunDeploymentCommand(deploymentId, scenarioName, NodesDeploymentData.empty, user)
@@ -106,6 +133,129 @@ class DeploymentServiceTest
     }
   }
 
+  "should allow to deploy scenario when active scenarios count is less than the limit" when {
+    "1st scenario is running, and the 2nd scenario is not deployed" in {
+      MockableDeploymentManager.configureScenarioStatuses(
+        Map(
+          "scenario1" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario2" -> BasicStatusDetails(SimpleStateStatus.NotDeployed, version = Some(VersionId(1))),
+        )
+      )
+
+      val result = deployExampleScenario(ProcessName("scenario3"))
+
+      inside(result) { case Right(DeploymentForeignKeys(_, _)) =>
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is cancelled" in {
+      MockableDeploymentManager.configureScenarioStatuses(
+        Map(
+          "scenario1" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario2" -> BasicStatusDetails(SimpleStateStatus.Canceled, version = Some(VersionId(1))),
+        )
+      )
+
+      val result = deployExampleScenario(ProcessName("scenario3"))
+
+      inside(result) { case Right(DeploymentForeignKeys(_, _)) =>
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is during cancel" in {
+      MockableDeploymentManager.configureScenarioStatuses(
+        Map(
+          "scenario1" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario2" -> BasicStatusDetails(SimpleStateStatus.DuringCancel, version = Some(VersionId(1))),
+        )
+      )
+
+      val result = deployExampleScenario(ProcessName("scenario3"))
+
+      inside(result) { case Right(DeploymentForeignKeys(_, _)) =>
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is finished" in {
+      MockableDeploymentManager.configureScenarioStatuses(
+        Map(
+          "scenario1" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario2" -> BasicStatusDetails(SimpleStateStatus.Finished, version = Some(VersionId(1))),
+        )
+      )
+
+      val result = deployExampleScenario(ProcessName("scenario3"))
+
+      inside(result) { case Right(DeploymentForeignKeys(_, _)) =>
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is problem" in {
+      MockableDeploymentManager.configureScenarioStatuses(
+        Map(
+          "scenario1" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario2" -> BasicStatusDetails(StateStatus("PROBLEM"), version = Some(VersionId(1))),
+        )
+      )
+
+      val result = deployExampleScenario(ProcessName("scenario3"))
+
+      inside(result) { case Right(DeploymentForeignKeys(_, _)) =>
+      }
+    }
+  }
+
+  "should not allow more scenarios than active scenario limits to be used" when {
+    "1st scenario is running, and the 2nd scenario is running, and the 3rd scenario is not deployed" in {
+      MockableDeploymentManager.configureScenarioStatuses(
+        Map(
+          "scenario1" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario2" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario3" -> BasicStatusDetails(SimpleStateStatus.NotDeployed, version = Some(VersionId(1))),
+        )
+      )
+
+      val result = deployExampleScenario(ProcessName("scenario4"))
+
+      inside(result) { case Left(ActiveScenariosLimitExceededError(2)) =>
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is during deploy, and the 3rd scenario is not deployed" in {
+      MockableDeploymentManager.configureScenarioStatuses(
+        Map(
+          "scenario1" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario2" -> BasicStatusDetails(SimpleStateStatus.DuringDeploy, version = Some(VersionId(1))),
+          "scenario3" -> BasicStatusDetails(SimpleStateStatus.NotDeployed, version = Some(VersionId(1))),
+        )
+      )
+
+      val result = deployExampleScenario(ProcessName("scenario4"))
+
+      inside(result) { case Left(ActiveScenariosLimitExceededError(2)) =>
+      }
+    }
+    "1st scenario is running, and the 2nd scenario is restarting, and the 3rd scenario is not deployed" in {
+      MockableDeploymentManager.configureScenarioStatuses(
+        Map(
+          "scenario1" -> BasicStatusDetails(SimpleStateStatus.Running, version = Some(VersionId(1))),
+          "scenario2" -> BasicStatusDetails(SimpleStateStatus.Restarting, version = Some(VersionId(1))),
+          "scenario3" -> BasicStatusDetails(SimpleStateStatus.NotDeployed, version = Some(VersionId(1))),
+        )
+      )
+
+      val result = deployExampleScenario(ProcessName("scenario4"))
+
+      inside(result) { case Left(ActiveScenariosLimitExceededError(2)) =>
+      }
+    }
+  }
+
+  override def beforeEach(): Unit = {
+    MockableDeploymentManager.clean()
+    super.beforeEach()
+  }
+
+  override def afterAll(): Unit = {
+    executionContextWithIORuntime.close()
+    super.afterAll()
+  }
+
   private def saveSampleScenario(scenario: CanonicalProcess) = {
     writeScenarioRepository
       .saveNewProcess(
@@ -116,13 +266,19 @@ class DeploymentServiceTest
           processingType = Streaming.stringify,
           isFragment = false,
         )
-      )(TestFactory.adminUser())
+      )(adminUser())
       .dbioActionValues
   }
 
-  override def beforeEach(): Unit = {
-    MockableDeploymentManager.clean()
-    super.beforeEach()
+  private def deployExampleScenario(scenarioName: ProcessName) = {
+    val scenario = ProcessTestData.validProcessWithName(scenarioName)
+    saveSampleScenario(scenario)
+
+    service
+      .runDeployment(
+        RunDeploymentCommand(DeploymentId.generate, scenarioName, NodesDeploymentData.empty, adminUser())
+      )
+      .futureValue
   }
 
 }
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/periodic/flink/PeriodicDeploymentManagerTest.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/periodic/flink/PeriodicDeploymentManagerTest.scala
index dc162a08b50..cc85b54dd06 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/periodic/flink/PeriodicDeploymentManagerTest.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/periodic/flink/PeriodicDeploymentManagerTest.scala
@@ -2,7 +2,6 @@ package pl.touk.nussknacker.ui.process.periodic.flink
 
 import org.scalatest.{Inside, OptionValues}
 import org.scalatest.concurrent.ScalaFutures
-import org.scalatest.exceptions.TestFailedException
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.prop.TableDrivenPropertyChecks
@@ -195,7 +194,7 @@ class PeriodicDeploymentManagerTest
     state shouldEqual SimpleStateStatus.NotDeployed
   }
 
-  test("getScenarioStatuss - should be scheduled when scenario scheduled and no job on Flink") {
+  test("getScenarioStatus - should be scheduled when scenario scheduled and no job on Flink") {
     val f = new Fixture
 
     val version = f.saveScenario()
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeDataSpec.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeDataSpec.scala
index 9434ace67b1..c36ce60edb4 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeDataSpec.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/processingtype/ProcessingTypeDataSpec.scala
@@ -86,7 +86,8 @@ class ProcessingTypeDataSpec extends AnyFunSuite with Matchers {
       ),
       deploymentData,
       category = "dummy category",
-      componentDefinitionExtractionMode = modelDependencies.componentDefinitionExtractionMode
+      componentDefinitionExtractionMode = modelDependencies.componentDefinitionExtractionMode,
+      activeScenariosLimit = None
     )
   }
 
diff --git a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/processingtype/provider/ProcessingTypeDataProviderAccessRestrictionTest.scala b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/processingtype/provider/ProcessingTypeDataProviderAccessRestrictionTest.scala
index 335fb7d9d5d..ffbde27cbdc 100644
--- a/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/processingtype/provider/ProcessingTypeDataProviderAccessRestrictionTest.scala
+++ b/designer/server/src/test/scala/pl/touk/nussknacker/ui/process/processingtype/provider/ProcessingTypeDataProviderAccessRestrictionTest.scala
@@ -63,7 +63,8 @@ class ProcessingTypeDataProviderAccessRestrictionTest
         classPath = modelClasspath,
         deploymentConfig = ConfigFactory.empty(),
         modelConfig = ConfigWithUnresolvedVersion(ConfigFactory.empty()),
-        category = s"${processingType}Category"
+        category = s"${processingType}Category",
+        activeScenariosLimit = None
       )
     }.toMap)
     TestProcessingTypeDataProviderFactory.create(
diff --git a/docs/Changelog.md b/docs/Changelog.md
index b6ce7d88dac..7f56fbf143c 100644
--- a/docs/Changelog.md
+++ b/docs/Changelog.md
@@ -155,6 +155,7 @@
 * [#7553](https://github.com/TouK/nussknacker/pull/7553) Key variable created in window components is not transformed into a string.
   Key will now have type of value given in `groupBy` field. Previously we wrapped `groupBy` value as a list in String, e.g. for `id = 1` before: `groupBy: id -> key: "[1]"`, after: `groupBy: id -> key: 1`
 * [#7824](https://github.com/TouK/nussknacker/pull/7824) Add Json type support for schemaless topic data with Json content type.
+* [#7970](https://github.com/TouK/nussknacker/pull/7970) Added "activeScenariosLimit" setting defined per processing type. 
 
 ## 1.18
 
diff --git a/engine/development/deploymentManager/src/main/scala/pl/touk/nussknacker/development/manager/MockableDeploymentManagerProvider.scala b/engine/development/deploymentManager/src/main/scala/pl/touk/nussknacker/development/manager/MockableDeploymentManagerProvider.scala
index 9011b216035..9020cfa4e75 100644
--- a/engine/development/deploymentManager/src/main/scala/pl/touk/nussknacker/development/manager/MockableDeploymentManagerProvider.scala
+++ b/engine/development/deploymentManager/src/main/scala/pl/touk/nussknacker/development/manager/MockableDeploymentManagerProvider.scala
@@ -56,7 +56,7 @@ class MockableDeploymentManagerProvider extends DeploymentManagerProvider {
 
 object MockableDeploymentManagerProvider {
 
-  type ScenarioName = String
+  private type ScenarioName = String
 
   class MockableDeploymentManager(modelDataProviderOpt: Option[BaseModelDataProvider])(
       implicit executionContext: ExecutionContext,
@@ -129,10 +129,22 @@ object MockableDeploymentManagerProvider {
       }
     }
 
-    override def deploymentSynchronisationSupport: DeploymentSynchronisationSupport = NoDeploymentSynchronisationSupport
+    override val deploymentSynchronisationSupport: DeploymentSynchronisationSupport = NoDeploymentSynchronisationSupport
 
-    override def deploymentsStatusesQueryForAllScenariosSupport: DeploymentsStatusesQueryForAllScenariosSupport =
-      NoDeploymentsStatusesQueryForAllScenariosSupport
+    override val deploymentsStatusesQueryForAllScenariosSupport: DeploymentsStatusesQueryForAllScenariosSupport =
+      new DeploymentsStatusesQueryForAllScenariosSupported {
+        override def getAllScenariosDeploymentsStatuses()(
+            implicit freshnessPolicy: DataFreshnessPolicy
+        ): Future[WithDataFreshnessStatus[Map[ProcessName, List[DeploymentStatusDetails]]]] = {
+          Future {
+            WithDataFreshnessStatus.fresh(
+              MockableDeploymentManager.scenarioStatuses.get().map { case (k, v) =>
+                (ProcessName(k), DeploymentStatusDetails(v.status, None, v.version) :: Nil)
+              }
+            )
+          }
+        }
+      }
 
     override def schedulingSupport: SchedulingSupport = NoSchedulingSupport
 
@@ -187,4 +199,4 @@ object MockableDeploymentManagerProvider {
 
 }
 
-case class BasicStatusDetails(status: StateStatus, version: Option[VersionId])
+final case class BasicStatusDetails(status: StateStatus, version: Option[VersionId])
diff --git a/engine/flink/management/src/main/scala/pl/touk/nussknacker/engine/management/FlinkConfig.scala b/engine/flink/management/src/main/scala/pl/touk/nussknacker/engine/management/FlinkConfig.scala
index dc107403212..af8429036e7 100644
--- a/engine/flink/management/src/main/scala/pl/touk/nussknacker/engine/management/FlinkConfig.scala
+++ b/engine/flink/management/src/main/scala/pl/touk/nussknacker/engine/management/FlinkConfig.scala
@@ -34,7 +34,8 @@ final case class FlinkConfig(
     jobConfigsCacheSize: Int = 1000,
     miniCluster: FlinkMiniClusterConfig = FlinkMiniClusterConfig(),
     scenarioTesting: ScenarioTestingConfig = ScenarioTestingConfig(),
-    scenarioStateVerification: ScenarioStateVerificationConfig = ScenarioStateVerificationConfig()
+    scenarioStateVerification: ScenarioStateVerificationConfig = ScenarioStateVerificationConfig(),
+    limitOfRunningScenarios: Option[Int] = None
 ) extends LazyLogging {
 
   def parseHttpClientConfig(
diff --git a/extensions-api/src/main/scala/pl/touk/nussknacker/engine/ProcessingTypeConfig.scala b/extensions-api/src/main/scala/pl/touk/nussknacker/engine/ProcessingTypeConfig.scala
index fe347fce8aa..ae527d8f270 100644
--- a/extensions-api/src/main/scala/pl/touk/nussknacker/engine/ProcessingTypeConfig.scala
+++ b/extensions-api/src/main/scala/pl/touk/nussknacker/engine/ProcessingTypeConfig.scala
@@ -1,7 +1,7 @@
 package pl.touk.nussknacker.engine
 
 import com.typesafe.config.Config
-import pl.touk.nussknacker.engine.ProcessingTypeConfig.DeploymentManagerType
+import pl.touk.nussknacker.engine.ProcessingTypeConfig.{ActiveScenariosLimit, DeploymentManagerType}
 import pl.touk.nussknacker.engine.deployment.EngineSetupName
 
 case class ProcessingTypeConfig(
@@ -10,7 +10,8 @@ case class ProcessingTypeConfig(
     classPath: List[String],
     deploymentConfig: Config,
     modelConfig: ConfigWithUnresolvedVersion,
-    category: String
+    category: String,
+    activeScenariosLimit: Option[ActiveScenariosLimit]
 )
 
 object ProcessingTypeConfig {
@@ -24,10 +25,12 @@ object ProcessingTypeConfig {
       config.resolved.as[List[String]]("modelConfig.classPath"),
       config.resolved.getConfig("deploymentConfig"),
       config.getConfig("modelConfig"),
-      config.resolved.as[String]("category")
+      config.resolved.as[String]("category"),
+      config.resolved.getAs[Int]("activeScenariosLimit").map(ActiveScenariosLimit.apply),
     )
   }
 
-  case class DeploymentManagerType(value: String)
+  final case class DeploymentManagerType(value: String) extends AnyVal
+  final case class ActiveScenariosLimit(value: Int)     extends AnyVal
 
 }