Replace data vending service with SendTransactionFlow (#964)

* WIP - Removed data Vending services, fixed all flow test

* * separated out extra data, extra data are sent after the SendTransactionFlow if required
* New SendProposalFlow for sending TradeProposal, which contains StateAndRef.
* WIP

* * removed TradeProposal interface.
* changed SendProposalFlow to SendStateAndRefFlow, same for receive side.
* fixup after rebase.

* * undo changes in .idea folder

* * remove unintended changes

* * Addressed PR issues

* * doc changes

* * addressed pr issues
* moved ResolveTransactionsFlow to internal
* changed FlowLogic<Unit> to FlowLogic<Void?> for java use case

* * addressed PR issues
* renamed DataVendingFlow in TestUtill to TestDataVendingFlow to avoid name confusion, and moved it to core/test

* * removed reference to ResolveTransactionsFlow

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

@@ -27,7 +27,7 @@ abstract class AbstractStateReplacementFlow {
      * @param M the type of a class representing proposed modification by the instigator.
      */
     @CordaSerializable
-    data class Proposal<out M>(val stateRef: StateRef, val modification: M, val stx: SignedTransaction)
+    data class Proposal<out M>(val stateRef: StateRef, val modification: M)
 
     /**
      * The assembled transaction for upgrading a contract.
@@ -114,9 +114,9 @@ abstract class AbstractStateReplacementFlow {
 
         @Suspendable
         private fun getParticipantSignature(party: Party, stx: SignedTransaction): DigitalSignature.WithKey {
-            val proposal = Proposal(originalState.ref, modification, stx)
+            val proposal = Proposal(originalState.ref, modification)
-            val response = sendAndReceive<DigitalSignature.WithKey>(party, proposal)
+            subFlow(SendTransactionFlow(party, stx))
-            return response.unwrap {
+            return sendAndReceive<DigitalSignature.WithKey>(party, proposal).unwrap {
                 check(party.owningKey.isFulfilledBy(it.by)) { "Not signed by the required participant" }
                 it.verify(stx.id)
                 it
@@ -149,24 +149,17 @@ abstract class AbstractStateReplacementFlow {
         @Throws(StateReplacementException::class)
         override fun call(): Void? {
             progressTracker.currentStep = VERIFYING
+            // We expect stx to have insufficient signatures here
+            val stx = subFlow(ReceiveTransactionFlow(otherSide, checkSufficientSignatures = false))
+            checkMySignatureRequired(stx)
             val maybeProposal: UntrustworthyData<Proposal<T>> = receive(otherSide)
-            val stx: SignedTransaction = maybeProposal.unwrap {
+            maybeProposal.unwrap {
-                verifyProposal(it)
+                verifyProposal(stx, it)
-                verifyTx(it.stx)
-                it.stx
             }
             approve(stx)
             return null
         }
 
-        @Suspendable
-        private fun verifyTx(stx: SignedTransaction) {
-            checkMySignatureRequired(stx)
-            checkDependenciesValid(stx)
-            // We expect stx to have insufficient signatures here
-            stx.verify(serviceHub, checkSufficientSignatures = false)
-        }
-
         @Suspendable
         private fun approve(stx: SignedTransaction) {
             progressTracker.currentStep = APPROVING
@@ -190,12 +183,12 @@ abstract class AbstractStateReplacementFlow {
         }
 
         /**
-         * Check the state change proposal to confirm that it's acceptable to this node. Rules for verification depend
+         * Check the state change proposal and the signed transaction to confirm that it's acceptable to this node.
-         * on the change proposed, and may further depend on the node itself (for example configuration). The
+         * Rules for verification depend on the change proposed, and may further depend on the node itself (for example configuration).
-         * proposal is returned if acceptable, otherwise a [StateReplacementException] is thrown.
+         * The proposal is returned if acceptable, otherwise a [StateReplacementException] is thrown.
          */
         @Throws(StateReplacementException::class)
-        abstract protected fun verifyProposal(proposal: Proposal<T>)
+        abstract protected fun verifyProposal(stx: SignedTransaction, proposal: Proposal<T>)
 
         private fun checkMySignatureRequired(stx: SignedTransaction) {
             // TODO: use keys from the keyManagementService instead
@@ -210,11 +203,6 @@ abstract class AbstractStateReplacementFlow {
             require(myKey in requiredKeys) { "Party is not a participant for any of the input states of transaction ${stx.id}" }
         }
 
-        @Suspendable
-        private fun checkDependenciesValid(stx: SignedTransaction) {
-            subFlow(ResolveTransactionsFlow(stx, otherSide))
-        }
-
         private fun sign(stx: SignedTransaction): DigitalSignature.WithKey {
             return serviceHub.createSignature(stx)
         }

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

@@ -2,7 +2,6 @@ package net.corda.core.flows
 
 import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.identity.Party
-import net.corda.core.serialization.CordaSerializable
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.NonEmptySet
 
@@ -18,16 +17,12 @@ import net.corda.core.utilities.NonEmptySet
 @InitiatingFlow
 class BroadcastTransactionFlow(val notarisedTransaction: SignedTransaction,
                                val participants: NonEmptySet<Party>) : FlowLogic<Unit>() {
-    @CordaSerializable
-    data class NotifyTxRequest(val tx: SignedTransaction)
-
     @Suspendable
     override fun call() {
         // TODO: Messaging layer should handle this broadcast for us
-        val msg = NotifyTxRequest(notarisedTransaction)
         participants.filter { it != serviceHub.myInfo.legalIdentity }.forEach { participant ->
-            // This pops out the other side in NotifyTransactionHandler
+            // SendTransactionFlow allows otherParty to access our data to resolve the transaction.
-            send(participant, msg)
+            subFlow(SendTransactionFlow(participant, notarisedTransaction))
         }
     }
 }

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

@@ -125,7 +125,9 @@ class CollectSignaturesFlow(val partiallySignedTx: SignedTransaction,
      * Get and check the required signature.
      */
     @Suspendable private fun collectSignature(counterparty: Party): DigitalSignature.WithKey {
-        return sendAndReceive<DigitalSignature.WithKey>(counterparty, partiallySignedTx).unwrap {
+        // SendTransactionFlow allows otherParty to access our data to resolve the transaction.
+        subFlow(SendTransactionFlow(counterparty, partiallySignedTx))
+        return receive<DigitalSignature.WithKey>(counterparty).unwrap {
             require(counterparty.owningKey.isFulfilledBy(it.by)) { "Not signed by the required Party." }
             it
         }
@@ -185,35 +187,30 @@ abstract class SignTransactionFlow(val otherParty: Party,
 
     @Suspendable override fun call(): SignedTransaction {
         progressTracker.currentStep = RECEIVING
-        val checkedProposal = receive<SignedTransaction>(otherParty).unwrap { proposal ->
+        // Receive transaction and resolve dependencies, check sufficient signatures is disabled as we don't have all signatures.
+        val stx = subFlow(ReceiveTransactionFlow(otherParty, checkSufficientSignatures = false))
         progressTracker.currentStep = VERIFYING
         // Check that the Responder actually needs to sign.
-            checkMySignatureRequired(proposal)
+        checkMySignatureRequired(stx)
         // Check the signatures which have already been provided. Usually the Initiators and possibly an Oracle's.
-            checkSignatures(proposal)
+        checkSignatures(stx)
-            // Resolve dependencies and verify, pass in the WireTransaction as we don't have all signatures.
+        stx.tx.toLedgerTransaction(serviceHub).verify()
-            subFlow(ResolveTransactionsFlow(proposal, otherParty))
-            proposal.tx.toLedgerTransaction(serviceHub).verify()
         // Perform some custom verification over the transaction.
         try {
-                checkTransaction(proposal)
+            checkTransaction(stx)
         } catch(e: Exception) {
             if (e is IllegalStateException || e is IllegalArgumentException || e is AssertionError)
                 throw FlowException(e)
             else
                 throw e
         }
-            // All good. Unwrap the proposal.
-            proposal
-        }
-
         // Sign and send back our signature to the Initiator.
         progressTracker.currentStep = SIGNING
-        val mySignature = serviceHub.createSignature(checkedProposal)
+        val mySignature = serviceHub.createSignature(stx)
         send(otherParty, mySignature)
 
         // Return the fully signed transaction once it has been committed.
-        return waitForLedgerCommit(checkedProposal.id)
+        return waitForLedgerCommit(stx.id)
     }
 
     @Suspendable private fun checkSignatures(stx: SignedTransaction) {

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

@@ -1,38 +0,0 @@
-package net.corda.core.flows
-
-import net.corda.core.contracts.AbstractAttachment
-import net.corda.core.contracts.Attachment
-import net.corda.core.crypto.SecureHash
-import net.corda.core.crypto.sha256
-import net.corda.core.identity.Party
-import net.corda.core.serialization.SerializationToken
-import net.corda.core.serialization.SerializeAsToken
-import net.corda.core.serialization.SerializeAsTokenContext
-
-/**
- * Given a set of hashes either loads from from local storage  or requests them from the other peer. Downloaded
- * attachments are saved to local storage automatically.
- */
-@InitiatingFlow
-class FetchAttachmentsFlow(requests: Set<SecureHash>,
-                           otherSide: Party) : FetchDataFlow<Attachment, ByteArray>(requests, otherSide, ByteArray::class.java) {
-    override fun load(txid: SecureHash): Attachment? = serviceHub.attachments.openAttachment(txid)
-
-    override fun convert(wire: ByteArray): Attachment = FetchedAttachment({ wire })
-
-    override fun maybeWriteToDisk(downloaded: List<Attachment>) {
-        for (attachment in downloaded) {
-            serviceHub.attachments.importAttachment(attachment.open())
-        }
-    }
-
-    private class FetchedAttachment(dataLoader: () -> ByteArray) : AbstractAttachment(dataLoader), SerializeAsToken {
-        override val id: SecureHash by lazy { attachmentData.sha256() }
-
-        private class Token(private val id: SecureHash) : SerializationToken {
-            override fun fromToken(context: SerializeAsTokenContext) = FetchedAttachment(context.attachmentDataLoader(id))
-        }
-
-        override fun toToken(context: SerializeAsTokenContext) = Token(id)
-    }
-}

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

@@ -1,20 +0,0 @@
-package net.corda.core.flows
-
-import net.corda.core.crypto.SecureHash
-import net.corda.core.identity.Party
-import net.corda.core.transactions.SignedTransaction
-
-/**
- * Given a set of tx hashes (IDs), either loads them from local disk or asks the remote peer to provide them.
- *
- * A malicious response in which the data provided by the remote peer does not hash to the requested hash results in
- * [FetchDataFlow.DownloadedVsRequestedDataMismatch] being thrown. If the remote peer doesn't have an entry, it
- * results in a [FetchDataFlow.HashNotFound] exception. Note that returned transactions are not inserted into
- * the database, because it's up to the caller to actually verify the transactions are valid.
- */
-@InitiatingFlow
-class FetchTransactionsFlow(requests: Set<SecureHash>, otherSide: Party) :
-        FetchDataFlow<SignedTransaction, SignedTransaction>(requests, otherSide, SignedTransaction::class.java) {
-
-    override fun load(txid: SecureHash): SignedTransaction? = serviceHub.validatedTransactions.getTransaction(txid)
-}

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

@@ -7,6 +7,7 @@ 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.internal.ResolveTransactionsFlow
 import net.corda.core.node.ServiceHub
 import net.corda.core.transactions.LedgerTransaction
 import net.corda.core.transactions.SignedTransaction

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

@@ -8,6 +8,7 @@ import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.SignedData
 import net.corda.core.crypto.keys
 import net.corda.core.identity.Party
+import net.corda.core.internal.FetchDataFlow
 import net.corda.core.node.services.NotaryService
 import net.corda.core.node.services.TrustedAuthorityNotaryService
 import net.corda.core.node.services.UniquenessProvider
@@ -63,18 +64,18 @@ object NotaryFlow {
                 throw NotaryException(NotaryError.TransactionInvalid(ex))
             }
 
-            val payload: Any = if (serviceHub.networkMapCache.isValidatingNotary(notaryParty)) {
+            val response = try {
-                stx
+                if (serviceHub.networkMapCache.isValidatingNotary(notaryParty)) {
+                    subFlow(SendTransactionWithRetry(notaryParty, stx))
+                    receive<List<DigitalSignature.WithKey>>(notaryParty)
                 } else {
-                if (stx.isNotaryChangeTransaction()) {
+                    val tx: Any = if (stx.isNotaryChangeTransaction()) {
                         stx.notaryChangeTx
                     } else {
                         stx.tx.buildFilteredTransaction(Predicate { it is StateRef || it is TimeWindow })
                     }
+                    sendAndReceiveWithRetry(notaryParty, tx)
                 }
-
-            val response = try {
-                sendAndReceiveWithRetry<List<DigitalSignature.WithKey>>(notaryParty, payload)
             } catch (e: NotaryException) {
                 if (e.error is NotaryError.Conflict) {
                     e.error.conflict.verified()
@@ -150,3 +151,12 @@ sealed class NotaryError {
         override fun toString() = cause.toString()
     }
 }
+
+/**
+ * The [SendTransactionWithRetry] flow is equivalent to [SendTransactionFlow] but using [sendAndReceiveWithRetry]
+ * instead of [sendAndReceive], [SendTransactionWithRetry] is intended to be use by the notary client only.
+ */
+private class SendTransactionWithRetry(otherSide: Party, stx: SignedTransaction) : SendTransactionFlow(otherSide, stx) {
+    @Suspendable
+    override fun sendPayloadAndReceiveDataRequest(otherSide: Party, payload: Any) = sendAndReceiveWithRetry<FetchDataFlow.Request>(otherSide, payload)
+}

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

@@ -0,0 +1,48 @@
+package net.corda.core.flows
+
+import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.contracts.*
+import net.corda.core.identity.Party
+import net.corda.core.internal.ResolveTransactionsFlow
+import net.corda.core.transactions.SignedTransaction
+import net.corda.core.utilities.unwrap
+import java.security.SignatureException
+
+/**
+ * The [ReceiveTransactionFlow] should be called in response to the [SendTransactionFlow].
+ *
+ * This flow is a combination of [receive], resolve and [SignedTransaction.verify]. This flow will receive the [SignedTransaction]
+ * and perform the resolution back-and-forth required to check the dependencies and download any missing attachments.
+ * The flow will return the [SignedTransaction] after it is resolved and then verified using [SignedTransaction.verify].
+ */
+class ReceiveTransactionFlow
+@JvmOverloads
+constructor(private val otherParty: Party, private val checkSufficientSignatures: Boolean = true) : FlowLogic<SignedTransaction>() {
+    @Suspendable
+    @Throws(SignatureException::class, AttachmentResolutionException::class, TransactionResolutionException::class, TransactionVerificationException::class)
+    override fun call(): SignedTransaction {
+        return receive<SignedTransaction>(otherParty).unwrap {
+            subFlow(ResolveTransactionsFlow(it, otherParty))
+            it.verify(serviceHub, checkSufficientSignatures)
+            it
+        }
+    }
+}
+
+/**
+ * The [ReceiveStateAndRefFlow] should be called in response to the [SendStateAndRefFlow].
+ *
+ * This flow is a combination of [receive] and resolve. This flow will receive a list of [StateAndRef]
+ * and perform the resolution back-and-forth required to check the dependencies.
+ * The flow will return the list of [StateAndRef] after it is resolved.
+ */
+// @JvmSuppressWildcards is used to suppress wildcards in return type when calling `subFlow(new ReceiveStateAndRef<T>(otherParty))` in java.
+class ReceiveStateAndRefFlow<out T : ContractState>(private val otherParty: Party) : FlowLogic<@JvmSuppressWildcards List<StateAndRef<T>>>() {
+    @Suspendable
+    override fun call(): List<StateAndRef<T>> {
+        return receive<List<StateAndRef<T>>>(otherParty).unwrap {
+            subFlow(ResolveTransactionsFlow(it.map { it.ref.txhash }.toSet(), otherParty))
+            it
+        }
+    }
+}

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

@@ -0,0 +1,69 @@
+package net.corda.core.flows
+
+import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.contracts.StateAndRef
+import net.corda.core.identity.Party
+import net.corda.core.internal.FetchDataFlow
+import net.corda.core.transactions.SignedTransaction
+import net.corda.core.utilities.unwrap
+
+/**
+ * The [SendTransactionFlow] should be used to send a transaction to another peer that wishes to verify that transaction's
+ * integrity by resolving and checking the dependencies as well. The other side should invoke [ReceiveTransactionFlow] at
+ * the right point in the conversation to receive the sent transaction and perform the resolution back-and-forth required
+ * to check the dependencies and download any missing attachments.
+ *
+ * @param otherSide the target party.
+ * @param stx the [SignedTransaction] being sent to the [otherSide].
+ */
+open class SendTransactionFlow(otherSide: Party, stx: SignedTransaction) : DataVendingFlow(otherSide, stx)
+
+/**
+ * The [SendStateAndRefFlow] should be used to send a list of input [StateAndRef] to another peer that wishes to verify
+ * the input's integrity by resolving and checking the dependencies as well. The other side should invoke [ReceiveStateAndRefFlow]
+ * at the right point in the conversation to receive the input state and ref and perform the resolution back-and-forth
+ * required to check the dependencies.
+ *
+ * @param otherSide the target party.
+ * @param stateAndRefs the list of [StateAndRef] being sent to the [otherSide].
+ */
+open class SendStateAndRefFlow(otherSide: Party, stateAndRefs: List<StateAndRef<*>>) : DataVendingFlow(otherSide, stateAndRefs)
+
+sealed class DataVendingFlow(val otherSide: Party, val payload: Any) : FlowLogic<Void?>() {
+    @Suspendable
+    protected open fun sendPayloadAndReceiveDataRequest(otherSide: Party, payload: Any) = sendAndReceive<FetchDataFlow.Request>(otherSide, payload)
+
+    @Suspendable
+    protected open fun verifyDataRequest(dataRequest: FetchDataFlow.Request.Data) {
+        // User can override this method to perform custom request verification.
+    }
+
+    @Suspendable
+    override fun call(): Void? {
+        // The first payload will be the transaction data, subsequent payload will be the transaction/attachment data.
+        var payload = payload
+        // This loop will receive [FetchDataFlow.Request] continuously until the `otherSide` has all the data they need
+        // to resolve the transaction, a [FetchDataFlow.EndRequest] will be sent from the `otherSide` to indicate end of
+        // data request.
+        while (true) {
+            val dataRequest = sendPayloadAndReceiveDataRequest(otherSide, payload).unwrap { request ->
+                when (request) {
+                    is FetchDataFlow.Request.Data -> {
+                        // Security TODO: Check for abnormally large or malformed data requests
+                        verifyDataRequest(request)
+                        request
+                    }
+                    FetchDataFlow.Request.End -> return null
+                }
+            }
+            payload = when (dataRequest.dataType) {
+                FetchDataFlow.DataType.TRANSACTION -> dataRequest.hashes.map {
+                    serviceHub.validatedTransactions.getTransaction(it) ?: throw FetchDataFlow.HashNotFound(it)
+                }
+                FetchDataFlow.DataType.ATTACHMENT -> dataRequest.hashes.map {
+                    serviceHub.attachments.openAttachment(it)?.open()?.readBytes() ?: throw FetchDataFlow.HashNotFound(it)
+                }
+            }
+        }
+    }
+}

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

@@ -1,12 +1,22 @@
-package net.corda.core.flows
+package net.corda.core.internal
 
 import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.contracts.AbstractAttachment
+import net.corda.core.contracts.Attachment
 import net.corda.core.contracts.NamedByHash
 import net.corda.core.crypto.SecureHash
-import net.corda.core.flows.FetchDataFlow.DownloadedVsRequestedDataMismatch
+import net.corda.core.crypto.sha256
-import net.corda.core.flows.FetchDataFlow.HashNotFound
+import net.corda.core.flows.FlowException
+import net.corda.core.flows.FlowLogic
 import net.corda.core.identity.Party
+import net.corda.core.internal.FetchDataFlow.DownloadedVsRequestedDataMismatch
+import net.corda.core.internal.FetchDataFlow.HashNotFound
 import net.corda.core.serialization.CordaSerializable
+import net.corda.core.serialization.SerializationToken
+import net.corda.core.serialization.SerializeAsToken
+import net.corda.core.serialization.SerializeAsTokenContext
+import net.corda.core.transactions.SignedTransaction
+import net.corda.core.utilities.NonEmptySet
 import net.corda.core.utilities.UntrustworthyData
 import net.corda.core.utilities.unwrap
 import java.util.*
@@ -27,10 +37,10 @@ import java.util.*
  * @param T The ultimate type of the data being fetched.
  * @param W The wire type of the data being fetched, for when it isn't the same as the ultimate type.
  */
-abstract class FetchDataFlow<T : NamedByHash, W : Any>(
+sealed class FetchDataFlow<T : NamedByHash, in W : Any>(
         protected val requests: Set<SecureHash>,
         protected val otherSide: Party,
-        protected val wrapperType: Class<W>) : FlowLogic<FetchDataFlow.Result<T>>() {
+        protected val dataType: DataType) : FlowLogic<FetchDataFlow.Result<T>>() {
 
     @CordaSerializable
     class DownloadedVsRequestedDataMismatch(val requested: SecureHash, val got: SecureHash) : IllegalArgumentException()
@@ -41,10 +51,18 @@ abstract class FetchDataFlow<T : NamedByHash, W : Any>(
     class HashNotFound(val requested: SecureHash) : FlowException()
 
     @CordaSerializable
-    data class Request(val hashes: List<SecureHash>)
+    data class Result<out T : NamedByHash>(val fromDisk: List<T>, val downloaded: List<T>)
 
     @CordaSerializable
-    data class Result<out T : NamedByHash>(val fromDisk: List<T>, val downloaded: List<T>)
+    sealed class Request {
+        data class Data(val hashes: NonEmptySet<SecureHash>, val dataType: DataType) : Request()
+        object End : Request()
+    }
+
+    @CordaSerializable
+    enum class DataType {
+        TRANSACTION, ATTACHMENT
+    }
 
     @Suspendable
     @Throws(HashNotFound::class)
@@ -65,11 +83,10 @@ abstract class FetchDataFlow<T : NamedByHash, W : Any>(
             // Above that, we start losing authentication data on the message fragments and take exceptions in the
             // network layer.
             val maybeItems = ArrayList<W>(toFetch.size)
-            send(otherSide, Request(toFetch))
             for (hash in toFetch) {
                 // We skip the validation here (with unwrap { it }) because we will do it below in validateFetchResponse.
                 // The only thing checked is the object type. It is a protocol violation to send results out of order.
-                maybeItems += receive(wrapperType, otherSide).unwrap { it }
+                maybeItems += sendAndReceive<List<W>>(otherSide, Request.Data(NonEmptySet.of(hash), dataType)).unwrap { it }
             }
             // Check for a buggy/malicious peer answering with something that we didn't ask for.
             val downloaded = validateFetchResponse(UntrustworthyData(maybeItems), toFetch)
@@ -117,3 +134,46 @@ abstract class FetchDataFlow<T : NamedByHash, W : Any>(
         }
     }
 }
+
+
+/**
+ * Given a set of hashes either loads from from local storage  or requests them from the other peer. Downloaded
+ * attachments are saved to local storage automatically.
+ */
+class FetchAttachmentsFlow(requests: Set<SecureHash>,
+                           otherSide: Party) : FetchDataFlow<Attachment, ByteArray>(requests, otherSide, DataType.ATTACHMENT) {
+
+    override fun load(txid: SecureHash): Attachment? = serviceHub.attachments.openAttachment(txid)
+
+    override fun convert(wire: ByteArray): Attachment = FetchedAttachment({ wire })
+
+    override fun maybeWriteToDisk(downloaded: List<Attachment>) {
+        for (attachment in downloaded) {
+            serviceHub.attachments.importAttachment(attachment.open())
+        }
+    }
+
+    private class FetchedAttachment(dataLoader: () -> ByteArray) : AbstractAttachment(dataLoader), SerializeAsToken {
+        override val id: SecureHash by lazy { attachmentData.sha256() }
+
+        private class Token(private val id: SecureHash) : SerializationToken {
+            override fun fromToken(context: SerializeAsTokenContext) = FetchedAttachment(context.attachmentDataLoader(id))
+        }
+
+        override fun toToken(context: SerializeAsTokenContext) = Token(id)
+    }
+}
+
+/**
+ * Given a set of tx hashes (IDs), either loads them from local disk or asks the remote peer to provide them.
+ *
+ * A malicious response in which the data provided by the remote peer does not hash to the requested hash results in
+ * [FetchDataFlow.DownloadedVsRequestedDataMismatch] being thrown. If the remote peer doesn't have an entry, it
+ * results in a [FetchDataFlow.HashNotFound] exception. Note that returned transactions are not inserted into
+ * the database, because it's up to the caller to actually verify the transactions are valid.
+ */
+class FetchTransactionsFlow(requests: Set<SecureHash>, otherSide: Party) :
+        FetchDataFlow<SignedTransaction, SignedTransaction>(requests, otherSide, DataType.TRANSACTION) {
+
+    override fun load(txid: SecureHash): SignedTransaction? = serviceHub.validatedTransactions.getTransaction(txid)
+}

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

@@ -1,11 +1,12 @@
-package net.corda.core.flows
+package net.corda.core.internal
 
 import co.paralleluniverse.fibers.Suspendable
-import net.corda.core.utilities.exactAdd
 import net.corda.core.crypto.SecureHash
+import net.corda.core.flows.FlowLogic
 import net.corda.core.identity.Party
 import net.corda.core.serialization.CordaSerializable
 import net.corda.core.transactions.SignedTransaction
+import net.corda.core.utilities.exactAdd
 import java.util.*
 
 // TODO: This code is currently unit tested by TwoPartyTradeFlowTests, it should have its own tests.
@@ -70,36 +71,37 @@ class ResolveTransactionsFlow(private val txHashes: Set<SecureHash>,
     // TODO: Figure out a more appropriate DOS limit here, 5000 is simply a very bad guess.
     /** The maximum number of transactions this flow will try to download before bailing out. */
     var transactionCountLimit = 5000
+        set(value) {
+            require(value > 0) { "$value is not a valid count limit" }
+            field = value
+        }
 
     @Suspendable
     @Throws(FetchDataFlow.HashNotFound::class)
     override fun call(): List<SignedTransaction> {
-        val newTxns: Iterable<SignedTransaction> = topologicalSort(downloadDependencies(txHashes))
+        // Start fetching data.
+        val newTxns = downloadDependencies(txHashes)
+        fetchMissingAttachments(signedTransaction?.let { newTxns + it } ?: newTxns)
+        send(otherSide, FetchDataFlow.Request.End)
+        // Finish fetching data.
 
+        val result = topologicalSort(newTxns)
+        result.forEach {
             // For each transaction, verify it and insert it into the database. As we are iterating over them in a
             // depth-first order, we should not encounter any verification failures due to missing data. If we fail
             // half way through, it's no big deal, although it might result in us attempting to re-download data
             // redundantly next time we attempt verification.
-        val result = ArrayList<SignedTransaction>()
+            it.verify(serviceHub)
-
+            serviceHub.recordTransactions(it)
-        for (stx in newTxns) {
-            // TODO: We could recover some parallelism from the dependency graph.
-            stx.verify(serviceHub)
-            serviceHub.recordTransactions(stx)
-            result += stx
         }
 
-        // If this flow is resolving a specific transaction, make sure we have its attachments as well
+        return signedTransaction?.let {
-        signedTransaction?.let {
+            result + it
-            fetchMissingAttachments(listOf(it))
+        } ?: result
-            result += it
-        }
-
-        return result
     }
 
     @Suspendable
-    private fun downloadDependencies(depsToCheck: Set<SecureHash>): Collection<SignedTransaction> {
+    private fun downloadDependencies(depsToCheck: Set<SecureHash>): List<SignedTransaction> {
         // Maintain a work queue of all hashes to load/download, initialised with our starting set. Then do a breadth
         // first traversal across the dependency graph.
         //
@@ -121,7 +123,6 @@ class ResolveTransactionsFlow(private val txHashes: Set<SecureHash>,
         val resultQ = LinkedHashMap<SecureHash, SignedTransaction>()
 
         val limit = transactionCountLimit
-        check(limit > 0) { "$limit is not a valid count limit" }
         var limitCounter = 0
         while (nextRequests.isNotEmpty()) {
             // Don't re-download the same tx when we haven't verified it yet but it's referenced multiple times in the
@@ -135,8 +136,6 @@ class ResolveTransactionsFlow(private val txHashes: Set<SecureHash>,
             // Request the standalone transaction data (which may refer to things we don't yet have).
             val downloads: List<SignedTransaction> = subFlow(FetchTransactionsFlow(notAlreadyFetched, otherSide)).downloaded
 
-            fetchMissingAttachments(downloads)
-
             for (stx in downloads)
                 check(resultQ.putIfAbsent(stx.id, stx) == null)   // Assert checks the filter at the start.
 
@@ -148,8 +147,7 @@ class ResolveTransactionsFlow(private val txHashes: Set<SecureHash>,
             if (limitCounter > limit)
                 throw ExcessivelyLargeTransactionGraph()
         }
-
+        return resultQ.values.toList()
-        return resultQ.values
     }
 
     /**

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

@@ -1,13 +1,15 @@
-package net.corda.node.messaging
+package net.corda.core.flows
 
+import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.contracts.Attachment
 import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.sha256
 import net.corda.core.getOrThrow
+import net.corda.core.identity.Party
+import net.corda.core.internal.FetchAttachmentsFlow
+import net.corda.core.internal.FetchDataFlow
 import net.corda.core.messaging.SingleMessageRecipient
 import net.corda.core.node.services.ServiceInfo
-import net.corda.core.flows.FetchAttachmentsFlow
-import net.corda.core.flows.FetchDataFlow
 import net.corda.node.services.config.NodeConfiguration
 import net.corda.node.services.database.RequeryConfiguration
 import net.corda.node.services.network.NetworkMapService
@@ -59,6 +61,8 @@ class AttachmentTests {
         val nodes = mockNet.createSomeNodes(2)
         val n0 = nodes.partyNodes[0]
         val n1 = nodes.partyNodes[1]
+        n0.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
+        n1.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
 
         // Insert an attachment into node zero's store directly.
         val id = n0.database.transaction {
@@ -67,7 +71,7 @@ class AttachmentTests {
 
         // Get node one to run a flow to fetch it and insert it.
         mockNet.runNetwork()
-        val f1 = n1.services.startFlow(FetchAttachmentsFlow(setOf(id), n0.info.legalIdentity))
+        val f1 = n1.startAttachmentFlow(setOf(id), n0.info.legalIdentity)
         mockNet.runNetwork()
         assertEquals(0, f1.resultFuture.getOrThrow().fromDisk.size)
 
@@ -81,7 +85,7 @@ class AttachmentTests {
         // Shut down node zero and ensure node one can still resolve the attachment.
         n0.stop()
 
-        val response: FetchDataFlow.Result<Attachment> = n1.services.startFlow(FetchAttachmentsFlow(setOf(id), n0.info.legalIdentity)).resultFuture.getOrThrow()
+        val response: FetchDataFlow.Result<Attachment> = n1.startAttachmentFlow(setOf(id), n0.info.legalIdentity).resultFuture.getOrThrow()
         assertEquals(attachment, response.fromDisk[0])
     }
 
@@ -90,11 +94,13 @@ class AttachmentTests {
         val nodes = mockNet.createSomeNodes(2)
         val n0 = nodes.partyNodes[0]
         val n1 = nodes.partyNodes[1]
+        n0.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
+        n1.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
 
         // Get node one to fetch a non-existent attachment.
         val hash = SecureHash.randomSHA256()
         mockNet.runNetwork()
-        val f1 = n1.services.startFlow(FetchAttachmentsFlow(setOf(hash), n0.info.legalIdentity))
+        val f1 = n1.startAttachmentFlow(setOf(hash), n0.info.legalIdentity)
         mockNet.runNetwork()
         val e = assertFailsWith<FetchDataFlow.HashNotFound> { f1.resultFuture.getOrThrow() }
         assertEquals(hash, e.requested)
@@ -118,6 +124,9 @@ class AttachmentTests {
         }, advertisedServices = *arrayOf(ServiceInfo(NetworkMapService.type), ServiceInfo(SimpleNotaryService.type)))
         val n1 = mockNet.createNode(n0.network.myAddress)
 
+        n0.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
+        n1.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
+
         val attachment = fakeAttachment()
         // Insert an attachment into node zero's store directly.
         val id = n0.database.transaction {
@@ -135,11 +144,24 @@ class AttachmentTests {
             n0.attachments.session.update(corruptAttachment)
         }
 
-
         // Get n1 to fetch the attachment. Should receive corrupted bytes.
         mockNet.runNetwork()
-        val f1 = n1.services.startFlow(FetchAttachmentsFlow(setOf(id), n0.info.legalIdentity))
+        val f1 = n1.startAttachmentFlow(setOf(id), n0.info.legalIdentity)
         mockNet.runNetwork()
         assertFailsWith<FetchDataFlow.DownloadedVsRequestedDataMismatch> { f1.resultFuture.getOrThrow() }
     }
+
+    private fun MockNetwork.MockNode.startAttachmentFlow(hashes: Set<SecureHash>, otherSide: Party) = services.startFlow(InitiatingFetchAttachmentsFlow(otherSide, hashes))
+
+    @InitiatingFlow
+    private class InitiatingFetchAttachmentsFlow(val otherSide: Party, val hashes: Set<SecureHash>) : FlowLogic<FetchDataFlow.Result<Attachment>>() {
+        @Suspendable
+        override fun call(): FetchDataFlow.Result<Attachment> = subFlow(FetchAttachmentsFlow(hashes, otherSide))
+    }
+
+    @InitiatedBy(InitiatingFetchAttachmentsFlow::class)
+    private class FetchAttachmentsResponse(val otherSide: Party) : FlowLogic<Void?>() {
+        @Suspendable
+        override fun call() = subFlow(TestDataVendingFlow(otherSide))
+    }
 }

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

@@ -0,0 +1,19 @@
+package net.corda.core.flows
+
+import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.identity.Party
+import net.corda.core.internal.FetchDataFlow
+import net.corda.core.utilities.UntrustworthyData
+
+// Flow to start data vending without sending transaction. For testing only.
+class TestDataVendingFlow(otherSide: Party) : SendStateAndRefFlow(otherSide, emptyList()) {
+    @Suspendable
+    override fun sendPayloadAndReceiveDataRequest(otherSide: Party, payload: Any): UntrustworthyData<FetchDataFlow.Request> {
+        return if (payload is List<*> && payload.isEmpty()) {
+            // Hack to not send the first message.
+            receive(otherSide)
+        } else {
+            super.sendPayloadAndReceiveDataRequest(otherSide, payload)
+        }
+    }
+}

core/src/test/kotlin/net/corda/core/internal/ResolveTransactionsFlowTest.kt

@@ -1,6 +1,11 @@
-package net.corda.core.flows
+package net.corda.core.internal
 
+import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.crypto.SecureHash
+import net.corda.core.flows.FlowLogic
+import net.corda.core.flows.InitiatedBy
+import net.corda.core.flows.InitiatingFlow
+import net.corda.core.flows.TestDataVendingFlow
 import net.corda.core.getOrThrow
 import net.corda.core.identity.Party
 import net.corda.core.transactions.SignedTransaction
@@ -38,6 +43,8 @@ class ResolveTransactionsFlowTest {
         val nodes = mockNet.createSomeNodes()
         a = nodes.partyNodes[0]
         b = nodes.partyNodes[1]
+        a.registerInitiatedFlow(TestResponseFlow::class.java)
+        b.registerInitiatedFlow(TestResponseFlow::class.java)
         notary = nodes.notaryNode.info.notaryIdentity
         mockNet.runNetwork()
     }
@@ -51,7 +58,7 @@ class ResolveTransactionsFlowTest {
     @Test
     fun `resolve from two hashes`() {
         val (stx1, stx2) = makeTransactions()
-        val p = ResolveTransactionsFlow(setOf(stx2.id), a.info.legalIdentity)
+        val p = TestFlow(setOf(stx2.id), a.info.legalIdentity)
         val future = b.services.startFlow(p).resultFuture
         mockNet.runNetwork()
         val results = future.getOrThrow()
@@ -66,7 +73,7 @@ class ResolveTransactionsFlowTest {
     @Test
     fun `dependency with an error`() {
         val stx = makeTransactions(signFirstTX = false).second
-        val p = ResolveTransactionsFlow(setOf(stx.id), a.info.legalIdentity)
+        val p = TestFlow(setOf(stx.id), a.info.legalIdentity)
         val future = b.services.startFlow(p).resultFuture
         mockNet.runNetwork()
         assertFailsWith(SignedTransaction.SignaturesMissingException::class) { future.getOrThrow() }
@@ -75,7 +82,7 @@ class ResolveTransactionsFlowTest {
     @Test
     fun `resolve from a signed transaction`() {
         val (stx1, stx2) = makeTransactions()
-        val p = ResolveTransactionsFlow(stx2, a.info.legalIdentity)
+        val p = TestFlow(stx2, a.info.legalIdentity)
         val future = b.services.startFlow(p).resultFuture
         mockNet.runNetwork()
         future.getOrThrow()
@@ -100,8 +107,7 @@ class ResolveTransactionsFlowTest {
             }
             cursor = stx
         }
-        val p = ResolveTransactionsFlow(setOf(cursor.id), a.info.legalIdentity)
+        val p = TestFlow(setOf(cursor.id), a.info.legalIdentity, 40)
-        p.transactionCountLimit = 40
         val future = b.services.startFlow(p).resultFuture
         mockNet.runNetwork()
         assertFailsWith<ResolveTransactionsFlow.ExcessivelyLargeTransactionGraph> { future.getOrThrow() }
@@ -125,7 +131,7 @@ class ResolveTransactionsFlowTest {
             a.services.recordTransactions(stx2, stx3)
         }
 
-        val p = ResolveTransactionsFlow(setOf(stx3.id), a.info.legalIdentity)
+        val p = TestFlow(setOf(stx3.id), a.info.legalIdentity)
         val future = b.services.startFlow(p).resultFuture
         mockNet.runNetwork()
         future.getOrThrow()
@@ -147,7 +153,7 @@ class ResolveTransactionsFlowTest {
             a.services.attachments.importAttachment(makeJar())
         }
         val stx2 = makeTransactions(withAttachment = id).second
-        val p = ResolveTransactionsFlow(stx2, a.info.legalIdentity)
+        val p = TestFlow(stx2, a.info.legalIdentity)
         val future = b.services.startFlow(p).resultFuture
         mockNet.runNetwork()
         future.getOrThrow()
@@ -184,4 +190,22 @@ class ResolveTransactionsFlowTest {
         return Pair(dummy1, dummy2)
     }
     // DOCEND 2
+
+    @InitiatingFlow
+    private class TestFlow(private val resolveTransactionsFlow: ResolveTransactionsFlow, private val txCountLimit: Int? = null) : FlowLogic<List<SignedTransaction>>() {
+        constructor(txHashes: Set<SecureHash>, otherSide: Party, txCountLimit: Int? = null) : this(ResolveTransactionsFlow(txHashes, otherSide), txCountLimit = txCountLimit)
+        constructor(stx: SignedTransaction, otherSide: Party) : this(ResolveTransactionsFlow(stx, otherSide))
+
+        @Suspendable
+        override fun call(): List<SignedTransaction> {
+            txCountLimit?.let { resolveTransactionsFlow.transactionCountLimit = it }
+            return subFlow(resolveTransactionsFlow)
+        }
+    }
+
+    @InitiatedBy(TestFlow::class)
+    private class TestResponseFlow(val otherSide: Party) : FlowLogic<Void?>() {
+        @Suspendable
+        override fun call() = subFlow(TestDataVendingFlow(otherSide))
+    }
 }

core/src/test/kotlin/net/corda/core/serialization/AttachmentSerializationTest.kt

@@ -7,17 +7,20 @@ import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.InitiatingFlow
 import net.corda.core.getOrThrow
 import net.corda.core.identity.Party
+import net.corda.core.internal.FetchAttachmentsFlow
+import net.corda.core.internal.FetchDataFlow
 import net.corda.core.messaging.RPCOps
 import net.corda.core.messaging.SingleMessageRecipient
 import net.corda.core.node.services.ServiceInfo
 import net.corda.core.utilities.unwrap
-import net.corda.core.flows.FetchAttachmentsFlow
 import net.corda.node.internal.InitiatedFlowFactory
 import net.corda.node.services.config.NodeConfiguration
 import net.corda.node.services.network.NetworkMapService
 import net.corda.node.services.persistence.NodeAttachmentService
 import net.corda.node.services.persistence.schemas.requery.AttachmentEntity
 import net.corda.node.services.statemachine.SessionInit
+import net.corda.core.flows.DataVendingFlow
+import net.corda.core.flows.TestDataVendingFlow
 import net.corda.testing.node.MockNetwork
 import org.junit.After
 import org.junit.Before
@@ -81,9 +84,12 @@ class AttachmentSerializationTest {
         mockNet.stopNodes()
     }
 
-    private class ServerLogic(private val client: Party) : FlowLogic<Unit>() {
+    private class ServerLogic(private val client: Party, private val sendData: Boolean) : FlowLogic<Unit>() {
         @Suspendable
         override fun call() {
+            if (sendData) {
+                subFlow(TestDataVendingFlow(client))
+            }
             receive<String>(client).unwrap { assertEquals("ping one", it) }
             sendAndReceive<String>(client, "pong").unwrap { assertEquals("ping two", it) }
         }
@@ -134,15 +140,16 @@ class AttachmentSerializationTest {
         @Suspendable
         override fun getAttachmentContent(): String {
             val (downloadedAttachment) = subFlow(FetchAttachmentsFlow(setOf(attachmentId), server)).downloaded
+            send(server, FetchDataFlow.Request.End)
             communicate()
             return downloadedAttachment.extractContent()
         }
     }
 
-    private fun launchFlow(clientLogic: ClientLogic, rounds: Int) {
+    private fun launchFlow(clientLogic: ClientLogic, rounds: Int, sendData: Boolean = false) {
         server.internalRegisterFlowFactory(ClientLogic::class.java, object : InitiatedFlowFactory<ServerLogic> {
             override fun createFlow(platformVersion: Int, otherParty: Party, sessionInit: SessionInit): ServerLogic {
-                return ServerLogic(otherParty)
+                return ServerLogic(otherParty, sendData)
             }
         }, ServerLogic::class.java, track = false)
         client.services.startFlow(clientLogic)
@@ -191,7 +198,7 @@ class AttachmentSerializationTest {
     @Test
     fun `only the hash of a FetchAttachmentsFlow attachment should be saved in checkpoint`() {
         val attachmentId = server.saveAttachment("genuine")
-        launchFlow(FetchAttachmentLogic(server, attachmentId), 2)
+        launchFlow(FetchAttachmentLogic(server, attachmentId), 2, sendData = true)
         client.hackAttachment(attachmentId, "hacked")
         assertEquals("hacked", rebootClientAndGetAttachmentContent(false))
     }

docs/source/api-flows.rst

@@ -396,7 +396,8 @@ Corda provides a number of built-in flows that should be used for handling commo
 
 * ``CollectSignaturesFlow``, which should be used to collect a transaction's required signatures
 * ``FinalityFlow``, which should be used to notarise and record a transaction
-* ``ResolveTransactionsFlow``, which should be used to verify the chain of inputs to a transaction
+* ``SendTransactionFlow``, which should be used to send a signed transaction if it needed to be resolved on the other side.
+* ``ReceiveTransactionFlow``, which should be used receive a signed transaction
 * ``ContractUpgradeFlow``, which should be used to change a state's contract
 * ``NotaryChangeFlow``, which should be used to change a state's notary
 
@@ -478,11 +479,29 @@ transaction and provide their signature if they are satisfied:
         :end-before: DOCEND 16
         :dedent: 12
 
-ResolveTransactionsFlow
+SendTransactionFlow/ReceiveTransactionFlow
-^^^^^^^^^^^^^^^^^^^^^^^
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-Verifying a transaction will also verify every transaction in the transaction's dependency chain. So if we receive a
+Verifying a transaction received from a counterparty also requires verification of every transaction in its
-transaction from a counterparty and it has any dependencies, we'd need to download all of these dependencies
+dependency chain. This means the receiving party needs to be able to ask the sender all the details of the chain.
-using``ResolveTransactionsFlow`` before verifying it:
+The sender will use ``SendTransactionFlow`` for sending the transaction and then for processing all subsequent
+transaction data vending requests as the receiver walks the dependency chain using ``ReceiveTransactionFlow``:
+
+.. container:: codeset
+
+    .. literalinclude:: ../../docs/source/example-code/src/main/kotlin/net/corda/docs/FlowCookbook.kt
+        :language: kotlin
+        :start-after: DOCSTART 12
+        :end-before: DOCEND 12
+        :dedent: 12
+
+    .. literalinclude:: ../../docs/source/example-code/src/main/java/net/corda/docs/FlowCookbookJava.java
+        :language: java
+        :start-after: DOCSTART 12
+        :end-before: DOCEND 12
+        :dedent: 12
+
+We can receive the transaction using ``ReceiveTransactionFlow``, which will automatically download all the
+dependencies and verify the transaction:
 
 .. container:: codeset
 
@@ -498,7 +517,7 @@ using``ResolveTransactionsFlow`` before verifying it:
         :end-before: DOCEND 13
         :dedent: 12
 
-We can also resolve a `StateRef` dependency chain:
+We can also send and receive a ``StateAndRef`` dependency chain and automatically resolve its dependencies:
 
 .. container:: codeset
 

docs/source/api-transactions.rst

@@ -359,7 +359,7 @@ Verifying the transaction's contents
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 To verify a transaction, we need to retrieve any states in the transaction chain that our node doesn't
 currently have in its local storage from the proposer(s) of the transaction. This process is handled by a built-in flow
-called ``ResolveTransactionsFlow``. See :doc:`api-flows` for more details.
+called ``ReceiveTransactionFlow``. See :doc:`api-flows` for more details.
 
 When verifying a ``SignedTransaction``, we don't verify the ``SignedTransaction`` *per se*, but rather the
 ``WireTransaction`` it contains. We extract this ``WireTransaction`` as follows:

docs/source/changelog.rst

@@ -72,6 +72,13 @@ UNRELEASED
     * ``stateMachineRecordedTransactionMapping``, replaced by ``stateMachineRecordedTransactionMappingFeed``
     * ``networkMapUpdates``, replaced by ``networkMapFeed``
 
+* Due to security concerns and the need to remove the concept of state relevancy (which isn't needed in Corda),
+  ``ResolveTransactionsFlow`` has been made internal. Instead merge the receipt of the ``SignedTransaction`` and the subsequent
+  sub-flow call to ``ResolveTransactionsFlow`` with a single call to ``ReceiveTransactionFlow``. The flow running on the counterparty
+  must use ``SendTransactionFlow`` at the correct place. There is also ``ReceiveStateAndRefFlow`` and ``SendStateAndRefFlow`` for
+  dealing with ``StateAndRef``s.
+
+
 Milestone 13
 ------------
 

docs/source/example-code/src/main/java/net/corda/docs/FlowCookbookJava.java

@@ -9,6 +9,7 @@ import net.corda.core.crypto.DigitalSignature;
 import net.corda.core.crypto.SecureHash;
 import net.corda.core.flows.*;
 import net.corda.core.identity.Party;
+import net.corda.core.internal.FetchDataFlow;
 import net.corda.core.node.services.ServiceType;
 import net.corda.core.node.services.Vault;
 import net.corda.core.node.services.Vault.Page;
@@ -23,11 +24,13 @@ import net.corda.core.utilities.UntrustworthyData;
 import net.corda.testing.contracts.DummyContract;
 import net.corda.testing.contracts.DummyState;
 import org.bouncycastle.asn1.x500.X500Name;
+import org.jetbrains.annotations.NotNull;
 
 import java.security.PublicKey;
 import java.security.SignatureException;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 
@@ -75,13 +78,15 @@ public class FlowCookbookJava {
         private static final Step SIGS_GATHERING = new Step("Gathering a transaction's signatures.") {
             // Wiring up a child progress tracker allows us to see the
             // subflow's progress steps in our flow's progress tracker.
-            @Override public ProgressTracker childProgressTracker() {
+            @Override
+            public ProgressTracker childProgressTracker() {
                 return CollectSignaturesFlow.Companion.tracker();
             }
         };
         private static final Step VERIFYING_SIGS = new Step("Verifying a transaction's signatures.");
         private static final Step FINALISATION = new Step("Finalising a transaction.") {
-            @Override public ProgressTracker childProgressTracker() {
+            @Override
+            public ProgressTracker childProgressTracker() {
                 return FinalityFlow.Companion.tracker();
             }
         };
@@ -390,18 +395,35 @@ public class FlowCookbookJava {
             ----------------------------*/
             progressTracker.setCurrentStep(TX_VERIFICATION);
 
-            // Verifying a transaction will also verify every transaction in
+            // Verifying a transaction will also verify every transaction in the transaction's dependency chain, which will require
-            // the transaction's dependency chain. So if this was a
+            // transaction data access on counterparty's node. The ``SendTransactionFlow`` can be used to automate the sending
-            // transaction we'd received from a counterparty and it had any
+            // and data vending process. The ``SendTransactionFlow`` will listen for data request until the transaction
-            // dependencies, we'd need to download all of these dependencies
+            // is resolved and verified on the other side:
-            // using``ResolveTransactionsFlow`` before verifying it.
+            // DOCSTART 12
+            subFlow(new SendTransactionFlow(counterparty, twiceSignedTx));
+
+            // Optional request verification to further restrict data access.
+            subFlow(new SendTransactionFlow(counterparty, twiceSignedTx){
+                @Override
+                protected void verifyDataRequest(@NotNull FetchDataFlow.Request.Data dataRequest) {
+                    // Extra request verification.
+                }
+            });
+            // DOCEND 12
+
+            // We can receive the transaction using ``ReceiveTransactionFlow``,
+            // which will automatically download all the dependencies and verify
+            // the transaction
             // DOCSTART 13
-            subFlow(new ResolveTransactionsFlow(twiceSignedTx, counterparty));
+            SignedTransaction verifiedTransaction = subFlow(new ReceiveTransactionFlow(counterparty));
             // DOCEND 13
 
-            // We can also resolve a `StateRef` dependency chain.
+            // We can also send and receive a `StateAndRef` dependency chain and automatically resolve its dependencies.
             // DOCSTART 14
-            subFlow(new ResolveTransactionsFlow(ImmutableSet.of(ourStateRef.getTxhash()), counterparty));
+            subFlow(new SendStateAndRefFlow(counterparty, dummyStates));
+
+            // On the receive side ...
+            List<StateAndRef<DummyState>> resolvedStateAndRef = subFlow(new ReceiveStateAndRefFlow<DummyState>(counterparty));
             // DOCEND 14
 
             // A ``SignedTransaction`` is a pairing of a ``WireTransaction``

docs/source/example-code/src/main/kotlin/net/corda/docs/CustomNotaryTutorial.kt

@@ -44,12 +44,18 @@ class MyValidatingNotaryFlow(otherSide: Party, service: MyCustomValidatingNotary
      */
     @Suspendable
     override fun receiveAndVerifyTx(): TransactionParts {
-        val stx = receive<SignedTransaction>(otherSide).unwrap { it }
+        try {
+            val stx = subFlow(ReceiveTransactionFlow(otherSide, checkSufficientSignatures = false))
             checkSignatures(stx)
-        resolveTransaction(stx)
-        validateTransaction(stx)
             val wtx = stx.tx
             return TransactionParts(wtx.id, wtx.inputs, wtx.timeWindow)
+        } catch (e: Exception) {
+            throw when (e) {
+                is TransactionVerificationException,
+                is SignatureException -> NotaryException(NotaryError.TransactionInvalid(e))
+                else -> e
+            }
+        }
     }
 
     fun processTransaction(stx: SignedTransaction) {
@@ -63,22 +69,5 @@ class MyValidatingNotaryFlow(otherSide: Party, service: MyCustomValidatingNotary
             throw NotaryException(NotaryError.TransactionInvalid(e))
         }
     }
-
-    @Suspendable
-    fun validateTransaction(stx: SignedTransaction) {
-        try {
-            resolveTransaction(stx)
-            stx.verify(serviceHub, false)
-        } catch (e: Exception) {
-            throw when (e) {
-                is TransactionVerificationException,
-                is SignatureException -> NotaryException(NotaryError.TransactionInvalid(e))
-                else -> e
-            }
-        }
-    }
-
-    @Suspendable
-    private fun resolveTransaction(stx: SignedTransaction) = subFlow(ResolveTransactionsFlow(stx, otherSide))
 }
 // END 2

docs/source/example-code/src/main/kotlin/net/corda/docs/FlowCookbook.kt

@@ -9,6 +9,7 @@ import net.corda.core.crypto.DigitalSignature
 import net.corda.core.crypto.SecureHash
 import net.corda.core.flows.*
 import net.corda.core.identity.Party
+import net.corda.core.internal.FetchDataFlow
 import net.corda.core.node.services.ServiceType
 import net.corda.core.node.services.Vault.Page
 import net.corda.core.node.services.queryBy
@@ -378,18 +379,34 @@ object FlowCookbook {
             ---------------------------**/
             progressTracker.currentStep = TX_VERIFICATION
 
-            // Verifying a transaction will also verify every transaction in
+            // Verifying a transaction will also verify every transaction in the transaction's dependency chain, which will require
-            // the transaction's dependency chain. So if this was a
+            // transaction data access on counterparty's node. The ``SendTransactionFlow`` can be used to automate the sending
-            // transaction we'd received from a counterparty and it had any
+            // and data vending process. The ``SendTransactionFlow`` will listen for data request until the transaction
-            // dependencies, we'd need to download all of these dependencies
+            // is resolved and verified on the other side:
-            // using``ResolveTransactionsFlow`` before verifying it.
+            // DOCSTART 12
+            subFlow(SendTransactionFlow(counterparty, twiceSignedTx))
+
+            // Optional request verification to further restrict data access.
+            subFlow(object :SendTransactionFlow(counterparty, twiceSignedTx){
+                override fun verifyDataRequest(dataRequest: FetchDataFlow.Request.Data) {
+                    // Extra request verification.
+                }
+            })
+            // DOCEND 12
+
+            // We can receive the transaction using ``ReceiveTransactionFlow``,
+            // which will automatically download all the dependencies and verify
+            // the transaction
             // DOCSTART 13
-            subFlow(ResolveTransactionsFlow(twiceSignedTx, counterparty))
+            val verifiedTransaction = subFlow(ReceiveTransactionFlow(counterparty))
             // DOCEND 13
 
-            // We can also resolve a `StateRef` dependency chain.
+            // We can also send and receive a `StateAndRef` dependency chain and automatically resolve its dependencies.
             // DOCSTART 14
-            subFlow(ResolveTransactionsFlow(setOf(ourStateRef.txhash), counterparty))
+            subFlow(SendStateAndRefFlow(counterparty, dummyStates))
+
+            // On the receive side ...
+            val resolvedStateAndRef = subFlow(ReceiveStateAndRefFlow<DummyState>(counterparty))
             // DOCEND 14
 
             // A ``SignedTransaction`` is a pairing of a ``WireTransaction``

docs/source/example-code/src/main/kotlin/net/corda/docs/FxTransactionBuildTutorial.kt

@@ -27,10 +27,6 @@ private data class FxRequest(val tradeId: String,
                              val counterparty: Party,
                              val notary: Party? = null)
 
-@CordaSerializable
-private data class FxResponse(val inputs: List<StateAndRef<Cash.State>>,
-                              val outputs: List<Cash.State>)
-
 // DOCSTART 1
 // This is equivalent to the VaultService.generateSpend
 // Which is brought here to make the filtering logic more visible in the example
@@ -69,7 +65,7 @@ private fun gatherOurInputs(serviceHub: ServiceHub,
 }
 // DOCEND 1
 
-private fun prepareOurInputsAndOutputs(serviceHub: ServiceHub, request: FxRequest): FxResponse {
+private fun prepareOurInputsAndOutputs(serviceHub: ServiceHub, request: FxRequest): Pair<List<StateAndRef<Cash.State>>, List<Cash.State>> {
     // Create amount with correct issuer details
     val sellAmount = request.amount
 
@@ -84,14 +80,15 @@ private fun prepareOurInputsAndOutputs(serviceHub: ServiceHub, request: FxReques
     // Build and an output state for the counterparty
     val transferedFundsOutput = Cash.State(sellAmount, request.counterparty)
 
-    if (residual > 0L) {
+    val outputs = if (residual > 0L) {
         // Build an output state for the residual change back to us
         val residualAmount = Amount(residual, sellAmount.token)
         val residualOutput = Cash.State(residualAmount, serviceHub.myInfo.legalIdentity)
-        return FxResponse(inputs, listOf(transferedFundsOutput, residualOutput))
+        listOf(transferedFundsOutput, residualOutput)
     } else {
-        return FxResponse(inputs, listOf(transferedFundsOutput))
+        listOf(transferedFundsOutput)
     }
+    return Pair(inputs, outputs)
     // DOCEND 2
 }
 
@@ -119,45 +116,45 @@ class ForeignExchangeFlow(val tradeId: String,
         } else throw IllegalArgumentException("Our identity must be one of the parties in the trade.")
 
         // Call the helper method to identify suitable inputs and make the outputs
-        val ourStates = prepareOurInputsAndOutputs(serviceHub, localRequest)
+        val (outInputStates, ourOutputStates) = prepareOurInputsAndOutputs(serviceHub, localRequest)
 
         // identify the notary for our states
-        val notary = ourStates.inputs.first().state.notary
+        val notary = outInputStates.first().state.notary
         // ensure request to other side is for a consistent notary
         val remoteRequestWithNotary = remoteRequest.copy(notary = notary)
 
         // Send the request to the counterparty to verify and call their version of prepareOurInputsAndOutputs
         // Then they can return their candidate states
-        val theirStates = sendAndReceive<FxResponse>(remoteRequestWithNotary.owner, remoteRequestWithNotary).unwrap {
+        send(remoteRequestWithNotary.owner, remoteRequestWithNotary)
-            require(it.inputs.all { it.state.notary == notary }) {
+        val theirInputStates = subFlow(ReceiveStateAndRefFlow<Cash.State>(remoteRequestWithNotary.owner))
+        val theirOutputStates = receive<List<Cash.State>>(remoteRequestWithNotary.owner).unwrap {
+            require(theirInputStates.all { it.state.notary == notary }) {
                 "notary of remote states must be same as for our states"
             }
-            require(it.inputs.all { it.state.data.amount.token == remoteRequestWithNotary.amount.token }) {
+            require(theirInputStates.all { it.state.data.amount.token == remoteRequestWithNotary.amount.token }) {
                 "Inputs not of the correct currency"
             }
-            require(it.outputs.all { it.amount.token == remoteRequestWithNotary.amount.token }) {
+            require(it.all { it.amount.token == remoteRequestWithNotary.amount.token }) {
                 "Outputs not of the correct currency"
             }
-            require(it.inputs.map { it.state.data.amount.quantity }.sum()
+            require(theirInputStates.map { it.state.data.amount.quantity }.sum()
                     >= remoteRequestWithNotary.amount.quantity) {
                 "the provided inputs don't provide sufficient funds"
             }
-            require(it.outputs.filter { it.owner == serviceHub.myInfo.legalIdentity }.
+            require(it.filter { it.owner == serviceHub.myInfo.legalIdentity }.
                     map { it.amount.quantity }.sum() == remoteRequestWithNotary.amount.quantity) {
                 "the provided outputs don't provide the request quantity"
             }
-            // Download their inputs chains to validate that they are OK
-            val dependencyTxIDs = it.inputs.map { it.ref.txhash }.toSet()
-            subFlow(ResolveTransactionsFlow(dependencyTxIDs, remoteRequestWithNotary.owner))
-
             it // return validated response
         }
 
         // having collated the data create the full transaction.
-        val signedTransaction = buildTradeProposal(ourStates, theirStates)
+        val signedTransaction = buildTradeProposal(outInputStates, ourOutputStates, theirInputStates, theirOutputStates)
 
         // pass transaction details to the counterparty to revalidate and confirm with a signature
-        val allPartySignedTx = sendAndReceive<DigitalSignature.WithKey>(remoteRequestWithNotary.owner, signedTransaction).unwrap {
+        // Allow otherParty to access our data to resolve the transaction.
+        subFlow(SendTransactionFlow(remoteRequestWithNotary.owner, signedTransaction))
+        val allPartySignedTx = receive<DigitalSignature.WithKey>(remoteRequestWithNotary.owner).unwrap {
             val withNewSignature = signedTransaction + it
             // check all signatures are present except the notary
             withNewSignature.verifySignaturesExcept(withNewSignature.tx.notary!!.owningKey)
@@ -177,22 +174,25 @@ class ForeignExchangeFlow(val tradeId: String,
     }
 
     // DOCSTART 3
-    private fun buildTradeProposal(ourStates: FxResponse, theirStates: FxResponse): SignedTransaction {
+    private fun buildTradeProposal(ourInputStates: List<StateAndRef<Cash.State>>,
+                                   ourOutputState: List<Cash.State>,
+                                   theirInputStates: List<StateAndRef<Cash.State>>,
+                                   theirOutputState: List<Cash.State>): SignedTransaction {
         // This is the correct way to create a TransactionBuilder,
         // do not construct directly.
         // We also set the notary to match the input notary
-        val builder = TransactionBuilder(ourStates.inputs.first().state.notary)
+        val builder = TransactionBuilder(ourInputStates.first().state.notary)
 
         // Add the move commands and key to indicate all the respective owners and need to sign
-        val ourSigners = ourStates.inputs.map { it.state.data.owner.owningKey }.toSet()
+        val ourSigners = ourInputStates.map { it.state.data.owner.owningKey }.toSet()
-        val theirSigners = theirStates.inputs.map { it.state.data.owner.owningKey }.toSet()
+        val theirSigners = theirInputStates.map { it.state.data.owner.owningKey }.toSet()
         builder.addCommand(Cash.Commands.Move(), (ourSigners + theirSigners).toList())
 
         // Build and add the inputs and outputs
-        builder.withItems(*ourStates.inputs.toTypedArray())
+        builder.withItems(*ourInputStates.toTypedArray())
-        builder.withItems(*theirStates.inputs.toTypedArray())
+        builder.withItems(*theirInputStates.toTypedArray())
-        builder.withItems(*ourStates.outputs.toTypedArray())
+        builder.withItems(*ourOutputState.toTypedArray())
-        builder.withItems(*theirStates.outputs.toTypedArray())
+        builder.withItems(*theirOutputState.toTypedArray())
 
         // We have already validated their response and trust our own data
         // so we can sign. Note the returned SignedTransaction is still not fully signed
@@ -228,23 +228,22 @@ class ForeignExchangeRemoteFlow(val source: Party) : FlowLogic<Unit>() {
         // we will use query manually in the helper function below.
         // Putting this into a non-suspendable function also prevent issues when
         // the flow is suspended.
-        val ourResponse = prepareOurInputsAndOutputs(serviceHub, request)
+        val (ourInputState, ourOutputState) = prepareOurInputsAndOutputs(serviceHub, request)
 
         // Send back our proposed states and await the full transaction to verify
-        val ourKey = serviceHub.keyManagementService.filterMyKeys(ourResponse.inputs.flatMap { it.state.data.participants }.map { it.owningKey }).single()
+        val ourKey = serviceHub.keyManagementService.filterMyKeys(ourInputState.flatMap { it.state.data.participants }.map { it.owningKey }).single()
-        val proposedTrade = sendAndReceive<SignedTransaction>(source, ourResponse).unwrap {
+        // SendStateAndRefFlow allows otherParty to access our transaction data to resolve the transaction.
+        subFlow(SendStateAndRefFlow(source, ourInputState))
+        send(source, ourOutputState)
+        val proposedTrade = subFlow(ReceiveTransactionFlow(source, checkSufficientSignatures = false)).let {
             val wtx = it.tx
             // check all signatures are present except our own and the notary
             it.verifySignaturesExcept(ourKey, wtx.notary!!.owningKey)
 
-            // We need to fetch their complete input states and dependencies so that verify can operate
-            checkDependencies(it)
-
             // This verifies that the transaction is contract-valid, even though it is missing signatures.
             // In a full solution there would be states tracking the trade request which
             // would be included in the transaction and enforce the amounts and tradeId
             wtx.toLedgerTransaction(serviceHub).verify()
-
             it // return the SignedTransaction
         }
 
@@ -256,12 +255,4 @@ class ForeignExchangeRemoteFlow(val source: Party) : FlowLogic<Unit>() {
         // N.B. The FinalityProtocol will be responsible for Notarising the SignedTransaction
         // and broadcasting the result to us.
     }
-
-    @Suspendable
-    private fun checkDependencies(stx: SignedTransaction) {
-        // Download and check all the transactions that this transaction depends on, but do not check this
-        // transaction itself.
-        val dependencyTxIDs = stx.tx.inputs.map { it.txhash }.toSet()
-        subFlow(ResolveTransactionsFlow(dependencyTxIDs, source))
-    }
 }

docs/source/flow-library.rst

@@ -47,17 +47,15 @@ responder side. Types of things you will need to check include:
 
 Typically after calling the ``CollectSignaturesFlow`` you then called the ``FinalityFlow``.
 
-ResolveTransactionsFlow
+SendTransactionFlow/ReceiveTransactionFlow
------------------------
+------------------------------------------
 
-This ``ResolveTransactionsFlow`` is used to verify the validity of a transaction by recursively checking the validity of
+The ``SendTransactionFlow`` and ``ReceiveTransactionFlow`` are used to automate the verification of the transaction by
-all the dependencies. Once a transaction is checked it's inserted into local storage so it can be relayed and won't be
+recursively checking the validity of all the dependencies. Once a transaction is received and checked it's inserted into
-checked again.
+local storage so it can be relayed and won't be checked again.
 
-A couple of constructors are provided that accept a single transaction. When these are used, the dependencies of that
+The ``SendTransactionFlow`` sends the transaction to the counterparty and listen for data request as the counterparty
-transaction are resolved and then the transaction itself is verified. Again, if successful, the results are inserted
+validating the transaction, extra checks can be implemented to restrict data access by overriding the ``verifyDataRequest``
-into the database as long as a [SignedTransaction] was provided. If only the ``WireTransaction`` form was provided
+method inside ``SendTransactionFlow``.
-then this isn't enough to put into the local database, so only the dependencies are checked and inserted. This way
-to use the flow is helpful when resolving and verifying an unfinished transaction.
 
-The flow returns a list of verified ``LedgerTransaction`` objects, in a depth-first order.
+The ``ReceiveTransactionFlow`` returns a verified ``SignedTransaction``.

docs/source/flow-state-machines.rst

@@ -388,7 +388,7 @@ returns the result of the flow's execution directly. Behind the scenes all this
 tracking (discussed more below) and then running the object's ``call`` method. Because the sub-flow might suspend,
 we must mark the method that invokes it as suspendable.
 
-Within FinalityFlow, we use a further sub-flow called ``ResolveTransactionsFlow``. This is responsible for downloading
+Within FinalityFlow, we use a further sub-flow called ``ReceiveTransactionFlow``. This is responsible for downloading
 and checking all the dependencies of a transaction, which in Corda are always retrievable from the party that sent you a
 transaction that uses them. This flow returns a list of ``LedgerTransaction`` objects.
 

docs/source/tutorial-attachments.rst

@@ -39,10 +39,9 @@ a JVM client.
 Protocol
 --------
 
-Normally attachments on transactions are fetched automatically via the ``ResolveTransactionsFlow``. Attachments
+Normally attachments on transactions are fetched automatically via the ``ReceiveTransactionFlow``. Attachments
 are needed in order to validate a transaction (they include, for example, the contract code), so must be fetched
-before the validation process can run. ``ResolveTransactionsFlow`` calls ``FetchTransactionsFlow`` to perform the
+before the validation process can run.
-actual retrieval.
 
 .. note:: Future versions of Corda may support non-critical attachments that are not used for transaction verification
    and which are shared explicitly. These are useful for attaching and signing auditing data with a transaction

finance/src/main/kotlin/net/corda/flows/TwoPartyTradeFlow.kt

@@ -47,7 +47,6 @@ object TwoPartyTradeFlow {
     // This object is serialised to the network and is the first flow message the seller sends to the buyer.
     @CordaSerializable
     data class SellerTradeInfo(
-            val assetForSale: StateAndRef<OwnableState>,
             val price: Amount<Currency>,
             val sellerOwner: AbstractParty
     )
@@ -75,11 +74,12 @@ object TwoPartyTradeFlow {
         override fun call(): SignedTransaction {
             progressTracker.currentStep = AWAITING_PROPOSAL
             // Make the first message we'll send to kick off the flow.
-            val hello = SellerTradeInfo(assetToSell, price, me)
+            val hello = SellerTradeInfo(price, me)
             // What we get back from the other side is a transaction that *might* be valid and acceptable to us,
             // but we must check it out thoroughly before we sign!
+            // SendTransactionFlow allows otherParty to access our data to resolve the transaction.
+            subFlow(SendStateAndRefFlow(otherParty, listOf(assetToSell)))
             send(otherParty, hello)
-
             // Verify and sign the transaction.
             progressTracker.currentStep = VERIFYING_AND_SIGNING
             // DOCSTART 5
@@ -113,11 +113,13 @@ object TwoPartyTradeFlow {
                      val typeToBuy: Class<out OwnableState>) : FlowLogic<SignedTransaction>() {
         // DOCSTART 2
         object RECEIVING : ProgressTracker.Step("Waiting for seller trading info")
+
         object VERIFYING : ProgressTracker.Step("Verifying seller assets")
         object SIGNING : ProgressTracker.Step("Generating and signing transaction proposal")
         object COLLECTING_SIGNATURES : ProgressTracker.Step("Collecting signatures from other parties") {
             override fun childProgressTracker() = CollectSignaturesFlow.tracker()
         }
+
         object RECORDING : ProgressTracker.Step("Recording completed transaction") {
             // TODO: Currently triggers a race condition on Team City. See https://github.com/corda/corda/issues/733.
             // override fun childProgressTracker() = FinalityFlow.tracker()
@@ -131,43 +133,34 @@ object TwoPartyTradeFlow {
         override fun call(): SignedTransaction {
             // Wait for a trade request to come in from the other party.
             progressTracker.currentStep = RECEIVING
-            val tradeRequest = receiveAndValidateTradeRequest()
+            val (assetForSale, tradeRequest) = receiveAndValidateTradeRequest()
 
             // Put together a proposed transaction that performs the trade, and sign it.
             progressTracker.currentStep = SIGNING
-            val (ptx, cashSigningPubKeys) = assembleSharedTX(tradeRequest)
+            val (ptx, cashSigningPubKeys) = assembleSharedTX(assetForSale, tradeRequest)
             val partSignedTx = signWithOurKeys(cashSigningPubKeys, ptx)
 
             // Send the signed transaction to the seller, who must then sign it themselves and commit
             // it to the ledger by sending it to the notary.
             progressTracker.currentStep = COLLECTING_SIGNATURES
             val twiceSignedTx = subFlow(CollectSignaturesFlow(partSignedTx, COLLECTING_SIGNATURES.childProgressTracker()))
-
             // Notarise and record the transaction.
             progressTracker.currentStep = RECORDING
             return subFlow(FinalityFlow(twiceSignedTx)).single()
         }
 
         @Suspendable
-        private fun receiveAndValidateTradeRequest(): SellerTradeInfo {
+        private fun receiveAndValidateTradeRequest(): Pair<StateAndRef<OwnableState>, SellerTradeInfo> {
-            val maybeTradeRequest = receive<SellerTradeInfo>(otherParty)
+            val assetForSale = subFlow(ReceiveStateAndRefFlow<OwnableState>(otherParty)).single()
-
+            return assetForSale to receive<SellerTradeInfo>(otherParty).unwrap {
                 progressTracker.currentStep = VERIFYING
-            maybeTradeRequest.unwrap {
+                val asset = assetForSale.state.data
-                // What is the seller trying to sell us?
-                val asset = it.assetForSale.state.data
                 val assetTypeName = asset.javaClass.name
-
                 if (it.price > acceptablePrice)
                     throw UnacceptablePriceException(it.price)
                 if (!typeToBuy.isInstance(asset))
                     throw AssetMismatchException(typeToBuy.name, assetTypeName)
-
+                it
-                // Check that the state being sold to us is in a valid chain of transactions, i.e. that the
-                // seller has a valid chain of custody proving that they own the thing they're selling.
-                subFlow(ResolveTransactionsFlow(setOf(it.assetForSale.ref.txhash), otherParty))
-
-                return it
             }
         }
 
@@ -177,23 +170,23 @@ object TwoPartyTradeFlow {
         }
 
         @Suspendable
-        private fun assembleSharedTX(tradeRequest: SellerTradeInfo): Pair<TransactionBuilder, List<PublicKey>> {
+        private fun assembleSharedTX(assetForSale: StateAndRef<OwnableState>, tradeRequest: SellerTradeInfo): Pair<TransactionBuilder, List<PublicKey>> {
             val ptx = TransactionBuilder(notary)
 
             // Add input and output states for the movement of cash, by using the Cash contract to generate the states
             val (tx, cashSigningPubKeys) = serviceHub.vaultService.generateSpend(ptx, tradeRequest.price, tradeRequest.sellerOwner)
 
             // Add inputs/outputs/a command for the movement of the asset.
-            tx.addInputState(tradeRequest.assetForSale)
+            tx.addInputState(assetForSale)
 
             // Just pick some new public key for now. This won't be linked with our identity in any way, which is what
             // we want for privacy reasons: the key is here ONLY to manage and control ownership, it is not intended to
             // reveal who the owner actually is. The key management service is expected to derive a unique key from some
             // initial seed in order to provide privacy protection.
             val freshKey = serviceHub.keyManagementService.freshKey()
-            val (command, state) = tradeRequest.assetForSale.state.data.withNewOwner(AnonymousParty(freshKey))
+            val (command, state) = assetForSale.state.data.withNewOwner(AnonymousParty(freshKey))
-            tx.addOutputState(state, tradeRequest.assetForSale.state.notary)
+            tx.addOutputState(state, assetForSale.state.notary)
-            tx.addCommand(command, tradeRequest.assetForSale.state.data.owner.owningKey)
+            tx.addCommand(command, assetForSale.state.data.owner.owningKey)
 
             // We set the transaction's time-window: it may be that none of the contracts need this!
             // But it can't hurt to have one.

finance/src/test/java/net/corda/flows/AbstractStateReplacementFlowTest.java

@@ -2,8 +2,9 @@ package net.corda.flows;
 
 import net.corda.core.flows.AbstractStateReplacementFlow;
 import net.corda.core.identity.Party;
-import net.corda.core.utilities.*;
+import net.corda.core.transactions.SignedTransaction;
-import org.jetbrains.annotations.*;
+import net.corda.core.utilities.ProgressTracker;
+import org.jetbrains.annotations.NotNull;
 
 @SuppressWarnings("unused")
 public class AbstractStateReplacementFlowTest {
@@ -15,7 +16,7 @@ public class AbstractStateReplacementFlowTest {
         }
 
         @Override
-        protected void verifyProposal(@NotNull AbstractStateReplacementFlow.Proposal proposal) {
+        protected void verifyProposal(@NotNull SignedTransaction stx, @NotNull AbstractStateReplacementFlow.Proposal proposal) {
         }
     }
 }

finance/src/test/kotlin/net/corda/flows/BroadcastTransactionFlowTest.kt

@@ -7,9 +7,9 @@ import com.pholser.junit.quickcheck.generator.Generator
 import com.pholser.junit.quickcheck.random.SourceOfRandomness
 import com.pholser.junit.quickcheck.runner.JUnitQuickcheck
 import net.corda.contracts.testing.SignedTransactionGenerator
-import net.corda.core.flows.BroadcastTransactionFlow.NotifyTxRequest
 import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.serialize
+import net.corda.core.transactions.SignedTransaction
 import net.corda.testing.initialiseTestSerialization
 import net.corda.testing.resetTestSerialization
 import org.junit.After
@@ -19,10 +19,10 @@ import kotlin.test.assertEquals
 @RunWith(JUnitQuickcheck::class)
 class BroadcastTransactionFlowTest {
 
-    class NotifyTxRequestMessageGenerator : Generator<NotifyTxRequest>(NotifyTxRequest::class.java) {
+    class NotifyTxRequestMessageGenerator : Generator<SignedTransaction>(SignedTransaction::class.java) {
-        override fun generate(random: SourceOfRandomness, status: GenerationStatus): NotifyTxRequest {
+        override fun generate(random: SourceOfRandomness, status: GenerationStatus): SignedTransaction {
             initialiseTestSerialization()
-            return NotifyTxRequest(tx = SignedTransactionGenerator().generate(random, status))
+            return SignedTransactionGenerator().generate(random, status)
         }
     }
 
@@ -32,9 +32,9 @@ class BroadcastTransactionFlowTest {
     }
 
     @Property
-    fun serialiseDeserialiseOfNotifyMessageWorks(@From(NotifyTxRequestMessageGenerator::class) message: NotifyTxRequest) {
+    fun serialiseDeserialiseOfNotifyMessageWorks(@From(NotifyTxRequestMessageGenerator::class) message: SignedTransaction) {
         val serialized = message.serialize().bytes
-        val deserialized = serialized.deserialize<NotifyTxRequest>()
+        val deserialized = serialized.deserialize<SignedTransaction>()
         assertEquals(deserialized, message)
     }
 }

node/src/integration-test/kotlin/net/corda/node/services/statemachine/LargeTransactionsTest.kt

@@ -34,7 +34,9 @@ class LargeTransactionsTest {
                     .addAttachment(hash4)
             val stx = serviceHub.signInitialTransaction(tx, serviceHub.legalIdentityKey)
             // Send to the other side and wait for it to trigger resolution from us.
-            sendAndReceive<Unit>(serviceHub.networkMapCache.getNodeByLegalName(BOB.name)!!.legalIdentity, stx)
+            val bob = serviceHub.networkMapCache.getNodeByLegalName(BOB.name)!!.legalIdentity
+            subFlow(SendTransactionFlow(bob, stx))
+            receive<Unit>(bob)
         }
     }
 
@@ -42,8 +44,7 @@ class LargeTransactionsTest {
     class ReceiveLargeTransactionFlow(private val counterParty: Party) : FlowLogic<Unit>() {
         @Suspendable
         override fun call() {
-            val stx = receive<SignedTransaction>(counterParty).unwrap { it }
+            subFlow(ReceiveTransactionFlow(counterParty))
-            subFlow(ResolveTransactionsFlow(stx, counterParty))
             // Unblock the other side by sending some dummy object (Unit is fine here as it's a singleton).
             send(counterParty, Unit)
         }

node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt

@@ -401,8 +401,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
     }
 
     private fun installCoreFlows() {
-        installCoreFlow(FetchTransactionsFlow::class) { otherParty, _ -> FetchTransactionsHandler(otherParty) }
-        installCoreFlow(FetchAttachmentsFlow::class) { otherParty, _ -> FetchAttachmentsHandler(otherParty) }
         installCoreFlow(BroadcastTransactionFlow::class) { otherParty, _ -> NotifyTransactionHandler(otherParty) }
         installCoreFlow(NotaryChangeFlow::class) { otherParty, _ -> NotaryChangeHandler(otherParty) }
         installCoreFlow(ContractUpgradeFlow::class) { otherParty, _ -> ContractUpgradeHandler(otherParty) }

node/src/main/kotlin/net/corda/node/services/CoreFlowHandlers.kt

@@ -5,7 +5,6 @@ import net.corda.core.contracts.ContractState
 import net.corda.core.contracts.UpgradeCommand
 import net.corda.core.contracts.UpgradedContract
 import net.corda.core.contracts.requireThat
-import net.corda.core.crypto.SecureHash
 import net.corda.core.flows.*
 import net.corda.core.identity.AnonymousPartyAndPath
 import net.corda.core.identity.Party
@@ -13,50 +12,6 @@ import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.unwrap
 
-/**
- * This class sets up network message handlers for requests from peers for data keyed by hash. It is a piece of simple
- * glue that sits between the network layer and the database layer.
- *
- * Note that in our data model, to be able to name a thing by hash automatically gives the power to request it. There
- * are no access control lists. If you want to keep some data private, then you must be careful who you give its name
- * to, and trust that they will not pass the name onwards. If someone suspects some data might exist but does not have
- * its name, then the 256-bit search space they'd have to cover makes it physically impossible to enumerate, and as
- * such the hash of a piece of data can be seen as a type of password allowing access to it.
- *
- * Additionally, because nodes do not store invalid transactions, requesting such a transaction will always yield null.
- */
-class FetchTransactionsHandler(otherParty: Party) : FetchDataHandler<SignedTransaction>(otherParty) {
-    override fun getData(id: SecureHash): SignedTransaction? {
-        return serviceHub.validatedTransactions.getTransaction(id)
-    }
-}
-
-class FetchAttachmentsHandler(otherParty: Party) : FetchDataHandler<ByteArray>(otherParty) {
-    override fun getData(id: SecureHash): ByteArray? {
-        return serviceHub.attachments.openAttachment(id)?.open()?.readBytes()
-    }
-}
-
-abstract class FetchDataHandler<out T>(val otherParty: Party) : FlowLogic<Unit>() {
-    @Suspendable
-    @Throws(FetchDataFlow.HashNotFound::class)
-    override fun call() {
-        val request = receive<FetchDataFlow.Request>(otherParty).unwrap {
-            if (it.hashes.isEmpty()) throw FlowException("Empty hash list")
-            it
-        }
-        // TODO: Use Artemis message streaming support here, called "large messages". This avoids the need to buffer.
-        // See the discussion in FetchDataFlow. We send each item individually here in a separate asynchronous send
-        // call, and the other side picks them up with a straight receive call, because we batching would push us over
-        // the (current) Artemis message size limit.
-        request.hashes.forEach {
-            send(otherParty, getData(it) ?: throw FetchDataFlow.HashNotFound(it))
-        }
-    }
-
-    protected abstract fun getData(id: SecureHash): T?
-}
-
 // TODO: We should have a whitelist of contracts we're willing to accept at all, and reject if the transaction
 //       includes us in any outside that list. Potentially just if it includes any outside that list at all.
 // TODO: Do we want to be able to reject specific transactions on more complex rules, for example reject incoming
@@ -64,10 +19,8 @@ abstract class FetchDataHandler<out T>(val otherParty: Party) : FlowLogic<Unit>(
 class NotifyTransactionHandler(val otherParty: Party) : FlowLogic<Unit>() {
     @Suspendable
     override fun call() {
-        val request = receive<BroadcastTransactionFlow.NotifyTxRequest>(otherParty).unwrap { it }
+        val stx = subFlow(ReceiveTransactionFlow(otherParty))
-        subFlow(ResolveTransactionsFlow(request.tx, otherParty))
+        serviceHub.recordTransactions(stx)
-        request.tx.verify(serviceHub)
-        serviceHub.recordTransactions(request.tx)
     }
 }
 
@@ -79,9 +32,9 @@ class NotaryChangeHandler(otherSide: Party) : AbstractStateReplacementFlow.Accep
      * and is also in a geographically convenient location we can just automatically approve the change.
      * TODO: In more difficult cases this should call for human attention to manually verify and approve the proposal
      */
-    override fun verifyProposal(proposal: AbstractStateReplacementFlow.Proposal<Party>): Unit {
+    override fun verifyProposal(stx: SignedTransaction, proposal: AbstractStateReplacementFlow.Proposal<Party>): Unit {
         val state = proposal.stateRef
-        val proposedTx = proposal.stx.resolveNotaryChangeTransaction(serviceHub)
+        val proposedTx = stx.resolveNotaryChangeTransaction(serviceHub)
         val newNotary = proposal.modification
 
         if (state !in proposedTx.inputs.map { it.ref }) {
@@ -99,15 +52,15 @@ class NotaryChangeHandler(otherSide: Party) : AbstractStateReplacementFlow.Accep
 class ContractUpgradeHandler(otherSide: Party) : AbstractStateReplacementFlow.Acceptor<Class<out UpgradedContract<ContractState, *>>>(otherSide) {
     @Suspendable
     @Throws(StateReplacementException::class)
-    override fun verifyProposal(proposal: AbstractStateReplacementFlow.Proposal<Class<out UpgradedContract<ContractState, *>>>) {
+    override fun verifyProposal(stx: SignedTransaction, proposal: AbstractStateReplacementFlow.Proposal<Class<out UpgradedContract<ContractState, *>>>) {
         // Retrieve signed transaction from our side, we will apply the upgrade logic to the transaction on our side, and
         // verify outputs matches the proposed upgrade.
-        val stx = subFlow(FetchTransactionsFlow(setOf(proposal.stateRef.txhash), otherSide)).fromDisk.singleOrNull()
+        val ourSTX = serviceHub.validatedTransactions.getTransaction(proposal.stateRef.txhash)
-        requireNotNull(stx) { "We don't have a copy of the referenced state" }
+        requireNotNull(ourSTX) { "We don't have a copy of the referenced state" }
-        val oldStateAndRef = stx!!.tx.outRef<ContractState>(proposal.stateRef.index)
+        val oldStateAndRef = ourSTX!!.tx.outRef<ContractState>(proposal.stateRef.index)
         val authorisedUpgrade = serviceHub.vaultService.getAuthorisedContractUpgrade(oldStateAndRef.ref) ?:
                 throw IllegalStateException("Contract state upgrade is unauthorised. State hash : ${oldStateAndRef.ref}")
-        val proposedTx = proposal.stx.tx
+        val proposedTx = stx.tx
         val expectedTx = ContractUpgradeFlow.assembleBareTx(oldStateAndRef, proposal.modification, proposedTx.privacySalt).toWireTransaction()
         requireThat {
             "The instigator is one of the participants" using (otherSide in oldStateAndRef.state.data.participants)

node/src/main/kotlin/net/corda/node/services/transactions/ValidatingNotaryFlow.kt

@@ -6,8 +6,6 @@ import net.corda.core.flows.*
 import net.corda.core.identity.Party
 import net.corda.core.node.services.TrustedAuthorityNotaryService
 import net.corda.core.transactions.SignedTransaction
-import net.corda.core.transactions.WireTransaction
-import net.corda.core.utilities.unwrap
 import java.security.SignatureException
 
 /**
@@ -24,26 +22,11 @@ class ValidatingNotaryFlow(otherSide: Party, service: TrustedAuthorityNotaryServ
      */
     @Suspendable
     override fun receiveAndVerifyTx(): TransactionParts {
-        val stx = receive<SignedTransaction>(otherSide).unwrap { it }
+        try {
+            val stx = subFlow(ReceiveTransactionFlow(otherSide, checkSufficientSignatures = false))
             checkSignatures(stx)
-        validateTransaction(stx)
             val wtx = stx.tx
             return TransactionParts(wtx.id, wtx.inputs, wtx.timeWindow)
-    }
-
-    private fun checkSignatures(stx: SignedTransaction) {
-        try {
-            stx.verifySignaturesExcept(serviceHub.myInfo.notaryIdentity.owningKey)
-        } catch(e: SignatureException) {
-            throw NotaryException(NotaryError.TransactionInvalid(e))
-        }
-    }
-
-    @Suspendable
-    fun validateTransaction(stx: SignedTransaction) {
-        try {
-            resolveTransaction(stx)
-            stx.verify(serviceHub, false)
         } catch (e: Exception) {
             throw when (e) {
                 is TransactionVerificationException,
@@ -53,6 +36,11 @@ class ValidatingNotaryFlow(otherSide: Party, service: TrustedAuthorityNotaryServ
         }
     }
 
-    @Suspendable
+    private fun checkSignatures(stx: SignedTransaction) {
-    private fun resolveTransaction(stx: SignedTransaction) = subFlow(ResolveTransactionsFlow(stx, otherSide))
+        try {
+            stx.verifySignaturesExcept(serviceHub.myInfo.notaryIdentity.owningKey)
+        } catch(e: SignatureException) {
+            throw NotaryException(NotaryError.TransactionInvalid(e))
+        }
+    }
 }

node/src/test/kotlin/net/corda/node/services/persistence/DataVendingServiceTests.kt

@@ -5,10 +5,10 @@ import net.corda.contracts.asset.Cash
 import net.corda.core.contracts.Amount
 import net.corda.core.contracts.Issued
 import net.corda.core.contracts.USD
-import net.corda.core.flows.BroadcastTransactionFlow.NotifyTxRequest
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.InitiatedBy
 import net.corda.core.flows.InitiatingFlow
+import net.corda.core.flows.SendTransactionFlow
 import net.corda.core.identity.Party
 import net.corda.core.node.services.queryBy
 import net.corda.core.transactions.SignedTransaction
@@ -102,9 +102,9 @@ class DataVendingServiceTests {
     }
 
     @InitiatingFlow
-    private class NotifyTxFlow(val otherParty: Party, val stx: SignedTransaction) : FlowLogic<Unit>() {
+    private class NotifyTxFlow(val otherParty: Party, val stx: SignedTransaction) : FlowLogic<Void?>() {
         @Suspendable
-        override fun call() = send(otherParty, NotifyTxRequest(stx))
+        override fun call() = subFlow(SendTransactionFlow(otherParty, stx))
     }
 
     @InitiatedBy(NotifyTxFlow::class)

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/flows/SimmFlow.kt

@@ -302,8 +302,8 @@ object SimmFlow {
             logger.info("Handshake finished, awaiting Simm update")
             send(replyToParty, Ack) // Hack to state that this party is ready.
             subFlow(object : StateRevisionFlow.Receiver<PortfolioState.Update>(replyToParty) {
-                override fun verifyProposal(proposal: Proposal<PortfolioState.Update>) {
+                override fun verifyProposal(stx:SignedTransaction, proposal: Proposal<PortfolioState.Update>) {
-                    super.verifyProposal(proposal)
+                    super.verifyProposal(stx, proposal)
                     if (proposal.modification.portfolio != portfolio.refs) throw StateReplacementException()
                 }
             })
@@ -315,8 +315,8 @@ object SimmFlow {
             val valuer = serviceHub.identityService.partyFromAnonymous(stateRef.state.data.valuer) ?: throw IllegalStateException("Unknown valuer party ${stateRef.state.data.valuer}")
             val valuation = agreeValuation(portfolio, offer.valuationDate, valuer)
             subFlow(object : StateRevisionFlow.Receiver<PortfolioState.Update>(replyToParty) {
-                override fun verifyProposal(proposal: Proposal<PortfolioState.Update>) {
+                override fun verifyProposal(stx: SignedTransaction, proposal: Proposal<PortfolioState.Update>) {
-                    super.verifyProposal(proposal)
+                    super.verifyProposal(stx, proposal)
                     if (proposal.modification.valuation != valuation) throw StateReplacementException()
                 }
             })

samples/simm-valuation-demo/src/main/kotlin/net/corda/vega/flows/StateRevisionFlow.kt

@@ -5,6 +5,7 @@ import net.corda.core.contracts.StateAndRef
 import net.corda.core.flows.AbstractStateReplacementFlow
 import net.corda.core.flows.StateReplacementException
 import net.corda.core.identity.Party
+import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.seconds
 import net.corda.vega.contracts.RevisionedState
 
@@ -31,8 +32,8 @@ object StateRevisionFlow {
     }
 
     open class Receiver<in T>(otherParty: Party) : AbstractStateReplacementFlow.Acceptor<T>(otherParty) {
-        override fun verifyProposal(proposal: AbstractStateReplacementFlow.Proposal<T>) {
+        override fun verifyProposal(stx: SignedTransaction, proposal: AbstractStateReplacementFlow.Proposal<T>) {
-            val proposedTx = proposal.stx.tx
+            val proposedTx = stx.tx
             val state = proposal.stateRef
             if (state !in proposedTx.inputs) {
                 throw StateReplacementException("The proposed state $state is not in the proposed transaction inputs")