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,
+      initThreadPool,
-      new DirectoriesInitialization(system.dispatcher, config.directories),
+      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()
+    }
+  }
 }

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

@@ -33,6 +33,7 @@ class LibrariesTest
     with ReportLogsOnFailure
     with FlakySpec {
   private val libraryRepositoryPort: Int = 47308
+  private val defaultTimeout             = 30.seconds
 
   private val exampleRepo = new ExampleRepository(
     locateRootDirectory().toPath
@@ -94,7 +95,7 @@ class LibrariesTest
         } yield libraryNames
 
       // The resolver may find the current project and other test projects on the path.
-      val msg1 = client.expectSomeJson()
+      val msg1 = client.expectSomeJson(timeout = defaultTimeout)
       inside(findLibraryNamesInResponse(msg1)) { case Some(libs) =>
         // Ensure that before running this test, the library did not exist.
         libs should not contain testLibraryName
@@ -125,12 +126,15 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 1,
             "result": null
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
 
       client.send(json"""
           { "jsonrpc": "2.0",
@@ -138,7 +142,7 @@ class LibrariesTest
             "id": 2
           }
           """)
-      val msg2 = client.expectSomeJson()
+      val msg2 = client.expectSomeJson(timeout = defaultTimeout)
       inside(findLibraryNamesInResponse(msg2)) { case Some(libs) =>
         libs should contain(testLibraryName)
       }
@@ -168,12 +172,15 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 0,
             "result": null
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
 
       client.send(json"""
           { "jsonrpc": "2.0",
@@ -188,7 +195,8 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 1,
             "result": {
@@ -196,7 +204,9 @@ class LibrariesTest
               "tagLine": null
             }
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
 
       client.send(json"""
           { "jsonrpc": "2.0",
@@ -210,12 +220,15 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 2,
             "result": null
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
 
       client.send(json"""
           { "jsonrpc": "2.0",
@@ -230,7 +243,8 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 3,
             "result": {
@@ -238,7 +252,9 @@ class LibrariesTest
               "tagLine": "tag-line"
             }
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
     }
 
     "return LibraryNotFound error when getting the metadata of unknown library" in {
@@ -256,7 +272,8 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 0,
             "error": {
@@ -264,7 +281,9 @@ class LibrariesTest
               "message": "Local library [user.Get_Package_Unknown] has not been found."
             }
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
     }
 
     "get the package config" in {
@@ -283,12 +302,15 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 0,
             "result": null
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
 
       val libraryRoot = getTestDirectory
         .resolve("test_home")
@@ -318,7 +340,7 @@ class LibrariesTest
             }
           }
           """)
-      val response = client.expectSomeJson()
+      val response = client.expectSomeJson(timeout = defaultTimeout)
 
       response.hcursor
         .downField("result")
@@ -351,7 +373,8 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 0,
             "error": {
@@ -359,7 +382,9 @@ class LibrariesTest
               "message": "Local library [user.Get_Package_Unknown] has not been found."
             }
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
     }
 
     "create, publish a library and fetch its manifest from the server" in {
@@ -389,12 +414,15 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 0,
             "result": null
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
 
       // Update Main.enso
       val libraryRoot = getTestDirectory
@@ -422,12 +450,15 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 1,
             "result": null
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
 
       val repoRoot        = getTestDirectory.resolve("libraries_repo_root")
       val rootDir         = locateRootDirectory()
@@ -516,7 +547,8 @@ class LibrariesTest
             }
           }
           """)
-        client.expectJson(json"""
+        client.expectJson(
+          json"""
           { "jsonrpc": "2.0",
             "id": 3,
             "result": {
@@ -524,7 +556,9 @@ class LibrariesTest
               "tagLine": "published-lib"
             }
           }
-          """)
+          """,
+          timeout = defaultTimeout
+        )
       }
 
       // Once the server is down, the metadata request should fail, especially as the published version is not cached.
@@ -543,7 +577,8 @@ class LibrariesTest
             }
           }
           """)
-      val errorMsg = client.expectSomeJson().asObject.value
+      val errorMsg =
+        client.expectSomeJson(timeout = defaultTimeout).asObject.value
       errorMsg("id").value.asNumber.value.toInt.value shouldEqual 4
       errorMsg("error").value.asObject shouldBe defined
     }
@@ -574,7 +609,8 @@ class LibrariesTest
         var waitingForResult = true
 
         while (waitingForTask || waitingForResult) {
-          val msg = client.expectSomeJson(10.seconds.dilated).asObject.value
+          val msg =
+            client.expectSomeJson(timeout = defaultTimeout).asObject.value
 
           msg("id") match {
             case Some(json) =>
@@ -653,7 +689,8 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 0,
             "result": {
@@ -662,7 +699,9 @@ class LibrariesTest
               ]
             }
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
     }
 
     "update the list of editions if requested" ignore {
@@ -681,7 +720,8 @@ class LibrariesTest
             }
           }
           """)
-        client.expectJson(json"""
+        client.expectJson(
+          json"""
           { "jsonrpc": "2.0",
             "id": 0,
             "result": {
@@ -691,7 +731,9 @@ class LibrariesTest
               ]
             }
           }
-          """)
+          """,
+          timeout = defaultTimeout
+        )
       }
     }
   }
@@ -733,7 +775,9 @@ class LibrariesTest
             }
           }
           """)
-      val published = extractPublishedLibraries(client.expectSomeJson())
+      val published = extractPublishedLibraries(
+        client.expectSomeJson(timeout = defaultTimeout)
+      )
 
       published should contain(
         PublishedLibrary("Standard", "Base", isCached = true)
@@ -755,7 +799,9 @@ class LibrariesTest
             }
           }
           """)
-      extractPublishedLibraries(client.expectSomeJson()) should contain(
+      extractPublishedLibraries(
+        client.expectSomeJson(timeout = defaultTimeout)
+      ) should contain(
         PublishedLibrary("Standard", "Base", isCached = true)
       )
     }
@@ -776,7 +822,7 @@ class LibrariesTest
           }
           """)
 
-      val response = client.expectSomeJson()
+      val response = client.expectSomeJson(timeout = defaultTimeout)
       val components = response.hcursor
         .downField("result")
         .downField("availableComponents")
@@ -802,7 +848,7 @@ class LibrariesTest
           }
           """)
 
-      val response2 = client.expectSomeJson()
+      val response2 = client.expectSomeJson(timeout = defaultTimeout)
       val components2 = response2.hcursor
         .downField("result")
         .downField("availableComponents")
@@ -833,14 +879,17 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 0,
             "result": {
               "engineVersion": $currentVersion
             }
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
 
       client.send(json"""
           { "jsonrpc": "2.0",
@@ -854,14 +903,17 @@ class LibrariesTest
             }
           }
           """)
-      client.expectJson(json"""
+      client.expectJson(
+        json"""
           { "jsonrpc": "2.0",
             "id": 1,
             "result": {
               "engineVersion": $currentVersion
             }
           }
-          """)
+          """,
+        timeout = defaultTimeout
+      )
     }
   }
 }

lib/scala/json-rpc-server-test/src/main/scala/org/enso/jsonrpc/test/JsonRpcServerTestKit.scala

@@ -57,18 +57,22 @@ abstract class JsonRpcServerTestKit
 
   def protocolFactory: ProtocolFactory
 
-  def clientControllerFactory: ClientControllerFactory
+  def clientControllerFactory(): ClientControllerFactory
+
+  var _clientControllerFactory: ClientControllerFactory = _
 
   override def beforeEach(): Unit = {
     super.beforeEach()
     val factory = protocolFactory
     factory.init()
-    server  = new JsonRpcServer(factory, clientControllerFactory)
+    _clientControllerFactory = clientControllerFactory()
-    binding = Await.result(server.bind(interface, port = 0), 3.seconds)
+    server                   = new JsonRpcServer(factory, _clientControllerFactory)
-    address = s"ws://$interface:${binding.localAddress.getPort}"
+    binding                  = Await.result(server.bind(interface, port = 0), 3.seconds)
+    address                  = s"ws://$interface:${binding.localAddress.getPort}"
   }
 
   override def afterEach(): Unit = {
+    _clientControllerFactory.shutdown()
     Await.ready(binding.terminate(10.seconds.dilated), 15.seconds.dilated)
     super.afterEach()
   }

lib/scala/json-rpc-server/src/main/scala/org/enso/jsonrpc/ClientControllerFactory.scala

@@ -17,4 +17,8 @@ trait ClientControllerFactory {
     */
   def createClientController(clientId: UUID): ActorRef
 
+  /** Shutdown resources neccessary for creating client controller actor.
+    */
+  def shutdown(): Unit
+
 }

lib/scala/project-manager/src/main/scala/org/enso/projectmanager/protocol/ManagerClientControllerFactory.scala

@@ -53,4 +53,6 @@ class ManagerClientControllerFactory[
       s"manager-client-controller-$clientId"
     )
 
+  override def shutdown(): Unit = {}
+
 }

lib/scala/project-manager/src/test/scala/org/enso/projectmanager/BaseServerSpec.scala

@@ -207,7 +207,7 @@ class BaseServerSpec extends JsonRpcServerTestKit with BeforeAndAfterAll {
       distributionConfiguration
     )
 
-  override def clientControllerFactory: ClientControllerFactory = {
+  override def clientControllerFactory(): ClientControllerFactory = {
     new ManagerClientControllerFactory[ZIO[ZAny, +*, +*]](
       system                          = system,
       projectService                  = projectService,

lib/scala/project-manager/src/test/scala/org/enso/projectmanager/protocol/ProjectShutdownSpec.scala

@@ -39,7 +39,7 @@ class ProjectShutdownSpec
 
   var clientUUID: UUID = null
 
-  override def clientControllerFactory: ClientControllerFactory = {
+  override def clientControllerFactory(): ClientControllerFactory = {
     new ManagerClientControllerFactory[ZIO[ZAny, +*, +*]](
       system                          = system,
       projectService                  = projectService,