Repository: BlueBrain/nexus (scala)

Issue:
When computing the file attributes asynchronously, events are created when they are not yet computed
**Describe the bug**
When a file is linked, its attributes (size + checksum) are computed asynchronously by a task.
This task currently creates events even if attributes are not yet computed

**What part of Nexus is affected**
- [ ] Fusion
- [X] Delta
- [ ] CLI
- [ ] Documentation

**Expected behavior**
An event is registered only when the attributes are computed

Pull request description:
Fixes #4743

Patch:
diff --git a/delta/plugins/storage/src/main/resources/storage.conf b/delta/plugins/storage/src/main/resources/storage.conf
index b8b2d55be5..dcd48cef48 100644
--- a/delta/plugins/storage/src/main/resources/storage.conf
+++ b/delta/plugins/storage/src/main/resources/storage.conf
@@ -67,8 +67,8 @@ plugins.storage {
       show-location = true
       # the default maximum allowed file size (in bytes) for uploaded files. 10 GB
       default-max-file-size = 10737418240
-      # Retry strategy for digest computation
-      digest-computation = ${app.defaults.retry-strategy}
+      # Retry delay for digest computation
+      digest-computation-retry-delay = 5s
     }
     # the storages event log configuration
     event-log = ${app.defaults.event-log}
diff --git a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/StoragePluginModule.scala b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/StoragePluginModule.scala
index c6f3f31543..e2e15596d9 100644
--- a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/StoragePluginModule.scala
+++ b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/StoragePluginModule.scala
@@ -6,7 +6,7 @@ import cats.effect.{Clock, IO}
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.{ClasspathResourceLoader, TransactionalFileCopier, UUIDF}
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.config.ElasticSearchViewsConfig
-import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.Files
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.{FileAttributesUpdateStream, Files}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.Files.FilesLog
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.batch.{BatchCopy, BatchFiles}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.contexts.{files => fileCtxId}
@@ -156,43 +156,40 @@ class StoragePluginModule(priority: Int) extends ModuleDef {
     ScopedEventLog(Files.definition(clock), cfg.files.eventLog, xas)
   }
 
-  make[Files]
-    .fromEffect {
-      (
-          cfg: StoragePluginConfig,
-          storageTypeConfig: StorageTypeConfig,
-          aclCheck: AclCheck,
-          fetchContext: FetchContext,
-          storages: Storages,
-          supervisor: Supervisor,
-          storagesStatistics: StoragesStatistics,
-          xas: Transactors,
-          clock: Clock[IO],
-          uuidF: UUIDF,
-          as: ActorSystem[Nothing],
-          remoteDiskStorageClient: RemoteDiskStorageClient
-      ) =>
-        IO
-          .delay(
-            Files(
-              fetchContext,
-              aclCheck,
-              storages,
-              storagesStatistics,
-              xas,
-              storageTypeConfig,
-              cfg.files,
-              remoteDiskStorageClient,
-              clock
-            )(
-              uuidF,
-              as
-            )
-          )
-          .flatTap { files =>
-            Files.startDigestStream(files, supervisor, storageTypeConfig)
-          }
-    }
+  make[Files].from {
+    (
+        cfg: StoragePluginConfig,
+        storageTypeConfig: StorageTypeConfig,
+        aclCheck: AclCheck,
+        fetchContext: FetchContext,
+        storages: Storages,
+        storagesStatistics: StoragesStatistics,
+        xas: Transactors,
+        clock: Clock[IO],
+        uuidF: UUIDF,
+        as: ActorSystem[Nothing],
+        remoteDiskStorageClient: RemoteDiskStorageClient
+    ) =>
+      Files(
+        fetchContext,
+        aclCheck,
+        storages,
+        storagesStatistics,
+        xas,
+        storageTypeConfig,
+        cfg.files,
+        remoteDiskStorageClient,
+        clock
+      )(
+        uuidF,
+        as
+      )
+  }
+
+  make[FileAttributesUpdateStream].fromEffect {
+    (files: Files, storages: Storages, storageTypeConfig: StorageTypeConfig, supervisor: Supervisor) =>
+      FileAttributesUpdateStream.start(files, storages, storageTypeConfig.remoteDisk, supervisor)
+  }
 
   make[TransactionalFileCopier].fromValue(TransactionalFileCopier.mk())
 
diff --git a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/Files.scala b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/Files.scala
index cf8394256d..adc9a0761f 100644
--- a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/Files.scala
+++ b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/Files.scala
@@ -6,10 +6,8 @@ import akka.http.scaladsl.model.ContentTypes.`application/octet-stream`
 import akka.http.scaladsl.model.{BodyPartEntity, ContentType, HttpEntity, Uri}
 import cats.effect.{Clock, IO}
 import cats.syntax.all._
-import ch.epfl.bluebrain.nexus.delta.kernel.cache.LocalCache
 import ch.epfl.bluebrain.nexus.delta.kernel.kamon.KamonMetricComponent
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.UUIDF
-import ch.epfl.bluebrain.nexus.delta.kernel.{Logger, RetryStrategy}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.Files._
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.Digest.{ComputedDigest, NotComputedDigest}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileAttributes.FileAttributesOrigin.Client
@@ -18,7 +16,7 @@ import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileEvent._
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileRejection._
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model._
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.schemas.{files => fileSchema}
-import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StoragesConfig.{RemoteDiskStorageConfig, StorageTypeConfig}
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StoragesConfig.StorageTypeConfig
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.StorageRejection.{StorageFetchRejection, StorageIsDeprecated}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.{DigestAlgorithm, Storage, StorageRejection, StorageType}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.StorageFileRejection.{FetchAttributeRejection, FetchFileRejection, SaveFileRejection}
@@ -42,11 +40,8 @@ import ch.epfl.bluebrain.nexus.delta.sourcing.ScopedEntityDefinition.Tagger
 import ch.epfl.bluebrain.nexus.delta.sourcing._
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Subject
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Tag.UserTag
-import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, ProjectRef, ResourceRef}
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, ProjectRef, ResourceRef, SuccessElemStream}
 import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset
-import ch.epfl.bluebrain.nexus.delta.sourcing.stream.Elem.{DroppedElem, SuccessElem}
-import ch.epfl.bluebrain.nexus.delta.sourcing.stream._
-import fs2.Stream
 
 import java.util.UUID
 
@@ -508,76 +503,7 @@ final class Files(
   private def generateId(pc: ProjectContext): IO[Iri] =
     uuidF().map(uuid => pc.base.iri / uuid.toString)
 
-  /**
-    * Starts a stream that attempts to update file attributes asynchronously for linked files in remote storages
-    * @param offset
-    *   the offset to start from
-    * @return
-    */
-  private[files] def attributesUpdateStream(offset: Offset): Stream[IO, Elem[Unit]] = {
-    for {
-      // The stream will start only if remote storage is enabled
-      retryStrategy <- Stream.iterable[IO, RemoteDiskStorageConfig](config.remoteDisk).map { c =>
-                         RetryStrategy[Throwable](
-                           c.digestComputation,
-                           {
-                             case FetchRejection(_, _, FetchFileRejection.UnexpectedFetchError(_, _)) => true
-                             case DigestNotComputed(_)                                                => true
-                             case _                                                                   => false
-                           },
-                           RetryStrategy.logError(logger, "file attributes update")(_, _)
-                         )
-                       }
-      // We cache storage information
-      storageCache  <-
-        Stream.eval[IO, LocalCache[ResourceRef.Revision, Storage]](LocalCache[ResourceRef.Revision, Storage]())
-      fetchStorage   = (f: FileState) =>
-                         storageCache.getOrElseUpdate(
-                           f.storage,
-                           storages
-                             .fetch(IdSegmentRef(f.storage), f.project)
-                             .map(_.value)
-                             .adaptError { case e: StorageRejection =>
-                               WrappedStorageRejection(e)
-                             }
-                         )
-      stream        <- log
-                         .states(Scope.root, offset)
-                         .map { elem =>
-                           elem.value match {
-                             case f
-                                 if f.storageType == StorageType.RemoteDiskStorage && !f.attributes.digest.computed && !f.deprecated =>
-                               SuccessElem(
-                                 entityType,
-                                 elem.id,
-                                 Some(elem.value.project),
-                                 elem.instant,
-                                 elem.offset,
-                                 f,
-                                 elem.rev
-                               )
-                             case _ =>
-                               DroppedElem(
-                                 entityType,
-                                 elem.id,
-                                 Some(elem.value.project),
-                                 elem.instant,
-                                 elem.offset,
-                                 elem.rev
-                               )
-                           }
-                         }
-                         .evalMap {
-                           _.traverse { f =>
-                             for {
-                               _       <- logger.info(s"Updating attributes for file ${f.id} in ${f.project}")
-                               storage <- fetchStorage(f)
-                               _       <- updateAttributes(f, storage)
-                             } yield ()
-                           }.retry(retryStrategy)
-                         }
-    } yield stream
-  }
+  def states(offset: Offset): SuccessElemStream[FileState] = log.states(Scope.root, offset)
 
   private[files] def updateAttributes(iri: Iri, project: ProjectRef): IO[Unit] =
     for {
@@ -588,7 +514,7 @@ final class Files(
       _       <- updateAttributes(f: FileState, storage: Storage)
     } yield ()
 
-  private def updateAttributes(f: FileState, storage: Storage): IO[Unit] = {
+  private[files] def updateAttributes(f: FileState, storage: Storage): IO[Unit] = {
     val attr = f.attributes
     for {
       _        <- IO.raiseWhen(f.attributes.digest.computed)(DigestAlreadyComputed(f.id))
@@ -608,8 +534,6 @@ final class Files(
 
 object Files {
 
-  private val logger = Logger[Files]
-
   /**
     * The file entity type.
     */
@@ -700,6 +624,7 @@ object Files {
       case Some(s) if s.rev != c.rev               => IO.raiseError(IncorrectRev(c.rev, s.rev))
       case Some(s) if s.deprecated                 => IO.raiseError(FileIsDeprecated(c.id))
       case Some(s) if s.attributes.digest.computed => IO.raiseError(DigestAlreadyComputed(s.id))
+      case Some(s) if !c.digest.computed           => IO.raiseError(DigestNotComputed(s.id))
       case Some(s)                                 =>
         // format: off
         clock.realTimeInstant
@@ -815,19 +740,4 @@ object Files {
       storageTypeConfig
     )
   }
-
-  val metadata: ProjectionMetadata = ProjectionMetadata("system", "file-attributes-update", None, None)
-
-  def startDigestStream(files: Files, supervisor: Supervisor, config: StorageTypeConfig): IO[Unit] =
-    IO.whenA(config.remoteDisk.isDefined) {
-      supervisor
-        .run(
-          CompiledProjection.fromStream(
-            metadata,
-            ExecutionStrategy.PersistentSingleNode,
-            files.attributesUpdateStream(_)
-          )
-        )
-        .void
-    }
 }
diff --git a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/StoragesConfig.scala b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/StoragesConfig.scala
index cab040b7d0..52747cb13b 100644
--- a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/StoragesConfig.scala
+++ b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/StoragesConfig.scala
@@ -2,7 +2,7 @@ package ch.epfl.bluebrain.nexus.delta.plugins.storage.storages
 
 import akka.http.scaladsl.model.Uri
 import cats.implicits._
-import ch.epfl.bluebrain.nexus.delta.kernel.{RetryStrategyConfig, Secret}
+import ch.epfl.bluebrain.nexus.delta.kernel.Secret
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StoragesConfig.StorageTypeConfig
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.{AbsolutePath, DigestAlgorithm, StorageType}
 import ch.epfl.bluebrain.nexus.delta.sdk.auth.Credentials
@@ -16,6 +16,7 @@ import pureconfig.generic.auto._
 import pureconfig.{ConfigConvert, ConfigReader}
 
 import scala.annotation.nowarn
+import scala.concurrent.duration.FiniteDuration
 
 /**
   * Configuration for the Storages module.
@@ -195,7 +196,7 @@ object StoragesConfig {
     *   flag to decide whether or not to show the absolute location of the files in the metadata response
     * @param defaultMaxFileSize
     *   the default maximum allowed file size (in bytes) for uploaded files
-    * @param digestComputation
+    * @param digestComputationRetryDelay
     *   retry configuration for the digest computation task
     */
   final case class RemoteDiskStorageConfig(
@@ -206,7 +207,7 @@ object StoragesConfig {
       defaultWritePermission: Permission,
       showLocation: Boolean,
       defaultMaxFileSize: Long,
-      digestComputation: RetryStrategyConfig
+      digestComputationRetryDelay: FiniteDuration
   ) extends StorageTypeEntryConfig
 
   implicit private val uriConverter: ConfigConvert[Uri] =


Test patch:
diff --git a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FileAttributesUpdateStream.scala b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FileAttributesUpdateStream.scala
new file mode 100644
index 0000000000..1858fad1da
--- /dev/null
+++ b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FileAttributesUpdateStream.scala
@@ -0,0 +1,122 @@
+package ch.epfl.bluebrain.nexus.delta.plugins.storage.files
+
+import cats.effect.IO
+import cats.syntax.all._
+import ch.epfl.bluebrain.nexus.delta.kernel.Logger
+import ch.epfl.bluebrain.nexus.delta.kernel.cache.LocalCache
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileRejection._
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileState
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.Storages
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StoragesConfig.RemoteDiskStorageConfig
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.{Storage, StorageRejection, StorageType}
+import ch.epfl.bluebrain.nexus.delta.sdk.model.IdSegmentRef
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{ProjectRef, ResourceRef, SuccessElemStream}
+import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset
+import ch.epfl.bluebrain.nexus.delta.sourcing.stream._
+import fs2.Stream
+import retry.RetryPolicies.constantDelay
+import retry.implicits.retrySyntaxError
+
+import scala.concurrent.duration.FiniteDuration
+
+/**
+  * Stream that attempts to update file attributes asynchronously for linked files in remote storages
+  */
+sealed trait FileAttributesUpdateStream
+
+object FileAttributesUpdateStream {
+  private val logger = Logger[FileAttributesUpdateStream]
+
+  private val metadata: ProjectionMetadata = ProjectionMetadata("system", "file-attributes-update", None, None)
+
+  final object Disabled extends FileAttributesUpdateStream
+
+  final class Impl(
+      streamFiles: Offset => SuccessElemStream[FileState],
+      fetchStorage: (ProjectRef, ResourceRef.Revision) => IO[Storage],
+      updateAttributes: (FileState, Storage) => IO[Unit],
+      retryDelay: FiniteDuration
+  ) extends FileAttributesUpdateStream {
+
+    def run(offset: Offset): Stream[IO, Elem[Unit]] =
+      streamFiles(offset).evalMap {
+        _.evalMapFilter {
+          processFile
+        }
+      }
+
+    private[files] def processFile(file: FileState) = {
+      if (file.storageType == StorageType.RemoteDiskStorage && !file.attributes.digest.computed && !file.deprecated) {
+        for {
+          _       <- logger.debug(s"Attempt to update attributes for file ${file.id} in ${file.project}")
+          storage <- fetchStorage(file.project, file.storage)
+          _       <- updateAttributes(file, storage)
+          _       <- logger.info(s"Attributes for file ${file.id} in ${file.project} have been updated.")
+        } yield Some(())
+      } else IO.none
+    }.retryingOnAllErrors(
+      constantDelay(retryDelay),
+      {
+        case (_: DigestNotComputed, details) =>
+          IO.whenA(details.retriesSoFar % 10 == 0)(
+            logger.info(
+              s"Digest for file '${file.id}' in '${file.project}' is not yet completed after '${details.cumulativeDelay}'."
+            )
+          )
+        case (error, details)                =>
+          logger.error(error)(
+            s"Digest for file '${file.id}' in '${file.project}' ended in an error at attempt '${details.retriesSoFar}' and after '${details.cumulativeDelay}'."
+          )
+      }
+    )
+  }
+
+  def apply(
+      files: Files,
+      storages: Storages,
+      configOpt: Option[RemoteDiskStorageConfig]
+  ): IO[FileAttributesUpdateStream] =
+    configOpt match {
+      case Some(config) =>
+        LocalCache[(ProjectRef, ResourceRef.Revision), Storage]().map { storageCache =>
+          def fetchStorage(project: ProjectRef, id: ResourceRef.Revision) =
+            storageCache.getOrElseUpdate(
+              (project, id),
+              storages
+                .fetch(IdSegmentRef(id), project)
+                .map(_.value)
+                .adaptError { case e: StorageRejection =>
+                  WrappedStorageRejection(e)
+                }
+            )
+
+          new Impl(
+            files.states,
+            fetchStorage,
+            files.updateAttributes,
+            config.digestComputationRetryDelay
+          )
+        }
+      case None         => IO.pure(Disabled)
+    }
+
+  def start(
+      files: Files,
+      storages: Storages,
+      configOpt: Option[RemoteDiskStorageConfig],
+      supervisor: Supervisor
+  ): IO[FileAttributesUpdateStream] =
+    apply(files, storages, configOpt).flatTap {
+      case enabled: Impl    =>
+        supervisor
+          .run(
+            CompiledProjection.fromStream(
+              metadata,
+              ExecutionStrategy.PersistentSingleNode,
+              enabled.run
+            )
+          )
+      case _: Disabled.type =>
+        logger.debug("Remote storage is disabled, the update attributes task is disabled too.")
+    }
+}
diff --git a/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FileAttributesUpdateStreamSuite.scala b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FileAttributesUpdateStreamSuite.scala
new file mode 100644
index 0000000000..2f76118936
--- /dev/null
+++ b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FileAttributesUpdateStreamSuite.scala
@@ -0,0 +1,93 @@
+package ch.epfl.bluebrain.nexus.delta.plugins.storage.files
+
+import akka.http.scaladsl.model.{ContentTypes, Uri}
+import cats.effect.IO
+import cats.effect.kernel.Ref
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.generators.FileGen
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.Digest.{ComputedDigest, NotComputedDigest}
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileAttributes.FileAttributesOrigin.Client
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileRejection.DigestNotComputed
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.{FileAttributes, FileState}
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StorageFixtures
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.Storage.RemoteDiskStorage
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.{DigestAlgorithm, StorageType}
+import ch.epfl.bluebrain.nexus.delta.rdf.Vocabulary.nxv
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{Label, ProjectRef, ResourceRef}
+import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
+import fs2.Stream
+
+import java.util.UUID
+import scala.concurrent.duration._
+
+class FileAttributesUpdateStreamSuite extends NexusSuite with StorageFixtures {
+
+  private val project = ProjectRef.unsafe("org", "projg")
+  private val id      = nxv + "file"
+
+  private val storage = RemoteDiskStorage(nxv + "remote", project, remoteVal, json"""{"disk": "value"}""")
+
+  private val storageRef = ResourceRef.Revision(storage.id, 1)
+
+  private val mediaType  = Some(ContentTypes.`text/plain(UTF-8)`)
+  private val attributes = FileAttributes(
+    UUID.randomUUID(),
+    location = "http://localhost/my/file.txt",
+    path = Uri.Path("my/file.txt"),
+    filename = "myfile.txt",
+    mediaType = mediaType,
+    keywords = Map(Label.unsafe("key") -> "value"),
+    bytes = 10,
+    NotComputedDigest,
+    Client
+  )
+  private val validFile  =
+    FileGen.state(id, project, storageRef, attributes, storageType = StorageType.RemoteDiskStorage)
+
+  private def updateStream(updateAttributes: IO[Unit]) = new FileAttributesUpdateStream.Impl(
+    _ => Stream.empty,
+    (_, _) => IO.pure(storage),
+    (_, _) => updateAttributes,
+    50.millis
+  )
+
+  private def assertProcess(file: FileState, expectedAttempts: Int) =
+    for {
+      ref <- Ref.of[IO, Int](0)
+      _   <- updateStream(ref.update(_ + 1)).processFile(file)
+      _   <- ref.get.assertEquals(expectedAttempts)
+    } yield ()
+
+  private def assertSuccess(file: FileState) = assertProcess(file, 1)
+
+  private def assertSkipped(file: FileState) = assertProcess(file, 0)
+
+  test("A valid file is successfully processed.") {
+    assertSuccess(validFile)
+  }
+
+  test("A deprecated file is skipped.") {
+    val deprecated = validFile.copy(deprecated = true)
+    assertSkipped(deprecated)
+  }
+
+  test("A local file is skipped.") {
+    val localFile = validFile.copy(storageType = StorageType.DiskStorage)
+    assertSkipped(localFile)
+  }
+
+  test("A file with a computed digest is skipped.") {
+    val newAttributes    = attributes.copy(digest = ComputedDigest(DigestAlgorithm.default, "something"))
+    val alreadyProcessed = validFile.copy(attributes = newAttributes)
+    assertSkipped(alreadyProcessed)
+  }
+
+  test("A file is processed again when the digest is not yet computed.") {
+    for {
+      ref             <- Ref.of[IO, Int](0)
+      updateAttributes = ref.updateAndGet(_ + 1).flatMap { attempt => IO.raiseWhen(attempt < 2)(DigestNotComputed(id)) }
+      _               <- updateStream(updateAttributes).processFile(validFile)
+      _               <- ref.get.assertEquals(2)
+    } yield ()
+  }
+
+}
diff --git a/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FilesStmSpec.scala b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FilesStmSpec.scala
index 22f234b1e0..60222fb2f3 100644
--- a/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FilesStmSpec.scala
+++ b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/files/FilesStmSpec.scala
@@ -165,13 +165,20 @@ class FilesStmSpec extends CatsEffectSpec with FileFixtures with StorageFixtures
           RevisionNotFound(provided = 3, current = 1)
       }
 
-      "reject with DigestNotComputed" in {
+      "reject with DigestNotComputed with an update command" in {
         val current = FileGen.state(id, projectRef, storageRef, attributes.copy(digest = NotComputedDigest))
         val cmd     = UpdateFile(id, projectRef, storageRef, DiskStorageType, attributes, 1, alice, None)
         evaluate(clock)(Some(current), cmd).rejected shouldEqual DigestNotComputed(id)
       }
 
-      "reject with DigestAlreadyComputed" in {
+      "reject with DigestNotComputed with an update attributes command" in {
+        val updateAttrCmd = UpdateFileAttributes(id, projectRef, mediaType, 10, NotComputedDigest, 1, alice)
+        val current       = FileGen.state(id, projectRef, remoteStorageRef, attributes.copy(bytes = 1))
+
+        evaluate(clock)(Some(current), updateAttrCmd).rejected shouldEqual DigestAlreadyComputed(id)
+      }
+
+      "reject with DigestAlreadyComputed with an update attributes command" in {
         val updateAttrCmd = UpdateFileAttributes(id, projectRef, mediaType, 10, dig, 1, alice)
         val current       = FileGen.state(id, projectRef, remoteStorageRef, attributes.copy(bytes = 1))
 
diff --git a/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/StorageFixtures.scala b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/StorageFixtures.scala
index 6f7bd6e87e..d395fcc1f7 100644
--- a/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/StorageFixtures.scala
+++ b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/StorageFixtures.scala
@@ -1,6 +1,6 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.storage.storages
 
-import ch.epfl.bluebrain.nexus.delta.kernel.{RetryStrategyConfig, Secret}
+import ch.epfl.bluebrain.nexus.delta.kernel.Secret
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StoragesConfig.{DiskStorageConfig, RemoteDiskStorageConfig, S3StorageConfig, StorageTypeConfig}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.StorageFields.{DiskStorageFields, RemoteDiskStorageFields, S3StorageFields}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.{AbsolutePath, DigestAlgorithm}
@@ -16,6 +16,7 @@ import ch.epfl.bluebrain.nexus.testkit.scalatest.ClasspathResources
 import software.amazon.awssdk.regions.Region
 
 import java.nio.file.{Files, Paths}
+import scala.concurrent.duration._
 
 @SuppressWarnings(Array("OptionGet"))
 trait StorageFixtures extends CirceLiteral {
@@ -33,7 +34,7 @@ trait StorageFixtures extends CirceLiteral {
   implicit val config: StorageTypeConfig = StorageTypeConfig(
     disk = DiskStorageConfig(diskVolume, Set(diskVolume,tmpVolume), DigestAlgorithm.default, permissions.read, permissions.write, showLocation = false, Some(5000), 50),
     amazon = Some(S3StorageConfig(DigestAlgorithm.default, Some("localhost"), Some(Secret(MinioDocker.RootUser)), Some(Secret(MinioDocker.RootPassword)), permissions.read, permissions.write, showLocation = false, 60)),
-    remoteDisk = Some(RemoteDiskStorageConfig(DigestAlgorithm.default, BaseUri("http://localhost", Label.unsafe("v1")), Anonymous, permissions.read, permissions.write, showLocation = false, 70, RetryStrategyConfig.AlwaysGiveUp)),
+    remoteDisk = Some(RemoteDiskStorageConfig(DigestAlgorithm.default, BaseUri("http://localhost", Label.unsafe("v1")), Anonymous, permissions.read, permissions.write, showLocation = false, 70, 50.millis)),
   )
   implicit val showLocation: StoragesConfig.ShowFileLocation = config.showFileLocation
   val diskFields        = DiskStorageFields(Some("diskName"), Some("diskDescription"), default = true, Some(tmpVolume), Some(Permission.unsafe("disk/read")), Some(Permission.unsafe("disk/write")), Some(1000), Some(50))