Fix integer overflow during passivation while computing its duration + use hours

Author: Simon Dumas

delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/query/RefreshOrStop.scala

@@ -13,6 +13,7 @@ import fs2.Stream
 import fs2.concurrent.SignallingRef
 
 import concurrent.duration._
+import scala.util.Try
 
 /**
   * Computes the outcome to apply when all elements are consumed by a projection
@@ -75,9 +76,20 @@ object RefreshOrStop {
   private def passivate(project: ProjectRef, signal: SignallingRef[IO, Boolean]) =
     for {
       _           <- logger.info(s"Project '$project' is inactive, pausing until some activity is seen again.")
-      durationOpt <- Stream.awakeEvery[IO](1.second).scanMonoid.interruptWhen(signal).compile.last
-      minutes      = durationOpt.getOrElse(0.minute).toMinutes
-      _           <- logger.info(s"Project '$project' is active again after `$minutes minutes`, querying will resume.")
+      durationOpt <- Stream
+                       .awakeEvery[IO](1.second)
+                       .fold(Option(Duration.Zero)) { case (accOpt, duration) =>
+                         accOpt.flatMap(safeAdd(_, duration))
+                       }
+                       .interruptWhen(signal)
+                       .compile
+                       .last
+                       .map(_.flatten)
+      hours        = durationOpt.getOrElse(0.hour).toHours
+      _           <- logger.info(s"Project '$project' is active again after `$hours hours`, querying will resume.")
     } yield Passivated
 
+  private[query] def safeAdd(d1: FiniteDuration, d2: FiniteDuration) =
+    Try { d1 + d2 }.toOption
+
 }

delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/query/RefreshOrStopSuite.scala

@@ -9,7 +9,8 @@ import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
 import ch.epfl.bluebrain.nexus.testkit.mu.ce.PatienceConfig
 import fs2.concurrent.SignallingRef
 
-import scala.concurrent.duration.DurationInt
+import java.util.concurrent.TimeUnit
+import scala.concurrent.duration.{DurationInt, FiniteDuration}
 
 class RefreshOrStopSuite extends NexusSuite {
 
@@ -82,4 +83,17 @@ class RefreshOrStopSuite extends NexusSuite {
     } yield ()
   }
 
+  test("Safely add two durations and return the result") {
+    val d1       = FiniteDuration(42L, TimeUnit.NANOSECONDS)
+    val d2       = FiniteDuration(8L, TimeUnit.NANOSECONDS)
+    val expected = FiniteDuration(50L, TimeUnit.NANOSECONDS)
+    assertEquals(RefreshOrStop.safeAdd(d1, d2), Some(expected))
+  }
+
+  test("Trigger an overflow and handle the error") {
+    val d1 = FiniteDuration(Long.MaxValue - 5L, TimeUnit.NANOSECONDS)
+    val d2 = FiniteDuration(8L, TimeUnit.NANOSECONDS)
+    assertEquals(RefreshOrStop.safeAdd(d1, d2), None)
+  }
+
 }