Detect duplicate inputs in NotaryFlow

Throw NotaryException when duplicate inputs are detected.

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

@@ -1,6 +1,7 @@
 package net.corda.flows
 
 import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.contracts.StateRef
 import net.corda.core.crypto.*
 import net.corda.core.flows.FlowLogic
 import net.corda.core.node.services.TimestampChecker
@@ -129,21 +130,36 @@ object NotaryFlow {
         open fun beforeCommit(stx: SignedTransaction) {
         }
 
+        /** Throw NotaryException if inputs are not unique. */
+        private fun detectDuplicateInputs(tx: WireTransaction) {
+            var seenInputs = emptySet<StateRef>()
+            var conflicts = emptyMap<StateRef, UniquenessProvider.ConsumingTx>()
+            tx.inputs.forEachIndexed { i, stateRef ->
+                if (seenInputs.contains(stateRef)) {
+                    conflicts += stateRef.to(UniquenessProvider.ConsumingTx(tx.id, i, otherSide))
+                }
+                seenInputs += stateRef
+            }
+            if (conflicts.isNotEmpty()) {
+                throw notaryException(tx, UniquenessException(UniquenessProvider.Conflict(conflicts)))
+            }
+        }
+
+        /** A NotaryException is thrown if any of the states have been consumed by a different transaction or input
+         * states are present multiple times within this transaction.
+         */
         private fun commitInputStates(tx: WireTransaction) {
+            detectDuplicateInputs(tx)
             try {
                 uniquenessProvider.commit(tx.inputs, tx.id, otherSide)
             } catch (e: UniquenessException) {
-                // Allow re-committing the transaction to make the NotaryFlow idempotent. Alternatively, we could make
-                // the underlying UniquenessProviders idempotent.
                 val conflicts = tx.inputs.filterIndexed { i, stateRef ->
                     val consumingTx = e.error.stateHistory[stateRef]
                     consumingTx != null && consumingTx != UniquenessProvider.ConsumingTx(tx.id, i, otherSide)
                 }
                 if (conflicts.isNotEmpty()) {
                     // TODO: Create a new UniquenessException that only contains the conflicts filtered above.
-                    val conflictData = e.error.serialize()
+                    throw notaryException(tx, e)
-                    val signedConflict = SignedData(conflictData, sign(conflictData.bytes))
-                    throw NotaryException(NotaryError.Conflict(tx, signedConflict))
                 }
             }
         }
@@ -152,6 +168,12 @@ object NotaryFlow {
             val mySigningKey = serviceHub.notaryIdentityKey
             return mySigningKey.signWithECDSA(bits)
         }
+
+        private fun notaryException(tx: WireTransaction, e: UniquenessException): NotaryException {
+            val conflictData = e.error.serialize()
+            val signedConflict = SignedData(conflictData, sign(conflictData.bytes))
+            return NotaryException(NotaryError.Conflict(tx, signedConflict))
+        }
     }
 
     data class SignRequest(val tx: SignedTransaction)

node/src/test/kotlin/net/corda/node/services/NotaryServiceTests.kt

@@ -95,15 +95,34 @@ class NotaryServiceTests {
 
         val firstAttempt = NotaryFlow.Client(stx)
         val secondAttempt = NotaryFlow.Client(stx)
-        clientNode.services.startFlow(firstAttempt)
+        val f1 = clientNode.services.startFlow(firstAttempt)
-        val future = clientNode.services.startFlow(secondAttempt)
+        val f2 = clientNode.services.startFlow(secondAttempt)
 
         net.runNetwork()
 
-        future.resultFuture.getOrThrow()
+        assertEquals(f1.resultFuture.getOrThrow(), f2.resultFuture.getOrThrow())
     }
 
-    @Test fun `should report conflict when inputs are reused accross transactions`() {
+    @Test fun `should report conflict when inputs are reused within a single transactions`() {
+        val stx = run {
+            val inputState = issueState(clientNode)
+            // Use inputState twice as input of a single transaction.
+            val tx = TransactionType.General.Builder(notaryNode.info.notaryIdentity).withItems(inputState, inputState)
+            tx.signWith(clientNode.keyPair!!)
+            tx.toSignedTransaction(false)
+        }
+
+        val future = clientNode.services.startFlow(NotaryFlow.Client(stx))
+
+        net.runNetwork()
+
+        val ex = assertFailsWith(NotaryException::class) { future.resultFuture.getOrThrow() }
+        val notaryError = ex.error as NotaryError.Conflict
+        assertEquals(notaryError.tx, stx.tx)
+        notaryError.conflict.verified()
+    }
+
+    @Test fun `should report conflict when inputs are reused across transactions`() {
         val inputState = issueState(clientNode)
         val stx = run {
             val tx = TransactionType.General.Builder(notaryNode.info.notaryIdentity).withItems(inputState)