Ack unhandled messages when recipient flow is shut or shutting down (#335)

.idea/compiler.xml

@@ -118,6 +118,7 @@
       <module name="node_test" target="1.8" />
       <module name="notary-demo_main" target="1.8" />
       <module name="notary-demo_test" target="1.8" />
+      <module name="perftestcordapp_integrationTest" target="1.8" />
       <module name="perftestcordapp_main" target="1.8" />
       <module name="perftestcordapp_test" target="1.8" />
       <module name="publish-utils_main" target="1.8" />

node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineManagerImpl.kt

@@ -324,6 +324,7 @@ class StateMachineManagerImpl(
             val recipientId = sessionMessage.recipientSessionId
             val flowId = sessionToFlow[recipientId]
             if (flowId == null) {
+                acknowledgeHandle.acknowledge()
                 if (sessionMessage.payload is EndSessionMessage) {
                     logger.debug {
                         "Got ${EndSessionMessage::class.java.simpleName} for " +
@@ -610,6 +611,7 @@ class StateMachineManagerImpl(
             removalReason: FlowRemovalReason.OrderlyFinish,
             lastState: StateMachineState
     ) {
+        drainFlowEventQueue(flow)
         // final sanity checks
         require(lastState.unacknowledgedMessages.isEmpty())
         require(lastState.isRemoved)
@@ -625,6 +627,7 @@ class StateMachineManagerImpl(
             removalReason: FlowRemovalReason.ErrorFinish,
             lastState: StateMachineState
     ) {
+        drainFlowEventQueue(flow)
         val flowError = removalReason.flowErrors[0] // TODO what to do with several?
         val exception = flowError.exception
         (exception as? FlowException)?.originalErrorId = flowError.errorId
@@ -632,6 +635,30 @@ class StateMachineManagerImpl(
         lastState.flowLogic.progressTracker?.endWithError(exception)
         changesPublisher.onNext(StateMachineManager.Change.Removed(lastState.flowLogic, Try.Failure<Nothing>(exception)))
     }
+
+    // The flow's event queue may be non-empty in case it shut down abruptly. We handle outstanding events here.
+    private fun drainFlowEventQueue(flow: Flow) {
+        while (true) {
+            val event = flow.fiber.transientValues!!.value.eventQueue.tryReceive() ?: return
+            when (event) {
+                is Event.DeliverSessionMessage -> {
+                    // Acknowledge the message so it doesn't leak in the broker.
+                    event.acknowledgeHandle.acknowledge()
+                    when (event.sessionMessage.payload) {
+                        EndSessionMessage -> {
+                            logger.debug { "Unhandled message ${event.sessionMessage} due to flow shutting down" }
+                        }
+                        else -> {
+                            logger.warn("Unhandled message ${event.sessionMessage} due to flow shutting down")
+                        }
+                    }
+                }
+                else -> {
+                    logger.warn("Unhandled event $event due to flow shutting down")
+                }
+            }
+        }
+    }
 }
 
 class SessionRejectException(reason: String) : CordaException(reason)

perftestcordapp/build.gradle

@@ -10,6 +10,30 @@ apply plugin: 'net.corda.plugins.cordapp'
 
 description 'Corda performance test modules'
 
+sourceSets {
+    integrationTest {
+        kotlin {
+            compileClasspath += main.output + test.output
+            runtimeClasspath += main.output + test.output
+            srcDir file('src/integration-test/kotlin')
+        }
+        java {
+            compileClasspath += main.output + test.output
+            runtimeClasspath += main.output + test.output
+            srcDir file('src/integration-test/java')
+        }
+        resources {
+            srcDir file('src/integration-test/resources')
+            srcDir file('../../testing/test-utils/src/main/resources')
+        }
+    }
+}
+
+task integrationTest(type: Test) {
+    testClassesDirs = sourceSets.integrationTest.output.classesDirs
+    classpath = sourceSets.integrationTest.runtimeClasspath
+}
+
 dependencies {
     // Note the :finance module is a CorDapp in its own right
     // and CorDapps using :finance features should use 'cordapp' not 'compile' linkage.
@@ -29,6 +53,9 @@ dependencies {
 
 configurations {
     testArtifacts.extendsFrom testRuntime
+
+    integrationTestCompile.extendsFrom testCompile
+    integrationTestRuntime.extendsFrom testRuntime
 }
 
 task testJar(type: Jar) {

perftestcordapp/src/integrationTest/kotlin/com/r3/corda/enterprise/perftestcordapp/flows/NoSelectionIntegrationTest.kt

@@ -0,0 +1,33 @@
+package com.r3.corda.enterprise.perftestcordapp.flows
+
+import net.corda.core.messaging.startFlow
+import net.corda.core.utilities.OpaqueBytes
+import net.corda.core.utilities.getOrThrow
+import net.corda.finance.DOLLARS
+import net.corda.node.services.Permissions
+import net.corda.nodeapi.internal.config.User
+import net.corda.testing.driver.PortAllocation
+import net.corda.testing.driver.driver
+import org.junit.Ignore
+import org.junit.Test
+
+@Ignore("Use to test no-selection locally")
+class NoSelectionIntegrationTest {
+
+    @Test
+    fun `single pay no selection`() {
+        val aliceUser = User("A", "A", setOf(Permissions.startFlow<CashIssueAndPaymentNoSelection>()))
+        driver(
+                startNodesInProcess = true,
+                extraCordappPackagesToScan = listOf("com.r3.corda.enterprise.perftestcordapp"),
+                portAllocation = PortAllocation.Incremental(20000)
+        ) {
+            val alice = startNode(rpcUsers = listOf(aliceUser)).get()
+            defaultNotaryNode.get()
+            alice.rpcClientToNode().use("A", "A") { connection ->
+                connection.proxy.startFlow(::CashIssueAndPaymentNoSelection, 1.DOLLARS, OpaqueBytes.of(0), alice.nodeInfo.legalIdentities[0], false, defaultNotaryIdentity).returnValue.getOrThrow()
+            }
+        }
+    }
+
+}