CORE-20867 Implement retry topic to handle persistent transient RPC Client errors

components/flow/flow-service/src/main/kotlin/net/corda/flow/external/events/impl/ExternalEventManager.kt

@@ -1,13 +1,13 @@
 package net.corda.flow.external.events.impl
 
-import java.time.Instant
 import net.corda.data.flow.event.external.ExternalEvent
 import net.corda.data.flow.event.external.ExternalEventResponse
 import net.corda.data.flow.state.external.ExternalEventState
 import net.corda.flow.external.events.factory.ExternalEventFactory
 import net.corda.flow.external.events.factory.ExternalEventRecord
 import net.corda.messaging.api.records.Record
 import java.time.Duration
+import java.time.Instant
 
 /**
  * [ExternalEventManager] encapsulates external event behaviour by creating and modifying [ExternalEventState]s.
@@ -86,4 +86,16 @@ interface ExternalEventManager {
         instant: Instant,
         retryWindow: Duration
     ): Pair<ExternalEventState, Record<*, *>?>
+
+    /**
+     * Get the external event to send for the transient error retry scenario.
+     * Returns the event as is from the state. No additional checks required.
+     * @param externalEventState The [ExternalEventState] to get the event from.
+     * @param instant The current time. Used to set timestamp.
+     * @return The external event request to resend
+     * */
+    fun getRetryEvent(
+        externalEventState: ExternalEventState,
+        instant: Instant,
+    ): Record<*, *>
 }

components/flow/flow-service/src/main/kotlin/net/corda/flow/external/events/impl/ExternalEventManagerImpl.kt

@@ -174,6 +174,13 @@ class ExternalEventManagerImpl(
         return externalEventState to record
     }
 
+    override fun getRetryEvent(
+        externalEventState: ExternalEventState,
+        instant: Instant,
+    ): Record<*, *> {
+        return generateRecord(externalEventState, instant)
+    }
+
     private fun checkRetry(externalEventState: ExternalEventState, instant: Instant, retryWindow: Duration) {
         when {
             (externalEventState.sendTimestamp + retryWindow) >= instant -> {

components/flow/flow-service/src/main/kotlin/net/corda/flow/messaging/mediator/FlowEventMediatorFactoryImpl.kt

@@ -1,8 +1,10 @@
 package net.corda.flow.messaging.mediator
 
+import com.typesafe.config.ConfigValueFactory
 import net.corda.avro.serialization.CordaAvroSerializationFactory
 import net.corda.data.crypto.wire.ops.flow.FlowOpsRequest
 import net.corda.data.flow.event.FlowEvent
+import net.corda.data.flow.event.external.ExternalEventRetryRequest
 import net.corda.data.flow.event.mapper.FlowMapperEvent
 import net.corda.data.flow.output.FlowStatus
 import net.corda.data.flow.state.checkpoint.Checkpoint
@@ -30,6 +32,7 @@ import net.corda.messaging.api.mediator.RoutingDestination.Companion.routeTo
 import net.corda.messaging.api.mediator.RoutingDestination.Type.ASYNCHRONOUS
 import net.corda.messaging.api.mediator.RoutingDestination.Type.SYNCHRONOUS
 import net.corda.messaging.api.mediator.config.EventMediatorConfigBuilder
+import net.corda.messaging.api.mediator.config.RetryConfig
 import net.corda.messaging.api.mediator.factory.MediatorConsumerFactory
 import net.corda.messaging.api.mediator.factory.MediatorConsumerFactoryFactory
 import net.corda.messaging.api.mediator.factory.MessageRouterFactory
@@ -47,12 +50,14 @@ import net.corda.schema.configuration.BootConfig.TOKEN_SELECTION_WORKER_REST_END
 import net.corda.schema.configuration.BootConfig.UNIQUENESS_WORKER_REST_ENDPOINT
 import net.corda.schema.configuration.BootConfig.VERIFICATION_WORKER_REST_ENDPOINT
 import net.corda.schema.configuration.BootConfig.WORKER_MEDIATOR_REPLICAS_FLOW_SESSION
+import net.corda.schema.configuration.MessagingConfig.Bus.KAFKA_CONSUMER_MAX_POLL_RECORDS
 import net.corda.schema.configuration.MessagingConfig.Subscription.MEDIATOR_PROCESSING_MIN_POOL_RECORD_COUNT
 import net.corda.schema.configuration.MessagingConfig.Subscription.MEDIATOR_PROCESSING_THREAD_POOL_SIZE
 import org.osgi.service.component.annotations.Activate
 import org.osgi.service.component.annotations.Component
 import org.osgi.service.component.annotations.Reference
 import org.slf4j.LoggerFactory
+import java.time.Instant
 import java.util.UUID
 
 @Suppress("LongParameterList")
@@ -77,7 +82,9 @@ class FlowEventMediatorFactoryImpl @Activate constructor(
         private const val CONSUMER_GROUP = "FlowEventConsumer"
         private const val MESSAGE_BUS_CLIENT = "MessageBusClient"
         private const val RPC_CLIENT = "RpcClient"
-
+        private const val RETRY_TOPIC_POLL_LIMIT = 5
+        private const val RETRY_TOPIC = FLOW_EVENT_TOPIC
+        private const val TOKEN_RETRY = "TokenRetry"
         private val logger = LoggerFactory.getLogger(this::class.java.enclosingClass)
     }
 
@@ -123,12 +130,65 @@ class FlowEventMediatorFactoryImpl @Activate constructor(
         .threadName("flow-event-mediator")
         .stateManager(stateManager)
         .minGroupSize(messagingConfig.getInt(MEDIATOR_PROCESSING_MIN_POOL_RECORD_COUNT))
+        .retryConfig(RetryConfig(RETRY_TOPIC, ::buildRetryRequest))
         .build()
 
+
+    /**
+     * Build a request to trigger a resend of external events via the flow event pipeline.
+     * Request id is calculated from the previous request payload when possible to allow for some validation in the pipeline.
+     * This validation is an enhancement and not strictly required.
+     * A new Timestamp is set on each request to ensure each request is unique for replay logic handling.
+     * @param key the key of the input record
+     * @param syncRpcRequest the previous sync request which failed.
+     * @return list of output retry events.
+     */
+    private fun buildRetryRequest(key: String, syncRpcRequest: MediatorMessage<Any>) : List<MediatorMessage<Any>> {
+        return try {
+            val requestId = getRequestId(syncRpcRequest)
+            val externalEventRetryRequest = ExternalEventRetryRequest.newBuilder()
+                .setRequestId(requestId)
+                .setTimestamp(Instant.now())
+                .build()
+            val flowEvent = FlowEvent.newBuilder()
+                .setFlowId(key)
+                .setPayload(externalEventRetryRequest)
+                .build()
+            //ensure key is set correctly on new message destined for flow topic
+            val properties = syncRpcRequest.properties.toMutableMap().apply { this["key"] = key }
+            listOf(MediatorMessage(flowEvent, properties))
+        } catch (ex: Exception) {
+            //In this scenario we failed to build the retry event. This will likely result in the flow hanging until the idle processor
+            // kicks in. This shouldn't be possible and is just a safety net.
+            logger.warn("Failed to generate a retry event for key $key. No retry will be triggered.", ex)
+            emptyList()
+        }
+    }
+
+    /**
+     * Determine the external event request id where possible.
+     * Note, some token events have no request id as there is no response.
+     * For these use a hardcoded request id which will be ignored at the validation step.
+     * @param syncRpcRequest the previous request
+     * @return Request ID to set in the retry event.
+     */
+    private fun getRequestId(syncRpcRequest: MediatorMessage<Any>): String {
+        return when (val entityRequest = deserializer.deserialize(syncRpcRequest.payload as ByteArray)) {
+            is EntityRequest -> entityRequest.flowExternalEventContext.requestId
+            is FlowOpsRequest -> entityRequest.flowExternalEventContext.requestId
+            is LedgerPersistenceRequest -> entityRequest.flowExternalEventContext.requestId
+            is TransactionVerificationRequest -> entityRequest.flowExternalEventContext.requestId
+            is UniquenessCheckRequestAvro -> entityRequest.flowExternalEventContext.requestId
+            is TokenPoolCacheEvent -> TOKEN_RETRY
+            else -> "InvalidEntityType"
+        }
+    }
+
     private fun createMediatorConsumerFactories(messagingConfig: SmartConfig, bootConfig: SmartConfig): List<MediatorConsumerFactory> {
+        val retryTopicMessagingConfig = getRetryTopicConfig(messagingConfig)
         val mediatorConsumerFactory: MutableList<MediatorConsumerFactory> = mutableListOf(
             mediatorConsumerFactory(FLOW_START, messagingConfig),
-            mediatorConsumerFactory(FLOW_EVENT_TOPIC, messagingConfig)
+            mediatorConsumerFactory(FLOW_EVENT_TOPIC, retryTopicMessagingConfig)
         )
 
         val mediatorReplicas = bootConfig.getIntOrDefault(WORKER_MEDIATOR_REPLICAS_FLOW_SESSION, 1)
@@ -140,6 +200,10 @@ class FlowEventMediatorFactoryImpl @Activate constructor(
         return mediatorConsumerFactory
     }
 
+    private fun getRetryTopicConfig(messagingConfig: SmartConfig): SmartConfig {
+        return messagingConfig.withValue(KAFKA_CONSUMER_MAX_POLL_RECORDS, ConfigValueFactory.fromAnyRef(RETRY_TOPIC_POLL_LIMIT))
+    }
+
     private fun mediatorConsumerFactory(
         topic: String,
         messagingConfig: SmartConfig
@@ -178,8 +242,9 @@ class FlowEventMediatorFactoryImpl @Activate constructor(
                     rpcEndpoint(VERIFICATION_WORKER_REST_ENDPOINT, VERIFICATION_PATH), SYNCHRONOUS)
                 is UniquenessCheckRequestAvro -> routeTo(rpcClient,
                     rpcEndpoint(UNIQUENESS_WORKER_REST_ENDPOINT, UNIQUENESS_PATH), SYNCHRONOUS)
+                //RETRIES will appear as FlowEvents
                 is FlowEvent -> routeTo(messageBusClient,
-                    FLOW_EVENT_TOPIC, ASYNCHRONOUS)
+                    RETRY_TOPIC, ASYNCHRONOUS)
                 is String -> routeTo(messageBusClient, // Handling external messaging
                     message.properties[MSG_PROP_TOPIC] as String, ASYNCHRONOUS)
                 else -> {

components/flow/flow-service/src/main/kotlin/net/corda/flow/pipeline/handlers/events/ExternalEventRetryRequestHandler.kt

@@ -0,0 +1,69 @@
+package net.corda.flow.pipeline.handlers.events
+
+import net.corda.data.flow.event.external.ExternalEventResponse
+import net.corda.data.flow.event.external.ExternalEventRetryRequest
+import net.corda.flow.pipeline.events.FlowEventContext
+import net.corda.flow.pipeline.exceptions.FlowEventException
+import net.corda.utilities.debug
+import org.osgi.service.component.annotations.Component
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+/**
+ * Handles pre-processing of events that are intended to trigger a resend of an external event.
+ * This can be triggered by the mediator in the event of transient errors.
+ */
+@Component(service = [FlowEventHandler::class])
+class ExternalEventRetryRequestHandler : FlowEventHandler<ExternalEventRetryRequest> {
+
+    private companion object {
+        val log: Logger = LoggerFactory.getLogger(this::class.java.enclosingClass)
+        private const val TOKEN_RETRY = "TokenRetry"
+    }
+
+    override val type = ExternalEventRetryRequest::class.java
+
+    override fun preProcess(context: FlowEventContext<ExternalEventRetryRequest>): FlowEventContext<ExternalEventRetryRequest> {
+        val checkpoint = context.checkpoint
+        val externalEventRetryRequest = context.inputEventPayload
+
+        if (!checkpoint.doesExist) {
+            log.debug {
+                "Received a ${ExternalEventRetryRequest::class.simpleName} for flow [${context.inputEvent.flowId}] that " +
+                        "does not exist. The event will be discarded. ${ExternalEventRetryRequest::class.simpleName}: " +
+                        externalEventRetryRequest
+            }
+            throw FlowEventException(
+                "ExternalEventRetryRequestHandler received a ${ExternalEventRetryRequest::class.simpleName} for flow" +
+                        " [${context.inputEvent.flowId}] that does not exist"
+            )
+        }
+
+        val externalEventState = checkpoint.externalEventState
+        val retryRequestId: String = externalEventRetryRequest.requestId
+        val externalEventStateRequestId = externalEventState?.requestId
+        if (externalEventState == null) {
+            log.debug {
+                "Received an ${ExternalEventRetryRequest::class.simpleName} with request id: " +
+                        "$retryRequestId while flow [${context.inputEvent.flowId} is not waiting " +
+                        "for an ${ExternalEventResponse::class.simpleName}. " +
+                        "${ExternalEventRetryRequest::class.simpleName}: $externalEventRetryRequest"
+            }
+            throw FlowEventException(
+                "ExternalEventRetryRequestHandler received an ${ExternalEventRetryRequest::class.simpleName} with request id: " +
+                        "$retryRequestId while flow [${context.inputEvent.flowId} is not waiting " +
+                        "for an ${ExternalEventResponse::class.simpleName}"
+            )
+        }
+        //Discard events not related. Some token requests do not contain the external event id so this validation will allow all token
+        // requests to be resent. e.g TokenForceClaimRelease
+        else if (externalEventStateRequestId != retryRequestId && retryRequestId != TOKEN_RETRY) {
+            throw FlowEventException(
+                "Discarding retry request received with requestId $retryRequestId. This is likely a stale record polled. Checkpoint " +
+                        "is currently waiting to receive a response for requestId $externalEventStateRequestId"
+            )
+        }
+
+        return context
+    }
+}

components/flow/flow-service/src/main/kotlin/net/corda/flow/pipeline/impl/FlowGlobalPostProcessorImpl.kt

@@ -1,7 +1,9 @@
 package net.corda.flow.pipeline.impl
 
+import net.corda.data.flow.event.external.ExternalEventRetryRequest
 import net.corda.data.flow.event.mapper.FlowMapperEvent
 import net.corda.data.flow.event.mapper.ScheduleCleanup
+import net.corda.data.flow.state.external.ExternalEventState
 import net.corda.data.flow.state.session.SessionState
 import net.corda.data.flow.state.session.SessionStateType
 import net.corda.flow.external.events.impl.ExternalEventManager
@@ -51,11 +53,11 @@ class FlowGlobalPostProcessorImpl @Activate constructor(
         postProcessPendingPlatformError(context)
 
         val outputRecords = getSessionEvents(context, now) +
-            getFlowMapperSessionCleanupEvents(context, now) +
-            getExternalEvent(context, now)
+                getFlowMapperSessionCleanupEvents(context, now) +
+                getExternalEvent(context, now)
 
         context.flowMetrics.flowEventCompleted(context.inputEvent.payload::class.java.name)
-        val metadata = getStateMetadata(context)
+        val metadata = updateFlowSessionMetadata(context)
 
         return context.copy(
             outputRecords = context.outputRecords + outputRecords,
@@ -111,7 +113,7 @@ class FlowGlobalPostProcessorImpl @Activate constructor(
         if (!counterpartyExists) {
             val msg =
                 "[${context.checkpoint.holdingIdentity.x500Name}] has failed to create a flow with counterparty: " +
-                    "[${counterparty}] as the recipient doesn't exist in the network."
+                        "[${counterparty}] as the recipient doesn't exist in the network."
             sessionManager.errorSession(sessionState)
             if (doesCheckpointExist) {
                 log.debug { "$msg. Throwing FlowFatalException" }
@@ -157,24 +159,26 @@ class FlowGlobalPostProcessorImpl @Activate constructor(
      * Check to see if any external events needs to be sent or resent.
      */
     private fun getExternalEvent(context: FlowEventContext<Any>, now: Instant): List<Record<*, *>> {
-        val externalEventState = context.checkpoint.externalEventState
-        return if (externalEventState == null) {
-            listOf()
-        } else {
-            val retryWindow = context.flowConfig.getLong(EXTERNAL_EVENT_MESSAGE_RESEND_WINDOW)
-            externalEventManager.getEventToSend(externalEventState, now, Duration.ofMillis(retryWindow))
-                .let { (updatedExternalEventState, record) ->
-                    context.checkpoint.externalEventState = updatedExternalEventState
-                    if (record != null) {
-                        listOf(record)
-                    } else {
-                        listOf()
-                    }
+        val externalEventState = context.checkpoint.externalEventState ?: return emptyList()
+
+        return when (context.inputEvent.payload) {
+            is ExternalEventRetryRequest -> getTransientRetryRequest(externalEventState, now)
+            else -> {
+                val retryWindow = Duration.ofMillis(context.flowConfig.getLong(EXTERNAL_EVENT_MESSAGE_RESEND_WINDOW))
+                externalEventManager.getEventToSend(externalEventState, now, retryWindow).let { (updatedState, record) ->
+                    context.checkpoint.externalEventState = updatedState
+                    listOfNotNull(record)
                 }
+            }
         }
     }
 
-    private fun getStateMetadata(context: FlowEventContext<Any>): Metadata? {
+    private fun getTransientRetryRequest(externalEventState: ExternalEventState, now: Instant):
+            List<Record<*, *>> {
+        return listOf(externalEventManager.getRetryEvent(externalEventState, now))
+    }
+
+    private fun updateFlowSessionMetadata(context: FlowEventContext<Any>): Metadata? {
         val checkpoint = context.checkpoint
         // Find the earliest expiry time for any open sessions.
         val lastReceivedMessageTime = checkpoint.sessions.filter {

components/flow/flow-service/src/test/kotlin/net/corda/flow/external/events/impl/ExternalEventManagerImplTest.kt

@@ -1,9 +1,5 @@
 package net.corda.flow.external.events.impl
 
-import java.nio.ByteBuffer
-import java.time.Instant
-import java.time.temporal.ChronoUnit
-import java.util.stream.Stream
 import net.corda.avro.serialization.CordaAvroDeserializer
 import net.corda.avro.serialization.CordaAvroSerializer
 import net.corda.data.ExceptionEnvelope
@@ -34,7 +30,11 @@ import org.mockito.kotlin.mock
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.verifyNoInteractions
 import org.mockito.kotlin.whenever
+import java.nio.ByteBuffer
 import java.time.Duration
+import java.time.Instant
+import java.time.temporal.ChronoUnit
+import java.util.stream.Stream
 
 class ExternalEventManagerImplTest {
 
@@ -530,4 +530,35 @@ class ExternalEventManagerImplTest {
 
         assertEquals(null, record)
     }
+
+    @Test
+    fun `getRetryEvent returns an external event`() {
+        val now = Instant.now().truncatedTo(ChronoUnit.MILLIS)
+        val key = ByteBuffer.wrap(KEY.toByteArray())
+        val payload = ByteBuffer.wrap(byteArrayOf(1, 2, 3))
+
+        val externalEvent = ExternalEvent().apply {
+            this.topic = TOPIC
+            this.key = key
+            this.payload = payload
+            this.timestamp = now.minusSeconds(10)
+        }
+
+        val externalEventState = ExternalEventState().apply {
+            requestId = REQUEST_ID_1
+            eventToSend = externalEvent
+            sendTimestamp = null
+            status = ExternalEventStateStatus(ExternalEventStateType.OK, null)
+        }
+
+        val record = externalEventManager.getRetryEvent(
+            externalEventState,
+            now,
+        )
+
+
+        assertEquals(TOPIC, record.topic)
+        assertEquals(key.array(), record.key)
+        assertEquals(payload.array(), record.value)
+    }
 }