Merge remote-tracking branch 'open/master' into tudor-merge-os-01_06

# Conflicts:
#	core/src/main/kotlin/net/corda/core/cordapp/Cordapp.kt
#	core/src/main/kotlin/net/corda/core/internal/cordapp/CordappImpl.kt
#	node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
#	node/src/main/kotlin/net/corda/node/internal/NodeStartup.kt
#	node/src/main/kotlin/net/corda/node/internal/cordapp/CordappLoader.kt

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

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
 

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<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 info: Info,
+    val jarHash: SecureHash.SHA256
 
     /**
      * CorDapp's information, including vendor and version.

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<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) : Cordapp {
+    override val name: String = jarPath.toPath().fileName.toString().removeSuffix(".jar")
 
     /**
      * An exhaustive list of all classes relevant to the node within this CorDapp

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

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

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<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()

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.*
-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.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.*
-import net.corda.core.messaging.FlowHandle
+import net.corda.core.node.*
-import net.corda.core.messaging.FlowHandleImpl
+import net.corda.core.node.services.*
-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.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.*
-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.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.*
-import net.corda.node.services.api.DummyAuditService
+import net.corda.node.services.config.*
-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.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<Any> {
         checkpointStorage = DBCheckpointStorage()
 
+        verifyCheckpointsCompatible(checkpointStorage, cordappProvider.cordapps, versionInfo.platformVersion)
+
         val keyManagementService = makeKeyManagementService(identityService, keyPairs, database)
         _services = ServiceHubInternalImpl(
                 identityService,

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<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.")
+}
+

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

@@ -144,6 +144,13 @@ open class NodeStartup(val args: Array<String>) {
         } 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.")

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<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")
@@ -165,6 +180,8 @@ 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").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<Restri
                     url,
                     info,
                     name)
+                    findAllFlows(scanResult),
+                    it.url,
+                    getJarHash(it.url)
+            )
         }
     }
 
+    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 +302,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 +389,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) }
+        }
     }
 }

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<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]
 }

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<Cordapp>
+    fun getCordappForFlow(flowLogic: FlowLogic<*>): Cordapp?
 }

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<Pair<StateMachineRunId, SerializedBytes<Checkpoint>>>
-}
+}

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<FlowLogic<*>>) : Event()
+    data class EnterSubFlow(val subFlowClass: Class<FlowLogic<*>>, val subFlowVersion: SubFlowVersion ) : Event()
 
     /**
      * Signal the leaving of a subflow.

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<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
         )

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<Unit>()
         val initialState = StateMachineState(
                 checkpoint = initialCheckpoint,

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<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()
+}

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.*
-import net.corda.core.flows.FlowLogic
-import net.corda.core.flows.InitiatingFlow
 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 " +

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(

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<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)
     }
 

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)))
         }