diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 9722f81219..399053ffe6 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -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 diff --git a/README.md b/README.md index 463af1b9a2..1601315cc8 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/core/src/main/kotlin/net/corda/core/cordapp/Cordapp.kt b/core/src/main/kotlin/net/corda/core/cordapp/Cordapp.kt index d2f2a4ba51..2c7b64022b 100644 --- a/core/src/main/kotlin/net/corda/core/cordapp/Cordapp.kt +++ b/core/src/main/kotlin/net/corda/core/cordapp/Cordapp.kt @@ -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 val serializationCustomSerializers: List> val customSchemas: Set + val allFlows: List>> val jarPath: URL val cordappClasses: List - val info: Info + val info: Info, + val jarHash: SecureHash.SHA256 /** * CorDapp's information, including vendor and version. diff --git a/core/src/main/kotlin/net/corda/core/internal/cordapp/CordappImpl.kt b/core/src/main/kotlin/net/corda/core/internal/cordapp/CordappImpl.kt index f121e155c0..05e6732334 100644 --- a/core/src/main/kotlin/net/corda/core/internal/cordapp/CordappImpl.kt +++ b/core/src/main/kotlin/net/corda/core/internal/cordapp/CordappImpl.kt @@ -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, override val serializationCustomSerializers: List>, override val customSchemas: Set, + override val allFlows: List>>, 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) : Cordapp { + override val name: String = jarPath.toPath().fileName.toString().removeSuffix(".jar") /** * An exhaustive list of all classes relevant to the node within this CorDapp diff --git a/docs/source/azure-vm.rst b/docs/source/azure-vm.rst index 8661422006..b736057275 100644 --- a/docs/source/azure-vm.rst +++ b/docs/source/azure-vm.rst @@ -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 diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst index 22e041e133..2ae98a62b5 100644 --- a/docs/source/changelog.rst +++ b/docs/source/changelog.rst @@ -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 diff --git a/node-api/src/main/kotlin/net/corda/nodeapi/internal/network/NetworkBootstrapper.kt b/node-api/src/main/kotlin/net/corda/nodeapi/internal/network/NetworkBootstrapper.kt index f3094ed808..f15033fb34 100644 --- a/node-api/src/main/kotlin/net/corda/nodeapi/internal/network/NetworkBootstrapper.kt +++ b/node-api/src/main/kotlin/net/corda/nodeapi/internal/network/NetworkBootstrapper.kt @@ -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): List { + 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) { 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): List { - 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, nodeDirs: List): List { - 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, nodeInfoFiles: List) { for (nodeDir in nodeDirs) { val additionalNodeInfosDir = (nodeDir / CordformNode.NODE_INFO_DIRECTORY).createDirectories() diff --git a/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt b/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt index c8e6d87bb6..af944b3fd1 100644 --- a/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt +++ b/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt @@ -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,8 @@ abstract class AbstractNode(val configuration: NodeConfiguration, networkParameters: NetworkParameters): MutableList { checkpointStorage = DBCheckpointStorage() + verifyCheckpointsCompatible(checkpointStorage, cordappProvider.cordapps, versionInfo.platformVersion) + val keyManagementService = makeKeyManagementService(identityService, keyPairs, database) _services = ServiceHubInternalImpl( identityService, diff --git a/node/src/main/kotlin/net/corda/node/internal/CheckpointVerifier.kt b/node/src/main/kotlin/net/corda/node/internal/CheckpointVerifier.kt new file mode 100644 index 0000000000..96b7704df3 --- /dev/null +++ b/node/src/main/kotlin/net/corda/node/internal/CheckpointVerifier.kt @@ -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, 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, 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>, 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>, 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>) : 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.") +} + diff --git a/node/src/main/kotlin/net/corda/node/internal/NodeStartup.kt b/node/src/main/kotlin/net/corda/node/internal/NodeStartup.kt index 3a0f1018f7..a9770d5f81 100644 --- a/node/src/main/kotlin/net/corda/node/internal/NodeStartup.kt +++ b/node/src/main/kotlin/net/corda/node/internal/NodeStartup.kt @@ -144,6 +144,13 @@ open class NodeStartup(val args: Array) { } catch (e: DatabaseMigrationException) { logger.error(e.message) return false + } catch (e: CheckpointIncompatibleException) { + if (conf.devMode) { + Node.printWarning(e.message) + } else { + 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.") diff --git a/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappLoader.kt b/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappLoader.kt index 27f87c1b53..112d74c76f 100644 --- a/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappLoader.kt +++ b/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappLoader.kt @@ -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 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>, 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") @@ -165,6 +180,8 @@ class CordappLoader private constructor(private val cordappJarPaths: List listOf("main", "production").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: @@ -241,9 +258,15 @@ class CordappLoader private constructor(private val cordappJarPaths: List> { return scanResult.getClassesWithAnnotation(SerializeAsToken::class, CordaService::class) } @@ -279,6 +302,10 @@ class CordappLoader private constructor(private val cordappJarPaths: List>> { + return scanResult.getConcreteClassesOfType(FlowLogic::class) + } + private fun findContractClassNames(scanResult: RestrictedScanResult): List { return coreContractClasses.flatMap { scanResult.getNamesOfClassesImplementing(it) }.distinct() } @@ -362,5 +389,12 @@ class CordappLoader private constructor(private val cordappJarPaths: List getConcreteClassesOfType(type: KClass): List> { + return scanResult.getNamesOfSubclassesOf(type.java) + .filter { it.startsWith(qualifiedNamePrefix) } + .mapNotNull { loadClass(it, type) } + .filterNot { Modifier.isAbstract(it.modifiers) } + } } } diff --git a/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderImpl.kt b/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderImpl.kt index b245840bba..523462017b 100644 --- a/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderImpl.kt +++ b/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderImpl.kt @@ -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>) : SingletonSerializeAsToken(), CordappProviderInternal { - companion object { private val log = loggerFor() } @@ -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] } diff --git a/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderInternal.kt b/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderInternal.kt index 4012bf739c..37f2afa7c4 100644 --- a/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderInternal.kt +++ b/node/src/main/kotlin/net/corda/node/internal/cordapp/CordappProviderInternal.kt @@ -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 + fun getCordappForFlow(flowLogic: FlowLogic<*>): Cordapp? } diff --git a/node/src/main/kotlin/net/corda/node/services/api/CheckpointStorage.kt b/node/src/main/kotlin/net/corda/node/services/api/CheckpointStorage.kt index bddbbc7e54..7ec9671954 100644 --- a/node/src/main/kotlin/net/corda/node/services/api/CheckpointStorage.kt +++ b/node/src/main/kotlin/net/corda/node/services/api/CheckpointStorage.kt @@ -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>> -} +} \ No newline at end of file diff --git a/node/src/main/kotlin/net/corda/node/services/statemachine/Event.kt b/node/src/main/kotlin/net/corda/node/services/statemachine/Event.kt index e35fdbbb50..319d656f2c 100644 --- a/node/src/main/kotlin/net/corda/node/services/statemachine/Event.kt +++ b/node/src/main/kotlin/net/corda/node/services/statemachine/Event.kt @@ -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>) : Event() + data class EnterSubFlow(val subFlowClass: Class>, val subFlowVersion: SubFlowVersion ) : Event() /** * Signal the leaving of a subflow. diff --git a/node/src/main/kotlin/net/corda/node/services/statemachine/FlowStateMachineImpl.kt b/node/src/main/kotlin/net/corda/node/services/statemachine/FlowStateMachineImpl.kt index abb917b6c8..e9da60308e 100644 --- a/node/src/main/kotlin/net/corda/node/services/statemachine/FlowStateMachineImpl.kt +++ b/node/src/main/kotlin/net/corda/node/services/statemachine/FlowStateMachineImpl.kt @@ -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(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(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(override val id: StateMachineRunId, @Suspendable override fun subFlow(subFlow: FlowLogic): R { processEventImmediately( - Event.EnterSubFlow(subFlow.javaClass), + Event.EnterSubFlow(subFlow.javaClass, + createSubFlowVersion( + serviceHub.cordappProvider.getCordappForFlow(subFlow), serviceHub.myInfo.platformVersion)), isDbTransactionOpenOnEntry = true, isDbTransactionOpenOnExit = true ) diff --git a/node/src/main/kotlin/net/corda/node/services/statemachine/SingleThreadedStateMachineManager.kt b/node/src/main/kotlin/net/corda/node/services/statemachine/SingleThreadedStateMachineManager.kt index eb7b1ad4e9..73c3490a60 100644 --- a/node/src/main/kotlin/net/corda/node/services/statemachine/SingleThreadedStateMachineManager.kt +++ b/node/src/main/kotlin/net/corda/node/services/statemachine/SingleThreadedStateMachineManager.kt @@ -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() val initialState = StateMachineState( checkpoint = initialCheckpoint, diff --git a/node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineState.kt b/node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineState.kt index da2bff6559..5f1339a1f7 100644 --- a/node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineState.kt +++ b/node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineState.kt @@ -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>, frozenFlowLogic: SerializedBytes>, ourIdentity: Party, - deduplicationSeed: String + deduplicationSeed: String, + subFlowVersion: SubFlowVersion ): Try { - 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() +} \ No newline at end of file diff --git a/node/src/main/kotlin/net/corda/node/services/statemachine/SubFlow.kt b/node/src/main/kotlin/net/corda/node/services/statemachine/SubFlow.kt index 177d52e584..0d8e0118a3 100644 --- a/node/src/main/kotlin/net/corda/node/services/statemachine/SubFlow.kt +++ b/node/src/main/kotlin/net/corda/node/services/statemachine/SubFlow.kt @@ -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> + // Version of the code. + abstract val subFlowVersion: SubFlowVersion + /** * An inlined subflow. */ - data class Inlined(override val flowClass: Class>) : SubFlow() + data class Inlined(override val flowClass: Class>, override val subFlowVersion: SubFlowVersion) : SubFlow() /** * An initiating subflow. @@ -40,21 +41,22 @@ sealed class SubFlow { data class Initiating( override val flowClass: Class>, val classToInitiateWith: Class>, - val flowInfo: FlowInfo + val flowInfo: FlowInfo, + override val subFlowVersion: SubFlowVersion ) : SubFlow() companion object { - fun create(flowClass: Class>): Try { + fun create(flowClass: Class>, subFlowVersion: SubFlowVersion): Try { // 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 " + diff --git a/node/src/main/kotlin/net/corda/node/services/statemachine/transitions/TopLevelTransition.kt b/node/src/main/kotlin/net/corda/node/services/statemachine/transitions/TopLevelTransition.kt index 3da28f65c7..3a977c0ed5 100644 --- a/node/src/main/kotlin/net/corda/node/services/statemachine/transitions/TopLevelTransition.kt +++ b/node/src/main/kotlin/net/corda/node/services/statemachine/transitions/TopLevelTransition.kt @@ -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( diff --git a/node/src/test/kotlin/net/corda/node/services/persistence/DBCheckpointStorageTests.kt b/node/src/test/kotlin/net/corda/node/services/persistence/DBCheckpointStorageTests.kt index b3df9c6c06..f9bc6b6f5e 100644 --- a/node/src/test/kotlin/net/corda/node/services/persistence/DBCheckpointStorageTests.kt +++ b/node/src/test/kotlin/net/corda/node/services/persistence/DBCheckpointStorageTests.kt @@ -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> { + private fun newCheckpoint(version: Int = 1): Pair> { val id = StateMachineRunId.createRandom() val logic: FlowLogic<*> = object : FlowLogic() { 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) } diff --git a/testing/test-utils/src/main/kotlin/net/corda/testing/internal/MockCordappProvider.kt b/testing/test-utils/src/main/kotlin/net/corda/testing/internal/MockCordappProvider.kt index 12640cd005..412b7d2ca4 100644 --- a/testing/test-utils/src/main/kotlin/net/corda/testing/internal/MockCordappProvider.kt +++ b/testing/test-utils/src/main/kotlin/net/corda/testing/internal/MockCordappProvider.kt @@ -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))) }