Repository: 7mind/izumi (scala)

Issue:
distage-framework-docker: Do not duplicate containers spawning in parallel when `reuse=true`
Currently even when `reuse` is turned on, containers may be duplicated under parallel tests when they launch at the same time, because their spawning is non-atomic.

Current plan: use `nio.FileLock` in temporary directory to signify that a reusable container is spawning and prevent duplicates from spawning

Pull request:
Fix #945

Patch:
diff --git a/distage/distage-core-api/src/main/scala/izumi/distage/model/effect/DIEffect.scala b/distage/distage-core-api/src/main/scala/izumi/distage/model/effect/DIEffect.scala
index 62778ed0a0..5329deee1e 100644
--- a/distage/distage-core-api/src/main/scala/izumi/distage/model/effect/DIEffect.scala
+++ b/distage/distage-core-api/src/main/scala/izumi/distage/model/effect/DIEffect.scala
@@ -14,6 +14,7 @@ trait DIEffect[F[_]] extends DIApplicative[F] {
   def flatMap[A, B](fa: F[A])(f: A => F[B]): F[B]
   def bracket[A, B](acquire: => F[A])(release: A => F[Unit])(use: A => F[B]): F[B]
   def bracketCase[A, B](acquire: => F[A])(release: (A, Option[Throwable]) => F[Unit])(use: A => F[B]): F[B]
+  final def bracketAuto[A <: AutoCloseable, B](acquire: => F[A])(use: A => F[B]): F[B] = bracket(acquire)(a => maybeSuspend(a.close()))(use)
 
   /** A weaker version of `delay`. Does not guarantee _actual_
     * suspension of side-effects, because DIEffect[Identity] is allowed */
diff --git a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/DockerContainer.scala b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/DockerContainer.scala
index 948e1e68bd..a0b9622932 100644
--- a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/DockerContainer.scala
+++ b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/DockerContainer.scala
@@ -18,6 +18,7 @@ import izumi.fundamentals.platform.language.Quirks._
 import izumi.fundamentals.platform.network.IzSockets
 import izumi.logstage.api.IzLogger
 
+import scala.concurrent.duration._
 import scala.jdk.CollectionConverters._
 
 trait ContainerDef {
@@ -179,54 +180,62 @@ object DockerContainer {
     }
 
     private[this] def runReused(ports: Seq[PortDecl]): F[DockerContainer[T]] = {
-      for {
-        containers <- DIEffect[F].maybeSuspend {
-          // FIXME: temporary hack to allow missing containers to skip tests (happens when both DockerWrapper & integration check that depends on Docker.Container are memoized)
-          try {
-            client
-              .listContainersCmd()
-              .withAncestorFilter(List(config.image).asJava)
-              .withStatusFilter(List("running").asJava)
-              .exec()
-          } catch {
-            case c: Throwable =>
-              throw new IntegrationCheckException(Seq(ResourceCheck.ResourceUnavailable(c.getMessage, Some(c))))
+      logger.info(s"Running container with reused option with ${config.pullTimeout}.")
+      FileLockMutex.withLocalMutex(
+        s"${config.image.replace("/", "_")}:${config.ports.mkString(";")}",
+        logger,
+        1.second,
+        config.pullTimeout.toSeconds.toInt
+      ) {
+        for {
+          containers <- DIEffect[F].maybeSuspend {
+            // FIXME: temporary hack to allow missing containers to skip tests (happens when both DockerWrapper & integration check that depends on Docker.Container are memoized)
+            try {
+              client
+                .listContainersCmd()
+                .withAncestorFilter(List(config.image).asJava)
+                .withStatusFilter(List("running").asJava)
+                .exec()
+            } catch {
+              case c: Throwable =>
+                throw new IntegrationCheckException(Seq(ResourceCheck.ResourceUnavailable(c.getMessage, Some(c))))
+            }
+          }.map(_.asScala.toList.sortBy(_.getId))
+
+          candidates = {
+            val portSet = ports.map(_.port).toSet
+            containers.iterator.flatMap {
+              c =>
+                val inspection = client.inspectContainerCmd(c.getId).exec()
+                mapContainerPorts(inspection) match {
+                  case Left(value) =>
+                    logger.info(s"Container ${c.getId} missing ports $value so will not be reused")
+                    Seq.empty
+                  case Right(value) =>
+                    Seq((c, inspection, value))
+                }
+            }.find {
+              case (_, _, eports) =>
+                portSet.diff(eports.keySet).isEmpty
+            }
           }
-        }.map(_.asScala.toList.sortBy(_.getId))
-
-        candidates = {
-          val portSet = ports.map(_.port).toSet
-          containers.iterator.flatMap {
-            c =>
-              val inspection = client.inspectContainerCmd(c.getId).exec()
-              mapContainerPorts(inspection) match {
-                case Left(value) =>
-                  logger.info(s"Container ${c.getId} missing ports $value so will not be reused")
-                  Seq.empty
-                case Right(value) =>
-                  Seq((c, inspection, value))
-              }
-          }.find {
-            case (_, _, eports) =>
-              portSet.diff(eports.keySet).isEmpty
+          existing <- candidates match {
+            case Some((c, inspection, existingPorts)) =>
+              val unverified = DockerContainer[T](
+                id = ContainerId(c.getId),
+                name = inspection.getName,
+                ports = existingPorts,
+                containerConfig = config,
+                clientConfig = clientw.clientConfig,
+                availablePorts = Map.empty,
+                hostName = inspection.getConfig.getHostName
+              )
+              await(unverified)
+            case None =>
+              doRun(ports)
           }
-        }
-        existing <- candidates match {
-          case Some((c, inspection, existingPorts)) =>
-            val unverified = DockerContainer[T](
-              id = ContainerId(c.getId),
-              name = inspection.getName,
-              ports = existingPorts,
-              containerConfig = config,
-              clientConfig = clientw.clientConfig,
-              availablePorts = Map.empty,
-              hostName = inspection.getConfig.getHostName
-            )
-            await(unverified)
-          case None =>
-            doRun(ports)
-        }
-      } yield existing
+        } yield existing
+      }
     }
 
     private[this] def doRun(ports: Seq[PortDecl]): F[DockerContainer[T]] = {
diff --git a/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/FileLockMutex.scala b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/FileLockMutex.scala
new file mode 100644
index 0000000000..80bff1c4f3
--- /dev/null
+++ b/distage/distage-framework-docker/src/main/scala/izumi/distage/docker/FileLockMutex.scala
@@ -0,0 +1,55 @@
+package izumi.distage.docker
+
+import java.io.File
+import java.nio.channels.{FileChannel, OverlappingFileLockException}
+import java.nio.file.StandardOpenOption
+
+import izumi.distage.model.effect.DIEffect.syntax._
+import izumi.distage.model.effect.{DIEffect, DIEffectAsync}
+import izumi.logstage.api.IzLogger
+
+import scala.concurrent.duration._
+
+object FileLockMutex {
+  def withLocalMutex[F[_]: DIEffect: DIEffectAsync, E](
+    filename: String,
+    logger: IzLogger,
+    waitFor: FiniteDuration = 1.second,
+    maxAttempts: Int = 10
+  )(eff: F[E]): F[E] = {
+    def acquireAndRun(chanel: FileChannel, attempts: Int = 0): F[E] = {
+      DIEffect[F].maybeSuspend {
+        logger.debug(s"Attempt ${attempts -> "num"} to acquire lock for $filename.")
+        try {
+          Option(chanel.tryLock())
+        } catch {
+          case _: OverlappingFileLockException => None
+        }
+      }.flatMap {
+        case Some(v) =>
+          eff.guarantee(DIEffect[F].maybeSuspend(v.close()))
+        case None if attempts < maxAttempts =>
+          DIEffectAsync[F].sleep(waitFor).flatMap(_ => acquireAndRun(chanel, attempts + 1))
+        case _ =>
+          logger.warn(s"Cannot acquire lock for image $filename after $attempts. This may lead to creation of a new container duplicate.")
+          eff
+      }
+    }
+
+    val tmpDir = System.getProperty("java.io.tmpdir")
+    val file = new File(s"$tmpDir/$filename.tmp")
+    file.createNewFile()
+    DIEffect[F].bracket(DIEffect[F].maybeSuspend(FileChannel.open(file.toPath, StandardOpenOption.WRITE))) {
+      ch =>
+        DIEffect[F].definitelyRecover {
+          DIEffect[F].maybeSuspend {
+            ch.close()
+            file.delete()
+            ()
+          }
+        }(_ => DIEffect[F].unit)
+    } {
+      acquireAndRun(_)
+    }
+  }
+}


Test patch:
diff --git a/distage/distage-framework-docker/src/test/scala/izumi/distage/testkit/docker/DistageTestDockerBIO.scala b/distage/distage-framework-docker/src/test/scala/izumi/distage/testkit/docker/DistageTestDockerBIO.scala
index d205206a9f..64e062e7df 100644
--- a/distage/distage-framework-docker/src/test/scala/izumi/distage/testkit/docker/DistageTestDockerBIO.scala
+++ b/distage/distage-framework-docker/src/test/scala/izumi/distage/testkit/docker/DistageTestDockerBIO.scala
@@ -1,17 +1,21 @@
 package izumi.distage.testkit.docker
 
+import java.util.UUID
+
 import distage.DIKey
+import izumi.distage.model.definition.ModuleDef
 import izumi.distage.testkit.TestConfig
 import izumi.distage.testkit.docker.fixtures.PgSvcExample
 import izumi.distage.testkit.scalatest.DistageBIOSpecScalatest
 import izumi.fundamentals.platform.properties.EnvVarsCI
+import izumi.logstage.api.Log
 import zio.IO
 
-final class DistageTestDockerBIO extends DistageBIOSpecScalatest[IO] {
-
+// this tests needed to check mutex for reusable containers during parallel test runs
+abstract class DistageTestDockerBIO extends DistageBIOSpecScalatest[IO] {
   // ignore docker tests on CI (nested docker trouble)
   if (!EnvVarsCI.isIzumiCI()) {
-    "distage test runner" should {
+    "distage test runner should start only one container for reusable" should {
       "support docker resources" in {
         service: PgSvcExample =>
           for {
@@ -30,10 +34,27 @@ final class DistageTestDockerBIO extends DistageBIOSpecScalatest[IO] {
 
   override protected def config: TestConfig = {
     super.config.copy(
-      memoizationRoots = Set(
-        DIKey.get[PgSvcExample],
-//        DIKey.get[DynamoDocker.Container],
-//        DIKey.get[PostgresDocker.Container],
-      ))
+      memoizationRoots = Set(DIKey.get[PgSvcExample]),
+      parallelTests = true,
+      parallelEnvs = true,
+      moduleOverrides = super.config.moduleOverrides overridenBy new ModuleDef { make[UUID].fromValue(UUID.randomUUID()) },
+      testRunnerLogLevel = Log.Level.Debug
+    )
   }
 }
+
+final class DistageTestDockerBIO1 extends DistageTestDockerBIO
+final class DistageTestDockerBIO2 extends DistageTestDockerBIO
+final class DistageTestDockerBIO3 extends DistageTestDockerBIO
+final class DistageTestDockerBIO4 extends DistageTestDockerBIO
+final class DistageTestDockerBIO5 extends DistageTestDockerBIO
+final class DistageTestDockerBIO6 extends DistageTestDockerBIO
+final class DistageTestDockerBIO7 extends DistageTestDockerBIO
+final class DistageTestDockerBIO8 extends DistageTestDockerBIO
+final class DistageTestDockerBIO9 extends DistageTestDockerBIO
+final class DistageTestDockerBIO10 extends DistageTestDockerBIO
+final class DistageTestDockerBIO11 extends DistageTestDockerBIO
+final class DistageTestDockerBIO12 extends DistageTestDockerBIO
+final class DistageTestDockerBIO13 extends DistageTestDockerBIO
+final class DistageTestDockerBIO14 extends DistageTestDockerBIO
+final class DistageTestDockerBIO15 extends DistageTestDockerBIO