ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2024-12-18 20:47:57 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

CORDA-990: Flows Draining Mode (#2483)

Browse Source 
[CORDA-990]: Flows draining mode.
This commit is contained in:
Michele Sollecito 2018-02-12 16:13:43 +00:00 committed by GitHub
parent fa4b5d16ba
commit da591363fd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
38 changed files with 1044 additions and 183 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
.ci/api-current.txt
View File

@ -1571,6 +1571,7 @@ public final class net.corda.core.identity.IdentityUtils extends java.lang.Objec
@org.jetbrains.annotations.NotNull public abstract Iterable getVaultTransactionNotes(net.corda.core.crypto.SecureHash)
@kotlin.Deprecated @net.corda.core.messaging.RPCReturnsObservables @org.jetbrains.annotations.NotNull public abstract net.corda.core.messaging.DataFeed internalVerifiedTransactionsFeed()
@kotlin.Deprecated @org.jetbrains.annotations.NotNull public abstract List internalVerifiedTransactionsSnapshot()
public abstract boolean isFlowsDrainingModeEnabled()
@net.corda.core.messaging.RPCReturnsObservables @org.jetbrains.annotations.NotNull public abstract net.corda.core.messaging.DataFeed networkMapFeed()
@org.jetbrains.annotations.NotNull public abstract List networkMapSnapshot()
@net.corda.core.messaging.RPCReturnsObservables @org.jetbrains.annotations.NotNull public abstract net.corda.core.messaging.DataFeed networkParametersFeed()
@ -1583,6 +1584,7 @@ public final class net.corda.core.identity.IdentityUtils extends java.lang.Objec
@org.jetbrains.annotations.Nullable public abstract net.corda.core.identity.Party partyFromKey(java.security.PublicKey)
@org.jetbrains.annotations.NotNull public abstract List queryAttachments(net.corda.core.node.services.vault.AttachmentQueryCriteria, net.corda.core.node.services.vault.AttachmentSort)
@org.jetbrains.annotations.NotNull public abstract List registeredFlows()
public abstract void setFlowsDrainingModeEnabled(boolean)
@net.corda.core.messaging.RPCReturnsObservables @org.jetbrains.annotations.NotNull public abstract net.corda.core.messaging.DataFeed stateMachineRecordedTransactionMappingFeed()
@org.jetbrains.annotations.NotNull public abstract List stateMachineRecordedTransactionMappingSnapshot()
@net.corda.core.messaging.RPCReturnsObservables @org.jetbrains.annotations.NotNull public abstract net.corda.core.messaging.DataFeed stateMachinesFeed()
@ -3901,13 +3903,13 @@ public class net.corda.testing.node.MessagingServiceSpy extends java.lang.Object
@org.jetbrains.annotations.NotNull public net.corda.core.messaging.SingleMessageRecipient getMyAddress()
public void removeMessageHandler(net.corda.node.services.messaging.MessageHandlerRegistration)
@co.paralleluniverse.fibers.Suspendable public void send(List)
@co.paralleluniverse.fibers.Suspendable public void send(net.corda.node.services.messaging.Message, net.corda.core.messaging.MessageRecipients, Long, Object)
@co.paralleluniverse.fibers.Suspendable public void send(net.corda.node.services.messaging.Message, net.corda.core.messaging.MessageRecipients, Long, Object, Map)
##
public final class net.corda.testing.node.MockKeyManagementService extends net.corda.core.serialization.SingletonSerializeAsToken implements net.corda.core.node.services.KeyManagementService
@org.jetbrains.annotations.NotNull public Iterable filterMyKeys(Iterable)
@org.jetbrains.annotations.NotNull public java.security.PublicKey freshKey()
@org.jetbrains.annotations.NotNull public net.corda.core.identity.PartyAndCertificate freshKeyAndCert(net.corda.core.identity.PartyAndCertificate, boolean)
@org.jetbrains.annotations.NotNull public final net.corda.node.services.api.IdentityServiceInternal getIdentityService()
@org.jetbrains.annotations.NotNull public final net.corda.core.node.services.IdentityService getIdentityService()
@org.jetbrains.annotations.NotNull public Set getKeys()
@org.jetbrains.annotations.NotNull public net.corda.core.crypto.TransactionSignature sign(net.corda.core.crypto.SignableData, java.security.PublicKey)
@org.jetbrains.annotations.NotNull public net.corda.core.crypto.DigitalSignature$WithKey sign(byte[], java.security.PublicKey)
@ -4000,9 +4002,9 @@ public final class net.corda.testing.node.MockNodeParameters extends java.lang.O
##
public class net.corda.testing.node.MockServices extends java.lang.Object implements net.corda.core.node.StateLoader, net.corda.core.node.ServiceHub
public <init>(List, net.corda.core.identity.CordaX500Name)
public <init>(List, net.corda.node.services.api.IdentityServiceInternal, net.corda.core.identity.CordaX500Name)
public <init>(List, net.corda.core.node.services.IdentityService, net.corda.core.identity.CordaX500Name)
public <init>(net.corda.core.identity.CordaX500Name)
public <init>(net.corda.node.services.api.IdentityServiceInternal, net.corda.core.identity.CordaX500Name)
public <init>(net.corda.core.node.services.IdentityService, net.corda.core.identity.CordaX500Name)
@org.jetbrains.annotations.NotNull public net.corda.core.transactions.SignedTransaction addSignature(net.corda.core.transactions.SignedTransaction)
@org.jetbrains.annotations.NotNull public net.corda.core.transactions.SignedTransaction addSignature(net.corda.core.transactions.SignedTransaction, java.security.PublicKey)
@org.jetbrains.annotations.NotNull public net.corda.core.serialization.SerializeAsToken cordaService(Class)
@ -4015,7 +4017,7 @@ public class net.corda.testing.node.MockServices extends java.lang.Object implem
@org.jetbrains.annotations.NotNull public net.corda.core.node.services.ContractUpgradeService getContractUpgradeService()
@org.jetbrains.annotations.NotNull public net.corda.core.cordapp.CordappProvider getCordappProvider()
@org.jetbrains.annotations.NotNull public final com.google.common.collect.MutableClassToInstanceMap getCordappServices()
@org.jetbrains.annotations.NotNull public net.corda.node.services.api.IdentityServiceInternal getIdentityService()
@org.jetbrains.annotations.NotNull public net.corda.core.node.services.IdentityService getIdentityService()
@org.jetbrains.annotations.NotNull public net.corda.core.node.services.KeyManagementService getKeyManagementService()
@org.jetbrains.annotations.NotNull public static final net.corda.node.VersionInfo getMOCK_VERSION_INFO()
@org.jetbrains.annotations.NotNull public final net.corda.testing.services.MockCordappProvider getMockCordappProvider()

109
client/rpc/src/integration-test/kotlin/net/corda/client/rpc/FlowsExecutionModeRpcTest.kt Normal file
View File

@ -0,0 +1,109 @@
package net.corda.client.rpc
import net.corda.core.context.Actor
import net.corda.core.context.Trace
import net.corda.core.internal.packageName
import net.corda.core.messaging.CordaRPCOps
import net.corda.core.utilities.getOrThrow
import net.corda.finance.schemas.CashSchemaV1
import net.corda.node.internal.Node
import net.corda.node.internal.StartedNode
import net.corda.node.services.Permissions
import net.corda.node.services.Permissions.Companion.invokeRpc
import net.corda.testing.core.*
import net.corda.testing.driver.NodeHandle
import net.corda.testing.driver.driver
import net.corda.testing.node.User
import net.corda.testing.node.internal.NodeBasedTest
import org.assertj.core.api.Assertions.assertThat
import org.junit.Assume.assumeFalse
import org.junit.Before
import org.junit.Test
class FlowsExecutionModeRpcTest {
@Test
fun `persistent state survives node restart`() {
// Temporary disable this test when executed on Windows. It is known to be sporadically failing.
// More investigation is needed to establish why.
assumeFalse(System.getProperty("os.name").toLowerCase().startsWith("win"))
val user = User("mark", "dadada", setOf(invokeRpc("setFlowsDrainingModeEnabled"), invokeRpc("isFlowsDrainingModeEnabled")))
driver(isDebug = true, startNodesInProcess = true) {
val nodeName = {
val nodeHandle = startNode(rpcUsers = listOf(user)).getOrThrow()
val nodeName = nodeHandle.nodeInfo.chooseIdentity().name
nodeHandle.rpc.setFlowsDrainingModeEnabled(true)
nodeHandle.stop()
nodeName
}()
val nodeHandle = startNode(providedName = nodeName, rpcUsers = listOf(user)).getOrThrow()
assertThat(nodeHandle.rpc.isFlowsDrainingModeEnabled()).isEqualTo(true)
nodeHandle.stop()
}
}
}
class FlowsExecutionModeTests : NodeBasedTest(listOf("net.corda.finance.contracts", CashSchemaV1::class.packageName)) {
private val rpcUser = User("user1", "test", permissions = setOf(Permissions.all()))
private lateinit var node: StartedNode<Node>
private lateinit var client: CordaRPCClient
@Before
fun setup() {
node = startNode(ALICE_NAME, rpcUsers = listOf(rpcUser))
client = CordaRPCClient(node.internals.configuration.rpcOptions.address!!)
}
@Test
fun `flows draining mode can be enabled and queried`() {
asALoggerUser { rpcOps ->
val newValue = true
rpcOps.setFlowsDrainingModeEnabled(true)
val flowsExecutionMode = rpcOps.isFlowsDrainingModeEnabled()
assertThat(flowsExecutionMode).isEqualTo(newValue)
}
}
@Test
fun `flows draining mode can be disabled and queried`() {
asALoggerUser { rpcOps ->
rpcOps.setFlowsDrainingModeEnabled(true)
val newValue = false
rpcOps.setFlowsDrainingModeEnabled(newValue)
val flowsExecutionMode = rpcOps.isFlowsDrainingModeEnabled()
assertThat(flowsExecutionMode).isEqualTo(newValue)
}
}
@Test
fun `node starts with flows draining mode disabled`() {
asALoggerUser { rpcOps ->
val defaultStartingMode = rpcOps.isFlowsDrainingModeEnabled()
assertThat(defaultStartingMode).isEqualTo(false)
}
}
private fun login(username: String, password: String, externalTrace: Trace? = null, impersonatedActor: Actor? = null): CordaRPCConnection {
return client.start(username, password, externalTrace, impersonatedActor)
}
private fun asALoggerUser(action: (CordaRPCOps) -> Unit) {
login(rpcUser.username, rpcUser.password).use {
action(it.proxy)
}
}
}

18
core/src/main/kotlin/net/corda/core/messaging/CordaRPCOps.kt
View File

@ -1,10 +1,7 @@
package net.corda.core.messaging
import net.corda.core.concurrent.CordaFuture
import net.corda.core.context.Actor
import net.corda.core.context.AuthServiceId
import net.corda.core.context.InvocationContext
import net.corda.core.context.InvocationOrigin
import net.corda.core.contracts.ContractState
import net.corda.core.crypto.SecureHash
import net.corda.core.flows.FlowInitiator
@ -353,6 +350,21 @@ interface CordaRPCOps : RPCOps {
/** Clear all network map data from local node cache. */
fun clearNetworkMapCache()
/** Sets the value of the node's flows draining mode.
* If this mode is [enabled], the node will reject new flows through RPC, ignore scheduled flows, and do not process
* initial session messages, meaning that P2P counter-parties will not be able to initiate new flows involving the node.
*
* @param enabled whether the flows draining mode will be enabled.
* */
fun setFlowsDrainingModeEnabled(enabled: Boolean)
/**
* Returns whether the flows draining mode is enabled.
*
* @see setFlowsDrainingModeEnabled
*/
fun isFlowsDrainingModeEnabled(): Boolean
}
inline fun <reified T : ContractState> CordaRPCOps.vaultQueryBy(criteria: QueryCriteria = QueryCriteria.VaultQueryCriteria(),

1
core/src/main/kotlin/net/corda/core/utilities/KotlinUtils.kt
View File

@ -5,6 +5,7 @@ import net.corda.core.internal.uncheckedCast
import net.corda.core.serialization.CordaSerializable
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import rx.Observable
import java.time.Duration
import java.util.concurrent.ExecutionException
import java.util.concurrent.Future

5
docs/source/changelog.rst
View File

@ -6,6 +6,11 @@ from the previous milestone release.
UNRELEASED
----------
* Introduced Flow Draining mode, in which a node continues executing existing flows, but does not start new. This is to support graceful node shutdown/restarts.
In particular, when this mode is on, new flows through RPC will be rejected, scheduled flows will be ignored, and initial session messages will not be consumed.
This will ensure that the number of checkpoints will strictly diminish with time, allowing for a clean shutdown.
* Make the serialisation finger-printer a pluggable entity rather than hard wiring into the factory
* Removed blacklisted word checks in Corda X.500 name to allow "Server" or "Node" to be use as part of the legal name.

4
docs/source/example-code/src/main/kotlin/net/corda/docs/tutorial/mocknetwork/TutorialMockNetwork.kt
View File

@ -79,10 +79,10 @@ class TutorialMockNetwork {
// DOCSTART 1
// modify message if it's 1
nodeB.setMessagingServiceSpy(object : MessagingServiceSpy(nodeB.network) {
override fun send(message: Message, target: MessageRecipients, retryId: Long?, sequenceKey: Any) {
override fun send(message: Message, target: MessageRecipients, retryId: Long?, sequenceKey: Any, additionalHeaders: Map<String, String>) {
val messageData = message.data.deserialize<Any>() as? ExistingSessionMessage
val payload = messageData?.payload
if (payload is DataSessionMessage && payload.payload.deserialize() == 1) {
val alteredMessageData = messageData.copy(payload = payload.copy(99.serialize())).serialize().bytes
messagingService.send(InMemoryMessagingNetwork.InMemoryMessage(message.topic, OpaqueBytes(alteredMessageData), message.uniqueMessageId), target, retryId)

16
docs/source/key-concepts-node.rst
View File

@ -76,4 +76,18 @@ The node also has several CorDapps installed by default to handle common tasks s
* Retrieving transactions and attachments from counterparties
* Upgrading contracts
* Broadcasting agreed ledger updates for recording by counterparties
* Broadcasting agreed ledger updates for recording by counterparties
Draining mode
^^^^^^^^^^^^^
In order to operate a clean shutdown of a node, it is important than no flows are in-flight, meaning no checkpoints should
be persisted. The node is able to be put in a Flows Draining Mode, during which:
* Commands requiring to start new flows through RPC will be rejected.
* Scheduled flows due will be ignored.
* Initial P2P session messages will not be processed, meaning peers will not be able to initiate new flows involving the node.
* All other activities will proceed as usual, ensuring that the number of in-flight flows will strictly diminish.
As their number - which can be monitored through RPC - reaches zero, it is safe to shut the node down.
This property is durable, meaning that restarting the node will not reset it to its default value and that a RPC command is required.

2
gradle-plugins/api-scanner/README.md
View File

@ -10,7 +10,7 @@ See [here](../../docs/source/corda-api.rst) for Corda's public API strategy. We
apply this plugin to other modules in future Corda releases as those modules' APIs stabilise.
Basically, this plugin will document a module's `public` and `protected` classes/methods/fields,
excluding those from our `*.internal.*` packgages, any synthetic methods, bridge methods, or methods
excluding those from our `*.internal.*` packages, any synthetic methods, bridge methods, or methods
identified as having Kotlin's `internal` scope. (Kotlin doesn't seem to have implemented `internal`
scope for classes or fields yet as these are currently `public` inside the `.class` file.)

8
node-api/src/main/kotlin/net/corda/nodeapi/exceptions/RejectedCommandException.kt Normal file
View File

@ -0,0 +1,8 @@
package net.corda.nodeapi.exceptions
import net.corda.core.CordaRuntimeException
/**
* Thrown to indicate that the command was rejected by the node, typically due to a special temporary mode.
*/
class RejectedCommandException(msg: String) : CordaRuntimeException(msg)

100
node/src/integration-test/kotlin/net/corda/node/modes/draining/P2PFlowsDrainingModeTest.kt Normal file
View File

@ -0,0 +1,100 @@
package net.corda.node.modes.draining
import co.paralleluniverse.fibers.Suspendable
import net.corda.core.flows.*
import net.corda.core.identity.Party
import net.corda.core.internal.concurrent.map
import net.corda.core.messaging.startFlow
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.loggerFor
import net.corda.core.utilities.unwrap
import net.corda.node.services.Permissions
import net.corda.testing.core.chooseIdentity
import net.corda.testing.driver.PortAllocation
import net.corda.testing.driver.driver
import net.corda.testing.node.User
import org.assertj.core.api.AssertionsForInterfaceTypes.assertThat
import org.junit.After
import org.junit.Before
import org.junit.Test
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import kotlin.test.fail
class P2PFlowsDrainingModeTest {
private val portAllocation = PortAllocation.Incremental(10000)
private val user = User("mark", "dadada", setOf(Permissions.all()))
private val users = listOf(user)
private var executor: ExecutorService? = null
companion object {
private val logger = loggerFor<P2PFlowsDrainingModeTest>()
}
@Before
fun setup() {
executor = Executors.newSingleThreadExecutor()
}
@After
fun cleanUp() {
executor!!.shutdown()
}
@Test
fun `flows draining mode suspends consumption of initial session messages`() {
driver(isDebug = true, startNodesInProcess = false, portAllocation = portAllocation) {
val initiatedNode = startNode().getOrThrow()
val initiating = startNode(rpcUsers = users).getOrThrow().rpc
val counterParty = initiatedNode.nodeInfo.chooseIdentity()
val initiated = initiatedNode.rpc
initiated.setFlowsDrainingModeEnabled(true)
var shouldFail = true
initiating.apply {
val flow = startFlow(::InitiateSessionFlow, counterParty)
// this should be really fast, for the flow has already started, so 5 seconds should never be a problem
executor!!.submit({
logger.info("Now disabling flows draining mode for $counterParty.")
shouldFail = false
initiated.setFlowsDrainingModeEnabled(false)
})
flow.returnValue.map { result ->
if (shouldFail) {
fail("Shouldn't happen until flows draining mode is switched off.")
} else {
assertThat(result).isEqualTo("Hi there answer")
}
}.getOrThrow()
}
}
}
@StartableByRPC
@InitiatingFlow
class InitiateSessionFlow(private val counterParty: Party) : FlowLogic<String>() {
@Suspendable
override fun call(): String {
val session = initiateFlow(counterParty)
session.send("Hi there")
return session.receive<String>().unwrap { it }
}
}
@InitiatedBy(InitiateSessionFlow::class)
class InitiatedFlow(private val initiatingSession: FlowSession) : FlowLogic<Unit>() {
@Suspendable
override fun call() {
val message = initiatingSession.receive<String>().unwrap { it }
initiatingSession.send("$message answer")
}
}
}

49
node/src/integration-test/kotlin/net/corda/node/modes/draining/RpcFlowsDrainingModeTest.kt Normal file
View File

@ -0,0 +1,49 @@
package net.corda.node.modes.draining
import co.paralleluniverse.fibers.Suspendable
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.StartableByRPC
import net.corda.core.messaging.startFlow
import net.corda.core.utilities.getOrThrow
import net.corda.node.services.Permissions
import net.corda.nodeapi.exceptions.RejectedCommandException
import net.corda.testing.driver.PortAllocation
import net.corda.testing.driver.driver
import net.corda.testing.node.User
import org.assertj.core.api.Assertions.assertThat
import org.assertj.core.api.Assertions.catchThrowable
import org.junit.Test
class RpcFlowsDrainingModeTest {
private val portAllocation = PortAllocation.Incremental(10000)
private val user = User("mark", "dadada", setOf(Permissions.all()))
private val users = listOf(user)
@Test
fun `flows draining mode rejects start flows commands through rpc`() {
driver(isDebug = true, startNodesInProcess = false, portAllocation = portAllocation) {
startNode(rpcUsers = users).getOrThrow().rpc.apply {
setFlowsDrainingModeEnabled(true)
val error: Throwable? = catchThrowable { startFlow(RpcFlowsDrainingModeTest::NoOpFlow) }
assertThat(error).isNotNull()
assertThat(error!!).isInstanceOf(RejectedCommandException::class.java)
}
}
}
@StartableByRPC
class NoOpFlow : FlowLogic<Unit>() {
@Suspendable
override fun call() {
println("NO OP!")
}
}
}

2
node/src/integration-test/kotlin/net/corda/test/node/NodeStatePersistenceTests.kt
View File

@ -33,7 +33,7 @@ import javax.persistence.Table
import kotlin.test.assertEquals
import kotlin.test.assertNotNull
class NodeStatePersistenceTests {
class ONodeStatePersistenceTests {
@Test
fun `persistent state survives node restart`() {
// Temporary disable this test when executed on Windows. It is known to be sporadically failing.

26
node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
View File

@ -204,7 +204,8 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
val (keyPairs, info) = initNodeInfo(networkMapCache, identity, identityKeyPair)
identityService.loadIdentities(info.legalIdentitiesAndCerts)
val transactionStorage = makeTransactionStorage(database, configuration.transactionCacheSizeBytes)
val nodeServices = makeServices(keyPairs, schemaService, transactionStorage, database, info, identityService, networkMapCache)
val nodeProperties = NodePropertiesPersistentStore(StubbedNodeUniqueIdProvider::value, database)
val nodeServices = makeServices(keyPairs, schemaService, transactionStorage, database, info, identityService, networkMapCache, nodeProperties)
val notaryService = makeNotaryService(nodeServices, database)
val smm = makeStateMachineManager(database)
val flowLogicRefFactory = FlowLogicRefFactoryImpl(cordappLoader.appClassLoader)
@ -216,7 +217,9 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
transactionStorage,
unfinishedSchedules = busyNodeLatch,
serverThread = serverThread,
flowLogicRefFactory = flowLogicRefFactory)
flowLogicRefFactory = flowLogicRefFactory,
drainingModePollPeriod = configuration.drainingModePollPeriod,
nodeProperties = nodeProperties)
if (serverThread is ExecutorService) {
runOnStop += {
// We wait here, even though any in-flight messages should have been drained away because the
@ -525,7 +528,14 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
* Builds node internal, advertised, and plugin services.
* Returns a list of tokenizable services to be added to the serialisation context.
*/
private fun makeServices(keyPairs: Set<KeyPair>, schemaService: SchemaService, transactionStorage: WritableTransactionStorage, database: CordaPersistence, info: NodeInfo, identityService: IdentityService, networkMapCache: NetworkMapCacheInternal): MutableList<Any> {
private fun makeServices(keyPairs: Set<KeyPair>,
schemaService: SchemaService,
transactionStorage: WritableTransactionStorage,
database: CordaPersistence,
info: NodeInfo,
identityService: IdentityServiceInternal,
networkMapCache: NetworkMapCacheInternal,
nodeProperties: NodePropertiesStore): MutableList<Any> {
checkpointStorage = DBCheckpointStorage()
val metrics = MetricRegistry()
attachments = NodeAttachmentService(metrics, configuration.attachmentContentCacheSizeBytes, configuration.attachmentCacheBound)
@ -540,8 +550,9 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
cordappProvider,
database,
info,
networkMapCache)
network = makeMessagingService(database, info)
networkMapCache,
nodeProperties)
network = makeMessagingService(database, info, nodeProperties)
val tokenizableServices = mutableListOf(attachments, network, services.vaultService,
services.keyManagementService, services.identityService, platformClock,
services.auditService, services.monitoringService, services.networkMapCache, services.schemaService,
@ -675,7 +686,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
_started = null
}
protected abstract fun makeMessagingService(database: CordaPersistence, info: NodeInfo): MessagingService
protected abstract fun makeMessagingService(database: CordaPersistence, info: NodeInfo, nodeProperties: NodePropertiesStore): MessagingService
protected abstract fun startMessagingService(rpcOps: RPCOps)
private fun obtainIdentity(notaryConfig: NotaryConfig?): Pair<PartyAndCertificate, KeyPair> {
@ -746,7 +757,8 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
override val cordappProvider: CordappProviderInternal,
override val database: CordaPersistence,
override val myInfo: NodeInfo,
override val networkMapCache: NetworkMapCacheInternal
override val networkMapCache: NetworkMapCacheInternal,
override val nodeProperties: NodePropertiesStore
) : SingletonSerializeAsToken(), ServiceHubInternal, StateLoader by validatedTransactions {
override val rpcFlows = ArrayList<Class<out FlowLogic<*>>>()
override val stateMachineRecordedTransactionMapping = DBTransactionMappingStorage()

14
node/src/main/kotlin/net/corda/node/internal/CordaRPCOpsImpl.kt
View File

@ -28,6 +28,7 @@ import net.corda.node.services.api.FlowStarter
import net.corda.node.services.api.ServiceHubInternal
import net.corda.node.services.messaging.context
import net.corda.node.services.statemachine.StateMachineManager
import net.corda.nodeapi.exceptions.RejectedCommandException
import net.corda.nodeapi.internal.persistence.CordaPersistence
import rx.Observable
import java.io.InputStream
@ -165,6 +166,9 @@ internal class CordaRPCOpsImpl(
private fun <T> startFlow(logicType: Class<out FlowLogic<T>>, args: Array<out Any?>): FlowStateMachine<T> {
require(logicType.isAnnotationPresent(StartableByRPC::class.java)) { "${logicType.name} was not designed for RPC" }
if (isFlowsDrainingModeEnabled()) {
throw RejectedCommandException("Node is draining before shutdown. Cannot start new flows through RPC.")
}
return flowStarter.invokeFlowAsync(logicType, context(), *args).getOrThrow()
}
@ -284,6 +288,14 @@ internal class CordaRPCOpsImpl(
return vaultTrackBy(criteria, PageSpecification(), sorting, contractStateType)
}
override fun setFlowsDrainingModeEnabled(enabled: Boolean) {
services.nodeProperties.flowsDrainingMode.setEnabled(enabled)
}
override fun isFlowsDrainingModeEnabled(): Boolean {
return services.nodeProperties.flowsDrainingMode.isEnabled()
}
private fun stateMachineInfoFromFlowLogic(flowLogic: FlowLogic<*>): StateMachineInfo {
return StateMachineInfo(flowLogic.runId, flowLogic.javaClass.name, flowLogic.stateMachine.context.toFlowInitiator(), flowLogic.track(), flowLogic.stateMachine.context)
}
@ -302,7 +314,7 @@ internal class CordaRPCOpsImpl(
is InvocationOrigin.RPC -> FlowInitiator.RPC(principal)
is InvocationOrigin.Peer -> services.identityService.wellKnownPartyFromX500Name((origin as InvocationOrigin.Peer).party)?.let { FlowInitiator.Peer(it) } ?: throw IllegalStateException("Unknown peer with name ${(origin as InvocationOrigin.Peer).party}.")
is InvocationOrigin.Service -> FlowInitiator.Service(principal)
is InvocationOrigin.Shell -> FlowInitiator.Shell
InvocationOrigin.Shell -> FlowInitiator.Shell
is InvocationOrigin.Scheduled -> FlowInitiator.Scheduled((origin as InvocationOrigin.Scheduled).scheduledState)
}
}

13
node/src/main/kotlin/net/corda/node/internal/LifecycleSupport.kt
View File

@ -2,12 +2,23 @@ package net.corda.node.internal
interface LifecycleSupport : Startable, Stoppable
interface Stoppable {
interface Stoppable : AutoCloseable {
fun stop()
override fun close() = stop()
}
interface Startable {
fun start()
val started: Boolean
}
interface Connectable {
val connected: Boolean
fun connect()
fun disconnect()
}

7
node/src/main/kotlin/net/corda/node/internal/Node.kt
View File

@ -22,6 +22,7 @@ import net.corda.node.internal.artemis.BrokerAddresses
import net.corda.node.internal.cordapp.CordappLoader
import net.corda.node.internal.security.RPCSecurityManagerImpl
import net.corda.node.serialization.KryoServerSerializationScheme
import net.corda.node.services.api.NodePropertiesStore
import net.corda.node.services.api.SchemaService
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.config.SecurityConfiguration
@ -144,7 +145,7 @@ open class Node(configuration: NodeConfiguration,
private var shutdownHook: ShutdownHook? = null
override fun makeMessagingService(database: CordaPersistence, info: NodeInfo): MessagingService {
override fun makeMessagingService(database: CordaPersistence, info: NodeInfo, nodeProperties: NodePropertiesStore): MessagingService {
// Construct security manager reading users data either from the 'security' config section
// if present or from rpcUsers list if the former is missing from config.
val securityManagerConfig = configuration.security?.authService ?:
@ -177,7 +178,9 @@ open class Node(configuration: NodeConfiguration,
database,
services.networkMapCache,
advertisedAddress,
networkParameters.maxMessageSize)
networkParameters.maxMessageSize,
isDrainingModeOn = nodeProperties.flowsDrainingMode::isEnabled,
drainingModeWasChangedEvents = nodeProperties.flowsDrainingMode.values)
}
private fun startLocalRpcBroker(): BrokerAddresses? {

13
node/src/main/kotlin/net/corda/node/internal/NodeUniqueIdProvider.kt Normal file
View File

@ -0,0 +1,13 @@
package net.corda.node.internal
interface NodeUniqueIdProvider {
val value: String
}
// this is stubbed because we still do not support clustered node setups.
// the moment we will, this will have to be changed to return a value unique for each physical node.
internal object StubbedNodeUniqueIdProvider : NodeUniqueIdProvider {
// TODO implement to return a value unique for each physical node when we will support clustered node setups.
override val value: String = "NABOB"
}

6
node/src/main/kotlin/net/corda/node/internal/RpcAuthorisationProxy.kt
View File

@ -161,6 +161,12 @@ class RpcAuthorisationProxy(private val implementation: CordaRPCOps, private val
implementation.vaultTrackByWithSorting(contractStateType, criteria, sorting)
}
override fun setFlowsDrainingModeEnabled(enabled: Boolean) = guard("setFlowsDrainingModeEnabled") {
implementation.setFlowsDrainingModeEnabled(enabled)
}
override fun isFlowsDrainingModeEnabled(): Boolean = guard("isFlowsDrainingModeEnabled", implementation::isFlowsDrainingModeEnabled)
// TODO change to KFunction reference after Kotlin fixes https://youtrack.jetbrains.com/issue/KT-12140
private inline fun <RESULT> guard(methodName: String, action: () -> RESULT) = guard(methodName, emptyList(), action)

94
node/src/main/kotlin/net/corda/node/internal/artemis/ReactiveArtemisConsumer.kt Normal file
View File

@ -0,0 +1,94 @@
package net.corda.node.internal.artemis
import net.corda.node.internal.Connectable
import net.corda.node.internal.LifecycleSupport
import org.apache.activemq.artemis.api.core.client.ClientConsumer
import org.apache.activemq.artemis.api.core.client.ClientMessage
import org.apache.activemq.artemis.api.core.client.ClientSession
import rx.Observable
import rx.subjects.PublishSubject
interface ReactiveArtemisConsumer : LifecycleSupport, Connectable {
val messages: Observable<ClientMessage>
companion object {
fun multiplex(createSession: () -> ClientSession, queueName: String, filter: String? = null, vararg queueNames: String): ReactiveArtemisConsumer {
return MultiplexingReactiveArtemisConsumer(setOf(queueName, *queueNames), createSession, filter)
}
fun multiplex(queueNames: Set<String>, createSession: () -> ClientSession, filter: String? = null): ReactiveArtemisConsumer {
return MultiplexingReactiveArtemisConsumer(queueNames, createSession, filter)
}
}
}
private class MultiplexingReactiveArtemisConsumer(private val queueNames: Set<String>, private val createSession: () -> ClientSession, private val filter: String?) : ReactiveArtemisConsumer {
private var startedFlag = false
override var connected = false
override val messages: PublishSubject<ClientMessage> = PublishSubject.create<ClientMessage>()
private val consumers = mutableSetOf<ClientConsumer>()
private val sessions = mutableSetOf<ClientSession>()
override fun start() {
synchronized(this) {
require(!startedFlag)
connect()
startedFlag = true
}
}
override fun stop() {
synchronized(this) {
if(startedFlag) {
disconnect()
startedFlag = false
}
messages.onCompleted()
}
}
override fun connect() {
synchronized(this) {
require(!connected)
queueNames.forEach { queue ->
createSession().apply {
start()
consumers += filter?.let { createConsumer(queue, it) } ?: createConsumer(queue)
sessions += this
}
}
consumers.forEach { consumer ->
consumer.setMessageHandler { message ->
messages.onNext(message)
}
}
connected = true
}
}
override fun disconnect() {
synchronized(this) {
if(connected) {
consumers.forEach(ClientConsumer::close)
sessions.forEach(ClientSession::close)
consumers.clear()
sessions.clear()
connected = false
}
}
}
override val started: Boolean
get() = startedFlag
}

8
core/src/main/kotlin/net/corda/core/internal/schemas/NodeInfoSchema.kt → node/src/main/kotlin/net/corda/node/internal/schemas/NodeInfoSchema.kt
View File

@ -1,4 +1,4 @@
package net.corda.core.internal.schemas
package net.corda.node.internal.schemas
import net.corda.core.crypto.toStringShort
import net.corda.core.identity.PartyAndCertificate
@ -9,7 +9,7 @@ import net.corda.core.serialization.deserialize
import net.corda.core.serialization.serialize
import net.corda.core.utilities.MAX_HASH_HEX_SIZE
import net.corda.core.utilities.NetworkHostAndPort
import java.io.Serializable
import net.corda.node.services.persistence.NodePropertiesPersistentStore
import javax.persistence.*
object NodeInfoSchema
@ -17,7 +17,7 @@ object NodeInfoSchema
object NodeInfoSchemaV1 : MappedSchema(
schemaFamily = NodeInfoSchema.javaClass,
version = 1,
mappedTypes = listOf(PersistentNodeInfo::class.java, DBPartyAndCertificate::class.java, DBHostAndPort::class.java)
mappedTypes = listOf(PersistentNodeInfo::class.java, DBPartyAndCertificate::class.java, DBHostAndPort::class.java, NodePropertiesPersistentStore.DBNodeProperty::class.java)
) {
@Entity
@Table(name = "node_infos")
@ -33,7 +33,7 @@ object NodeInfoSchemaV1 : MappedSchema(
@Column(name = "addresses")
@OneToMany(cascade = arrayOf(CascadeType.ALL), orphanRemoval = true)
@JoinColumn(name = "node_info_id", foreignKey = ForeignKey(name = "FK__info_hosts__infos"))
val addresses: List<NodeInfoSchemaV1.DBHostAndPort>,
val addresses: List<DBHostAndPort>,
@Column(name = "legal_identities_certs")
@ManyToMany(cascade = arrayOf(CascadeType.ALL))

2
node/src/main/kotlin/net/corda/node/internal/security/RPCSecurityManagerImpl.kt
View File

@ -180,7 +180,7 @@ private object RPCPermissionResolver : PermissionResolver {
// Leaving empty set of targets and actions to match everything
return RPCPermission()
}
else -> throw IllegalArgumentException("Unkwnow permission action specifier: $action")
else -> throw IllegalArgumentException("Unknown permission action specifier: $action")
}
}
}

17
node/src/main/kotlin/net/corda/node/services/api/NodePropertiesStore.kt Normal file
View File

@ -0,0 +1,17 @@
package net.corda.node.services.api
import rx.Observable
interface NodePropertiesStore {
val flowsDrainingMode: FlowsDrainingModeOperations
interface FlowsDrainingModeOperations {
fun setEnabled(enabled: Boolean)
fun isEnabled(): Boolean
val values: Observable<Pair<Boolean, Boolean>>
}
}

1
node/src/main/kotlin/net/corda/node/services/api/ServiceHubInternal.kt
View File

@ -65,6 +65,7 @@ interface ServiceHubInternal : ServiceHub {
val networkService: MessagingService
val database: CordaPersistence
val configuration: NodeConfiguration
val nodeProperties: NodePropertiesStore
val networkMapUpdater: NetworkMapUpdater
override val cordappProvider: CordappProviderInternal
override fun recordTransactions(statesToRecord: StatesToRecord, txs: Iterable<SignedTransaction>) {

3
node/src/main/kotlin/net/corda/node/services/config/NodeConfiguration.kt
View File

@ -16,6 +16,7 @@ import net.corda.nodeapi.internal.config.parseAs
import net.corda.nodeapi.internal.persistence.DatabaseConfig
import java.net.URL
import java.nio.file.Path
import java.time.Duration
import java.util.*
@ -49,6 +50,8 @@ interface NodeConfiguration : NodeSSLConfiguration {
val transactionCacheSizeBytes: Long get() = defaultTransactionCacheSize
val attachmentContentCacheSizeBytes: Long get() = defaultAttachmentContentCacheSize
val attachmentCacheBound: Long get() = defaultAttachmentCacheBound
// do not change this value without syncing it with ScheduledFlowsDrainingModeTest
val drainingModePollPeriod: Duration get() = Duration.ofSeconds(5)
fun validate(): List<String>

23
node/src/main/kotlin/net/corda/node/services/events/NodeSchedulerService.kt
View File

@ -24,12 +24,14 @@ import net.corda.core.utilities.trace
import net.corda.node.internal.CordaClock
import net.corda.node.internal.MutableClock
import net.corda.node.services.api.FlowStarter
import net.corda.node.services.api.NodePropertiesStore
import net.corda.node.services.api.SchedulerService
import net.corda.node.utilities.PersistentMap
import net.corda.nodeapi.internal.persistence.CordaPersistence
import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
import org.apache.activemq.artemis.utils.ReusableLatch
import org.slf4j.Logger
import java.time.Duration
import java.time.Instant
import java.util.*
import java.util.concurrent.*
@ -60,6 +62,8 @@ class NodeSchedulerService(private val clock: CordaClock,
private val unfinishedSchedules: ReusableLatch = ReusableLatch(),
private val serverThread: Executor,
private val flowLogicRefFactory: FlowLogicRefFactory,
private val nodeProperties: NodePropertiesStore,
private val drainingModePollPeriod: Duration,
private val log: Logger = staticLog,
private val scheduledStates: MutableMap<StateRef, ScheduledStateRef> = createMap())
: SchedulerService, SingletonSerializeAsToken() {
@ -285,10 +289,19 @@ class NodeSchedulerService(private val clock: CordaClock,
scheduledStatesQueue.add(newState)
} else {
val flowLogic = flowLogicRefFactory.toFlowLogic(scheduledActivity.logicRef)
log.trace { "Scheduler starting FlowLogic $flowLogic" }
scheduledFlow = flowLogic
scheduledStates.remove(scheduledState.ref)
scheduledStatesQueue.remove(scheduledState)
scheduledFlow = when {
nodeProperties.flowsDrainingMode.isEnabled() -> {
log.warn("Ignoring scheduled flow start because of draining mode. FlowLogic: $flowLogic.")
awaitWithDeadline(clock, Instant.now() + drainingModePollPeriod)
null
}
else -> {
log.trace { "Scheduler starting FlowLogic $flowLogic" }
scheduledStates.remove(scheduledState.ref)
scheduledStatesQueue.remove(scheduledState)
flowLogic
}
}
}
}
// and schedule the next one
@ -308,4 +321,4 @@ class NodeSchedulerService(private val clock: CordaClock,
null
}
}
}
}

13
node/src/main/kotlin/net/corda/node/services/messaging/Messaging.kt
View File

@ -66,7 +66,8 @@ interface MessagingService {
message: Message,
target: MessageRecipients,
retryId: Long? = null,
sequenceKey: Any = target
sequenceKey: Any = target,
additionalHeaders: Map<String, String> = emptyMap()
)
/** A message with a target and sequenceKey specified. */
@ -92,6 +93,8 @@ interface MessagingService {
/**
* Returns an initialised [Message] with the current time, etc, already filled in.
*
* @param topicSession identifier for the topic and session the message is sent to.
* @param additionalProperties optional additional message headers.
* @param topic identifier for the topic the message is sent to.
*/
fun createMessage(topic: String, data: ByteArray, deduplicationId: String = UUID.randomUUID().toString()): Message
@ -143,3 +146,11 @@ object TopicStringValidator {
/** @throws IllegalArgumentException if the given topic contains invalid characters */
fun check(tag: String) = require(regex.matcher(tag).matches())
}
object P2PMessagingHeaders {
object Type {
const val KEY = "corda_p2p_message_type"
const val SESSION_INIT_VALUE = "session_init"
}
}

353
node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt
View File

@ -12,16 +12,21 @@ import net.corda.core.node.services.PartyInfo
import net.corda.core.serialization.SerializationDefaults
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.serialization.deserialize
import net.corda.core.serialization.internal.nodeSerializationEnv
import net.corda.core.serialization.serialize
import net.corda.core.utilities.*
import net.corda.node.VersionInfo
import net.corda.node.internal.LifecycleSupport
import net.corda.node.internal.artemis.ReactiveArtemisConsumer.Companion.multiplex
import net.corda.node.services.api.NetworkMapCacheInternal
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.statemachine.StateMachineManagerImpl
import net.corda.node.utilities.AffinityExecutor
import net.corda.node.utilities.AppendOnlyPersistentMap
import net.corda.node.utilities.PersistentMap
import net.corda.nodeapi.internal.ArtemisMessagingClient
import net.corda.nodeapi.ArtemisTcpTransport
import net.corda.nodeapi.ConnectionDirection
import net.corda.nodeapi.internal.ArtemisMessagingComponent
import net.corda.nodeapi.internal.ArtemisMessagingComponent.*
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_CONTROL
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_NOTIFY
@ -34,11 +39,11 @@ import org.apache.activemq.artemis.api.core.ActiveMQObjectClosedException
import org.apache.activemq.artemis.api.core.Message.*
import org.apache.activemq.artemis.api.core.RoutingType
import org.apache.activemq.artemis.api.core.SimpleString
import org.apache.activemq.artemis.api.core.client.ClientConsumer
import org.apache.activemq.artemis.api.core.client.ClientMessage
import org.apache.activemq.artemis.api.core.client.ClientSession
import org.apache.activemq.artemis.api.core.client.*
import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
import rx.Observable
import rx.Subscription
import rx.subjects.PublishSubject
import java.security.PublicKey
import java.time.Instant
import java.util.*
@ -77,17 +82,19 @@ import javax.persistence.Lob
* @param maxMessageSize A bound applied to the message size.
*/
@ThreadSafe
class P2PMessagingClient(config: NodeConfiguration,
class P2PMessagingClient(private val config: NodeConfiguration,
private val versionInfo: VersionInfo,
serverAddress: NetworkHostAndPort,
private val serverAddress: NetworkHostAndPort,
private val myIdentity: PublicKey,
private val serviceIdentity: PublicKey?,
private val nodeExecutor: AffinityExecutor.ServiceAffinityExecutor,
private val database: CordaPersistence,
private val networkMap: NetworkMapCacheInternal,
advertisedAddress: NetworkHostAndPort = serverAddress,
maxMessageSize: Int
) : SingletonSerializeAsToken(), MessagingService {
private val maxMessageSize: Int,
private val isDrainingModeOn: () -> Boolean,
private val drainingModeWasChangedEvents: Observable<Pair<Boolean, Boolean>>
) : SingletonSerializeAsToken(), MessagingService, AutoCloseable {
companion object {
private val log = contextLogger()
// This is a "property" attached to an Artemis MQ message object, which contains our own notion of "topic".
@ -99,7 +106,7 @@ class P2PMessagingClient(config: NodeConfiguration,
private val releaseVersionProperty = SimpleString("release-version")
private val platformVersionProperty = SimpleString("platform-version")
private val amqDelayMillis = System.getProperty("amq.delivery.delay.ms", "0").toInt()
private val messageMaxRetryCount: Int = 3
private const val messageMaxRetryCount: Int = 3
fun createProcessedMessage(): AppendOnlyPersistentMap<String, Instant, ProcessedMessage, String> {
return AppendOnlyPersistentMap(
@ -142,11 +149,18 @@ class P2PMessagingClient(config: NodeConfiguration,
}
private class InnerState {
var started = false
var running = false
var p2pConsumer: ClientConsumer? = null
var serviceConsumer: ClientConsumer? = null
var eventsSubscription: Subscription? = null
var p2pConsumer: P2PMessagingConsumer? = null
var locator: ServerLocator? = null
var producer: ClientProducer? = null
var producerSession: ClientSession? = null
var bridgeSession: ClientSession? = null
var bridgeNotifyConsumer: ClientConsumer? = null
var networkChangeSubscription: Subscription? = null
fun sendMessage(address: String, message: ClientMessage) = producer!!.send(address, message)
}
private val messagesToRedeliver = database.transaction {
@ -166,7 +180,6 @@ class P2PMessagingClient(config: NodeConfiguration,
override val myAddress: SingleMessageRecipient = NodeAddress(myIdentity, advertisedAddress)
private val messageRedeliveryDelaySeconds = config.messageRedeliveryDelaySeconds.toLong()
private val artemis = ArtemisMessagingClient(config, serverAddress, maxMessageSize)
private val state = ThreadBox(InnerState())
private val knownQueues = Collections.newSetFromMap(ConcurrentHashMap<String, Boolean>())
private val handlers = CopyOnWriteArrayList<Handler>()
@ -201,28 +214,45 @@ class P2PMessagingClient(config: NodeConfiguration,
fun start() {
state.locked {
val session = artemis.start().session
val inbox = RemoteInboxAddress(myIdentity).queueName
val inboxes = mutableListOf(inbox)
// Create a queue, consumer and producer for handling P2P network messages.
createQueueIfAbsent(inbox)
p2pConsumer = session.createConsumer(inbox)
if (serviceIdentity != null) {
val serviceAddress = RemoteInboxAddress(serviceIdentity).queueName
inboxes += serviceAddress
createQueueIfAbsent(serviceAddress)
val serviceHandler = session.createConsumer(serviceAddress)
serviceHandler.setMessageHandler { msg ->
val message: ReceivedMessage? = artemisToCordaMessage(msg)
if (message != null)
deliver(message)
state.locked {
msg.individualAcknowledge()
}
}
started = true
log.info("Connecting to message broker: $serverAddress")
// TODO Add broker CN to config for host verification in case the embedded broker isn't used
val tcpTransport = ArtemisTcpTransport.tcpTransport(ConnectionDirection.Outbound(), serverAddress, config)
locator = ActiveMQClient.createServerLocatorWithoutHA(tcpTransport).apply {
// Never time out on our loopback Artemis connections. If we switch back to using the InVM transport this
// would be the default and the two lines below can be deleted.
connectionTTL = -1
clientFailureCheckPeriod = -1
minLargeMessageSize = maxMessageSize
isUseGlobalPools = nodeSerializationEnv != null
}
registerBridgeControl(session, inboxes)
enumerateBridges(session, inboxes)
val sessionFactory = locator!!.createSessionFactory()
// Login using the node username. The broker will authenticate us as its node (as opposed to another peer)
// using our TLS certificate.
// Note that the acknowledgement of messages is not flushed to the Artermis journal until the default buffer
// size of 1MB is acknowledged.
val createNewSession = { sessionFactory!!.createSession(ArtemisMessagingComponent.NODE_USER, ArtemisMessagingComponent.NODE_USER, false, true, true, locator!!.isPreAcknowledge, ActiveMQClient.DEFAULT_ACK_BATCH_SIZE) }
producerSession = createNewSession()
bridgeSession = createNewSession()
producerSession!!.start()
bridgeSession!!.start()
val inboxes = mutableSetOf<String>()
// Create a queue, consumer and producer for handling P2P network messages.
// Create a general purpose producer.
producer = producerSession!!.createProducer()
inboxes += RemoteInboxAddress(myIdentity).queueName
serviceIdentity?.let {
inboxes += RemoteInboxAddress(it).queueName
}
inboxes.forEach { createQueueIfAbsent(it, producerSession!!) }
p2pConsumer = P2PMessagingConsumer(inboxes, createNewSession, isDrainingModeOn, drainingModeWasChangedEvents)
registerBridgeControl(bridgeSession!!, inboxes.toList())
enumerateBridges(bridgeSession!!, inboxes.toList())
}
resumeMessageRedelivery()
@ -240,7 +270,7 @@ class P2PMessagingClient(config: NodeConfiguration,
log.info(notifyMessage.toString())
when (notifyMessage) {
is BridgeControl.BridgeToNodeSnapshotRequest -> enumerateBridges(session, inboxes)
else -> log.error("Unexpected Bridge Control message type on notify topc $notifyMessage")
else -> log.error("Unexpected Bridge Control message type on notify topic $notifyMessage")
}
msg.acknowledge()
}
@ -249,20 +279,23 @@ class P2PMessagingClient(config: NodeConfiguration,
}
private fun sendBridgeControl(message: BridgeControl) {
val client = artemis.started!!
val controlPacket = message.serialize(context = SerializationDefaults.P2P_CONTEXT).bytes
val artemisMessage = client.session.createMessage(false)
artemisMessage.writeBodyBufferBytes(controlPacket)
client.producer.send(BRIDGE_CONTROL, artemisMessage)
state.locked {
val controlPacket = message.serialize(context = SerializationDefaults.P2P_CONTEXT).bytes
val artemisMessage = producerSession!!.createMessage(false)
artemisMessage.writeBodyBufferBytes(controlPacket)
sendMessage(BRIDGE_CONTROL, artemisMessage)
}
}
private fun updateBridgesOnNetworkChange(change: NetworkMapCache.MapChange) {
log.info("Updating bridges on network map change: ${change.node}")
fun gatherAddresses(node: NodeInfo): Sequence<BridgeEntry> {
return node.legalIdentitiesAndCerts.map {
val messagingAddress = NodeAddress(it.party.owningKey, node.addresses.first())
BridgeEntry(messagingAddress.queueName, node.addresses, node.legalIdentities.map { it.name })
}.filter { artemis.started!!.session.queueQuery(SimpleString(it.queueName)).isExists }.asSequence()
return state.locked {
node.legalIdentitiesAndCerts.map {
val messagingAddress = NodeAddress(it.party.owningKey, node.addresses.first())
BridgeEntry(messagingAddress.queueName, node.addresses, node.legalIdentities.map { it.name })
}.filter { producerSession!!.queueQuery(SimpleString(it.queueName)).isExists }.asSequence()
}
}
fun deployBridges(node: NodeInfo) {
@ -315,62 +348,43 @@ class P2PMessagingClient(config: NodeConfiguration,
private fun resumeMessageRedelivery() {
messagesToRedeliver.forEach { retryId, (message, target) ->
send(message, target, retryId)
sendInternal(message, target, retryId)
}
}
private val shutdownLatch = CountDownLatch(1)
private fun processMessage(consumer: ClientConsumer): Boolean {
// Two possibilities here:
//
// 1. We block waiting for a message and the consumer is closed in another thread. In this case
// receive returns null and we break out of the loop.
// 2. We receive a message and process it, and stop() is called during delivery. In this case,
// calling receive will throw and we break out of the loop.
//
// It's safe to call into receive simultaneous with other threads calling send on a producer.
val artemisMessage: ClientMessage = try {
consumer.receive()
} catch (e: ActiveMQObjectClosedException) {
null
} ?: return false
val message: ReceivedMessage? = artemisToCordaMessage(artemisMessage)
if (message != null)
deliver(message)
// Ack the message so it won't be redelivered. We should only really do this when there were no
// transient failures. If we caught an exception in the handler, we could back off and retry delivery
// a few times before giving up and redirecting the message to a dead-letter address for admin or
// developer inspection. Artemis has the features to do this for us, we just need to enable them.
//
// TODO: Setup Artemis delayed redelivery and dead letter addresses.
//
// ACKing a message calls back into the session which isn't thread safe, so we have to ensure it
// doesn't collide with a send here. Note that stop() could have been called whilst we were
// processing a message but if so, it'll be parked waiting for us to count down the latch, so
// the session itself is still around and we can still ack messages as a result.
state.locked {
artemisMessage.acknowledge()
}
return true
}
/**
* Starts the p2p event loop: this method only returns once [stop] has been called.
*/
fun run() {
val latch = CountDownLatch(1)
try {
val consumer = state.locked {
check(artemis.started != null) { "start must be called first" }
check(started) { "start must be called first" }
check(!running) { "run can't be called twice" }
running = true
// If it's null, it means we already called stop, so return immediately.
p2pConsumer ?: return
if (p2pConsumer == null) {
return
}
eventsSubscription = p2pConsumer!!.messages
.doOnError { error -> throw error }
.doOnNext { artemisMessage ->
val receivedMessage = artemisToCordaMessage(artemisMessage)
receivedMessage?.let {
deliver(it)
}
artemisMessage.acknowledge()
}
// this `run()` method is semantically meant to block until the message consumption runs, hence the latch here
.doOnCompleted(latch::countDown)
.subscribe()
p2pConsumer!!
}
while (processMessage(consumer)) { }
consumer.start()
latch.await()
} finally {
shutdownLatch.countDown()
}
@ -458,30 +472,24 @@ class P2PMessagingClient(config: NodeConfiguration,
fun stop() {
val running = state.locked {
// We allow stop() to be called without a run() in between, but it must have at least been started.
check(artemis.started != null)
check(started)
val prevRunning = running
running = false
networkChangeSubscription?.unsubscribe()
val c = p2pConsumer ?: throw IllegalStateException("stop can't be called twice")
try {
c.close()
} catch (e: ActiveMQObjectClosedException) {
// Ignore it: this can happen if the server has gone away before we do.
}
try {
bridgeNotifyConsumer!!.close()
} catch (e: ActiveMQObjectClosedException) {
// Ignore it: this can happen if the server has gone away before we do.
}
require(p2pConsumer != null, {"stop can't be called twice"})
require(producer != null, {"stop can't be called twice"})
close(p2pConsumer)
p2pConsumer = null
val s = serviceConsumer
try {
s?.close()
} catch (e: ActiveMQObjectClosedException) {
// Ignore it: this can happen if the server has gone away before we do.
}
serviceConsumer = null
close(producer)
producer = null
producerSession!!.commit()
close(bridgeNotifyConsumer)
knownQueues.clear()
eventsSubscription?.unsubscribe()
eventsSubscription = null
prevRunning
}
if (running && !nodeExecutor.isOnThread) {
@ -489,21 +497,32 @@ class P2PMessagingClient(config: NodeConfiguration,
shutdownLatch.await()
}
// Only first caller to gets running true to protect against double stop, which seems to happen in some integration tests.
if (running) {
state.locked {
artemis.stop()
}
state.locked {
locator?.close()
}
}
override fun send(message: Message, target: MessageRecipients, retryId: Long?, sequenceKey: Any) {
private fun close(target: AutoCloseable?) {
try {
target?.close()
} catch (ignored: ActiveMQObjectClosedException) {
// swallow
}
}
override fun close() = stop()
override fun send(message: Message, target: MessageRecipients, retryId: Long?, sequenceKey: Any, additionalHeaders: Map<String, String>) {
sendInternal(message, target, retryId, additionalHeaders)
}
private fun sendInternal(message: Message, target: MessageRecipients, retryId: Long?, additionalHeaders: Map<String, String> = emptyMap()) {
// We have to perform sending on a different thread pool, since using the same pool for messaging and
// fibers leads to Netty buffer memory leaks, caused by both Netty and Quasar fiddling with thread-locals.
messagingExecutor.fetchFrom {
state.locked {
val mqAddress = getMQAddress(target)
val artemis = artemis.started!!
val artemisMessage = artemis.session.createMessage(true).apply {
val artemisMessage = producerSession!!.createMessage(true).apply {
putStringProperty(cordaVendorProperty, cordaVendor)
putStringProperty(releaseVersionProperty, releaseVersion)
putIntProperty(platformVersionProperty, versionInfo.platformVersion)
@ -516,11 +535,12 @@ class P2PMessagingClient(config: NodeConfiguration,
if (amqDelayMillis > 0 && message.topic == StateMachineManagerImpl.sessionTopic) {
putLongProperty(HDR_SCHEDULED_DELIVERY_TIME, System.currentTimeMillis() + amqDelayMillis)
}
additionalHeaders.forEach { key, value -> putStringProperty(key, value)}
}
log.trace {
"Send to: $mqAddress topic: ${message.topic} uuid: ${message.uniqueMessageId}"
}
artemis.producer.send(mqAddress, artemisMessage)
sendMessage(mqAddress, artemisMessage)
retryId?.let {
database.transaction {
messagesToRedeliver.computeIfAbsent(it, { Pair(message, target) })
@ -556,7 +576,7 @@ class P2PMessagingClient(config: NodeConfiguration,
state.locked {
log.trace { "Retry #$retryCount sending message $message to $address for $retryId" }
artemis.started!!.producer.send(address, message)
sendMessage(address, message)
}
scheduledMessageRedeliveries[retryId] = messagingExecutor.schedule({
@ -575,6 +595,9 @@ class P2PMessagingClient(config: NodeConfiguration,
}
}
private fun Pair<ClientMessage, ReceivedMessage?>.deliver() = deliver(second!!)
private fun Pair<ClientMessage, ReceivedMessage?>.acknowledge() = first.acknowledge()
private fun getMQAddress(target: MessageRecipients): String {
return if (target == myAddress) {
// If we are sending to ourselves then route the message directly to our P2P queue.
@ -583,28 +606,27 @@ class P2PMessagingClient(config: NodeConfiguration,
// Otherwise we send the message to an internal queue for the target residing on our broker. It's then the
// broker's job to route the message to the target's P2P queue.
val internalTargetQueue = (target as? ArtemisAddress)?.queueName ?: throw IllegalArgumentException("Not an Artemis address")
createQueueIfAbsent(internalTargetQueue)
state.locked {
createQueueIfAbsent(internalTargetQueue, producerSession!!)
}
internalTargetQueue
}
}
/** Attempts to create a durable queue on the broker which is bound to an address of the same name. */
private fun createQueueIfAbsent(queueName: String) {
private fun createQueueIfAbsent(queueName: String, session: ClientSession) {
if (!knownQueues.contains(queueName)) {
state.alreadyLocked {
val session = artemis.started!!.session
val queueQuery = session.queueQuery(SimpleString(queueName))
if (!queueQuery.isExists) {
log.info("Create fresh queue $queueName bound on same address")
session.createQueue(queueName, RoutingType.ANYCAST, queueName, true)
if (queueName.startsWith(PEERS_PREFIX)) {
val keyHash = queueName.substring(PEERS_PREFIX.length)
val peers = networkMap.getNodesByOwningKeyIndex(keyHash)
for (node in peers) {
val bridge = BridgeEntry(queueName, node.addresses, node.legalIdentities.map { it.name })
val createBridgeMessage = BridgeControl.Create(myIdentity.toStringShort(), bridge)
sendBridgeControl(createBridgeMessage)
}
val queueQuery = session.queueQuery(SimpleString(queueName))
if (!queueQuery.isExists) {
log.info("Create fresh queue $queueName bound on same address")
session.createQueue(queueName, RoutingType.ANYCAST, queueName, true)
if (queueName.startsWith(PEERS_PREFIX)) {
val keyHash = queueName.substring(PEERS_PREFIX.length)
val peers = networkMap.getNodesByOwningKeyIndex(keyHash)
for (node in peers) {
val bridge = BridgeEntry(queueName, node.addresses, node.legalIdentities.map { it.name })
val createBridgeMessage = BridgeControl.Create(myIdentity.toStringShort(), bridge)
sendBridgeControl(createBridgeMessage)
}
}
}
@ -637,3 +659,78 @@ class P2PMessagingClient(config: NodeConfiguration,
}
}
}
private class P2PMessagingConsumer(
queueNames: Set<String>,
createSession: () -> ClientSession,
private val isDrainingModeOn: () -> Boolean,
private val drainingModeWasChangedEvents: Observable<Pair<Boolean, Boolean>>) : LifecycleSupport {
private companion object {
private const val initialSessionMessages = "${P2PMessagingHeaders.Type.KEY}=${P2PMessagingHeaders.Type.SESSION_INIT_VALUE}"
private const val existingSessionMessages = "${P2PMessagingHeaders.Type.KEY}<>${P2PMessagingHeaders.Type.SESSION_INIT_VALUE}"
}
private var startedFlag = false
val messages: PublishSubject<ClientMessage> = PublishSubject.create<ClientMessage>()
private var initialConsumer = multiplex(queueNames, createSession, initialSessionMessages)
private var existingConsumer = multiplex(queueNames, createSession, existingSessionMessages)
private val subscriptions = mutableSetOf<Subscription>()
override fun start() {
synchronized(this) {
require(!startedFlag)
drainingModeWasChangedEvents.filter { change -> change.switchedOn() }.doOnNext { pauseInitial() }.subscribe()
drainingModeWasChangedEvents.filter { change -> change.switchedOff() }.doOnNext { resumeInitial() }.subscribe()
subscriptions += initialConsumer.messages.doOnNext(messages::onNext).subscribe()
subscriptions += existingConsumer.messages.doOnNext(messages::onNext).subscribe()
if (!isDrainingModeOn()) {
initialConsumer.start()
}
existingConsumer.start()
startedFlag = true
}
}
override fun stop() {
synchronized(this) {
if (startedFlag) {
initialConsumer.stop()
existingConsumer.stop()
subscriptions.forEach(Subscription::unsubscribe)
subscriptions.clear()
startedFlag = false
}
messages.onCompleted()
}
}
override val started: Boolean
get() = startedFlag
private fun pauseInitial() {
if (initialConsumer.started && initialConsumer.connected) {
initialConsumer.disconnect()
}
}
private fun resumeInitial() {
if(!initialConsumer.started) {
initialConsumer.start()
}
if (!initialConsumer.connected) {
initialConsumer.connect()
}
}
private fun Pair<Boolean, Boolean>.switchedOff() = first && !second
private fun Pair<Boolean, Boolean>.switchedOn() = !first && second
}

2
node/src/main/kotlin/net/corda/node/services/network/PersistentNetworkMapCache.kt
View File

@ -10,7 +10,7 @@ import net.corda.core.identity.PartyAndCertificate
import net.corda.core.node.NotaryInfo
import net.corda.core.internal.bufferUntilSubscribed
import net.corda.core.internal.concurrent.openFuture
import net.corda.core.internal.schemas.NodeInfoSchemaV1
import net.corda.node.internal.schemas.NodeInfoSchemaV1
import net.corda.core.messaging.DataFeed
import net.corda.core.node.NodeInfo
import net.corda.core.node.services.IdentityService

67
node/src/main/kotlin/net/corda/node/services/persistence/NodePropertiesPersistentStore.kt Normal file
View File

@ -0,0 +1,67 @@
package net.corda.node.services.persistence
import net.corda.core.utilities.debug
import net.corda.core.utilities.loggerFor
import net.corda.node.services.api.NodePropertiesStore
import net.corda.node.services.api.NodePropertiesStore.FlowsDrainingModeOperations
import net.corda.node.utilities.PersistentMap
import net.corda.nodeapi.internal.persistence.CordaPersistence
import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
import org.slf4j.Logger
import rx.subjects.PublishSubject
import javax.persistence.Column
import javax.persistence.Entity
import javax.persistence.Id
import javax.persistence.Table
/**
* Simple node properties key value store in DB.
*/
class NodePropertiesPersistentStore(readPhysicalNodeId: () -> String, persistence: CordaPersistence) : NodePropertiesStore {
private companion object {
val logger = loggerFor<NodePropertiesStore>()
}
override val flowsDrainingMode: FlowsDrainingModeOperations = FlowsDrainingModeOperationsImpl(readPhysicalNodeId, persistence, logger)
@Entity
@Table(name = "${NODE_DATABASE_PREFIX}properties")
class DBNodeProperty(
@Id
@Column(name = "key")
val key: String = "",
@Column(name = "value")
var value: String? = ""
)
}
private class FlowsDrainingModeOperationsImpl(readPhysicalNodeId: () -> String, private val persistence: CordaPersistence, logger: Logger) : FlowsDrainingModeOperations {
private val nodeSpecificFlowsExecutionModeKey = "${readPhysicalNodeId()}_flowsExecutionMode"
init {
logger.debug { "Node's flow execution mode property key: $nodeSpecificFlowsExecutionModeKey" }
}
private val map = PersistentMap({ key -> key }, { entity -> entity.key to entity.value!! }, NodePropertiesPersistentStore::DBNodeProperty, NodePropertiesPersistentStore.DBNodeProperty::class.java)
override val values = PublishSubject.create<Pair<Boolean, Boolean>>()!!
override fun setEnabled(enabled: Boolean) {
var oldValue: Boolean? = null
persistence.transaction {
oldValue = map.put(nodeSpecificFlowsExecutionModeKey, enabled.toString())?.toBoolean() ?: false
}
values.onNext(oldValue!! to enabled)
}
override fun isEnabled(): Boolean {
return persistence.transaction {
map[nodeSpecificFlowsExecutionModeKey]?.toBoolean() ?: false
}
}
}

2
node/src/main/kotlin/net/corda/node/services/schema/NodeSchemaService.kt
View File

@ -3,7 +3,7 @@ package net.corda.node.services.schema
import net.corda.core.contracts.ContractState
import net.corda.core.contracts.FungibleAsset
import net.corda.core.contracts.LinearState
import net.corda.core.internal.schemas.NodeInfoSchemaV1
import net.corda.node.internal.schemas.NodeInfoSchemaV1
import net.corda.core.schemas.CommonSchemaV1
import net.corda.core.schemas.MappedSchema
import net.corda.core.schemas.PersistentState

17
node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineManagerImpl.kt
View File

@ -14,14 +14,17 @@ import net.corda.core.concurrent.CordaFuture
import net.corda.core.context.InvocationContext
import net.corda.core.crypto.SecureHash
import net.corda.core.crypto.newSecureRandom
import net.corda.core.crypto.random63BitValue
import net.corda.core.flows.FlowException
import net.corda.core.flows.FlowInfo
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.StateMachineRunId
import net.corda.core.identity.Party
import net.corda.core.internal.*
import net.corda.core.internal.FlowStateMachine
import net.corda.core.internal.ThreadBox
import net.corda.core.internal.bufferUntilSubscribed
import net.corda.core.internal.castIfPossible
import net.corda.core.internal.concurrent.doneFuture
import net.corda.core.internal.uncheckedCast
import net.corda.core.messaging.DataFeed
import net.corda.core.serialization.SerializationDefaults.CHECKPOINT_CONTEXT
import net.corda.core.serialization.SerializationDefaults.SERIALIZATION_FACTORY
@ -37,6 +40,7 @@ import net.corda.node.services.api.Checkpoint
import net.corda.node.services.api.CheckpointStorage
import net.corda.node.services.api.ServiceHubInternal
import net.corda.node.services.config.shouldCheckCheckpoints
import net.corda.node.services.messaging.P2PMessagingHeaders
import net.corda.node.services.messaging.ReceivedMessage
import net.corda.node.utilities.AffinityExecutor
import net.corda.node.utilities.newNamedSingleThreadExecutor
@ -640,11 +644,18 @@ class StateMachineManagerImpl(
}
serviceHub.networkService.apply {
send(createMessage(sessionTopic, serialized.bytes), address, retryId = retryId)
send(createMessage(sessionTopic, serialized.bytes), address, retryId = retryId, additionalHeaders = message.additionalHeaders())
}
}
}
private fun SessionMessage.additionalHeaders(): Map<String, String> {
return when (this) {
is InitialSessionMessage -> mapOf(P2PMessagingHeaders.Type.KEY to P2PMessagingHeaders.Type.SESSION_INIT_VALUE)
else -> emptyMap()
}
}
class SessionRejectException(val rejectMessage: String, val logMessage: String) : CordaException(rejectMessage) {
constructor(message: String) : this(message, message)
}

153
node/src/test/kotlin/net/corda/node/modes/draining/ScheduledFlowsDrainingModeTest.kt Normal file
View File

@ -0,0 +1,153 @@
package net.corda.node.modes.draining
import co.paralleluniverse.fibers.Suspendable
import net.corda.core.contracts.LinearState
import net.corda.core.contracts.SchedulableState
import net.corda.core.contracts.ScheduledActivity
import net.corda.core.contracts.StateRef
import net.corda.core.contracts.UniqueIdentifier
import net.corda.core.flows.FinalityFlow
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.FlowLogicRefFactory
import net.corda.core.flows.SchedulableFlow
import net.corda.core.identity.Party
import net.corda.core.transactions.TransactionBuilder
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.loggerFor
import net.corda.node.internal.StartedNode
import net.corda.testing.contracts.DummyContract
import net.corda.testing.core.ALICE_NAME
import net.corda.testing.core.BOB_NAME
import net.corda.testing.core.dummyCommand
import net.corda.testing.core.singleIdentity
import net.corda.testing.node.MockNetwork
import net.corda.testing.node.MockNodeParameters
import net.corda.testing.node.internal.InternalMockNetwork
import net.corda.testing.node.internal.startFlow
import org.junit.After
import org.junit.Before
import org.junit.Test
import java.time.Instant
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit
import kotlin.reflect.jvm.jvmName
import kotlin.test.fail
class ScheduledFlowsDrainingModeTest {
private lateinit var mockNet: InternalMockNetwork
private lateinit var aliceNode: StartedNode<InternalMockNetwork.MockNode>
private lateinit var bobNode: StartedNode<InternalMockNetwork.MockNode>
private lateinit var notary: Party
private lateinit var alice: Party
private lateinit var bob: Party
private var executor: ScheduledExecutorService? = null
companion object {
private val logger = loggerFor<ScheduledFlowsDrainingModeTest>()
}
@Before
fun setup() {
mockNet = InternalMockNetwork(threadPerNode = true, cordappPackages = listOf("net.corda.testing.contracts"))
aliceNode = mockNet.createNode(MockNodeParameters(legalName = ALICE_NAME))
bobNode = mockNet.createNode(MockNodeParameters(legalName = BOB_NAME))
notary = mockNet.defaultNotaryIdentity
alice = aliceNode.info.singleIdentity()
bob = bobNode.info.singleIdentity()
executor = Executors.newSingleThreadScheduledExecutor()
}
@After
fun cleanUp() {
mockNet.stopNodes()
executor!!.shutdown()
}
@Test
fun `flows draining mode ignores scheduled flows until unset`() {
val latch = CountDownLatch(1)
var shouldFail = true
aliceNode.services.nodeProperties.flowsDrainingMode.setEnabled(true)
val scheduledStates = aliceNode.services
.vaultService
.updates
.filter { update -> update.containsType<ScheduledState>() }
.map { update -> update.produced.single().state.data as ScheduledState }
scheduledStates.filter { state -> !state.processed }.doOnNext { _ ->
// this is needed because there is a delay between the moment a SchedulableState gets in the Vault and the first time nextScheduledActivity is called
executor!!.schedule({
logger.info("Disabling flows draining mode")
shouldFail = false
aliceNode.services.nodeProperties.flowsDrainingMode.setEnabled(false)
}, 5, TimeUnit.SECONDS)
}.subscribe()
scheduledStates.filter { state -> state.processed }.doOnNext { _ ->
if (shouldFail) {
fail("Should not have happened before draining is switched off.")
}
latch.countDown()
}.subscribe()
val flow = aliceNode.services.startFlow(InsertInitialStateFlow(bob, notary))
flow.resultFuture.getOrThrow()
mockNet.waitQuiescent()
latch.await()
}
data class ScheduledState(private val creationTime: Instant, val source: Party, val destination: Party, val processed: Boolean = false, override val linearId: UniqueIdentifier = UniqueIdentifier()) : SchedulableState, LinearState {
override fun nextScheduledActivity(thisStateRef: StateRef, flowLogicRefFactory: FlowLogicRefFactory): ScheduledActivity? {
return if (!processed) {
val logicRef = flowLogicRefFactory.create(ScheduledFlow::class.jvmName, thisStateRef)
ScheduledActivity(logicRef, creationTime)
} else {
null
}
}
override val participants: List<Party> get() = listOf(source, destination)
}
class InsertInitialStateFlow(private val destination: Party, private val notary: Party) : FlowLogic<Unit>() {
@Suspendable
override fun call() {
val scheduledState = ScheduledState(serviceHub.clock.instant(), ourIdentity, destination)
val builder = TransactionBuilder(notary).addOutputState(scheduledState, DummyContract.PROGRAM_ID).addCommand(dummyCommand(ourIdentity.owningKey))
val tx = serviceHub.signInitialTransaction(builder)
subFlow(FinalityFlow(tx))
}
}
@SchedulableFlow
class ScheduledFlow(private val stateRef: StateRef) : FlowLogic<Unit>() {
@Suspendable
override fun call() {
val state = serviceHub.toStateAndRef<ScheduledState>(stateRef)
val scheduledState = state.state.data
// Only run flow over states originating on this node
if (!serviceHub.myInfo.isLegalIdentity(scheduledState.source)) {
return
}
require(!scheduledState.processed) { "State should not have been previously processed" }
val notary = state.state.notary
val newStateOutput = scheduledState.copy(processed = true)
val builder = TransactionBuilder(notary).addInputState(state).addOutputState(newStateOutput, DummyContract.PROGRAM_ID).addCommand(dummyCommand(ourIdentity.owningKey))
val tx = serviceHub.signInitialTransaction(builder)
subFlow(FinalityFlow(tx, setOf(scheduledState.destination)))
}
}
}

27
node/src/test/kotlin/net/corda/node/services/RPCSecurityManagerTest.kt
View File

@ -6,8 +6,9 @@ import net.corda.core.messaging.CordaRPCOps
import net.corda.node.internal.security.Password
import net.corda.node.internal.security.RPCSecurityManagerImpl
import net.corda.node.internal.security.tryAuthenticate
import net.corda.nodeapi.internal.config.User
import net.corda.node.services.Permissions.Companion.invokeRpc
import net.corda.node.services.config.SecurityConfiguration
import net.corda.nodeapi.internal.config.User
import org.assertj.core.api.Assertions.assertThatThrownBy
import org.junit.Test
import javax.security.auth.login.FailedLoginException
@ -30,8 +31,8 @@ class RPCSecurityManagerTest {
checkUserActions(
permitted = setOf(arrayListOf("nodeInfo"), arrayListOf("notaryIdentities")),
permissions = setOf(
Permissions.invokeRpc(CordaRPCOps::nodeInfo),
Permissions.invokeRpc(CordaRPCOps::notaryIdentities)))
invokeRpc(CordaRPCOps::nodeInfo),
invokeRpc(CordaRPCOps::notaryIdentities)))
}
@Test
@ -46,7 +47,7 @@ class RPCSecurityManagerTest {
@Test
fun `Check startFlow RPC permission implies startFlowDynamic`() {
checkUserActions(
permissions = setOf(Permissions.invokeRpc("startFlow")),
permissions = setOf(invokeRpc("startFlow")),
permitted = setOf(arrayListOf("startFlow"), arrayListOf("startFlowDynamic")))
}
@ -54,7 +55,7 @@ class RPCSecurityManagerTest {
fun `Check startTrackedFlow RPC permission implies startTrackedFlowDynamic`() {
checkUserActions(
permitted = setOf(arrayListOf("startTrackedFlow"), arrayListOf("startTrackedFlowDynamic")),
permissions = setOf(Permissions.invokeRpc("startTrackedFlow")))
permissions = setOf(invokeRpc("startTrackedFlow")))
}
@Test
@ -64,6 +65,18 @@ class RPCSecurityManagerTest {
permitted = allActions.map { arrayListOf(it) }.toSet())
}
@Test
fun `flows draining mode permissions`() {
checkUserActions(
permitted = setOf(arrayListOf("setFlowsDrainingModeEnabled")),
permissions = setOf(invokeRpc(CordaRPCOps::setFlowsDrainingModeEnabled))
)
checkUserActions(
permitted = setOf(arrayListOf("isFlowsDrainingModeEnabled")),
permissions = setOf(invokeRpc(CordaRPCOps::isFlowsDrainingModeEnabled))
)
}
@Test
fun `Malformed permission strings`() {
assertMalformedPermission("bar")
@ -131,11 +144,11 @@ class RPCSecurityManagerTest {
val call = request.first()
val args = request.drop(1).toTypedArray()
assert(subject.isPermitted(request.first(), *args)) {
"User ${subject.principal} should be permitted ${call} with target '${request.toList()}'"
"User ${subject.principal} should be permitted $call with target '${request.toList()}'"
}
if (args.isEmpty()) {
assert(subject.isPermitted(request.first(), "XXX")) {
"User ${subject.principal} should be permitted ${call} with any target"
"User ${subject.principal} should be permitted $call with any target"
}
}
}

12
node/src/test/kotlin/net/corda/node/services/events/NodeSchedulerServiceTest.kt
View File

@ -12,6 +12,7 @@ import net.corda.core.internal.uncheckedCast
import net.corda.core.node.StateLoader
import net.corda.core.utilities.days
import net.corda.node.services.api.FlowStarter
import net.corda.node.services.api.NodePropertiesStore
import net.corda.nodeapi.internal.persistence.CordaPersistence
import net.corda.nodeapi.internal.persistence.DatabaseTransaction
import net.corda.testing.internal.doLookup
@ -23,6 +24,7 @@ import org.junit.rules.TestWatcher
import org.junit.runner.Description
import org.slf4j.Logger
import java.time.Clock
import java.time.Duration
import java.time.Instant
class NodeSchedulerServiceTest {
@ -39,6 +41,12 @@ class NodeSchedulerServiceTest {
private val flowStarter = rigorousMock<FlowStarter>().also {
doReturn(openFuture<FlowStateMachine<*>>()).whenever(it).startFlow(any<FlowLogic<*>>(), any())
}
private val flowsDraingMode = rigorousMock<NodePropertiesStore.FlowsDrainingModeOperations>().also {
doReturn(false).whenever(it).isEnabled()
}
private val nodeProperties = rigorousMock<NodePropertiesStore>().also {
doReturn(flowsDraingMode).whenever(it).flowsDrainingMode
}
private val transactionStates = mutableMapOf<StateRef, TransactionState<*>>()
private val stateLoader = rigorousMock<StateLoader>().also {
doLookup(transactionStates).whenever(it).loadState(any())
@ -58,6 +66,8 @@ class NodeSchedulerServiceTest {
stateLoader,
serverThread = MoreExecutors.directExecutor(),
flowLogicRefFactory = flowLogicRefFactory,
nodeProperties = nodeProperties,
drainingModePollPeriod = Duration.ofSeconds(5),
log = log,
scheduledStates = mutableMapOf()).apply { start() }
@Rule
@ -173,4 +183,4 @@ class NodeSchedulerServiceTest {
scheduler.unscheduleStateActivity(schedule(mark + 1.days).stateRef)
testClock.advanceBy(1.days)
}
}
}

8
node/src/test/kotlin/net/corda/node/services/messaging/ArtemisMessagingTest.kt
View File

@ -26,6 +26,7 @@ import org.junit.Before
import org.junit.Rule
import org.junit.Test
import org.junit.rules.TemporaryFolder
import rx.subjects.PublishSubject
import java.net.ServerSocket
import java.util.concurrent.BlockingQueue
import java.util.concurrent.LinkedBlockingQueue
@ -150,10 +151,11 @@ class ArtemisMessagingTest {
createMessagingServer().start()
val messagingClient = createMessagingClient(platformVersion = platformVersion)
startNodeMessagingClient()
messagingClient.addMessageHandler(TOPIC) { message, _ ->
receivedMessages.add(message)
}
startNodeMessagingClient()
// Run after the handlers are added, otherwise (some of) the messages get delivered and discarded / dead-lettered.
thread(isDaemon = true) { messagingClient.run() }
@ -171,7 +173,9 @@ class ArtemisMessagingTest {
ServiceAffinityExecutor("ArtemisMessagingTests", 1),
database,
networkMapCache,
maxMessageSize = maxMessageSize).apply {
maxMessageSize = maxMessageSize,
isDrainingModeOn = { false },
drainingModeWasChangedEvents = PublishSubject.create<Pair<Boolean, Boolean>>()).apply {
config.configureWithDevSSLCertificate()
messagingClient = this
}

2
testing/node-driver/src/main/kotlin/net/corda/testing/node/InMemoryMessagingNetwork.kt
View File

@ -330,7 +330,7 @@ class InMemoryMessagingNetwork internal constructor(
state.locked { check(handlers.remove(registration as Handler)) }
}
override fun send(message: Message, target: MessageRecipients, retryId: Long?, sequenceKey: Any) {
override fun send(message: Message, target: MessageRecipients, retryId: Long?, sequenceKey: Any, additionalHeaders: Map<String, String>) {
check(running)
msgSend(this, message, target)
if (!sendManuallyPumped) {

2
testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/DriverDSLImpl.kt
View File

@ -159,7 +159,7 @@ class DriverDSLImpl(
config.corda.rpcUsers[0].run { client.start(username, password) }
} catch (e: Exception) {
if (processDeathFuture.isDone) throw e
log.error("Exception while connecting to RPC, retrying to connect at $rpcAddress", e)
log.info("Exception while connecting to RPC, retrying to connect at $rpcAddress", e)
null
}
}

16
testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/InternalMockNetwork.kt
View File

@ -30,8 +30,13 @@ import net.corda.node.VersionInfo
import net.corda.node.internal.AbstractNode
import net.corda.node.internal.StartedNode
import net.corda.node.internal.cordapp.CordappLoader
import net.corda.node.services.api.NodePropertiesStore
import net.corda.node.services.api.SchemaService
import net.corda.node.services.config.*
import net.corda.node.services.config.BFTSMaRtConfiguration
import net.corda.node.services.config.CertChainPolicyConfig
import net.corda.node.services.config.NodeConfiguration
import net.corda.node.services.config.NotaryConfig
import net.corda.node.services.config.VerifierType
import net.corda.node.services.keys.E2ETestKeyManagementService
import net.corda.node.services.messaging.MessagingService
import net.corda.node.services.transactions.BFTNonValidatingNotaryService
@ -48,9 +53,14 @@ import net.corda.testing.common.internal.testNetworkParameters
import net.corda.testing.core.setGlobalSerialization
import net.corda.testing.internal.rigorousMock
import net.corda.testing.internal.testThreadFactory
import net.corda.testing.node.*
import net.corda.testing.node.InMemoryMessagingNetwork
import net.corda.testing.node.MessagingServiceSpy
import net.corda.testing.node.MockNetworkNotarySpec
import net.corda.testing.node.MockNetworkParameters
import net.corda.testing.node.MockNodeParameters
import net.corda.testing.node.MockServices.Companion.MOCK_VERSION_INFO
import net.corda.testing.node.MockServices.Companion.makeTestDataSourceProperties
import net.corda.testing.node.TestClock
import org.apache.activemq.artemis.utils.ReusableLatch
import org.apache.sshd.common.util.security.SecurityUtils
import rx.internal.schedulers.CachedThreadScheduler
@ -259,7 +269,7 @@ open class InternalMockNetwork(private val cordappPackages: List<String>,
// We only need to override the messaging service here, as currently everything that hits disk does so
// through the java.nio API which we are already mocking via Jimfs.
override fun makeMessagingService(database: CordaPersistence, info: NodeInfo): MessagingService {
override fun makeMessagingService(database: CordaPersistence, info: NodeInfo, nodeProperties: NodePropertiesStore): MessagingService {
require(id >= 0) { "Node ID must be zero or positive, was passed: " + id }
return mockNet.messagingNetwork.createNodeWithID(
!mockNet.threadPerNode,