Fix BFT refresh retry loop (#1162)

---------

Signed-off-by: Oriol Muñoz <oriol.munoz@digitalasset.com>

Author: OriolMunoz-da

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala

@@ -884,75 +884,88 @@ object BftScanConnection {
     )(implicit tc: TraceContext): Future[ScanConnections] = {
       val currentState @ BftState(currentScanConnections, currentFailed) =
         currentScanConnectionsRef.get()
-      val currentScans = (currentScanConnections.keys ++ currentFailed.keys).toSet
       logger.info(s"Started refreshing scan list from $currentState")
-      getScans(connection).flatMap { scansInDsoRules =>
-        val newScans = scansInDsoRules.filter(scan => !currentScans.contains(scan.publicUrl))
-        val removedScans = currentScans.filter(url => !scansInDsoRules.exists(_.publicUrl == url))
-        if (scansInDsoRules.isEmpty) {
-          // This is expected on app init, and is retried when building the BftScanConnection
-          Future.failed(
-            io.grpc.Status.FAILED_PRECONDITION
-              .withDescription(
-                s"Scan list in DsoRules is empty. Last known list: $currentState"
-              )
-              .asRuntimeException()
-          )
-        } else if (newScans.isEmpty && removedScans.isEmpty && currentFailed.isEmpty) {
-          logger.debug("Not updating scan list as there are no changes.")
-          Future.successful(currentState.scanConnections)
-        } else {
-          for {
-            (newScansFailedConnections, newScansSuccessfulConnections) <- attemptConnections(
-              newScans
+
+      for {
+        // if the previous state had too many failed scans, we cannot fetch the new list of scans.
+        // thus, we retry all failed connections first.
+        (retriedScansFailedConnections, retriedScansSuccessfulConnections) <- attemptConnections(
+          currentFailed.map { case (uri, (_, svName)) =>
+            DsoScan(uri, svName)
+          }.toSeq
+        )
+        retriedCurrentState = BftState(
+          currentScanConnections ++ retriedScansSuccessfulConnections,
+          retriedScansFailedConnections.toMap,
+        )
+        // this state will be used to fetch the scans in the next call
+        _ = currentScanConnectionsRef.set(retriedCurrentState)
+        // these will be BFT-read, failing if there's no consensus
+        scansInDsoRules <- getScans(connection)
+        newState <- computeNewState(retriedCurrentState, scansInDsoRules)
+      } yield {
+        currentScanConnectionsRef.set(newState)
+        logger.info(s"Updated scan list to $newState")
+
+        val connections = newState.scanConnections
+        val defaultCallConfig = BftCallConfig.default(connections)
+        // Most but not all calls will use the default config.
+        // Fail early if there are not enough Scans for the default config
+        if (!defaultCallConfig.enoughAvailableScans) {
+          throw io.grpc.Status.FAILED_PRECONDITION
+            .withDescription(
+              s"There are not enough Scans to satisfy f=${connections.f}. Will be retried. State: $newState"
             )
-            toRetry = currentFailed -- removedScans
-            (retriedScansFailedConnections, retriedScansSuccessfulConnections) <-
-              attemptConnections(
-                toRetry.map { case (url, (_, svName)) => DsoScan(url, svName) }.toSeq
-              )
-          } yield {
-            removedScans.foreach { url =>
-              currentScanConnections.get(url).foreach { case (connection, svName) =>
-                logger.info(
-                  s"Closing connection to scan of $svName ($url) as it's been removed from the DsoRules scan list."
-                )
-                attemptToClose(connection)
-              }
-            }
-            (newScansFailedConnections ++ retriedScansFailedConnections).foreach {
-              case (url, (err, svName)) =>
-                // TODO(#815): abstract this pattern into the RetryProvider
-                if (retryProvider.isClosing)
-                  logger.info(
-                    s"Suppressed warning, as we're shutting down: Failed to connect to scan of $svName ($url).",
-                    err,
-                  )
-                else
-                  logger.warn(s"Failed to connect to scan of $svName ($url).", err)
-            }
+            .asRuntimeException()
+        } else {
+          connections
+        }
+      }
+    }
+
+    private def computeNewState(
+        currentState: BftState,
+        scansInDsoRules: Seq[DsoScan],
+    )(implicit tc: TraceContext): Future[BftState] = {
+      val BftState(currentScanConnections, currentFailed) = currentState
+      val currentScans = (currentScanConnections.keys ++ currentFailed.keys).toSet
 
-            val newState = BftState(
-              currentScanConnections -- removedScans ++ newScansSuccessfulConnections ++ retriedScansSuccessfulConnections,
-              (retriedScansFailedConnections ++ newScansFailedConnections).toMap,
+      val newScans = scansInDsoRules.filter(scan => !currentScans.contains(scan.publicUrl))
+      val removedScans = currentScans.filter(url => !scansInDsoRules.exists(_.publicUrl == url))
+      if (scansInDsoRules.isEmpty) {
+        // This is expected on app init, and is retried when building the BftScanConnection
+        Future.failed(
+          io.grpc.Status.FAILED_PRECONDITION
+            .withDescription(
+              s"Scan list in DsoRules is empty. Last known list: $currentState"
             )
-            currentScanConnectionsRef.set(newState)
-            logger.info(s"Updated scan list to $newState")
-
-            val connections = newState.scanConnections
-            val defaultCallConfig = BftCallConfig.default(connections)
-            // Most but not all calls will use the default config.
-            // Fail early if there are not enough Scans for the default config
-            if (!defaultCallConfig.enoughAvailableScans) {
-              throw io.grpc.Status.FAILED_PRECONDITION
-                .withDescription(
-                  s"There are not enough Scans to satisfy f=${connections.f}. Will be retried. State: $newState"
-                )
-                .asRuntimeException()
-            } else {
-              connections
+            .asRuntimeException()
+        )
+      } else {
+        for {
+          (newScansFailedConnections, newScansSuccessfulConnections) <- attemptConnections(
+            newScans
+          )
+        } yield {
+          logger.info(
+            s"New successful scans: ${newScansSuccessfulConnections.map(_._1)}, " +
+              s"new failed scans: ${newScansFailedConnections.map(_._1)}, " +
+              s"removed scans: $removedScans"
+          )
+
+          removedScans.foreach { url =>
+            currentScanConnections.get(url).foreach { case (connection, svName) =>
+              logger.info(
+                s"Closing connection to scan of $svName ($url) as it's been removed from the DsoRules scan list."
+              )
+              attemptToClose(connection)
             }
           }
+
+          BftState(
+            (currentScanConnections -- removedScans) ++ newScansSuccessfulConnections,
+            (currentFailed -- removedScans) ++ newScansFailedConnections,
+          )
         }
       }
     }
@@ -969,12 +982,27 @@ object BftScanConnection {
         .traverse { scan =>
           logger.info(s"Attempting to connect to Scan: $scan.")
           connectionBuilder(scan.publicUrl)
-            .transformWith(result =>
+            .transformWith { result =>
+              // logging
+              result.failed.foreach { err =>
+                // TODO(#815): abstract this pattern into the RetryProvider
+                if (retryProvider.isClosing)
+                  logger.info(
+                    s"Suppressed warning, as we're shutting down: Failed to connect to scan of ${scan.svName} (${scan.publicUrl}).",
+                    err,
+                  )
+                else
+                  logger.warn(
+                    s"Failed to connect to scan of ${scan.svName} (${scan.publicUrl}).",
+                    err,
+                  )
+              }
+              // actual result
               Future.successful(
                 result.toEither
                   .bimap(scan.publicUrl -> (_, scan.svName), scan.publicUrl -> (_, scan.svName))
               )
-            )
+            }
         }
         .map(_.partitionEither(identity))
     }

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnectionTest.scala

@@ -15,6 +15,7 @@ import com.google.protobuf.ByteString
 import org.apache.pekko.http.scaladsl.model.*
 import org.lfdecentralizedtrust.splice.admin.http.HttpErrorWithHttpCode
 import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules as amuletrulesCodegen
+import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.AmuletRules
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.{
   holdingv1,
   metadatav1,
@@ -49,6 +50,7 @@ import org.scalatest.wordspec.AsyncWordSpec
 import org.slf4j.event.Level
 
 import java.time.{Duration, Instant}
+import java.util.concurrent.atomic.AtomicInteger
 import scala.concurrent.{ExecutionContext, Future}
 
 // mock verification triggers this
@@ -76,7 +78,11 @@ class BftScanConnectionTest
     }
     connections.foreach { connection =>
       // all of this is noise...
-      when(connection.getAmuletRulesWithState(eqTo(None))(any[ExecutionContext], any[TraceContext]))
+      when(
+        connection.getAmuletRulesWithState(
+          any[Option[ContractWithState[AmuletRules.ContractId, AmuletRules]]]
+        )(any[ExecutionContext], any[TraceContext])
+      )
         .thenReturn(
           Future.successful(
             ContractWithState(
@@ -184,6 +190,7 @@ class BftScanConnectionTest
       synchronizerId = synchronizerId,
     )
   }
+  val refreshSeconds = 10000L
   def getBft(
       initialConnections: Seq[SingleScanConnection],
       connectionBuilder: Uri => Future[SingleScanConnection] = _ =>
@@ -198,7 +205,7 @@ class BftScanConnectionTest
         initialFailedConnections,
         connectionBuilder,
         Bft.getScansInDsoRules,
-        NonNegativeFiniteDuration.ofSeconds(1),
+        NonNegativeFiniteDuration.ofSeconds(refreshSeconds),
         retryProvider,
         loggerFactory,
       ),
@@ -278,18 +285,85 @@ class BftScanConnectionTest
       connections.foreach(makeMockReturn(_, partyIdA))
 
       // we initialize with just the first one, and the second one will be "built" when we refresh
-      val bft = getBft(connections.take(1), _ => Future.successful(connections(1)))
-      clock.advance(Duration.ofSeconds(2))
+      val refreshCalled = new AtomicInteger(0)
+      val bft = getBft(
+        connections.take(1),
+        _ => {
+          refreshCalled.incrementAndGet()
+          Future.successful(connections(1))
+        },
+      )
+      clock.advance(Duration.ofSeconds(refreshSeconds + 1))
+      // even after advancing it shouldn't refresh yet, as that's less than refreshSeconds
+      clock.advance(Duration.ofSeconds(1))
+      clock.advance(Duration.ofSeconds(1))
+      clock.advance(Duration.ofSeconds(1))
+      clock.advance(Duration.ofSeconds(1))
 
       // eventually the refresh goes through and the second connection is used
       eventually() {
+        refreshCalled.intValue() should be(1)
         val result = bft.getDsoPartyId().futureValue
         try { verify(connections(1), atLeast(1)).getDsoPartyId() }
         catch { case cause: MockitoAssertionError => fail("Mockito fail", cause) }
         result should be(partyIdA)
       }
     }
 
+    "refresh the list of scans faster if there are not enough available scans" in {
+      val connections = getMockedConnections(n = 4)
+      val connectionsMap = connections.map(c => c.config.adminApi.url -> c).toMap
+
+      connections.foreach(makeMockReturn(_, partyIdA))
+      val refreshCalled = connections.map(_.config.adminApi.url -> new AtomicInteger(0)).toMap
+
+      loggerFactory.assertLogsSeq(SuppressionRule.Level(Level.WARN))(
+        {
+          // all failed until retried enough times
+          val bft = getBft(
+            Seq.empty,
+            uri => {
+              val calls = refreshCalled(uri).incrementAndGet()
+              if (calls > 3) {
+                Future.successful(connectionsMap(uri))
+              } else {
+                Future.failed(new RuntimeException("some'rror"))
+              }
+            },
+            initialFailedConnections = connections
+              .map(connection => connection.config.adminApi.url -> new RuntimeException("Failed"))
+              .toMap,
+          )
+          // trigger the first refresh, this is only required in tests, prod code retries already on BftScanConnection init
+          clock.advance(Duration.ofSeconds(refreshSeconds + 1))
+          // and then refresh until it's called enough times
+          eventually() {
+            clock.advance(Duration.ofSeconds(1))
+            forAll(refreshCalled) { case (_, calls) =>
+              calls.intValue() should be >= 3
+            }
+          }
+
+          // eventually the refresh goes through and the second connection is used
+          eventually() {
+            bft.scanList.scanConnections.open should have size 4
+            bft.scanList.scanConnections.failed should be(0)
+            val result = bft.getDsoPartyId().futureValue
+            try { verify(connections(1), atLeast(1)).getDsoPartyId() }
+            catch { case cause: MockitoAssertionError => fail("Mockito fail", cause) }
+            result should be(partyIdA)
+          }
+        },
+        entries =>
+          forAll(entries) { entry =>
+            entry.warningMessage should include("Failed to connect to scan").or(
+              include("which are fewer than the necessary")
+            )
+          },
+      )
+
+    }
+
     "fail if too many Scans failed to connect" in {
       // f = (1ok + 3bad - 1) / 3 = 1
       // 1 Scan is not enough for f+1=2