DA5-22: Distribution and validation of draft transactions

components/ledger/ledger-utxo-flow/src/main/kotlin/net/corda/ledger/utxo/flow/impl/UtxoLedgerServiceImpl.kt

@@ -8,7 +8,9 @@ import net.corda.ledger.utxo.flow.impl.flows.finality.UtxoFinalityFlow
 import net.corda.ledger.utxo.flow.impl.flows.finality.UtxoReceiveFinalityFlow
 import net.corda.ledger.utxo.flow.impl.flows.transactionbuilder.ReceiveAndUpdateTransactionBuilderFlow
 import net.corda.ledger.utxo.flow.impl.flows.transactionbuilder.SendTransactionBuilderDiffFlow
+import net.corda.ledger.utxo.flow.impl.flows.transactiontransmission.ReceiveLedgerTransactionFlow
 import net.corda.ledger.utxo.flow.impl.flows.transactiontransmission.ReceiveTransactionFlow
+import net.corda.ledger.utxo.flow.impl.flows.transactiontransmission.SendAsLedgerTransactionFlow
 import net.corda.ledger.utxo.flow.impl.flows.transactiontransmission.SendTransactionFlow
 import net.corda.ledger.utxo.flow.impl.persistence.UtxoLedgerPersistenceService
 import net.corda.ledger.utxo.flow.impl.persistence.UtxoLedgerStateQueryService
@@ -279,6 +281,16 @@ class UtxoLedgerServiceImpl @Activate constructor(
         flowEngine.subFlow(SendTransactionFlow(signedTransaction, sessions))
     }
 
+    @Suspendable
+    override fun receiveLedgerTransaction(session: FlowSession): UtxoLedgerTransaction {
+        return flowEngine.subFlow(ReceiveLedgerTransactionFlow(session))
+    }
+
+    @Suspendable
+    override fun sendAsLedgerTransaction(signedTransaction: UtxoSignedTransaction, sessions: List<FlowSession>) {
+        flowEngine.subFlow(SendAsLedgerTransactionFlow(signedTransaction, sessions))
+    }
+
     @Suspendable
     override fun sendUpdatedTransactionBuilder(
         transactionBuilder: UtxoTransactionBuilder,

components/ledger/ledger-utxo-flow/src/main/kotlin/net/corda/ledger/utxo/flow/impl/flows/backchain/TransactionBackchainExtensions.kt

@@ -1,10 +1,17 @@
 package net.corda.ledger.utxo.flow.impl.flows.backchain
 
 import net.corda.v5.crypto.SecureHash
+import net.corda.v5.ledger.utxo.transaction.UtxoLedgerTransaction
 import net.corda.v5.ledger.utxo.transaction.UtxoSignedTransaction
 
 val UtxoSignedTransaction.dependencies: Set<SecureHash>
     get() = this
         .let { it.inputStateRefs.asSequence() + it.referenceStateRefs.asSequence() }
         .map { it.transactionId }
         .toSet()
+
+val UtxoLedgerTransaction.dependencies: Set<SecureHash>
+    get() = this
+        .let { it.inputStateRefs.asSequence() + it.referenceStateRefs.asSequence() }
+        .map { it.transactionId }
+        .toSet()

components/ledger/ledger-utxo-flow/src/main/kotlin/net/corda/ledger/utxo/flow/impl/flows/transactiontransmission/ReceiveLedgerTransactionFlow.kt

@@ -0,0 +1,61 @@
+package net.corda.ledger.utxo.flow.impl.flows.transactiontransmission
+
+import net.corda.ledger.common.data.transaction.WireTransaction
+import net.corda.ledger.common.flow.flows.Payload
+import net.corda.ledger.utxo.flow.impl.flows.backchain.InvalidBackchainException
+import net.corda.ledger.utxo.flow.impl.flows.backchain.TransactionBackchainResolutionFlow
+import net.corda.ledger.utxo.flow.impl.flows.backchain.dependencies
+import net.corda.ledger.utxo.flow.impl.transaction.factory.UtxoLedgerTransactionFactory
+import net.corda.sandbox.CordaSystemFlow
+import net.corda.utilities.trace
+import net.corda.v5.application.crypto.DigitalSignatureAndMetadata
+import net.corda.v5.application.flows.CordaInject
+import net.corda.v5.application.flows.FlowEngine
+import net.corda.v5.application.flows.SubFlow
+import net.corda.v5.application.messaging.FlowSession
+import net.corda.v5.base.annotations.Suspendable
+import net.corda.v5.ledger.utxo.transaction.UtxoLedgerTransaction
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+@CordaSystemFlow
+class ReceiveLedgerTransactionFlow(
+    private val session: FlowSession
+) : SubFlow<UtxoLedgerTransaction> {
+
+    private companion object {
+        private val log: Logger = LoggerFactory.getLogger(this::class.java)
+    }
+
+    @CordaInject
+    lateinit var flowEngine: FlowEngine
+
+    @CordaInject
+    lateinit var utxoLedgerTransactionFactory: UtxoLedgerTransactionFactory
+
+    @Suspendable
+    override fun call(): UtxoLedgerTransaction {
+        val wireTransaction = session.receive(WireTransaction::class.java)
+        val ledgerTransaction = utxoLedgerTransactionFactory.create(wireTransaction)
+
+        val transactionDependencies = ledgerTransaction.dependencies
+        if (transactionDependencies.isNotEmpty()) {
+            try {
+                flowEngine.subFlow(TransactionBackchainResolutionFlow(transactionDependencies, session))
+            } catch (e: InvalidBackchainException) {
+                val message = "Invalid transaction: ${wireTransaction.id} found during back-chain resolution."
+                log.warn(message, e)
+                session.send(Payload.Failure<List<DigitalSignatureAndMetadata>>(message))
+                throw e
+            }
+        } else {
+            log.trace {
+                "Transaction with id ${wireTransaction.id} has no dependencies so backchain resolution will not be performed."
+            }
+        }
+
+        session.send(Payload.Success("Successfully received transaction."))
+
+        return ledgerTransaction
+    }
+}

components/ledger/ledger-utxo-flow/src/main/kotlin/net/corda/ledger/utxo/flow/impl/flows/transactiontransmission/SendAsLedgerTransactionFlow.kt

@@ -0,0 +1,56 @@
+package net.corda.ledger.utxo.flow.impl.flows.transactiontransmission
+
+import net.corda.ledger.common.flow.flows.Payload
+import net.corda.ledger.utxo.flow.impl.flows.backchain.TransactionBackchainSenderFlow
+import net.corda.ledger.utxo.flow.impl.flows.backchain.dependencies
+import net.corda.ledger.utxo.flow.impl.transaction.UtxoSignedTransactionInternal
+import net.corda.sandbox.CordaSystemFlow
+import net.corda.utilities.trace
+import net.corda.v5.application.flows.CordaInject
+import net.corda.v5.application.flows.FlowEngine
+import net.corda.v5.application.flows.SubFlow
+import net.corda.v5.application.messaging.FlowMessaging
+import net.corda.v5.application.messaging.FlowSession
+import net.corda.v5.base.annotations.Suspendable
+import net.corda.v5.base.exceptions.CordaRuntimeException
+import net.corda.v5.ledger.utxo.transaction.UtxoSignedTransaction
+import org.slf4j.LoggerFactory
+
+@CordaSystemFlow
+class SendAsLedgerTransactionFlow(
+    private val transaction: UtxoSignedTransaction,
+    private val sessions: List<FlowSession>
+) : SubFlow<Unit> {
+
+    private companion object {
+        val log = LoggerFactory.getLogger(this::class.java)
+    }
+
+    @CordaInject
+    lateinit var flowMessaging: FlowMessaging
+
+    @CordaInject
+    lateinit var flowEngine: FlowEngine
+
+    @Suspendable
+    override fun call() {
+        flowMessaging.sendAll((transaction as UtxoSignedTransactionInternal).wireTransaction, sessions.toSet())
+
+        sessions.forEach {
+            if (transaction.dependencies.isNotEmpty()) {
+                flowEngine.subFlow(TransactionBackchainSenderFlow(transaction.id, it))
+            } else {
+                log.trace {
+                    "Transaction with id ${transaction.id} has no dependencies so backchain resolution will not be performed."
+                }
+            }
+
+            val sendingTransactionResult = it.receive(Payload::class.java)
+            if (sendingTransactionResult is Payload.Failure) {
+                throw CordaRuntimeException(
+                    sendingTransactionResult.message
+                )
+            }
+        }
+    }
+}

components/ledger/ledger-utxo-flow/src/test/kotlin/net/corda/ledger/utxo/flow/impl/flows/transactiontransmission/ReceiveLedgerTransactionFlowTest.kt

@@ -0,0 +1,90 @@
+package net.corda.ledger.utxo.flow.impl.flows.transactiontransmission
+
+import net.corda.crypto.core.SecureHashImpl
+import net.corda.ledger.common.data.transaction.WireTransaction
+import net.corda.ledger.common.flow.flows.Payload
+import net.corda.ledger.utxo.data.transaction.UtxoLedgerTransactionInternal
+import net.corda.ledger.utxo.flow.impl.flows.backchain.TransactionBackchainResolutionFlow
+import net.corda.ledger.utxo.flow.impl.flows.backchain.dependencies
+import net.corda.ledger.utxo.flow.impl.transaction.factory.UtxoLedgerTransactionFactory
+import net.corda.v5.application.flows.FlowEngine
+import net.corda.v5.application.messaging.FlowSession
+import net.corda.v5.ledger.utxo.StateRef
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.mockito.kotlin.any
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.never
+import org.mockito.kotlin.spy
+import org.mockito.kotlin.verify
+import org.mockito.kotlin.whenever
+
+class ReceiveLedgerTransactionFlowTest {
+    private companion object {
+        val TX_ID_1 = SecureHashImpl("SHA", byteArrayOf(2, 2, 2, 2))
+        val TX_ID_2 = SecureHashImpl("SHA", byteArrayOf(3, 3, 3, 3))
+
+        val TX_2_INPUT_DEPENDENCY_STATE_REF_1 = StateRef(TX_ID_2, 0)
+        val TX_3_INPUT_DEPENDENCY_STATE_REF_1 = StateRef(TX_ID_2, 0)
+        val TX_3_INPUT_DEPENDENCY_STATE_REF_2 = StateRef(TX_ID_2, 1)
+
+        val TX_3_INPUT_REFERENCE_DEPENDENCY_STATE_REF_1 = StateRef(TX_ID_2, 0)
+        val TX_3_INPUT_REFERENCE_DEPENDENCY_STATE_REF_2 = StateRef(TX_ID_2, 1)
+    }
+
+    private val flowEngine = mock<FlowEngine>()
+    private var utxoLedgerTransactionFactory = mock<UtxoLedgerTransactionFactory>()
+
+    private val sessionAlice = mock<FlowSession>()
+
+    private val transaction = mock<WireTransaction>()
+    private val ledgerTransaction = mock<UtxoLedgerTransactionInternal>()
+
+    @BeforeEach
+    fun beforeEach() {
+        whenever(transaction.id).thenReturn(TX_ID_1)
+        whenever(flowEngine.subFlow(any<TransactionBackchainResolutionFlow>())).thenReturn(Unit)
+        whenever(utxoLedgerTransactionFactory.create(transaction)).thenReturn(ledgerTransaction)
+    }
+
+    @Test
+    fun `receiving transaction with dependencies should call backchain resolution flow`() {
+        whenever(sessionAlice.receive(WireTransaction::class.java)).thenReturn(transaction)
+        whenever(ledgerTransaction.inputStateRefs).thenReturn(
+            listOf(
+                TX_2_INPUT_DEPENDENCY_STATE_REF_1,
+                TX_3_INPUT_DEPENDENCY_STATE_REF_1,
+                TX_3_INPUT_DEPENDENCY_STATE_REF_2
+            )
+        )
+        whenever(ledgerTransaction.referenceStateRefs).thenReturn(
+            listOf(
+                TX_3_INPUT_REFERENCE_DEPENDENCY_STATE_REF_1,
+                TX_3_INPUT_REFERENCE_DEPENDENCY_STATE_REF_2
+            )
+        )
+
+        callReceiveTransactionFlow(sessionAlice)
+
+        verify(flowEngine).subFlow(TransactionBackchainResolutionFlow(ledgerTransaction.dependencies, sessionAlice))
+        verify(sessionAlice).send(Payload.Success("Successfully received transaction."))
+    }
+
+    @Test
+    fun `receiving transaction with no dependencies shouldn't call backchain resolution flow`() {
+        whenever(sessionAlice.receive(WireTransaction::class.java)).thenReturn(transaction)
+        whenever(ledgerTransaction.inputStateRefs).thenReturn(listOf())
+        whenever(ledgerTransaction.referenceStateRefs).thenReturn(listOf())
+        callReceiveTransactionFlow(sessionAlice)
+
+        verify(sessionAlice).send(Payload.Success("Successfully received transaction."))
+        verify(flowEngine, never()).subFlow(TransactionBackchainResolutionFlow(ledgerTransaction.dependencies, sessionAlice))
+    }
+
+    private fun callReceiveTransactionFlow(session: FlowSession) {
+        val flow = spy(ReceiveLedgerTransactionFlow(session))
+        flow.utxoLedgerTransactionFactory = utxoLedgerTransactionFactory
+        flow.flowEngine = flowEngine
+        flow.call()
+    }
+}

components/ledger/ledger-utxo-flow/src/test/kotlin/net/corda/ledger/utxo/flow/impl/flows/transactiontransmission/SendAsLedgerTransactionFlowTest.kt

@@ -0,0 +1,109 @@
+package net.corda.ledger.utxo.flow.impl.flows.transactiontransmission
+
+import net.corda.crypto.core.SecureHashImpl
+import net.corda.ledger.common.flow.flows.Payload
+import net.corda.ledger.utxo.flow.impl.flows.backchain.TransactionBackchainSenderFlow
+import net.corda.ledger.utxo.flow.impl.transaction.UtxoSignedTransactionInternal
+import net.corda.v5.application.crypto.DigitalSignatureAndMetadata
+import net.corda.v5.application.flows.FlowEngine
+import net.corda.v5.application.messaging.FlowMessaging
+import net.corda.v5.application.messaging.FlowSession
+import net.corda.v5.base.exceptions.CordaRuntimeException
+import net.corda.v5.ledger.utxo.transaction.UtxoSignedTransaction
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.mockito.kotlin.any
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.never
+import org.mockito.kotlin.spy
+import org.mockito.kotlin.verify
+import org.mockito.kotlin.whenever
+
+class SendAsLedgerTransactionFlowTest {
+
+    private companion object {
+        val TX_ID = SecureHashImpl("SHA", byteArrayOf(1, 1, 1, 1))
+    }
+
+    private val flowEngine = mock<FlowEngine>()
+    private val flowMessaging = mock<FlowMessaging>()
+
+    private val sessionAlice = mock<FlowSession>()
+    private val sessionBob = mock<FlowSession>()
+    private val sessions = listOf(sessionAlice, sessionBob)
+
+    private val transaction = mock<UtxoSignedTransactionInternal>()
+    private val successMessage = "Successfully received transaction."
+
+    @BeforeEach
+    fun beforeEach() {
+        whenever(transaction.id).thenReturn(TX_ID)
+        whenever(flowEngine.subFlow(any<TransactionBackchainSenderFlow>())).thenReturn(Unit)
+    }
+
+    @Test
+    fun `does nothing when receiving payload successfully`() {
+        whenever(sessionAlice.receive(Payload::class.java)).thenReturn(
+            Payload.Success(successMessage)
+        )
+        whenever(sessionBob.receive(Payload::class.java)).thenReturn(
+            Payload.Success(successMessage)
+        )
+
+        callSendTransactionFlow(transaction, sessions)
+
+        verify(flowMessaging).sendAll(transaction.wireTransaction, sessions.toSet())
+        verify(sessionAlice).receive(Payload::class.java)
+    }
+
+    @Test
+    fun `sending transaction with dependencies should call backchain flow`() {
+        whenever(transaction.inputStateRefs).thenReturn(listOf(mock()))
+
+        whenever(sessionAlice.receive(Payload::class.java)).thenReturn(
+            Payload.Success(successMessage)
+        )
+        whenever(sessionBob.receive(Payload::class.java)).thenReturn(
+            Payload.Success(successMessage)
+        )
+
+        callSendTransactionFlow(transaction, sessions)
+
+        verify(flowEngine).subFlow(TransactionBackchainSenderFlow(TX_ID, sessionAlice))
+        verify(flowEngine).subFlow(TransactionBackchainSenderFlow(TX_ID, sessionBob))
+    }
+
+    @Test
+    fun `sending transaction with no dependencies should not call backchain flow`() {
+        whenever(sessionAlice.receive(Payload::class.java)).thenReturn(
+            Payload.Success(successMessage)
+        )
+        whenever(sessionBob.receive(Payload::class.java)).thenReturn(
+            Payload.Success(successMessage)
+        )
+
+        callSendTransactionFlow(transaction, sessions)
+
+        verify(flowEngine, never()).subFlow(TransactionBackchainSenderFlow(TX_ID, sessionAlice))
+    }
+
+    @Test
+    fun `exceptions get propagated back from receiver`() {
+        whenever(sessionAlice.receive(Payload::class.java)).thenReturn(
+            Payload.Failure<List<DigitalSignatureAndMetadata>>("fail")
+        )
+
+        assertThatThrownBy { callSendTransactionFlow(transaction, sessions) }
+            .isInstanceOf(CordaRuntimeException::class.java)
+            .hasMessageContaining("fail")
+    }
+
+    private fun callSendTransactionFlow(signedTransaction: UtxoSignedTransaction, sessions: List<FlowSession>) {
+        val flow = spy(SendAsLedgerTransactionFlow(signedTransaction, sessions))
+
+        flow.flowEngine = flowEngine
+        flow.flowMessaging = flowMessaging
+        flow.call()
+    }
+}

gradle.properties

@@ -44,7 +44,7 @@ commonsLangVersion = 3.12.0
 commonsTextVersion = 1.10.0
 # Corda API libs revision (change in 4th digit indicates a breaking change)
 # Change to 5.2.0.xx-SNAPSHOT to pick up maven local published copy
-cordaApiVersion=5.2.0.26-beta+
+cordaApiVersion=5.2.0.27-beta+
 
 disruptorVersion=3.4.4
 felixConfigAdminVersion=1.9.26