CORDA-1847 - Fix hibernate observer issue (#3694)

* CORDA-1847 Fix hibernate observer issue * CORDA-1847 Fix merge
2024-12-18 20:47:57 +00:00 · 2018-07-26 12:12:53 +01:00 · 2018-07-26 12:12:53 +01:00 · 0bce5311a7
commit 0bce5311a7
parent f5d26aa25d
25 changed files with 73 additions and 84 deletions
--- a/confidential-identities/src/test/kotlin/net/corda/confidential/IdentitySyncFlowTests.kt
+++ b/confidential-identities/src/test/kotlin/net/corda/confidential/IdentitySyncFlowTests.kt
@ -34,7 +34,7 @@ class IdentitySyncFlowTests {
    fun before() {
        // We run this in parallel threads to help catch any race conditions that may exist.
        mockNet = InternalMockNetwork(
-                cordappPackages = listOf("net.corda.finance.contracts.asset"),
+                cordappPackages = listOf("net.corda.finance.contracts.asset", "net.corda.finance.schemas"),
                networkSendManuallyPumped = false,
                threadPerNode = true
        )
--- a/core/src/test/kotlin/net/corda/core/flows/CollectSignaturesFlowTests.kt
+++ b/core/src/test/kotlin/net/corda/core/flows/CollectSignaturesFlowTests.kt
@ -8,6 +8,7 @@ import net.corda.core.identity.CordaX500Name
 import net.corda.core.identity.Party
 import net.corda.core.identity.excludeHostNode
 import net.corda.core.identity.groupAbstractPartyByWellKnownParty
+import net.corda.core.node.services.IdentityService
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.transactions.TransactionBuilder
 import net.corda.core.utilities.getOrThrow
@ -137,7 +138,7 @@ class CollectSignaturesFlowTests {
    @Test
    fun `fails when not signed by initiator`() {
        val onePartyDummyContract = DummyContract.generateInitial(1337, notary, alice.ref(1))
-        val miniCorpServices = MockServices(listOf("net.corda.testing.contracts"), miniCorp, rigorousMock())
+        val miniCorpServices = MockServices(listOf("net.corda.testing.contracts"), miniCorp, rigorousMock<IdentityService>())
        val ptx = miniCorpServices.signInitialTransaction(onePartyDummyContract)
        val flow = aliceNode.services.startFlow(CollectSignaturesFlow(ptx, emptySet()))
        mockNet.runNetwork()
--- a/core/src/test/kotlin/net/corda/core/flows/ContractUpgradeFlowTest.kt
+++ b/core/src/test/kotlin/net/corda/core/flows/ContractUpgradeFlowTest.kt
@ -46,7 +46,7 @@ class ContractUpgradeFlowTest {

    @Before
    fun setup() {
-        mockNet = InternalMockNetwork(cordappPackages = listOf("net.corda.testing.contracts", "net.corda.finance.contracts.asset", "net.corda.core.flows"))
+        mockNet = InternalMockNetwork(cordappPackages = listOf("net.corda.testing.contracts", "net.corda.finance.contracts.asset", "net.corda.core.flows", "net.corda.finance.schemas"))
        aliceNode = mockNet.createPartyNode(ALICE_NAME)
        bobNode = mockNet.createPartyNode(BOB_NAME)
        notary = mockNet.defaultNotaryIdentity
--- a/docs/source/example-code/src/integration-test/kotlin/net/corda/docs/IntegrationTestingTutorial.kt
+++ b/docs/source/example-code/src/integration-test/kotlin/net/corda/docs/IntegrationTestingTutorial.kt
@ -26,7 +26,7 @@ class IntegrationTestingTutorial {
    fun `alice bob cash exchange example`() {
        // START 1
        driver(DriverParameters(startNodesInProcess = true,
-                extraCordappPackagesToScan = listOf("net.corda.finance.contracts.asset"))) {
+                extraCordappPackagesToScan = listOf("net.corda.finance.contracts.asset", "net.corda.finance.schemas"))) {
            val aliceUser = User("aliceUser", "testPassword1", permissions = setOf(
                    startFlow<CashIssueFlow>(),
                    startFlow<CashPaymentFlow>(),
--- a/finance/src/integration-test/kotlin/net/corda/finance/flows/CashSelectionTest.kt
+++ b/finance/src/integration-test/kotlin/net/corda/finance/flows/CashSelectionTest.kt
@ -12,6 +12,7 @@ import net.corda.testing.core.singleIdentity
 import net.corda.testing.driver.DriverParameters
 import net.corda.testing.driver.driver
 import net.corda.testing.driver.internal.InProcessImpl
+import net.corda.testing.driver.internal.OutOfProcessImpl
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.Test

--- a/finance/src/test/kotlin/net/corda/finance/contracts/CommercialPaperTests.kt
+++ b/finance/src/test/kotlin/net/corda/finance/contracts/CommercialPaperTests.kt
@ -109,7 +109,7 @@ class CommercialPaperTestsGeneric {
    val testSerialization = SerializationEnvironmentRule()

    private val megaCorpRef = megaCorp.ref(123)
-    private val ledgerServices = MockServices(megaCorp, miniCorp)
+    private val ledgerServices = MockServices(listOf("net.corda.finance.schemas"), megaCorp, miniCorp)

    @Test
    fun `trade lifecycle test`() {
@ -245,10 +245,10 @@ class CommercialPaperTestsGeneric {
        // of the dummy cash issuer.

        val allIdentities = arrayOf(megaCorp.identity, miniCorp.identity, dummyCashIssuer.identity, dummyNotary.identity)
-        val notaryServices = MockServices(dummyNotary)
-        val issuerServices = MockServices(dummyCashIssuer, dummyNotary)
+        val notaryServices = MockServices(listOf("net.corda.finance.contracts", "net.corda.finance.contracts.asset", "net.corda.finance.schemas"), dummyNotary)
+        val issuerServices = MockServices(listOf("net.corda.finance.contracts", "net.corda.finance.contracts.asset", "net.corda.finance.schemas"), dummyCashIssuer, dummyNotary)
        val (aliceDatabase, aliceServices) = makeTestDatabaseAndMockServices(
-                listOf("net.corda.finance.contracts"),
+                listOf("net.corda.finance.contracts", "net.corda.finance.contracts.asset", "net.corda.finance.schemas"),
                makeTestIdentityService(*allIdentities),
                alice
        )
@ -257,7 +257,7 @@ class CommercialPaperTestsGeneric {
        }

        val (megaCorpDatabase, megaCorpServices) = makeTestDatabaseAndMockServices(
-                listOf("net.corda.finance.contracts"),
+                listOf("net.corda.finance.contracts", "net.corda.finance.contracts.asset", "net.corda.finance.schemas"),
                makeTestIdentityService(*allIdentities),
                megaCorp
        )
--- a/finance/src/test/kotlin/net/corda/finance/contracts/asset/CashTests.kt
+++ b/finance/src/test/kotlin/net/corda/finance/contracts/asset/CashTests.kt
@ -80,10 +80,10 @@ class CashTests {
    @Before
    fun setUp() {
        LogHelper.setLevel(NodeVaultService::class)
-        megaCorpServices = MockServices(megaCorp)
-        miniCorpServices = MockServices(miniCorp)
+        megaCorpServices = MockServices(listOf("net.corda.finance.contracts.asset", "net.corda.finance.schemas"), megaCorp)
+        miniCorpServices = MockServices(listOf("net.corda.finance.contracts.asset", "net.corda.finance.schemas"), miniCorp)
        val databaseAndServices = makeTestDatabaseAndMockServices(
-                listOf("net.corda.finance.contracts.asset"),
+                listOf("net.corda.finance.contracts.asset", "net.corda.finance.schemas"),
                makeTestIdentityService(megaCorp.identity, miniCorp.identity, dummyCashIssuer.identity, dummyNotary.identity),
                TestIdentity(CordaX500Name("Me", "London", "GB"))
        )
--- a/finance/src/test/kotlin/net/corda/finance/contracts/asset/ObligationTests.kt
+++ b/finance/src/test/kotlin/net/corda/finance/contracts/asset/ObligationTests.kt
@ -9,6 +9,7 @@ import net.corda.core.crypto.sha256
 import net.corda.core.identity.AbstractParty
 import net.corda.core.identity.AnonymousParty
 import net.corda.core.identity.CordaX500Name
+import net.corda.core.node.services.IdentityService
 import net.corda.core.transactions.TransactionBuilder
 import net.corda.core.utilities.NonEmptySet
 import net.corda.core.utilities.OpaqueBytes
@ -79,7 +80,7 @@ class ObligationTests {
            beneficiary = CHARLIE
    )
    private val outState = inState.copy(beneficiary = AnonymousParty(BOB_PUBKEY))
-    private val miniCorpServices = MockServices(listOf("net.corda.finance.contracts.asset"), miniCorp, rigorousMock())
+    private val miniCorpServices = MockServices(listOf("net.corda.finance.contracts.asset"), miniCorp, rigorousMock<IdentityService>())
    private val notaryServices = MockServices(emptyList(), MEGA_CORP.name, rigorousMock(), dummyNotary.keyPair)
    private val identityService = rigorousMock<IdentityServiceInternal>().also {
        doReturn(null).whenever(it).partyFromKey(ALICE_PUBKEY)
--- a/finance/src/test/kotlin/net/corda/finance/flows/CashExitFlowTests.kt
+++ b/finance/src/test/kotlin/net/corda/finance/flows/CashExitFlowTests.kt
@ -27,7 +27,7 @@ class CashExitFlowTests {
    @Before
    fun start() {
        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin(),
-                cordappPackages = listOf("net.corda.finance.contracts.asset"))
+                cordappPackages = listOf("net.corda.finance.contracts.asset", "net.corda.finance.schemas"))
        bankOfCordaNode = mockNet.createPartyNode(BOC_NAME)
        bankOfCorda = bankOfCordaNode.info.identityFromX500Name(BOC_NAME)
        notary = mockNet.defaultNotaryIdentity
--- a/finance/src/test/kotlin/net/corda/finance/flows/CashIssueFlowTests.kt
+++ b/finance/src/test/kotlin/net/corda/finance/flows/CashIssueFlowTests.kt
@ -24,7 +24,7 @@ class CashIssueFlowTests {

    @Before
    fun start() {
-        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin(), cordappPackages = listOf("net.corda.finance.contracts.asset"))
+        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin(), cordappPackages = listOf("net.corda.finance.contracts", "net.corda.finance.contracts.asset", "net.corda.finance.schemas"))
        bankOfCordaNode = mockNet.createPartyNode(BOC_NAME)
        bankOfCorda =  bankOfCordaNode.info.identityFromX500Name(BOC_NAME)
        notary = mockNet.defaultNotaryIdentity
--- a/finance/src/test/kotlin/net/corda/finance/flows/CashPaymentFlowTests.kt
+++ b/finance/src/test/kotlin/net/corda/finance/flows/CashPaymentFlowTests.kt
@ -30,7 +30,7 @@ class CashPaymentFlowTests {

    @Before
    fun start() {
-        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin(), cordappPackages = listOf("net.corda.finance.contracts.asset"))
+        mockNet = MockNetwork(servicePeerAllocationStrategy = RoundRobin(), cordappPackages = listOf("net.corda.finance.contracts.asset", "net.corda.finance.schemas"))
        bankOfCordaNode = mockNet.createPartyNode(BOC_NAME)
        bankOfCorda = bankOfCordaNode.info.identityFromX500Name(BOC_NAME)
        aliceNode = mockNet.createPartyNode(ALICE_NAME)
--- a/node/src/integration-test/kotlin/net/corda/node/services/DistributedServiceTests.kt
+++ b/node/src/integration-test/kotlin/net/corda/node/services/DistributedServiceTests.kt
@ -42,7 +42,7 @@ class DistributedServiceTests {
                invokeRpc(CordaRPCOps::stateMachinesFeed))
        )
        driver(DriverParameters(
-                extraCordappPackagesToScan = listOf("net.corda.finance.contracts"),
+                extraCordappPackagesToScan = listOf("net.corda.finance.contracts", "net.corda.finance.schemas"),
                notarySpecs = listOf(
                        NotarySpec(
                                DUMMY_NOTARY_NAME,
--- a/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
+++ b/node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
@ -48,7 +48,6 @@ import net.corda.node.services.keys.PersistentKeyManagementService
 import net.corda.node.services.messaging.MessagingService
 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.*
 import net.corda.node.services.transactions.*
@ -637,7 +636,6 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
    private fun makeVaultObservers(schedulerService: SchedulerService, hibernateConfig: HibernateConfiguration, smm: StateMachineManager, schemaService: SchemaService, flowLogicRefFactory: FlowLogicRefFactory) {
        VaultSoftLockManager.install(services.vaultService, smm)
        ScheduledActivityObserver.install(services.vaultService, schedulerService, flowLogicRefFactory)
-        HibernateObserver.install(services.vaultService.rawUpdates, hibernateConfig, schemaService)
    }

    @VisibleForTesting
@ -813,8 +811,8 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
    }

    protected open fun generateKeyPair() = cryptoGenerateKeyPair()
-    protected open fun makeVaultService(keyManagementService: KeyManagementService, services: ServicesForResolution, hibernateConfig: HibernateConfiguration): VaultServiceInternal {
-        return NodeVaultService(platformClock, keyManagementService, services, hibernateConfig)
+    protected open fun makeVaultService(keyManagementService: KeyManagementService, services: ServicesForResolution, hibernateConfig: HibernateConfiguration, schemaService: SchemaService): VaultServiceInternal {
+        return NodeVaultService(platformClock, keyManagementService, services, hibernateConfig, schemaService)
    }

    private inner class ServiceHubInternalImpl(
@ -838,7 +836,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
        override val stateMachineRecordedTransactionMapping = DBTransactionMappingStorage()
        override val auditService = DummyAuditService()
        override val transactionVerifierService by lazy { makeTransactionVerifierService() }
-        override val vaultService by lazy { makeVaultService(keyManagementService, servicesForResolution, database.hibernateConfig) }
+        override val vaultService by lazy { makeVaultService(keyManagementService, servicesForResolution, database.hibernateConfig, schemaService) }
        override val contractUpgradeService by lazy { ContractUpgradeServiceImpl() }
        override val attachments: AttachmentStorage get() = this@AbstractNode.attachments
        override val networkService: MessagingService get() = network
--- a/node/src/main/kotlin/net/corda/node/services/schema/PersistentStateService.kt
+++ b/node/src/main/kotlin/net/corda/node/services/schema/PersistentStateService.kt
@ -4,16 +4,12 @@ import net.corda.core.contracts.ContractState
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.StateRef
 import net.corda.core.internal.VisibleForTesting
-import net.corda.core.node.services.Vault
 import net.corda.core.schemas.MappedSchema
 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.HibernateConfiguration
-import net.corda.nodeapi.internal.persistence.contextTransaction
-import org.hibernate.FlushMode
-import rx.Observable
+import net.corda.nodeapi.internal.persistence.currentDBSession


 /**
@ -26,17 +22,12 @@ data class ContractStateAndRef(val state: ContractState, val ref: StateRef)
 * A vault observer that extracts Object Relational Mappings for contract states that support it, and persists them with Hibernate.
 */
 // TODO: Manage version evolution of the schemas via additional tooling.
-class HibernateObserver private constructor(private val config: HibernateConfiguration, private val schemaService: SchemaService) {
+class PersistentStateService(private val schemaService: SchemaService) {
    companion object {
        private val log = contextLogger()
-        fun install(vaultUpdates: Observable<Vault.Update<ContractState>>, config: HibernateConfiguration, schemaService: SchemaService): HibernateObserver {
-            val observer = HibernateObserver(config, schemaService)
-            vaultUpdates.subscribe { observer.persist(it.produced) }
-            return observer
-        }
    }

-    private fun persist(produced: Set<StateAndRef<ContractState>>) {
+    fun persist(produced: Set<StateAndRef<ContractState>>) {
        val stateBySchema: MutableMap<MappedSchema, MutableList<ContractStateAndRef>> = mutableMapOf()
        // map all states by their referenced schemas
        produced.forEach {
@ -52,18 +43,10 @@ class HibernateObserver private constructor(private val config: HibernateConfigu

    @VisibleForTesting
    internal fun persistStatesWithSchema(statesAndRefs: List<ContractStateAndRef>, schema: MappedSchema) {
-        val sessionFactory = config.sessionFactoryForSchemas(setOf(schema))
-        val session = sessionFactory.withOptions().
-                connection(contextTransaction.connection).
-                flushMode(FlushMode.MANUAL).
-                openSession()
-        session.use { thisSession ->
-            statesAndRefs.forEach {
-                val mappedObject = schemaService.generateMappedObject(it.state, schema)
-                mappedObject.stateRef = PersistentStateRef(it.ref)
-                thisSession.persist(mappedObject)
-            }
-            thisSession.flush()
+        statesAndRefs.forEach {
+            val mappedObject = schemaService.generateMappedObject(it.state, schema)
+            mappedObject.stateRef = PersistentStateRef(it.ref)
+            currentDBSession().persist(mappedObject)
        }
    }
 }
--- a/node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt
+++ b/node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt
@ -14,7 +14,9 @@ import net.corda.core.schemas.PersistentStateRef
 import net.corda.core.serialization.SingletonSerializeAsToken
 import net.corda.core.transactions.*
 import net.corda.core.utilities.*
+import net.corda.node.services.api.SchemaService
 import net.corda.node.services.api.VaultServiceInternal
+import net.corda.node.services.schema.PersistentStateService
 import net.corda.node.services.statemachine.FlowStateMachineImpl
 import net.corda.nodeapi.internal.persistence.*
 import org.hibernate.Session
@ -48,7 +50,8 @@ class NodeVaultService(
        private val clock: Clock,
        private val keyManagementService: KeyManagementService,
        private val servicesForResolution: ServicesForResolution,
-        hibernateConfig: HibernateConfiguration
+        private val hibernateConfig: HibernateConfiguration,
+        private val schemaService: SchemaService
 ) : SingletonSerializeAsToken(), VaultServiceInternal {
    private companion object {
        private val log = contextLogger()
@ -64,6 +67,7 @@ class NodeVaultService(
    }

    private val mutex = ThreadBox(InnerState())
+    private val persistentStateService = PersistentStateService(schemaService)

    private fun recordUpdate(update: Vault.Update<ContractState>): Vault.Update<ContractState> {
        if (!update.isEmpty()) {
@ -200,6 +204,7 @@ class NodeVaultService(
                // flowId required by SoftLockManager to perform auto-registration of soft locks for new states
                val uuid = (Strand.currentStrand() as? FlowStateMachineImpl<*>)?.id?.uuid
                val vaultUpdate = if (uuid != null) netUpdate.copy(flowId = uuid) else netUpdate
+                persistentStateService.persist(vaultUpdate.produced)
                updatesPublisher.onNext(vaultUpdate)
            }
        }
--- a/node/src/test/kotlin/net/corda/node/CordaRPCOpsImplTest.kt
+++ b/node/src/test/kotlin/net/corda/node/CordaRPCOpsImplTest.kt
@ -80,7 +80,7 @@ class CordaRPCOpsImplTest {

    @Before
    fun setup() {
-        mockNet = InternalMockNetwork(cordappPackages = listOf("net.corda.finance.contracts.asset"))
+        mockNet = InternalMockNetwork(cordappPackages = listOf("net.corda.finance.contracts.asset", "net.corda.finance.schemas"))
        aliceNode = mockNet.createNode(InternalMockNodeParameters(legalName = ALICE_NAME))
        rpc = SecureCordaRPCOps(aliceNode.services, aliceNode.smm, aliceNode.database, aliceNode.services)
        CURRENT_RPC_CONTEXT.set(RpcAuthContext(InvocationContext.rpc(testActor()), buildSubject("TEST_USER", emptySet())))
--- a/node/src/test/kotlin/net/corda/node/messaging/TwoPartyTradeFlowTests.kt
+++ b/node/src/test/kotlin/net/corda/node/messaging/TwoPartyTradeFlowTests.kt
@ -78,7 +78,7 @@ import kotlin.test.assertTrue
@RunWith(Parameterized::class)
 class TwoPartyTradeFlowTests(private val anonymous: Boolean) {
    companion object {
-        private val cordappPackages = listOf("net.corda.finance.contracts")
+        private val cordappPackages = listOf("net.corda.finance.contracts", "net.corda.finance.schemas")
        @JvmStatic
        @Parameterized.Parameters(name = "Anonymous = {0}")
        fun data(): Collection<Boolean> = listOf(true, false)
--- a/node/src/test/kotlin/net/corda/node/services/persistence/HibernateConfigurationTest.kt
+++ b/node/src/test/kotlin/net/corda/node/services/persistence/HibernateConfigurationTest.kt
@ -25,12 +25,13 @@ import net.corda.finance.SWISS_FRANCS
 import net.corda.finance.contracts.asset.Cash
 import net.corda.finance.contracts.asset.DummyFungibleContract
 import net.corda.finance.schemas.CashSchemaV1
+import net.corda.finance.schemas.SampleCashSchemaV1
 import net.corda.finance.schemas.SampleCashSchemaV2
 import net.corda.finance.schemas.SampleCashSchemaV3
 import net.corda.finance.utils.sumCash
 import net.corda.node.internal.configureDatabase
 import net.corda.node.services.schema.ContractStateAndRef
-import net.corda.node.services.schema.HibernateObserver
+import net.corda.node.services.schema.PersistentStateService
 import net.corda.node.services.schema.NodeSchemaService
 import net.corda.node.services.vault.VaultSchemaV1
 import net.corda.node.services.api.IdentityServiceInternal
@ -41,6 +42,7 @@ import net.corda.nodeapi.internal.persistence.DatabaseConfig
 import net.corda.nodeapi.internal.persistence.HibernateConfiguration
 import net.corda.testing.core.*
 import net.corda.testing.internal.rigorousMock
+import net.corda.testing.internal.vault.DummyDealStateSchemaV1
 import net.corda.testing.internal.vault.VaultFiller
 import net.corda.testing.node.MockServices
 import net.corda.testing.node.MockServices.Companion.makeTestDataSourceProperties
@ -82,7 +84,7 @@ class HibernateConfigurationTest {

    // Hibernate configuration objects
    lateinit var hibernateConfig: HibernateConfiguration
-    lateinit var hibernatePersister: HibernateObserver
+    lateinit var hibernatePersister: PersistentStateService
    lateinit var sessionFactory: SessionFactory
    lateinit var entityManager: EntityManager
    lateinit var criteriaBuilder: CriteriaBuilder
@ -97,10 +99,10 @@ class HibernateConfigurationTest {

    @Before
    fun setUp() {
-        val cordappPackages = listOf("net.corda.testing.internal.vault", "net.corda.finance.contracts.asset")
+        val cordappPackages = listOf("net.corda.testing.internal.vault", "net.corda.finance.contracts.asset", "net.corda.finance.schemas")
        bankServices = MockServices(cordappPackages, BOC.name, rigorousMock(), BOC_KEY)
-        issuerServices = MockServices(cordappPackages, dummyCashIssuer, rigorousMock())
-        notaryServices = MockServices(cordappPackages, dummyNotary, rigorousMock())
+        issuerServices = MockServices(cordappPackages, dummyCashIssuer, rigorousMock<IdentityService>())
+        notaryServices = MockServices(cordappPackages, dummyNotary, rigorousMock<IdentityService>())
        notary = notaryServices.myInfo.singleIdentity()
        val dataSourceProps = makeTestDataSourceProperties()
        val identityService = rigorousMock<IdentityService>().also { mock ->
@ -110,7 +112,7 @@ class HibernateConfigurationTest {
                doReturn(it.party).whenever(mock).wellKnownPartyFromX500Name(it.name)
            }
        }
-        val schemaService = NodeSchemaService()
+        val schemaService = NodeSchemaService(extraSchemas = setOf(CashSchemaV1, SampleCashSchemaV1, SampleCashSchemaV2, SampleCashSchemaV3, DummyLinearStateSchemaV1, DummyLinearStateSchemaV2, DummyDealStateSchemaV1))
        database = configureDatabase(dataSourceProps, DatabaseConfig(), identityService, schemaService)
        database.transaction {
            hibernateConfig = database.hibernateConfig
@ -119,7 +121,7 @@ class HibernateConfigurationTest {
            services = object : MockServices(cordappPackages, BOB_NAME, rigorousMock<IdentityServiceInternal>().also {
                doNothing().whenever(it).justVerifyAndRegisterIdentity(argThat { name == BOB_NAME })
            }, generateKeyPair(), dummyNotary.keyPair) {
-                override val vaultService = NodeVaultService(Clock.systemUTC(), keyManagementService, servicesForResolution, hibernateConfig)
+                override val vaultService = NodeVaultService(Clock.systemUTC(), keyManagementService, servicesForResolution, hibernateConfig, schemaService)
                override fun recordTransactions(statesToRecord: StatesToRecord, txs: Iterable<SignedTransaction>) {
                    for (stx in txs) {
                        (validatedTransactions as WritableTransactionStorage).addTransaction(stx)
@ -131,7 +133,7 @@ class HibernateConfigurationTest {
                override fun jdbcSession() = database.createSession()
            }
            vaultFiller = VaultFiller(services, dummyNotary, notary, ::Random)
-            hibernatePersister = HibernateObserver.install(services.vaultService.rawUpdates, hibernateConfig, schemaService)
+            hibernatePersister = PersistentStateService(schemaService)
        }

        identity = services.myInfo.singleIdentity()
--- a/node/src/test/kotlin/net/corda/node/services/schema/PersistentStateServiceTests.kt
+++ b/node/src/test/kotlin/net/corda/node/services/schema/PersistentStateServiceTests.kt
@ -21,19 +21,21 @@ import net.corda.testing.node.MockServices.Companion.makeTestDataSourcePropertie
 import net.corda.testing.internal.rigorousMock
 import org.junit.After
 import org.junit.Before
+import org.junit.Ignore
 import org.junit.Test
 import rx.subjects.PublishSubject
 import kotlin.test.assertEquals

-class HibernateObserverTests {
+@Ignore("needs reworking")
+class PersistentStateServiceTests {
    @Before
    fun setUp() {
-        LogHelper.setLevel(HibernateObserver::class)
+        LogHelper.setLevel(PersistentStateService::class)
    }

    @After
    fun cleanUp() {
-        LogHelper.reset(HibernateObserver::class)
+        LogHelper.reset(PersistentStateService::class)
    }

    class TestState : QueryableState {
@ -54,7 +56,7 @@ class HibernateObserverTests {
        val testSchema = TestSchema
        val rawUpdatesPublisher = PublishSubject.create<Vault.Update<ContractState>>()
        val schemaService = object : SchemaService {
-            override val schemaOptions: Map<MappedSchema, SchemaService.SchemaOptions> = emptyMap()
+            override val schemaOptions: Map<MappedSchema, SchemaService.SchemaOptions> = mapOf(testSchema to SchemaService.SchemaOptions())

            override fun selectSchemas(state: ContractState): Iterable<MappedSchema> = setOf(testSchema)

@ -66,7 +68,6 @@ class HibernateObserverTests {
            }
        }
        val database = configureDatabase(makeTestDataSourceProperties(), DatabaseConfig(), rigorousMock(), schemaService)
-        HibernateObserver.install(rawUpdatesPublisher, database.hibernateConfig, schemaService)
        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)))))
--- a/node/src/test/kotlin/net/corda/node/services/vault/NodeVaultServiceTest.kt
+++ b/node/src/test/kotlin/net/corda/node/services/vault/NodeVaultServiceTest.kt
@ -14,10 +14,7 @@ import net.corda.core.identity.*
 import net.corda.core.internal.NotaryChangeTransactionBuilder
 import net.corda.core.internal.packageName
 import net.corda.core.node.StatesToRecord
-import net.corda.core.node.services.StatesNotAvailableException
-import net.corda.core.node.services.Vault
-import net.corda.core.node.services.VaultService
-import net.corda.core.node.services.queryBy
+import net.corda.core.node.services.*
 import net.corda.core.node.services.vault.PageSpecification
 import net.corda.core.node.services.vault.QueryCriteria
 import net.corda.core.node.services.vault.QueryCriteria.*
@ -100,8 +97,8 @@ class NodeVaultServiceTest {
        vaultFiller = VaultFiller(services, dummyNotary)
        // This is safe because MockServices only ever have a single identity
        identity = services.myInfo.singleIdentityAndCert()
-        issuerServices = MockServices(cordappPackages, dummyCashIssuer, rigorousMock())
-        bocServices = MockServices(cordappPackages, bankOfCorda, rigorousMock())
+        issuerServices = MockServices(cordappPackages, dummyCashIssuer, rigorousMock<IdentityService>())
+        bocServices = MockServices(cordappPackages, bankOfCorda, rigorousMock<IdentityService>())
        services.identityService.verifyAndRegisterIdentity(DUMMY_CASH_ISSUER_IDENTITY)
        services.identityService.verifyAndRegisterIdentity(BOC_IDENTITY)
    }
--- a/node/src/test/kotlin/net/corda/node/services/vault/VaultQueryTests.kt
+++ b/node/src/test/kotlin/net/corda/node/services/vault/VaultQueryTests.kt
@ -1596,9 +1596,6 @@ class VaultQueryTests {
        cordappPackages -= SampleCashSchemaV3::class.packageName
        setUp()
        database.transaction {
-            vaultFiller.fillWithSomeTestCash(100.DOLLARS, notaryServices, 1, DUMMY_CASH_ISSUER)
-            vaultFiller.fillWithSomeTestCash(100.POUNDS, notaryServices, 1, DUMMY_CASH_ISSUER)
-            vaultFiller.fillWithSomeTestCash(100.SWISS_FRANCS, notaryServices, 1, DUMMY_CASH_ISSUER)
            // CashSchemaV3 NOT registered with NodeSchemaService
            val logicalExpression = builder { SampleCashSchemaV3.PersistentCashState::currency.equal(GBP.currencyCode) }
            val criteria = VaultCustomQueryCriteria(logicalExpression)
--- a/node/src/test/kotlin/net/corda/node/services/vault/VaultSoftLockManagerTest.kt
+++ b/node/src/test/kotlin/net/corda/node/services/vault/VaultSoftLockManagerTest.kt
@ -24,6 +24,7 @@ import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.getOrThrow
 import net.corda.core.utilities.unwrap
 import net.corda.node.internal.InitiatedFlowFactory
+import net.corda.node.services.api.SchemaService
 import net.corda.node.services.api.VaultServiceInternal
 import net.corda.nodeapi.internal.persistence.HibernateConfiguration
 import net.corda.testing.core.singleIdentity
@ -83,8 +84,8 @@ class VaultSoftLockManagerTest {
    }
    private val mockNet = InternalMockNetwork(cordappPackages = listOf(ContractImpl::class.packageName), defaultFactory = { args ->
        object : InternalMockNetwork.MockNode(args) {
-            override fun makeVaultService(keyManagementService: KeyManagementService, services: ServicesForResolution, hibernateConfig: HibernateConfiguration): VaultServiceInternal {
-                val realVault = super.makeVaultService(keyManagementService, services, hibernateConfig)
+            override fun makeVaultService(keyManagementService: KeyManagementService, services: ServicesForResolution, hibernateConfig: HibernateConfiguration, schemaService: SchemaService): VaultServiceInternal {
+                val realVault = super.makeVaultService(keyManagementService, services, hibernateConfig, schemaService)
                return object : VaultServiceInternal by realVault {
                    override fun softLockRelease(lockId: UUID, stateRefs: NonEmptySet<StateRef>?) {
                        mockVault.softLockRelease(lockId, stateRefs) // No need to also call the real one for these tests.
--- a/node/src/test/kotlin/net/corda/node/services/vault/VaultWithCashTest.kt
+++ b/node/src/test/kotlin/net/corda/node/services/vault/VaultWithCashTest.kt
@ -11,6 +11,7 @@ import net.corda.core.identity.Party
 import net.corda.core.internal.concurrent.fork
 import net.corda.core.internal.concurrent.transpose
 import net.corda.core.internal.packageName
+import net.corda.core.node.services.IdentityService
 import net.corda.core.node.services.Vault
 import net.corda.core.node.services.VaultService
 import net.corda.core.node.services.queryBy
@ -44,7 +45,7 @@ import kotlin.test.fail

 class VaultWithCashTest {
    private companion object {
-        val cordappPackages = listOf("net.corda.testing.internal.vault", "net.corda.finance.contracts.asset", CashSchemaV1::class.packageName)
+        val cordappPackages = listOf("net.corda.testing.internal.vault", "net.corda.finance.contracts.asset", CashSchemaV1::class.packageName, "net.corda.core.contracts")
        val BOB = TestIdentity(BOB_NAME, 80).party
        val dummyCashIssuer = TestIdentity(CordaX500Name("Snake Oil Issuer", "London", "GB"), 10)
        val DUMMY_CASH_ISSUER = dummyCashIssuer.ref(1)
@ -82,7 +83,7 @@ class VaultWithCashTest {
        services = databaseAndServices.second
        vaultFiller = VaultFiller(services, dummyNotary)
        issuerServices = MockServices(cordappPackages, dummyCashIssuer, rigorousMock(), MEGA_CORP_KEY)
-        notaryServices = MockServices(cordappPackages, dummyNotary, rigorousMock())
+        notaryServices = MockServices(cordappPackages, dummyNotary, rigorousMock<IdentityService>())
        notary = notaryServices.myInfo.legalIdentitiesAndCerts.single().party
    }

--- a/samples/trader-demo/src/test/kotlin/net/corda/traderdemo/TransactionGraphSearchTests.kt
+++ b/samples/trader-demo/src/test/kotlin/net/corda/traderdemo/TransactionGraphSearchTests.kt
@ -3,6 +3,7 @@ package net.corda.traderdemo
 import net.corda.core.contracts.CommandData
 import net.corda.core.crypto.newSecureRandom
 import net.corda.core.identity.CordaX500Name
+import net.corda.core.node.services.IdentityService
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.transactions.TransactionBuilder
 import net.corda.core.transactions.WireTransaction
@ -46,8 +47,8 @@ class TransactionGraphSearchTests {
     * @param signer signer for the two transactions and their commands.
     */
    fun buildTransactions(command: CommandData): GraphTransactionStorage {
-        val megaCorpServices = MockServices(listOf("net.corda.testing.contracts"), megaCorp, rigorousMock())
-        val notaryServices = MockServices(listOf("net.corda.testing.contracts"), dummyNotary, rigorousMock())
+        val megaCorpServices = MockServices(listOf("net.corda.testing.contracts"), megaCorp, rigorousMock<IdentityService>())
+        val notaryServices = MockServices(listOf("net.corda.testing.contracts"), dummyNotary, rigorousMock<IdentityService>())
        val originBuilder = TransactionBuilder(dummyNotary.party)
                .addOutputState(DummyState(random31BitValue()), DummyContract.PROGRAM_ID)
                .addCommand(command, megaCorp.publicKey)
--- a/testing/node-driver/src/main/kotlin/net/corda/testing/node/MockServices.kt
+++ b/testing/node-driver/src/main/kotlin/net/corda/testing/node/MockServices.kt
@ -2,8 +2,6 @@ package net.corda.testing.node

 import com.google.common.collect.MutableClassToInstanceMap
 import net.corda.core.contracts.ContractClassName
-import net.corda.core.contracts.ContractState
-import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.StateRef
 import net.corda.core.cordapp.CordappProvider
 import net.corda.core.crypto.*
@ -17,7 +15,6 @@ import net.corda.core.node.services.*
 import net.corda.core.serialization.SerializeAsToken
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.NetworkHostAndPort
-import net.corda.node.VersionInfo
 import net.corda.node.internal.ServicesForResolutionImpl
 import net.corda.node.internal.configureDatabase
 import net.corda.node.internal.cordapp.CordappLoader
@ -25,7 +22,6 @@ import net.corda.node.services.api.SchemaService
 import net.corda.node.services.api.VaultServiceInternal
 import net.corda.node.services.api.WritableTransactionStorage
 import net.corda.node.services.identity.InMemoryIdentityService
-import net.corda.node.services.schema.HibernateObserver
 import net.corda.node.services.schema.NodeSchemaService
 import net.corda.node.services.transactions.InMemoryTransactionVerifierService
 import net.corda.node.services.vault.NodeVaultService
@ -196,6 +192,12 @@ open class MockServices private constructor(
    constructor(firstIdentity: TestIdentity, vararg moreIdentities: TestIdentity) : this(
            listOf(getCallerPackage()),
            firstIdentity,
+            *moreIdentities
+    )
+
+    constructor(cordappPackages: List<String>, firstIdentity: TestIdentity, vararg moreIdentities: TestIdentity) : this(
+            cordappPackages,
+            firstIdentity,
            makeTestIdentityService(*listOf(firstIdentity, *moreIdentities).map { it.identity }.toTypedArray()),
            firstIdentity.keyPair
    )
@ -229,9 +231,7 @@ open class MockServices private constructor(
    protected val servicesForResolution: ServicesForResolution get() = ServicesForResolutionImpl(identityService, attachments, cordappProvider, networkParameters, validatedTransactions)

    internal fun makeVaultService(hibernateConfig: HibernateConfiguration, schemaService: SchemaService): VaultServiceInternal {
-        val vaultService = NodeVaultService(Clock.systemUTC(), keyManagementService, servicesForResolution, hibernateConfig)
-        HibernateObserver.install(vaultService.rawUpdates, hibernateConfig, schemaService)
-        return vaultService
+        return NodeVaultService(Clock.systemUTC(), keyManagementService, servicesForResolution, hibernateConfig, schemaService)
    }

    // This needs to be internal as MutableClassToInstanceMap is a guava type and shouldn't be part of our public API