ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-04-07 11:27:01 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

CORDA-866 - backport event horizon + serialization fixes (#3489)

Browse Source 
Implement removal of stale nodes from network - backport (#3128)
Implement removal of stale nodes from network

Add eventHorizon to network parameters structure. Add republishing of
node info on 1 day intervals - it is treated by network map as heartbeat from node indicating if it's alive or not. Add removal of old node infos on network map signing.

Add copy method to network parameters data class

Add JvmOverloads annotation to the constructor, because it's data class
exposed in API
make NetworkParameters serialization compatible (#3234)

Cope with API checker inlining annotations
This commit is contained in:
Katarzyna Streich 2018-07-03 14:38:48 +01:00 committed by Katelyn Baker
parent 881509ab0d
commit ba37b3ea57
14 changed files with 149 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.ci/api-current.txt
View File

@ -1847,7 +1847,8 @@ public @interface net.corda.core.messaging.RPCReturnsObservables
@org.jetbrains.annotations.NotNull public abstract net.corda.core.messaging.FlowProgressHandle startTrackedFlow(net.corda.core.flows.FlowLogic)
##
@net.corda.core.serialization.CordaSerializable public final class net.corda.core.node.NetworkParameters extends java.lang.Object
public <init>(int, List, int, int, java.time.Instant, int, Map)
@net.corda.core.serialization.DeprecatedConstructorForDeserialization public <init>(int, List, int, int, java.time.Instant, int, Map)
public <init>(int, List, int, int, java.time.Instant, int, Map, java.time.Duration)
public final int component1()
@org.jetbrains.annotations.NotNull public final List component2()
public final int component3()

47
core/src/main/kotlin/net/corda/core/node/NetworkParameters.kt
View File

@ -3,6 +3,9 @@ package net.corda.core.node
import net.corda.core.identity.Party
import net.corda.core.node.services.AttachmentId
import net.corda.core.serialization.CordaSerializable
import net.corda.core.serialization.DeprecatedConstructorForDeserialization
import net.corda.core.utilities.days
import java.time.Duration
import java.time.Instant
/**
@ -17,9 +20,9 @@ import java.time.Instant
* of parameters.
* @property whitelistedContractImplementations List of whitelisted jars containing contract code for each contract class.
* This will be used by [net.corda.core.contracts.WhitelistedByZoneAttachmentConstraint]. Read more about contract constraints here: <https://docs.corda.net/api-contract-constraints.html>
* @property eventHorizon Time after which nodes will be removed from the network map if they have not been seen
* during this period
*/
// TODO Add eventHorizon - how many days a node can be offline before being automatically ejected from the network.
// It needs separate design.
@CordaSerializable
data class NetworkParameters(
val minimumPlatformVersion: Int,
@ -28,14 +31,52 @@ data class NetworkParameters(
val maxTransactionSize: Int,
val modifiedTime: Instant,
val epoch: Int,
val whitelistedContractImplementations: Map<String, List<AttachmentId>>
val whitelistedContractImplementations: Map<String, List<AttachmentId>>,
val eventHorizon: Duration
) {
@DeprecatedConstructorForDeserialization(1)
constructor (minimumPlatformVersion: Int,
notaries: List<NotaryInfo>,
maxMessageSize: Int,
maxTransactionSize: Int,
modifiedTime: Instant,
epoch: Int,
whitelistedContractImplementations: Map<String, List<AttachmentId>>
) : this(minimumPlatformVersion,
notaries,
maxMessageSize,
maxTransactionSize,
modifiedTime,
epoch,
whitelistedContractImplementations,
Int.MAX_VALUE.days
)
init {
require(minimumPlatformVersion > 0) { "minimumPlatformVersion must be at least 1" }
require(notaries.distinctBy { it.identity } == notaries) { "Duplicate notary identities" }
require(epoch > 0) { "epoch must be at least 1" }
require(maxMessageSize > 0) { "maxMessageSize must be at least 1" }
require(maxTransactionSize > 0) { "maxTransactionSize must be at least 1" }
require(!eventHorizon.isNegative) { "eventHorizon must be positive value" }
}
fun copy(minimumPlatformVersion: Int,
notaries: List<NotaryInfo>,
maxMessageSize: Int,
maxTransactionSize: Int,
modifiedTime: Instant,
epoch: Int,
whitelistedContractImplementations: Map<String, List<AttachmentId>>
): NetworkParameters {
return copy(minimumPlatformVersion = minimumPlatformVersion,
notaries = notaries,
maxMessageSize = maxMessageSize,
maxTransactionSize = maxTransactionSize,
modifiedTime = modifiedTime,
epoch = epoch,
whitelistedContractImplementations = whitelistedContractImplementations,
eventHorizon = eventHorizon)
}
}

3
docs/source/network-map.rst
View File

@ -119,6 +119,9 @@ The current set of network parameters:
For each contract class there is a list of hashes of the approved CorDapp jar versions containing that contract.
Read more about *Zone constraints* here :doc:`api-contract-constraints`
:eventHorizon: Time after which nodes are considered to be unresponsive and removed from network map. Nodes republish their
``NodeInfo`` on a regular interval. Network map treats that as a heartbeat from the node.
More parameters will be added in future releases to regulate things like allowed port numbers, how long a node can be
offline before it is evicted from the zone, whether or not IPv6 connectivity is required for zone members, required
cryptographic algorithms and rollout schedules (e.g. for moving to post quantum cryptography), parameters related to

3
docs/source/serialization-default-evolution.rst
View File

@ -64,7 +64,7 @@ above. A sensible default for the missing value is provided for instantiation of
order, see the discussion below.
As before, instances of the class at version A will be able to deserialize serialized forms of example B as it
will simply treat them as if the property has been removed (as from its perspective, they will have been.)
will simply treat them as if the property has been removed (as from its perspective, they will have been).
Constructor Versioning
@ -144,7 +144,6 @@ be:
// The third alteration, and how it currently exists, property e added
data class Example3 (val a: Int, val b: Int, val c: Int, val d: Int, val: Int e) {
// NOTE: version number purposefully omitted from annotation for demonstration purposes
@DeprecatedConstructorForDeserialization(1)
constructor (a: Int, b: Int) : this(a, b, -1, -1, -1) // alt constructor 1
@DeprecatedConstructorForDeserialization(2)

4
node-api/src/main/kotlin/net/corda/nodeapi/internal/network/NetworkBootstrapper.kt
View File

@ -18,6 +18,7 @@ import net.corda.core.serialization.SerializationContext
import net.corda.core.serialization.internal.SerializationEnvironmentImpl
import net.corda.core.serialization.internal._contextSerializationEnv
import net.corda.core.utilities.ByteSequence
import net.corda.core.utilities.days
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.seconds
import net.corda.nodeapi.internal.SignedNodeInfo
@ -181,7 +182,8 @@ class NetworkBootstrapper {
maxMessageSize = 10485760,
maxTransactionSize = Int.MAX_VALUE,
epoch = 1,
whitelistedContractImplementations = whitelist
whitelistedContractImplementations = whitelist,
eventHorizon = 30.days
), overwriteFile = true)
nodeDirs.forEach { copier.install(it) }

2
node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/EvolutionSerializer.kt
View File

@ -114,7 +114,7 @@ abstract class EvolutionSerializer(
this.resultsIndex = it.index
} ?: if (!it.value.type.isMarkedNullable) {
throw NotSerializableException(
"New parameter ${it.value.name} is mandatory, should be nullable for evolution to worK")
"New parameter ${it.value.name} is mandatory, should be nullable for evolution to work")
}
}
return EvolutionSerializerViaConstructor (new.type, factory, readersAsSerialized, constructor, constructorArgs)

18
node/src/integration-test/kotlin/net/corda/node/services/network/NetworkMapTest.kt
View File

@ -8,6 +8,7 @@ import net.corda.core.internal.exists
import net.corda.core.internal.list
import net.corda.core.internal.readObject
import net.corda.core.node.NodeInfo
import net.corda.core.serialization.serialize
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.seconds
import net.corda.nodeapi.internal.network.NETWORK_PARAMS_FILE_NAME
@ -170,6 +171,23 @@ class NetworkMapTest(var initFunc: (URL, NetworkMapServer) -> CompatibilityZoneP
}
}
@Test
fun `test node heartbeat`() {
internalDriver(
portAllocation = portAllocation,
compatibilityZone = compatibilityZone,
initialiseSerialization = false,
systemProperties = mapOf("net.corda.node.internal.nodeinfo.publish.interval" to 1.seconds.toString())
) {
val aliceNode = startNode(providedName = ALICE_NAME).getOrThrow()
assertThat(networkMapServer.networkMapHashes()).contains(aliceNode.nodeInfo.serialize().hash)
networkMapServer.removeNodeInfo(aliceNode.nodeInfo)
assertThat(networkMapServer.networkMapHashes()).doesNotContain(aliceNode.nodeInfo.serialize().hash)
Thread.sleep(2000)
assertThat(networkMapServer.networkMapHashes()).contains(aliceNode.nodeInfo.serialize().hash)
}
}
private fun NodeHandle.onlySees(vararg nodes: NodeInfo) {
// Make sure the nodes aren't getting the node infos from their additional directories
val nodeInfosDir = baseDirectory / CordformNode.NODE_INFO_DIRECTORY

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

@ -24,11 +24,8 @@ import net.corda.core.node.services.*
import net.corda.core.serialization.SerializationWhitelist
import net.corda.core.serialization.SerializeAsToken
import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.serialization.serialize
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.core.utilities.debug
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.*
import net.corda.node.CordaClock
import net.corda.node.VersionInfo
import net.corda.node.internal.classloading.requireAnnotation
@ -83,6 +80,7 @@ import java.security.cert.X509Certificate
import java.sql.Connection
import java.time.Clock
import java.time.Duration
import java.time.format.DateTimeParseException
import java.util.*
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ExecutorService
@ -198,8 +196,8 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
val identityService = makeIdentityService(identity.certificate)
networkMapClient = configuration.networkServices?.let { NetworkMapClient(it.networkMapURL, identityService.trustRoot) }
val networkParameters = NetworkParametersReader(identityService.trustRoot, networkMapClient, configuration.baseDirectory).networkParameters
val networkParameteresReader = NetworkParametersReader(identityService.trustRoot, networkMapClient, configuration.baseDirectory)
val networkParameters = networkParameteresReader.networkParameters
check(networkParameters.minimumPlatformVersion <= versionInfo.platformVersion) {
"Node's platform version is lower than network's required minimumPlatformVersion"
}
@ -267,7 +265,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
networkMapUpdater = NetworkMapUpdater(services.networkMapCache,
NodeInfoWatcher(configuration.baseDirectory, getRxIoScheduler(), Duration.ofMillis(configuration.additionalNodeInfoPollingFrequencyMsec)),
networkMapClient,
networkParameters.serialize().hash,
networkParameteresReader.hash,
configuration.baseDirectory)
runOnStop += networkMapUpdater::close
@ -343,6 +341,7 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
// Write the node-info file even if nothing's changed, just in case the file has been deleted.
NodeInfoWatcher.saveToFile(configuration.baseDirectory, nodeInfoAndSigned)
// Always republish on startup, it's treated by network map server as a heartbeat.
if (networkMapClient != null) {
tryPublishNodeInfoAsync(nodeInfoAndSigned.signed, networkMapClient)
}
@ -350,18 +349,31 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
return Pair(keyPairs, nodeInfo)
}
// Publish node info on startup and start task that sends every day a heartbeat - republishes node info.
private fun tryPublishNodeInfoAsync(signedNodeInfo: SignedNodeInfo, networkMapClient: NetworkMapClient) {
// By default heartbeat interval should be set to 1 day, but for testing we may change it.
val republishProperty = System.getProperty("net.corda.node.internal.nodeinfo.publish.interval")
val heartbeatInterval = if (republishProperty != null) {
try {
Duration.parse(republishProperty)
} catch (e: DateTimeParseException) {
1.days
}
} else {
1.days
}
val executor = Executors.newSingleThreadScheduledExecutor(NamedThreadFactory("Network Map Updater", Executors.defaultThreadFactory()))
executor.submit(object : Runnable {
override fun run() {
try {
val republishInterval = try {
networkMapClient.publish(signedNodeInfo)
heartbeatInterval
} catch (t: Throwable) {
log.warn("Error encountered while publishing node info, will retry again", t)
// TODO: Exponential backoff?
executor.schedule(this, 1, TimeUnit.MINUTES)
// TODO: Exponential backoff? It should reach max interval of eventHorizon/2.
1.minutes
}
executor.schedule(this, republishInterval.toMinutes(), TimeUnit.MINUTES)
}
})
}

22
node/src/main/kotlin/net/corda/node/internal/NetworkParametersReader.kt
View File

@ -22,11 +22,14 @@ class NetworkParametersReader(private val trustRoot: X509Certificate,
private val logger = contextLogger()
}
private data class NetworkParamsAndHash(val networkParameters: NetworkParameters, val hash: SecureHash)
private val networkParamsFile = baseDirectory / NETWORK_PARAMS_FILE_NAME
private val parametersUpdateFile = baseDirectory / NETWORK_PARAMS_UPDATE_FILE_NAME
val networkParameters by lazy { retrieveNetworkParameters() }
private val netParamsAndHash by lazy { retrieveNetworkParameters() }
val networkParameters get() = netParamsAndHash.networkParameters
val hash get() = netParamsAndHash.hash
private fun retrieveNetworkParameters(): NetworkParameters {
private fun retrieveNetworkParameters(): NetworkParamsAndHash {
val advertisedParametersHash = try {
networkMapClient?.getNetworkMap()?.payload?.networkParameterHash
} catch (e: ConnectException) {
@ -44,18 +47,18 @@ class NetworkParametersReader(private val trustRoot: X509Certificate,
// on the other we have parameters update process - it needs to be unified. Say you start the node, you don't have matching parameters,
// you get them from network map, but you have to run the approval step.
if (signedParametersFromFile == null) { // Node joins for the first time.
downloadParameters(trustRoot, advertisedParametersHash)
downloadParameters(advertisedParametersHash)
}
else if (signedParametersFromFile.raw.hash == advertisedParametersHash) { // Restarted with the same parameters.
signedParametersFromFile.verifiedNetworkMapCert(trustRoot)
signedParametersFromFile
} else { // Update case.
readParametersUpdate(advertisedParametersHash, signedParametersFromFile.raw.hash).verifiedNetworkMapCert(trustRoot)
readParametersUpdate(advertisedParametersHash, signedParametersFromFile.raw.hash)
}
} else { // No compatibility zone configured. Node should proceed with parameters from file.
signedParametersFromFile?.verifiedNetworkMapCert(trustRoot) ?: throw IllegalArgumentException("Couldn't find network parameters file and compatibility zone wasn't configured/isn't reachable")
signedParametersFromFile ?: throw IllegalArgumentException("Couldn't find network parameters file and compatibility zone wasn't configured/isn't reachable")
}
logger.info("Loaded network parameters: $parameters")
return parameters
return NetworkParamsAndHash(parameters.verifiedNetworkMapCert(trustRoot), parameters.raw.hash)
}
private fun readParametersUpdate(advertisedParametersHash: SecureHash, previousParametersHash: SecureHash): SignedNetworkParameters {
@ -75,14 +78,13 @@ class NetworkParametersReader(private val trustRoot: X509Certificate,
}
// Used only when node joins for the first time.
private fun downloadParameters(trustRoot: X509Certificate, parametersHash: SecureHash): NetworkParameters {
private fun downloadParameters(parametersHash: SecureHash): SignedNetworkParameters {
logger.info("No network-parameters file found. Expecting network parameters to be available from the network map.")
val networkMapClient = checkNotNull(networkMapClient) {
"Node hasn't been configured to connect to a network map from which to get the network parameters"
}
val signedParams = networkMapClient.getNetworkParameters(parametersHash)
val verifiedParams = signedParams.verifiedNetworkMapCert(trustRoot)
signedParams.serialize().open().copyTo(baseDirectory / NETWORK_PARAMS_FILE_NAME)
return verifiedParams
return signedParams
}
}

8
node/src/main/kotlin/net/corda/node/services/network/NetworkMapUpdater.kt
View File

@ -147,15 +147,17 @@ class NetworkMapUpdater(private val networkMapCache: NetworkMapCacheInternal,
networkMapClient ?: throw IllegalStateException("Network parameters updates are not supported without compatibility zone configured")
// TODO This scenario will happen if node was restarted and didn't download parameters yet, but we accepted them.
// Add persisting of newest parameters from update.
val (_, newParams) = requireNotNull(newNetworkParameters) { "Couldn't find parameters update for the hash: $parametersHash" }
val (update, signedNewNetParams) = requireNotNull(newNetworkParameters) { "Couldn't find parameters update for the hash: $parametersHash" }
// We should check that we sign the right data structure hash.
val newParametersHash = newParams.verifiedNetworkMapCert(networkMapClient.trustedRoot).serialize().hash
val newNetParams = signedNewNetParams.verifiedNetworkMapCert(networkMapClient.trustedRoot)
val newParametersHash = signedNewNetParams.raw.hash
if (parametersHash == newParametersHash) {
// The latest parameters have priority.
newParams.serialize()
signedNewNetParams.serialize()
.open()
.copyTo(baseDirectory / NETWORK_PARAMS_UPDATE_FILE_NAME, StandardCopyOption.REPLACE_EXISTING)
networkMapClient.ackNetworkParametersUpdate(sign(parametersHash))
logger.info("Accepted network parameter update $update: $newNetParams")
} else {
throw IllegalArgumentException("Refused to accept parameters with hash $parametersHash because network map " +
"advertises update with hash $newParametersHash. Please check newest version")

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

@ -6,6 +6,8 @@ import net.corda.core.internal.createDirectories
import net.corda.core.internal.div
import net.corda.core.internal.exists
import net.corda.core.internal.readObject
import net.corda.core.serialization.deserialize
import net.corda.core.utilities.days
import net.corda.core.utilities.seconds
import net.corda.node.internal.NetworkParametersReader
import net.corda.nodeapi.internal.network.*
@ -14,6 +16,7 @@ import net.corda.testing.core.SerializationEnvironmentRule
import net.corda.testing.driver.PortAllocation
import net.corda.testing.internal.DEV_ROOT_CA
import net.corda.testing.node.internal.network.NetworkMapServer
import org.assertj.core.api.Assertions.assertThat
import org.junit.After
import org.junit.Before
import org.junit.Rule
@ -21,12 +24,14 @@ import org.junit.Test
import java.net.URL
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertNotNull
class NetworkParametersReaderTest {
@Rule
@JvmField
val testSerialization = SerializationEnvironmentRule(true)
val fs = Jimfs.newFileSystem(Configuration.unix())
private val cacheTimeout = 100000.seconds
private lateinit var server: NetworkMapServer
@ -42,11 +47,11 @@ class NetworkParametersReaderTest {
@After
fun tearDown() {
server.close()
fs.close()
}
@Test
fun `read correct set of parameters from file`() {
val fs = Jimfs.newFileSystem(Configuration.unix())
val baseDirectory = fs.getPath("/node").createDirectories()
val oldParameters = testNetworkParameters(epoch = 1)
NetworkParametersCopier(oldParameters).install(baseDirectory)
@ -60,4 +65,24 @@ class NetworkParametersReaderTest {
.verifiedNetworkMapCert(DEV_ROOT_CA.certificate)
assertEquals(server.networkParameters, parametersFromFile)
}
@Test
fun `read network parameters from file when network map server is down`() {
server.close()
val baseDirectory = fs.getPath("/node").createDirectories()
val fileParameters = testNetworkParameters(epoch = 1)
NetworkParametersCopier(fileParameters).install(baseDirectory)
val parameters = NetworkParametersReader(DEV_ROOT_CA.certificate, networkMapClient, baseDirectory).networkParameters
assertThat(parameters).isEqualTo(fileParameters)
}
@Test
fun `serialized parameters compatibility`() {
// Network parameters file from before eventHorizon extension
val inputStream = javaClass.classLoader.getResourceAsStream("network-compatibility/network-parameters")
assertNotNull(inputStream)
val inByteArray: ByteArray = inputStream.readBytes()
val parameters = inByteArray.deserialize<SignedNetworkParameters>()
assertThat(parameters.verified().eventHorizon).isEqualTo(Int.MAX_VALUE.days)
}
}

BIN
node/src/test/resources/network-compatibility/network-parameters Normal file
View File

Binary file not shown.

7
testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/network/NetworkMapServer.kt
View File

@ -13,6 +13,7 @@ import net.corda.nodeapi.internal.crypto.CertificateAndKeyPair
import net.corda.core.node.NetworkParameters
import net.corda.nodeapi.internal.network.NetworkMap
import net.corda.nodeapi.internal.network.ParametersUpdate
import net.corda.testing.common.internal.testNetworkParameters
import org.eclipse.jetty.server.Server
import org.eclipse.jetty.server.ServerConnector
import org.eclipse.jetty.server.handler.HandlerCollection
@ -39,7 +40,7 @@ class NetworkMapServer(private val cacheTimeout: Duration,
private val myHostNameValue: String = "test.host.name",
vararg additionalServices: Any) : Closeable {
companion object {
private val stubNetworkParameters = NetworkParameters(1, emptyList(), 10485760, Int.MAX_VALUE, Instant.now(), 10, emptyMap())
private val stubNetworkParameters = testNetworkParameters(epoch = 10)
}
private val server: Server
@ -81,6 +82,8 @@ class NetworkMapServer(private val cacheTimeout: Duration,
.let { NetworkHostAndPort(it.host, it.localPort) }
}
fun networkMapHashes(): List<SecureHash> = service.nodeInfoMap.keys.toList()
fun removeNodeInfo(nodeInfo: NodeInfo) {
service.removeNodeInfo(nodeInfo)
}
@ -106,7 +109,7 @@ class NetworkMapServer(private val cacheTimeout: Duration,
@Path("network-map")
inner class InMemoryNetworkMapService {
private val nodeInfoMap = mutableMapOf<SecureHash, SignedNodeInfo>()
val nodeInfoMap = mutableMapOf<SecureHash, SignedNodeInfo>()
val latestAcceptedParametersMap = mutableMapOf<PublicKey, SecureHash>()
private val signedNetParams by lazy {
networkParameters.signWithCert(networkMapCa.keyPair.private, networkMapCa.certificate)

8
testing/test-common/src/main/kotlin/net/corda/testing/common/internal/ParametersUtilities.kt
View File

@ -2,6 +2,8 @@ package net.corda.testing.common.internal
import net.corda.core.node.NetworkParameters
import net.corda.core.node.NotaryInfo
import net.corda.core.utilities.days
import java.time.Duration
import java.time.Instant
fun testNetworkParameters(
@ -11,7 +13,8 @@ fun testNetworkParameters(
maxMessageSize: Int = 10485760,
// TODO: Make this configurable and consistence across driver, bootstrapper, demobench and NetworkMapServer
maxTransactionSize: Int = maxMessageSize,
epoch: Int = 1
epoch: Int = 1,
eventHorizon: Duration = 30.days
): NetworkParameters {
return NetworkParameters(
minimumPlatformVersion = minimumPlatformVersion,
@ -20,6 +23,7 @@ fun testNetworkParameters(
maxMessageSize = maxMessageSize,
maxTransactionSize = maxTransactionSize,
epoch = epoch,
whitelistedContractImplementations = emptyMap()
whitelistedContractImplementations = emptyMap(),
eventHorizon = eventHorizon
)
}