ENT-4601 Public API to run external operations from a flow (#5833)

Deprecate FlowAsyncOperation and reimplement public versions FlowExternalOperation and FlowExternalAsyncOperation.

await added to FlowLogic to allow easy calling from both Java and Kotlin. There are two overrides of await (one for FlowExternalOperation and FlowExternalAsyncOperation).

Implementations of FlowExternalOperation return a result (written as blocking code) from their execute function. This operation will then be executed using a thread provided by the externalOperationExecutor.

Implementations of FlowExternalAsyncOperation return a future from their execute function. This operation must be executed on a newly spawned thread or one provided by a thread pool. It is up to developers to handle threading in this scenario.

The default thread pool (externalOperationExecutor) can be configured through the flowExternalOperationThreadPoolSize node config.

The current implementation leaves FlowAsyncOperation alone, meaning that any developers that have used it (even though it is internal) won't need to change their apps. If this was not concern I would delete it completely and replumb the state machine code. Instead, it has been marked with @DoNotImplement and executeAsync is annotated with @Deprecated

core/src/main/kotlin/net/corda/core/flows/FlowExternalOperations.kt

@@ -0,0 +1,67 @@
+package net.corda.core.flows
+
+import net.corda.core.internal.ServiceHubCoreInternal
+import net.corda.core.node.ServiceHub
+import java.util.concurrent.CompletableFuture
+
+/**
+ * [FlowExternalAsyncOperation] represents an external future that blocks a flow from continuing until the future returned by
+ * [FlowExternalAsyncOperation.execute] has completed. Examples of external processes where [FlowExternalAsyncOperation] would be useful
+ * include, triggering a long running process on an external system or retrieving information from a service that might be down.
+ *
+ * The flow will suspend while it is blocked to free up a flow worker thread, which allows other flows to continue processing while waiting
+ * for the result of this process.
+ *
+ * Implementations of [FlowExternalAsyncOperation] should ideally hold references to any external values required by [execute]. These
+ * references should be passed into the implementation's constructor. For example, an amount or a reference to a Corda Service could be
+ * passed in.
+ *
+ * It is discouraged to insert into the node's database from a [FlowExternalAsyncOperation], except for keeping track of [deduplicationId]s
+ * that have been processed. It is possible to interact with the database from inside a [FlowExternalAsyncOperation] but, for most
+ * operations, is not currently supported.
+ */
+interface FlowExternalAsyncOperation<R : Any> {
+
+    /**
+     * Executes a future.
+     *
+     * The future created and returned from [execute] must handle its own threads. If a new thread is not spawned or taken from a thread
+     * pool, then the flow worker thread will be used. This removes any benefit from using an [FlowExternalAsyncOperation].
+     *
+     * @param deduplicationId  If the flow restarts from a checkpoint (due to node restart, or via a visit to the flow
+     * hospital following an error) the execute method might be called more than once by the Corda flow state machine.
+     * For each duplicate call, the deduplicationId is guaranteed to be the same allowing duplicate requests to be
+     * de-duplicated if necessary inside the execute method.
+     */
+    fun execute(deduplicationId: String): CompletableFuture<R>
+}
+
+/**
+ * [FlowExternalOperation] represents an external process that blocks a flow from continuing until the result of [execute]
+ * has been retrieved. Examples of external processes where [FlowExternalOperation] would be useful include, triggering a long running
+ * process on an external system or retrieving information from a service that might be down.
+ *
+ * The flow will suspend while it is blocked to free up a flow worker thread, which allows other flows to continue processing while waiting
+ * for the result of this process.
+ *
+ * Implementations of [FlowExternalOperation] should ideally hold references to any external values required by [execute]. These references
+ * should be passed into the implementation's constructor. For example, an amount or a reference to a Corda Service could be passed in.
+ *
+ * It is discouraged to insert into the node's database from a [FlowExternalOperation], except for keeping track of [deduplicationId]s that
+ * have been processed. It is possible to interact with the database from inside a [FlowExternalOperation] but, for most operations, is not
+ * currently supported.
+ */
+interface FlowExternalOperation<R : Any> {
+
+    /**
+     * Executes a blocking operation.
+     *
+     * The execution of [execute] will be run on a thread from the node's external process thread pool when called by [FlowLogic.await].
+     *
+     * @param deduplicationId  If the flow restarts from a checkpoint (due to node restart, or via a visit to the flow
+     * hospital following an error) the execute method might be called more than once by the Corda flow state machine.
+     * For each duplicate call, the deduplicationId is guaranteed to be the same allowing duplicate requests to be
+     * de-duplicated if necessary inside the execute method.
+     */
+    fun execute(deduplicationId: String): R
+}

core/src/main/kotlin/net/corda/core/flows/FlowLogic.kt

@@ -4,13 +4,22 @@ import co.paralleluniverse.fibers.Suspendable
 import co.paralleluniverse.strands.Strand
 import net.corda.core.CordaInternal
 import net.corda.core.DeleteForDJVM
+import net.corda.core.concurrent.CordaFuture
 import net.corda.core.contracts.StateRef
 import net.corda.core.crypto.SecureHash
 import net.corda.core.identity.AbstractParty
 import net.corda.core.identity.AnonymousParty
 import net.corda.core.identity.Party
 import net.corda.core.identity.PartyAndCertificate
-import net.corda.core.internal.*
+import net.corda.core.internal.FlowAsyncOperation
+import net.corda.core.internal.FlowIORequest
+import net.corda.core.internal.FlowStateMachine
+import net.corda.core.internal.ServiceHubCoreInternal
+import net.corda.core.internal.WaitForStateConsumption
+import net.corda.core.internal.abbreviate
+import net.corda.core.internal.checkPayloadIs
+import net.corda.core.internal.concurrent.asCordaFuture
+import net.corda.core.internal.uncheckedCast
 import net.corda.core.messaging.DataFeed
 import net.corda.core.node.NodeInfo
 import net.corda.core.node.ServiceHub
@@ -24,7 +33,10 @@ import net.corda.core.utilities.debug
 import net.corda.core.utilities.toNonEmptySet
 import org.slf4j.Logger
 import java.time.Duration
-import java.util.*
+import java.util.HashMap
+import java.util.LinkedHashMap
+import java.util.concurrent.CompletableFuture
+import java.util.function.Supplier
 
 /**
  * A sub-class of [FlowLogic<T>] implements a flow using direct, straight line blocking code. Thus you
@@ -432,7 +444,12 @@ abstract class FlowLogic<out T> {
      * @param stateRefs the StateRefs which will be consumed in the future.
      */
     @Suspendable
-    fun waitForStateConsumption(stateRefs: Set<StateRef>) = executeAsync(WaitForStateConsumption(stateRefs, serviceHub))
+    fun waitForStateConsumption(stateRefs: Set<StateRef>) {
+        // Manually call the equivalent of [await] to remove extra wrapping of objects
+        // Makes serializing of object easier for [CheckpointDumper] as well
+        val request = FlowIORequest.ExecuteAsyncOperation(WaitForStateConsumption(stateRefs, serviceHub))
+        return stateMachine.suspend(request, false)
+    }
 
     /**
      * Returns a shallow copy of the Quasar stack frames at the time of call to [flowStackSnapshot]. Use this to inspect
@@ -503,6 +520,72 @@ abstract class FlowLogic<out T> {
     private fun <R> castMapValuesToKnownType(map: Map<FlowSession, UntrustworthyData<Any>>): List<UntrustworthyData<R>> {
         return map.values.map { uncheckedCast<Any, UntrustworthyData<R>>(it) }
     }
+
+    /**
+     * Executes the specified [operation] and suspends until operation completion.
+     *
+     * An implementation of [FlowExternalAsyncOperation] should be provided that creates a new future that the state machine awaits
+     * completion of.
+     *
+     */
+    @Suspendable
+    fun <R : Any> await(operation: FlowExternalAsyncOperation<R>): R {
+        // Wraps the passed in [FlowExternalAsyncOperation] so its [CompletableFuture] can be converted into a [CordaFuture]
+        val flowAsyncOperation = object : FlowAsyncOperation<R>, WrappedFlowExternalAsyncOperation<R> {
+            override val operation = operation
+            override fun execute(deduplicationId: String): CordaFuture<R> {
+                return this.operation.execute(deduplicationId).asCordaFuture()
+            }
+        }
+        val request = FlowIORequest.ExecuteAsyncOperation(flowAsyncOperation)
+        return stateMachine.suspend(request, false)
+    }
+
+    /**
+     * Executes the specified [operation] and suspends until operation completion.
+     *
+     * An implementation of [FlowExternalOperation] should be provided that returns a result which the state machine will run on a separate
+     * thread (using the node's external operation thread pool).
+     *
+     */
+    @Suspendable
+    fun <R : Any> await(operation: FlowExternalOperation<R>): R {
+        val flowAsyncOperation = object : FlowAsyncOperation<R>, WrappedFlowExternalOperation<R> {
+            override val serviceHub = this@FlowLogic.serviceHub as ServiceHubCoreInternal
+            override val operation = operation
+            override fun execute(deduplicationId: String): CordaFuture<R> {
+                // Using a [CompletableFuture] allows unhandled exceptions to be thrown inside the background operation
+                // the exceptions will be set on the future by [CompletableFuture.AsyncSupply.run]
+                return CompletableFuture.supplyAsync(
+                    Supplier { this.operation.execute(deduplicationId) },
+                    serviceHub.externalOperationExecutor
+                ).asCordaFuture()
+            }
+        }
+        val request = FlowIORequest.ExecuteAsyncOperation(flowAsyncOperation)
+        return stateMachine.suspend(request, false)
+    }
+}
+
+/**
+ * [WrappedFlowExternalAsyncOperation] is added to allow jackson to properly reference the data stored within the wrapped
+ * [FlowExternalAsyncOperation].
+ */
+private interface WrappedFlowExternalAsyncOperation<R : Any> {
+    val operation: FlowExternalAsyncOperation<R>
+}
+
+/**
+ * [WrappedFlowExternalOperation] is added to allow jackson to properly reference the data stored within the wrapped
+ * [FlowExternalOperation].
+ *
+ * The reference to [ServiceHub] is is also needed by Kryo to properly keep a reference to [ServiceHub] so that
+ * [FlowExternalOperation] can be run from the [ServiceHubCoreInternal.externalOperationExecutor] without causing errors when retrying a
+ * flow. A [NullPointerException] is thrown if [FlowLogic.serviceHub] is accessed from [FlowLogic.await] when retrying a flow.
+ */
+private interface WrappedFlowExternalOperation<R : Any> {
+    val serviceHub: ServiceHub
+    val operation: FlowExternalOperation<R>
 }
 
 /**

core/src/main/kotlin/net/corda/core/internal/FlowAsyncOperation.kt

@@ -5,7 +5,6 @@ import net.corda.core.concurrent.CordaFuture
 import net.corda.core.flows.FlowLogic
 import net.corda.core.serialization.CordaSerializable
 
-// DOCSTART FlowAsyncOperation
 /**
  * Interface for arbitrary operations that can be invoked in a flow asynchronously - the flow will suspend until the
  * operation completes. Operation parameters are expected to be injected via constructor.
@@ -21,13 +20,14 @@ interface FlowAsyncOperation<R : Any> {
      */
     fun execute(deduplicationId: String): CordaFuture<R>
 }
-// DOCEND FlowAsyncOperation
 
-// DOCSTART executeAsync
 /** Executes the specified [operation] and suspends until operation completion. */
+@Deprecated(
+    "This has been replaced by [FlowLogic.await] that provides an improved and public API",
+    ReplaceWith("net.corda.core.flows.FlowLogic.await")
+)
 @Suspendable
 fun <T, R : Any> FlowLogic<T>.executeAsync(operation: FlowAsyncOperation<R>, maySkipCheckpoint: Boolean = false): R {
     val request = FlowIORequest.ExecuteAsyncOperation(operation)
     return stateMachine.suspend(request, maySkipCheckpoint)
 }
-// DOCEND executeAsync

core/src/main/kotlin/net/corda/core/internal/ServiceHubCoreInternal.kt

@@ -4,11 +4,14 @@ import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.DeleteForDJVM
 import net.corda.core.node.ServiceHub
 import net.corda.core.node.StatesToRecord
+import java.util.concurrent.ExecutorService
 
 // TODO: This should really be called ServiceHubInternal but that name is already taken by net.corda.node.services.api.ServiceHubInternal.
 @DeleteForDJVM
 interface ServiceHubCoreInternal : ServiceHub {
 
+    val externalOperationExecutor: ExecutorService
+
     val attachmentTrustCalculator: AttachmentTrustCalculator
 
     fun createTransactionsResolver(flow: ResolveTransactionsFlow): TransactionsResolver

core/src/main/kotlin/net/corda/core/internal/concurrent/CordaFutureImpl.kt

@@ -27,6 +27,17 @@ fun <V, W, X> CordaFuture<out V>.thenMatch(success: (V) -> W, failure: (Throwabl
 /** When this future is done and the outcome is failure, log the throwable. */
 fun CordaFuture<*>.andForget(log: Logger) = thenMatch({}, { log.error("Background task failed:", it) })
 
+fun <RESULT> CordaFuture<out RESULT>.doOnComplete(accept: (RESULT) -> Unit): CordaFuture<RESULT> {
+    return CordaFutureImpl<RESULT>().also { result ->
+        thenMatch({
+            accept(it)
+            result.capture { it }
+        }, {
+            result.setException(it)
+        })
+    }
+}
+
 /**
  * Returns a future that will have an outcome of applying the given transform to this future's value.
  * But if this future fails, the transform is not invoked and the returned future becomes done with the same throwable.

core/src/main/kotlin/net/corda/core/internal/notary/SinglePartyNotaryService.kt

@@ -1,7 +1,6 @@
 package net.corda.core.internal.notary
 
 import co.paralleluniverse.fibers.Suspendable
-import net.corda.core.concurrent.CordaFuture
 import net.corda.core.contracts.StateRef
 import net.corda.core.contracts.TimeWindow
 import net.corda.core.crypto.Crypto
@@ -9,16 +8,16 @@ import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.SignableData
 import net.corda.core.crypto.SignatureMetadata
 import net.corda.core.crypto.TransactionSignature
+import net.corda.core.flows.FlowExternalAsyncOperation
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.NotarisationRequestSignature
 import net.corda.core.identity.Party
-import net.corda.core.internal.FlowAsyncOperation
-import net.corda.core.internal.executeAsync
 import net.corda.core.internal.notary.UniquenessProvider.Result
 import net.corda.core.serialization.CordaSerializable
 import net.corda.core.utilities.contextLogger
 import org.slf4j.Logger
 import java.time.Duration
+import java.util.concurrent.CompletableFuture
 
 /** Base implementation for a notary service operated by a singe party. */
 abstract class SinglePartyNotaryService : NotaryService() {
@@ -48,7 +47,7 @@ abstract class SinglePartyNotaryService : NotaryService() {
         val callingFlow = FlowLogic.currentTopLevel
                 ?: throw IllegalStateException("This method should be invoked in a flow context.")
 
-        val result = callingFlow.executeAsync(
+        val result = callingFlow.await(
                 CommitOperation(
                         this,
                         inputs,
@@ -87,10 +86,10 @@ abstract class SinglePartyNotaryService : NotaryService() {
             val requestSignature: NotarisationRequestSignature,
             val timeWindow: TimeWindow?,
             val references: List<StateRef>
-    ) : FlowAsyncOperation<Result> {
+    ) : FlowExternalAsyncOperation<Result> {
 
-        override fun execute(deduplicationId: String): CordaFuture<Result> {
-            return service.uniquenessProvider.commit(inputs, txId, caller, requestSignature, timeWindow, references)
+        override fun execute(deduplicationId: String): CompletableFuture<Result> {
+            return service.uniquenessProvider.commit(inputs, txId, caller, requestSignature, timeWindow, references).toCompletableFuture()
         }
     }