Pick up new implementation of H2CashSelection

perftestcordapp/build.gradle

@@ -5,20 +5,26 @@ apply plugin: 'kotlin-jpa'
 apply plugin: CanonicalizerPlugin
 apply plugin: 'net.corda.plugins.publish-utils'
 apply plugin: 'net.corda.plugins.quasar-utils'
-apply plugin: 'net.corda.plugins.cordformation'
+apply plugin: 'net.corda.plugins.cordapp'
 //apply plugin: 'com.jfrog.artifactory'
 
 description 'Corda performance test modules'
 
 dependencies {
-    // Note the :perftestflows module is a CorDapp in its own right
+    // Note the :finance module is a CorDapp in its own right
-    // and CorDapps using :perftestflows features should use 'cordapp' not 'compile' linkage.
+    // and CorDapps using :finance features should use 'cordapp' not 'compile' linkage.
     cordaCompile project(':core')
     cordaCompile project(':confidential-identities')
 
+    // TODO Remove this once we have app configs
+    compile "com.typesafe:config:$typesafe_config_version"
+
     testCompile project(':test-utils')
     testCompile project(path: ':core', configuration: 'testArtifacts')
     testCompile "junit:junit:$junit_version"
+
+    // AssertJ: for fluent assertions for testing
+    testCompile "org.assertj:assertj-core:$assertj_version"
 }
 
 configurations {

perftestcordapp/src/main/kotlin/com/r3/corda/enterprise/perftestcordapp/contracts/asset/Cash.kt

@@ -17,78 +17,21 @@ import net.corda.core.schemas.PersistentState
 import net.corda.core.schemas.QueryableState
 import net.corda.core.transactions.LedgerTransaction
 import net.corda.core.transactions.TransactionBuilder
-import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.toBase58String
 import com.r3.corda.enterprise.perftestcordapp.schemas.CashSchemaV1
 import com.r3.corda.enterprise.perftestcordapp.utils.sumCash
 import com.r3.corda.enterprise.perftestcordapp.utils.sumCashOrNull
 import com.r3.corda.enterprise.perftestcordapp.utils.sumCashOrZero
+import com.r3.corda.enterprise.perftestcordapp.contracts.asset.cash.selection.AbstractCashSelection
 import java.math.BigInteger
 import java.security.PublicKey
-import java.sql.DatabaseMetaData
 import java.util.*
-import java.util.concurrent.atomic.AtomicReference
 
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 //
 // Cash
 //
 
-/**
- * Pluggable interface to allow for different cash selection provider implementations
- * Default implementation [CashSelectionH2Impl] uses H2 database and a custom function within H2 to perform aggregation.
- * Custom implementations must implement this interface and declare their implementation in
- * META-INF/services/net.corda.contracts.asset.CashSelection
- */
-interface CashSelection {
-    companion object {
-        val instance = AtomicReference<CashSelection>()
-
-        fun getInstance(metadata: () -> java.sql.DatabaseMetaData): CashSelection {
-            return instance.get() ?: {
-                val _metadata = metadata()
-                val cashSelectionAlgos = ServiceLoader.load(CashSelection::class.java).toList()
-                val cashSelectionAlgo = cashSelectionAlgos.firstOrNull { it.isCompatible(_metadata) }
-                cashSelectionAlgo?.let {
-                    instance.set(cashSelectionAlgo)
-                    cashSelectionAlgo
-                } ?: throw ClassNotFoundException("\nUnable to load compatible cash selection algorithm implementation for JDBC driver ($_metadata)." +
-                        "\nPlease specify an implementation in META-INF/services/com.r3.corda.enterprise.perftestcordapp.contracts.asset.CashSelection")
-            }.invoke()
-        }
-    }
-
-    /**
-     * Upon dynamically loading configured Cash Selection algorithms declared in META-INF/services
-     * this method determines whether the loaded implementation is compatible and usable with the currently
-     * loaded JDBC driver.
-     * Note: the first loaded implementation to pass this check will be used at run-time.
-     */
-    fun isCompatible(metadata: DatabaseMetaData): Boolean
-
-    /**
-     * Query to gather Cash states that are available
-     * @param services The service hub to allow access to the database session
-     * @param amount The amount of currency desired (ignoring issues, but specifying the currency)
-     * @param onlyFromIssuerParties If empty the operation ignores the specifics of the issuer,
-     * otherwise the set of eligible states wil be filtered to only include those from these issuers.
-     * @param notary If null the notary source is ignored, if specified then only states marked
-     * with this notary are included.
-     * @param lockId The FlowLogic.runId.uuid of the flow, which is used to soft reserve the states.
-     * Also, previous outputs of the flow will be eligible as they are implicitly locked with this id until the flow completes.
-     * @param withIssuerRefs If not empty the specific set of issuer references to match against.
-     * @return The matching states that were found. If sufficient funds were found these will be locked,
-     * otherwise what is available is returned unlocked for informational purposes.
-     */
-    @Suspendable
-    fun unconsumedCashStatesForSpending(services: ServiceHub,
-                                        amount: Amount<Currency>,
-                                        onlyFromIssuerParties: Set<AbstractParty> = emptySet(),
-                                        notary: Party? = null,
-                                        lockId: UUID,
-                                        withIssuerRefs: Set<OpaqueBytes> = emptySet()): List<StateAndRef<Cash.State>>
-}
-
 /**
  * A cash transaction may split and merge money represented by a set of (issuer, depositRef) pairs, across multiple
  * input and output states. Imagine a Bitcoin transaction but in which all UTXOs had a colour
@@ -103,8 +46,8 @@ interface CashSelection {
  * vaults can ignore the issuer/depositRefs and just examine the amount fields.
  */
 class Cash : OnLedgerAsset<Currency, Cash.Commands, Cash.State>() {
-    override fun extractCommands(commands: Collection<CommandWithParties<CommandData>>): List<CommandWithParties<Commands>>
+    override fun extractCommands(commands: Collection<CommandWithParties<CommandData>>): List<CommandWithParties<Cash.Commands>>
-            = commands.select<Commands>()
+            = commands.select<Cash.Commands>()
 
     // DOCSTART 1
     /** A state representing a cash claim against some party. */
@@ -126,10 +69,10 @@ class Cash : OnLedgerAsset<Currency, Cash.Commands, Cash.State>() {
         override fun toString() = "${Emoji.bagOfCash}Cash($amount at ${amount.token.issuer} owned by $owner)"
 
         override fun withNewOwner(newOwner: AbstractParty) = CommandAndState(Commands.Move(), copy(owner = newOwner))
-        fun ownedBy(owner: AbstractParty) = copy(owner = owner)
+        infix fun ownedBy(owner: AbstractParty) = copy(owner = owner)
-        fun issuedBy(party: AbstractParty) = copy(amount = Amount(amount.quantity, amount.token.copy(issuer = amount.token.issuer.copy(party = party))))
+        infix fun issuedBy(party: AbstractParty) = copy(amount = Amount(amount.quantity, amount.token.copy(issuer = amount.token.issuer.copy(party = party))))
-        fun issuedBy(deposit: PartyAndReference) = copy(amount = Amount(amount.quantity, amount.token.copy(issuer = deposit)))
+        infix fun issuedBy(deposit: PartyAndReference) = copy(amount = Amount(amount.quantity, amount.token.copy(issuer = deposit)))
-        fun withDeposit(deposit: PartyAndReference): State = copy(amount = amount.copy(token = amount.token.copy(issuer = deposit)))
+        infix fun withDeposit(deposit: PartyAndReference): Cash.State = copy(amount = amount.copy(token = amount.token.copy(issuer = deposit)))
 
         /** Object Relational Mapping support. */
         override fun generateMappedObject(schema: MappedSchema): PersistentState {
@@ -196,7 +139,7 @@ class Cash : OnLedgerAsset<Currency, Cash.Commands, Cash.State>() {
     override fun verify(tx: LedgerTransaction) {
         // Each group is a set of input/output states with distinct (reference, currency) attributes. These types
         // of cash are not fungible and must be kept separated for bookkeeping purposes.
-        val groups = tx.groupStates { it: State -> it.amount.token }
+        val groups = tx.groupStates { it: Cash.State -> it.amount.token }
 
         for ((inputs, outputs, key) in groups) {
             // Either inputs or outputs could be empty.
@@ -376,12 +319,12 @@ class Cash : OnLedgerAsset<Currency, Cash.Commands, Cash.State>() {
                           payments: List<PartyAndAmount<Currency>>,
                           ourIdentity: PartyAndCertificate,
                           onlyFromParties: Set<AbstractParty> = emptySet()): Pair<TransactionBuilder, List<PublicKey>> {
-            fun deriveState(txState: TransactionState<State>, amt: Amount<Issued<Currency>>, owner: AbstractParty)
+            fun deriveState(txState: TransactionState<Cash.State>, amt: Amount<Issued<Currency>>, owner: AbstractParty)
                     = txState.copy(data = txState.data.copy(amount = amt, owner = owner))
 
             // Retrieve unspent and unlocked cash states that meet our spending criteria.
             val totalAmount = payments.map { it.amount }.sumOrThrow()
-            val cashSelection = CashSelection.getInstance({ services.jdbcSession().metaData })
+            val cashSelection = AbstractCashSelection.getInstance({ services.jdbcSession().metaData })
             val acceptableCoins = cashSelection.unconsumedCashStatesForSpending(services, totalAmount, onlyFromParties, tx.notary, tx.lockId)
             val revocationEnabled = false // Revocation is currently unsupported
             // Generate a new identity that change will be sent to for confidentiality purposes. This means that a
@@ -396,13 +339,6 @@ class Cash : OnLedgerAsset<Currency, Cash.Commands, Cash.State>() {
     }
 }
 
-// Small DSL extensions.
-
-/** @suppress */ infix fun Cash.State.`owned by`(owner: AbstractParty) = ownedBy(owner)
-/** @suppress */ infix fun Cash.State.`issued by`(party: AbstractParty) = issuedBy(party)
-/** @suppress */ infix fun Cash.State.`issued by`(deposit: PartyAndReference) = issuedBy(deposit)
-/** @suppress */ infix fun Cash.State.`with deposit`(deposit: PartyAndReference): Cash.State = withDeposit(deposit)
-
 // Unit testing helpers. These could go in a separate file but it's hardly worth it for just a few functions.
 
 /** A randomly generated key. */

perftestcordapp/src/main/kotlin/com/r3/corda/enterprise/perftestcordapp/contracts/asset/OnLedgerAsset.kt

@@ -231,12 +231,36 @@ abstract class OnLedgerAsset<T : Any, C : CommandData, S : FungibleAsset<T>> : C
          */
         @Throws(InsufficientBalanceException::class)
         @JvmStatic
+        @Deprecated("Replaced with generateExit() which takes in a party to pay change to")
         fun <S : FungibleAsset<T>, T: Any> generateExit(tx: TransactionBuilder, amountIssued: Amount<Issued<T>>,
                                                         assetStates: List<StateAndRef<S>>,
                                                         deriveState: (TransactionState<S>, Amount<Issued<T>>, AbstractParty) -> TransactionState<S>,
                                                         generateMoveCommand: () -> CommandData,
                                                         generateExitCommand: (Amount<Issued<T>>) -> CommandData): Set<PublicKey> {
-            val owner = assetStates.map { it.state.data.owner }.toSet().singleOrNull() ?: throw InsufficientBalanceException(amountIssued)
+            val owner = assetStates.map { it.state.data.owner }.toSet().firstOrNull() ?: throw InsufficientBalanceException(amountIssued)
+            return generateExit(tx, amountIssued, assetStates, owner, deriveState, generateMoveCommand, generateExitCommand)
+        }
+
+        /**
+         * Generate an transaction exiting fungible assets from the ledger.
+         *
+         * @param tx transaction builder to add states and commands to.
+         * @param amountIssued the amount to be exited, represented as a quantity of issued currency.
+         * @param assetStates the asset states to take funds from. No checks are done about ownership of these states, it is
+         * the responsibility of the caller to check that they do not attempt to exit funds held by others.
+         * @param payChangeTo party to pay any change to; this is normally a confidential identity of the calling
+         * party.
+         * @return the public keys which must sign the transaction for it to be valid.
+         */
+        @Throws(InsufficientBalanceException::class)
+        @JvmStatic
+        fun <S : FungibleAsset<T>, T: Any> generateExit(tx: TransactionBuilder, amountIssued: Amount<Issued<T>>,
+                                                        assetStates: List<StateAndRef<S>>,
+                                                        payChangeTo: AbstractParty,
+                                                        deriveState: (TransactionState<S>, Amount<Issued<T>>, AbstractParty) -> TransactionState<S>,
+                                                        generateMoveCommand: () -> CommandData,
+                                                        generateExitCommand: (Amount<Issued<T>>) -> CommandData): Set<PublicKey> {
+            require(assetStates.isNotEmpty()) { "List of states to exit cannot be empty." }
             val currency = amountIssued.token.product
             val amount = Amount(amountIssued.quantity, currency)
             var acceptableCoins = assetStates.filter { ref -> ref.state.data.amount.token == amountIssued.token }
@@ -256,7 +280,7 @@ abstract class OnLedgerAsset<T : Any, C : CommandData, S : FungibleAsset<T>> : C
 
             val outputs = if (change != null) {
                 // Add a change output and adjust the last output downwards.
-                listOf(deriveState(gathered.last().state, change, owner))
+                listOf(deriveState(gathered.last().state, change, payChangeTo))
             } else emptyList()
 
             for (state in gathered) tx.addInputState(state)
@@ -296,9 +320,12 @@ abstract class OnLedgerAsset<T : Any, C : CommandData, S : FungibleAsset<T>> : C
      * @param amountIssued the amount to be exited, represented as a quantity of issued currency.
      * @param assetStates the asset states to take funds from. No checks are done about ownership of these states, it is
      * the responsibility of the caller to check that they do not exit funds held by others.
+     * @param payChangeTo party to pay any change to; this is normally a confidential identity of the calling
+     * party.
      * @return the public keys which must sign the transaction for it to be valid.
      */
     @Throws(InsufficientBalanceException::class)
+    @Deprecated("Replaced with generateExit() which takes in a party to pay change to")
     fun generateExit(tx: TransactionBuilder, amountIssued: Amount<Issued<T>>,
                      assetStates: List<StateAndRef<S>>): Set<PublicKey> {
         return generateExit(
@@ -311,6 +338,30 @@ abstract class OnLedgerAsset<T : Any, C : CommandData, S : FungibleAsset<T>> : C
         )
     }
 
+    /**
+     * Generate an transaction exiting assets from the ledger.
+     *
+     * @param tx transaction builder to add states and commands to.
+     * @param amountIssued the amount to be exited, represented as a quantity of issued currency.
+     * @param assetStates the asset states to take funds from. No checks are done about ownership of these states, it is
+     * the responsibility of the caller to check that they do not exit funds held by others.
+     * @return the public keys which must sign the transaction for it to be valid.
+     */
+    @Throws(InsufficientBalanceException::class)
+    fun generateExit(tx: TransactionBuilder, amountIssued: Amount<Issued<T>>,
+                     assetStates: List<StateAndRef<S>>,
+                     payChangeTo: AbstractParty): Set<PublicKey> {
+        return generateExit(
+                tx,
+                amountIssued,
+                assetStates,
+                payChangeTo,
+                deriveState = { state, amount, owner -> deriveState(state, amount, owner) },
+                generateMoveCommand = { -> generateMoveCommand() },
+                generateExitCommand = { amount -> generateExitCommand(amount) }
+        )
+    }
+
     abstract fun generateExitCommand(amount: Amount<Issued<T>>): CommandData
     abstract fun generateMoveCommand(): MoveCommand
 

perftestcordapp/src/main/kotlin/com/r3/corda/enterprise/perftestcordapp/contracts/asset/cash/selection/AbstractCashSelection.kt

@@ -0,0 +1,168 @@
+package com.r3.corda.enterprise.perftestcordapp.contracts.asset.cash.selection
+
+import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.contracts.Amount
+import net.corda.core.contracts.StateAndRef
+import net.corda.core.contracts.StateRef
+import net.corda.core.contracts.TransactionState
+import net.corda.core.crypto.SecureHash
+import net.corda.core.flows.FlowLogic
+import net.corda.core.identity.AbstractParty
+import net.corda.core.identity.Party
+import net.corda.core.node.ServiceHub
+import net.corda.core.node.services.StatesNotAvailableException
+import net.corda.core.serialization.SerializationDefaults
+import net.corda.core.serialization.deserialize
+import net.corda.core.utilities.*
+import com.r3.corda.enterprise.perftestcordapp.contracts.asset.Cash
+import java.sql.*
+import java.util.*
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
+
+/**
+ * Pluggable interface to allow for different cash selection provider implementations
+ * Default implementation [CashSelectionH2Impl] uses H2 database and a custom function within H2 to perform aggregation.
+ * Custom implementations must implement this interface and declare their implementation in
+ * META-INF/services/net.corda.contracts.asset.CashSelection
+ */
+abstract class AbstractCashSelection {
+    companion object {
+        val instance = AtomicReference<AbstractCashSelection>()
+
+        fun getInstance(metadata: () -> java.sql.DatabaseMetaData): AbstractCashSelection {
+            return instance.get() ?: {
+                val _metadata = metadata()
+                val cashSelectionAlgos = ServiceLoader.load(AbstractCashSelection::class.java).toList()
+                val cashSelectionAlgo = cashSelectionAlgos.firstOrNull { it.isCompatible(_metadata) }
+                cashSelectionAlgo?.let {
+                    instance.set(cashSelectionAlgo)
+                    cashSelectionAlgo
+                } ?: throw ClassNotFoundException("\nUnable to load compatible cash selection algorithm implementation for JDBC driver ($_metadata)." +
+                        "\nPlease specify an implementation in META-INF/services/${AbstractCashSelection::class.java}")
+            }.invoke()
+        }
+
+        val log = loggerFor<AbstractCashSelection>()
+    }
+
+    // coin selection retry loop counter, sleep (msecs) and lock for selecting states
+    // TODO: make parameters configurable when we get CorDapp configuration.
+    private val MAX_RETRIES = 8
+    private val RETRY_SLEEP = 100
+    private val RETRY_CAP = 2000
+    private val spendLock: ReentrantLock = ReentrantLock()
+
+    /**
+     * Upon dynamically loading configured Cash Selection algorithms declared in META-INF/services
+     * this method determines whether the loaded implementation is compatible and usable with the currently
+     * loaded JDBC driver.
+     * Note: the first loaded implementation to pass this check will be used at run-time.
+     */
+    abstract fun isCompatible(metadata: DatabaseMetaData): Boolean
+
+    /**
+     * A vendor specific query(ies) to gather Cash states that are available.
+     * @param statement The service hub to allow access to the database session
+     * @param amount The amount of currency desired (ignoring issues, but specifying the currency)
+     * @param lockId The FlowLogic.runId.uuid of the flow, which is used to soft reserve the states.
+     * Also, previous outputs of the flow will be eligible as they are implicitly locked with this id until the flow completes.
+     * @param notary If null the notary source is ignored, if specified then only states marked
+     * with this notary are included.
+     * @param onlyFromIssuerParties Optional issuer parties to match against.
+     * @param withIssuerRefs Optional issuer references to match against.
+     * @return JDBC ResultSet with the matching states that were found. If sufficient funds were found these will be locked,
+     * otherwise what is available is returned unlocked for informational purposes.
+     */
+    abstract fun executeQuery(connection: Connection, amount: Amount<Currency>, lockId: UUID, notary: Party?,
+                              onlyFromIssuerParties: Set<AbstractParty>, withIssuerRefs: Set<OpaqueBytes>) : ResultSet
+
+    override abstract fun toString() : String
+
+    /**
+     * Query to gather Cash states that are available and retry if they are temporarily unavailable.
+     * @param services The service hub to allow access to the database session
+     * @param amount The amount of currency desired (ignoring issues, but specifying the currency)
+     * @param onlyFromIssuerParties If empty the operation ignores the specifics of the issuer,
+     * otherwise the set of eligible states wil be filtered to only include those from these issuers.
+     * @param notary If null the notary source is ignored, if specified then only states marked
+     * with this notary are included.
+     * @param lockId The FlowLogic.runId.uuid of the flow, which is used to soft reserve the states.
+     * Also, previous outputs of the flow will be eligible as they are implicitly locked with this id until the flow completes.
+     * @param withIssuerRefs If not empty the specific set of issuer references to match against.
+     * @return The matching states that were found. If sufficient funds were found these will be locked,
+     * otherwise what is available is returned unlocked for informational purposes.
+     */
+    @Suspendable
+    fun unconsumedCashStatesForSpending(services: ServiceHub,
+                                        amount: Amount<Currency>,
+                                        onlyFromIssuerParties: Set<AbstractParty> = emptySet(),
+                                        notary: Party? = null,
+                                        lockId: UUID,
+                                        withIssuerRefs: Set<OpaqueBytes> = emptySet()): List<StateAndRef<Cash.State>> {
+        val stateAndRefs = mutableListOf<StateAndRef<Cash.State>>()
+
+        for (retryCount in 1..MAX_RETRIES) {
+            if (!attemptSpend(services, amount, lockId, notary, onlyFromIssuerParties, withIssuerRefs, stateAndRefs)) {
+                log.warn("Coin selection failed on attempt $retryCount")
+                // TODO: revisit the back off strategy for contended spending.
+                if (retryCount != MAX_RETRIES) {
+                    stateAndRefs.clear()
+                    val durationMillis = (minOf(RETRY_SLEEP.shl(retryCount), RETRY_CAP / 2) * (1.0 + Math.random())).toInt()
+                    FlowLogic.sleep(durationMillis.millis)
+                } else {
+                    log.warn("Insufficient spendable states identified for $amount")
+                }
+            } else {
+                break
+            }
+        }
+        return stateAndRefs
+    }
+
+    private fun attemptSpend(services: ServiceHub, amount: Amount<Currency>, lockId: UUID, notary: Party?, onlyFromIssuerParties: Set<AbstractParty>, withIssuerRefs: Set<OpaqueBytes>, stateAndRefs: MutableList<StateAndRef<Cash.State>>): Boolean {
+        spendLock.withLock {
+            val connection = services.jdbcSession()
+            try {
+                // we select spendable states irrespective of lock but prioritised by unlocked ones (Eg. null)
+                // the softLockReserve update will detect whether we try to lock states locked by others
+                val rs = executeQuery(connection, amount, lockId, notary, onlyFromIssuerParties, withIssuerRefs)
+                stateAndRefs.clear()
+
+                var totalPennies = 0L
+                while (rs.next()) {
+                    val txHash = SecureHash.parse(rs.getString(1))
+                    val index = rs.getInt(2)
+                    val stateRef = StateRef(txHash, index)
+                    val state = rs.getBlob(3).deserialize<TransactionState<Cash.State>>(context = SerializationDefaults.STORAGE_CONTEXT)
+                    val pennies = rs.getLong(4)
+                    totalPennies = rs.getLong(5)
+                    val rowLockId = rs.getString(6)
+                    stateAndRefs.add(StateAndRef(state, stateRef))
+                    log.trace { "ROW: $rowLockId ($lockId): $stateRef : $pennies ($totalPennies)" }
+                }
+
+                if (stateAndRefs.isNotEmpty() && totalPennies >= amount.quantity) {
+                    // we should have a minimum number of states to satisfy our selection `amount` criteria
+                    log.trace("Coin selection for $amount retrieved ${stateAndRefs.count()} states totalling $totalPennies pennies: $stateAndRefs")
+
+                    // With the current single threaded state machine available states are guaranteed to lock.
+                    // TODO However, we will have to revisit these methods in the future multi-threaded.
+                    services.vaultService.softLockReserve(lockId, (stateAndRefs.map { it.ref }).toNonEmptySet())
+                    return true
+                }
+                log.trace("Coin selection requested $amount but retrieved $totalPennies pennies with state refs: ${stateAndRefs.map { it.ref }}")
+                // retry as more states may become available
+            } catch (e: SQLException) {
+                log.error("""Failed retrieving unconsumed states for: amount [$amount], onlyFromIssuerParties [$onlyFromIssuerParties], notary [$notary], lockId [$lockId]
+                            $e.
+                        """)
+            } catch (e: StatesNotAvailableException) { // Should never happen with single threaded state machine
+                log.warn(e.message)
+                // retry only if there are locked states that may become available again (or consumed with change)
+            }
+        }
+        return false
+    }
+}

perftestcordapp/src/main/kotlin/com/r3/corda/enterprise/perftestcordapp/contracts/asset/cash/selection/CashSelectionH2Impl.kt

@@ -1,29 +1,16 @@
 package com.r3.corda.enterprise.perftestcordapp.contracts.asset.cash.selection
 
-
-import co.paralleluniverse.fibers.Suspendable
-import co.paralleluniverse.strands.Strand
-import com.r3.corda.enterprise.perftestcordapp.contracts.asset.Cash
-import com.r3.corda.enterprise.perftestcordapp.contracts.asset.CashSelection
 import net.corda.core.contracts.Amount
-import net.corda.core.contracts.StateAndRef
-import net.corda.core.contracts.StateRef
-import net.corda.core.contracts.TransactionState
-import net.corda.core.crypto.SecureHash
 import net.corda.core.identity.AbstractParty
 import net.corda.core.identity.Party
-import net.corda.core.node.ServiceHub
-import net.corda.core.node.services.StatesNotAvailableException
-import net.corda.core.serialization.SerializationDefaults
-import net.corda.core.serialization.deserialize
 import net.corda.core.utilities.*
+import com.r3.corda.enterprise.perftestcordapp.contracts.asset.cash.selection.AbstractCashSelection
+import java.sql.Connection
 import java.sql.DatabaseMetaData
-import java.sql.SQLException
+import java.sql.ResultSet
 import java.util.*
-import java.util.concurrent.locks.ReentrantLock
-import kotlin.concurrent.withLock
 
-class CashSelectionH2Impl : CashSelection {
+class CashSelectionH2Impl : AbstractCashSelection() {
 
     companion object {
         const val JDBC_DRIVER_NAME = "H2 JDBC Driver"
@@ -34,37 +21,8 @@ class CashSelectionH2Impl : CashSelection {
         return metadata.driverName == JDBC_DRIVER_NAME
     }
 
-    // coin selection retry loop counter, sleep (msecs) and lock for selecting states
+    override fun toString() = "${this::class.java} for $JDBC_DRIVER_NAME"
-    private val MAX_RETRIES = 5
-    private val RETRY_SLEEP = 100
-    private val spendLock: ReentrantLock = ReentrantLock()
 
-    /**
-     * An optimised query to gather Cash states that are available and retry if they are temporarily unavailable.
-     * @param services The service hub to allow access to the database session
-     * @param amount The amount of currency desired (ignoring issues, but specifying the currency)
-     * @param onlyFromIssuerParties If empty the operation ignores the specifics of the issuer,
-     * otherwise the set of eligible states wil be filtered to only include those from these issuers.
-     * @param notary If null the notary source is ignored, if specified then only states marked
-     * with this notary are included.
-     * @param lockId The FlowLogic.runId.uuid of the flow, which is used to soft reserve the states.
-     * Also, previous outputs of the flow will be eligible as they are implicitly locked with this id until the flow completes.
-     * @param withIssuerRefs If not empty the specific set of issuer references to match against.
-     * @return The matching states that were found. If sufficient funds were found these will be locked,
-     * otherwise what is available is returned unlocked for informational purposes.
-     */
-    @Suspendable
-    override fun unconsumedCashStatesForSpending(services: ServiceHub,
-                                                 amount: Amount<Currency>,
-                                                 onlyFromIssuerParties: Set<AbstractParty>,
-                                                 notary: Party?,
-                                                 lockId: UUID,
-                                                 withIssuerRefs: Set<OpaqueBytes>): List<StateAndRef<Cash.State>> {
-
-        val issuerKeysStr = onlyFromIssuerParties.fold("") { left, right -> left + "('${right.owningKey.toBase58String()}')," }.dropLast(1)
-        val issuerRefsStr = withIssuerRefs.fold("") { left, right -> left + "('${right.bytes.toHexString()}')," }.dropLast(1)
-
-        val stateAndRefs = mutableListOf<StateAndRef<Cash.State>>()
 
     //       We are using an H2 specific means of selecting a minimum set of rows that match a request amount of coins:
     //       1) There is no standard SQL mechanism of calculating a cumulative total on a field and restricting row selection on the
@@ -72,80 +30,39 @@ class CashSelectionH2Impl : CashSelection {
     //       2) H2 uses session variables to perform this accumulator function:
     //          http://www.h2database.com/html/functions.html#set
     //       3) H2 does not support JOIN's in FOR UPDATE (hence we are forced to execute 2 queries)
+    override fun executeQuery(connection: Connection, amount: Amount<Currency>, lockId: UUID, notary: Party?,
+                              onlyFromIssuerParties: Set<AbstractParty>, withIssuerRefs: Set<OpaqueBytes>) : ResultSet {
+        connection.createStatement().execute("CALL SET(@t, 0);")
 
-        for (retryCount in 1..MAX_RETRIES) {
-
-            spendLock.withLock {
-                val statement = services.jdbcSession().createStatement()
-                try {
-                    statement.execute("CALL SET(@t, 0);")
-
-                    // we select spendable states irrespective of lock but prioritised by unlocked ones (Eg. null)
-                    // the softLockReserve update will detect whether we try to lock states locked by others
         val selectJoin = """
                     SELECT vs.transaction_id, vs.output_index, vs.contract_state, ccs.pennies, SET(@t, ifnull(@t,0)+ccs.pennies) total_pennies, vs.lock_id
                     FROM vault_states AS vs, contract_pt_cash_states AS ccs
                     WHERE vs.transaction_id = ccs.transaction_id AND vs.output_index = ccs.output_index
                     AND vs.state_status = 0
-                    AND ccs.ccy_code = '${amount.token}' and @t < ${amount.quantity}
+                    AND ccs.ccy_code = ? and @t < ?
-                    AND (vs.lock_id = '$lockId' OR vs.lock_id is null)
+                    AND (vs.lock_id = ? OR vs.lock_id is null)
                     """ +
                 (if (notary != null)
-                                " AND vs.notary_name = '${notary.name}'" else "") +
+                    " AND vs.notary_name = ?" else "") +
                 (if (onlyFromIssuerParties.isNotEmpty())
-                                " AND ccs.issuer_key IN ($issuerKeysStr)" else "") +
+                    " AND ccs.issuer_key IN (?)" else "") +
                 (if (withIssuerRefs.isNotEmpty())
-                                " AND ccs.issuer_ref IN ($issuerRefsStr)" else "")
+                    " AND ccs.issuer_ref IN (?)" else "")
 
-                    // Retrieve spendable state refs
+        // Use prepared statement for protection against SQL Injection (http://www.h2database.com/html/advanced.html#sql_injection)
-                    val rs = statement.executeQuery(selectJoin)
+        val psSelectJoin = connection.prepareStatement(selectJoin)
-                    stateAndRefs.clear()
+        var pIndex = 0
-                    log.debug(selectJoin)
+        psSelectJoin.setString(++pIndex, amount.token.currencyCode)
-                    var totalPennies = 0L
+        psSelectJoin.setLong(++pIndex, amount.quantity)
-                    while (rs.next()) {
+        psSelectJoin.setString(++pIndex, lockId.toString())
-                        val txHash = SecureHash.parse(rs.getString(1))
+        if (notary != null)
-                        val index = rs.getInt(2)
+            psSelectJoin.setString(++pIndex, notary.name.toString())
-                        val stateRef = StateRef(txHash, index)
+        if (onlyFromIssuerParties.isNotEmpty())
-                        val state = rs.getBytes(3).deserialize<TransactionState<Cash.State>>(context = SerializationDefaults.STORAGE_CONTEXT)
+            psSelectJoin.setObject(++pIndex, onlyFromIssuerParties.map { it.owningKey.toBase58String() as Any}.toTypedArray() )
-                        val pennies = rs.getLong(4)
+        if (withIssuerRefs.isNotEmpty())
-                        totalPennies = rs.getLong(5)
+            psSelectJoin.setObject(++pIndex, withIssuerRefs.map { it.bytes.toHexString() as Any }.toTypedArray())
-                        val rowLockId = rs.getString(6)
+        log.debug { psSelectJoin.toString() }
-                        stateAndRefs.add(StateAndRef(state, stateRef))
-                        log.trace { "ROW: $rowLockId ($lockId): $stateRef : $pennies ($totalPennies)" }
-                    }
 
-                    if (stateAndRefs.isNotEmpty() && totalPennies >= amount.quantity) {
+        return psSelectJoin.executeQuery()
-                        // we should have a minimum number of states to satisfy our selection `amount` criteria
-                        log.trace("Coin selection for $amount retrieved ${stateAndRefs.count()} states totalling $totalPennies pennies: $stateAndRefs")
-
-                        // With the current single threaded state machine available states are guaranteed to lock.
-                        // TODO However, we will have to revisit these methods in the future multi-threaded.
-                        services.vaultService.softLockReserve(lockId, (stateAndRefs.map { it.ref }).toNonEmptySet())
-                        return stateAndRefs
-                    }
-                    log.trace("Coin selection requested $amount but retrieved $totalPennies pennies with state refs: ${stateAndRefs.map { it.ref }}")
-                    // retry as more states may become available
-                } catch (e: SQLException) {
-                    log.error("""Failed retrieving unconsumed states for: amount [$amount], onlyFromIssuerParties [$onlyFromIssuerParties], notary [$notary], lockId [$lockId]
-                        $e.
-                    """)
-                } catch (e: StatesNotAvailableException) { // Should never happen with single threaded state machine
-                    stateAndRefs.clear()
-                    log.warn(e.message)
-                    // retry only if there are locked states that may become available again (or consumed with change)
-                } finally {
-                    statement.close()
-                }
-            }
-
-            log.warn("Coin selection failed on attempt $retryCount")
-            // TODO: revisit the back off strategy for contended spending.
-            if (retryCount != MAX_RETRIES) {
-                Strand.sleep(RETRY_SLEEP * retryCount.toLong())
-            }
-        }
-
-        log.warn("Insufficient spendable states identified for $amount")
-        return stateAndRefs
     }
 }

perftestcordapp/src/main/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/AbstractCashFlow.kt

@@ -31,7 +31,7 @@ abstract class AbstractCashFlow<out T>(override val progressTracker: ProgressTra
         try {
             return subFlow(FinalityFlow(tx, extraParticipants))
         } catch (e: NotaryException) {
-            throw PtCashException(message, e)
+            throw CashException(message, e)
         }
     }
 
@@ -50,4 +50,4 @@ abstract class AbstractCashFlow<out T>(override val progressTracker: ProgressTra
     abstract class AbstractRequest(val amount: Amount<Currency>)
 }
 
-class PtCashException(message: String, cause: Throwable) : FlowException(message, cause)
+class CashException(message: String, cause: Throwable) : FlowException(message, cause)

perftestcordapp/src/main/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/CashConfigDataFlow.kt

@@ -1,40 +1,61 @@
 package com.r3.corda.enterprise.perftestcordapp.flows
 
 import co.paralleluniverse.fibers.Suspendable
-import net.corda.core.flows.FlowException
+import com.typesafe.config.ConfigFactory
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.StartableByRPC
+import net.corda.core.internal.declaredField
+import net.corda.core.internal.div
+import net.corda.core.internal.read
+import net.corda.core.node.AppServiceHub
+import net.corda.core.node.services.CordaService
 import net.corda.core.serialization.CordaSerializable
+import net.corda.core.serialization.SingletonSerializeAsToken
 import com.r3.corda.enterprise.perftestcordapp.CHF
 import com.r3.corda.enterprise.perftestcordapp.EUR
 import com.r3.corda.enterprise.perftestcordapp.GBP
 import com.r3.corda.enterprise.perftestcordapp.USD
+import com.r3.corda.enterprise.perftestcordapp.flows.ConfigHolder.Companion.supportedCurrencies
+import java.nio.file.Path
 import java.util.*
 
+// TODO Until apps have access to their own config, we'll hack things by first getting the baseDirectory, read the node.conf
+// again to get our config and store it here for access by our flow
+@CordaService
+class ConfigHolder(services: AppServiceHub) : SingletonSerializeAsToken() {
+    companion object {
+        val supportedCurrencies = listOf(USD, GBP, CHF, EUR)
+    }
+
+    val issuableCurrencies: List<Currency>
+
+    init {
+        // Warning!! You are about to see a major hack!
+        val baseDirectory = services.declaredField<Any>("serviceHub").value
+                .let { it.javaClass.getMethod("getConfiguration").apply { isAccessible = true }.invoke(it) }
+                .declaredField<Path>("baseDirectory").value
+        val config = (baseDirectory / "node.conf").read { ConfigFactory.parseReader(it.reader()) }
+        if (config.hasPath("issuableCurrencies")) {
+            issuableCurrencies = config.getStringList("issuableCurrencies").map { Currency.getInstance(it) }
+            require(supportedCurrencies.containsAll(issuableCurrencies))
+        } else {
+            issuableCurrencies = emptyList()
+        }
+    }
+}
+
+
 /**
  * Flow to obtain cash cordapp app configuration.
  */
 @StartableByRPC
-class CashConfigDataFlow : FlowLogic<PtCashConfiguration>() {
+class CashConfigDataFlow : FlowLogic<CashConfiguration>() {
-    companion object {
-        private val supportedCurrencies = listOf(USD, GBP, CHF, EUR)
-    }
-
     @Suspendable
-    override fun call(): PtCashConfiguration {
+    override fun call(): CashConfiguration {
-        val issuableCurrencies = supportedCurrencies.mapNotNull {
+        val configHolder = serviceHub.cordaService(ConfigHolder::class.java)
-            try {
+        return CashConfiguration(configHolder.issuableCurrencies, supportedCurrencies)
-                // Currently it uses checkFlowPermission to determine the list of issuable currency as a temporary hack.
-                // TODO: get the config from proper configuration source.
-                checkFlowPermission("corda.issuer.$it", emptyMap())
-                it
-            } catch (e: FlowException) {
-                null
-            }
-        }
-        return PtCashConfiguration(issuableCurrencies, supportedCurrencies)
     }
 }
 
 @CordaSerializable
-data class PtCashConfiguration(val issuableCurrencies: List<Currency>, val supportedCurrencies: List<Currency>)
+data class CashConfiguration(val issuableCurrencies: List<Currency>, val supportedCurrencies: List<Currency>)

perftestcordapp/src/main/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/CashExitFlow.kt

@@ -14,7 +14,7 @@ import net.corda.core.transactions.TransactionBuilder
 import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.ProgressTracker
 import com.r3.corda.enterprise.perftestcordapp.contracts.asset.Cash
-import com.r3.corda.enterprise.perftestcordapp.contracts.asset.CashSelection
+import com.r3.corda.enterprise.perftestcordapp.contracts.asset.cash.selection.AbstractCashSelection
 import com.r3.corda.enterprise.perftestcordapp.issuedBy
 import java.util.*
 
@@ -41,12 +41,12 @@ class CashExitFlow(private val amount: Amount<Currency>,
      * (for this flow this map is always empty).
      */
     @Suspendable
-    @Throws(PtCashException::class)
+    @Throws(CashException::class)
     override fun call(): AbstractCashFlow.Result {
         progressTracker.currentStep = GENERATING_TX
         val builder = TransactionBuilder(notary = null)
         val issuer = ourIdentity.ref(issuerRef)
-        val exitStates = CashSelection
+        val exitStates = AbstractCashSelection
                 .getInstance { serviceHub.jdbcSession().metaData }
                 .unconsumedCashStatesForSpending(serviceHub, amount, setOf(issuer.party), builder.notary, builder.lockId, setOf(issuer.reference))
         val signers = try {
@@ -55,11 +55,11 @@ class CashExitFlow(private val amount: Amount<Currency>,
                     amount.issuedBy(issuer),
                     exitStates)
         } catch (e: InsufficientBalanceException) {
-            throw PtCashException("Exiting more cash than exists", e)
+            throw CashException("Exiting more cash than exists", e)
         }
 
         // Work out who the owners of the burnt states were (specify page size so we don't silently drop any if > DEFAULT_PAGE_SIZE)
-        val inputStates = serviceHub.vaultQueryService.queryBy<Cash.State>(VaultQueryCriteria(stateRefs = builder.inputStates()),
+        val inputStates = serviceHub.vaultService.queryBy<Cash.State>(VaultQueryCriteria(stateRefs = builder.inputStates()),
                 PageSpecification(pageNumber = DEFAULT_PAGE_NUM, pageSize = builder.inputStates().size)).states
 
         // TODO: Is it safe to drop participants we don't know how to contact? Does not knowing how to contact them

perftestcordapp/src/main/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/CashPaymentFlow.kt

@@ -55,7 +55,7 @@ open class CashPaymentFlow(
                     anonymousRecipient,
                     issuerConstraint)
         } catch (e: InsufficientBalanceException) {
-            throw PtCashException("Insufficient cash for spend: ${e.message}", e)
+            throw CashException("Insufficient cash for spend: ${e.message}", e)
         }
 
         progressTracker.currentStep = SIGNING_TX

perftestcordapp/src/test/kotlin/com/r3/corda/enterprise/perftestcordapp/contracts/CommercialPaperTests.kt

@@ -15,6 +15,7 @@ import com.r3.corda.enterprise.perftestcordapp.contracts.asset.*
 import net.corda.testing.*
 import com.r3.corda.enterprise.perftestcordapp.contracts.asset.fillWithSomeTestCash
 import net.corda.testing.node.MockServices
+import net.corda.testing.node.MockServices.Companion.makeTestDatabaseAndMockServices
 import org.junit.Ignore
 import org.junit.Test
 import org.junit.runner.RunWith
@@ -81,8 +82,8 @@ class CommercialPaperTestsGeneric {
         ledger {
             unverifiedTransaction {
                 attachment(Cash.PROGRAM_ID)
-                output(Cash.PROGRAM_ID, "alice's $900", 900.DOLLARS.CASH `issued by` issuer `owned by` ALICE)
+                output(Cash.PROGRAM_ID, "alice's $900", 900.DOLLARS.CASH issuedBy issuer ownedBy ALICE)
-                output(Cash.PROGRAM_ID, "some profits", someProfits.STATE `owned by` MEGA_CORP)
+                output(Cash.PROGRAM_ID, "some profits", someProfits.STATE ownedBy MEGA_CORP)
             }
 
             // Some CP is issued onto the ledger by MegaCorp.
@@ -100,7 +101,7 @@ class CommercialPaperTestsGeneric {
                 attachments(Cash.PROGRAM_ID, CommercialPaper.CP_PROGRAM_ID)
                 input("paper")
                 input("alice's $900")
-                output(Cash.PROGRAM_ID, "borrowed $900") { 900.DOLLARS.CASH `issued by` issuer `owned by` MEGA_CORP }
+                output(Cash.PROGRAM_ID, "borrowed $900") { 900.DOLLARS.CASH issuedBy issuer ownedBy MEGA_CORP }
                 output(thisTest.getContract(), "alice's paper") { "paper".output<CommercialPaper.State>().withOwner(ALICE) }
                 command(ALICE_PUBKEY) { Cash.Commands.Move() }
                 command(MEGA_CORP_PUBKEY) { thisTest.getMoveCommand() }
@@ -115,8 +116,8 @@ class CommercialPaperTestsGeneric {
                 input("some profits")
 
                 fun TransactionDSL<TransactionDSLInterpreter>.outputs(aliceGetsBack: Amount<Issued<Currency>>) {
-                    output(Cash.PROGRAM_ID, "Alice's profit") { aliceGetsBack.STATE `owned by` ALICE }
+                    output(Cash.PROGRAM_ID, "Alice's profit") { aliceGetsBack.STATE ownedBy ALICE }
-                    output(Cash.PROGRAM_ID, "Change") { (someProfits - aliceGetsBack).STATE `owned by` MEGA_CORP }
+                    output(Cash.PROGRAM_ID, "Change") { (someProfits - aliceGetsBack).STATE ownedBy MEGA_CORP }
                 }
 
                 command(MEGA_CORP_PUBKEY) { Cash.Commands.Move() }
@@ -216,7 +217,6 @@ class CommercialPaperTestsGeneric {
     //    @Test
     @Ignore
     fun `issue move and then redeem`() {
-        setCordappPackages("com.r3.enterprise.perftestcordapp.contracts")
         initialiseTestSerialization()
         val aliceDatabaseAndServices = MockServices.makeTestDatabaseAndMockServices(keys = listOf(ALICE_KEY))
         val databaseAlice = aliceDatabaseAndServices.first
@@ -300,5 +300,3 @@ class CommercialPaperTestsGeneric {
         resetTestSerialization()
     }
 }
-
-

perftestcordapp/src/test/kotlin/com/r3/corda/enterprise/perftestcordapp/contracts/asset/CashSelectionH2Test.kt

@@ -0,0 +1,40 @@
+package com.r3.corda.enterprise.perftestcordapp.contracts.asset
+
+import net.corda.core.utilities.getOrThrow
+import com.r3.corda.enterprise.perftestcordapp.DOLLARS
+import com.r3.corda.enterprise.perftestcordapp.flows.CashException
+import com.r3.corda.enterprise.perftestcordapp.flows.CashPaymentFlow
+import net.corda.testing.chooseIdentity
+import net.corda.testing.node.MockNetwork
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.Test
+
+
+class CashSelectionH2Test {
+
+    @Test
+    fun `check does not hold connection over retries`() {
+        val mockNet = MockNetwork(threadPerNode = true)
+        try {
+            val notaryNode = mockNet.createNotaryNode()
+            val bankA = mockNet.createNode(configOverrides = { existingConfig ->
+                // Tweak connections to be minimal to make this easier (1 results in a hung node during start up, so use 2 connections).
+                existingConfig.dataSourceProperties.setProperty("maximumPoolSize", "2")
+                existingConfig
+            })
+
+            mockNet.startNodes()
+
+            // Start more cash spends than we have connections.  If spend leaks a connection on retry, we will run out of connections.
+            val flow1 = bankA.services.startFlow(CashPaymentFlow(amount = 100.DOLLARS, anonymous = false, recipient = notaryNode.info.chooseIdentity()))
+            val flow2 = bankA.services.startFlow(CashPaymentFlow(amount = 100.DOLLARS, anonymous = false, recipient = notaryNode.info.chooseIdentity()))
+            val flow3 = bankA.services.startFlow(CashPaymentFlow(amount = 100.DOLLARS, anonymous = false, recipient = notaryNode.info.chooseIdentity()))
+
+            assertThatThrownBy { flow1.resultFuture.getOrThrow() }.isInstanceOf(CashException::class.java)
+            assertThatThrownBy { flow2.resultFuture.getOrThrow() }.isInstanceOf(CashException::class.java)
+            assertThatThrownBy { flow3.resultFuture.getOrThrow() }.isInstanceOf(CashException::class.java)
+        } finally {
+            mockNet.stopNodes()
+        }
+    }
+}

perftestcordapp/src/test/kotlin/com/r3/corda/enterprise/perftestcordapp/contracts/asset/CashTests.kt

@@ -11,6 +11,7 @@ import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.generateKeyPair
 import net.corda.core.identity.AbstractParty
 import net.corda.core.identity.AnonymousParty
+import net.corda.core.identity.CordaX500Name
 import net.corda.core.identity.Party
 import net.corda.core.node.ServiceHub
 import net.corda.core.node.services.Vault
@@ -80,25 +81,25 @@ fun ServiceHub.fillWithSomeTestCash(howMuch: Amount<Currency>,
 
 
 class CashTests : TestDependencyInjectionBase() {
-    val defaultRef = OpaqueBytes(ByteArray(1, { 1 }))
+    private val defaultRef = OpaqueBytes(ByteArray(1, { 1 }))
-    val defaultIssuer = MEGA_CORP.ref(defaultRef)
+    private val defaultIssuer = MEGA_CORP.ref(defaultRef)
-    val inState = Cash.State(
+    private val inState = Cash.State(
             amount = 1000.DOLLARS `issued by` defaultIssuer,
             owner = AnonymousParty(ALICE_PUBKEY)
     )
     // Input state held by the issuer
-    val issuerInState = inState.copy(owner = defaultIssuer.party)
+    private val issuerInState = inState.copy(owner = defaultIssuer.party)
-    val outState = issuerInState.copy(owner = AnonymousParty(BOB_PUBKEY))
+    private val outState = issuerInState.copy(owner = AnonymousParty(BOB_PUBKEY))
 
-    fun Cash.State.editDepositRef(ref: Byte) = copy(
+    private fun Cash.State.editDepositRef(ref: Byte) = copy(
             amount = Amount(amount.quantity, token = amount.token.copy(amount.token.issuer.copy(reference = OpaqueBytes.of(ref))))
     )
 
-    lateinit var miniCorpServices: MockServices
+    private lateinit var miniCorpServices: MockServices
-    lateinit var megaCorpServices: MockServices
+    private lateinit var megaCorpServices: MockServices
     val vault: VaultService get() = miniCorpServices.vaultService
     lateinit var database: CordaPersistence
-    lateinit var vaultStatesUnconsumed: List<StateAndRef<Cash.State>>
+    private lateinit var vaultStatesUnconsumed: List<StateAndRef<Cash.State>>
 
     @Before
     fun setUp() {
@@ -120,7 +121,7 @@ class CashTests : TestDependencyInjectionBase() {
                     ownedBy = OUR_IDENTITY_1, issuedBy = MINI_CORP.ref(1), issuerServices = miniCorpServices)
         }
         database.transaction {
-            vaultStatesUnconsumed = miniCorpServices.vaultQueryService.queryBy<Cash.State>().states
+            vaultStatesUnconsumed = miniCorpServices.vaultService.queryBy<Cash.State>().states
         }
         resetTestSerialization()
     }
@@ -154,7 +155,7 @@ class CashTests : TestDependencyInjectionBase() {
             }
             tweak {
                 output(Cash.PROGRAM_ID) { outState }
-                output(Cash.PROGRAM_ID) { outState `issued by` MINI_CORP }
+                output(Cash.PROGRAM_ID) { outState issuedBy MINI_CORP }
                 command(ALICE_PUBKEY) { Cash.Commands.Move() }
                 this `fails with` "at least one cash input"
             }
@@ -358,7 +359,7 @@ class CashTests : TestDependencyInjectionBase() {
         transaction {
             attachment(Cash.PROGRAM_ID)
             input(Cash.PROGRAM_ID) { inState }
-            output(Cash.PROGRAM_ID) { outState `issued by` MINI_CORP }
+            output(Cash.PROGRAM_ID) { outState issuedBy MINI_CORP }
             command(ALICE_PUBKEY) { Cash.Commands.Move() }
             this `fails with` "the amounts balance"
         }
@@ -397,7 +398,7 @@ class CashTests : TestDependencyInjectionBase() {
         transaction {
             attachment(Cash.PROGRAM_ID)
             input(Cash.PROGRAM_ID) { inState }
-            input(Cash.PROGRAM_ID) { inState `issued by` MINI_CORP }
+            input(Cash.PROGRAM_ID) { inState issuedBy MINI_CORP }
             output(Cash.PROGRAM_ID) { outState }
             command(ALICE_PUBKEY) { Cash.Commands.Move() }
             this `fails with` "the amounts balance"
@@ -445,9 +446,9 @@ class CashTests : TestDependencyInjectionBase() {
         transaction {
             attachment(Cash.PROGRAM_ID)
             input(Cash.PROGRAM_ID) { issuerInState }
-            input(Cash.PROGRAM_ID) { issuerInState.copy(owner = MINI_CORP) `issued by` MINI_CORP }
+            input(Cash.PROGRAM_ID) { issuerInState.copy(owner = MINI_CORP) issuedBy MINI_CORP }
 
-            output(Cash.PROGRAM_ID) { issuerInState.copy(amount = issuerInState.amount - (200.DOLLARS `issued by` defaultIssuer)) `issued by` MINI_CORP }
+            output(Cash.PROGRAM_ID) { issuerInState.copy(amount = issuerInState.amount - (200.DOLLARS `issued by` defaultIssuer)) issuedBy MINI_CORP }
             output(Cash.PROGRAM_ID) { issuerInState.copy(owner = MINI_CORP, amount = issuerInState.amount - (200.DOLLARS `issued by` defaultIssuer)) }
 
             command(MEGA_CORP_PUBKEY, MINI_CORP_PUBKEY) { Cash.Commands.Move() }
@@ -481,7 +482,7 @@ class CashTests : TestDependencyInjectionBase() {
             attachment(Cash.PROGRAM_ID)
             // Gather 2000 dollars from two different issuers.
             input(Cash.PROGRAM_ID) { inState }
-            input(Cash.PROGRAM_ID) { inState `issued by` MINI_CORP }
+            input(Cash.PROGRAM_ID) { inState issuedBy MINI_CORP }
             command(ALICE_PUBKEY) { Cash.Commands.Move() }
 
             // Can't merge them together.
@@ -498,7 +499,7 @@ class CashTests : TestDependencyInjectionBase() {
 
             // This works.
             output(Cash.PROGRAM_ID) { inState.copy(owner = AnonymousParty(BOB_PUBKEY)) }
-            output(Cash.PROGRAM_ID) { inState.copy(owner = AnonymousParty(BOB_PUBKEY)) `issued by` MINI_CORP }
+            output(Cash.PROGRAM_ID) { inState.copy(owner = AnonymousParty(BOB_PUBKEY)) issuedBy MINI_CORP }
             this.verifies()
         }
     }
@@ -509,10 +510,10 @@ class CashTests : TestDependencyInjectionBase() {
         transaction {
             attachment(Cash.PROGRAM_ID)
             val pounds = Cash.State(658.POUNDS `issued by` MINI_CORP.ref(3, 4, 5), AnonymousParty(BOB_PUBKEY))
-            input(Cash.PROGRAM_ID) { inState `owned by` AnonymousParty(ALICE_PUBKEY) }
+            input(Cash.PROGRAM_ID) { inState ownedBy AnonymousParty(ALICE_PUBKEY) }
             input(Cash.PROGRAM_ID) { pounds }
-            output(Cash.PROGRAM_ID) { inState `owned by` AnonymousParty(BOB_PUBKEY) }
+            output(Cash.PROGRAM_ID) { inState ownedBy AnonymousParty(BOB_PUBKEY) }
-            output(Cash.PROGRAM_ID) { pounds `owned by` AnonymousParty(ALICE_PUBKEY) }
+            output(Cash.PROGRAM_ID) { pounds ownedBy AnonymousParty(ALICE_PUBKEY) }
             command(ALICE_PUBKEY, BOB_PUBKEY) { Cash.Commands.Move() }
 
             this.verifies()
@@ -523,19 +524,20 @@ class CashTests : TestDependencyInjectionBase() {
     //
     // Spend tx generation
 
-    val OUR_KEY: KeyPair by lazy { generateKeyPair() }
+    private val OUR_KEY: KeyPair by lazy { generateKeyPair() }
-    val OUR_IDENTITY_1: AbstractParty get() = AnonymousParty(OUR_KEY.public)
+    private val OUR_IDENTITY_1: AbstractParty get() = AnonymousParty(OUR_KEY.public)
+    private val OUR_IDENTITY_AND_CERT = getTestPartyAndCertificate(CordaX500Name(organisation = "Me", locality = "London", country = "GB"), OUR_KEY.public)
 
-    val THEIR_IDENTITY_1 = AnonymousParty(MINI_CORP_PUBKEY)
+    private val THEIR_IDENTITY_1 = AnonymousParty(MINI_CORP_PUBKEY)
-    val THEIR_IDENTITY_2 = AnonymousParty(CHARLIE_PUBKEY)
+    private val THEIR_IDENTITY_2 = AnonymousParty(CHARLIE_PUBKEY)
 
-    fun makeCash(amount: Amount<Currency>, corp: Party, depositRef: Byte = 1) =
+    private fun makeCash(amount: Amount<Currency>, issuer: AbstractParty, depositRef: Byte = 1) =
             StateAndRef(
-                    TransactionState<Cash.State>(Cash.State(amount `issued by` corp.ref(depositRef), OUR_IDENTITY_1), Cash.PROGRAM_ID, DUMMY_NOTARY),
+                    TransactionState(Cash.State(amount `issued by` issuer.ref(depositRef), OUR_IDENTITY_1), Cash.PROGRAM_ID, DUMMY_NOTARY),
                     StateRef(SecureHash.randomSHA256(), Random().nextInt(32))
             )
 
-    val WALLET = listOf(
+    private val WALLET = listOf(
             makeCash(100.DOLLARS, MEGA_CORP),
             makeCash(400.DOLLARS, MEGA_CORP),
             makeCash(80.DOLLARS, MINI_CORP),
@@ -545,16 +547,17 @@ class CashTests : TestDependencyInjectionBase() {
     /**
      * Generate an exit transaction, removing some amount of cash from the ledger.
      */
-    private fun makeExit(amount: Amount<Currency>, corp: Party, depositRef: Byte = 1): WireTransaction {
+    private fun makeExit(serviceHub: ServiceHub, amount: Amount<Currency>, issuer: Party, depositRef: Byte = 1): WireTransaction {
         val tx = TransactionBuilder(DUMMY_NOTARY)
-        Cash().generateExit(tx, Amount(amount.quantity, Issued(corp.ref(depositRef), amount.token)), WALLET)
+        val payChangeTo = serviceHub.keyManagementService.freshKeyAndCert(MINI_CORP_IDENTITY, false).party
-        return tx.toWireTransaction(miniCorpServices)
+        Cash().generateExit(tx, Amount(amount.quantity, Issued(issuer.ref(depositRef), amount.token)), WALLET, payChangeTo)
+        return tx.toWireTransaction(serviceHub)
     }
 
     private fun makeSpend(amount: Amount<Currency>, dest: AbstractParty): WireTransaction {
         val tx = TransactionBuilder(DUMMY_NOTARY)
         database.transaction {
-            Cash.generateSpend(miniCorpServices, tx, amount, dest)
+            Cash.generateSpend(miniCorpServices, tx, amount, OUR_IDENTITY_AND_CERT, dest)
         }
         return tx.toWireTransaction(miniCorpServices)
     }
@@ -565,7 +568,7 @@ class CashTests : TestDependencyInjectionBase() {
     @Test
     fun generateSimpleExit() {
         initialiseTestSerialization()
-        val wtx = makeExit(100.DOLLARS, MEGA_CORP, 1)
+        val wtx = makeExit(miniCorpServices, 100.DOLLARS, MEGA_CORP, 1)
         assertEquals(WALLET[0].ref, wtx.inputs[0])
         assertEquals(0, wtx.outputs.size)
 
@@ -581,10 +584,16 @@ class CashTests : TestDependencyInjectionBase() {
     @Test
     fun generatePartialExit() {
         initialiseTestSerialization()
-        val wtx = makeExit(50.DOLLARS, MEGA_CORP, 1)
+        val wtx = makeExit(miniCorpServices, 50.DOLLARS, MEGA_CORP, 1)
-        assertEquals(WALLET[0].ref, wtx.inputs[0])
+        val actualInput = wtx.inputs.single()
-        assertEquals(1, wtx.outputs.size)
+        // Filter the available inputs and confirm exactly one has been used
-        assertEquals(WALLET[0].state.data.copy(amount = WALLET[0].state.data.amount.splitEvenly(2).first()), wtx.getOutput(0))
+        val expectedInputs = WALLET.filter { it.ref == actualInput }
+        assertEquals(1, expectedInputs.size)
+        val inputState = expectedInputs.single()
+        val actualChange = wtx.outputs.single().data as Cash.State
+        val expectedChangeAmount = inputState.state.data.amount.quantity - 50.DOLLARS.quantity
+        val expectedChange = WALLET[0].state.data.copy(amount = WALLET[0].state.data.amount.copy(quantity = expectedChangeAmount), owner = actualChange.owner)
+        assertEquals(expectedChange, wtx.getOutput(0))
     }
 
     /**
@@ -593,7 +602,7 @@ class CashTests : TestDependencyInjectionBase() {
     @Test
     fun generateAbsentExit() {
         initialiseTestSerialization()
-        assertFailsWith<InsufficientBalanceException> { makeExit(100.POUNDS, MEGA_CORP, 1) }
+        assertFailsWith<InsufficientBalanceException> { makeExit(miniCorpServices, 100.POUNDS, MEGA_CORP, 1) }
     }
 
     /**
@@ -602,7 +611,7 @@ class CashTests : TestDependencyInjectionBase() {
     @Test
     fun generateInvalidReferenceExit() {
         initialiseTestSerialization()
-        assertFailsWith<InsufficientBalanceException> { makeExit(100.POUNDS, MEGA_CORP, 2) }
+        assertFailsWith<InsufficientBalanceException> { makeExit(miniCorpServices, 100.POUNDS, MEGA_CORP, 2) }
     }
 
     /**
@@ -611,7 +620,7 @@ class CashTests : TestDependencyInjectionBase() {
     @Test
     fun generateInsufficientExit() {
         initialiseTestSerialization()
-        assertFailsWith<InsufficientBalanceException> { makeExit(1000.DOLLARS, MEGA_CORP, 1) }
+        assertFailsWith<InsufficientBalanceException> { makeExit(miniCorpServices, 1000.DOLLARS, MEGA_CORP, 1) }
     }
 
     /**
@@ -620,7 +629,7 @@ class CashTests : TestDependencyInjectionBase() {
     @Test
     fun generateOwnerWithNoStatesExit() {
         initialiseTestSerialization()
-        assertFailsWith<InsufficientBalanceException> { makeExit(100.POUNDS, CHARLIE, 1) }
+        assertFailsWith<InsufficientBalanceException> { makeExit(miniCorpServices, 100.POUNDS, CHARLIE, 1) }
     }
 
     /**
@@ -629,9 +638,9 @@ class CashTests : TestDependencyInjectionBase() {
     @Test
     fun generateExitWithEmptyVault() {
         initialiseTestSerialization()
-        assertFailsWith<InsufficientBalanceException> {
+        assertFailsWith<IllegalArgumentException> {
             val tx = TransactionBuilder(DUMMY_NOTARY)
-            Cash().generateExit(tx, Amount(100, Issued(CHARLIE.ref(1), GBP)), emptyList())
+            Cash().generateExit(tx, Amount(100, Issued(CHARLIE.ref(1), GBP)), emptyList(), OUR_IDENTITY_1)
         }
     }
 
@@ -643,7 +652,6 @@ class CashTests : TestDependencyInjectionBase() {
                     makeSpend(100.DOLLARS, THEIR_IDENTITY_1)
                 }
         database.transaction {
-            @Suppress("UNCHECKED_CAST")
             val vaultState = vaultStatesUnconsumed.elementAt(0)
             assertEquals(vaultState.ref, wtx.inputs[0])
             assertEquals(vaultState.state.data.copy(owner = THEIR_IDENTITY_1), wtx.getOutput(0))
@@ -657,7 +665,7 @@ class CashTests : TestDependencyInjectionBase() {
         database.transaction {
 
             val tx = TransactionBuilder(DUMMY_NOTARY)
-            Cash.generateSpend(miniCorpServices, tx, 80.DOLLARS, ALICE, setOf(MINI_CORP))
+            Cash.generateSpend(miniCorpServices, tx, 80.DOLLARS, OUR_IDENTITY_AND_CERT, ALICE, setOf(MINI_CORP))
 
             assertEquals(vaultStatesUnconsumed.elementAt(2).ref, tx.inputStates()[0])
         }
@@ -671,16 +679,15 @@ class CashTests : TestDependencyInjectionBase() {
                     makeSpend(10.DOLLARS, THEIR_IDENTITY_1)
                 }
         database.transaction {
-            @Suppress("UNCHECKED_CAST")
             val vaultState = vaultStatesUnconsumed.elementAt(0)
             val changeAmount = 90.DOLLARS `issued by` defaultIssuer
-            val likelyChangeState = wtx.outputs.map(TransactionState<*>::data).filter { state ->
+            val likelyChangeState = wtx.outputs.map(TransactionState<*>::data).single { state ->
                 if (state is Cash.State) {
                     state.amount == changeAmount
                 } else {
                     false
                 }
-            }.single()
+            }
             val changeOwner = (likelyChangeState as Cash.State).owner
             assertEquals(1, miniCorpServices.keyManagementService.filterMyKeys(setOf(changeOwner.owningKey)).toList().size)
             assertEquals(vaultState.ref, wtx.inputs[0])
@@ -698,7 +705,6 @@ class CashTests : TestDependencyInjectionBase() {
                     makeSpend(500.DOLLARS, THEIR_IDENTITY_1)
                 }
         database.transaction {
-            @Suppress("UNCHECKED_CAST")
             val vaultState0 = vaultStatesUnconsumed.elementAt(0)
             val vaultState1 = vaultStatesUnconsumed.elementAt(1)
             assertEquals(vaultState0.ref, wtx.inputs[0])
@@ -770,8 +776,8 @@ class CashTests : TestDependencyInjectionBase() {
                 Cash.State(1000.POUNDS `issued by` MINI_CORP.ref(3), MEGA_CORP).amount.token)
 
         // States cannot be aggregated if the reference differs
-        assertNotEquals(fiveThousandDollarsFromMega.amount.token, (fiveThousandDollarsFromMega `with deposit` defaultIssuer).amount.token)
+        assertNotEquals(fiveThousandDollarsFromMega.amount.token, (fiveThousandDollarsFromMega withDeposit defaultIssuer).amount.token)
-        assertNotEquals((fiveThousandDollarsFromMega `with deposit` defaultIssuer).amount.token, fiveThousandDollarsFromMega.amount.token)
+        assertNotEquals((fiveThousandDollarsFromMega withDeposit defaultIssuer).amount.token, fiveThousandDollarsFromMega.amount.token)
     }
 
     @Test

perftestcordapp/src/test/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/CashExitFlowTests.kt

@@ -1,22 +1,18 @@
-package com.r3.corda.enterprise.perftestcordapp.contracts.flows
+package com.r3.corda.enterprise.perftestcordapp.flows
 
-import net.corda.core.identity.Party
-import net.corda.core.utilities.OpaqueBytes
-import net.corda.core.utilities.getOrThrow
-import com.r3.corda.enterprise.perftestcordapp.flows.PtCashException
-import com.r3.corda.enterprise.perftestcordapp.flows.CashExitFlow
-import com.r3.corda.enterprise.perftestcordapp.flows.CashIssueFlow
 import com.r3.corda.enterprise.perftestcordapp.DOLLARS
 import com.r3.corda.enterprise.perftestcordapp.`issued by`
 import com.r3.corda.enterprise.perftestcordapp.contracts.asset.Cash
+import net.corda.core.identity.Party
+import net.corda.core.utilities.OpaqueBytes
+import net.corda.core.utilities.getOrThrow
 import net.corda.node.internal.StartedNode
+import net.corda.testing.BOC
 import net.corda.testing.chooseIdentity
 import net.corda.testing.getDefaultNotary
 import net.corda.testing.node.InMemoryMessagingNetwork.ServicePeerAllocationStrategy.RoundRobin
 import net.corda.testing.node.MockNetwork
 import net.corda.testing.node.MockNetwork.MockNode
-import net.corda.testing.setCordappPackages
-import net.corda.testing.unsetCordappPackages
 import org.junit.After
 import org.junit.Before
 import org.junit.Test
@@ -34,11 +30,10 @@ class CashExitFlowTests {
 
     @Before
     fun start() {
-        setCordappPackages("com.r3.corda.enterprise.perftestcordapp.contracts.asset")
+        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin(), cordappPackages = listOf("com.r3.corda.enterprise.perftestcordapp.contracts.asset"))
-        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin())
+        notaryNode = mockNet.createNotaryNode()
-        val nodes = mockNet.createSomeNodes(1)
+        bankOfCordaNode = mockNet.createPartyNode(BOC.name)
-        notaryNode = nodes.notaryNode
+        notary = notaryNode.services.getDefaultNotary()
-        bankOfCordaNode = nodes.partyNodes[0]
         bankOfCorda = bankOfCordaNode.info.chooseIdentity()
 
         mockNet.runNetwork()
@@ -51,7 +46,6 @@ class CashExitFlowTests {
     @After
     fun cleanUp() {
         mockNet.stopNodes()
-        unsetCordappPackages()
     }
 
     @Test
@@ -72,7 +66,7 @@ class CashExitFlowTests {
         val expected = 0.DOLLARS
         val future = bankOfCordaNode.services.startFlow(CashExitFlow(expected, ref)).resultFuture
         mockNet.runNetwork()
-        assertFailsWith<PtCashException> {
+        assertFailsWith<CashException> {
             future.getOrThrow()
         }
     }

perftestcordapp/src/test/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/CashIssueFlowTests.kt

@@ -1,4 +1,4 @@
-package com.r3.corda.enterprise.perftestcordapp.contracts.flows
+package com.r3.corda.enterprise.perftestcordapp.flows
 
 import net.corda.core.identity.Party
 import net.corda.core.utilities.OpaqueBytes
@@ -7,13 +7,12 @@ import com.r3.corda.enterprise.perftestcordapp.DOLLARS
 import com.r3.corda.enterprise.perftestcordapp.`issued by`
 import com.r3.corda.enterprise.perftestcordapp.contracts.asset.Cash
 import net.corda.node.internal.StartedNode
-import com.r3.corda.enterprise.perftestcordapp.flows.CashIssueFlow
 import net.corda.testing.chooseIdentity
 import net.corda.testing.getDefaultNotary
+import net.corda.testing.BOC
 import net.corda.testing.node.InMemoryMessagingNetwork.ServicePeerAllocationStrategy.RoundRobin
 import net.corda.testing.node.MockNetwork
 import net.corda.testing.node.MockNetwork.MockNode
-import net.corda.testing.setCordappPackages
 import org.junit.After
 import org.junit.Before
 import org.junit.Test
@@ -29,15 +28,12 @@ class CashIssueFlowTests {
 
     @Before
     fun start() {
-        setCordappPackages("com.r3.corda.enterprise.perftestcordapp.contracts.asset")
+        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin(), cordappPackages = listOf("com.r3.corda.enterprise.perftestcordapp.contracts.asset"))
-        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin())
+        notaryNode = mockNet.createNotaryNode()
-        val nodes = mockNet.createSomeNodes(1)
+        bankOfCordaNode = mockNet.createPartyNode(BOC.name)
-        notaryNode = nodes.notaryNode
-        bankOfCordaNode = nodes.partyNodes[0]
         bankOfCorda = bankOfCordaNode.info.chooseIdentity()
-
+        notary = notaryNode.services.getDefaultNotary()
         mockNet.runNetwork()
-        notary = bankOfCordaNode.services.getDefaultNotary()
     }
 
     @After

perftestcordapp/src/test/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/CashPaymentFlowTests.kt

@@ -1,26 +1,19 @@
-package com.r3.corda.enterprise.perftestcordapp.contracts.flows
+package com.r3.corda.enterprise.perftestcordapp.flows
 
+import com.r3.corda.enterprise.perftestcordapp.DOLLARS
+import com.r3.corda.enterprise.perftestcordapp.`issued by`
+import com.r3.corda.enterprise.perftestcordapp.contracts.asset.Cash
 import net.corda.core.identity.Party
 import net.corda.core.node.services.Vault
 import net.corda.core.node.services.trackBy
 import net.corda.core.node.services.vault.QueryCriteria
 import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.getOrThrow
-import com.r3.corda.enterprise.perftestcordapp.DOLLARS
-import com.r3.corda.enterprise.perftestcordapp.`issued by`
-import com.r3.corda.enterprise.perftestcordapp.contracts.asset.Cash
-import com.r3.corda.enterprise.perftestcordapp.flows.PtCashException
-import com.r3.corda.enterprise.perftestcordapp.flows.CashIssueFlow
-import com.r3.corda.enterprise.perftestcordapp.flows.CashPaymentFlow
 import net.corda.node.internal.StartedNode
-import net.corda.testing.chooseIdentity
+import net.corda.testing.*
-import net.corda.testing.expect
-import net.corda.testing.expectEvents
-import net.corda.testing.getDefaultNotary
 import net.corda.testing.node.InMemoryMessagingNetwork.ServicePeerAllocationStrategy.RoundRobin
 import net.corda.testing.node.MockNetwork
 import net.corda.testing.node.MockNetwork.MockNode
-import net.corda.testing.setCordappPackages
 import org.junit.After
 import org.junit.Before
 import org.junit.Test
@@ -38,11 +31,9 @@ class CashPaymentFlowTests {
 
     @Before
     fun start() {
-        setCordappPackages("com.r3.corda.enterprise.perftestcordapp.contracts.asset")
+        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin(), cordappPackages = listOf("com.r3.corda.enterprise.perftestcordapp.contracts.asset"))
-        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin())
+        notaryNode = mockNet.createNotaryNode()
-        val nodes = mockNet.createSomeNodes(1)
+        bankOfCordaNode = mockNet.createPartyNode(BOC.name)
-        notaryNode = nodes.notaryNode
-        bankOfCordaNode = nodes.partyNodes[0]
         bankOfCorda = bankOfCordaNode.info.chooseIdentity()
         notary = notaryNode.services.getDefaultNotary()
         val future = bankOfCordaNode.services.startFlow(CashIssueFlow(initialBalance, ref, notary)).resultFuture
@@ -64,8 +55,8 @@ class CashPaymentFlowTests {
         bankOfCordaNode.database.transaction {
             // Register for vault updates
             val criteria = QueryCriteria.VaultQueryCriteria(status = Vault.StateStatus.ALL)
-            val (_, vaultUpdatesBoc) = bankOfCordaNode.services.vaultQueryService.trackBy<Cash.State>(criteria)
+            val (_, vaultUpdatesBoc) = bankOfCordaNode.services.vaultService.trackBy<Cash.State>(criteria)
-            val (_, vaultUpdatesBankClient) = notaryNode.services.vaultQueryService.trackBy<Cash.State>(criteria)
+            val (_, vaultUpdatesBankClient) = notaryNode.services.vaultService.trackBy<Cash.State>(criteria)
 
             val future = bankOfCordaNode.services.startFlow(CashPaymentFlow(expectedPayment,
                     payTo)).resultFuture
@@ -102,7 +93,7 @@ class CashPaymentFlowTests {
         val future = bankOfCordaNode.services.startFlow(CashPaymentFlow(expected,
                 payTo)).resultFuture
         mockNet.runNetwork()
-        assertFailsWith<PtCashException> {
+        assertFailsWith<CashException> {
             future.getOrThrow()
         }
     }

perftestcordapp/src/test/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/TwoPartyTradeFlowTest.kt

@@ -1,4 +1,7 @@
-package com.r3.corda.enterprise.perftestcordapp.contracts.flows
+package com.r3.corda.enterprise.perftestcordapp.flows
+
+// NB: Unlike the other flow tests in this package, this is not originally copied from net.corda.finance, but
+// from net.corda.node.messaging
 
 import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.concurrent.CordaFuture
@@ -31,8 +34,6 @@ import com.r3.corda.enterprise.perftestcordapp.`issued by`
 import com.r3.corda.enterprise.perftestcordapp.contracts.CommercialPaper
 import com.r3.corda.enterprise.perftestcordapp.contracts.asset.CASH
 import com.r3.corda.enterprise.perftestcordapp.contracts.asset.Cash
-import com.r3.corda.enterprise.perftestcordapp.contracts.asset.`issued by`
-import com.r3.corda.enterprise.perftestcordapp.contracts.asset.`owned by`
 import com.r3.corda.enterprise.perftestcordapp.flows.TwoPartyTradeFlow.Buyer
 import com.r3.corda.enterprise.perftestcordapp.flows.TwoPartyTradeFlow.Seller
 import net.corda.node.internal.StartedNode
@@ -43,13 +44,18 @@ import net.corda.node.utilities.CordaPersistence
 import net.corda.nodeapi.internal.ServiceInfo
 import net.corda.testing.*
 import com.r3.corda.enterprise.perftestcordapp.contracts.asset.fillWithSomeTestCash
+import net.corda.node.services.api.Checkpoint
+import net.corda.node.services.api.CheckpointStorage
 import net.corda.testing.node.InMemoryMessagingNetwork
 import net.corda.testing.node.MockNetwork
+import net.corda.testing.node.MockServices
 import net.corda.testing.node.pumpReceive
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.After
 import org.junit.Before
 import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
 import rx.Observable
 import java.io.ByteArrayInputStream
 import java.io.ByteArrayOutputStream
@@ -62,18 +68,41 @@ import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
 import kotlin.test.assertTrue
 
+
+/**
+ * Copied from DBCheckpointStorageTests as it is required as helper for this test
+ */
+internal fun CheckpointStorage.checkpoints(): List<Checkpoint> {
+    val checkpoints = mutableListOf<Checkpoint>()
+    forEach {
+        checkpoints += it
+        true
+    }
+    return checkpoints
+}
+
+
 /**
  * In this example, Alice wishes to sell her commercial paper to Bob in return for $1,000,000 and they wish to do
  * it on the ledger atomically. Therefore they must work together to build a transaction.
  *
  * We assume that Alice and Bob already found each other via some market, and have agreed the details already.
  */
-class TwoPartyTradeFlowTests {
+@RunWith(Parameterized::class)
+class TwoPartyTradeFlowTests(val anonymous: Boolean) {
+    companion object {
+        private val cordappPackages = listOf("com.r3.corda.enterprise.perftestcordapp.contracts")
+        @JvmStatic
+        @Parameterized.Parameters
+        fun data(): Collection<Boolean> {
+            return listOf(true, false)
+        }
+    }
+
     private lateinit var mockNet: MockNetwork
 
     @Before
     fun before() {
-        setCordappPackages("com.r3.corda.enterprise.perftestcordapp.contracts")
         LogHelper.setLevel("platform.trade", "core.contract.TransactionGroup", "recordingmap")
     }
 
@@ -81,7 +110,6 @@ class TwoPartyTradeFlowTests {
     fun after() {
         mockNet.stopNodes()
         LogHelper.reset("platform.trade", "core.contract.TransactionGroup", "recordingmap")
-        unsetCordappPackages()
     }
 
     @Test
@@ -89,18 +117,17 @@ class TwoPartyTradeFlowTests {
         // We run this in parallel threads to help catch any race conditions that may exist. The other tests
         // we run in the unit test thread exclusively to speed things up, ensure deterministic results and
         // allow interruption half way through.
-        mockNet = MockNetwork(false, true)
+        mockNet = MockNetwork(false, true, cordappPackages = cordappPackages)
-
+        ledger(MockServices(cordappPackages), initialiseSerialization = false) {
-        ledger(initialiseSerialization = false) {
+            val notaryNode = mockNet.createNotaryNode()
-            val basketOfNodes = mockNet.createSomeNodes(3)
+            val aliceNode = mockNet.createPartyNode(ALICE_NAME)
-            val notaryNode = basketOfNodes.notaryNode
+            val bobNode = mockNet.createPartyNode(BOB_NAME)
-            val aliceNode = basketOfNodes.partyNodes[0]
+            val bankNode = mockNet.createPartyNode(BOC_NAME)
-            val bobNode = basketOfNodes.partyNodes[1]
+            val alice = aliceNode.info.singleIdentity()
-            val bankNode = basketOfNodes.partyNodes[2]
+            val bank = bankNode.info.singleIdentity()
-            val cashIssuer = bankNode.info.chooseIdentity().ref(1)
+            val notary = notaryNode.services.getDefaultNotary()
-            val cpIssuer = bankNode.info.chooseIdentity().ref(1, 2, 3)
+            val cashIssuer = bank.ref(1)
-            val notary = aliceNode.services.getDefaultNotary()
+            val cpIssuer = bank.ref(1, 2, 3)
-
 
             aliceNode.internals.disableDBCloseOnStop()
             bobNode.internals.disableDBCloseOnStop()
@@ -111,8 +138,8 @@ class TwoPartyTradeFlowTests {
             }
 
             val alicesFakePaper = aliceNode.database.transaction {
-                fillUpForSeller(false, cpIssuer, aliceNode.info.chooseIdentity(),
+                fillUpForSeller(false, cpIssuer, alice,
-                        1200.DOLLARS `issued by` bankNode.info.chooseIdentity().ref(0), null, notary).second
+                        1200.DOLLARS `issued by` bank.ref(0), null, notary).second
             }
 
             insertFakeTransactions(alicesFakePaper, aliceNode, notaryNode, bankNode)
@@ -127,27 +154,28 @@ class TwoPartyTradeFlowTests {
             aliceNode.dispose()
             bobNode.dispose()
 
-//            aliceNode.database.transaction {
+            aliceNode.database.transaction {
-//                assertThat(aliceNode.checkpointStorage.checkpoints()).isEmpty()
+                assertThat(aliceNode.checkpointStorage.checkpoints()).isEmpty()
-//            }
+            }
             aliceNode.internals.manuallyCloseDB()
-//            bobNode.database.transaction {
+            bobNode.database.transaction {
-//                assertThat(bobNode.checkpointStorage.checkpoints()).isEmpty()
+                assertThat(bobNode.checkpointStorage.checkpoints()).isEmpty()
-//            }
+            }
             bobNode.internals.manuallyCloseDB()
         }
     }
 
     @Test(expected = InsufficientBalanceException::class)
     fun `trade cash for commercial paper fails using soft locking`() {
-        mockNet = MockNetwork(false, true)
+        mockNet = MockNetwork(false, true, cordappPackages = cordappPackages)
-
+        ledger(MockServices(cordappPackages), initialiseSerialization = false) {
-        ledger(initialiseSerialization = false) {
+            val notaryNode = mockNet.createNotaryNode()
-            val notaryNode = mockNet.createNotaryNode(null, DUMMY_NOTARY.name)
+            val aliceNode = mockNet.createPartyNode(ALICE_NAME)
-            val aliceNode = mockNet.createPartyNode(notaryNode.network.myAddress, ALICE.name)
+            val bobNode = mockNet.createPartyNode(BOB_NAME)
-            val bobNode = mockNet.createPartyNode(notaryNode.network.myAddress, BOB.name)
+            val bankNode = mockNet.createPartyNode(BOC_NAME)
-            val bankNode = mockNet.createPartyNode(notaryNode.network.myAddress, BOC.name)
+            val alice = aliceNode.info.singleIdentity()
-            val issuer = bankNode.info.chooseIdentity().ref(1)
+            val bank = bankNode.info.singleIdentity()
+            val issuer = bank.ref(1)
             val notary = aliceNode.services.getDefaultNotary()
 
             aliceNode.internals.disableDBCloseOnStop()
@@ -159,8 +187,8 @@ class TwoPartyTradeFlowTests {
             }
 
             val alicesFakePaper = aliceNode.database.transaction {
-                fillUpForSeller(false, issuer, aliceNode.info.chooseIdentity(),
+                fillUpForSeller(false, issuer, alice,
-                        1200.DOLLARS `issued by` bankNode.info.chooseIdentity().ref(0), null, notary).second
+                        1200.DOLLARS `issued by` bank.ref(0), null, notary).second
             }
 
             insertFakeTransactions(alicesFakePaper, aliceNode, notaryNode, bankNode)
@@ -182,52 +210,43 @@ class TwoPartyTradeFlowTests {
             aliceNode.dispose()
             bobNode.dispose()
 
-//            aliceNode.database.transaction {
+            aliceNode.database.transaction {
-//                assertThat(aliceNode.checkpointStorage.checkpoints()).isEmpty()
+                assertThat(aliceNode.checkpointStorage.checkpoints()).isEmpty()
-//            }
+            }
             aliceNode.internals.manuallyCloseDB()
-//            bobNode.database.transaction {
+            bobNode.database.transaction {
-//                assertThat(bobNode.checkpointStorage.checkpoints()).isEmpty()
+                assertThat(bobNode.checkpointStorage.checkpoints()).isEmpty()
-//            }
+            }
             bobNode.internals.manuallyCloseDB()
         }
     }
 
     @Test
     fun `shutdown and restore`() {
-        mockNet = MockNetwork(false)
+        mockNet = MockNetwork(false, cordappPackages = cordappPackages)
-        ledger(initialiseSerialization = false) {
+        ledger(MockServices(cordappPackages), initialiseSerialization = false) {
-            val notaryNode = mockNet.createNotaryNode(null, DUMMY_NOTARY.name)
+            val notaryNode = mockNet.createNotaryNode()
-            val aliceNode = mockNet.createPartyNode(notaryNode.network.myAddress, ALICE.name)
+            val aliceNode = mockNet.createPartyNode(ALICE_NAME)
-            var bobNode = mockNet.createPartyNode(notaryNode.network.myAddress, BOB.name)
+            var bobNode = mockNet.createPartyNode(BOB_NAME)
-            val bankNode = mockNet.createPartyNode(notaryNode.network.myAddress, BOC.name)
+            val bankNode = mockNet.createPartyNode(BOC_NAME)
-            val issuer = bankNode.info.chooseIdentity().ref(1, 2, 3)
-
-            // Let the nodes know about each other - normally the network map would handle this
-            mockNet.registerIdentities()
-
-            aliceNode.database.transaction {
-                aliceNode.services.identityService.verifyAndRegisterIdentity(bobNode.info.chooseIdentityAndCert())
-            }
-            bobNode.database.transaction {
-                bobNode.services.identityService.verifyAndRegisterIdentity(aliceNode.info.chooseIdentityAndCert())
-            }
             aliceNode.internals.disableDBCloseOnStop()
             bobNode.internals.disableDBCloseOnStop()
 
             val bobAddr = bobNode.network.myAddress as InMemoryMessagingNetwork.PeerHandle
-            val networkMapAddress = notaryNode.network.myAddress
-
             mockNet.runNetwork() // Clear network map registration messages
-            val notary = aliceNode.services.getDefaultNotary()
+
+            val notary = notaryNode.services.getDefaultNotary()
+            val alice = aliceNode.info.singleIdentity()
+            val bank = bankNode.info.singleIdentity()
+            val issuer = bank.ref(1, 2, 3)
 
             bobNode.database.transaction {
                 bobNode.services.fillWithSomeTestCash(2000.DOLLARS, bankNode.services, outputNotary = notary,
                         issuedBy = issuer)
             }
             val alicesFakePaper = aliceNode.database.transaction {
-                fillUpForSeller(false, issuer, aliceNode.info.chooseIdentity(),
+                fillUpForSeller(false, issuer, alice,
-                        1200.DOLLARS `issued by` bankNode.info.chooseIdentity().ref(0), null, notary).second
+                        1200.DOLLARS `issued by` bank.ref(0), null, notary).second
             }
             insertFakeTransactions(alicesFakePaper, aliceNode, notaryNode, bankNode)
             val aliceFuture = runBuyerAndSeller(notary, aliceNode, bobNode, "alice's paper".outputStateAndRef()).sellerResult
@@ -244,10 +263,10 @@ class TwoPartyTradeFlowTests {
             aliceNode.pumpReceive()
             bobNode.pumpReceive()
 
-//            // OK, now Bob has sent the partial transaction back to Alice and is waiting for Alice's signature.
+            // OK, now Bob has sent the partial transaction back to Alice and is waiting for Alice's signature.
-//            bobNode.database.transaction {
+            bobNode.database.transaction {
-//                assertThat(bobNode.checkpointStorage.checkpoints()).hasSize(1)
+                assertThat(bobNode.checkpointStorage.checkpoints()).hasSize(1)
-//            }
+            }
 
             val storage = bobNode.services.validatedTransactions
             val bobTransactionsBeforeCrash = bobNode.database.transaction {
@@ -267,13 +286,12 @@ class TwoPartyTradeFlowTests {
 
             // ... bring the node back up ... the act of constructing the SMM will re-register the message handlers
             // that Bob was waiting on before the reboot occurred.
-            bobNode = mockNet.createNode(networkMapAddress, bobAddr.id, object : MockNetwork.Factory<MockNetwork.MockNode> {
+            bobNode = mockNet.createNode(bobAddr.id, object : MockNetwork.Factory<MockNetwork.MockNode> {
                 override fun create(config: NodeConfiguration, network: MockNetwork, networkMapAddr: SingleMessageRecipient?,
-                                    advertisedServices: Set<ServiceInfo>, id: Int, overrideServices: Map<ServiceInfo, KeyPair>?,
+                                    id: Int, notaryIdentity: Pair<ServiceInfo, KeyPair>?, entropyRoot: BigInteger): MockNetwork.MockNode {
-                                    entropyRoot: BigInteger): MockNetwork.MockNode {
+                    return MockNetwork.MockNode(config, network, networkMapAddr, bobAddr.id, notaryIdentity, entropyRoot)
-                    return MockNetwork.MockNode(config, network, networkMapAddr, advertisedServices, bobAddr.id, overrideServices, entropyRoot)
                 }
-            }, BOB.name)
+            }, BOB_NAME)
 
             // Find the future representing the result of this state machine again.
             val bobFuture = bobNode.smm.findStateMachines(BuyerAcceptor::class.java).single().second
@@ -285,12 +303,12 @@ class TwoPartyTradeFlowTests {
             assertThat(bobFuture.getOrThrow()).isEqualTo(aliceFuture.getOrThrow())
 
             assertThat(bobNode.smm.findStateMachines(Buyer::class.java)).isEmpty()
-//            bobNode.database.transaction {
+            bobNode.database.transaction {
-//                assertThat(bobNode.checkpointStorage.checkpoints()).isEmpty()
+                assertThat(bobNode.checkpointStorage.checkpoints()).isEmpty()
-//            }
+            }
-//            aliceNode.database.transaction {
+            aliceNode.database.transaction {
-//                assertThat(aliceNode.checkpointStorage.checkpoints()).isEmpty()
+                assertThat(aliceNode.checkpointStorage.checkpoints()).isEmpty()
-//            }
+            }
 
             bobNode.database.transaction {
                 val restoredBobTransactions = bobTransactionsBeforeCrash.filter {
@@ -306,18 +324,15 @@ class TwoPartyTradeFlowTests {
 
     // Creates a mock node with an overridden storage service that uses a RecordingMap, that lets us test the order
     // of gets and puts.
-    private fun makeNodeWithTracking(
+    private fun makeNodeWithTracking(name: CordaX500Name): StartedNode<MockNetwork.MockNode> {
-            networkMapAddress: SingleMessageRecipient?,
-            name: CordaX500Name): StartedNode<MockNetwork.MockNode> {
         // Create a node in the mock network ...
-        return mockNet.createNode(networkMapAddress, nodeFactory = object : MockNetwork.Factory<MockNetwork.MockNode> {
+        return mockNet.createNode(nodeFactory = object : MockNetwork.Factory<MockNetwork.MockNode> {
             override fun create(config: NodeConfiguration,
                                 network: MockNetwork,
                                 networkMapAddr: SingleMessageRecipient?,
-                                advertisedServices: Set<ServiceInfo>, id: Int,
+                                id: Int, notaryIdentity: Pair<ServiceInfo, KeyPair>?,
-                                overrideServices: Map<ServiceInfo, KeyPair>?,
                                 entropyRoot: BigInteger): MockNetwork.MockNode {
-                return object : MockNetwork.MockNode(config, network, networkMapAddr, advertisedServices, id, overrideServices, entropyRoot) {
+                return object : MockNetwork.MockNode(config, network, networkMapAddr, id, notaryIdentity, entropyRoot) {
                     // That constructs a recording tx storage
                     override fun makeTransactionStorage(): WritableTransactionStorage {
                         return RecordingTransactionStorage(database, super.makeTransactionStorage())
@@ -329,18 +344,18 @@ class TwoPartyTradeFlowTests {
 
     @Test
     fun `check dependencies of sale asset are resolved`() {
-        mockNet = MockNetwork(false)
+        mockNet = MockNetwork(false, cordappPackages = cordappPackages)
-
+        val notaryNode = mockNet.createNotaryNode()
-        val notaryNode = mockNet.createNotaryNode(null, DUMMY_NOTARY.name)
+        val aliceNode = makeNodeWithTracking(ALICE_NAME)
-        val aliceNode = makeNodeWithTracking(notaryNode.network.myAddress, ALICE.name)
+        val bobNode = makeNodeWithTracking(BOB_NAME)
-        val bobNode = makeNodeWithTracking(notaryNode.network.myAddress, BOB.name)
+        val bankNode = makeNodeWithTracking(BOC_NAME)
-        val bankNode = makeNodeWithTracking(notaryNode.network.myAddress, BOC.name)
-        val issuer = bankNode.info.chooseIdentity().ref(1, 2, 3)
         mockNet.runNetwork()
         notaryNode.internals.ensureRegistered()
         val notary = aliceNode.services.getDefaultNotary()
-
+        val alice = aliceNode.info.singleIdentity()
-        mockNet.registerIdentities()
+        val bob = bobNode.info.singleIdentity()
+        val bank = bankNode.info.singleIdentity()
+        val issuer = bank.ref(1, 2, 3)
 
         ledger(aliceNode.services, initialiseSerialization = false) {
 
@@ -356,12 +371,12 @@ class TwoPartyTradeFlowTests {
             }
 
             val bobsFakeCash = bobNode.database.transaction {
-                fillUpForBuyer(false, issuer, AnonymousParty(bobNode.info.chooseIdentity().owningKey), notary)
+                fillUpForBuyer(false, issuer, AnonymousParty(bob.owningKey), notary)
             }.second
             val bobsSignedTxns = insertFakeTransactions(bobsFakeCash, bobNode, notaryNode, bankNode)
             val alicesFakePaper = aliceNode.database.transaction {
-                fillUpForSeller(false, issuer, aliceNode.info.chooseIdentity(),
+                fillUpForSeller(false, issuer, alice,
-                        1200.DOLLARS `issued by` bankNode.info.chooseIdentity().ref(0), attachmentID, notary).second
+                        1200.DOLLARS `issued by` bank.ref(0), attachmentID, notary).second
             }
             val alicesSignedTxns = insertFakeTransactions(alicesFakePaper, aliceNode, notaryNode, bankNode)
 
@@ -436,19 +451,18 @@ class TwoPartyTradeFlowTests {
 
     @Test
     fun `track works`() {
-        mockNet = MockNetwork(false)
+        mockNet = MockNetwork(false, cordappPackages = cordappPackages)
-
+        val notaryNode = mockNet.createNotaryNode()
-        val notaryNode = mockNet.createNotaryNode(null, DUMMY_NOTARY.name)
+        val aliceNode = makeNodeWithTracking(ALICE_NAME)
-        val aliceNode = makeNodeWithTracking(notaryNode.network.myAddress, ALICE.name)
+        val bobNode = makeNodeWithTracking(BOB_NAME)
-        val bobNode = makeNodeWithTracking(notaryNode.network.myAddress, BOB.name)
+        val bankNode = makeNodeWithTracking(BOC_NAME)
-        val bankNode = makeNodeWithTracking(notaryNode.network.myAddress, BOC.name)
-        val issuer = bankNode.info.chooseIdentity().ref(1, 2, 3)
 
         mockNet.runNetwork()
         notaryNode.internals.ensureRegistered()
         val notary = aliceNode.services.getDefaultNotary()
-
+        val alice: Party = aliceNode.info.singleIdentity()
-        mockNet.registerIdentities()
+        val bank: Party = bankNode.info.singleIdentity()
+        val issuer = bank.ref(1, 2, 3)
 
         ledger(aliceNode.services, initialiseSerialization = false) {
             // Insert a prospectus type attachment into the commercial paper transaction.
@@ -469,8 +483,8 @@ class TwoPartyTradeFlowTests {
             insertFakeTransactions(bobsFakeCash, bobNode, notaryNode, bankNode)
 
             val alicesFakePaper = aliceNode.database.transaction {
-                fillUpForSeller(false, issuer, aliceNode.info.chooseIdentity(),
+                fillUpForSeller(false, issuer, alice,
-                        1200.DOLLARS `issued by` bankNode.info.chooseIdentity().ref(0), attachmentID, notary).second
+                        1200.DOLLARS `issued by` bank.ref(0), attachmentID, notary).second
             }
 
             insertFakeTransactions(alicesFakePaper, aliceNode, notaryNode, bankNode)
@@ -519,16 +533,16 @@ class TwoPartyTradeFlowTests {
 
     @Test
     fun `dependency with error on buyer side`() {
-        mockNet = MockNetwork(false)
+        mockNet = MockNetwork(false, cordappPackages = cordappPackages)
-        ledger(initialiseSerialization = false) {
+        ledger(MockServices(cordappPackages), initialiseSerialization = false) {
             runWithError(true, false, "at least one cash input")
         }
     }
 
     @Test
     fun `dependency with error on seller side`() {
-        mockNet = MockNetwork(false)
+        mockNet = MockNetwork(false, cordappPackages = cordappPackages)
-        ledger(initialiseSerialization = false) {
+        ledger(MockServices(cordappPackages), initialiseSerialization = false) {
             runWithError(false, true, "Issuances have a time-window")
         }
     }
@@ -543,8 +557,7 @@ class TwoPartyTradeFlowTests {
     private fun runBuyerAndSeller(notary: Party,
                                   sellerNode: StartedNode<MockNetwork.MockNode>,
                                   buyerNode: StartedNode<MockNetwork.MockNode>,
-                                  assetToSell: StateAndRef<OwnableState>,
+                                  assetToSell: StateAndRef<OwnableState>): RunResult {
-                                  anonymous: Boolean = true): RunResult {
         val buyerFlows: Observable<out FlowLogic<*>> = buyerNode.internals.registerInitiatedFlow(BuyerAcceptor::class.java)
         val firstBuyerFiber = buyerFlows.toFuture().map { it.stateMachine }
         val seller = SellerInitiator(buyerNode.info.chooseIdentity(), notary, assetToSell, 1000.DOLLARS, anonymous)
@@ -595,26 +608,24 @@ class TwoPartyTradeFlowTests {
             aliceError: Boolean,
             expectedMessageSubstring: String
     ) {
-        val notaryNode = mockNet.createNotaryNode(null, DUMMY_NOTARY.name)
+        val notaryNode = mockNet.createNotaryNode()
-        val aliceNode = mockNet.createPartyNode(notaryNode.network.myAddress, ALICE.name)
+        val aliceNode = mockNet.createPartyNode(ALICE_NAME)
-        val bobNode = mockNet.createPartyNode(notaryNode.network.myAddress, BOB.name)
+        val bobNode = mockNet.createPartyNode(BOB_NAME)
-        val bankNode = mockNet.createPartyNode(notaryNode.network.myAddress, BOC.name)
+        val bankNode = mockNet.createPartyNode(BOC_NAME)
-        val issuer = bankNode.info.chooseIdentity().ref(1, 2, 3)
 
         mockNet.runNetwork()
         notaryNode.internals.ensureRegistered()
         val notary = aliceNode.services.getDefaultNotary()
-
+        val alice = aliceNode.info.singleIdentity()
-        // Let the nodes know about each other - normally the network map would handle this
+        val bob = bobNode.info.singleIdentity()
-        mockNet.registerIdentities()
+        val bank = bankNode.info.singleIdentity()
+        val issuer = bank.ref(1, 2, 3)
 
         val bobsBadCash = bobNode.database.transaction {
-            fillUpForBuyer(bobError, issuer, bobNode.info.chooseIdentity(),
+            fillUpForBuyer(bobError, issuer, bob, notary).second
-                    notary).second
         }
         val alicesFakePaper = aliceNode.database.transaction {
-            fillUpForSeller(aliceError, issuer, aliceNode.info.chooseIdentity(),
+            fillUpForSeller(aliceError, issuer, alice,1200.DOLLARS `issued by` issuer, null, notary).second
-                    1200.DOLLARS `issued by` issuer, null, notary).second
         }
 
         insertFakeTransactions(bobsBadCash, bobNode, notaryNode, bankNode)
@@ -680,8 +691,8 @@ class TwoPartyTradeFlowTests {
         // wants to sell to Bob.
         val eb1 = transaction(transactionBuilder = TransactionBuilder(notary = notary)) {
             // Issued money to itself.
-            output(Cash.PROGRAM_ID, "elbonian money 1", notary = notary) { 800.DOLLARS.CASH `issued by` issuer `owned by` interimOwner }
+            output(Cash.PROGRAM_ID, "elbonian money 1", notary = notary) { 800.DOLLARS.CASH issuedBy issuer ownedBy interimOwner }
-            output(Cash.PROGRAM_ID, "elbonian money 2", notary = notary) { 1000.DOLLARS.CASH `issued by` issuer `owned by` interimOwner }
+            output(Cash.PROGRAM_ID, "elbonian money 2", notary = notary) { 1000.DOLLARS.CASH issuedBy issuer ownedBy interimOwner }
             if (!withError) {
                 command(issuer.party.owningKey) { Cash.Commands.Issue() }
             } else {
@@ -699,15 +710,15 @@ class TwoPartyTradeFlowTests {
         // Bob gets some cash onto the ledger from BoE
         val bc1 = transaction(transactionBuilder = TransactionBuilder(notary = notary)) {
             input("elbonian money 1")
-            output(Cash.PROGRAM_ID, "bob cash 1", notary = notary) { 800.DOLLARS.CASH `issued by` issuer `owned by` owner }
+            output(Cash.PROGRAM_ID, "bob cash 1", notary = notary) { 800.DOLLARS.CASH issuedBy issuer ownedBy owner }
             command(interimOwner.owningKey) { Cash.Commands.Move() }
             this.verifies()
         }
 
         val bc2 = transaction(transactionBuilder = TransactionBuilder(notary = notary)) {
             input("elbonian money 2")
-            output(Cash.PROGRAM_ID, "bob cash 2", notary = notary) { 300.DOLLARS.CASH `issued by` issuer `owned by` owner }
+            output(Cash.PROGRAM_ID, "bob cash 2", notary = notary) { 300.DOLLARS.CASH issuedBy issuer ownedBy owner }
-            output(Cash.PROGRAM_ID, notary = notary) { 700.DOLLARS.CASH `issued by` issuer `owned by` interimOwner }   // Change output.
+            output(Cash.PROGRAM_ID, notary = notary) { 700.DOLLARS.CASH issuedBy issuer ownedBy interimOwner }   // Change output.
             command(interimOwner.owningKey) { Cash.Commands.Move() }
             this.verifies()
         }