CORE-18532: Retry Transient Errors in Flow Engine

Instead of automatically publishing the original event back to the "flow.event" topic whenever a transient exception occurs when executing the flow pipeline, automatically retry the exceptions using an exponential backoff retry mechanism and permanently fail the flow if the configured time or attemps is reached. - Create internal retry utility to manage retries in an automated fashion with plugable backoff strategy (constant, linear and exponential provided out of the box). - Replace existing utilities in the crypto components and tests with the new one and delete unused classes. - Update FlowEventProcessor to automatically retry transient exceptions through the new retry utility using an exponential backoff with a growth factor of 250ms. - Remove unnecessary code and update tests to accommodate for the new internal retry.
corda · Dec 29, 2023 · 19bd760 · 19bd760
1 parent 9a8b705
commit 19bd760
Show file tree

Hide file tree

Showing 29 changed files with 582 additions and 527 deletions.
diff --git a/...o-service-impl/src/main/kotlin/net/corda/crypto/service/impl/bus/CryptoOpsBusProcessor.kt b/...o-service-impl/src/main/kotlin/net/corda/crypto/service/impl/bus/CryptoOpsBusProcessor.kt
@@ -10,7 +10,7 @@ import net.corda.crypto.core.KeyAlreadyExistsException
 import net.corda.crypto.core.KeyOrderBy
 import net.corda.crypto.core.SecureHashImpl
 import net.corda.crypto.core.ShortHash
-import net.corda.crypto.impl.retrying.BackoffStrategy
+import net.corda.crypto.impl.retrying.CryptoBackoffStrategy
 import net.corda.crypto.impl.retrying.CryptoRetryingExecutor
 import net.corda.crypto.impl.toMap
 import net.corda.crypto.impl.toSignatureSpec
@@ -68,7 +68,7 @@ class CryptoOpsBusProcessor(
 
     private val executor = CryptoRetryingExecutor(
         logger,
-        BackoffStrategy.createBackoff(config.maxAttempts, config.waitBetweenMills)
+        CryptoBackoffStrategy(config.maxAttempts, config.waitBetweenMills)
     )
 
     override fun onNext(request: RpcOpsRequest, respFuture: CompletableFuture<RpcOpsResponse>) {

diff --git a/...ice-impl/src/main/kotlin/net/corda/crypto/service/impl/bus/HSMRegistrationBusProcessor.kt b/...ice-impl/src/main/kotlin/net/corda/crypto/service/impl/bus/HSMRegistrationBusProcessor.kt
@@ -2,7 +2,7 @@ package net.corda.crypto.service.impl.bus
 
 import net.corda.crypto.config.impl.RetryingConfig
 import net.corda.crypto.core.CryptoService
-import net.corda.crypto.impl.retrying.BackoffStrategy
+import net.corda.crypto.impl.retrying.CryptoBackoffStrategy
 import net.corda.crypto.impl.retrying.CryptoRetryingExecutor
 import net.corda.crypto.softhsm.TenantInfoService
 import net.corda.data.crypto.wire.CryptoNoContentValue
@@ -31,7 +31,7 @@ class HSMRegistrationBusProcessor(
 
     private val executor = CryptoRetryingExecutor(
         logger,
-        BackoffStrategy.createBackoff(config.maxAttempts, config.waitBetweenMills)
+        CryptoBackoffStrategy(config.maxAttempts, config.waitBetweenMills)
     )
 
     override fun onNext(request: HSMRegistrationRequest, respFuture: CompletableFuture<HSMRegistrationResponse>) {

diff --git a/...-service-impl/src/main/kotlin/net/corda/crypto/service/impl/rpc/CryptoFlowOpsProcessor.kt b/...-service-impl/src/main/kotlin/net/corda/crypto/service/impl/rpc/CryptoFlowOpsProcessor.kt
@@ -6,7 +6,7 @@ import net.corda.crypto.core.CryptoService
 import net.corda.crypto.core.SecureHashImpl
 import net.corda.crypto.core.ShortHash
 import net.corda.crypto.core.publicKeyIdFromBytes
-import net.corda.crypto.impl.retrying.BackoffStrategy
+import net.corda.crypto.impl.retrying.CryptoBackoffStrategy
 import net.corda.crypto.impl.retrying.CryptoRetryingExecutor
 import net.corda.crypto.impl.toMap
 import net.corda.crypto.impl.toSignatureSpec
@@ -53,7 +53,7 @@ class CryptoFlowOpsProcessor(
 
     private val executor = CryptoRetryingExecutor(
         logger,
-        BackoffStrategy.createBackoff(config.maxAttempts, config.waitBetweenMills)
+        CryptoBackoffStrategy(config.maxAttempts, config.waitBetweenMills)
     )
 
     override fun process(request: FlowOpsRequest): FlowEvent {

diff --git a/...vice-impl/src/main/kotlin/net/corda/crypto/service/impl/rpc/SessionDecryptionProcessor.kt b/...vice-impl/src/main/kotlin/net/corda/crypto/service/impl/rpc/SessionDecryptionProcessor.kt
@@ -3,7 +3,7 @@ package net.corda.crypto.service.impl.rpc
 import net.corda.crypto.config.impl.RetryingConfig
 import net.corda.crypto.core.CryptoService
 import net.corda.crypto.core.CryptoTenants
-import net.corda.crypto.impl.retrying.BackoffStrategy
+import net.corda.crypto.impl.retrying.CryptoBackoffStrategy
 import net.corda.crypto.impl.retrying.CryptoRetryingExecutor
 import net.corda.data.ExceptionEnvelope
 import net.corda.data.crypto.wire.ops.encryption.request.DecryptRpcCommand
@@ -25,7 +25,7 @@ class SessionDecryptionProcessor(
 
     private val executor = CryptoRetryingExecutor(
         logger,
-        BackoffStrategy.createBackoff(config.maxAttempts, config.waitBetweenMills)
+        CryptoBackoffStrategy(config.maxAttempts, config.waitBetweenMills)
     )
 
     override val requestClass = DecryptRpcCommand::class.java

diff --git a/...vice-impl/src/main/kotlin/net/corda/crypto/service/impl/rpc/SessionEncryptionProcessor.kt b/...vice-impl/src/main/kotlin/net/corda/crypto/service/impl/rpc/SessionEncryptionProcessor.kt
@@ -3,7 +3,7 @@ package net.corda.crypto.service.impl.rpc
 import net.corda.crypto.config.impl.RetryingConfig
 import net.corda.crypto.core.CryptoService
 import net.corda.crypto.core.CryptoTenants
-import net.corda.crypto.impl.retrying.BackoffStrategy
+import net.corda.crypto.impl.retrying.CryptoBackoffStrategy
 import net.corda.crypto.impl.retrying.CryptoRetryingExecutor
 import net.corda.data.ExceptionEnvelope
 import net.corda.data.crypto.wire.ops.encryption.request.EncryptRpcCommand
@@ -25,7 +25,7 @@ class SessionEncryptionProcessor(
 
     private val executor = CryptoRetryingExecutor(
         logger,
-        BackoffStrategy.createBackoff(config.maxAttempts, config.waitBetweenMills)
+        CryptoBackoffStrategy(config.maxAttempts, config.waitBetweenMills)
     )
 
     override val requestClass = EncryptRpcCommand::class.java

diff --git a/...rvice/src/integrationTest/kotlin/net/corda/flow/testing/context/FlowServiceTestContext.kt b/...rvice/src/integrationTest/kotlin/net/corda/flow/testing/context/FlowServiceTestContext.kt
@@ -108,6 +108,7 @@ class FlowServiceTestContext @Activate constructor(
 
     private val testConfig = mutableMapOf<String, Any>(
         FlowConfig.EXTERNAL_EVENT_MAX_RETRIES to 2,
+        FlowConfig.PROCESSING_MAX_RETRY_ATTEMPTS to 5,
         FlowConfig.EXTERNAL_EVENT_MESSAGE_RESEND_WINDOW to 500000,
         FlowConfig.SESSION_TIMEOUT_WINDOW to 500000,
         FlowConfig.SESSION_FLOW_CLEANUP_TIME to 30000,

diff --git a/...ice/src/integrationTest/kotlin/net/corda/flow/testing/tests/FlowFinishedAcceptanceTest.kt b/...ice/src/integrationTest/kotlin/net/corda/flow/testing/tests/FlowFinishedAcceptanceTest.kt
@@ -3,7 +3,6 @@ package net.corda.flow.testing.tests
 import net.corda.data.flow.output.FlowStates
 import net.corda.flow.testing.context.ALICE_FLOW_KEY_MAPPER
 import net.corda.flow.testing.context.BOB_FLOW_KEY_MAPPER
-import net.corda.flow.testing.context.CHARLIE_FLOW_KEY_MAPPER
 import net.corda.flow.testing.context.FlowServiceTestBase
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
@@ -53,39 +52,6 @@ class FlowFinishedAcceptanceTest : FlowServiceTestBase() {
         }
     }
 
-    @Test
-    fun `A flow finishing when previously in a retry state publishes a completed flow status and schedules flow cleanup`() {
-        // Trigger a retry state
-        `when` {
-            startFlowEventReceived(FLOW_ID1, REQUEST_ID1, CHARLIE_HOLDING_IDENTITY, CPI1, "flow start data")
-        }
-
-        then {
-            expectOutputForFlow(FLOW_ID1) {
-                checkpointHasRetry(1)
-            }
-        }
-
-        // Now add the missing vnode and trigger a flow completion
-        given {
-            virtualNode(CPI1, CHARLIE_HOLDING_IDENTITY)
-            membershipGroupFor(CHARLIE_HOLDING_IDENTITY)
-        }
-
-        `when` {
-            startFlowEventReceived(FLOW_ID1, REQUEST_ID1, CHARLIE_HOLDING_IDENTITY, CPI1, "flow start data")
-                .completedSuccessfullyWith(DONE)
-        }
-
-        then {
-            expectOutputForFlow(FLOW_ID1) {
-                nullStateRecord()
-                flowStatus(FlowStates.COMPLETED, result = DONE)
-                scheduleFlowMapperCleanupEvents(CHARLIE_FLOW_KEY_MAPPER)
-            }
-        }
-    }
-
     @Test
     fun `A flow finishing with FlowFinished removes fiber from fiber cache`() {
         `when` {
@@ -118,4 +84,4 @@ class FlowFinishedAcceptanceTest : FlowServiceTestBase() {
             }
         }
     }
-}
+}
diff --git a/...low/flow-service/src/integrationTest/kotlin/net/corda/flow/testing/tests/StartFlowTest.kt b/...low/flow-service/src/integrationTest/kotlin/net/corda/flow/testing/tests/StartFlowTest.kt
@@ -11,14 +11,15 @@ import org.junit.jupiter.api.extension.ExtendWith
 import org.junit.jupiter.api.parallel.Execution
 import org.junit.jupiter.api.parallel.ExecutionMode
 import org.osgi.test.junit5.service.ServiceExtension
+import java.util.concurrent.CountDownLatch
+import kotlin.concurrent.thread
 
 @ExtendWith(ServiceExtension::class)
 @Execution(ExecutionMode.SAME_THREAD)
 class StartFlowTest : FlowServiceTestBase() {
 
     @Test
-    fun `RPC Start Flow - Flow starts and updates stats to running`() {
-
+    fun `Flow starts and runs to completion`() {
         given {
             virtualNode(CPI1, BOB_HOLDING_IDENTITY)
             cpkMetadata(CPI1, CPK1, CPK1_CHECKSUM)
@@ -41,14 +42,12 @@ class StartFlowTest : FlowServiceTestBase() {
         }
     }
 
-
     /**
      * When a virtual node has an INACTIVE StartFlowOperationalStatus, it should throw a FlowMarkedForKillException and
      * have a Killed status.
      */
     @Test
     fun `Flow is marked as killed if startFlowOperationalStatus of vNode is INACTIVE`() {
-
         given {
             virtualNode(CPI1, CHARLIE_HOLDING_IDENTITY, flowStartOperationalStatus = OperationalStatus.INACTIVE)
             cpkMetadata(CPI1, CPK1, CPK1_CHECKSUM)
@@ -67,98 +66,82 @@ class StartFlowTest : FlowServiceTestBase() {
                 flowStatus(
                     state = FlowStates.KILLED,
                     flowTerminatedReason = "flowStartOperationalStatus is INACTIVE, new flows cannot be started for " +
-                            "virtual node with shortHash ${CHARLIE_HOLDING_IDENTITY.toCorda().shortHash}"
+                        "virtual node with shortHash ${CHARLIE_HOLDING_IDENTITY.toCorda().shortHash}"
                 )
             }
         }
     }
 
     /**
-     * When a flow event fails with a transient exception then the flow will be put into a retry
-     * state. In this case the flow engine will publish the problematic event back to itself to be processed at some
-     * later time. This could then fail again, triggering the same loop.
+     * When a flow event fails with an internal transient exception, the flow engine will automatically retry the failed
+     * operation without publishing any extra messages to Kafka. Instead, it will try to re-process the event at a later
+     * time using exponential backoff. This could then fail again, triggering the same loop until a threshold is reached.
      *
-     * Scenario 1 - Fails multiple times but completes before the retry limit
-     * Scenario 2 - Fails multiple times and hits the retry limit failing the flow to the DLQ
+     * Scenario 1 - Fails multiple times, transient error is fixed before retry limit => flow completes.
+     * Scenario 2 - Fails multiple times, transient error persists, hits the retry limit => flow fails and DLQ event.
      */
     @Test
-    fun `RPC Start Flow - Retry scenario 1 - Fail then succeeds`() {
+    fun `Retry scenario 1 - Transient exception is fixed within the retry limit and flow succeeds`() {
         given {
             cpkMetadata(CPI1, CPK1, CPK1_CHECKSUM)
             sandboxCpk(CPK1_CHECKSUM)
             membershipGroupFor(BOB_HOLDING_IDENTITY)
         }
 
-        `when` {
-            startFlowEventReceived(FLOW_ID1, REQUEST_ID1, BOB_HOLDING_IDENTITY, CPI1, "flow start data")
-        }
-
-        then {
-            expectOutputForFlow(FLOW_ID1) {
-                checkpointHasRetry(1)
-                flowFiberCacheDoesNotContainKey(BOB_HOLDING_IDENTITY, REQUEST_ID1)
+        val latch = CountDownLatch(1)
+        val startFlow = thread {
+            latch.countDown()
+            `when` {
+                startFlowEventReceived(FLOW_ID1, REQUEST_ID1, BOB_HOLDING_IDENTITY, CPI1, "flow start data")
+                    .suspendsWith(FlowIORequest.InitialCheckpoint)
+                    .completedSuccessfullyWith("hello")
             }
         }
 
-        // Now fix the issue and expect the wake-up event to
-        // retry the start flow successfully
-        given {
-            virtualNode(CPI1, BOB_HOLDING_IDENTITY)
-        }
-
-        `when` {
-            startFlowEventReceived(FLOW_ID1, REQUEST_ID1, BOB_HOLDING_IDENTITY, CPI1, "flow start data")
-                .completedSuccessfullyWith("hello")
-        }
+        latch.await()
+        Thread.sleep(1000)
+        testContext.virtualNode(CPI1, BOB_HOLDING_IDENTITY)
+        startFlow.join()
 
         then {
             expectOutputForFlow(FLOW_ID1) {
-                checkpointDoesNotHaveRetry()
+                nullStateRecord()
                 flowStatus(FlowStates.COMPLETED, result = "hello")
                 flowFiberCacheDoesNotContainKey(BOB_HOLDING_IDENTITY, REQUEST_ID1)
             }
         }
     }
 
     @Test
-    fun `RPC Start Flow - Retry scenario 2 - Hit the retry limit and fail the flow`() {
+    fun `Retry scenario 2 - Transient exception is not fixed within the retry limit and flow permanently fails`() {
         given {
             cpkMetadata(CPI1, CPK1, CPK1_CHECKSUM)
             sandboxCpk(CPK1_CHECKSUM)
             membershipGroupFor(BOB_HOLDING_IDENTITY)
-            flowConfiguration(FlowConfig.PROCESSING_MAX_RETRY_WINDOW_DURATION, 0)
+            flowConfiguration(FlowConfig.PROCESSING_MAX_RETRY_ATTEMPTS, 1)
         }
 
         `when` {
             startFlowEventReceived(FLOW_ID1, REQUEST_ID1, BOB_HOLDING_IDENTITY, CPI1, "flow start data")
         }
 
-        then {
-            expectOutputForFlow(FLOW_ID1) {
-                checkpointHasRetry(1)
-                flowFiberCacheDoesNotContainKey(BOB_HOLDING_IDENTITY, REQUEST_ID1)
-            }
-        }
-
-        `when` {
-            startFlowEventReceived(FLOW_ID1, REQUEST_ID1, BOB_HOLDING_IDENTITY, CPI1, "flow start data")
-        }
-
         then {
             expectOutputForFlow(FLOW_ID1) {
                 nullStateRecord()
                 markedForDlq()
                 noFlowEvents()
                 flowFiberCacheDoesNotContainKey(BOB_HOLDING_IDENTITY, REQUEST_ID1)
-                //we can't return a status record after the change to checkpoint initialization
+                // we can't return a status record after the change to checkpoint initialization
                 // Story to deal with change in status records -> CORE-10571: Re-design how status record is published
-/*                flowStatus(
+                /*
+                flowStatus(
                     state = FlowStates.FAILED,
                     errorType = FlowProcessingExceptionTypes.FLOW_FAILED,
                     errorMessage = "Execution failed with \"Failed to find the virtual node info for holder " +
-                            "'HoldingIdentity(x500Name=${BOB_HOLDING_IDENTITY.x500Name}, groupId=${BOB_HOLDING_IDENTITY.groupId})'\" " +
-                            "after 1 retry attempts."
-                )*/
+                        "'HoldingIdentity(x500Name=${BOB_HOLDING_IDENTITY.x500Name}, groupId=${BOB_HOLDING_IDENTITY.groupId})'\" " +
+                        "after 1 retry attempts."
+                )
+                */
             }
         }
     }

diff --git a/.../flow/flow-service/src/main/kotlin/net/corda/flow/pipeline/FlowEventExceptionProcessor.kt b/.../flow/flow-service/src/main/kotlin/net/corda/flow/pipeline/FlowEventExceptionProcessor.kt
@@ -6,7 +6,6 @@ import net.corda.flow.pipeline.exceptions.FlowEventException
 import net.corda.flow.pipeline.exceptions.FlowFatalException
 import net.corda.flow.pipeline.exceptions.FlowMarkedForKillException
 import net.corda.flow.pipeline.exceptions.FlowPlatformException
-import net.corda.flow.pipeline.exceptions.FlowTransientException
 import net.corda.libs.configuration.SmartConfig
 
 /**
@@ -34,18 +33,6 @@ interface FlowEventExceptionProcessor {
      */
     fun process(throwable: Throwable, context: FlowEventContext<*>): FlowEventContext<*>
 
-    /**
-     * Processes a [FlowTransientException] and provides the pipeline response.
-     *
-     * Used to handle event retries.
-     *
-     * @param exception The [FlowTransientException] thrown during processing
-     * @param context The [FlowEventContext] at the point of failure.
-     *
-     * @return The updated context.
-     */
-    fun process(exception: FlowTransientException, context: FlowEventContext<*>): FlowEventContext<*>
-
     /**
      * Processes a [FlowFatalException] and provides the pipeline response.
      *

diff --git a/...flow-service/src/main/kotlin/net/corda/flow/pipeline/exceptions/FlowTransientException.kt b/...flow-service/src/main/kotlin/net/corda/flow/pipeline/exceptions/FlowTransientException.kt
@@ -3,9 +3,8 @@ package net.corda.flow.pipeline.exceptions
 import net.corda.v5.base.exceptions.CordaRuntimeException
 
 /**
- * The [FlowTransientException] is thrown for a recoverable error, this exception will cause the event processing
- * to be retried
+ * The [FlowTransientException] should only be thrown for transient errors that affect the flow engine and any of its
+ * dependant services, it will cause the pipeline execution to be automatically retried at source.
  */
 class FlowTransientException(override val message: String, cause: Throwable? = null) :
     CordaRuntimeException(message, cause)
-