CORDA-540: Ensure that registration of a test node was a success prior to performing further testing with it (#1379)

Registration may fail due to low level serialization problems especially when running in AMQP mode
Also some minor improvements for exceptions reporting and test coverage

core/src/test/java/net/corda/core/flows/FlowsInJavaTest.java

@@ -7,6 +7,7 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
@@ -18,11 +19,13 @@ public class FlowsInJavaTest {
     private MockNetwork.MockNode node2;
 
     @Before
-    public void setUp() {
+    public void setUp() throws Exception {
         MockNetwork.BasketOfNodes someNodes = mockNet.createSomeNodes(2);
         node1 = someNodes.getPartyNodes().get(0);
         node2 = someNodes.getPartyNodes().get(1);
         mockNet.runNetwork();
+        // Ensure registration was successful
+        node1.getNodeReadyFuture().get();
     }
 
     @After

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

@@ -56,6 +56,11 @@ class AttachmentTests {
         val nodes = mockNet.createSomeNodes(2)
         val n0 = nodes.partyNodes[0]
         val n1 = nodes.partyNodes[1]
+
+        // Ensure that registration was successful before progressing any further
+        mockNet.runNetwork()
+        n0.ensureRegistered()
+
         n0.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
         n1.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
 
@@ -89,6 +94,11 @@ class AttachmentTests {
         val nodes = mockNet.createSomeNodes(2)
         val n0 = nodes.partyNodes[0]
         val n1 = nodes.partyNodes[1]
+
+        // Ensure that registration was successful before progressing any further
+        mockNet.runNetwork()
+        n0.ensureRegistered()
+
         n0.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
         n1.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
 
@@ -119,6 +129,10 @@ class AttachmentTests {
         }, advertisedServices = *arrayOf(ServiceInfo(NetworkMapService.type), ServiceInfo(SimpleNotaryService.type)))
         val n1 = mockNet.createNode(n0.network.myAddress)
 
+        // Ensure that registration was successful before progressing any further
+        mockNet.runNetwork()
+        n0.ensureRegistered()
+
         n0.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
         n1.registerInitiatedFlow(FetchAttachmentsResponse::class.java)
 

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

@@ -36,6 +36,7 @@ class CollectSignaturesFlowTests {
         c = nodes.partyNodes[2]
         notary = nodes.notaryNode.info.notaryIdentity
         mockNet.runNetwork()
+        a.ensureRegistered()
     }
 
     @After

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

@@ -46,6 +46,11 @@ class ContractUpgradeFlowTest {
         val nodes = mockNet.createSomeNodes(notaryKeyPair = null) // prevent generation of notary override
         a = nodes.partyNodes[0]
         b = nodes.partyNodes[1]
+
+        // Process registration
+        mockNet.runNetwork()
+        a.ensureRegistered()
+
         notary = nodes.notaryNode.info.notaryIdentity
 
         val nodeIdentity = nodes.notaryNode.info.legalIdentitiesAndCerts.single { it.party == nodes.notaryNode.info.notaryIdentity }

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

@@ -29,6 +29,7 @@ class FinalityFlowTests {
         nodeB = nodes.partyNodes[1]
         notary = nodes.notaryNode.info.notaryIdentity
         mockNet.runNetwork()
+        nodeA.ensureRegistered()
     }
 
     @After

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

@@ -32,6 +32,7 @@ class ManualFinalityFlowTests {
         nodeC = nodes.partyNodes[2]
         notary = nodes.notaryNode.info.notaryIdentity
         mockNet.runNetwork()
+        nodeA.ensureRegistered()
     }
 
     @After

core/src/test/kotlin/net/corda/core/internal/concurrent/CordaFutureImplTest.kt

@@ -112,6 +112,18 @@ class CordaFutureTest {
         }
         verify(log).error(any(), same(throwable))
     }
+
+    @Test
+    fun `captureLater works`() {
+        val failingFuture = CordaFutureImpl<Int>()
+        val anotherFailingFuture = CordaFutureImpl<Int>()
+        anotherFailingFuture.captureLater(failingFuture)
+
+        val exception = Exception()
+        failingFuture.setException(exception)
+
+        Assertions.assertThatThrownBy { anotherFailingFuture.getOrThrow() }.isSameAs(exception)
+    }
 }
 
 class TransposeTest {

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

@@ -46,13 +46,13 @@ private fun MockNetwork.MockNode.saveAttachment(content: String) = database.tran
     attachments.importAttachment(createAttachmentData(content).inputStream())
 }
 private fun MockNetwork.MockNode.hackAttachment(attachmentId: SecureHash, content: String) = database.transaction {
-    attachments.updateAttachment(attachmentId, createAttachmentData(content))
+    updateAttachment(attachmentId, createAttachmentData(content))
 }
 
 /**
  * @see NodeAttachmentService.importAttachment
  */
-private fun NodeAttachmentService.updateAttachment(attachmentId: SecureHash, data: ByteArray) {
+private fun updateAttachment(attachmentId: SecureHash, data: ByteArray) {
     val session = DatabaseTransactionManager.current().session
     val attachment = session.get<NodeAttachmentService.DBAttachment>(NodeAttachmentService.DBAttachment::class.java, attachmentId.toString())
     attachment?.let {
@@ -73,6 +73,7 @@ class AttachmentSerializationTest {
         client = mockNet.createNode(server.network.myAddress)
         client.disableDBCloseOnStop() // Otherwise the in-memory database may disappear (taking the checkpoint with it) while we reboot the client.
         mockNet.runNetwork()
+        server.ensureRegistered()
     }
 
     @After

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

@@ -605,8 +605,8 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
         val address: SingleMessageRecipient = networkMapAddress ?:
                 network.getAddressOfParty(PartyInfo.Node(info)) as SingleMessageRecipient
         // Register for updates, even if we're the one running the network map.
-        return sendNetworkMapRegistration(address).flatMap { (error) ->
+        return sendNetworkMapRegistration(address).flatMap { response: RegistrationResponse ->
-            check(error == null) { "Unable to register with the network map service: $error" }
+            check(response.error == null) { "Unable to register with the network map service: ${response.error}" }
             // The future returned addMapService will complete on the same executor as sendNetworkMapRegistration, namely the one used by net
             services.networkMapCache.addMapService(network, address, true, null)
         }

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

@@ -330,7 +330,9 @@ open class Node(override val configuration: FullNodeConfiguration,
 
                 _startupComplete.set(Unit)
             }
-        }, {})
+        },
+        { th -> logger.error("Unexpected exception", th)}
+        )
         shutdownHook = addShutdownHook {
             stop()
         }

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

@@ -117,7 +117,10 @@ open class NodeStartup(val args: Array<String>) {
                     logger.error("Shell failed to start", e)
                 }
             }
-        }, {})
+        },
+        {
+            th -> logger.error("Unexpected exception during registration", th)
+        })
         node.run()
     }
 

node/src/main/kotlin/net/corda/node/services/network/NetworkMapService.kt

@@ -35,6 +35,7 @@ import net.corda.node.services.network.NetworkMapService.Companion.SUBSCRIPTION_
 import net.corda.node.utilities.AddOrRemove
 import net.corda.node.utilities.AddOrRemove.ADD
 import net.corda.node.utilities.AddOrRemove.REMOVE
+import java.io.IOException
 import java.security.PublicKey
 import java.security.SignatureException
 import java.time.Instant
@@ -243,6 +244,10 @@ abstract class AbstractNetworkMapService(services: ServiceHubInternal,
             request.wireReg.verified()
         } catch (e: SignatureException) {
             return RegistrationResponse("Invalid signature on request")
+        } catch (e: IOException) {
+            val msg = "Unexpected IO exception: ${e.message}"
+            logger.error(msg, e)
+            return RegistrationResponse(msg)
         }
 
         val node = change.node

test-utils/src/main/kotlin/net/corda/testing/node/MockNode.kt

@@ -268,6 +268,15 @@ class MockNetwork(private val networkSendManuallyPumped: Boolean = false,
                 }
             })
         }
+
+        /**
+         * Makes sure that the [MockNode] is correctly registered on the [MockNetwork]
+         * Please note that [MockNetwork.runNetwork] should be invoked to ensure that all the pending registration requests
+         * were duly processed
+         */
+        fun ensureRegistered() {
+            _nodeReadyFuture.getOrThrow()
+        }
     }
 
     /**