CORDA-866, ENT-1933 - Remove stale nodes from Network, Fix NetParam serialization (#3255) (#3128)

*  CORDA-866: Implement removal of stale nodes from network - backport (#3128)

* CORDA-866: Implement removal of stale nodes from network

Backported

* Implement removal of stale nodes from network

Add eventHorizon to NetworkParameters 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 NetworkParameters data class

Add JvmOverloads annotation to the constructor, because it's data class
exposed in API

* Fix test

* ENT-1933: make NetworkParameters serialization compatible (#3234)

* ENT-1933: make NetworkParameters serialization compatible

* Fixes after cherry-pick

core/src/main/kotlin/net/corda/core/node/NetworkParameters.kt

@@ -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)
     }
 
     override fun toString(): String {
@@ -47,6 +88,7 @@ data class NetworkParameters(
   whitelistedContractImplementations {
     ${whitelistedContractImplementations.entries.joinToString("\n    ")}
   }
+  eventHorizon=$eventHorizon
   modifiedTime=$modifiedTime
   epoch=$epoch
 }"""

docs/source/network-map.rst

@@ -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

docs/source/serialization-default-evolution.rst

@@ -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)

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

@@ -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) }

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

@@ -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)

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

@@ -193,6 +193,23 @@ class NetworkMapTest {
         }
     }
 
+    @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

node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt

@@ -26,9 +26,7 @@ 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.*
-import net.corda.core.utilities.debug
-import net.corda.core.utilities.getOrThrow
 import net.corda.node.CordaClock
 import net.corda.node.VersionInfo
 import net.corda.node.internal.classloading.requireAnnotation
@@ -83,6 +81,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 +197,8 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
         val identityService = makeIdentityService(identity.certificate)
 
         networkMapClient = configuration.compatibilityZoneURL?.let { NetworkMapClient(it, identityService.trustRoot) }
-
+        val networkParameteresReader = NetworkParametersReader(identityService.trustRoot, networkMapClient, configuration.baseDirectory)
-        val networkParameters = NetworkParametersReader(identityService.trustRoot, networkMapClient, configuration.baseDirectory).networkParameters
+        val networkParameters = networkParameteresReader.networkParameters
         check(networkParameters.minimumPlatformVersion <= versionInfo.platformVersion) {
             "Node's platform version is lower than network's required minimumPlatformVersion"
         }
@@ -267,7 +266,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 +342,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 +350,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?
+                    // TODO: Exponential backoff? It should reach max interval of eventHorizon/2.
-                    executor.schedule(this, 1, TimeUnit.MINUTES)
+                    1.minutes
                 }
+                executor.schedule(this, republishInterval.toMinutes(), TimeUnit.MINUTES)
             }
         })
     }

node/src/main/kotlin/net/corda/node/internal/NetworkParametersReader.kt

@@ -21,11 +21,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: Exception) {
@@ -43,17 +46,17 @@ 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 {
@@ -74,14 +77,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
     }
 }

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

@@ -151,7 +151,7 @@ class NetworkMapUpdater(private val networkMapCache: NetworkMapCacheInternal,
         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 newNetParams = signedNewNetParams.verifiedNetworkMapCert(networkMapClient.trustedRoot)
-        val newParametersHash = newNetParams.serialize().hash
+        val newParametersHash = signedNewNetParams.raw.hash
         if (parametersHash == newParametersHash) {
             // The latest parameters have priority.
             signedNewNetParams.serialize()

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

@@ -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.*
@@ -22,6 +24,7 @@ import org.junit.Test
 import java.net.URL
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
+import kotlin.test.assertNotNull
 
 class NetworkParametersReaderTest {
     @Rule
@@ -72,4 +75,14 @@ class NetworkParametersReaderTest {
         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)
+    }
 }

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

@@ -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
@@ -77,6 +78,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)
     }
@@ -102,7 +105,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)

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

@@ -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
     )
 }