mirror of
https://github.com/corda/corda.git
synced 2025-06-17 06:38:21 +00:00
NetworkMapCache database backed (#1135)
Work on database backed NetworkMapCache Make NodeInfo JPA entity. Enable node startup with it's database network map cache. Fix schema. Make node not wait for finishing network map service registration if it successfully loaded data from database. Add tests for startup without NetworkMapService. * Rename networkMapRegistrationFuture Change networkMapRegistrationFuture to nodeReadyFuture, it no longer indicates the NetworkMapService registration, because we are able to run network without map service configured. * Partially integrate database into NetworkMapCache Full integrtion will come with service removal. Move MockServiceHubInternal to net.corda.node.testing * Add workaround to transaction scope race Temporary workaround to force isolated transaction (otherwise it causes race conditions when processing network map registration on network map node). * Remove WorldMapLocation from NodeInfo Infer the node's location based on X500 name Add serial number on NodeInfo For tests of running without NetworkMap, start nodes with nonexistent NetworkMap address Make clearNetworkMapCache callable via RPC.
This commit is contained in:
Katarzyna Streich
committed by
GitHub
GitHub
parent
3e5fa9ee6a
commit
472ecc65c6
@ -43,6 +43,7 @@ class P2PMessagingTest : NodeBasedTest() {
|
||||
val startUpDuration = elapsedTime { startNodes().getOrThrow() }
|
||||
// Start the network map a second time - this will restore message queues from the journal.
|
||||
// This will hang and fail prior the fix. https://github.com/corda/corda/issues/37
|
||||
clearAllNodeInfoDb() // Clear network map data from nodes databases.
|
||||
stopAllNodes()
|
||||
startNodes().getOrThrow(timeout = startUpDuration * 3)
|
||||
}
|
||||
@ -149,7 +150,6 @@ class P2PMessagingTest : NodeBasedTest() {
|
||||
// Restart the node and expect a response
|
||||
val aliceRestarted = startNode(ALICE.name, configOverrides = mapOf("messageRedeliveryDelaySeconds" to 1)).getOrThrow()
|
||||
val response = aliceRestarted.network.onNext<Any>(dummyTopic, sessionId).getOrThrow(5.seconds)
|
||||
|
||||
assertThat(requestsReceived.get()).isGreaterThanOrEqualTo(2)
|
||||
assertThat(response).isEqualTo(responseMessage)
|
||||
}
|
||||
|
||||
@ -67,7 +67,7 @@ class P2PSecurityTest : NodeBasedTest() {
|
||||
|
||||
private fun SimpleNode.registerWithNetworkMap(registrationName: X500Name): CordaFuture<NetworkMapService.RegistrationResponse> {
|
||||
val legalIdentity = getTestPartyAndCertificate(registrationName, identity.public)
|
||||
val nodeInfo = NodeInfo(listOf(MOCK_HOST_AND_PORT), legalIdentity, NonEmptySet.of(legalIdentity), 1)
|
||||
val nodeInfo = NodeInfo(listOf(MOCK_HOST_AND_PORT), legalIdentity, NonEmptySet.of(legalIdentity), 1, serial = 1)
|
||||
val registration = NodeRegistration(nodeInfo, System.currentTimeMillis(), AddOrRemove.ADD, Instant.MAX)
|
||||
val request = RegistrationRequest(registration.toWire(keyService, identity.public), network.myAddress)
|
||||
return network.sendRequest<NetworkMapService.RegistrationResponse>(NetworkMapService.REGISTER_TOPIC, request, networkMapNode.network.myAddress)
|
||||
|
||||
@ -12,6 +12,7 @@ import net.corda.core.flows.*
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.identity.PartyAndCertificate
|
||||
import net.corda.core.internal.*
|
||||
import net.corda.core.internal.concurrent.doneFuture
|
||||
import net.corda.core.internal.concurrent.flatMap
|
||||
import net.corda.core.internal.concurrent.openFuture
|
||||
import net.corda.core.messaging.CordaRPCOps
|
||||
@ -39,7 +40,7 @@ import net.corda.node.services.identity.InMemoryIdentityService
|
||||
import net.corda.node.services.keys.PersistentKeyManagementService
|
||||
import net.corda.node.services.messaging.MessagingService
|
||||
import net.corda.node.services.messaging.sendRequest
|
||||
import net.corda.node.services.network.InMemoryNetworkMapCache
|
||||
import net.corda.node.services.network.PersistentNetworkMapCache
|
||||
import net.corda.node.services.network.NetworkMapService
|
||||
import net.corda.node.services.network.NetworkMapService.RegistrationRequest
|
||||
import net.corda.node.services.network.NetworkMapService.RegistrationResponse
|
||||
@ -138,10 +139,11 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
var isPreviousCheckpointsPresent = false
|
||||
private set
|
||||
|
||||
protected val _networkMapRegistrationFuture = openFuture<Unit>()
|
||||
/** Completes once the node has successfully registered with the network map service */
|
||||
val networkMapRegistrationFuture: CordaFuture<Unit>
|
||||
get() = _networkMapRegistrationFuture
|
||||
protected val _nodeReadyFuture = openFuture<Unit>()
|
||||
/** Completes once the node has successfully registered with the network map service
|
||||
* or has loaded network map data from local database */
|
||||
val nodeReadyFuture: CordaFuture<Unit>
|
||||
get() = _nodeReadyFuture
|
||||
|
||||
/** Fetch CordaPluginRegistry classes registered in META-INF/services/net.corda.core.node.CordaPluginRegistry files that exist in the classpath */
|
||||
open val pluginRegistries: List<CordaPluginRegistry> by lazy {
|
||||
@ -155,10 +157,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
/** The implementation of the [CordaRPCOps] interface used by this node. */
|
||||
open val rpcOps: CordaRPCOps by lazy { CordaRPCOpsImpl(services, smm, database) } // Lazy to avoid init ordering issue with the SMM.
|
||||
|
||||
open fun findMyLocation(): WorldMapLocation? {
|
||||
return configuration.myLegalName.locationOrNull?.let { CityDatabase[it] }
|
||||
}
|
||||
|
||||
open fun start() {
|
||||
require(!started) { "Node has already been started" }
|
||||
|
||||
@ -208,7 +206,10 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
}
|
||||
|
||||
runOnStop += network::stop
|
||||
_networkMapRegistrationFuture.captureLater(registerWithNetworkMapIfConfigured())
|
||||
}
|
||||
// If we successfully loaded network data from database, we set this future to Unit.
|
||||
_nodeReadyFuture.captureLater(registerWithNetworkMapIfConfigured())
|
||||
database.transaction {
|
||||
smm.start()
|
||||
// Shut down the SMM so no Fibers are scheduled.
|
||||
runOnStop += { smm.stop(acceptableLiveFiberCountOnStop()) }
|
||||
@ -483,9 +484,9 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
|
||||
private fun makeInfo(legalIdentity: PartyAndCertificate): NodeInfo {
|
||||
val advertisedServiceEntries = makeServiceEntries()
|
||||
val allIdentities = (advertisedServiceEntries.map { it.identity } + legalIdentity).toNonEmptySet()
|
||||
val allIdentities = advertisedServiceEntries.map { it.identity }.toSet() // TODO Add node's legalIdentity (after services removal).
|
||||
val addresses = myAddresses() // TODO There is no support for multiple IP addresses yet.
|
||||
return NodeInfo(addresses, legalIdentity, allIdentities, platformVersion, advertisedServiceEntries, findMyLocation())
|
||||
return NodeInfo(addresses, legalIdentity, allIdentities, platformVersion, advertisedServiceEntries, platformClock.instant().toEpochMilli())
|
||||
}
|
||||
|
||||
/**
|
||||
@ -580,7 +581,15 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
services.networkMapCache.runWithoutMapService()
|
||||
noNetworkMapConfigured() // TODO This method isn't needed as runWithoutMapService sets the Future in the cache
|
||||
} else {
|
||||
registerWithNetworkMap()
|
||||
val netMapRegistration = registerWithNetworkMap()
|
||||
// We may want to start node immediately with database data and not wait for network map registration (but send it either way).
|
||||
// So we are ready to go.
|
||||
if (services.networkMapCache.loadDBSuccess) {
|
||||
log.info("Node successfully loaded network map data from the database.")
|
||||
doneFuture(Unit)
|
||||
} else {
|
||||
netMapRegistration
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -606,7 +615,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
// Register this node against the network
|
||||
val instant = platformClock.instant()
|
||||
val expires = instant + NetworkMapService.DEFAULT_EXPIRATION_PERIOD
|
||||
val reg = NodeRegistration(info, instant.toEpochMilli(), ADD, expires)
|
||||
val reg = NodeRegistration(info, info.serial, ADD, expires)
|
||||
val request = RegistrationRequest(reg.toWire(services.keyManagementService, info.legalIdentityAndCert.owningKey), network.myAddress)
|
||||
return network.sendRequest(NetworkMapService.REGISTER_TOPIC, request, networkMapAddress)
|
||||
}
|
||||
@ -616,9 +625,13 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
|
||||
/** This is overriden by the mock node implementation to enable operation without any network map service */
|
||||
protected open fun noNetworkMapConfigured(): CordaFuture<Unit> {
|
||||
// TODO: There should be a consistent approach to configuration error exceptions.
|
||||
throw IllegalStateException("Configuration error: this node isn't being asked to act as the network map, nor " +
|
||||
"has any other map node been configured.")
|
||||
if (services.networkMapCache.loadDBSuccess) {
|
||||
return doneFuture(Unit)
|
||||
} else {
|
||||
// TODO: There should be a consistent approach to configuration error exceptions.
|
||||
throw IllegalStateException("Configuration error: this node isn't being asked to act as the network map, nor " +
|
||||
"has any other map node been configured.")
|
||||
}
|
||||
}
|
||||
|
||||
protected open fun makeKeyManagementService(identityService: IdentityService): KeyManagementService {
|
||||
@ -754,7 +767,8 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
override val monitoringService = MonitoringService(MetricRegistry())
|
||||
override val validatedTransactions = makeTransactionStorage()
|
||||
override val transactionVerifierService by lazy { makeTransactionVerifierService() }
|
||||
override val networkMapCache by lazy { InMemoryNetworkMapCache(this) }
|
||||
override val schemaService by lazy { NodeSchemaService(pluginRegistries.flatMap { it.requiredSchemas }.toSet()) }
|
||||
override val networkMapCache by lazy { PersistentNetworkMapCache(this) }
|
||||
override val vaultService by lazy { NodeVaultService(this) }
|
||||
override val vaultQueryService by lazy {
|
||||
HibernateVaultQueryImpl(database.hibernateConfig, vaultService)
|
||||
@ -776,7 +790,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
override val networkService: MessagingService get() = network
|
||||
override val clock: Clock get() = platformClock
|
||||
override val myInfo: NodeInfo get() = info
|
||||
override val schemaService by lazy { NodeSchemaService(pluginRegistries.flatMap { it.requiredSchemas }.toSet()) }
|
||||
override val database: CordaPersistence get() = this@AbstractNode.database
|
||||
override val configuration: NodeConfiguration get() = this@AbstractNode.configuration
|
||||
|
||||
|
||||
@ -173,7 +173,7 @@ class CordaRPCOpsImpl(
|
||||
override fun authoriseContractUpgrade(state: StateAndRef<*>, upgradedContractClass: Class<out UpgradedContract<*, *>>) = services.vaultService.authoriseContractUpgrade(state, upgradedContractClass)
|
||||
override fun deauthoriseContractUpgrade(state: StateAndRef<*>) = services.vaultService.deauthoriseContractUpgrade(state)
|
||||
override fun currentNodeTime(): Instant = Instant.now(services.clock)
|
||||
override fun waitUntilRegisteredWithNetworkMap() = services.networkMapCache.mapServiceRegistered
|
||||
override fun waitUntilNetworkReady() = services.networkMapCache.nodeReady
|
||||
override fun partyFromAnonymous(party: AbstractParty): Party? = services.identityService.partyFromAnonymous(party)
|
||||
override fun partyFromKey(key: PublicKey) = services.identityService.partyFromKey(key)
|
||||
override fun partyFromX500Name(x500Name: X500Name) = services.identityService.partyFromX500Name(x500Name)
|
||||
@ -182,6 +182,10 @@ class CordaRPCOpsImpl(
|
||||
|
||||
override fun registeredFlows(): List<String> = services.rpcFlows.map { it.name }.sorted()
|
||||
|
||||
override fun clearNetworkMapCache() {
|
||||
services.networkMapCache.clearNetworkMapCache()
|
||||
}
|
||||
|
||||
companion object {
|
||||
private fun stateMachineInfoFromFlowLogic(flowLogic: FlowLogic<*>): StateMachineInfo {
|
||||
return StateMachineInfo(flowLogic.runId, flowLogic.javaClass.name, flowLogic.stateMachine.flowInitiator, flowLogic.track())
|
||||
|
||||
@ -155,7 +155,7 @@ open class Node(override val configuration: FullNodeConfiguration,
|
||||
myIdentityOrNullIfNetworkMapService,
|
||||
serverThread,
|
||||
database,
|
||||
networkMapRegistrationFuture,
|
||||
nodeReadyFuture,
|
||||
services.monitoringService,
|
||||
advertisedAddress)
|
||||
}
|
||||
@ -210,15 +210,17 @@ open class Node(override val configuration: FullNodeConfiguration,
|
||||
log.trace { "Trying to detect public hostname through the Network Map Service at $serverAddress" }
|
||||
val tcpTransport = ArtemisTcpTransport.tcpTransport(ConnectionDirection.Outbound(), serverAddress, configuration)
|
||||
val locator = ActiveMQClient.createServerLocatorWithoutHA(tcpTransport).apply {
|
||||
initialConnectAttempts = 5
|
||||
retryInterval = 5.seconds.toMillis()
|
||||
initialConnectAttempts = 2 // TODO Public host discovery needs rewriting, as we may start nodes without network map, and we don't want to wait that long on startup.
|
||||
retryInterval = 2.seconds.toMillis()
|
||||
retryIntervalMultiplier = 1.5
|
||||
maxRetryInterval = 3.minutes.toMillis()
|
||||
}
|
||||
val clientFactory = try {
|
||||
locator.createSessionFactory()
|
||||
} catch (e: ActiveMQNotConnectedException) {
|
||||
throw IOException("Unable to connect to the Network Map Service at $serverAddress for IP address discovery", e)
|
||||
log.warn("Unable to connect to the Network Map Service at $serverAddress for IP address discovery. " +
|
||||
"Using the provided \"${configuration.p2pAddress.host}\" as the advertised address.")
|
||||
return null
|
||||
}
|
||||
|
||||
val session = clientFactory.createSession(PEER_USER, PEER_USER, false, true, true, locator.isPreAcknowledge, ActiveMQClient.DEFAULT_ACK_BATCH_SIZE)
|
||||
@ -306,7 +308,7 @@ open class Node(override val configuration: FullNodeConfiguration,
|
||||
}
|
||||
super.start()
|
||||
|
||||
networkMapRegistrationFuture.thenMatch({
|
||||
nodeReadyFuture.thenMatch({
|
||||
serverThread.execute {
|
||||
// Begin exporting our own metrics via JMX. These can be monitored using any agent, e.g. Jolokia:
|
||||
//
|
||||
|
||||
@ -102,7 +102,7 @@ open class NodeStartup(val args: Array<String>) {
|
||||
node.start()
|
||||
printPluginsAndServices(node)
|
||||
|
||||
node.networkMapRegistrationFuture.thenMatch({
|
||||
node.nodeReadyFuture.thenMatch({
|
||||
val elapsed = (System.currentTimeMillis() - startTime) / 10 / 100.0
|
||||
// TODO: Replace this with a standard function to get an unambiguous rendering of the X.500 name.
|
||||
val name = node.info.legalIdentity.name.orgName ?: node.info.legalIdentity.name.commonName
|
||||
|
||||
@ -53,6 +53,9 @@ interface NetworkMapCacheInternal : NetworkMapCache {
|
||||
/** For testing where the network map cache is manipulated marks the service as immediately ready. */
|
||||
@VisibleForTesting
|
||||
fun runWithoutMapService()
|
||||
|
||||
/** Indicates if loading network map data from database was successful. */
|
||||
val loadDBSuccess: Boolean
|
||||
}
|
||||
|
||||
@CordaSerializable
|
||||
|
||||
@ -338,6 +338,7 @@ class ArtemisMessagingServer(override val config: NodeConfiguration,
|
||||
* TODO : Create the bridge directly from the list of queues on start up when we have a persisted network map service.
|
||||
*/
|
||||
private fun updateBridgesOnNetworkChange(change: MapChange) {
|
||||
log.debug { "Updating bridges on network map change: ${change.node}" }
|
||||
fun gatherAddresses(node: NodeInfo): Sequence<ArtemisPeerAddress> {
|
||||
val peerAddress = getArtemisPeerAddress(node)
|
||||
val addresses = mutableListOf(peerAddress)
|
||||
|
||||
@ -1,185 +0,0 @@
|
||||
package net.corda.node.services.network
|
||||
|
||||
import net.corda.core.concurrent.CordaFuture
|
||||
import net.corda.core.identity.AbstractParty
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.internal.VisibleForTesting
|
||||
import net.corda.core.internal.bufferUntilSubscribed
|
||||
import net.corda.core.internal.concurrent.map
|
||||
import net.corda.core.internal.concurrent.openFuture
|
||||
import net.corda.core.messaging.DataFeed
|
||||
import net.corda.core.messaging.SingleMessageRecipient
|
||||
import net.corda.core.node.NodeInfo
|
||||
import net.corda.core.node.ServiceHub
|
||||
import net.corda.core.node.services.NetworkMapCache.MapChange
|
||||
import net.corda.core.node.services.PartyInfo
|
||||
import net.corda.core.serialization.SingletonSerializeAsToken
|
||||
import net.corda.core.serialization.deserialize
|
||||
import net.corda.core.serialization.serialize
|
||||
import net.corda.core.utilities.loggerFor
|
||||
import net.corda.node.services.api.NetworkCacheError
|
||||
import net.corda.node.services.api.NetworkMapCacheInternal
|
||||
import net.corda.node.services.messaging.MessagingService
|
||||
import net.corda.node.services.messaging.createMessage
|
||||
import net.corda.node.services.messaging.sendRequest
|
||||
import net.corda.node.services.network.NetworkMapService.FetchMapResponse
|
||||
import net.corda.node.services.network.NetworkMapService.SubscribeResponse
|
||||
import net.corda.node.utilities.AddOrRemove
|
||||
import net.corda.node.utilities.bufferUntilDatabaseCommit
|
||||
import net.corda.node.utilities.wrapWithDatabaseTransaction
|
||||
import rx.Observable
|
||||
import rx.subjects.PublishSubject
|
||||
import java.security.PublicKey
|
||||
import java.security.SignatureException
|
||||
import java.util.*
|
||||
import javax.annotation.concurrent.ThreadSafe
|
||||
|
||||
/**
|
||||
* Extremely simple in-memory cache of the network map.
|
||||
*
|
||||
* @param serviceHub an optional service hub from which we'll take the identity service. We take a service hub rather
|
||||
* than the identity service directly, as this avoids problems with service start sequence (network map cache
|
||||
* and identity services depend on each other). Should always be provided except for unit test cases.
|
||||
*/
|
||||
@ThreadSafe
|
||||
open class InMemoryNetworkMapCache(private val serviceHub: ServiceHub?) : SingletonSerializeAsToken(), NetworkMapCacheInternal {
|
||||
companion object {
|
||||
val logger = loggerFor<InMemoryNetworkMapCache>()
|
||||
}
|
||||
|
||||
override val partyNodes: List<NodeInfo> get() = registeredNodes.map { it.value }
|
||||
override val networkMapNodes: List<NodeInfo> get() = getNodesWithService(NetworkMapService.type)
|
||||
private val _changed = PublishSubject.create<MapChange>()
|
||||
// We use assignment here so that multiple subscribers share the same wrapped Observable.
|
||||
override val changed: Observable<MapChange> = _changed.wrapWithDatabaseTransaction()
|
||||
private val changePublisher: rx.Observer<MapChange> get() = _changed.bufferUntilDatabaseCommit()
|
||||
|
||||
private val _registrationFuture = openFuture<Void?>()
|
||||
override val mapServiceRegistered: CordaFuture<Void?> get() = _registrationFuture
|
||||
|
||||
private var registeredForPush = false
|
||||
protected var registeredNodes: MutableMap<PublicKey, NodeInfo> = Collections.synchronizedMap(HashMap())
|
||||
|
||||
override fun getPartyInfo(party: Party): PartyInfo? {
|
||||
val node = registeredNodes[party.owningKey]
|
||||
if (node != null) {
|
||||
return PartyInfo.Node(node)
|
||||
}
|
||||
for ((_, value) in registeredNodes) {
|
||||
for (service in value.advertisedServices) {
|
||||
if (service.identity.party == party) {
|
||||
return PartyInfo.Service(service)
|
||||
}
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
override fun getNodeByLegalIdentityKey(identityKey: PublicKey): NodeInfo? = registeredNodes[identityKey]
|
||||
override fun getNodeByLegalIdentity(party: AbstractParty): NodeInfo? {
|
||||
val wellKnownParty = if (serviceHub != null) {
|
||||
serviceHub.identityService.partyFromAnonymous(party)
|
||||
} else {
|
||||
party
|
||||
}
|
||||
|
||||
return wellKnownParty?.let {
|
||||
getNodeByLegalIdentityKey(it.owningKey)
|
||||
}
|
||||
}
|
||||
|
||||
override fun track(): DataFeed<List<NodeInfo>, MapChange> {
|
||||
synchronized(_changed) {
|
||||
return DataFeed(partyNodes, _changed.bufferUntilSubscribed().wrapWithDatabaseTransaction())
|
||||
}
|
||||
}
|
||||
|
||||
override fun addMapService(network: MessagingService, networkMapAddress: SingleMessageRecipient, subscribe: Boolean,
|
||||
ifChangedSinceVer: Int?): CordaFuture<Unit> {
|
||||
if (subscribe && !registeredForPush) {
|
||||
// Add handler to the network, for updates received from the remote network map service.
|
||||
network.addMessageHandler(NetworkMapService.PUSH_TOPIC) { message, _ ->
|
||||
try {
|
||||
val req = message.data.deserialize<NetworkMapService.Update>()
|
||||
val ackMessage = network.createMessage(NetworkMapService.PUSH_ACK_TOPIC,
|
||||
data = NetworkMapService.UpdateAcknowledge(req.mapVersion, network.myAddress).serialize().bytes)
|
||||
network.send(ackMessage, req.replyTo)
|
||||
processUpdatePush(req)
|
||||
} catch(e: NodeMapError) {
|
||||
logger.warn("Failure during node map update due to bad update: ${e.javaClass.name}")
|
||||
} catch(e: Exception) {
|
||||
logger.error("Exception processing update from network map service", e)
|
||||
}
|
||||
}
|
||||
registeredForPush = true
|
||||
}
|
||||
|
||||
// Fetch the network map and register for updates at the same time
|
||||
val req = NetworkMapService.FetchMapRequest(subscribe, ifChangedSinceVer, network.myAddress)
|
||||
val future = network.sendRequest<FetchMapResponse>(NetworkMapService.FETCH_TOPIC, req, networkMapAddress).map { (nodes) ->
|
||||
// We may not receive any nodes back, if the map hasn't changed since the version specified
|
||||
nodes?.forEach { processRegistration(it) }
|
||||
Unit
|
||||
}
|
||||
_registrationFuture.captureLater(future.map { null })
|
||||
|
||||
return future
|
||||
}
|
||||
|
||||
override fun addNode(node: NodeInfo) {
|
||||
synchronized(_changed) {
|
||||
val previousNode = registeredNodes.put(node.legalIdentity.owningKey, node)
|
||||
if (previousNode == null) {
|
||||
changePublisher.onNext(MapChange.Added(node))
|
||||
} else if (previousNode != node) {
|
||||
changePublisher.onNext(MapChange.Modified(node, previousNode))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun removeNode(node: NodeInfo) {
|
||||
synchronized(_changed) {
|
||||
registeredNodes.remove(node.legalIdentity.owningKey)
|
||||
changePublisher.onNext(MapChange.Removed(node))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Unsubscribes from updates from the given map service.
|
||||
* @param service the network map service to listen to updates from.
|
||||
*/
|
||||
override fun deregisterForUpdates(network: MessagingService, service: NodeInfo): CordaFuture<Unit> {
|
||||
// Fetch the network map and register for updates at the same time
|
||||
val req = NetworkMapService.SubscribeRequest(false, network.myAddress)
|
||||
// `network.getAddressOfParty(partyInfo)` is a work-around for MockNetwork and InMemoryMessaging to get rid of SingleMessageRecipient in NodeInfo.
|
||||
val address = network.getAddressOfParty(PartyInfo.Node(service))
|
||||
val future = network.sendRequest<SubscribeResponse>(NetworkMapService.SUBSCRIPTION_TOPIC, req, address).map {
|
||||
if (it.confirmed) Unit else throw NetworkCacheError.DeregistrationFailed()
|
||||
}
|
||||
_registrationFuture.captureLater(future.map { null })
|
||||
return future
|
||||
}
|
||||
|
||||
fun processUpdatePush(req: NetworkMapService.Update) {
|
||||
try {
|
||||
val reg = req.wireReg.verified()
|
||||
processRegistration(reg)
|
||||
} catch (e: SignatureException) {
|
||||
throw NodeMapError.InvalidSignature()
|
||||
}
|
||||
}
|
||||
|
||||
private fun processRegistration(reg: NodeRegistration) {
|
||||
// TODO: Implement filtering by sequence number, so we only accept changes that are
|
||||
// more recent than the latest change we've processed.
|
||||
when (reg.type) {
|
||||
AddOrRemove.ADD -> addNode(reg.node)
|
||||
AddOrRemove.REMOVE -> removeNode(reg.node)
|
||||
}
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
override fun runWithoutMapService() {
|
||||
_registrationFuture.set(null)
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,339 @@
|
||||
package net.corda.node.services.network
|
||||
|
||||
import net.corda.core.concurrent.CordaFuture
|
||||
import net.corda.core.internal.bufferUntilSubscribed
|
||||
import net.corda.core.crypto.parsePublicKeyBase58
|
||||
import net.corda.core.crypto.toBase58String
|
||||
import net.corda.core.identity.AbstractParty
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.internal.VisibleForTesting
|
||||
import net.corda.core.internal.concurrent.map
|
||||
import net.corda.core.internal.concurrent.openFuture
|
||||
import net.corda.core.messaging.DataFeed
|
||||
import net.corda.core.messaging.SingleMessageRecipient
|
||||
import net.corda.core.node.NodeInfo
|
||||
import net.corda.core.node.services.NetworkMapCache.MapChange
|
||||
import net.corda.core.node.services.PartyInfo
|
||||
import net.corda.core.schemas.NodeInfoSchemaV1
|
||||
import net.corda.core.serialization.SingletonSerializeAsToken
|
||||
import net.corda.core.serialization.deserialize
|
||||
import net.corda.core.serialization.serialize
|
||||
import net.corda.core.utilities.NetworkHostAndPort
|
||||
import net.corda.core.utilities.loggerFor
|
||||
import net.corda.node.services.api.NetworkCacheError
|
||||
import net.corda.node.services.api.NetworkMapCacheInternal
|
||||
import net.corda.node.services.api.ServiceHubInternal
|
||||
import net.corda.node.services.messaging.MessagingService
|
||||
import net.corda.node.services.messaging.createMessage
|
||||
import net.corda.node.services.messaging.sendRequest
|
||||
import net.corda.node.services.network.NetworkMapService.FetchMapResponse
|
||||
import net.corda.node.services.network.NetworkMapService.SubscribeResponse
|
||||
import net.corda.node.utilities.AddOrRemove
|
||||
import net.corda.node.utilities.DatabaseTransactionManager
|
||||
import net.corda.node.utilities.bufferUntilDatabaseCommit
|
||||
import net.corda.node.utilities.wrapWithDatabaseTransaction
|
||||
import org.bouncycastle.asn1.x500.X500Name
|
||||
import org.hibernate.Session
|
||||
import rx.Observable
|
||||
import rx.subjects.PublishSubject
|
||||
import java.security.PublicKey
|
||||
import java.security.SignatureException
|
||||
import java.util.*
|
||||
import javax.annotation.concurrent.ThreadSafe
|
||||
|
||||
/**
|
||||
* Extremely simple in-memory cache of the network map.
|
||||
*
|
||||
* @param serviceHub an optional service hub from which we'll take the identity service. We take a service hub rather
|
||||
* than the identity service directly, as this avoids problems with service start sequence (network map cache
|
||||
* and identity services depend on each other). Should always be provided except for unit test cases.
|
||||
*/
|
||||
@ThreadSafe
|
||||
open class PersistentNetworkMapCache(private val serviceHub: ServiceHubInternal) : SingletonSerializeAsToken(), NetworkMapCacheInternal {
|
||||
companion object {
|
||||
val logger = loggerFor<PersistentNetworkMapCache>()
|
||||
}
|
||||
|
||||
private var registeredForPush = false
|
||||
// TODO Small explanation, partyNodes and registeredNodes is left in memory as it was before, because it will be removed in
|
||||
// next PR that gets rid of services. These maps are used only for queries by service.
|
||||
override val partyNodes: List<NodeInfo> get() = registeredNodes.map { it.value }
|
||||
override val networkMapNodes: List<NodeInfo> get() = getNodesWithService(NetworkMapService.type)
|
||||
private val _changed = PublishSubject.create<MapChange>()
|
||||
// We use assignment here so that multiple subscribers share the same wrapped Observable.
|
||||
override val changed: Observable<MapChange> = _changed.wrapWithDatabaseTransaction()
|
||||
private val changePublisher: rx.Observer<MapChange> get() = _changed.bufferUntilDatabaseCommit()
|
||||
|
||||
private val _registrationFuture = openFuture<Void?>()
|
||||
override val nodeReady: CordaFuture<Void?> get() = _registrationFuture
|
||||
protected val registeredNodes: MutableMap<PublicKey, NodeInfo> = Collections.synchronizedMap(HashMap())
|
||||
private var _loadDBSuccess: Boolean = false
|
||||
override val loadDBSuccess get() = _loadDBSuccess
|
||||
|
||||
init {
|
||||
serviceHub.database.transaction { loadFromDB() }
|
||||
}
|
||||
|
||||
override fun getPartyInfo(party: Party): PartyInfo? {
|
||||
val nodes = serviceHub.database.transaction { queryByIdentityKey(party.owningKey) }
|
||||
if (nodes.size == 1 && nodes[0].legalIdentity == party) {
|
||||
return PartyInfo.Node(nodes[0])
|
||||
}
|
||||
for (node in nodes) {
|
||||
for (service in node.advertisedServices) {
|
||||
if (service.identity.party == party) {
|
||||
return PartyInfo.Service(service)
|
||||
}
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
// TODO See comment to queryByLegalName why it's left like that.
|
||||
override fun getNodeByLegalName(principal: X500Name): NodeInfo? = partyNodes.singleOrNull { it.legalIdentity.name == principal }
|
||||
//serviceHub!!.database.transaction { queryByLegalName(principal).firstOrNull() }
|
||||
override fun getNodeByLegalIdentityKey(identityKey: PublicKey): NodeInfo? =
|
||||
serviceHub.database.transaction { queryByIdentityKey(identityKey).firstOrNull() }
|
||||
override fun getNodeByLegalIdentity(party: AbstractParty): NodeInfo? {
|
||||
val wellKnownParty = serviceHub.identityService.partyFromAnonymous(party)
|
||||
return wellKnownParty?.let {
|
||||
getNodeByLegalIdentityKey(it.owningKey)
|
||||
}
|
||||
}
|
||||
|
||||
override fun getNodeByAddress(address: NetworkHostAndPort): NodeInfo? = serviceHub.database.transaction { queryByAddress(address) }
|
||||
|
||||
override fun track(): DataFeed<List<NodeInfo>, MapChange> {
|
||||
synchronized(_changed) {
|
||||
return DataFeed(partyNodes, _changed.bufferUntilSubscribed().wrapWithDatabaseTransaction())
|
||||
}
|
||||
}
|
||||
|
||||
override fun addMapService(network: MessagingService, networkMapAddress: SingleMessageRecipient, subscribe: Boolean,
|
||||
ifChangedSinceVer: Int?): CordaFuture<Unit> {
|
||||
if (subscribe && !registeredForPush) {
|
||||
// Add handler to the network, for updates received from the remote network map service.
|
||||
network.addMessageHandler(NetworkMapService.PUSH_TOPIC) { message, _ ->
|
||||
try {
|
||||
val req = message.data.deserialize<NetworkMapService.Update>()
|
||||
val ackMessage = network.createMessage(NetworkMapService.PUSH_ACK_TOPIC,
|
||||
data = NetworkMapService.UpdateAcknowledge(req.mapVersion, network.myAddress).serialize().bytes)
|
||||
network.send(ackMessage, req.replyTo)
|
||||
processUpdatePush(req)
|
||||
} catch(e: NodeMapError) {
|
||||
logger.warn("Failure during node map update due to bad update: ${e.javaClass.name}")
|
||||
} catch(e: Exception) {
|
||||
logger.error("Exception processing update from network map service", e)
|
||||
}
|
||||
}
|
||||
registeredForPush = true
|
||||
}
|
||||
|
||||
// Fetch the network map and register for updates at the same time
|
||||
val req = NetworkMapService.FetchMapRequest(subscribe, ifChangedSinceVer, network.myAddress)
|
||||
val future = network.sendRequest<FetchMapResponse>(NetworkMapService.FETCH_TOPIC, req, networkMapAddress).map { (nodes) ->
|
||||
// We may not receive any nodes back, if the map hasn't changed since the version specified
|
||||
nodes?.forEach { processRegistration(it) }
|
||||
Unit
|
||||
}
|
||||
_registrationFuture.captureLater(future.map { null })
|
||||
|
||||
return future
|
||||
}
|
||||
|
||||
override fun addNode(node: NodeInfo) {
|
||||
synchronized(_changed) {
|
||||
val previousNode = registeredNodes.put(node.legalIdentity.owningKey, node)
|
||||
if (previousNode == null) {
|
||||
serviceHub.database.transaction {
|
||||
updateInfoDB(node)
|
||||
changePublisher.onNext(MapChange.Added(node))
|
||||
}
|
||||
} else if (previousNode != node) {
|
||||
serviceHub.database.transaction {
|
||||
updateInfoDB(node)
|
||||
changePublisher.onNext(MapChange.Modified(node, previousNode))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun removeNode(node: NodeInfo) {
|
||||
synchronized(_changed) {
|
||||
registeredNodes.remove(node.legalIdentity.owningKey)
|
||||
serviceHub.database.transaction {
|
||||
removeInfoDB(node)
|
||||
changePublisher.onNext(MapChange.Removed(node))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Unsubscribes from updates from the given map service.
|
||||
* @param service the network map service to listen to updates from.
|
||||
*/
|
||||
override fun deregisterForUpdates(network: MessagingService, service: NodeInfo): CordaFuture<Unit> {
|
||||
// Fetch the network map and register for updates at the same time
|
||||
val req = NetworkMapService.SubscribeRequest(false, network.myAddress)
|
||||
// `network.getAddressOfParty(partyInfo)` is a work-around for MockNetwork and InMemoryMessaging to get rid of SingleMessageRecipient in NodeInfo.
|
||||
val address = network.getAddressOfParty(PartyInfo.Node(service))
|
||||
val future = network.sendRequest<SubscribeResponse>(NetworkMapService.SUBSCRIPTION_TOPIC, req, address).map {
|
||||
if (it.confirmed) Unit else throw NetworkCacheError.DeregistrationFailed()
|
||||
}
|
||||
_registrationFuture.captureLater(future.map { null })
|
||||
return future
|
||||
}
|
||||
|
||||
fun processUpdatePush(req: NetworkMapService.Update) {
|
||||
try {
|
||||
val reg = req.wireReg.verified()
|
||||
processRegistration(reg)
|
||||
} catch (e: SignatureException) {
|
||||
throw NodeMapError.InvalidSignature()
|
||||
}
|
||||
}
|
||||
|
||||
private fun processRegistration(reg: NodeRegistration) {
|
||||
// TODO: Implement filtering by sequence number, so we only accept changes that are
|
||||
// more recent than the latest change we've processed.
|
||||
when (reg.type) {
|
||||
AddOrRemove.ADD -> addNode(reg.node)
|
||||
AddOrRemove.REMOVE -> removeNode(reg.node)
|
||||
}
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
override fun runWithoutMapService() {
|
||||
_registrationFuture.set(null)
|
||||
}
|
||||
|
||||
// Changes related to NetworkMap redesign
|
||||
// TODO It will be properly merged into network map cache after services removal.
|
||||
|
||||
private inline fun <T> createSession(block: (Session) -> T): T {
|
||||
return DatabaseTransactionManager.current().session.let { block(it) }
|
||||
}
|
||||
|
||||
private fun getAllInfos(session: Session): List<NodeInfoSchemaV1.PersistentNodeInfo> {
|
||||
val criteria = session.criteriaBuilder.createQuery(NodeInfoSchemaV1.PersistentNodeInfo::class.java)
|
||||
criteria.select(criteria.from(NodeInfoSchemaV1.PersistentNodeInfo::class.java))
|
||||
return session.createQuery(criteria).resultList
|
||||
}
|
||||
|
||||
/**
|
||||
* Load NetworkMap data from the database if present. Node can start without having NetworkMapService configured.
|
||||
*/
|
||||
private fun loadFromDB() {
|
||||
logger.info("Loading network map from database...")
|
||||
createSession {
|
||||
val result = getAllInfos(it)
|
||||
for (nodeInfo in result) {
|
||||
try {
|
||||
logger.info("Loaded node info: $nodeInfo")
|
||||
val publicKey = parsePublicKeyBase58(nodeInfo.legalIdentitiesAndCerts.single { it.isMain }.owningKey)
|
||||
val node = nodeInfo.toNodeInfo()
|
||||
registeredNodes.put(publicKey, node)
|
||||
changePublisher.onNext(MapChange.Added(node)) // Redeploy bridges after reading from DB on startup.
|
||||
_loadDBSuccess = true // This is used in AbstractNode to indicate that node is ready.
|
||||
} catch (e: Exception) {
|
||||
logger.warn("Exception parsing network map from the database.", e)
|
||||
}
|
||||
}
|
||||
if (loadDBSuccess) {
|
||||
_registrationFuture.set(null) // Useful only if we don't have NetworkMapService configured so StateMachineManager can start.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun updateInfoDB(nodeInfo: NodeInfo) {
|
||||
// TODO Temporary workaround to force isolated transaction (otherwise it causes race conditions when processing
|
||||
// network map registration on network map node)
|
||||
val session = serviceHub.database.entityManagerFactory.withOptions().connection(serviceHub.database.dataSource.connection
|
||||
.apply {
|
||||
transactionIsolation = 1
|
||||
}).openSession()
|
||||
|
||||
val tx = session.beginTransaction()
|
||||
// TODO For now the main legal identity is left in NodeInfo, this should be set comparision/come up with index for NodeInfo?
|
||||
val info = findByIdentityKey(session, nodeInfo.legalIdentity.owningKey)
|
||||
val nodeInfoEntry = generateMappedObject(nodeInfo)
|
||||
if (info.isNotEmpty()) {
|
||||
nodeInfoEntry.id = info[0].id
|
||||
}
|
||||
session.merge(nodeInfoEntry)
|
||||
tx.commit()
|
||||
session.close()
|
||||
}
|
||||
|
||||
private fun removeInfoDB(nodeInfo: NodeInfo) {
|
||||
createSession {
|
||||
val info = findByIdentityKey(it, nodeInfo.legalIdentity.owningKey).single()
|
||||
it.remove(info)
|
||||
}
|
||||
}
|
||||
|
||||
private fun findByIdentityKey(session: Session, identityKey: PublicKey): List<NodeInfoSchemaV1.PersistentNodeInfo> {
|
||||
val query = session.createQuery(
|
||||
"SELECT n FROM ${NodeInfoSchemaV1.PersistentNodeInfo::class.java.name} n JOIN n.legalIdentitiesAndCerts l WHERE l.owningKey = :owningKey",
|
||||
NodeInfoSchemaV1.PersistentNodeInfo::class.java)
|
||||
query.setParameter("owningKey", identityKey.toBase58String())
|
||||
return query.resultList
|
||||
}
|
||||
|
||||
private fun queryByIdentityKey(identityKey: PublicKey): List<NodeInfo> {
|
||||
createSession {
|
||||
val result = findByIdentityKey(it, identityKey)
|
||||
return result.map { it.toNodeInfo() }
|
||||
}
|
||||
}
|
||||
|
||||
// TODO It's useless for now, because toString on X500 names is inconsistent and we have:
|
||||
// C=ES,L=Madrid,O=Alice Corp,CN=Alice Corp
|
||||
// CN=Alice Corp,O=Alice Corp,L=Madrid,C=ES
|
||||
private fun queryByLegalName(name: X500Name): List<NodeInfo> {
|
||||
createSession {
|
||||
val query = it.createQuery(
|
||||
"SELECT n FROM ${NodeInfoSchemaV1.PersistentNodeInfo::class.java.name} n JOIN n.legalIdentitiesAndCerts l WHERE l.name = :name",
|
||||
NodeInfoSchemaV1.PersistentNodeInfo::class.java)
|
||||
query.setParameter("name", name.toString())
|
||||
val result = query.resultList
|
||||
return result.map { it.toNodeInfo() }
|
||||
}
|
||||
}
|
||||
|
||||
private fun queryByAddress(hostAndPort: NetworkHostAndPort): NodeInfo? {
|
||||
createSession {
|
||||
val query = it.createQuery(
|
||||
"SELECT n FROM ${NodeInfoSchemaV1.PersistentNodeInfo::class.java.name} n JOIN n.addresses a WHERE a.pk.host = :host AND a.pk.port = :port",
|
||||
NodeInfoSchemaV1.PersistentNodeInfo::class.java)
|
||||
query.setParameter("host", hostAndPort.host)
|
||||
query.setParameter("port", hostAndPort.port)
|
||||
val result = query.resultList
|
||||
return if (result.isEmpty()) null
|
||||
else result.map { it.toNodeInfo() }.singleOrNull() ?: throw IllegalStateException("More than one node with the same host and port")
|
||||
}
|
||||
}
|
||||
|
||||
/** Object Relational Mapping support. */
|
||||
private fun generateMappedObject(nodeInfo: NodeInfo): NodeInfoSchemaV1.PersistentNodeInfo {
|
||||
return NodeInfoSchemaV1.PersistentNodeInfo(
|
||||
id = 0,
|
||||
addresses = nodeInfo.addresses.map { NodeInfoSchemaV1.DBHostAndPort.fromHostAndPort(it) },
|
||||
legalIdentitiesAndCerts = nodeInfo.legalIdentitiesAndCerts.map { NodeInfoSchemaV1.DBPartyAndCertificate(it) }.toSet()
|
||||
// TODO It's workaround to keep the main identity, will be removed in future PR getting rid of services.
|
||||
+ NodeInfoSchemaV1.DBPartyAndCertificate(nodeInfo.legalIdentityAndCert, isMain = true),
|
||||
platformVersion = nodeInfo.platformVersion,
|
||||
advertisedServices = nodeInfo.advertisedServices.map { NodeInfoSchemaV1.DBServiceEntry(it.serialize().bytes) },
|
||||
serial = nodeInfo.serial
|
||||
)
|
||||
}
|
||||
|
||||
override fun clearNetworkMapCache() {
|
||||
serviceHub.database.transaction {
|
||||
createSession {
|
||||
val result = getAllInfos(it)
|
||||
for (nodeInfo in result) it.remove(nodeInfo)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -5,6 +5,7 @@ import net.corda.core.contracts.FungibleAsset
|
||||
import net.corda.core.contracts.LinearState
|
||||
import net.corda.core.schemas.CommonSchemaV1
|
||||
import net.corda.core.schemas.MappedSchema
|
||||
import net.corda.core.schemas.NodeInfoSchemaV1
|
||||
import net.corda.core.schemas.PersistentState
|
||||
import net.corda.core.schemas.QueryableState
|
||||
import net.corda.core.serialization.SingletonSerializeAsToken
|
||||
@ -58,6 +59,7 @@ class NodeSchemaService(customSchemas: Set<MappedSchema> = emptySet()) : SchemaS
|
||||
val requiredSchemas: Map<MappedSchema, SchemaService.SchemaOptions> =
|
||||
mapOf(Pair(CommonSchemaV1, SchemaService.SchemaOptions()),
|
||||
Pair(VaultSchemaV1, SchemaService.SchemaOptions()),
|
||||
Pair(NodeInfoSchemaV1, SchemaService.SchemaOptions()),
|
||||
Pair(NodeServicesV1, SchemaService.SchemaOptions()))
|
||||
|
||||
override val schemaOptions: Map<MappedSchema, SchemaService.SchemaOptions> = requiredSchemas.plus(customSchemas.map {
|
||||
|
||||
@ -39,6 +39,7 @@ import net.corda.node.utilities.wrapWithDatabaseTransaction
|
||||
import net.corda.nodeapi.internal.serialization.SerializeAsTokenContextImpl
|
||||
import net.corda.nodeapi.internal.serialization.withTokenContext
|
||||
import org.apache.activemq.artemis.utils.ReusableLatch
|
||||
import org.bouncycastle.asn1.x500.X500Name
|
||||
import org.slf4j.Logger
|
||||
import rx.Observable
|
||||
import rx.subjects.PublishSubject
|
||||
@ -170,7 +171,7 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
|
||||
checkQuasarJavaAgentPresence()
|
||||
restoreFibersFromCheckpoints()
|
||||
listenToLedgerTransactions()
|
||||
serviceHub.networkMapCache.mapServiceRegistered.then { executor.execute(this::resumeRestoredFibers) }
|
||||
serviceHub.networkMapCache.nodeReady.then { executor.execute(this::resumeRestoredFibers) }
|
||||
}
|
||||
|
||||
private fun checkQuasarJavaAgentPresence() {
|
||||
|
||||
@ -1,85 +0,0 @@
|
||||
package net.corda.node.services
|
||||
|
||||
import com.codahale.metrics.MetricRegistry
|
||||
import net.corda.core.flows.FlowInitiator
|
||||
import net.corda.core.flows.FlowLogic
|
||||
import net.corda.core.node.NodeInfo
|
||||
import net.corda.core.node.services.*
|
||||
import net.corda.core.serialization.SerializeAsToken
|
||||
import net.corda.core.utilities.NonEmptySet
|
||||
import net.corda.node.internal.InitiatedFlowFactory
|
||||
import net.corda.node.serialization.NodeClock
|
||||
import net.corda.node.services.api.*
|
||||
import net.corda.node.services.config.NodeConfiguration
|
||||
import net.corda.node.services.messaging.MessagingService
|
||||
import net.corda.node.services.schema.NodeSchemaService
|
||||
import net.corda.node.services.statemachine.FlowStateMachineImpl
|
||||
import net.corda.node.services.statemachine.StateMachineManager
|
||||
import net.corda.node.services.transactions.InMemoryTransactionVerifierService
|
||||
import net.corda.node.utilities.CordaPersistence
|
||||
import net.corda.testing.DUMMY_IDENTITY_1
|
||||
import net.corda.testing.MOCK_HOST_AND_PORT
|
||||
import net.corda.testing.MOCK_IDENTITY_SERVICE
|
||||
import net.corda.testing.node.MockAttachmentStorage
|
||||
import net.corda.testing.node.MockNetworkMapCache
|
||||
import net.corda.testing.node.MockStateMachineRecordedTransactionMappingStorage
|
||||
import net.corda.testing.node.MockTransactionStorage
|
||||
import java.sql.Connection
|
||||
import java.time.Clock
|
||||
|
||||
open class MockServiceHubInternal(
|
||||
override val database: CordaPersistence,
|
||||
override val configuration: NodeConfiguration,
|
||||
val customVault: VaultService? = null,
|
||||
val customVaultQuery: VaultQueryService? = null,
|
||||
val keyManagement: KeyManagementService? = null,
|
||||
val network: MessagingService? = null,
|
||||
val identity: IdentityService? = MOCK_IDENTITY_SERVICE,
|
||||
override val attachments: AttachmentStorage = MockAttachmentStorage(),
|
||||
override val validatedTransactions: WritableTransactionStorage = MockTransactionStorage(),
|
||||
override val stateMachineRecordedTransactionMapping: StateMachineRecordedTransactionMappingStorage = MockStateMachineRecordedTransactionMappingStorage(),
|
||||
val mapCache: NetworkMapCacheInternal? = null,
|
||||
val scheduler: SchedulerService? = null,
|
||||
val overrideClock: Clock? = NodeClock(),
|
||||
val schemas: SchemaService? = NodeSchemaService(),
|
||||
val customTransactionVerifierService: TransactionVerifierService? = InMemoryTransactionVerifierService(2)
|
||||
) : ServiceHubInternal {
|
||||
override val vaultQueryService: VaultQueryService
|
||||
get() = customVaultQuery ?: throw UnsupportedOperationException()
|
||||
override val transactionVerifierService: TransactionVerifierService
|
||||
get() = customTransactionVerifierService ?: throw UnsupportedOperationException()
|
||||
override val vaultService: VaultService
|
||||
get() = customVault ?: throw UnsupportedOperationException()
|
||||
override val keyManagementService: KeyManagementService
|
||||
get() = keyManagement ?: throw UnsupportedOperationException()
|
||||
override val identityService: IdentityService
|
||||
get() = identity ?: throw UnsupportedOperationException()
|
||||
override val networkService: MessagingService
|
||||
get() = network ?: throw UnsupportedOperationException()
|
||||
override val networkMapCache: NetworkMapCacheInternal
|
||||
get() = mapCache ?: MockNetworkMapCache(this)
|
||||
override val schedulerService: SchedulerService
|
||||
get() = scheduler ?: throw UnsupportedOperationException()
|
||||
override val clock: Clock
|
||||
get() = overrideClock ?: throw UnsupportedOperationException()
|
||||
override val myInfo: NodeInfo
|
||||
get() = NodeInfo(listOf(MOCK_HOST_AND_PORT), DUMMY_IDENTITY_1, NonEmptySet.of(DUMMY_IDENTITY_1), 1) // Required to get a dummy platformVersion when required for tests.
|
||||
override val monitoringService: MonitoringService = MonitoringService(MetricRegistry())
|
||||
override val rpcFlows: List<Class<out FlowLogic<*>>>
|
||||
get() = throw UnsupportedOperationException()
|
||||
override val schemaService: SchemaService
|
||||
get() = schemas ?: throw UnsupportedOperationException()
|
||||
override val auditService: AuditService = DummyAuditService()
|
||||
|
||||
lateinit var smm: StateMachineManager
|
||||
|
||||
override fun <T : SerializeAsToken> cordaService(type: Class<T>): T = throw UnsupportedOperationException()
|
||||
|
||||
override fun <T> startFlow(logic: FlowLogic<T>, flowInitiator: FlowInitiator): FlowStateMachineImpl<T> {
|
||||
return smm.executor.fetchFrom { smm.add(logic, flowInitiator) }
|
||||
}
|
||||
|
||||
override fun getFlowFactory(initiatingFlowClass: Class<out FlowLogic<*>>): InitiatedFlowFactory<*>? = null
|
||||
|
||||
override fun jdbcSession(): Connection = database.createSession()
|
||||
}
|
||||
@ -11,12 +11,12 @@ import net.corda.core.node.services.VaultService
|
||||
import net.corda.core.serialization.SingletonSerializeAsToken
|
||||
import net.corda.core.transactions.TransactionBuilder
|
||||
import net.corda.core.utilities.days
|
||||
import net.corda.node.services.MockServiceHubInternal
|
||||
import net.corda.node.services.identity.InMemoryIdentityService
|
||||
import net.corda.node.services.persistence.DBCheckpointStorage
|
||||
import net.corda.node.services.statemachine.FlowLogicRefFactoryImpl
|
||||
import net.corda.node.services.statemachine.StateMachineManager
|
||||
import net.corda.node.services.vault.NodeVaultService
|
||||
import net.corda.node.testing.MockServiceHubInternal
|
||||
import net.corda.node.utilities.AffinityExecutor
|
||||
import net.corda.node.utilities.CordaPersistence
|
||||
import net.corda.node.utilities.configureDatabase
|
||||
|
||||
@ -12,9 +12,10 @@ import net.corda.node.services.RPCUserServiceImpl
|
||||
import net.corda.node.services.api.MonitoringService
|
||||
import net.corda.node.services.config.NodeConfiguration
|
||||
import net.corda.node.services.config.configureWithDevSSLCertificate
|
||||
import net.corda.node.services.network.InMemoryNetworkMapCache
|
||||
import net.corda.node.services.network.PersistentNetworkMapCache
|
||||
import net.corda.node.services.network.NetworkMapService
|
||||
import net.corda.node.services.transactions.PersistentUniquenessProvider
|
||||
import net.corda.node.testing.MockServiceHubInternal
|
||||
import net.corda.node.utilities.AffinityExecutor.ServiceAffinityExecutor
|
||||
import net.corda.node.utilities.CordaPersistence
|
||||
import net.corda.node.utilities.configureDatabase
|
||||
@ -54,8 +55,7 @@ class ArtemisMessagingTests : TestDependencyInjectionBase() {
|
||||
var messagingClient: NodeMessagingClient? = null
|
||||
var messagingServer: ArtemisMessagingServer? = null
|
||||
|
||||
// TODO: We should have a dummy service hub rather than change behaviour in tests
|
||||
val networkMapCache = InMemoryNetworkMapCache(serviceHub = null)
|
||||
lateinit var networkMapCache: PersistentNetworkMapCache
|
||||
|
||||
val rpcOps = object : RPCOps {
|
||||
override val protocolVersion: Int get() = throw UnsupportedOperationException()
|
||||
@ -71,6 +71,7 @@ class ArtemisMessagingTests : TestDependencyInjectionBase() {
|
||||
LogHelper.setLevel(PersistentUniquenessProvider::class)
|
||||
database = configureDatabase(makeTestDataSourceProperties(), makeTestDatabaseProperties(), createIdentityService = ::makeTestIdentityService)
|
||||
networkMapRegistrationFuture = doneFuture(Unit)
|
||||
networkMapCache = PersistentNetworkMapCache(serviceHub = object : MockServiceHubInternal(database, config) {})
|
||||
}
|
||||
|
||||
@After
|
||||
|
||||
@ -6,13 +6,14 @@ import net.corda.core.utilities.getOrThrow
|
||||
import net.corda.testing.ALICE
|
||||
import net.corda.testing.BOB
|
||||
import net.corda.testing.node.MockNetwork
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.junit.After
|
||||
import org.junit.Before
|
||||
import org.junit.Test
|
||||
import java.math.BigInteger
|
||||
import kotlin.test.assertEquals
|
||||
|
||||
class InMemoryNetworkMapCacheTest {
|
||||
class NetworkMapCacheTest {
|
||||
lateinit var mockNet: MockNetwork
|
||||
|
||||
@Before
|
||||
@ -47,9 +48,7 @@ class InMemoryNetworkMapCacheTest {
|
||||
// Node A currently knows only about itself, so this returns node A
|
||||
assertEquals(nodeA.services.networkMapCache.getNodeByLegalIdentityKey(nodeA.info.legalIdentity.owningKey), nodeA.info)
|
||||
|
||||
nodeA.database.transaction {
|
||||
nodeA.services.networkMapCache.addNode(nodeB.info)
|
||||
}
|
||||
nodeA.services.networkMapCache.addNode(nodeB.info)
|
||||
// The details of node B write over those for node A
|
||||
assertEquals(nodeA.services.networkMapCache.getNodeByLegalIdentityKey(nodeA.info.legalIdentity.owningKey), nodeB.info)
|
||||
}
|
||||
@ -68,4 +67,20 @@ class InMemoryNetworkMapCacheTest {
|
||||
|
||||
// TODO: Should have a test case with anonymous lookup
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `remove node from cache`() {
|
||||
val nodes = mockNet.createSomeNodes(1)
|
||||
val n0 = nodes.mapNode
|
||||
val n1 = nodes.partyNodes[0]
|
||||
val node0Cache = n0.services.networkMapCache as PersistentNetworkMapCache
|
||||
mockNet.runNetwork()
|
||||
n0.database.transaction {
|
||||
assertThat(node0Cache.getNodeByLegalIdentity(n1.info.legalIdentity) != null)
|
||||
node0Cache.removeNode(n1.info)
|
||||
assertThat(node0Cache.getNodeByLegalIdentity(n1.info.legalIdentity) == null)
|
||||
assertThat(node0Cache.getNodeByLegalIdentity(n0.info.legalIdentity) != null)
|
||||
assertThat(node0Cache.getNodeByLegalName(n1.info.legalIdentity.name) == null)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,194 @@
|
||||
package net.corda.node.services.network
|
||||
|
||||
import co.paralleluniverse.fibers.Suspendable
|
||||
import net.corda.core.crypto.toBase58String
|
||||
import net.corda.core.flows.FlowLogic
|
||||
import net.corda.core.flows.InitiatedBy
|
||||
import net.corda.core.flows.InitiatingFlow
|
||||
import net.corda.core.identity.Party
|
||||
import net.corda.core.node.NodeInfo
|
||||
import net.corda.core.utilities.NetworkHostAndPort
|
||||
import net.corda.core.utilities.getOrThrow
|
||||
import net.corda.core.utilities.seconds
|
||||
import net.corda.core.utilities.unwrap
|
||||
import net.corda.node.internal.Node
|
||||
import net.corda.testing.ALICE
|
||||
import net.corda.testing.BOB
|
||||
import net.corda.testing.CHARLIE
|
||||
import net.corda.testing.DUMMY_NOTARY
|
||||
import net.corda.testing.node.NodeBasedTest
|
||||
import org.assertj.core.api.Assertions.assertThat
|
||||
import org.bouncycastle.asn1.x500.X500Name
|
||||
import org.junit.Before
|
||||
import org.junit.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFails
|
||||
|
||||
class PersistentNetworkMapCacheTest : NodeBasedTest() {
|
||||
val partiesList = listOf(DUMMY_NOTARY, ALICE, BOB)
|
||||
val addressesMap: HashMap<X500Name, NetworkHostAndPort> = HashMap()
|
||||
val infos: MutableSet<NodeInfo> = HashSet()
|
||||
|
||||
@Before
|
||||
fun start() {
|
||||
val nodes = startNodesWithPort(partiesList)
|
||||
nodes.forEach { it.nodeReadyFuture.get() } // Need to wait for network map registration, as these tests are ran without waiting.
|
||||
nodes.forEach {
|
||||
infos.add(it.info)
|
||||
addressesMap[it.info.legalIdentity.name] = it.info.addresses[0]
|
||||
it.stop() // We want them to communicate with NetworkMapService to save data to cache.
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `get nodes by owning key and by name, no network map service`() {
|
||||
val alice = startNodesWithPort(listOf(ALICE), noNetworkMap = true)[0]
|
||||
val netCache = alice.services.networkMapCache as PersistentNetworkMapCache
|
||||
alice.database.transaction {
|
||||
val res = netCache.getNodeByLegalIdentity(alice.info.legalIdentity)
|
||||
assertEquals(alice.info, res)
|
||||
val res2 = netCache.getNodeByLegalName(DUMMY_NOTARY.name)
|
||||
assertEquals(infos.filter { it.legalIdentity.name == DUMMY_NOTARY.name }.singleOrNull(), res2)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `get nodes by address no network map service`() {
|
||||
val alice = startNodesWithPort(listOf(ALICE), noNetworkMap = true)[0]
|
||||
val netCache = alice.services.networkMapCache as PersistentNetworkMapCache
|
||||
alice.database.transaction {
|
||||
val res = netCache.getNodeByAddress(alice.info.addresses[0])
|
||||
assertEquals(alice.info, res)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `restart node with DB map cache and no network map`() {
|
||||
val alice = startNodesWithPort(listOf(ALICE), noNetworkMap = true)[0]
|
||||
val partyNodes = alice.services.networkMapCache.partyNodes
|
||||
assert(NetworkMapService.type !in alice.info.advertisedServices.map { it.info.type })
|
||||
assertEquals(null, alice.inNodeNetworkMapService)
|
||||
assertEquals(infos.size, partyNodes.size)
|
||||
assertEquals(infos.map { it.legalIdentity }.toSet(), partyNodes.map { it.legalIdentity }.toSet())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `start 2 nodes without pointing at NetworkMapService and communicate with each other`() {
|
||||
val parties = partiesList.subList(1, partiesList.size)
|
||||
val nodes = startNodesWithPort(parties, noNetworkMap = true)
|
||||
assert(nodes.all { it.inNodeNetworkMapService == null })
|
||||
assert(nodes.all { NetworkMapService.type !in it.info.advertisedServices.map { it.info.type } })
|
||||
nodes.forEach {
|
||||
val partyNodes = it.services.networkMapCache.partyNodes
|
||||
assertEquals(infos.size, partyNodes.size)
|
||||
assertEquals(infos.map { it.legalIdentity }.toSet(), partyNodes.map { it.legalIdentity }.toSet())
|
||||
}
|
||||
checkConnectivity(nodes)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `start 2 nodes pointing at NetworkMapService but don't start network map node`() {
|
||||
val parties = partiesList.subList(1, partiesList.size)
|
||||
val nodes = startNodesWithPort(parties, noNetworkMap = false)
|
||||
assert(nodes.all { it.inNodeNetworkMapService == null })
|
||||
assert(nodes.all { NetworkMapService.type !in it.info.advertisedServices.map { it.info.type } })
|
||||
nodes.forEach {
|
||||
val partyNodes = it.services.networkMapCache.partyNodes
|
||||
assertEquals(infos.size, partyNodes.size)
|
||||
assertEquals(infos.map { it.legalIdentity }.toSet(), partyNodes.map { it.legalIdentity }.toSet())
|
||||
}
|
||||
checkConnectivity(nodes)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `start node and network map communicate`() {
|
||||
val parties = partiesList.subList(0, 2)
|
||||
val nodes = startNodesWithPort(parties, noNetworkMap = false)
|
||||
checkConnectivity(nodes)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `start node without networkMapService and no database - fail`() {
|
||||
assertFails { startNode(CHARLIE.name, noNetworkMap = true).getOrThrow(2.seconds) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `new node joins network without network map started`() {
|
||||
val parties = partiesList.subList(1, partiesList.size)
|
||||
// Start 2 nodes pointing at network map, but don't start network map service.
|
||||
val otherNodes = startNodesWithPort(parties, noNetworkMap = false)
|
||||
otherNodes.forEach { node ->
|
||||
assert(infos.any { it.legalIdentity == node.info.legalIdentity })
|
||||
}
|
||||
// Start node that is not in databases of other nodes. Point to NMS. Which has't started yet.
|
||||
val charlie = startNodesWithPort(listOf(CHARLIE), noNetworkMap = false)[0]
|
||||
otherNodes.forEach {
|
||||
assert(charlie.info.legalIdentity !in it.services.networkMapCache.partyNodes.map { it.legalIdentity })
|
||||
}
|
||||
// Start Network Map and see that charlie node appears in caches.
|
||||
val nms = startNodesWithPort(listOf(DUMMY_NOTARY), noNetworkMap = false)[0]
|
||||
nms.startupComplete.get()
|
||||
assert(nms.inNodeNetworkMapService != null)
|
||||
assert(infos.any {it.legalIdentity == nms.info.legalIdentity})
|
||||
otherNodes.forEach {
|
||||
assert(nms.info.legalIdentity in it.services.networkMapCache.partyNodes.map { it.legalIdentity })
|
||||
}
|
||||
charlie.nodeReadyFuture.get() // Finish registration.
|
||||
checkConnectivity(listOf(otherNodes[0], nms)) // Checks connectivity from A to NMS.
|
||||
val cacheA = otherNodes[0].services.networkMapCache.partyNodes
|
||||
val cacheB = otherNodes[1].services.networkMapCache.partyNodes
|
||||
val cacheC = charlie.services.networkMapCache.partyNodes
|
||||
assertEquals(4, cacheC.size) // Charlie fetched data from NetworkMap
|
||||
assert(charlie.info.legalIdentity in cacheB.map { it.legalIdentity }) // Other nodes also fetched data from Network Map with node C.
|
||||
assertEquals(cacheA.toSet(), cacheB.toSet())
|
||||
assertEquals(cacheA.toSet(), cacheC.toSet())
|
||||
}
|
||||
|
||||
// HELPERS
|
||||
// Helper function to restart nodes with the same host and port.
|
||||
private fun startNodesWithPort(nodesToStart: List<Party>, noNetworkMap: Boolean = false): List<Node> {
|
||||
return nodesToStart.map { party ->
|
||||
val configOverrides = addressesMap[party.name]?.let { mapOf("p2pAddress" to it.toString()) } ?: emptyMap()
|
||||
if (party == DUMMY_NOTARY) {
|
||||
startNetworkMapNode(party.name, configOverrides = configOverrides)
|
||||
}
|
||||
else {
|
||||
startNode(party.name,
|
||||
configOverrides = configOverrides,
|
||||
noNetworkMap = noNetworkMap,
|
||||
waitForConnection = false).getOrThrow()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check that nodes are functional, communicate each with each.
|
||||
private fun checkConnectivity(nodes: List<Node>) {
|
||||
nodes.forEach { node1 ->
|
||||
nodes.forEach { node2 ->
|
||||
node2.registerInitiatedFlow(SendBackFlow::class.java)
|
||||
val resultFuture = node1.services.startFlow(SendFlow(node2.info.legalIdentity)).resultFuture
|
||||
assertThat(resultFuture.getOrThrow()).isEqualTo("Hello!")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@InitiatingFlow
|
||||
private class SendFlow(val otherParty: Party) : FlowLogic<String>() {
|
||||
@Suspendable
|
||||
override fun call(): String {
|
||||
println("SEND FLOW to $otherParty")
|
||||
println("Party key ${otherParty.owningKey.toBase58String()}")
|
||||
return sendAndReceive<String>(otherParty, "Hi!").unwrap { it }
|
||||
}
|
||||
}
|
||||
|
||||
@InitiatedBy(SendFlow::class)
|
||||
private class SendBackFlow(val otherParty: Party) : FlowLogic<Unit>() {
|
||||
@Suspendable
|
||||
override fun call() {
|
||||
println("SEND BACK FLOW to $otherParty")
|
||||
println("Party key ${otherParty.owningKey.toBase58String()}")
|
||||
send(otherParty, "Hello!")
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -220,6 +220,7 @@ class FlowFrameworkTests {
|
||||
node2.stop()
|
||||
node2.database.transaction {
|
||||
assertEquals(1, node2.checkpointStorage.checkpoints().size) // confirm checkpoint
|
||||
node2.services.networkMapCache.clearNetworkMapCache()
|
||||
}
|
||||
val node2b = mockNet.createNode(node1.network.myAddress, node2.id, advertisedServices = *node2.advertisedServices.toTypedArray())
|
||||
node2.manuallyCloseDB()
|
||||
|
||||
Reference in New Issue
Block a user