Add extension hooks to FinalityFlow

Add extension hooks to FinalityFlow to support different behaviours depending on whether the node knows who all of the parties in a transaction are.

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

@@ -5,6 +5,7 @@ import net.corda.core.contracts.ContractState
 import net.corda.core.contracts.StateRef
 import net.corda.core.contracts.TransactionState
 import net.corda.core.crypto.isFulfilledBy
+import net.corda.core.identity.AbstractParty
 import net.corda.core.identity.Party
 import net.corda.core.node.ServiceHub
 import net.corda.core.transactions.LedgerTransaction
@@ -32,9 +33,10 @@ import net.corda.core.utilities.toNonEmptySet
  * @param transactions What to commit.
  * @param extraRecipients A list of additional participants to inform of the transaction.
  */
-class FinalityFlow(val transactions: Iterable<SignedTransaction>,
+open class FinalityFlow(val transactions: Iterable<SignedTransaction>,
                    val extraRecipients: Set<Party>,
                    override val progressTracker: ProgressTracker) : FlowLogic<List<SignedTransaction>>() {
+    val extraParticipants: Set<Participant> = extraRecipients.map { it -> Participant(it, it) }.toSet()
     constructor(transaction: SignedTransaction, extraParticipants: Set<Party>) : this(listOf(transaction), extraParticipants, tracker())
     constructor(transaction: SignedTransaction) : this(listOf(transaction), emptySet(), tracker())
     constructor(transaction: SignedTransaction, progressTracker: ProgressTracker) : this(listOf(transaction), emptySet(), progressTracker)
@@ -50,6 +52,9 @@ class FinalityFlow(val transactions: Iterable<SignedTransaction>,
         fun tracker() = ProgressTracker(NOTARISING, BROADCASTING)
     }
 
+    open protected val me
+        get() = serviceHub.myInfo.legalIdentity
+
     @Suspendable
     @Throws(NotaryException::class)
     override fun call(): List<SignedTransaction> {
@@ -59,34 +64,43 @@ class FinalityFlow(val transactions: Iterable<SignedTransaction>,
         // Lookup the resolved transactions and use them to map each signed transaction to the list of participants.
         // Then send to the notary if needed, record locally and distribute.
         progressTracker.currentStep = NOTARISING
-        val notarisedTxns = notariseAndRecord(lookupParties(resolveDependenciesOf(transactions)))
+        val notarisedTxns: List<Pair<SignedTransaction, Set<Participant>>> = resolveDependenciesOf(transactions)
+                .map { (stx, ltx) -> Pair(notariseAndRecord(stx), lookupParties(ltx)) }
 
         // Each transaction has its own set of recipients, but extra recipients get them all.
         progressTracker.currentStep = BROADCASTING
-        val me = serviceHub.myInfo.legalIdentity
         for ((stx, parties) in notarisedTxns) {
-            val participants = parties + extraRecipients - me
+            broadcastTransaction(stx, (parties + extraParticipants).filter { it.wellKnown != me })
-            if (participants.isNotEmpty()) {
-                subFlow(BroadcastTransactionFlow(stx, participants.toNonEmptySet()))
-            }
         }
         return notarisedTxns.map { it.first }
     }
 
-    // TODO: API: Make some of these protected?
+    /**
+     * Broadcast a transaction to the participants. By default calls [BroadcastTransactionFlow], however can be
+     * overridden for more complex transaction delivery protocols (for example where not all parties know each other).
+     * This implementation will filter out any participants for who there is no well known identity.
+     *
+     * @param participants the participants to send the transaction to. This is expected to include extra participants
+     * and exclude the local node.
+     */
+    @Suspendable
+    open protected fun broadcastTransaction(stx: SignedTransaction, participants: Iterable<Participant>) {
+        val wellKnownParticipants = participants.map { it.wellKnown }.filterNotNull()
+        if (wellKnownParticipants.isNotEmpty()) {
+            subFlow(BroadcastTransactionFlow(stx, wellKnownParticipants.toNonEmptySet()))
+        }
+    }
 
     @Suspendable
-    private fun notariseAndRecord(stxnsAndParties: List<Pair<SignedTransaction, Set<Party>>>): List<Pair<SignedTransaction, Set<Party>>> {
+    private fun notariseAndRecord(stx: SignedTransaction): SignedTransaction {
-        return stxnsAndParties.map { (stx, parties) ->
+        val notarised = if (needsNotarySignature(stx)) {
-            val notarised = if (needsNotarySignature(stx)) {
+            val notarySignatures = subFlow(NotaryFlow.Client(stx))
-                val notarySignatures = subFlow(NotaryFlow.Client(stx))
+            stx + notarySignatures
-                stx + notarySignatures
+        } else {
-            } else {
+            stx
-                stx
-            }
-            serviceHub.recordTransactions(notarised)
-            Pair(notarised, parties)
         }
+        serviceHub.recordTransactions(notarised)
+        return notarised
     }
 
     private fun needsNotarySignature(stx: SignedTransaction): Boolean {
@@ -102,14 +116,31 @@ class FinalityFlow(val transactions: Iterable<SignedTransaction>,
         return !(notaryKey?.isFulfilledBy(signers) ?: false)
     }
 
-    private fun lookupParties(ltxns: List<Pair<SignedTransaction, LedgerTransaction>>): List<Pair<SignedTransaction, Set<Party>>> {
+    /**
-        return ltxns.map { (stx, ltx) ->
+     * Resolve the parties involved in a transaction.
-            // Calculate who is meant to see the results based on the participants involved.
+     *
-            val keys = ltx.outputs.flatMap { it.data.participants } + ltx.inputs.flatMap { it.state.data.participants }
+     * @return the set of participants and their resolved well known identities (where known).
-            // TODO: Is it safe to drop participants we don't know how to contact? Does not knowing how to contact them count as a reason to fail?
+     */
-            val parties = keys.mapNotNull { serviceHub.identityService.partyFromAnonymous(it) }.toSet()
+    open protected fun lookupParties(ltx: LedgerTransaction): Set<Participant> {
-            Pair(stx, parties)
+        // Calculate who is meant to see the results based on the participants involved.
-        }
+        return extractParticipants(ltx)
+                .map(this::partyFromAnonymous)
+                .toSet()
+    }
+
+    /**
+     * Helper function to extract all participants from a ledger transaction. Intended to help implement [lookupParties]
+     * overriding functions.
+     */
+    protected fun extractParticipants(ltx: LedgerTransaction): List<AbstractParty> {
+        return ltx.outputs.flatMap { it.data.participants } + ltx.inputs.flatMap { it.state.data.participants }
+    }
+
+    /**
+     * Helper function which wraps [IdentityService.partyFromAnonymous] so it can be called as a lambda function.
+     */
+    protected fun partyFromAnonymous(anon: AbstractParty): Participant {
+        return Participant(anon, serviceHub.identityService.partyFromAnonymous(anon))
     }
 
     private fun resolveDependenciesOf(signedTransactions: Iterable<SignedTransaction>): List<Pair<SignedTransaction, LedgerTransaction>> {
@@ -134,4 +165,6 @@ class FinalityFlow(val transactions: Iterable<SignedTransaction>,
             stx to ltx
         }
     }
+
+    data class Participant(val participant: AbstractParty, val wellKnown: Party?)
 }

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

@@ -0,0 +1,20 @@
+package net.corda.core.flows
+
+import net.corda.core.identity.Party
+import net.corda.core.transactions.LedgerTransaction
+import net.corda.core.transactions.SignedTransaction
+import net.corda.core.utilities.ProgressTracker
+
+/**
+ * Alternative finality flow which only does not attempt to take participants from the transaction, but instead all
+ * participating parties must be provided manually.
+ *
+ * @param transactions What to commit.
+ * @param extraRecipients A list of additional participants to inform of the transaction.
+ */
+class ManualFinalityFlow(transactions: Iterable<SignedTransaction>,
+                   recipients: Set<Party>,
+                   progressTracker: ProgressTracker) : FinalityFlow(transactions, recipients, progressTracker) {
+    constructor(transaction: SignedTransaction, extraParticipants: Set<Party>) : this(listOf(transaction), extraParticipants, tracker())
+    override fun lookupParties(ltx: LedgerTransaction): Set<Participant> = emptySet()
+}

core/src/test/kotlin/net/corda/core/flows/FinalityFlowTests.kt

@@ -0,0 +1,56 @@
+package net.corda.core.flows
+
+import net.corda.contracts.asset.Cash
+import net.corda.core.contracts.Amount
+import net.corda.core.contracts.GBP
+import net.corda.core.contracts.Issued
+import net.corda.core.contracts.TransactionType
+import net.corda.core.getOrThrow
+import net.corda.core.identity.Party
+import net.corda.core.transactions.TransactionBuilder
+import net.corda.testing.node.MockNetwork
+import net.corda.testing.node.MockServices
+import org.junit.After
+import org.junit.Before
+import org.junit.Test
+import kotlin.test.assertEquals
+
+class FinalityFlowTests {
+    lateinit var mockNet: MockNetwork
+    lateinit var nodeA: MockNetwork.MockNode
+    lateinit var nodeB: MockNetwork.MockNode
+    lateinit var notary: Party
+    val services = MockServices()
+
+    @Before
+    fun setup() {
+        mockNet = MockNetwork()
+        val nodes = mockNet.createSomeNodes(2)
+        nodeA = nodes.partyNodes[0]
+        nodeB = nodes.partyNodes[1]
+        notary = nodes.notaryNode.info.notaryIdentity
+        mockNet.runNetwork()
+    }
+
+    @After
+    fun tearDown() {
+        mockNet.stopNodes()
+    }
+
+    @Test
+    fun `finalise a simple transaction`() {
+        val amount = Amount(1000, Issued(nodeA.info.legalIdentity.ref(0), GBP))
+        val builder = TransactionBuilder(TransactionType.General, notary)
+        Cash().generateIssue(builder, amount, nodeB.info.legalIdentity, notary)
+        val stx = nodeA.services.signInitialTransaction(builder)
+        val flow = nodeA.services.startFlow(FinalityFlow(stx))
+        mockNet.runNetwork()
+        val result = flow.resultFuture.getOrThrow()
+        val notarisedTx = result.single()
+        notarisedTx.verifySignatures()
+        val transactionSeenByB = nodeB.services.database.transaction {
+            nodeB.services.validatedTransactions.getTransaction(notarisedTx.id)
+        }
+        assertEquals(notarisedTx, transactionSeenByB)
+    }
+}

core/src/test/kotlin/net/corda/core/flows/ManualFinalityFlowTests.kt

@@ -0,0 +1,64 @@
+package net.corda.core.flows
+
+import net.corda.contracts.asset.Cash
+import net.corda.core.contracts.Amount
+import net.corda.core.contracts.GBP
+import net.corda.core.contracts.Issued
+import net.corda.core.contracts.TransactionType
+import net.corda.core.getOrThrow
+import net.corda.core.identity.Party
+import net.corda.core.transactions.TransactionBuilder
+import net.corda.testing.node.MockNetwork
+import net.corda.testing.node.MockServices
+import org.junit.After
+import org.junit.Before
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+
+class ManualFinalityFlowTests {
+    lateinit var mockNet: MockNetwork
+    lateinit var nodeA: MockNetwork.MockNode
+    lateinit var nodeB: MockNetwork.MockNode
+    lateinit var nodeC: MockNetwork.MockNode
+    lateinit var notary: Party
+    val services = MockServices()
+
+    @Before
+    fun setup() {
+        mockNet = MockNetwork()
+        val nodes = mockNet.createSomeNodes(3)
+        nodeA = nodes.partyNodes[0]
+        nodeB = nodes.partyNodes[1]
+        nodeC = nodes.partyNodes[2]
+        notary = nodes.notaryNode.info.notaryIdentity
+        mockNet.runNetwork()
+    }
+
+    @After
+    fun tearDown() {
+        mockNet.stopNodes()
+    }
+
+    @Test
+    fun `finalise a simple transaction`() {
+        val amount = Amount(1000, Issued(nodeA.info.legalIdentity.ref(0), GBP))
+        val builder = TransactionBuilder(TransactionType.General, notary)
+        Cash().generateIssue(builder, amount, nodeB.info.legalIdentity, notary)
+        val stx = nodeA.services.signInitialTransaction(builder)
+        val flow = nodeA.services.startFlow(ManualFinalityFlow(stx, setOf(nodeC.info.legalIdentity)))
+        mockNet.runNetwork()
+        val result = flow.resultFuture.getOrThrow()
+        val notarisedTx = result.single()
+        notarisedTx.verifySignatures()
+        // We override the participants, so node C will get a copy despite not being involved, and B won't
+        val transactionSeenByB = nodeB.services.database.transaction {
+            nodeB.services.validatedTransactions.getTransaction(notarisedTx.id)
+        }
+        assertNull(transactionSeenByB)
+        val transactionSeenByC = nodeC.services.database.transaction {
+            nodeC.services.validatedTransactions.getTransaction(notarisedTx.id)
+        }
+        assertEquals(notarisedTx, transactionSeenByC)
+    }
+}

docs/source/changelog.rst

@@ -34,6 +34,10 @@ UNRELEASED
 * Moved the core flows previously found in ``net.corda.flows`` into ``net.corda.core.flows``. This is so that all packages
   in the ``core`` module begin with ``net.corda.core``.
 
+* ``FinalityFlow`` now has can be subclassed, and the ``broadcastTransaction`` and ``lookupParties`` function can be
+  overriden in order to handle cases where no single transaction participant is aware of all parties, and therefore
+  the transaction must be relayed between participants rather than sent from a single node.
+
 Milestone 13
 ------------
 

docs/source/release-notes.rst

@@ -6,6 +6,9 @@ Here are release notes for each snapshot release from M9 onwards.
 Unreleased
 ----------
 
+The transaction finalisation flow (``FinalityFlow``) has had hooks for alternative implementations, for example in
+scenarios where no single participant in a transaction is aware of the well known identities of all parties.
+
 Milestone 13
 ------------