Further optimizations to test resources (#8936)

TestRuntime should be deprecated as it creates a number of threads and doesn't allow to easily modify ZIO's runtime.
But the biggest drop stems from fixing leaking `FileSystemService` that weren't being closed for every `TextOperationsTest` test.
The change is a follow up on #8892 but this time focused on ZIO usage.

Hopefully fixes #8806 for good.

# Important Notes
Running `language-server/test`.
Before:
![Screenshot from 2024-02-02 09-48-32](https://github.com/enso-org/enso/assets/292128/fb414c74-7d7a-4e7b-8b0c-d25dc3721bbf)

After:
![Screenshot from 2024-02-02 09-46-02](https://github.com/enso-org/enso/assets/292128/db9429df-d861-4f48-818f-888d5bbbb089)

engine/language-server/src/main/scala/org/enso/languageserver/filemanager/ReceivesTreeUpdatesHandler.scala

@@ -1,6 +1,6 @@
 package org.enso.languageserver.filemanager
 
-import akka.actor.{Actor, ActorRef, Props, Stash, Terminated}
+import akka.actor.{Actor, ActorRef, PoisonPill, Props, Stash, Terminated}
 import com.typesafe.scalalogging.LazyLogging
 import org.enso.filewatcher.WatcherFactory
 import org.enso.languageserver.capability.CapabilityProtocol.{
@@ -166,6 +166,11 @@ final class ReceivesTreeUpdatesHandler(
 
     case Terminated(watcher) =>
       context.become(withStore(store.removeWatcher(watcher)))
+
+    case Stop =>
+      store.watchers.foreach { case (_, watcher) =>
+        watcher ! PoisonPill
+      }
   }
 }
 
@@ -211,6 +216,9 @@ object ReceivesTreeUpdatesHandler {
       new Store(Map())
   }
 
+  // A message sent to the handler to stop all watchers and the handler itself
+  case object Stop
+
   /** Creates a configuration object used to create a
     * [[ReceivesTreeUpdatesHandler]].
     *

engine/language-server/src/main/scala/org/enso/languageserver/protocol/json/JsonConnectionControllerFactory.scala

@@ -75,4 +75,7 @@ class JsonConnectionControllerFactory(
       ),
       s"json-connection-controller-$clientId"
     )
+
+  def shutdown(): Unit = {}
+
 }

engine/language-server/src/test/scala/org/enso/languageserver/websocket/binary/BaseBinaryServerTest.scala

@@ -15,7 +15,7 @@ import org.enso.languageserver.data.{
   ProjectDirectoriesConfig,
   VcsManagerConfig
 }
-import org.enso.languageserver.effect.{TestRuntime, ZioExec}
+import org.enso.languageserver.effect.{ExecutionContextRuntime, ZioExec}
 import org.enso.languageserver.filemanager.{
   ContentRoot,
   ContentRootManager,
@@ -33,6 +33,8 @@ import org.enso.languageserver.websocket.binary.factory.{
   SessionInitFactory
 }
 
+import java.util.concurrent.{ExecutorService, Executors}
+import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
 abstract class BaseBinaryServerTest extends BinaryServerTestKit {
@@ -54,15 +56,30 @@ abstract class BaseBinaryServerTest extends BinaryServerTestKit {
     None
   )
 
+  var threadPool: ExecutorService = _
+
   sys.addShutdownHook(FileUtils.deleteQuietly(testContentRoot.file))
 
   @volatile
   protected var lastConnectionController: ActorRef = _
 
+  override def beforeEach(): Unit = {
+    threadPool = Executors.newWorkStealingPool(4)
+    super.beforeEach()
+  }
+
+  override def afterEach(): Unit = {
+    threadPool.close()
+    super.afterEach()
+  }
+
   override def connectionControllerFactory: ConnectionControllerFactory = {
     (clientIp: RemoteAddress.IP) =>
       {
-        val zioExec = ZioExec(new TestRuntime)
+        val testExecutor = ExecutionContext.fromExecutor(threadPool)
+        val zioRuntime   = new ExecutionContextRuntime(testExecutor)
+        zioRuntime.init()
+        val zioExec = ZioExec(zioRuntime)
 
         val contentRootManagerActor =
           system.actorOf(ContentRootManagerActor.props(config))

engine/language-server/src/test/scala/org/enso/languageserver/websocket/json/BaseServerTest.scala

@@ -1,5 +1,6 @@
 package org.enso.languageserver.websocket.json
 
+import akka.actor.{ActorRef, ActorSystem}
 import akka.testkit.TestProbe
 import io.circe.literal._
 import io.circe.parser.parse
@@ -20,13 +21,14 @@ import org.enso.languageserver.boot.{
 }
 import org.enso.languageserver.boot.resource.{
   DirectoriesInitialization,
+  InitializationComponent,
   RepoInitialization,
   SequentialResourcesInitialization,
   ZioRuntimeInitialization
 }
 import org.enso.languageserver.capability.CapabilityRouter
 import org.enso.languageserver.data._
-import org.enso.languageserver.effect.{TestRuntime, ZioExec}
+import org.enso.languageserver.effect.{ExecutionContextRuntime, ZioExec}
 import org.enso.languageserver.event.InitializedEvent
 import org.enso.languageserver.filemanager._
 import org.enso.languageserver.io._
@@ -67,6 +69,9 @@ import java.io.File
 import java.net.URISyntaxException
 import java.nio.file.{Files, Path}
 import java.util.UUID
+import java.util.concurrent.{Executors, ThreadFactory}
+import java.util.concurrent.atomic.AtomicInteger
+import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
 abstract class BaseServerTest
@@ -141,22 +146,36 @@ abstract class BaseServerTest
       InputRedirectionController.props(stdIn, stdInSink, sessionRouter)
     )
 
-  val zioRuntime      = new TestRuntime
+  class TestThreadFactory(name: String) extends ThreadFactory {
+    private val counter = new AtomicInteger(0)
+    override def newThread(r: Runnable): Thread = {
+      val t = new Thread();
+      t.setName(name + "-" + counter.getAndIncrement())
+      t
+    }
+  }
+
+  val initThreadPool = Executors.newWorkStealingPool(4)
+  val threadPool =
+    Executors.newWorkStealingPool(10)
+  val testExecutor = ExecutionContext.fromExecutor(threadPool)
+  val zioRuntime   = new ExecutionContextRuntime(testExecutor)
+
   val zioExec         = ZioExec(zioRuntime)
   val sqlDatabase     = SqlDatabase(config.directories.suggestionsDatabaseFile)
   val suggestionsRepo = new SqlSuggestionsRepo(sqlDatabase)(system.dispatcher)
 
   private def initializationComponent =
     new SequentialResourcesInitialization(
-      system.dispatcher,
-      new DirectoriesInitialization(system.dispatcher, config.directories),
+      initThreadPool,
+      new DirectoriesInitialization(initThreadPool, config.directories),
       new ZioRuntimeInitialization(
-        system.dispatcher,
+        initThreadPool,
         zioRuntime,
         system.eventStream
       ),
       new RepoInitialization(
-        system.dispatcher,
+        initThreadPool,
         config.directories,
         system.eventStream,
         sqlDatabase,
@@ -179,7 +198,9 @@ abstract class BaseServerTest
   }
 
   override def afterAll(): Unit = {
-    sqlDatabase.close()
+    suggestionsRepo.close()
+    threadPool.shutdown()
+    initThreadPool.shutdown()
     super.afterAll()
   }
 
@@ -208,7 +229,7 @@ abstract class BaseServerTest
     rootDir
   }
 
-  override def clientControllerFactory: ClientControllerFactory = {
+  override def clientControllerFactory(): ClientControllerFactory = {
     val contentRootManagerWrapper: ContentRootManager =
       new ContentRootManagerWrapper(config, contentRootManagerActor)
 
@@ -379,10 +400,11 @@ abstract class BaseServerTest
       )
     )
 
-    new JsonConnectionControllerFactory(
+    new TestJsonConnectionControllerFactory(
       mainComponent          = initializationComponent,
       bufferRegistry         = bufferRegistry,
       capabilityRouter       = capabilityRouter,
+      fileEventRegistry      = fileEventRegistry,
       fileManager            = fileManager,
       vcsManager             = vcsManager,
       contentRootManager     = contentRootManagerActor,
@@ -502,4 +524,49 @@ abstract class BaseServerTest
         fail("expected OpenFile notification got " + msg)
     }
   }
+
+  class TestJsonConnectionControllerFactory(
+    mainComponent: InitializationComponent,
+    bufferRegistry: ActorRef,
+    capabilityRouter: ActorRef,
+    fileEventRegistry: ActorRef,
+    fileManager: ActorRef,
+    vcsManager: ActorRef,
+    contentRootManager: ActorRef,
+    contextRegistry: ActorRef,
+    suggestionsHandler: ActorRef,
+    stdOutController: ActorRef,
+    stdErrController: ActorRef,
+    stdInController: ActorRef,
+    runtimeConnector: ActorRef,
+    idlenessMonitor: ActorRef,
+    projectSettingsManager: ActorRef,
+    profilingManager: ActorRef,
+    libraryConfig: LibraryConfig,
+    config: Config
+  )(implicit system: ActorSystem)
+      extends JsonConnectionControllerFactory(
+        mainComponent,
+        bufferRegistry,
+        capabilityRouter,
+        fileManager,
+        vcsManager,
+        contentRootManager,
+        contextRegistry,
+        suggestionsHandler,
+        stdOutController,
+        stdErrController,
+        stdInController,
+        runtimeConnector,
+        idlenessMonitor,
+        projectSettingsManager,
+        profilingManager,
+        libraryConfig,
+        config
+      )(system) {
+    override def shutdown(): Unit = {
+      fileEventRegistry ! ReceivesTreeUpdatesHandler.Stop
+      super.shutdown()
+    }
+  }
 }