make loading more async

build.gradle

@@ -5,7 +5,7 @@ buildscript {
         corda_release_group = 'net.corda'
         corda_release_version = '4.3'
         tokens_release_group = "com.r3.corda.lib.tokens"
-        tokens_release_version = "1.2.4-SNAPSHOT"
+        tokens_release_version = "1.2.5-ASYNC-LOADING-SNAPSHOT"
         corda_gradle_plugins_version = '5.0.12'
         kotlin_version = '1.2.71'
         junit_version = '4.12'

contracts/build.gradle

@@ -47,7 +47,14 @@ cordapp {
         versionId 2
     }
     signing {
-        enabled false
+        enabled true
+        options {
+            keystore "./signing-infra/keyStore.jks"
+            alias "cordapp_signing"
+            storepass "monkey"
+            keypass "monkey"
+            storetype "PKCS12"
+        }
     }
 }
 

freighter-tests/build.gradle

@@ -26,8 +26,13 @@ task freighterTest(type: Test, dependsOn: [project(":workflows").jar]) {
     classpath = sourceSets.freighterTest.runtimeClasspath
     useJUnitPlatform {
         includeTags "DOCKER"
-        excludeTags "AZURE", "FULL_LINUX_KERNEL", "ORACLE"
+        excludeTags "AZURE", "FULL_LINUX_KERNEL", "ORACLE", "LARGE_TEST"
     }
+
+    testLogging {
+        showStandardStreams = true
+    }
+
 }
 
 configurations {
@@ -37,10 +42,11 @@ configurations {
 
 dependencies {
     freighterTestCompile "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlin_version"
-    freighterTestCompile "freighter:freighter-testing-core-junit5:0.7.3-TEST-SNAPSHOT"
+    freighterTestCompile "freighter:freighter-testing-core-junit5:0.9.0-SNAPSHOT"
 
     freighterTestCompile project(":contracts")
     freighterTestCompile project(":workflows")
+    freighterTestCompile project(":workflows-integration-test")
 }
 
 

freighter-tests/src/freighterTest/kotlin/freighter/testing/HugeTokensLoadedOnRestartTest.kt

@@ -0,0 +1,170 @@
+package freighter.testing
+
+import com.r3.corda.lib.tokens.contracts.states.FungibleToken
+import com.r3.corda.lib.tokens.contracts.types.IssuedTokenType
+import com.r3.corda.lib.tokens.contracts.types.TokenType
+import com.r3.corda.lib.tokens.integration.workflows.LockEverythingGetValue
+import com.r3.corda.lib.tokens.workflows.flows.rpc.IssueTokens
+import com.stress.flows.CreateNewCIFlow
+import freighter.deployments.DeploymentContext
+import freighter.deployments.NodeBuilder
+import freighter.deployments.SingleNodeDeployment
+import freighter.deployments.UnitOfDeployment
+import freighter.machine.DeploymentMachineProvider
+import freighter.machine.generateRandomString
+import net.corda.core.contracts.Amount
+import net.corda.core.messaging.startFlow
+import net.corda.core.utilities.getOrThrow
+import org.junit.jupiter.api.Tag
+import org.junit.jupiter.api.Test
+import utility.getOrThrow
+import java.time.Duration
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.locks.ReentrantLock
+import java.util.stream.StreamSupport
+import kotlin.concurrent.withLock
+import kotlin.streams.toList
+
+@Tag("LARGE_TEST")
+annotation class LargeTest
+
+class HugeTokensLoadedOnRestartTest : DockerRemoteMachineBasedTest() {
+
+    val loadingThreads = 8
+    val pageSize = 10_000
+
+    val tokenCurrentContracts =
+            NodeBuilder.DeployedCordapp.fromClassPath("tokens-contracts").signedWithFreighterKey()
+
+    val testFlows =
+            NodeBuilder.DeployedCordapp.fromClassPath("workflows-integration-test").signedWithFreighterKey()
+
+    val tokenCurrentWorkflows =
+            NodeBuilder.DeployedCordapp.fromClassPath("tokens-workflows").withConfig(
+                    """
+				stateSelection.inMemory.enabled=true
+				stateSelection.inMemory.indexingStrategies=[EXTERNAL_ID, PUBLIC_KEY]
+				stateSelection.inMemory.pageSize=${pageSize}
+				stateSelection.inMemory.loadingSleepSeconds=-1
+				stateSelection.inMemory.loadingThreads=${loadingThreads}
+			""".trimIndent().byteInputStream()
+            )
+
+    val modernCiV1 = NodeBuilder.DeployedCordapp.fromGradleArtifact(
+            group = "com.r3.corda.lib.ci",
+            artifact = "ci-workflows",
+            version = "1.0"
+    )
+
+    val freighterHelperCordapp = NodeBuilder.DeployedCordapp.fromClassPath("freighter-cordapp-flows")
+
+    @Test
+    fun `tokens can be loaded async during node startup on postgres 9_6`() {
+        run(DeploymentMachineProvider.DatabaseType.PG_9_6)
+    }
+
+    @Test
+    fun `tokens can be loaded async during node startup on H2`() {
+        run(DeploymentMachineProvider.DatabaseType.H2)
+    }
+
+    private fun run(db: DeploymentMachineProvider.DatabaseType) {
+        val randomString = generateRandomString()
+        val deploymentContext = DeploymentContext(machineProvider, nms, artifactoryUsername, artifactoryPassword)
+        val node1 = SingleNodeDeployment(
+                NodeBuilder().withX500("O=PartyB, C=GB, L=LONDON, CN=$randomString")
+                        .withCordapp(tokenCurrentContracts)
+                        .withCordapp(tokenCurrentWorkflows)
+                        .withCordapp(modernCiV1)
+                        .withCordapp(freighterHelperCordapp)
+                        .withCordapp(testFlows)
+                        .withDatabase(machineProvider.requestDatabase(db))
+        ).withVersion(UnitOfDeployment.CORDA_4_7)
+                .deploy(deploymentContext)
+
+        val nodeMachine1 = node1.getOrThrow().nodeMachines.single()
+
+        val createdCi = nodeMachine1.rpc {
+            startFlow(::CreateNewCIFlow).returnValue.getOrThrow().also {
+                println("Successfully created CI: $it")
+            }
+        }
+
+        val tokenType = TokenType("StefCoin", 2)
+        val issuedTokenType = IssuedTokenType(nodeMachine1.identity(), tokenType)
+        val issuedTotal = AtomicLong(0)
+
+        val tokenToIssue = (0.until(100)).map { FungibleToken(Amount(1, issuedTokenType), createdCi) }.toList()
+
+        val numberIssued = StreamSupport.stream((0.until(500)).chunked(200).spliterator(), true).map { toIssue ->
+            nodeMachine1.rpc {
+                repeat(toIssue.size) {
+                    startFlow(
+                            ::IssueTokens,
+                            tokenToIssue, listOf()
+                    ).returnValue.getOrThrow(Duration.ofMinutes(1))
+                    println("[${Thread.currentThread().name}] Total number issued: ${issuedTotal.addAndGet(tokenToIssue.size * 1L)}")
+                }
+            }
+
+            toIssue.size * tokenToIssue.size
+        }.toList().sum()
+
+        nodeMachine1.stopNode()
+        println()
+        nodeMachine1.startNode()
+
+        val nodeStartTime = System.currentTimeMillis()
+
+        var amountLoaded = 0L
+
+        val lock = ReentrantLock()
+        val condition = lock.newCondition()
+
+
+        //whilst we are loading, issue 500 more tokens to see if they are correctly loaded
+        CompletableFuture.runAsync {
+            lock.withLock {
+                condition.await()
+            }
+            nodeMachine1.rpc {
+                repeat(5) {
+                    startFlow(
+                            ::IssueTokens,
+                            tokenToIssue, listOf()
+                    ).returnValue.getOrThrow(Duration.ofMinutes(1))
+                    println("[${Thread.currentThread().name}] Total number issued: ${issuedTotal.addAndGet(tokenToIssue.size * 1L)}")
+                }
+            }
+        }
+
+        while (amountLoaded != (issuedTotal.toLong())) {
+            nodeMachine1.rpc {
+                amountLoaded = startFlow(
+                        ::LockEverythingGetValue,
+                        tokenType
+                ).returnValue.getOrThrow(Duration.ofMinutes(1))
+            }
+            println("TOTAL TOKEN VALUE IN CACHE: $amountLoaded")
+
+            if (amountLoaded > 0) {
+                lock.withLock {
+                    condition.signal()
+                }
+            }
+
+            if (amountLoaded != issuedTotal.toLong()) {
+                Thread.sleep(20000)
+            }
+        }
+
+        val loadEndTime = System.currentTimeMillis()
+
+        println("it took: ${(loadEndTime - nodeStartTime) / 1000} seconds to populate $amountLoaded states using $loadingThreads loading threads and pageSize: $pageSize")
+
+
+    }
+
+
+}

freighter-tests/src/freighterTest/kotlin/freighter/testing/NullHolderOnObserverTest.kt

@@ -67,12 +67,6 @@ class NullHolderOnObserverTest : DockerRemoteMachineBasedTest() {
 		runTokensOnNodeRunningDatabase(DeploymentMachineProvider.DatabaseType.MS_SQL)
 	}
 
-	@Test
-	@OracleTest
-	fun `tokens can be observed on node that does not know CI running oracle 12 r2`() {
-		runTokensOnNodeRunningDatabase(DeploymentMachineProvider.DatabaseType.ORACLE_12_R2)
-	}
-
 	private fun runTokensOnNodeRunningDatabase(db: DeploymentMachineProvider.DatabaseType) {
 		val randomString = generateRandomString()
 		val deploymentContext = DeploymentContext(machineProvider, nms, artifactoryUsername, artifactoryPassword)
@@ -83,7 +77,7 @@ class NullHolderOnObserverTest : DockerRemoteMachineBasedTest() {
 				.withCordapp(modernCiV1)
 				.withCordapp(freighterHelperCordapp)
 				.withDatabase(machineProvider.requestDatabase(db))
-		).withVersion(UnitOfDeployment.CORDA_4_6)
+		).withVersion(UnitOfDeployment.CORDA_4_7)
 			.deploy(deploymentContext)
 
 		val node2 = SingleNodeDeployment(
@@ -93,7 +87,7 @@ class NullHolderOnObserverTest : DockerRemoteMachineBasedTest() {
 				.withCordapp(modernCiV1)
 				.withCordapp(freighterHelperCordapp)
 				.withDatabase(machineProvider.requestDatabase(db))
-		).withVersion(UnitOfDeployment.CORDA_4_6)
+		).withVersion(UnitOfDeployment.CORDA_4_7)
 			.deploy(deploymentContext)
 
 		val nodeMachine1 = node1.getOrThrow().nodeMachines.single()