diff --git a/config/dev/generalnodea.conf b/config/dev/generalnodea.conf index 12a3200105..7547300ca6 100644 --- a/config/dev/generalnodea.conf +++ b/config/dev/generalnodea.conf @@ -4,4 +4,3 @@ trustStorePassword : "trustpass" p2pAddress : "localhost:10002" rpcAddress : "localhost:10003" webAddress : "localhost:10004" -useHTTPS : false diff --git a/config/dev/generalnodeb.conf b/config/dev/generalnodeb.conf index 55ebb1e690..f922e91536 100644 --- a/config/dev/generalnodeb.conf +++ b/config/dev/generalnodeb.conf @@ -4,4 +4,3 @@ trustStorePassword : "trustpass" p2pAddress : "localhost:10005" rpcAddress : "localhost:10006" webAddress : "localhost:10007" -useHTTPS : false diff --git a/config/dev/nameservernode.conf b/config/dev/nameservernode.conf index 0519e56872..3f700628dc 100644 --- a/config/dev/nameservernode.conf +++ b/config/dev/nameservernode.conf @@ -6,4 +6,3 @@ webAddress : "localhost:10001" notary : { validating : true } -useHTTPS : false diff --git a/docs/source/corda-configuration-file.rst b/docs/source/corda-configuration-file.rst index 25490f055d..fe034e92c6 100644 --- a/docs/source/corda-configuration-file.rst +++ b/docs/source/corda-configuration-file.rst @@ -43,7 +43,6 @@ Simple Notary configuration file. notary : { validating : false } - useHTTPS : false devMode : true compatibilityZoneURL : "https://cz.corda.net" @@ -136,10 +135,6 @@ path to the node's base directory. Only one of ``raft``, ``bftSMaRt`` or ``custom`` configuration values may be specified. -:useHTTPS: If false the node's web server will be plain HTTP. If true the node will use the same certificate and private - key from the ``/certificates/sslkeystore.jks`` file as the ArtemisMQ port for HTTPS. If HTTPS is enabled - then unencrypted HTTP traffic to the node's **webAddress** port is not supported. - :rpcUsers: A list of users who are authorised to access the RPC system. Each user in the list is a config object with the following fields: diff --git a/docs/source/deploying-a-node.rst b/docs/source/deploying-a-node.rst index 4f94c02c95..6b17b9dd5d 100644 --- a/docs/source/deploying-a-node.rst +++ b/docs/source/deploying-a-node.rst @@ -46,7 +46,6 @@ handling, and ensures the Corda service is run at boot. myLegalName : "O=Bank of Breakfast Tea, L=London, C=GB" keyStorePassword : "cordacadevpass" trustStorePassword : "trustpass" - useHTTPS : false devMode : false rpcUsers=[ { @@ -217,7 +216,6 @@ at boot, and means the Corda service stays running with no users connected to th keyStorePassword : "cordacadevpass" trustStorePassword : "trustpass" extraAdvertisedServiceIds: [ "" ] - useHTTPS : false devMode : false rpcUsers=[ { diff --git a/docs/source/example-code/src/main/resources/example-network-map-node.conf b/docs/source/example-code/src/main/resources/example-network-map-node.conf index e75807bbc7..980f5f7031 100644 --- a/docs/source/example-code/src/main/resources/example-network-map-node.conf +++ b/docs/source/example-code/src/main/resources/example-network-map-node.conf @@ -4,4 +4,3 @@ trustStorePassword : "trustpass" p2pAddress : "my-network-map:10000" webAddress : "localhost:10001" sshdAddress : "localhost:10002" -useHTTPS : false diff --git a/docs/source/example-code/src/main/resources/example-node.conf b/docs/source/example-code/src/main/resources/example-node.conf index c01a10fd78..3715ff22e8 100644 --- a/docs/source/example-code/src/main/resources/example-node.conf +++ b/docs/source/example-code/src/main/resources/example-node.conf @@ -10,7 +10,6 @@ dataSourceProperties : { p2pAddress : "my-corda-node:10002" rpcAddress : "my-corda-node:10003" webAddress : "localhost:10004" -useHTTPS : false rpcUsers : [ { username=user1, password=letmein, permissions=[ StartFlow.net.corda.protocols.CashProtocol ] } ] diff --git a/docs/source/glossary.rst b/docs/source/glossary.rst index 4432ddb2a2..1fa01d7a59 100644 --- a/docs/source/glossary.rst +++ b/docs/source/glossary.rst @@ -1,6 +1,8 @@ Glossary ======== +AMQP + The serialisation mechanism used within Corda for everything except flow checkpoints and RPC. Artemis The message queuing middleware used within Corda Attachment @@ -30,7 +32,7 @@ Gradle Kotlin The language used to code Corda. Fully compatible with any JVM language, including (obviously) Java. Kryo - The serialisation mechanism used within Corda - which is subject to change in a future release. + The serialisation mechanism used within Corda for flow checkpoints and RPC. Input In Corda terms, an input state is one that is used and consumed within a transaction. Once consumed, it cannot be re-used. JVM diff --git a/docs/source/serialization.rst b/docs/source/serialization.rst index 4e07a37048..96c74e6b2f 100644 --- a/docs/source/serialization.rst +++ b/docs/source/serialization.rst @@ -37,7 +37,7 @@ It's reproduced here as an example of both ways you can do this for a couple of :end-before: END 7 .. note:: Several of the core interfaces at the heart of Corda are already annotated and so any classes that implement - them will automatically be whitelisted. This includes `Contract`, `ContractState` and `CommandData`. + them will automatically be whitelisted. This includes ``Contract``, ``ContractState`` and ``CommandData``. .. warning:: Java 8 Lambda expressions are not serializable except in flow checkpoints, and then not by default. The syntax to declare a serializable Lambda expression that will work with Corda is ``Runnable r = (Runnable & Serializable) () -> System.out.println("Hello World");``, or @@ -52,62 +52,62 @@ AMQP ==== Originally Corda used a ``Kryo``-based serialization scheme throughout for all serialization contexts. However, it was realised there -was a compelling use case for the definition and development of a custom format based upon AMQP 1.0. The primary drivers for this were +was a compelling use case for the definition and development of a custom format based upon AMQP 1.0. The primary drivers for this were: - #. A desire to have a schema describing what has been serialized along-side the actual data: + #. A desire to have a schema describing what has been serialized alongside the actual data: - #. To assist with versioning, both in terms of being able to interpret long ago archived data (e.g. trades from - a decade ago, long after the code has changed) and between differing code versions. - #. To make it easier to write user interfaces that can navigate the serialized form of data. - #. To support cross platform (non-JVM) interaction, where the format of a class file is not so easily interpreted. + #. To assist with versioning, both in terms of being able to interpret data archived long ago (e.g. trades from + a decade ago, long after the code has changed) and between differing code versions + #. To make it easier to write user interfaces that can navigate the serialized form of data + #. To support cross platform (non-JVM) interaction, where the format of a class file is not so easily interpreted #. A desire to use a documented and static wire format that is platform independent, and is not subject to change with - 3rd party library upgrades etc. + 3rd party library upgrades, etc. #. A desire to support open-ended polymorphism, where the number of subclasses of a superclass can expand over time - and do not need to be defined in the schema *upfront*, which is key to many Corda concepts, such as contract states. - #. Increased security from deserialized objects being constructed through supported constructors rather than having - data poked directly into their fields without an opportunity to validate consistency or intercept attempts to manipulate - supposed invariants. + and the subclasses do not need to be defined in the schema *upfront*. This is key to many Corda concepts, such as states. + #. Increased security by constructing deserialized objects through supported constructors, rather than having + data inserted directly into their fields without an opportunity to validate consistency or intercept attempts to manipulate + supposed invariants Delivering this is an ongoing effort by the Corda development team. At present, the ``Kryo``-based format is still used by the RPC framework on -both the client and server side. However, it is planned that this will move to the AMQP framework when ready. +both the client and server side. However, it is planned that the RPC framework will move to the AMQP framework when ready. The AMQP framework is currently used for: - #. The peer to peer context, representing inter-node communication. - #. The persistence layer, representing contract states persisted into the vault. + #. The peer-to-peer context, representing inter-node communication + #. The persistence layer, representing contract states persisted into the vault -Finally, for the checkpointing of flows Corda will continue to use the existing ``Kryo`` scheme. +Finally, for the checkpointing of flows, Corda will continue to use the existing ``Kryo`` scheme. This separation of serialization schemes into different contexts allows us to use the most suitable framework for that context rather than -attempting to force a one size fits all approach. Where ``Kryo`` is more suited to the serialization of a programs stack frames, being more flexible -than our AMQP framework in what it can construct and serialize, that flexibility makes it exceptionally difficult to make secure. Conversely -our AMQP framework allows us to concentrate on a robust a secure framework that can be reasoned about thus made safer with far fewer unforeseen +attempting to force a one-size-fits-all approach. ``Kryo`` is more suited to the serialization of a program's stack frames, as it is more flexible +than our AMQP framework in what it can construct and serialize. However, that flexibility makes it exceptionally difficult to make secure. Conversely, +our AMQP framework allows us to concentrate on a secure framework that can be reasoned about and thus made safer, with far fewer security holes. .. note:: Selection of serialization context should, for the most part, be opaque to CorDapp developers, the Corda framework selecting - the correct context as confugred. + the correct context as configured. -.. For information on our choice of AMQP 1.0, see :doc:`amqp-choice`. For detail on how we utilise AMQP 1.0 and represent +.. note:: For information on our choice of AMQP 1.0, see :doc:`amqp-choice`. For detail on how we utilise AMQP 1.0 and represent objects in AMQP types, see :doc:`amqp-format`. -We describe here what is and will be supported in the Corda AMQP format from the perspective -of CorDapp developers, to allow for CorDapps to take into consideration the future state. The AMQP serialization format will of -course continue to apply the whitelisting functionality that is already in place and described in :doc:`serialization`. +This document describes what is currently and what will be supported in the Corda AMQP format from the perspective +of CorDapp developers, to allow CorDapps to take into consideration the future state. The AMQP serialization format will +continue to apply the whitelisting functionality that is already in place and described in :doc:`serialization`. Core Types ---------- -Here we describe the classes and interfaces that the AMQP serialization format will support. +This section describes the classes and interfaces that the AMQP serialization format supports. Collection Types ```````````````` The following collection types are supported. Any implementation of the following will be mapped to *an* implementation of the interface or class on the other end. -e.g. If you, for example, use a Guava implementation of a collection it will deserialize as a different implementation, -but will continue to adhere to the most specific of any of the following interfaces. You should use only these types -as the declared types of fields and properties, and not the concrete implementation types. Collections must be used -in their generic form, the generic type parameters will be included in the schema, and the elements type checked against the -generic parameters when deserialized. +For example, if you use a Guava implementation of a collection, it will deserialize as the primitive collection type. + +The declared types of properties should only use these types, and not any concrete implementation types (e.g. +Guava implementations). Collections must specify their generic type, the generic type parameters will be included in +the schema, and the element's type will be checked against the generic parameters when deserialized. :: @@ -121,8 +121,8 @@ generic parameters when deserialized. java.util.SortedMap java.util.NavigableMap -However, we will support the concrete implementation types below explicitly and also as the declared type of a field, as -a convenience. +However, as a convenience, we explicitly support the concrete implementation types below, and they can be used as the +declared types of properties. :: @@ -151,12 +151,12 @@ All the primitive types are supported. Arrays `````` -We also support arrays of any supported type, primitive or otherwise. +Arrays of any type are supported, primitive or otherwise. JDK Types ````````` -The following types are supported from the JDK libraries. +The following JDK library types are supported: :: @@ -200,10 +200,10 @@ The following types are supported from the JDK libraries. java.util.Currency java.util.UUID -Third Party Types +Third-Party Types ````````````````` -The following 3rd party types are supported. +The following 3rd-party types are supported: :: @@ -215,11 +215,11 @@ The following 3rd party types are supported. Corda Types ``````````` -Classes and interfaces in the Corda codebase annotated with ``@CordaSerializable`` are of course supported. +Any classes and interfaces in the Corda codebase annotated with ``@CordaSerializable`` are supported. -All Corda exceptions that are expected to be serialized inherit from ``CordaThrowable`` via either ``CordaException``, for -checked exceptions, or ``CordaRuntimeException``, for unchecked exceptions. Any ``Throwable`` that is serialized but does -not conform to ``CordaThrowable`` will be converted to a ``CordaRuntimeException`` with the original exception type +All Corda exceptions that are expected to be serialized inherit from ``CordaThrowable`` via either ``CordaException`` (for +checked exceptions) or ``CordaRuntimeException`` (for unchecked exceptions). Any ``Throwable`` that is serialized but does +not conform to ``CordaThrowable`` will be converted to a ``CordaRuntimeException``, with the original exception type and other properties retained within it. .. _amqp_custom_types_ref: @@ -227,7 +227,7 @@ and other properties retained within it. Custom Types ------------ -Here are the rules to adhere to for support of your own types: +You own types must adhere to the following rules to be supported: Classes ``````` @@ -236,34 +236,40 @@ General Rules ''''''''''''' #. The class must be compiled with parameter names included in the ``.class`` file. This is the default in Kotlin - but must be turned on in Java (``-parameters`` command line option to ``javac``). - #. The class is annotated with ``@CordaSerializable``. - #. The declared types of constructor arguments, getters, and setters must be supported, and where generics are used the + but must be turned on in Java using the ``-parameters`` command line option to ``javac`` + + .. note:: In circumstances where classes cannot be recompiled, such as when using a third-party library, a + proxy serializer can be used to avoid this problem. Details on creating such an object can be found on the + :doc:`cordapp-custom-serializers` page. + + #. The class must be annotated with ``@CordaSerializable`` + #. The declared types of constructor arguments, getters, and setters must be supported, and where generics are used, the generic parameter must be a supported type, an open wildcard (``*``), or a bounded wildcard which is currently - widened to an open wildcard. - #. Any superclass must adhere to the same rules, but can be abstract. - #. Object graph cycles are not supported, so an object cannot refer to itself, directly or indirectly. + widened to an open wildcard + #. Any superclass must adhere to the same rules, but can be abstract + #. Object graph cycles are not supported, so an object cannot refer to itself, directly or indirectly Constructor Instantiation ''''''''''''''''''''''''' -The primary way the AMQP serialization framework for Corda instantiates objects is via a defined constructor. This is -used to first determine which properties of an object are to be serialised then, on deserialization, it is used to +The primary way Corda's AMQP serialization framework instantiates objects is via a specified constructor. This is +used to first determine which properties of an object are to be serialised, then, on deserialization, it is used to instantiate the object with the serialized values. -This is the recommended design idiom for serializable objects in Corda as it allows for immutable state objects to -be created +It is recommended that serializable objects in Corda adhere to the following rules, as they allow immutable state +objects to be deserialised: - #. A Java Bean getter for each of the properties in the constructor, with the names matching up. For example, for a constructor - parameter ``foo``, there must be a getter called ``getFoo()``. If the type of ``foo`` is boolean, the getter may - optionally be called ``isFoo()``. This is why the class must be compiled with parameter names turned on. + #. A Java Bean getter for each of the properties in the constructor, with a name of the form ``getX``. For example, for a constructor + parameter ``foo``, there must be a getter called ``getFoo()``. If ``foo`` is a boolean, the getter may + optionally be called ``isFoo()`` (this is why the class must be compiled with parameter names turned on) #. A constructor which takes all of the properties that you wish to record in the serialized form. This is required in - order for the serialization framework to reconstruct an instance of your class. + order for the serialization framework to reconstruct an instance of your class #. If more than one constructor is provided, the serialization framework needs to know which one to use. The ``@ConstructorForDeserialization`` annotation can be used to indicate which one. For a Kotlin class, without the ``@ConstructorForDeserialization`` annotation, the - *primary constructor* will be selected. + *primary constructor* will be selected -In Kotlin, this maps cleanly to a data class where there getters are synthesized automatically. For example, +In Kotlin, this maps cleanly to a data class where there getters are synthesized automatically. For example, suppose we +have the following data class: .. container:: codeset @@ -271,9 +277,10 @@ In Kotlin, this maps cleanly to a data class where there getters are synthesized data class Example (val a: Int, val b: String) -Both properties a and b will be included in the serialised form. However, as stated above, properties not mentioned in -the constructor will not be serialised. For example, in the following code property c will not be considered part of the -serialised form +Properties ``a`` and ``b`` will be included in the serialised form. + +However, properties not mentioned in the constructor will not be serialised. For example, in the following code, +property ``c`` will not be considered part of the serialised form: .. container:: codeset @@ -291,14 +298,14 @@ serialised form Setter Instantiation '''''''''''''''''''' -As an alternative to constructor based initialisation Corda can also determine the important elements of an -object by inspecting the getter and setter methods present on a class. If a class has **only** a default +As an alternative to constructor-based initialisation, Corda can also determine the important elements of an +object by inspecting the getter and setter methods present on the class. If a class has **only** a default constructor **and** properties then the serializable properties will be determined by the presence of -both a getter and setter for that property that are both publicly visible. I.e. the class adheres to -the classic *idiom* of mutable JavaBeans. +both a getter and setter for that property that are both publicly visible (i.e. the class adheres to +the classic *idiom* of mutable JavaBeans). -On deserialization, a default instance will first be created and then, in turn, the setters invoked -on that object to populate the correct values. +On deserialization, a default instance will first be created, and then the setters will be invoked on that object to +populate it with the correct values. For example: @@ -324,7 +331,7 @@ Inaccessible Private Properties ``````````````````````````````` Whilst the Corda AMQP serialization framework supports private object properties without publicly -accessible getter methods this development idiom is strongly discouraged. +accessible getter methods, this development idiom is strongly discouraged. For example. @@ -350,15 +357,14 @@ For example. } } -When designing stateful objects is should be remembered that they are not, despite appearances, traditional +When designing stateful objects, is should be remembered that they are not, despite appearances, traditional programmatic constructs. They are signed over, transformed, serialised, and relationally mapped. As such, -all elements should be publicly accessible by design +all elements should be publicly accessible by design. -.. warning:: IDEs will indiciate erroneously that properties can be given something other than public - visibility. Ignore this as whilst it will work, as discussed above there are many reasons why this isn't - a good idea and those are beyond the scope of the IDEs inference rules +.. warning:: IDEs will indicate erroneously that properties can be given something other than public visibility. Ignore + this, as whilst it will work, as discussed above there are many reasons why this isn't a good idea. -Providing a public getter, as per the following example, is acceptable +Providing a public getter, as per the following example, is acceptable: .. container:: codeset @@ -392,7 +398,7 @@ Providing a public getter, as per the following example, is acceptable Enums ````` - #. All enums are supported, provided they are annotated with ``@CordaSerializable``. + #. All enums are supported, provided they are annotated with ``@CordaSerializable`` Exceptions @@ -401,24 +407,25 @@ Exceptions The following rules apply to supported ``Throwable`` implementations. #. If you wish for your exception to be serializable and transported type safely it should inherit from either - ``CordaException`` or ``CordaRuntimeException``. + ``CordaException`` or ``CordaRuntimeException`` #. If not, the ``Throwable`` will deserialize to a ``CordaRuntimeException`` with the details of the original - ``Throwable`` contained within it, including the class name of the original ``Throwable``. + ``Throwable`` contained within it, including the class name of the original ``Throwable`` Kotlin Objects `````````````` - #. Kotlin ``object`` s are singletons and treated differently. They are recorded into the stream with no properties - and deserialize back to the singleton instance. Currently, the same is not true of Java singletons, - and they will deserialize to new instances of the class. - #. Kotlin's anonymous ``object`` s are not currently supported. I.e. constructs like: - ``object : Contract {...}`` will not serialize correctly and need to be re-written as an explicit class declaration. + #. Kotlin's non-anonymous ``object`` s (i.e. constructs like ``object foo : Contract {...}``) are singletons and + treated differently. They are recorded into the stream with no properties, and deserialize back to the + singleton instance. Currently, the same is not true of Java singletons, which will deserialize to new instances + of the class + #. Kotlin's anonymous ``object`` s (i.e. constructs like ``object : Contract {...}``) are not currently supported + and will not serialize correctly. They need to be re-written as an explicit class declaration The Carpenter ````````````` -We will support a class carpenter that can dynamically manufacture classes from the supplied schema when deserializing -in the JVM without the supporting classes on the classpath. This can be useful where other components might expect to +We support a class carpenter that can dynamically manufacture classes from the supplied schema when deserializing, +without the supporting classes being present on the classpath. This can be useful where other components might expect to be able to use reflection over the deserialized data, and also for ensuring classes not on the classpath can be deserialized without loading potentially malicious code dynamically without security review outside of a fully sandboxed environment. A more detailed discussion of the carpenter will be provided in a future update to the documentation. @@ -427,25 +434,25 @@ Future Enhancements ``````````````````` #. Java singleton support. We will add support for identifying classes which are singletons and identifying the - static method responsible for returning the singleton instance. + static method responsible for returning the singleton instance #. Instance internalizing support. We will add support for identifying classes that should be resolved against an instances map to avoid - creating many duplicate instances that are equal. Similar to ``String.intern()``. + creating many duplicate instances that are equal (similar to ``String.intern()``) .. Type Evolution: Type Evolution -------------- -Type evolution is the mechanisms by which classes can be altered over time yet still remain serializable and deserializable across +Type evolution is the mechanism by which classes can be altered over time yet still remain serializable and deserializable across all versions of the class. This ensures an object serialized with an older idea of what the class "looked like" can be deserialized and a version of the current state of the class instantiated. -More detail can be found in :doc:`serialization-default-evolution` +More detail can be found in :doc:`serialization-default-evolution`. Enum Evolution `````````````` Corda supports interoperability of enumerated type versions. This allows such types to be changed over time without breaking -backward (or forward) compatibility. The rules and mechanisms for doing this are discussed in :doc:`serialization-enum-evolution`` +backward (or forward) compatibility. The rules and mechanisms for doing this are discussed in :doc:`serialization-enum-evolution``. diff --git a/experimental/flow-hook/src/main/kotlin/net/corda/flowhook/FlowHookContainer.kt b/experimental/flow-hook/src/main/kotlin/net/corda/flowhook/FlowHookContainer.kt index 686e3d46a4..abb55cca2c 100644 --- a/experimental/flow-hook/src/main/kotlin/net/corda/flowhook/FlowHookContainer.kt +++ b/experimental/flow-hook/src/main/kotlin/net/corda/flowhook/FlowHookContainer.kt @@ -2,7 +2,7 @@ package net.corda.flowhook import co.paralleluniverse.fibers.Fiber import net.corda.node.services.statemachine.Event -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager +import net.corda.nodeapi.internal.persistence.contextTransactionOrNull import java.sql.Connection @Suppress("UNUSED") @@ -156,7 +156,7 @@ object FlowHookContainer { private fun currentTransactionOrThread(): Any { return try { - DatabaseTransactionManager.currentOrNull() + contextTransactionOrNull } catch (exception: IllegalStateException) { null } ?: Thread.currentThread() diff --git a/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/CordaPersistence.kt b/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/CordaPersistence.kt index 5303b202f3..a1c6210a5e 100644 --- a/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/CordaPersistence.kt +++ b/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/CordaPersistence.kt @@ -1,13 +1,16 @@ package net.corda.nodeapi.internal.persistence +import co.paralleluniverse.strands.Strand import net.corda.core.schemas.MappedSchema import net.corda.core.utilities.contextLogger import rx.Observable import rx.Subscriber +import rx.subjects.PublishSubject import rx.subjects.UnicastSubject import java.io.Closeable import java.sql.Connection import java.sql.SQLException +import java.util.* import java.util.concurrent.CopyOnWriteArrayList import javax.persistence.AttributeConverter import javax.sql.DataSource @@ -40,6 +43,11 @@ enum class TransactionIsolationLevel { val jdbcValue: Int = java.sql.Connection::class.java.getField(jdbcString).get(null) as Int } +private val _contextDatabase = ThreadLocal() +var contextDatabase: CordaPersistence + get() = _contextDatabase.get() ?: error("Was expecting to find CordaPersistence set on current thread: ${Strand.currentStrand()}") + set(database) = _contextDatabase.set(database) + class CordaPersistence( val dataSource: DataSource, databaseConfig: DatabaseConfig, @@ -51,7 +59,7 @@ class CordaPersistence( private val log = contextLogger() } - val defaultIsolationLevel = databaseConfig.transactionIsolationLevel + private val defaultIsolationLevel = databaseConfig.transactionIsolationLevel val hibernateConfig: HibernateConfiguration by lazy { transaction { @@ -60,8 +68,19 @@ class CordaPersistence( } val entityManagerFactory get() = hibernateConfig.sessionFactoryForRegisteredSchemas + data class Boundary(val txId: UUID) + + internal val transactionBoundaries = PublishSubject.create().toSerialized() + init { - DatabaseTransactionManager(this) + // Found a unit test that was forgetting to close the database transactions. When you close() on the top level + // database transaction it will reset the threadLocalTx back to null, so if it isn't then there is still a + // database transaction open. The [transaction] helper above handles this in a finally clause for you + // but any manual database transaction management is liable to have this problem. + contextTransactionOrNull?.let { + error("Was not expecting to find existing database transaction on current strand when setting database: ${Strand.currentStrand()}, $it") + } + _contextDatabase.set(this) // Check not in read-only mode. transaction { check(!connection.metaData.isReadOnly) { "Database should not be readonly." } @@ -72,25 +91,29 @@ class CordaPersistence( const val DATA_SOURCE_URL = "dataSource.url" } - /** - * Creates an instance of [DatabaseTransaction], with the given transaction isolation level. - */ - fun createTransaction(isolationLevel: TransactionIsolationLevel): DatabaseTransaction { - // We need to set the database for the current [Thread] or [Fiber] here as some tests share threads across databases. - DatabaseTransactionManager.dataSource = this - return DatabaseTransactionManager.currentOrNew(isolationLevel) + fun currentOrNew(isolation: TransactionIsolationLevel = defaultIsolationLevel): DatabaseTransaction { + return contextTransactionOrNull ?: newTransaction(isolation) + } + + fun newTransaction(isolation: TransactionIsolationLevel = defaultIsolationLevel): DatabaseTransaction { + return DatabaseTransaction(isolation.jdbcValue, contextTransactionOrNull, this).also { + contextTransactionOrNull = it + } } /** - * Creates an instance of [DatabaseTransaction], with the default transaction isolation level. + * Creates an instance of [DatabaseTransaction], with the given transaction isolation level. */ - fun createTransaction(): DatabaseTransaction = createTransaction(defaultIsolationLevel) + fun createTransaction(isolationLevel: TransactionIsolationLevel = defaultIsolationLevel): DatabaseTransaction { + // We need to set the database for the current [Thread] or [Fiber] here as some tests share threads across databases. + _contextDatabase.set(this) + return currentOrNew(isolationLevel) + } fun createSession(): Connection { // We need to set the database for the current [Thread] or [Fiber] here as some tests share threads across databases. - DatabaseTransactionManager.dataSource = this - val ctx = DatabaseTransactionManager.currentOrNull() - return ctx?.connection ?: throw IllegalStateException("Was expecting to find database transaction: must wrap calling code within a transaction.") + _contextDatabase.set(this) + return contextTransaction.connection } /** @@ -99,7 +122,7 @@ class CordaPersistence( * @param statement to be executed in the scope of this transaction. */ fun transaction(isolationLevel: TransactionIsolationLevel, statement: DatabaseTransaction.() -> T): T { - DatabaseTransactionManager.dataSource = this + _contextDatabase.set(this) return transaction(isolationLevel, 2, statement) } @@ -110,7 +133,7 @@ class CordaPersistence( fun transaction(statement: DatabaseTransaction.() -> T): T = transaction(defaultIsolationLevel, statement) private fun transaction(isolationLevel: TransactionIsolationLevel, recoverableFailureTolerance: Int, statement: DatabaseTransaction.() -> T): T { - val outer = DatabaseTransactionManager.currentOrNull() + val outer = contextTransactionOrNull return if (outer != null) { outer.statement() } else { @@ -126,7 +149,7 @@ class CordaPersistence( log.warn("Cleanup task failed:", t) } while (true) { - val transaction = DatabaseTransactionManager.currentOrNew(isolationLevel) + val transaction = contextDatabase.currentOrNew(isolationLevel) // XXX: Does this code really support statement changing the contextDatabase? try { val answer = transaction.statement() transaction.commit() @@ -160,8 +183,8 @@ class CordaPersistence( * For examples, see the call hierarchy of this function. */ fun rx.Observer.bufferUntilDatabaseCommit(): rx.Observer { - val currentTxId = DatabaseTransactionManager.transactionId - val databaseTxBoundary: Observable = DatabaseTransactionManager.transactionBoundaries.first { it.txId == currentTxId } + val currentTxId = contextTransaction.id + val databaseTxBoundary: Observable = contextDatabase.transactionBoundaries.first { it.txId == currentTxId } val subject = UnicastSubject.create() subject.delaySubscription(databaseTxBoundary).subscribe(this) databaseTxBoundary.doOnCompleted { subject.onCompleted() } @@ -169,12 +192,12 @@ fun rx.Observer.bufferUntilDatabaseCommit(): rx.Observer { } // A subscriber that delegates to multiple others, wrapping a database transaction around the combination. -private class DatabaseTransactionWrappingSubscriber(val db: CordaPersistence?) : Subscriber() { +private class DatabaseTransactionWrappingSubscriber(private val db: CordaPersistence?) : Subscriber() { // Some unsubscribes happen inside onNext() so need something that supports concurrent modification. val delegates = CopyOnWriteArrayList>() fun forEachSubscriberWithDbTx(block: Subscriber.() -> Unit) { - (db ?: DatabaseTransactionManager.dataSource).transaction { + (db ?: contextDatabase).transaction { delegates.filter { !it.isUnsubscribed }.forEach { it.block() } diff --git a/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/DatabaseTransaction.kt b/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/DatabaseTransaction.kt index 4bb1bbd42e..ffcdd55c24 100644 --- a/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/DatabaseTransaction.kt +++ b/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/DatabaseTransaction.kt @@ -1,23 +1,29 @@ package net.corda.nodeapi.internal.persistence +import co.paralleluniverse.strands.Strand import org.hibernate.Session import org.hibernate.Transaction -import rx.subjects.Subject import java.sql.Connection import java.util.* +fun currentDBSession(): Session = contextTransaction.session +private val _contextTransaction = ThreadLocal() +var contextTransactionOrNull: DatabaseTransaction? + get() = _contextTransaction.get() + set(transaction) = _contextTransaction.set(transaction) +val contextTransaction get() = contextTransactionOrNull ?: error("Was expecting to find transaction set on current strand: ${Strand.currentStrand()}") + class DatabaseTransaction( isolation: Int, - private val threadLocal: ThreadLocal, - private val transactionBoundaries: Subject, - val cordaPersistence: CordaPersistence + val outerTransaction: DatabaseTransaction?, + val database: CordaPersistence ) { val id: UUID = UUID.randomUUID() private var _connectionCreated = false val connectionCreated get() = _connectionCreated val connection: Connection by lazy(LazyThreadSafetyMode.NONE) { - cordaPersistence.dataSource.connection + database.dataSource.connection .apply { _connectionCreated = true // only set the transaction isolation level if it's actually changed - setting isn't free. @@ -28,16 +34,13 @@ class DatabaseTransaction( } private val sessionDelegate = lazy { - val session = cordaPersistence.entityManagerFactory.withOptions().connection(connection).openSession() + val session = database.entityManagerFactory.withOptions().connection(connection).openSession() hibernateTransaction = session.beginTransaction() session } val session: Session by sessionDelegate private lateinit var hibernateTransaction: Transaction - - val outerTransaction: DatabaseTransaction? = threadLocal.get() - fun commit() { if (sessionDelegate.isInitialized()) { hibernateTransaction.commit() @@ -63,9 +66,9 @@ class DatabaseTransaction( if (_connectionCreated) { connection.close() } - threadLocal.set(outerTransaction) + contextTransactionOrNull = outerTransaction if (outerTransaction == null) { - transactionBoundaries.onNext(DatabaseTransactionManager.Boundary(id)) + database.transactionBoundaries.onNext(CordaPersistence.Boundary(id)) } } } diff --git a/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/DatabaseTransactionManager.kt b/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/DatabaseTransactionManager.kt deleted file mode 100644 index ade1603002..0000000000 --- a/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/DatabaseTransactionManager.kt +++ /dev/null @@ -1,77 +0,0 @@ -package net.corda.nodeapi.internal.persistence - -import co.paralleluniverse.strands.Strand -import org.hibernate.Session -import rx.subjects.PublishSubject -import rx.subjects.Subject -import java.util.* -import java.util.concurrent.ConcurrentHashMap - -fun currentDBSession(): Session = DatabaseTransactionManager.current().session - -class DatabaseTransactionManager(initDataSource: CordaPersistence) { - companion object { - private val threadLocalDb = ThreadLocal() - private val threadLocalTx = ThreadLocal() - private val databaseToInstance = ConcurrentHashMap() - - fun setThreadLocalTx(tx: DatabaseTransaction?): DatabaseTransaction? { - val oldTx = threadLocalTx.get() - threadLocalTx.set(tx) - return oldTx - } - - fun restoreThreadLocalTx(context: DatabaseTransaction?) { - if (context != null) { - threadLocalDb.set(context.cordaPersistence) - } - threadLocalTx.set(context) - } - - var dataSource: CordaPersistence - get() = threadLocalDb.get() ?: throw IllegalStateException("Was expecting to find CordaPersistence set on current thread: ${Strand.currentStrand()}") - set(value) = threadLocalDb.set(value) - - val transactionId: UUID - get() = threadLocalTx.get()?.id ?: throw IllegalStateException("Was expecting to find transaction set on current strand: ${Strand.currentStrand()}") - - val manager: DatabaseTransactionManager get() = databaseToInstance[dataSource]!! - - val transactionBoundaries: Subject get() = manager._transactionBoundaries - - fun currentOrNull(): DatabaseTransaction? = manager.currentOrNull() - - fun currentOrNew(isolation: TransactionIsolationLevel = dataSource.defaultIsolationLevel): DatabaseTransaction { - return currentOrNull() ?: manager.newTransaction(isolation.jdbcValue) - } - - fun current(): DatabaseTransaction = currentOrNull() ?: error("No transaction in context.") - - fun newTransaction(isolation: TransactionIsolationLevel = dataSource.defaultIsolationLevel): DatabaseTransaction { - return manager.newTransaction(isolation.jdbcValue) - } - } - - data class Boundary(val txId: UUID) - - private val _transactionBoundaries = PublishSubject.create().toSerialized() - - init { - // Found a unit test that was forgetting to close the database transactions. When you close() on the top level - // database transaction it will reset the threadLocalTx back to null, so if it isn't then there is still a - // database transaction open. The [transaction] helper above handles this in a finally clause for you - // but any manual database transaction management is liable to have this problem. - if (threadLocalTx.get() != null) { - throw IllegalStateException("Was not expecting to find existing database transaction on current strand when setting database: ${Strand.currentStrand()}, ${threadLocalTx.get()}") - } - dataSource = initDataSource - databaseToInstance[dataSource] = this - } - - private fun newTransaction(isolation: Int) = - DatabaseTransaction(isolation, threadLocalTx, transactionBoundaries, dataSource).apply { - threadLocalTx.set(this) - } - - private fun currentOrNull(): DatabaseTransaction? = threadLocalTx.get() -} diff --git a/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/HibernateConfiguration.kt b/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/HibernateConfiguration.kt index 0e17c06e4c..3861edb4a3 100644 --- a/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/HibernateConfiguration.kt +++ b/node-api/src/main/kotlin/net/corda/nodeapi/internal/persistence/HibernateConfiguration.kt @@ -128,15 +128,16 @@ class HibernateConfiguration( class NodeDatabaseConnectionProvider : ConnectionProvider { override fun closeConnection(conn: Connection) { conn.autoCommit = false - val tx = DatabaseTransactionManager.current() - tx.commit() - tx.close() + contextTransaction.run { + commit() + close() + } } override fun supportsAggressiveRelease(): Boolean = true override fun getConnection(): Connection { - return DatabaseTransactionManager.newTransaction().connection + return contextDatabase.newTransaction().connection } override fun unwrap(unwrapType: Class): T { 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 2c07f39c19..fb1a294345 100644 --- a/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt +++ b/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt @@ -222,7 +222,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration, check(started == null) { "Node has already been started" } log.info("Node starting up ...") initCertificate() - val schemaService = NodeSchemaService(cordappLoader.cordappSchemas) + val schemaService = NodeSchemaService(cordappLoader.cordappSchemas, configuration.notary != null) val (identity, identityKeyPair) = obtainIdentity(notaryConfig = null) val lh = lazyHub() configure(lh) diff --git a/node/src/main/kotlin/net/corda/node/services/config/NodeConfiguration.kt b/node/src/main/kotlin/net/corda/node/services/config/NodeConfiguration.kt index 62cbf3fa37..d8ad742d74 100644 --- a/node/src/main/kotlin/net/corda/node/services/config/NodeConfiguration.kt +++ b/node/src/main/kotlin/net/corda/node/services/config/NodeConfiguration.kt @@ -18,8 +18,6 @@ import java.util.* val Int.MB: Long get() = this * 1024L * 1024L interface NodeConfiguration : NodeSSLConfiguration { - // myLegalName should be only used in the initial network registration, we should use the name from the certificate instead of this. - // TODO: Remove this so we don't accidentally use this identity in the code? val myLegalName: CordaX500Name val emailAddress: String val exportJMXto: String @@ -35,8 +33,6 @@ interface NodeConfiguration : NodeSSLConfiguration { val notary: NotaryConfig? val activeMQServer: ActiveMqServerConfiguration val additionalNodeInfoPollingFrequencyMsec: Long - // TODO Remove as this is only used by the driver - val useHTTPS: Boolean val p2pAddress: NetworkHostAndPort val rpcAddress: NetworkHostAndPort? val messagingServerAddress: NetworkHostAndPort? @@ -127,7 +123,6 @@ data class NodeConfigurationImpl( // TODO typesafe config supports the notion of durations. Make use of that by mapping it to java.time.Duration. // Then rename this to messageRedeliveryDelay and make it of type Duration override val messageRedeliveryDelaySeconds: Int = 30, - override val useHTTPS: Boolean, override val p2pAddress: NetworkHostAndPort, override val rpcAddress: NetworkHostAndPort?, override val relay: RelayConfiguration?, diff --git a/node/src/main/kotlin/net/corda/node/services/persistence/DBCheckpointStorage.kt b/node/src/main/kotlin/net/corda/node/services/persistence/DBCheckpointStorage.kt index 91b64b81e6..6fed52b7d8 100644 --- a/node/src/main/kotlin/net/corda/node/services/persistence/DBCheckpointStorage.kt +++ b/node/src/main/kotlin/net/corda/node/services/persistence/DBCheckpointStorage.kt @@ -5,7 +5,6 @@ import net.corda.core.serialization.SerializedBytes import net.corda.core.utilities.debug import net.corda.node.services.api.CheckpointStorage import net.corda.node.services.statemachine.Checkpoint -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX import net.corda.nodeapi.internal.persistence.currentDBSession import org.slf4j.LoggerFactory @@ -43,7 +42,7 @@ class DBCheckpointStorage : CheckpointStorage { } override fun removeCheckpoint(id: StateMachineRunId): Boolean { - val session = DatabaseTransactionManager.current().session + val session = currentDBSession() val criteriaBuilder = session.criteriaBuilder val delete = criteriaBuilder.createCriteriaDelete(DBCheckpoint::class.java) val root = delete.from(DBCheckpoint::class.java) diff --git a/node/src/main/kotlin/net/corda/node/services/persistence/NodeAttachmentService.kt b/node/src/main/kotlin/net/corda/node/services/persistence/NodeAttachmentService.kt index da6dd489f8..60c5eb373a 100644 --- a/node/src/main/kotlin/net/corda/node/services/persistence/NodeAttachmentService.kt +++ b/node/src/main/kotlin/net/corda/node/services/persistence/NodeAttachmentService.kt @@ -17,7 +17,6 @@ import net.corda.core.node.services.vault.AttachmentSort import net.corda.core.serialization.* import net.corda.core.utilities.contextLogger import net.corda.node.services.vault.HibernateAttachmentQueryCriteriaParser -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX import net.corda.nodeapi.internal.persistence.currentDBSession import java.io.* @@ -242,8 +241,7 @@ class NodeAttachmentService(metrics: MetricRegistry) : AttachmentStorage, Single override fun queryAttachments(criteria: AttachmentQueryCriteria, sorting: AttachmentSort?): List { log.info("Attachment query criteria: $criteria, sorting: $sorting") - - val session = DatabaseTransactionManager.current().session + val session = currentDBSession() val criteriaBuilder = session.criteriaBuilder val criteriaQuery = criteriaBuilder.createQuery(DBAttachment::class.java) diff --git a/node/src/main/kotlin/net/corda/node/services/schema/HibernateObserver.kt b/node/src/main/kotlin/net/corda/node/services/schema/HibernateObserver.kt index eb58d0871e..62d5cee4d8 100644 --- a/node/src/main/kotlin/net/corda/node/services/schema/HibernateObserver.kt +++ b/node/src/main/kotlin/net/corda/node/services/schema/HibernateObserver.kt @@ -10,8 +10,8 @@ import net.corda.core.schemas.PersistentStateRef import net.corda.core.utilities.contextLogger import net.corda.core.utilities.debug import net.corda.node.services.api.SchemaService -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager import net.corda.nodeapi.internal.persistence.HibernateConfiguration +import net.corda.nodeapi.internal.persistence.contextTransaction import org.hibernate.FlushMode import rx.Observable @@ -54,7 +54,7 @@ class HibernateObserver private constructor(private val config: HibernateConfigu internal fun persistStatesWithSchema(statesAndRefs: List, schema: MappedSchema) { val sessionFactory = config.sessionFactoryForSchemas(setOf(schema)) val session = sessionFactory.withOptions(). - connection(DatabaseTransactionManager.current().connection). + connection(contextTransaction.connection). flushMode(FlushMode.MANUAL). openSession() session.use { thisSession -> diff --git a/node/src/main/kotlin/net/corda/node/services/schema/NodeSchemaService.kt b/node/src/main/kotlin/net/corda/node/services/schema/NodeSchemaService.kt index 108488479e..9f0cd31b61 100644 --- a/node/src/main/kotlin/net/corda/node/services/schema/NodeSchemaService.kt +++ b/node/src/main/kotlin/net/corda/node/services/schema/NodeSchemaService.kt @@ -29,24 +29,20 @@ import net.corda.node.services.vault.VaultSchemaV1 * TODO: support plugins for schema version upgrading or custom mapping not supported by original [QueryableState]. * TODO: create whitelisted tables when a CorDapp is first installed */ -class NodeSchemaService(extraSchemas: Set = emptySet()) : SchemaService, SingletonSerializeAsToken() { - // Entities for compulsory services - object NodeServices +class NodeSchemaService(extraSchemas: Set = emptySet(), includeNotarySchemas: Boolean = false) : SchemaService, SingletonSerializeAsToken() { + // Core Entities used by a Node + object NodeCore - object NodeServicesV1 : MappedSchema(schemaFamily = NodeServices.javaClass, version = 1, + object NodeCoreV1 : MappedSchema(schemaFamily = NodeCore.javaClass, version = 1, mappedTypes = listOf(DBCheckpointStorage.DBCheckpoint::class.java, DBTransactionStorage.DBTransaction::class.java, DBTransactionMappingStorage.DBTransactionMapping::class.java, PersistentKeyManagementService.PersistentKey::class.java, - PersistentUniquenessProvider.PersistentUniqueness::class.java, - PersistentUniquenessProvider.PersistentNotaryCommit::class.java, NodeSchedulerService.PersistentScheduledState::class.java, NodeAttachmentService.DBAttachment::class.java, P2PMessagingClient.ProcessedMessage::class.java, P2PMessagingClient.RetryMessage::class.java, NodeAttachmentService.DBAttachment::class.java, - RaftUniquenessProvider.RaftState::class.java, - BFTNonValidatingNotaryService.PersistedCommittedState::class.java, PersistentIdentityService.PersistentIdentity::class.java, PersistentIdentityService.PersistentIdentityNames::class.java, ContractUpgradeServiceImpl.DBContractUpgrade::class.java, @@ -55,15 +51,25 @@ class NodeSchemaService(extraSchemas: Set = emptySet()) : SchemaSe override val migrationResource = "node-services.changelog-master" } + // Entities used by a Notary + object NodeNotary + + object NodeNotaryV1 : MappedSchema(schemaFamily = NodeNotary.javaClass, version = 1, + mappedTypes = listOf(PersistentUniquenessProvider.PersistentUniqueness::class.java, + PersistentUniquenessProvider.PersistentNotaryCommit::class.java, + RaftUniquenessProvider.RaftState::class.java, + BFTNonValidatingNotaryService.PersistedCommittedState::class.java + )) + // Required schemas are those used by internal Corda services - // For example, cash is used by the vault for coin selection (but will be extracted as a standalone CorDapp in future) private val requiredSchemas: Map = mapOf(Pair(CommonSchemaV1, SchemaOptions()), - Pair(VaultSchemaV1, SchemaOptions()), - Pair(NodeInfoSchemaV1, SchemaOptions()), - Pair(NodeServicesV1, SchemaOptions())) + Pair(VaultSchemaV1, SchemaOptions()), + Pair(NodeInfoSchemaV1, SchemaOptions()), + Pair(NodeCoreV1, SchemaOptions())) + private val notarySchemas = if (includeNotarySchemas) mapOf(Pair(NodeNotaryV1, SchemaOptions())) else emptyMap() - override val schemaOptions: Map = requiredSchemas + extraSchemas.associateBy({ it }, { SchemaOptions() }) + override val schemaOptions: Map = requiredSchemas + notarySchemas + extraSchemas.associateBy({ it }, { SchemaOptions() }) // Currently returns all schemas supported by the state, with no filtering or enrichment. override fun selectSchemas(state: ContractState): Iterable { diff --git a/node/src/main/kotlin/net/corda/node/services/statemachine/ActionExecutorImpl.kt b/node/src/main/kotlin/net/corda/node/services/statemachine/ActionExecutorImpl.kt index 6449a8b090..6efdc09f7b 100644 --- a/node/src/main/kotlin/net/corda/node/services/statemachine/ActionExecutorImpl.kt +++ b/node/src/main/kotlin/net/corda/node/services/statemachine/ActionExecutorImpl.kt @@ -10,7 +10,9 @@ import net.corda.core.utilities.contextLogger import net.corda.core.utilities.trace import net.corda.node.services.api.CheckpointStorage import net.corda.node.services.api.ServiceHubInternal -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager +import net.corda.nodeapi.internal.persistence.contextDatabase +import net.corda.nodeapi.internal.persistence.contextTransaction +import net.corda.nodeapi.internal.persistence.contextTransactionOrNull import java.time.Duration import java.time.Instant import java.util.concurrent.TimeUnit @@ -163,24 +165,24 @@ class ActionExecutorImpl( @Suspendable private fun executeCreateTransaction() { - if (DatabaseTransactionManager.currentOrNull() != null) { + if (contextTransactionOrNull != null) { throw IllegalStateException("Refusing to create a second transaction") } - DatabaseTransactionManager.newTransaction() + contextDatabase.newTransaction() } @Suspendable private fun executeRollbackTransaction() { - DatabaseTransactionManager.currentOrNull()?.close() + contextTransactionOrNull?.close() } @Suspendable private fun executeCommitTransaction() { try { - DatabaseTransactionManager.current().commit() + contextTransaction.commit() } finally { - DatabaseTransactionManager.current().close() - DatabaseTransactionManager.setThreadLocalTx(null) + contextTransaction.close() + contextTransactionOrNull = null } } 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 4b0408c622..dca556ea59 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 @@ -27,7 +27,8 @@ import net.corda.node.services.statemachine.transitions.FlowContinuation import net.corda.node.services.statemachine.transitions.StateMachine import net.corda.nodeapi.internal.persistence.CordaPersistence import net.corda.nodeapi.internal.persistence.DatabaseTransaction -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager +import net.corda.nodeapi.internal.persistence.contextTransaction +import net.corda.nodeapi.internal.persistence.contextTransactionOrNull import org.slf4j.Logger import org.slf4j.LoggerFactory import java.nio.file.Paths @@ -58,8 +59,8 @@ class FlowStateMachineImpl(override val id: StateMachineRunId, } private fun extractThreadLocalTransaction(): TransientReference { - val transaction = DatabaseTransactionManager.current() - DatabaseTransactionManager.setThreadLocalTx(null) + val transaction = contextTransaction + contextTransactionOrNull = null return TransientReference(transaction) } } @@ -234,7 +235,7 @@ class FlowStateMachineImpl(override val id: StateMachineRunId, parkAndSerialize { _, _ -> logger.trace { "Suspended on $ioRequest" } - DatabaseTransactionManager.setThreadLocalTx(transaction.value) + contextTransactionOrNull = transaction.value val event = try { Event.Suspend( ioRequest = ioRequest, diff --git a/node/src/main/kotlin/net/corda/node/services/statemachine/TransitionExecutorImpl.kt b/node/src/main/kotlin/net/corda/node/services/statemachine/TransitionExecutorImpl.kt index b895a40863..2cf328a450 100644 --- a/node/src/main/kotlin/net/corda/node/services/statemachine/TransitionExecutorImpl.kt +++ b/node/src/main/kotlin/net/corda/node/services/statemachine/TransitionExecutorImpl.kt @@ -5,7 +5,8 @@ import net.corda.core.utilities.contextLogger import net.corda.node.services.statemachine.transitions.FlowContinuation import net.corda.node.services.statemachine.transitions.TransitionResult import net.corda.nodeapi.internal.persistence.CordaPersistence -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager +import net.corda.nodeapi.internal.persistence.contextDatabase +import net.corda.nodeapi.internal.persistence.contextTransactionOrNull import java.security.SecureRandom /** @@ -31,12 +32,12 @@ class TransitionExecutorImpl( transition: TransitionResult, actionExecutor: ActionExecutor ): Pair { - DatabaseTransactionManager.dataSource = database + contextDatabase = database for (action in transition.actions) { try { actionExecutor.executeAction(fiber, action) } catch (exception: Throwable) { - DatabaseTransactionManager.currentOrNull()?.close() + contextTransactionOrNull?.close() if (transition.newState.checkpoint.errorState is ErrorState.Errored) { // If we errored while transitioning to an error state then we cannot record the additional // error as that may result in an infinite loop, e.g. error propagation fails -> record error -> propagate fails again. diff --git a/node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt b/node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt index f215d87cee..a731f4841e 100644 --- a/node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt +++ b/node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt @@ -17,12 +17,8 @@ import net.corda.core.transactions.NotaryChangeWireTransaction import net.corda.core.transactions.WireTransaction import net.corda.core.utilities.* import net.corda.node.services.api.VaultServiceInternal -import net.corda.nodeapi.internal.persistence.HibernateConfiguration import net.corda.node.services.statemachine.FlowStateMachineImpl -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager -import net.corda.nodeapi.internal.persistence.bufferUntilDatabaseCommit -import net.corda.nodeapi.internal.persistence.currentDBSession -import net.corda.nodeapi.internal.persistence.wrapWithDatabaseTransaction +import net.corda.nodeapi.internal.persistence.* import org.hibernate.Session import rx.Observable import rx.subjects.PublishSubject @@ -479,8 +475,7 @@ class NodeVaultService( } } - private fun getSession() = DatabaseTransactionManager.currentOrNew().session - + private fun getSession() = contextDatabase.currentOrNew().session /** * Derive list from existing vault states and then incrementally update using vault observables */ diff --git a/node/src/main/kotlin/net/corda/node/services/vault/VaultSoftLockManager.kt b/node/src/main/kotlin/net/corda/node/services/vault/VaultSoftLockManager.kt index 7e13cfab85..33cd73e74b 100644 --- a/node/src/main/kotlin/net/corda/node/services/vault/VaultSoftLockManager.kt +++ b/node/src/main/kotlin/net/corda/node/services/vault/VaultSoftLockManager.kt @@ -10,7 +10,7 @@ import net.corda.core.utilities.toNonEmptySet import net.corda.core.utilities.trace import net.corda.node.services.statemachine.FlowStateMachineImpl import net.corda.node.services.statemachine.StateMachineManager -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager +import net.corda.nodeapi.internal.persistence.contextDatabase import java.util.* class VaultSoftLockManager private constructor(private val vault: VaultService) { @@ -52,14 +52,14 @@ class VaultSoftLockManager private constructor(private val vault: VaultService) private fun registerSoftLocks(flowId: UUID, stateRefs: NonEmptySet) { log.trace { "Reserving soft locks for flow id $flowId and states $stateRefs" } - DatabaseTransactionManager.dataSource.transaction { + contextDatabase.transaction { vault.softLockReserve(flowId, stateRefs) } } private fun unregisterSoftLocks(flowId: UUID, logic: FlowLogic<*>) { log.trace { "Releasing soft locks for flow ${logic.javaClass.simpleName} with flow id $flowId" } - DatabaseTransactionManager.dataSource.transaction { + contextDatabase.transaction { vault.softLockRelease(flowId) } } diff --git a/node/src/main/resources/reference.conf b/node/src/main/resources/reference.conf index 7ce5aaa6b7..eab9ede6e0 100644 --- a/node/src/main/resources/reference.conf +++ b/node/src/main/resources/reference.conf @@ -14,7 +14,6 @@ database = { exportHibernateJMXStatistics = "false" } devMode = true -useHTTPS = false h2port = 0 useTestClock = false verifierType = InMemory diff --git a/node/src/test/kotlin/net/corda/node/services/config/NodeConfigurationImplTest.kt b/node/src/test/kotlin/net/corda/node/services/config/NodeConfigurationImplTest.kt index 18388b6338..9df1ef43c6 100644 --- a/node/src/test/kotlin/net/corda/node/services/config/NodeConfigurationImplTest.kt +++ b/node/src/test/kotlin/net/corda/node/services/config/NodeConfigurationImplTest.kt @@ -76,7 +76,6 @@ class NodeConfigurationImplTest { dataSourceProperties = makeTestDataSourceProperties(ALICE_NAME.organisation), rpcUsers = emptyList(), verifierType = VerifierType.InMemory, - useHTTPS = false, p2pAddress = NetworkHostAndPort("localhost", 0), rpcAddress = NetworkHostAndPort("localhost", 1), messagingServerAddress = null, diff --git a/node/src/test/kotlin/net/corda/node/services/schema/HibernateObserverTests.kt b/node/src/test/kotlin/net/corda/node/services/schema/HibernateObserverTests.kt index 0a657ad95e..c83b75b89f 100644 --- a/node/src/test/kotlin/net/corda/node/services/schema/HibernateObserverTests.kt +++ b/node/src/test/kotlin/net/corda/node/services/schema/HibernateObserverTests.kt @@ -15,7 +15,6 @@ import net.corda.core.schemas.QueryableState import net.corda.node.services.api.SchemaService import net.corda.node.internal.configureDatabase import net.corda.nodeapi.internal.persistence.DatabaseConfig -import net.corda.nodeapi.internal.persistence.DatabaseTransactionManager import net.corda.testing.internal.LogHelper import net.corda.testing.TestIdentity import net.corda.testing.contracts.DummyContract @@ -74,11 +73,11 @@ class HibernateObserverTests { database.transaction { val MEGA_CORP = TestIdentity(CordaX500Name("MegaCorp", "London", "GB")).party rawUpdatesPublisher.onNext(Vault.Update(emptySet(), setOf(StateAndRef(TransactionState(TestState(), DummyContract.PROGRAM_ID, MEGA_CORP), StateRef(SecureHash.sha256("dummy"), 0))))) - val parentRowCountResult = DatabaseTransactionManager.current().connection.prepareStatement("select count(*) from Parents").executeQuery() + val parentRowCountResult = connection.prepareStatement("select count(*) from Parents").executeQuery() parentRowCountResult.next() val parentRows = parentRowCountResult.getInt(1) parentRowCountResult.close() - val childrenRowCountResult = DatabaseTransactionManager.current().connection.prepareStatement("select count(*) from Children").executeQuery() + val childrenRowCountResult = connection.prepareStatement("select count(*) from Children").executeQuery() childrenRowCountResult.next() val childrenRows = childrenRowCountResult.getInt(1) childrenRowCountResult.close() diff --git a/node/src/test/kotlin/net/corda/node/services/schema/NodeSchemaServiceTest.kt b/node/src/test/kotlin/net/corda/node/services/schema/NodeSchemaServiceTest.kt index ac9a88afe0..521a1f34b3 100644 --- a/node/src/test/kotlin/net/corda/node/services/schema/NodeSchemaServiceTest.kt +++ b/node/src/test/kotlin/net/corda/node/services/schema/NodeSchemaServiceTest.kt @@ -9,15 +9,19 @@ import net.corda.core.schemas.MappedSchema import net.corda.core.schemas.PersistentState import net.corda.core.utilities.getOrThrow import net.corda.node.services.api.ServiceHubInternal +import net.corda.node.services.schema.NodeSchemaService.NodeCoreV1 +import net.corda.node.services.schema.NodeSchemaService.NodeNotaryV1 import net.corda.testing.driver.NodeHandle import net.corda.testing.driver.driver -import net.corda.testing.node.MockNetwork import net.corda.testing.internal.vault.DummyLinearStateSchemaV1 +import net.corda.testing.node.MockNetwork import org.hibernate.annotations.Cascade import org.hibernate.annotations.CascadeType +import org.junit.Ignore import org.junit.Test import javax.persistence.* import kotlin.test.assertEquals +import kotlin.test.assertFalse import kotlin.test.assertTrue class NodeSchemaServiceTest { @@ -30,7 +34,30 @@ class NodeSchemaServiceTest { val mockNode = mockNet.createNode() val schemaService = mockNode.services.schemaService assertTrue(schemaService.schemaOptions.containsKey(DummyLinearStateSchemaV1)) + mockNet.stopNodes() + } + @Test + fun `check node runs with minimal core schema set`() { + val mockNet = MockNetwork(cordappPackages = emptyList()) + val mockNode = mockNet.createNode() + val schemaService = mockNode.services.schemaService + + // check against NodeCore schemas + assertTrue(schemaService.schemaOptions.containsKey(NodeCoreV1)) + assertFalse(schemaService.schemaOptions.containsKey(NodeNotaryV1)) + mockNet.stopNodes() + } + + @Test + fun `check node runs inclusive of notary node schema set`() { + val mockNet = MockNetwork(cordappPackages = emptyList()) + val mockNotaryNode = mockNet.notaryNodes.first() + val schemaService = mockNotaryNode.services.schemaService + + // check against NodeCore + NodeNotary Schemas + assertTrue(schemaService.schemaOptions.containsKey(NodeCoreV1)) + assertTrue(schemaService.schemaOptions.containsKey(NodeNotaryV1)) mockNet.stopNodes() } @@ -59,6 +86,34 @@ class NodeSchemaServiceTest { assertEquals>(expected, tables.toMutableSet().apply { retainAll(expected) }) } + @Ignore + @Test + fun `check node runs with minimal core schema set using driverDSL`() { + // TODO: driver limitation: cannot restrict CorDapps that get automatically created by default, + // can ONLY specify additional ones using `extraCordappPackagesToScan` constructor argument. + driver(startNodesInProcess = true, notarySpecs = emptyList()) { + val node = startNode().getOrThrow() + val result = node.rpc.startFlow(::MappedSchemasFlow) + val mappedSchemas = result.returnValue.getOrThrow() + // check against NodeCore schemas + assertTrue(mappedSchemas.contains(NodeCoreV1.name)) + assertFalse(mappedSchemas.contains(NodeNotaryV1.name)) // still gets loaded due TODO restriction + } + + } + + @Test + fun `check node runs inclusive of notary node schema set using driverDSL`() { + driver(startNodesInProcess = true) { + val notaryNode = defaultNotaryNode.getOrThrow().rpc.startFlow(::MappedSchemasFlow) + val mappedSchemas = notaryNode.returnValue.getOrThrow() + // check against NodeCore + NodeNotary Schemas + assertTrue(mappedSchemas.contains(NodeCoreV1.name)) + assertTrue(mappedSchemas.contains(NodeNotaryV1.name)) + } + + } + @StartableByRPC class MappedSchemasFlow : FlowLogic>() { @Suspendable diff --git a/node/src/test/kotlin/net/corda/node/services/transactions/DistributedImmutableMapTests.kt b/node/src/test/kotlin/net/corda/node/services/transactions/DistributedImmutableMapTests.kt index 1c60db0ba6..6b78bb53c2 100644 --- a/node/src/test/kotlin/net/corda/node/services/transactions/DistributedImmutableMapTests.kt +++ b/node/src/test/kotlin/net/corda/node/services/transactions/DistributedImmutableMapTests.kt @@ -11,6 +11,7 @@ import net.corda.core.internal.concurrent.transpose import net.corda.core.utilities.NetworkHostAndPort import net.corda.core.utilities.getOrThrow import net.corda.node.internal.configureDatabase +import net.corda.node.services.schema.NodeSchemaService import net.corda.nodeapi.internal.persistence.CordaPersistence import net.corda.nodeapi.internal.persistence.DatabaseConfig import net.corda.testing.internal.LogHelper @@ -88,7 +89,7 @@ class DistributedImmutableMapTests { private fun createReplica(myAddress: NetworkHostAndPort, clusterAddress: NetworkHostAndPort? = null): CompletableFuture { val storage = Storage.builder().withStorageLevel(StorageLevel.MEMORY).build() val address = Address(myAddress.host, myAddress.port) - val database = configureDatabase(makeTestDataSourceProperties(), DatabaseConfig(runMigration = true), rigorousMock()) + val database = configureDatabase(makeTestDataSourceProperties(), DatabaseConfig(runMigration = true), rigorousMock(), NodeSchemaService(includeNotarySchemas = true)) databases.add(database) val stateMachineFactory = { DistributedImmutableMap(database, RaftUniquenessProvider.Companion::createMap) } diff --git a/node/src/test/kotlin/net/corda/node/services/transactions/PersistentUniquenessProviderTests.kt b/node/src/test/kotlin/net/corda/node/services/transactions/PersistentUniquenessProviderTests.kt index 7e25d42c9e..cf833dbcbe 100644 --- a/node/src/test/kotlin/net/corda/node/services/transactions/PersistentUniquenessProviderTests.kt +++ b/node/src/test/kotlin/net/corda/node/services/transactions/PersistentUniquenessProviderTests.kt @@ -4,6 +4,7 @@ import net.corda.core.crypto.SecureHash import net.corda.core.identity.CordaX500Name import net.corda.core.node.services.UniquenessException import net.corda.node.internal.configureDatabase +import net.corda.node.services.schema.NodeSchemaService import net.corda.nodeapi.internal.persistence.CordaPersistence import net.corda.nodeapi.internal.persistence.DatabaseConfig import net.corda.testing.* @@ -29,7 +30,7 @@ class PersistentUniquenessProviderTests { @Before fun setUp() { LogHelper.setLevel(PersistentUniquenessProvider::class) - database = configureDatabase(makeTestDataSourceProperties(), DatabaseConfig(runMigration = true), rigorousMock()) + database = configureDatabase(makeTestDataSourceProperties(), DatabaseConfig(runMigration = true), rigorousMock(), NodeSchemaService(includeNotarySchemas = true)) } @After diff --git a/node/src/test/kotlin/net/corda/node/utilities/ObservablesTests.kt b/node/src/test/kotlin/net/corda/node/utilities/ObservablesTests.kt index 68918167f0..9aeb4a96d8 100644 --- a/node/src/test/kotlin/net/corda/node/utilities/ObservablesTests.kt +++ b/node/src/test/kotlin/net/corda/node/utilities/ObservablesTests.kt @@ -16,9 +16,7 @@ import java.io.Closeable import java.util.* class ObservablesTests { - - private fun isInDatabaseTransaction(): Boolean = (DatabaseTransactionManager.currentOrNull() != null) - + private fun isInDatabaseTransaction() = contextTransactionOrNull != null private val toBeClosed = mutableListOf() private fun createDatabase(): CordaPersistence { @@ -168,7 +166,7 @@ class ObservablesTests { observableWithDbTx.first().subscribe { undelayedEvent.set(it to isInDatabaseTransaction()) } fun observeSecondEvent(event: Int, future: SettableFuture>) { - future.set(event to if (isInDatabaseTransaction()) DatabaseTransactionManager.transactionId else null) + future.set(event to if (isInDatabaseTransaction()) contextTransaction.id else null) } observableWithDbTx.skip(1).first().subscribe { observeSecondEvent(it, delayedEventFromSecondObserver) } diff --git a/samples/irs-demo/src/integration-test/kotlin/net/corda/test/spring/SpringDriver.kt b/samples/irs-demo/src/integration-test/kotlin/net/corda/test/spring/SpringDriver.kt index c7d7a57ffd..9d83d481ba 100644 --- a/samples/irs-demo/src/integration-test/kotlin/net/corda/test/spring/SpringDriver.kt +++ b/samples/irs-demo/src/integration-test/kotlin/net/corda/test/spring/SpringDriver.kt @@ -71,7 +71,7 @@ data class SpringBootDriverDSL(private val driverDSL: DriverDSLImpl) : InternalD } private fun queryWebserver(handle: NodeHandle, process: Process, checkUrl: String): WebserverHandle { - val protocol = if (handle.configuration.useHTTPS) "https://" else "http://" + val protocol = if (handle.useHTTPS) "https://" else "http://" val url = URL(URL("$protocol${handle.webAddress}"), checkUrl) val client = OkHttpClient.Builder().connectTimeout(5, TimeUnit.SECONDS).readTimeout(10, TimeUnit.SECONDS).build() diff --git a/testing/node-driver/src/main/kotlin/net/corda/testing/driver/Driver.kt b/testing/node-driver/src/main/kotlin/net/corda/testing/driver/Driver.kt index 4a8c6101c9..a7316c18f4 100644 --- a/testing/node-driver/src/main/kotlin/net/corda/testing/driver/Driver.kt +++ b/testing/node-driver/src/main/kotlin/net/corda/testing/driver/Driver.kt @@ -15,8 +15,8 @@ import net.corda.node.internal.StartedNode import net.corda.node.services.config.NodeConfiguration import net.corda.node.services.config.VerifierType import net.corda.testing.DUMMY_NOTARY_NAME -import net.corda.testing.node.User import net.corda.testing.node.NotarySpec +import net.corda.testing.node.User import net.corda.testing.node.internal.DriverDSLImpl import net.corda.testing.node.internal.genericDriver import net.corda.testing.node.internal.getTimestampAsDirectoryName @@ -41,6 +41,7 @@ sealed class NodeHandle { abstract val rpc: CordaRPCOps abstract val configuration: NodeConfiguration abstract val webAddress: NetworkHostAndPort + abstract val useHTTPS: Boolean /** * Stops the referenced node. @@ -52,6 +53,7 @@ sealed class NodeHandle { override val rpc: CordaRPCOps, override val configuration: NodeConfiguration, override val webAddress: NetworkHostAndPort, + override val useHTTPS: Boolean, val debugPort: Int?, val process: Process, private val onStopCallback: () -> Unit @@ -70,6 +72,7 @@ sealed class NodeHandle { override val rpc: CordaRPCOps, override val configuration: NodeConfiguration, override val webAddress: NetworkHostAndPort, + override val useHTTPS: Boolean, val node: StartedNode, val nodeThread: Thread, private val onStopCallback: () -> Unit diff --git a/testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/DriverDSLImpl.kt b/testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/DriverDSLImpl.kt index f9df016fdb..962469c20e 100644 --- a/testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/DriverDSLImpl.kt +++ b/testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/DriverDSLImpl.kt @@ -330,7 +330,7 @@ class DriverDSLImpl( } private fun queryWebserver(handle: NodeHandle, process: Process): WebserverHandle { - val protocol = if (handle.configuration.useHTTPS) "https://" else "http://" + val protocol = if (handle.useHTTPS) "https://" else "http://" val url = URL("$protocol${handle.webAddress}/api/status") val client = OkHttpClient.Builder().connectTimeout(5, TimeUnit.SECONDS).readTimeout(60, TimeUnit.SECONDS).build() @@ -604,10 +604,14 @@ class DriverDSLImpl( val baseDirectory = config.corda.baseDirectory.createDirectories() localNetworkMap?.networkParametersCopier?.install(baseDirectory) localNetworkMap?.nodeInfosCopier?.addConfig(baseDirectory) + val onNodeExit: () -> Unit = { localNetworkMap?.nodeInfosCopier?.removeConfig(baseDirectory) countObservables.remove(config.corda.myLegalName) } + + val useHTTPS = config.typesafe.run { hasPath("useHTTPS") && getBoolean("useHTTPS") } + if (startInProcess ?: startNodesInProcess) { val nodeAndThreadFuture = startInProcessNode(executorService, config, cordappPackages) shutdownManager.registerShutdown( @@ -621,7 +625,7 @@ class DriverDSLImpl( return nodeAndThreadFuture.flatMap { (node, thread) -> establishRpc(config, openFuture()).flatMap { rpc -> allNodesConnected(rpc).map { - NodeHandle.InProcess(rpc.nodeInfo(), rpc, config.corda, webAddress, node, thread, onNodeExit) + NodeHandle.InProcess(rpc.nodeInfo(), rpc, config.corda, webAddress, useHTTPS, node, thread, onNodeExit) } } } @@ -650,7 +654,7 @@ class DriverDSLImpl( } processDeathFuture.cancel(false) log.info("Node handle is ready. NodeInfo: ${rpc.nodeInfo()}, WebAddress: $webAddress") - NodeHandle.OutOfProcess(rpc.nodeInfo(), rpc, config.corda, webAddress, debugPort, process, onNodeExit) + NodeHandle.OutOfProcess(rpc.nodeInfo(), rpc, config.corda, webAddress, useHTTPS, debugPort, process, onNodeExit) } } }