Retry on ChecksumError (#797)

Fix for #780

BUILD.tools

@@ -7,6 +7,7 @@ jar_library(
             rev = "2.12.4",
         ),
     ],
+    scope = 'runtime'
 )
 
 jar_library(

cache/src/main/scala/coursier/Cache.scala

@@ -926,6 +926,14 @@ object Cache {
     }
   }
 
+
+  /**
+    * This method computes the task needed to get a file.
+    *
+    * Retry only applies to [[coursier.FileError.WrongChecksum]].
+    *
+    * [[coursier.FileError.DownloadError]] is handled separately at [[downloading]]
+    */
   def file[F[_]](
     artifact: Artifact,
     cache: File = default,
@@ -933,7 +941,8 @@ object Cache {
     checksums: Seq[Option[String]] = defaultChecksums,
     logger: Option[Logger] = None,
     pool: ExecutorService = defaultPool,
-    ttl: Option[Duration] = defaultTtl
+    ttl: Option[Duration] = defaultTtl,
+    retry: Int = 1
   )(implicit S: Schedulable[F]): EitherT[F, FileError, File] = {
 
     val checksums0 = if (checksums.isEmpty) Seq(None) else checksums
@@ -975,6 +984,35 @@ object Cache {
       case (f, None) => EitherT(S.point[Either[FileError, File]](Right(f)))
       case (f, Some(c)) =>
         validateChecksum(artifact, c, cache, pool).map(_ => f)
+    }.leftFlatMap {
+      case err: FileError.WrongChecksum =>
+        if (retry <= 0) {
+          EitherT(S.point(Left(err)))
+        }
+        else {
+          EitherT {
+            S.schedule[Either[FileError, Unit]](pool) {
+              val badFile = localFile(artifact.url, cache, artifact.authentication.map(_.user))
+              badFile.delete()
+              logger.foreach(_.removedCorruptFile(artifact.url, badFile, Some(err)))
+              Right(())
+            }
+          }.flatMap {
+            _ =>
+              file(
+                artifact,
+                cache,
+                cachePolicy,
+                checksums,
+                logger,
+                pool,
+                ttl,
+                retry - 1
+              )
+          }
+        }
+      case err =>
+        EitherT(S.point(Left(err)))
     }
   }
 
@@ -1123,6 +1161,8 @@ object Cache {
 
     def gettingLength(url: String): Unit = {}
     def gettingLengthResult(url: String, length: Option[Long]): Unit = {}
+
+    def removedCorruptFile(url: String, file: File, reason: Option[FileError]): Unit
   }
 
   var bufferSize = 1024*1024

cache/src/main/scala/coursier/TermDisplay.scala

@@ -509,4 +509,7 @@ class TermDisplay(
     }
   }
 
+  // TODO(wisechengyi,alexarchambault): implement this
+  override def removedCorruptFile(url: String, file: File, reason: Option[FileError]): Unit = {}
+
 }

cli/src/main/scala-2.12/coursier/cli/Helper.scala

@@ -644,7 +644,8 @@ class Helper(
         checksums = checksums,
         logger = logger,
         pool = pool,
-        ttl = ttl0
+        ttl = ttl0,
+        retry = common.retryCount
       )
 
       (file(cachePolicies.head) /: cachePolicies.tail)(_ orElse file(_))

cli/src/main/scala-2.12/coursier/cli/options/CacheOptions.scala

@@ -5,7 +5,6 @@ import coursier.Cache
 
 final case class CacheOptions(
   @Help("Cache directory (defaults to environment variable COURSIER_CACHE or ~/.coursier/cache/v1)")
-  @Short("C")
     cache: String = Cache.default.toString
 )
 

cli/src/main/scala-2.12/coursier/cli/options/CommonOptions.scala

@@ -95,7 +95,11 @@ final case class CommonOptions(
   @Help("Swap the mainline Scala JARs by Typelevel ones")
     typelevel: Boolean = false,
   @Recurse
-    cacheOptions: CacheOptions = CacheOptions()
+    cacheOptions: CacheOptions = CacheOptions(),
+
+  @Help("Retry limit for Checksum error when fetching a file")
+    retryCount: Int = 1
+
 ) {
   val verbosityLevel = Tag.unwrap(verbose) - (if (quiet) 1 else 0)
   lazy val classifier0 = classifier.flatMap(_.split(',')).filter(_.nonEmpty).toSet

cli/src/test/scala-2.12/coursier/cli/BUILD

@@ -11,7 +11,7 @@ junit_tests(
 scala_library(
     name='lib',
     dependencies = [
-        "3rdparty/jvm:caseapp",
+        "cache/src/main/scala:cache"
     ],
     sources = ["CliTestLib.scala"],
 )

cli/src/test/scala-2.12/coursier/cli/CliFetchIntegrationTest.scala

@@ -4,9 +4,13 @@ import java.io._
 
 import java.net.URLEncoder.encode
 import argonaut.Argonaut._
-import coursier.cli.util.{DepNode, ReportNode}
 import caseapp.core.RemainingArgs
 import coursier.cli.options._
+import coursier.cli.util.{DepNode, ReportNode}
+import coursier.internal.FileUtil
+import java.io._
+import java.net.URLEncoder.encode
+import java.nio.charset.StandardCharsets.UTF_8
 import org.junit.runner.RunWith
 import org.scalatest.FlatSpec
 import org.scalatest.junit.JUnitRunner
@@ -809,4 +813,58 @@ class CliFetchIntegrationTest extends FlatSpec with CliTestLib {
       assert(depNode.get.dependencies.head.contains("org.tukaani:xz:1.2"))
     }
   }
+
+  "Bad pom resolve" should "succeed with retry" in withTempDir("tmp_dir") {
+    dir => {
+      def runFetchJunit = {
+        val fetchOpt = FetchOptions(common = CommonOptions(cacheOptions = CacheOptions(cache = dir.getAbsolutePath)))
+        val fetch = Fetch(fetchOpt, RemainingArgs(Seq("junit:junit:4.12"), Seq()))
+        assert(fetch.files0.map(_.getName).toSet
+          .equals(Set("junit-4.12.jar", "hamcrest-core-1.3.jar")))
+        val junitJarPath = fetch.files0.map(_.getAbsolutePath()).filter(_.contains("junit-4.12.jar"))
+          .head
+        val junitPomFile = new File(junitJarPath.replace(".jar", ".pom"))
+        val junitPomShaFile = new File(junitJarPath.replace(".jar", ".pom.sha1"))
+        assert(junitPomFile.exists())
+        assert(junitPomShaFile.exists())
+        junitPomFile
+      }
+
+      val junitPomFile: File = runFetchJunit
+      val originalPomContent = FileUtil.readAllBytes(junitPomFile)
+
+      // Corrupt the pom content
+      FileUtil.write(junitPomFile, "bad pom".getBytes(UTF_8))
+
+      // Run fetch again and it should pass because of retrying om the bad pom.
+      val pom = runFetchJunit
+      assert(FileUtil.readAllBytes(pom).sameElements(originalPomContent))
+    }
+  }
+
+  "Bad jar resolve" should "succeed with retry" in withTempDir("tmp_dir") {
+    dir => {
+      def runFetchJunit = {
+        val fetchOpt = FetchOptions(common = CommonOptions(cacheOptions = CacheOptions(cache = dir.getAbsolutePath)))
+        val fetch = Fetch(fetchOpt, RemainingArgs(Seq("junit:junit:4.12"), Seq()))
+        assert(fetch.files0.map(_.getName).toSet
+          .equals(Set("junit-4.12.jar", "hamcrest-core-1.3.jar")))
+        val junitJarPath = fetch.files0.map(_.getAbsolutePath()).filter(_.contains("junit-4.12.jar"))
+          .head
+        val junitJarFile = new File(junitJarPath)
+        junitJarFile
+      }
+
+      val originalJunitJar: File = runFetchJunit
+      val originalJunitJarContent = FileUtil.readAllBytes(originalJunitJar)
+
+      // Corrupt the jar content
+      FileUtil.write(originalJunitJar, "bad jar".getBytes(UTF_8))
+
+      // Run fetch again and it should pass because of retrying on the bad jar.
+      val jar = runFetchJunit
+      assert(FileUtil.readAllBytes(jar).sameElements(originalJunitJarContent))
+    }
+  }
+
 }

cli/src/test/scala-2.12/coursier/cli/CliTestLib.scala

@@ -1,5 +1,6 @@
 package coursier.cli
 
+import coursier.internal.FileUtil
 import java.io.{File, FileWriter}
 
 
@@ -14,10 +15,36 @@ trait CliTestLib {
     writer.flush()
     try {
       testCode(file, writer) // "loan" the fixture to the test
-    }
-    finally {
+    } finally {
       writer.close()
       file.delete()
     }
   }
+
+  def withTempDir(
+      prefix: String
+  )(testCode: File => Any) {
+    val dir = FileUtil.createTempDirectory(prefix) // create the fixture
+    try {
+      testCode(dir) // "loan" the fixture to the test
+    } finally {
+      cleanDir(dir)
+    }
+  }
+
+  def cleanDir(tmpDir: File): Unit = {
+    def delete(f: File): Boolean =
+      if (f.isDirectory) {
+        val removedContent =
+          Option(f.listFiles()).toSeq.flatten.map(delete).forall(x => x)
+        val removedDir = f.delete()
+
+        removedContent && removedDir
+      } else
+        f.delete()
+
+    if (!delete(tmpDir))
+      Console.err.println(
+        s"Warning: unable to remove temporary directory $tmpDir")
+  }
 }

core/shared/src/main/scala/coursier/util/EitherT.scala

@@ -24,6 +24,16 @@ final case class EitherT[F[_], L, R](run: F[Either[L, R]]) {
       M.map(run)(e => e.left.map(f))
     )
 
+  def leftFlatMap[S](f: L => EitherT[F, S, R])(implicit M: Monad[F]): EitherT[F, S, R] =
+    EitherT(
+      M.bind(run) {
+        case Left(l) =>
+          f(l).run
+        case Right(r) =>
+          M.point(Right(r))
+      }
+    )
+
   def orElse(other: => EitherT[F, L, R])(implicit M: Monad[F]): EitherT[F, L, R] =
     EitherT(
       M.bind(run) {