ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-01-31 16:35:43 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge pull request #910 from corda/tudor-merge-os-01_06

Browse Source 
Tudor merge os 01 06
This commit is contained in:
Tudor Malene 2018-06-04 15:37:33 +01:00 committed by GitHub
commit 4f1a22dfac
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 293 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CONTRIBUTORS.md
View File

@ -127,9 +127,11 @@ see changes to this list.
* Mike Hearn (R3)
* Mike Ward (R3)
* Mike Reichelt (US Bank)
* Mohamed Amine LEGHERABA
* Mustafa Ozturk (Natixis)
* Nick Skinner (Northern Trust)
* Nigel King (R3)
* Nitesh Solanki (Persistent Systems Limited)
* Nuam Athaweth (MUFG)
* Oscar Zibordi de Paiva (Bradesco)
* OP Financial

2
README.md
View File

@ -7,7 +7,7 @@
Corda Enterprise is R3's closed source patch set on top of Corda Open Source. It adds features and improvements that we
plan to charge for.
Corda is a decentralised database system in which nodes trust each other as little as possible.
Corda is an open source blockchain project, designed for business from the start. Only Corda allows you to build interoperable blockchain networks that transact in strict privacy. Corda's smart contract technology allows businesses to transact directly, with value.
## Features

5
core/src/main/kotlin/net/corda/core/cordapp/Cordapp.kt
View File

@ -11,6 +11,7 @@
package net.corda.core.cordapp
import net.corda.core.DoNotImplement
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.FlowLogic
import net.corda.core.schemas.MappedSchema
import net.corda.core.serialization.SerializationCustomSerializer
@ -35,7 +36,9 @@ import java.net.URL
* @property serializationWhitelists List of Corda plugin registries
* @property serializationCustomSerializers List of serializers
* @property customSchemas List of custom schemas
* @property allFlows List of all flow classes
* @property jarPath The path to the JAR for this CorDapp
* @property jarHash Hash of the jar
*/
@DoNotImplement
interface Cordapp {
@ -49,9 +52,11 @@ interface Cordapp {
val serializationWhitelists: List<SerializationWhitelist>
val serializationCustomSerializers: List<SerializationCustomSerializer<*, *>>
val customSchemas: Set<MappedSchema>
val allFlows: List<Class<out FlowLogic<*>>>
val jarPath: URL
val cordappClasses: List<String>
val info: Info
val jarHash: SecureHash.SHA256
/**
* CorDapp's information, including vendor and version.

5
core/src/main/kotlin/net/corda/core/internal/cordapp/CordappImpl.kt
View File

@ -11,6 +11,7 @@
package net.corda.core.internal.cordapp
import net.corda.core.cordapp.Cordapp
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.FlowLogic
import net.corda.core.internal.toPath
import net.corda.core.schemas.MappedSchema
@ -29,9 +30,11 @@ data class CordappImpl(
override val serializationWhitelists: List<SerializationWhitelist>,
override val serializationCustomSerializers: List<SerializationCustomSerializer<*, *>>,
override val customSchemas: Set<MappedSchema>,
override val allFlows: List<Class<out FlowLogic<*>>>,
override val jarPath: URL,
override val info: Cordapp.Info = CordappImpl.Info.UNKNOWN,
override val name: String = jarPath.toPath().fileName.toString().removeSuffix(".jar")) : Cordapp {
override val jarHash: SecureHash.SHA256,
override val name: String = jarPath.toPath().fileName.toString().removeSuffix(".jar") ) : Cordapp {
/**
* An exhaustive list of all classes relevant to the node within this CorDapp

9
docs/source/azure-vm.rst
View File

@ -115,6 +115,15 @@ For Corda nodes running release M11
cd /opt/corda/cordapps
wget http://downloads.corda.net/cordapps/net/corda/yo/0.11.0/yo-0.11.0.jar
For Corda nodes running version 2
.. sourcecode:: shell
cd /opt/corda/plugins
wget http://ci-artifactory.corda.r3cev.com/artifactory/cordapp-showcase/yo-4.jar
Now restart Corda and the Corda webserver using the following commands or restart your Corda VM from the Azure portal:
.. sourcecode:: shell

2
docs/source/changelog.rst
View File

@ -14,6 +14,8 @@ Unreleased
* Shell now kills an ongoing flow when CTRL+C is pressed in the terminal.
* Add check at startup that all persisted Checkpoints are compatible with the current version of the code.
* ``ServiceHub`` and ``CordaRPCOps`` can now safely be used from multiple threads without incurring in database transaction problems.
* Doorman and NetworkMap url's can now be configured individually rather than being assumed to be

73
node-api/src/main/kotlin/net/corda/nodeapi/internal/network/NetworkBootstrapper.kt
View File

@ -18,6 +18,7 @@ import net.corda.core.identity.CordaX500Name
import net.corda.core.identity.Party
import net.corda.core.internal.*
import net.corda.core.internal.concurrent.fork
import net.corda.core.internal.concurrent.transpose
import net.corda.core.node.NetworkParameters
import net.corda.core.node.NodeInfo
import net.corda.core.node.NotaryInfo
@ -41,11 +42,13 @@ import java.nio.file.Path
import java.nio.file.Paths
import java.nio.file.StandardCopyOption.REPLACE_EXISTING
import java.time.Instant
import java.util.*
import java.util.concurrent.Executors
import java.util.concurrent.TimeoutException
import java.util.concurrent.TimeUnit
import kotlin.collections.component1
import kotlin.collections.component2
import kotlin.collections.set
import kotlin.concurrent.schedule
import kotlin.streams.toList
/**
@ -118,11 +121,10 @@ class NetworkBootstrapper {
println("Nodes found in the following sub-directories: ${nodeDirs.map { it.fileName }}")
val configs = nodeDirs.associateBy({ it }, { ConfigFactory.parseFile((it / "node.conf").toFile()) })
generateServiceIdentitiesForNotaryClusters(configs)
val processes = startNodeInfoGeneration(nodeDirs)
initialiseSerialization()
try {
println("Waiting for all nodes to generate their node-info files...")
val nodeInfoFiles = gatherNodeInfoFiles(processes, nodeDirs)
val nodeInfoFiles = generateNodeInfos(nodeDirs)
println("Checking for duplicate nodes")
checkForDuplicateLegalNames(nodeInfoFiles)
println("Distributing all node-info files to all nodes")
@ -139,10 +141,41 @@ class NetworkBootstrapper {
println("Bootstrapping complete!")
} finally {
_contextSerializationEnv.set(null)
processes.forEach { if (it.isAlive) it.destroyForcibly() }
}
}
private fun generateNodeInfos(nodeDirs: List<Path>): List<Path> {
val numParallelProcesses = Runtime.getRuntime().availableProcessors()
val timePerNode = 40.seconds // On the test machine, generating the node info takes 7 seconds for a single node.
val tExpected = maxOf(timePerNode, timePerNode * nodeDirs.size.toLong() / numParallelProcesses.toLong())
val warningTimer = Timer("WarnOnSlowMachines", false).schedule(tExpected.toMillis()) {
println("...still waiting. If this is taking longer than usual, check the node logs.")
}
val executor = Executors.newFixedThreadPool(numParallelProcesses)
return try {
nodeDirs.map { executor.fork { generateNodeInfo(it) } }.transpose().getOrThrow()
} finally {
warningTimer.cancel()
executor.shutdownNow()
}
}
private fun generateNodeInfo(nodeDir: Path): Path {
val logsDir = (nodeDir / LOGS_DIR_NAME).createDirectories()
val process = ProcessBuilder(nodeInfoGenCmd)
.directory(nodeDir.toFile())
.redirectErrorStream(true)
.redirectOutput((logsDir / "node-info-gen.log").toFile())
.apply { environment()["CAPSULE_CACHE_DIR"] = "../.cache" }
.start()
if (!process.waitFor(3, TimeUnit.MINUTES)) {
process.destroyForcibly()
throw IllegalStateException("Error while generating node info file. Please check the logs in $logsDir.")
}
check(process.exitValue() == 0) { "Error while generating node info file. Please check the logs in $logsDir." }
return nodeDir.list { paths -> paths.filter { it.fileName.toString().startsWith(NODE_INFO_FILE_NAME_PREFIX) }.findFirst().get() }
}
private fun generateDirectoriesIfNeeded(directory: Path, cordappJars: List<Path>) {
val confFiles = directory.list { it.filter { it.toString().endsWith("_node.conf") }.toList() }
val webServerConfFiles = directory.list { it.filter { it.toString().endsWith("_web-server.conf") }.toList() }
@ -170,38 +203,6 @@ class NetworkBootstrapper {
return cordaJarPath
}
private fun startNodeInfoGeneration(nodeDirs: List<Path>): List<Process> {
return nodeDirs.map { nodeDir ->
val logsDir = (nodeDir / LOGS_DIR_NAME).createDirectories()
ProcessBuilder(nodeInfoGenCmd)
.directory(nodeDir.toFile())
.redirectErrorStream(true)
.redirectOutput((logsDir / "node-info-gen.log").toFile())
.apply { environment()["CAPSULE_CACHE_DIR"] = "../.cache" }
.start()
}
}
private fun gatherNodeInfoFiles(processes: List<Process>, nodeDirs: List<Path>): List<Path> {
val executor = Executors.newSingleThreadExecutor()
val future = executor.fork {
processes.zip(nodeDirs).map { (process, nodeDir) ->
check(process.waitFor() == 0) {
"Node in ${nodeDir.fileName} exited with ${process.exitValue()} when generating its node-info - see logs in ${nodeDir / LOGS_DIR_NAME}"
}
nodeDir.list { paths -> paths.filter { it.fileName.toString().startsWith(NODE_INFO_FILE_NAME_PREFIX) }.findFirst().get() }
}
}
return try {
future.getOrThrow(timeout = 60.seconds)
} catch (e: TimeoutException) {
println("...still waiting. If this is taking longer than usual, check the node logs.")
future.getOrThrow()
}
}
private fun distributeNodeInfos(nodeDirs: List<Path>, nodeInfoFiles: List<Path>) {
for (nodeDir in nodeDirs) {
val additionalNodeInfosDir = (nodeDir / CordformNode.NODE_INFO_DIRECTORY).createDirectories()

2
node/src/integration-test/kotlin/net/corda/node/services/statemachine/HardRestartTest.kt
View File

@ -20,12 +20,14 @@ import net.corda.testing.internal.IntegrationTestSchemas
import net.corda.testing.internal.toDatabaseSchemaName
import net.corda.testing.node.User
import org.junit.ClassRule
import org.junit.Ignore
import org.junit.Test
import java.util.*
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import kotlin.concurrent.thread
@Ignore
class HardRestartTest : IntegrationTest() {
companion object {
@ClassRule

72
node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
View File

@ -20,16 +20,7 @@ import net.corda.core.concurrent.CordaFuture
import net.corda.core.context.InvocationContext
import net.corda.core.crypto.newSecureRandom
import net.corda.core.crypto.sign
import net.corda.core.flows.ContractUpgradeFlow
import net.corda.core.flows.FinalityFlow
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.FlowLogicRefFactory
import net.corda.core.flows.FlowSession
import net.corda.core.flows.InitiatedBy
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.NotaryChangeFlow
import net.corda.core.flows.NotaryFlow
import net.corda.core.flows.StartableByService
import net.corda.core.flows.*
import net.corda.core.identity.AbstractParty
import net.corda.core.identity.CordaX500Name
import net.corda.core.identity.Party
@ -41,33 +32,17 @@ import net.corda.core.internal.concurrent.map
import net.corda.core.internal.concurrent.openFuture
import net.corda.core.internal.notary.NotaryService
import net.corda.core.internal.uncheckedCast
import net.corda.core.messaging.CordaRPCOps
import net.corda.core.messaging.FlowHandle
import net.corda.core.messaging.FlowHandleImpl
import net.corda.core.messaging.FlowProgressHandle
import net.corda.core.messaging.FlowProgressHandleImpl
import net.corda.core.messaging.RPCOps
import net.corda.core.node.AppServiceHub
import net.corda.core.node.NetworkParameters
import net.corda.core.node.NodeInfo
import net.corda.core.node.ServiceHub
import net.corda.core.node.ServicesForResolution
import net.corda.core.node.services.AttachmentStorage
import net.corda.core.node.services.CordaService
import net.corda.core.node.services.IdentityService
import net.corda.core.node.services.KeyManagementService
import net.corda.core.node.services.TransactionVerifierService
import net.corda.core.messaging.*
import net.corda.core.node.*
import net.corda.core.node.services.*
import net.corda.core.serialization.SerializationWhitelist
import net.corda.core.serialization.SerializeAsToken
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.serialization.serialize
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.core.utilities.days
import net.corda.core.utilities.debug
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.minutes
import net.corda.core.utilities.*
import net.corda.node.CordaClock
import net.corda.node.VersionInfo
import net.corda.node.internal.CheckpointVerifier.verifyCheckpointsCompatible
import net.corda.node.internal.classloading.requireAnnotation
import net.corda.node.internal.cordapp.CordappConfigFileProvider
import net.corda.node.internal.cordapp.CordappLoader
@ -80,26 +55,9 @@ import net.corda.node.internal.security.RPCSecurityManager
import net.corda.node.services.ContractUpgradeHandler
import net.corda.node.services.FinalityHandler
import net.corda.node.services.NotaryChangeHandler
import net.corda.node.services.api.CheckpointStorage
import net.corda.node.services.api.DummyAuditService
import net.corda.node.services.api.FlowStarter
import net.corda.node.services.api.IdentityServiceInternal
import net.corda.node.services.api.MonitoringService
import net.corda.node.services.api.NetworkMapCacheBaseInternal
import net.corda.node.services.api.NetworkMapCacheInternal
import net.corda.node.services.api.NodePropertiesStore
import net.corda.node.services.api.SchedulerService
import net.corda.node.services.api.SchemaService
import net.corda.node.services.api.ServiceHubInternal
import net.corda.node.services.api.StartedNodeServices
import net.corda.node.services.api.VaultServiceInternal
import net.corda.node.services.api.WritableTransactionStorage
import net.corda.node.services.config.BFTSMaRtConfiguration
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.config.NotaryConfig
import net.corda.node.services.config.configureWithDevSSLCertificate
import net.corda.node.services.api.*
import net.corda.node.services.config.*
import net.corda.node.services.config.shell.toShellConfig
import net.corda.node.services.config.shouldInitCrashShell
import net.corda.node.services.events.NodeSchedulerService
import net.corda.node.services.events.ScheduledActivityObserver
import net.corda.node.services.identity.PersistentIdentityService
@ -119,6 +77,8 @@ import net.corda.node.services.persistence.DBTransactionStorage
import net.corda.node.services.persistence.NodeAttachmentService
import net.corda.node.services.persistence.NodePropertiesPersistentStore
import net.corda.node.services.persistence.RunOnceService
import net.corda.node.services.network.*
import net.corda.node.services.persistence.*
import net.corda.node.services.schema.HibernateObserver
import net.corda.node.services.schema.NodeSchemaService
import net.corda.node.services.statemachine.ExternalEvent
@ -136,6 +96,8 @@ import net.corda.node.services.transactions.RaftUniquenessProvider
import net.corda.node.services.transactions.RaftValidatingNotaryService
import net.corda.node.services.transactions.SimpleNotaryService
import net.corda.node.services.transactions.ValidatingNotaryService
import net.corda.node.services.statemachine.*
import net.corda.node.services.transactions.*
import net.corda.node.services.upgrade.ContractUpgradeServiceImpl
import net.corda.node.services.vault.NodeVaultService
import net.corda.node.utilities.AffinityExecutor
@ -757,6 +719,16 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
networkParameters: NetworkParameters): MutableList<Any> {
checkpointStorage = DBCheckpointStorage()
try {
verifyCheckpointsCompatible(checkpointStorage, cordappProvider.cordapps, versionInfo.platformVersion)
} catch (e: CheckpointIncompatibleException) {
if (configuration.devMode) {
Node.printWarning(e.message)
} else {
throw e
}
}
val keyManagementService = makeKeyManagementService(identityService, keyPairs, database)
_services = ServiceHubInternalImpl(
identityService,

71
node/src/main/kotlin/net/corda/node/internal/CheckpointVerifier.kt Normal file
View File

@ -0,0 +1,71 @@
package net.corda.node.internal
import net.corda.core.cordapp.Cordapp
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.FlowLogic
import net.corda.core.serialization.SerializationDefaults
import net.corda.core.serialization.deserialize
import net.corda.node.services.api.CheckpointStorage
import net.corda.node.services.statemachine.SubFlow
import net.corda.node.services.statemachine.SubFlowVersion
object CheckpointVerifier {
/**
* Verifies that all Checkpoints stored in the db can be safely loaded with the currently installed version.
* @throws CheckpointIncompatibleException if any offending checkpoint is found.
*/
fun verifyCheckpointsCompatible(checkpointStorage: CheckpointStorage, currentCordapps: List<Cordapp>, platformVersion: Int) {
checkpointStorage.getAllCheckpoints().forEach { (_, serializedCheckpoint) ->
val checkpoint = try {
serializedCheckpoint.deserialize(context = SerializationDefaults.CHECKPOINT_CONTEXT)
} catch (e: Exception) {
throw CheckpointIncompatibleException.CannotBeDeserialisedException(e)
}
// For each Subflow, compare the checkpointed version to the current version.
checkpoint.subFlowStack.forEach { checkFlowCompatible(it, currentCordapps, platformVersion) }
}
}
// Throws exception when the flow is incompatible
private fun checkFlowCompatible(subFlow: SubFlow, currentCordapps: List<Cordapp>, platformVersion: Int) {
val corDappInfo = subFlow.subFlowVersion
if (corDappInfo.platformVersion != platformVersion) {
throw CheckpointIncompatibleException.SubFlowCoreVersionIncompatibleException(subFlow.flowClass, corDappInfo.platformVersion)
}
if (corDappInfo is SubFlowVersion.CorDappFlow) {
val installedCordapps = currentCordapps.filter { it.name == corDappInfo.corDappName }
when (installedCordapps.size) {
0 -> throw CheckpointIncompatibleException.FlowNotInstalledException(subFlow.flowClass)
1 -> {
val currenCordapp = installedCordapps.first()
if (corDappInfo.corDappHash != currenCordapp.jarHash) {
throw CheckpointIncompatibleException.FlowVersionIncompatibleException(subFlow.flowClass, currenCordapp, corDappInfo.corDappHash)
}
}
else -> throw IllegalStateException("Multiple Cordapps with name ${corDappInfo.corDappName} installed.") // This should not happen
}
}
}
}
/**
* Thrown at startup, if a checkpoint is found that is incompatible with the current code
*/
sealed class CheckpointIncompatibleException(override val message: String) : Exception() {
class CannotBeDeserialisedException(val e: Exception) : CheckpointIncompatibleException(
"Found checkpoint that cannot be deserialised using the current Corda version. Please revert to the previous version of Corda, drain your node (see https://docs.corda.net/upgrading-cordapps.html#flow-drains), and try again. Cause: ${e.message}")
class SubFlowCoreVersionIncompatibleException(val flowClass: Class<out FlowLogic<*>>, oldVersion: Int) : CheckpointIncompatibleException(
"Found checkpoint for flow: ${flowClass} that is incompatible with the current Corda platform. Please revert to the previous version of Corda (version ${oldVersion}), drain your node (see https://docs.corda.net/upgrading-cordapps.html#flow-drains), and try again.")
class FlowVersionIncompatibleException(val flowClass: Class<out FlowLogic<*>>, val cordapp: Cordapp, oldHash: SecureHash) : CheckpointIncompatibleException(
"Found checkpoint for flow: ${flowClass} that is incompatible with the current installed version of ${cordapp.name}. Please reinstall the previous version of the CorDapp (with hash: ${oldHash}), drain your node (see https://docs.corda.net/upgrading-cordapps.html#flow-drains), and try again.")
class FlowNotInstalledException(val flowClass: Class<out FlowLogic<*>>) : CheckpointIncompatibleException(
"Found checkpoint for flow: ${flowClass} that is no longer installed. Please install the missing CorDapp, drain your node (see https://docs.corda.net/upgrading-cordapps.html#flow-drains), and try again.")
}

3
node/src/main/kotlin/net/corda/node/internal/NodeStartup.kt
View File

@ -144,6 +144,9 @@ open class NodeStartup(val args: Array<String>) {
} catch (e: DatabaseMigrationException) {
logger.error(e.message)
return false
} catch (e: CheckpointIncompatibleException) {
logger.error(e.message)
return false
} catch (e: Exception) {
if (e is Errors.NativeIoException && e.message?.contains("Address already in use") == true) {
logger.error("One of the ports required by the Corda node is already in use.")

63
node/src/main/kotlin/net/corda/node/internal/cordapp/CordappLoader.kt
View File

@ -14,6 +14,10 @@ import com.github.benmanes.caffeine.cache.Caffeine
import io.github.lukehutch.fastclasspathscanner.FastClasspathScanner
import io.github.lukehutch.fastclasspathscanner.scanner.ScanResult
import net.corda.core.cordapp.Cordapp
import net.corda.core.crypto.SecureHash
import net.corda.core.crypto.sha256
import net.corda.core.flows.*
import net.corda.core.internal.*
import net.corda.core.flows.ContractUpgradeFlow
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatedBy
@ -70,6 +74,17 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
val cordapps: List<Cordapp> by lazy { loadCordapps() + coreCordapp }
val appClassLoader: ClassLoader = URLClassLoader(cordappJarPaths.stream().map { it.url }.toTypedArray(), javaClass.classLoader)
// Create a map of the CorDapps that provide a Flow. If a flow is not in this map it is a Core flow.
// It also checks that there is only one CorDapp containing that flow class
val flowCordappMap: Map<Class<out FlowLogic<*>>, Cordapp> by lazy {
cordapps.flatMap { corDapp -> corDapp.allFlows.map { flow -> flow to corDapp } }
.groupBy { it.first }
.mapValues {
require(it.value.size == 1) { "There are multiple CorDapp jars on the classpath for flow ${it.value.first().first.name}: ${it.value.map { it.second.name }.joinToString()}." }
it.value.single().second
}
}
init {
if (cordappJarPaths.isEmpty()) {
logger.info("No CorDapp paths provided")
@ -93,7 +108,9 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
serializationCustomSerializers = listOf(),
customSchemas = setOf(),
jarPath = ContractUpgradeFlow.javaClass.protectionDomain.codeSource.location, // Core JAR location
info = CordappImpl.Info("corda-core", versionInfo.vendor, versionInfo.releaseVersion)
info = CordappImpl.Info("corda-core", versionInfo.vendor, versionInfo.releaseVersion),
allFlows = listOf(),
jarHash = SecureHash.allOnesHash
)
companion object {
@ -165,6 +182,10 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
val resource = scanPackage.replace('.', '/')
return this::class.java.classLoader.getResources(resource)
.asSequence()
// This is to only scan classes from test folders.
.filter { url ->
listOf("main", "production/classes").none { url.toString().contains("$it/$resource") } || listOf("net.corda.core", "net.corda.node", "net.corda.finance").none { scanPackage.startsWith(it) }
}
.map { url ->
if (url.protocol == "jar") {
// When running tests from gradle this may be a corda module jar, so restrict to scanPackage:
@ -229,21 +250,26 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
val name = url.toPath().fileName.toString().removeSuffix(".jar")
val info = url.openStream().let(::JarInputStream).use { it.manifest }.toCordappInfo(name)
val scanResult = scanCordapp(it)
CordappImpl(findContractClassNames(scanResult),
findInitiatedFlows(scanResult),
findRPCFlows(scanResult),
findServiceFlows(scanResult),
findSchedulableFlows(scanResult),
findServices(scanResult),
findPlugins(it),
findSerializers(scanResult),
findCustomSchemas(scanResult),
url,
info,
name)
CordappImpl(contractClassNames = findContractClassNames(scanResult),
initiatedFlows = findInitiatedFlows(scanResult),
rpcFlows = findRPCFlows(scanResult),
serviceFlows = findServiceFlows(scanResult),
schedulableFlows = findSchedulableFlows(scanResult),
services = findServices(scanResult),
serializationWhitelists = findPlugins(it),
serializationCustomSerializers = findSerializers(scanResult),
customSchemas = findCustomSchemas(scanResult),
allFlows = findAllFlows(scanResult),
jarPath = it.url,
info = info,
jarHash = getJarHash(it.url),
name = name
)
}
}
private fun getJarHash(url: URL): SecureHash.SHA256 = url.openStream().readFully().sha256()
private fun findServices(scanResult: RestrictedScanResult): List<Class<out SerializeAsToken>> {
return scanResult.getClassesWithAnnotation(SerializeAsToken::class, CordaService::class)
}
@ -279,6 +305,10 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
return scanResult.getClassesWithAnnotation(FlowLogic::class, SchedulableFlow::class)
}
private fun findAllFlows(scanResult: RestrictedScanResult): List<Class<out FlowLogic<*>>> {
return scanResult.getConcreteClassesOfType(FlowLogic::class)
}
private fun findContractClassNames(scanResult: RestrictedScanResult): List<String> {
return coreContractClasses.flatMap { scanResult.getNamesOfClassesImplementing(it) }.distinct()
}
@ -362,5 +392,12 @@ class CordappLoader private constructor(private val cordappJarPaths: List<Restri
.mapNotNull { loadClass(it, type) }
.filterNot { Modifier.isAbstract(it.modifiers) }
}
fun <T : Any> getConcreteClassesOfType(type: KClass<T>): List<Class<out T>> {
return scanResult.getNamesOfSubclassesOf(type.java)
.filter { it.startsWith(qualifiedNamePrefix) }
.mapNotNull { loadClass(it, type) }
.filterNot { Modifier.isAbstract(it.modifiers) }
}
}
}

4
node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderImpl.kt
View File

@ -16,6 +16,7 @@ import net.corda.core.contracts.ContractClassName
import net.corda.core.cordapp.Cordapp
import net.corda.core.cordapp.CordappContext
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.FlowLogic
import net.corda.core.internal.DEPLOYED_CORDAPP_UPLOADER
import net.corda.core.internal.createCordappContext
import net.corda.core.node.services.AttachmentId
@ -32,7 +33,6 @@ open class CordappProviderImpl(private val cordappLoader: CordappLoader,
private val cordappConfigProvider: CordappConfigProvider,
attachmentStorage: AttachmentStorage,
private val whitelistedContractImplementations: Map<String, List<AttachmentId>>) : SingletonSerializeAsToken(), CordappProviderInternal {
companion object {
private val log = loggerFor<CordappProviderImpl>()
}
@ -127,4 +127,6 @@ open class CordappProviderImpl(private val cordappLoader: CordappLoader,
* @return cordapp A cordapp or null if no cordapp has the given class loaded
*/
fun getCordappForClass(className: String): Cordapp? = cordapps.find { it.cordappClasses.contains(className) }
override fun getCordappForFlow(flowLogic: FlowLogic<*>) = cordappLoader.flowCordappMap[flowLogic.javaClass]
}

2
node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderInternal.kt
View File

@ -12,7 +12,9 @@ package net.corda.node.internal.cordapp
import net.corda.core.cordapp.Cordapp
import net.corda.core.cordapp.CordappProvider
import net.corda.core.flows.FlowLogic
interface CordappProviderInternal : CordappProvider {
val cordapps: List<Cordapp>
fun getCordappForFlow(flowLogic: FlowLogic<*>): Cordapp?
}

3
node/src/main/kotlin/net/corda/node/services/api/CheckpointStorage.kt
View File

@ -10,6 +10,7 @@
package net.corda.node.services.api
import net.corda.core.cordapp.Cordapp
import net.corda.core.flows.StateMachineRunId
import net.corda.core.serialization.SerializedBytes
import net.corda.node.services.statemachine.Checkpoint
@ -41,4 +42,4 @@ interface CheckpointStorage {
* underlying database connection is closed, so any processing should happen before it is closed.
*/
fun getAllCheckpoints(): Stream<Pair<StateMachineRunId, SerializedBytes<Checkpoint>>>
}
}

2
node/src/main/kotlin/net/corda/node/services/statemachine/Event.kt
View File

@ -90,7 +90,7 @@ sealed class Event {
*
* @param subFlowClass the [Class] of the subflow, to be used to determine whether it's Initiating or inlined.
*/
data class EnterSubFlow(val subFlowClass: Class<FlowLogic<*>>) : Event()
data class EnterSubFlow(val subFlowClass: Class<FlowLogic<*>>, val subFlowVersion: SubFlowVersion ) : Event()
/**
* Signal the leaving of a subflow.

13
node/src/main/kotlin/net/corda/node/services/statemachine/FlowStateMachineImpl.kt
View File

@ -18,6 +18,7 @@ import co.paralleluniverse.strands.Strand
import co.paralleluniverse.strands.channels.Channel
import net.corda.core.concurrent.CordaFuture
import net.corda.core.context.InvocationContext
import net.corda.core.cordapp.Cordapp
import net.corda.core.flows.*
import net.corda.core.identity.Party
import net.corda.core.internal.*
@ -56,6 +57,12 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
*/
fun currentStateMachine(): FlowStateMachineImpl<*>? = Strand.currentStrand() as? FlowStateMachineImpl<*>
// If no CorDapp found then it is a Core flow.
internal fun createSubFlowVersion(cordapp: Cordapp?, platformVersion: Int): SubFlowVersion {
return cordapp?.let { SubFlowVersion.CorDappFlow(platformVersion, it.name, it.jarHash) }
?: SubFlowVersion.CoreFlow(platformVersion)
}
private val log: Logger = LoggerFactory.getLogger("net.corda.flow")
private val SERIALIZER_BLOCKER = Fiber::class.java.getDeclaredField("SERIALIZER_BLOCKER").apply { isAccessible = true }.get(null)
@ -109,7 +116,7 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
if (value) field = value else throw IllegalArgumentException("Can only set to true")
}
/**
/**
* Processes an event by creating the associated transition and executing it using the given executor.
* Try to avoid using this directly, instead use [processEventsUntilFlowIsResumed] or [processEventImmediately]
* instead.
@ -247,7 +254,9 @@ class FlowStateMachineImpl<R>(override val id: StateMachineRunId,
@Suspendable
override fun <R> subFlow(subFlow: FlowLogic<R>): R {
processEventImmediately(
Event.EnterSubFlow(subFlow.javaClass),
Event.EnterSubFlow(subFlow.javaClass,
createSubFlowVersion(
serviceHub.cordappProvider.getCordappForFlow(subFlow), serviceHub.myInfo.platformVersion)),
isDbTransactionOpenOnEntry = true,
isDbTransactionOpenOnExit = true
)

3
node/src/main/kotlin/net/corda/node/services/statemachine/MultiThreadedStateMachineManager.kt
View File

@ -532,7 +532,8 @@ class MultiThreadedStateMachineManager(
flowLogic.stateMachine = flowStateMachineImpl
val frozenFlowLogic = (flowLogic as FlowLogic<*>).serialize(context = checkpointSerializationContext!!)
val initialCheckpoint = Checkpoint.create(invocationContext, flowStart, flowLogic.javaClass, frozenFlowLogic, ourIdentity, deduplicationSeed).getOrThrow()
val flowCorDappVersion= FlowStateMachineImpl.createSubFlowVersion(serviceHub.cordappProvider.getCordappForFlow(flowLogic), serviceHub.myInfo.platformVersion)
val initialCheckpoint = Checkpoint.create(invocationContext, flowStart, flowLogic.javaClass, frozenFlowLogic, ourIdentity, deduplicationSeed, flowCorDappVersion).getOrThrow()
val startedFuture = openFuture<Unit>()
val initialState = StateMachineState(
checkpoint = initialCheckpoint,

7
node/src/main/kotlin/net/corda/node/services/statemachine/SingleThreadedStateMachineManager.kt
View File

@ -18,6 +18,7 @@ import co.paralleluniverse.strands.channels.Channels
import com.codahale.metrics.Gauge
import net.corda.core.concurrent.CordaFuture
import net.corda.core.context.InvocationContext
import net.corda.core.context.InvocationOrigin
import net.corda.core.flows.FlowException
import net.corda.core.flows.FlowInfo
import net.corda.core.flows.FlowLogic
@ -46,6 +47,8 @@ import net.corda.node.services.api.ServiceHubInternal
import net.corda.node.services.config.shouldCheckCheckpoints
import net.corda.node.services.messaging.DeduplicationHandler
import net.corda.node.services.messaging.ReceivedMessage
import net.corda.node.services.statemachine.FlowStateMachineImpl.Companion.createSubFlowVersion
import net.corda.node.services.statemachine.interceptors.*
import net.corda.node.services.statemachine.interceptors.DumpHistoryOnErrorInterceptor
import net.corda.node.services.statemachine.interceptors.FiberDeserializationChecker
import net.corda.node.services.statemachine.interceptors.FiberDeserializationCheckingInterceptor
@ -539,7 +542,9 @@ class SingleThreadedStateMachineManager(
flowLogic.stateMachine = flowStateMachineImpl
val frozenFlowLogic = (flowLogic as FlowLogic<*>).serialize(context = checkpointSerializationContext!!)
val initialCheckpoint = Checkpoint.create(invocationContext, flowStart, flowLogic.javaClass, frozenFlowLogic, ourIdentity, deduplicationSeed).getOrThrow()
val flowCorDappVersion= createSubFlowVersion(serviceHub.cordappProvider.getCordappForFlow(flowLogic), serviceHub.myInfo.platformVersion)
val initialCheckpoint = Checkpoint.create(invocationContext, flowStart, flowLogic.javaClass, frozenFlowLogic, ourIdentity, deduplicationSeed, flowCorDappVersion).getOrThrow()
val startedFuture = openFuture<Unit>()
val initialState = StateMachineState(
checkpoint = initialCheckpoint,

15
node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineState.kt
View File

@ -11,6 +11,7 @@
package net.corda.node.services.statemachine
import net.corda.core.context.InvocationContext
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.FlowInfo
import net.corda.core.flows.FlowLogic
import net.corda.core.identity.Party
@ -80,9 +81,10 @@ data class Checkpoint(
flowLogicClass: Class<FlowLogic<*>>,
frozenFlowLogic: SerializedBytes<FlowLogic<*>>,
ourIdentity: Party,
deduplicationSeed: String
deduplicationSeed: String,
subFlowVersion: SubFlowVersion
): Try<Checkpoint> {
return SubFlow.create(flowLogicClass).map { topLevelSubFlow ->
return SubFlow.create(flowLogicClass, subFlowVersion).map { topLevelSubFlow ->
Checkpoint(
invocationContext = invocationContext,
ourIdentity = ourIdentity,
@ -241,3 +243,12 @@ sealed class ErrorState {
}
}
}
/**
* Stored per [SubFlow]. Contains metadata around the version of the code at the Checkpointing moment.
*/
sealed class SubFlowVersion {
abstract val platformVersion: Int
data class CoreFlow(override val platformVersion: Int) : SubFlowVersion()
data class CorDappFlow(override val platformVersion: Int, val corDappName: String, val corDappHash: SecureHash) : SubFlowVersion()
}

18
node/src/main/kotlin/net/corda/node/services/statemachine/SubFlow.kt
View File

@ -10,9 +10,7 @@
package net.corda.node.services.statemachine
import net.corda.core.flows.FlowInfo
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.InitiatingFlow
import net.corda.core.flows.*
import net.corda.core.utilities.Try
/**
@ -25,10 +23,13 @@ import net.corda.core.utilities.Try
sealed class SubFlow {
abstract val flowClass: Class<out FlowLogic<*>>
// Version of the code.
abstract val subFlowVersion: SubFlowVersion
/**
* An inlined subflow.
*/
data class Inlined(override val flowClass: Class<FlowLogic<*>>) : SubFlow()
data class Inlined(override val flowClass: Class<FlowLogic<*>>, override val subFlowVersion: SubFlowVersion) : SubFlow()
/**
* An initiating subflow.
@ -40,21 +41,22 @@ sealed class SubFlow {
data class Initiating(
override val flowClass: Class<FlowLogic<*>>,
val classToInitiateWith: Class<in FlowLogic<*>>,
val flowInfo: FlowInfo
val flowInfo: FlowInfo,
override val subFlowVersion: SubFlowVersion
) : SubFlow()
companion object {
fun create(flowClass: Class<FlowLogic<*>>): Try<SubFlow> {
fun create(flowClass: Class<FlowLogic<*>>, subFlowVersion: SubFlowVersion): Try<SubFlow> {
// Are we an InitiatingFlow?
val initiatingAnnotations = getInitiatingFlowAnnotations(flowClass)
return when (initiatingAnnotations.size) {
0 -> {
Try.Success(Inlined(flowClass))
Try.Success(Inlined(flowClass, subFlowVersion))
}
1 -> {
val initiatingAnnotation = initiatingAnnotations[0]
val flowContext = FlowInfo(initiatingAnnotation.second.version, flowClass.appName)
Try.Success(Initiating(flowClass, initiatingAnnotation.first, flowContext))
Try.Success(Initiating(flowClass, initiatingAnnotation.first, flowContext, subFlowVersion))
}
else -> {
Try.Failure(IllegalArgumentException("${InitiatingFlow::class.java.name} can only be annotated " +

2
node/src/main/kotlin/net/corda/node/services/statemachine/transitions/TopLevelTransition.kt
View File

@ -102,7 +102,7 @@ class TopLevelTransition(
private fun enterSubFlowTransition(event: Event.EnterSubFlow): TransitionResult {
return builder {
val subFlow = SubFlow.create(event.subFlowClass)
val subFlow = SubFlow.create(event.subFlowClass, event.subFlowVersion)
when (subFlow) {
is Try.Success -> {
currentState = currentState.copy(

32
node/src/test/kotlin/net/corda/node/services/persistence/DBCheckpointStorageTests.kt
View File

@ -16,9 +16,12 @@ import net.corda.core.flows.StateMachineRunId
import net.corda.core.serialization.SerializationDefaults
import net.corda.core.serialization.SerializedBytes
import net.corda.core.serialization.serialize
import net.corda.node.internal.CheckpointIncompatibleException
import net.corda.node.internal.CheckpointVerifier
import net.corda.node.internal.configureDatabase
import net.corda.node.services.api.CheckpointStorage
import net.corda.node.services.statemachine.Checkpoint
import net.corda.node.services.statemachine.SubFlowVersion
import net.corda.node.services.statemachine.FlowStart
import net.corda.node.services.transactions.PersistentUniquenessProvider
import net.corda.nodeapi.internal.persistence.CordaPersistence
@ -28,6 +31,7 @@ import net.corda.testing.core.SerializationEnvironmentRule
import net.corda.testing.core.TestIdentity
import net.corda.testing.internal.LogHelper
import net.corda.testing.node.MockServices.Companion.makeTestDataSourceProperties
import org.assertj.core.api.Assertions
import org.assertj.core.api.Assertions.assertThat
import org.junit.After
import org.junit.Before
@ -44,6 +48,7 @@ class DBCheckpointStorageTests {
private companion object {
val ALICE = TestIdentity(ALICE_NAME, 70).party
}
@Rule
@JvmField
val testSerialization = SerializationEnvironmentRule()
@ -158,19 +163,42 @@ class DBCheckpointStorageTests {
}
}
@Test
fun `verify checkpoints compatible`() {
database.transaction {
val (id, checkpoint) = newCheckpoint(1)
checkpointStorage.addCheckpoint(id, checkpoint)
}
database.transaction {
CheckpointVerifier.verifyCheckpointsCompatible(checkpointStorage, emptyList(), 1)
}
database.transaction {
val (id1, checkpoint1) = newCheckpoint(2)
checkpointStorage.addCheckpoint(id1, checkpoint1)
}
Assertions.assertThatThrownBy {
database.transaction {
CheckpointVerifier.verifyCheckpointsCompatible(checkpointStorage, emptyList(), 1)
}
}.isInstanceOf(CheckpointIncompatibleException::class.java)
}
private fun newCheckpointStorage() {
database.transaction {
checkpointStorage = DBCheckpointStorage()
}
}
private fun newCheckpoint(): Pair<StateMachineRunId, SerializedBytes<Checkpoint>> {
private fun newCheckpoint(version: Int = 1): Pair<StateMachineRunId, SerializedBytes<Checkpoint>> {
val id = StateMachineRunId.createRandom()
val logic: FlowLogic<*> = object : FlowLogic<Unit>() {
override fun call() {}
}
val frozenLogic = logic.serialize(context = SerializationDefaults.CHECKPOINT_CONTEXT)
val checkpoint = Checkpoint.create(InvocationContext.shell(), FlowStart.Explicit, logic.javaClass, frozenLogic, ALICE, "").getOrThrow()
val checkpoint = Checkpoint.create(InvocationContext.shell(), FlowStart.Explicit, logic.javaClass, frozenLogic, ALICE, "", SubFlowVersion.CoreFlow(version)).getOrThrow()
return id to checkpoint.serialize(context = SerializationDefaults.CHECKPOINT_CONTEXT)
}

5
testing/test-utils/src/main/kotlin/net/corda/testing/internal/MockCordappProvider.kt
View File

@ -12,6 +12,7 @@ package net.corda.testing.internal
import net.corda.core.contracts.ContractClassName
import net.corda.core.cordapp.Cordapp
import net.corda.core.crypto.SecureHash
import net.corda.core.internal.TEST_UPLOADER
import net.corda.core.internal.cordapp.CordappImpl
import net.corda.core.node.services.AttachmentId
@ -43,7 +44,9 @@ class MockCordappProvider(
serializationWhitelists = emptyList(),
serializationCustomSerializers = emptyList(),
customSchemas = emptySet(),
jarPath = Paths.get("").toUri().toURL())
jarPath = Paths.get("").toUri().toURL(),
allFlows = emptyList(),
jarHash = SecureHash.allOnesHash)
if (cordappRegistry.none { it.first.contractClassNames.contains(contractClassName) }) {
cordappRegistry.add(Pair(cordapp, findOrImportAttachment(listOf(contractClassName), contractClassName.toByteArray(), attachments)))
}