Fix merge upstream options (#182)

* Fix merge upstream options

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -18,7 +18,9 @@ package io.emeraldpay.dshackle.config
 
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode
+import org.apache.commons.lang3.ObjectUtils.firstNonNull
 import java.net.URI
+import java.time.Duration
 import java.util.Arrays
 import java.util.Locale
 
@@ -26,46 +28,67 @@ open class UpstreamsConfig {
     var defaultOptions: MutableList<DefaultOptions> = ArrayList<DefaultOptions>()
     var upstreams: MutableList<Upstream<*>> = ArrayList<Upstream<*>>()
 
-    open class Options {
+    data class Options(
+        val disableValidation: Boolean,
+        val validationInterval: Int,
+        val timeout: Duration,
+        val providesBalance: Boolean?,
+        val validatePeers: Boolean,
+        val minPeers: Int,
+        val validateSyncing: Boolean
+    )
+
+    open class PartialOptions {
         var disableValidation: Boolean? = null
-        var validationInterval: Int = 30
+        var validationInterval: Int? = null
             set(value) {
-                require(value > 0) {
+                require(value == null || value > 0) {
                     "validation-interval must be a positive number: $value"
                 }
                 field = value
             }
-        var timeout = Defaults.timeout
+        var timeout: Duration? = null
         var providesBalance: Boolean? = null
-        var validatePeers: Boolean = true
+        var validatePeers: Boolean? = null
-        var minPeers: Int? = 1
+        var minPeers: Int? = null
             set(value) {
-                require(value != null && value >= 0) {
+                require(value == null || value >= 0) {
                     "min-peers must be a positive number: $value"
                 }
                 field = value
             }
-        var validateSyncing: Boolean = true
+        var validateSyncing: Boolean? = null
-        fun merge(overwrites: Options?): Options {
+
+        fun merge(overwrites: PartialOptions?): PartialOptions {
             if (overwrites == null) {
                 return this
             }
-            val copy = Options()
+            val copy = PartialOptions()
-            copy.validatePeers = this.validatePeers && overwrites.validatePeers
+            copy.validatePeers = firstNonNull(overwrites.validatePeers, this.validatePeers)
-            copy.minPeers = if (this.minPeers != null) this.minPeers else overwrites.minPeers
+            copy.minPeers = firstNonNull(overwrites.minPeers, this.minPeers)
-            copy.disableValidation =
+            copy.disableValidation = firstNonNull(overwrites.disableValidation, this.disableValidation)
-                if (this.disableValidation != null) this.disableValidation else overwrites.disableValidation
+            copy.validationInterval = firstNonNull(overwrites.validationInterval, this.validationInterval)
-            copy.validationInterval = overwrites.validationInterval
+            copy.providesBalance = firstNonNull(overwrites.providesBalance, this.providesBalance)
-            copy.providesBalance =
+            copy.validateSyncing = firstNonNull(overwrites.validateSyncing, this.validateSyncing)
-                if (this.providesBalance != null) this.providesBalance else overwrites.providesBalance
+            copy.timeout = firstNonNull(overwrites.timeout, this.timeout)
-            copy.validateSyncing = this.validateSyncing && overwrites.validateSyncing
             return copy
         }
 
+        fun buildOptions(): Options =
+            Options(
+                firstNonNull(this.disableValidation, false)!!,
+                firstNonNull(this.validationInterval, 30)!!,
+                firstNonNull(this.timeout, Defaults.timeout)!!,
+                this.providesBalance,
+                firstNonNull(this.validatePeers, true)!!,
+                firstNonNull(this.minPeers, 1)!!,
+                firstNonNull(this.validateSyncing, true)!!
+            )
+
         companion object {
             @JvmStatic
-            fun getDefaults(): Options {
+            fun getDefaults(): PartialOptions {
-                val options = Options()
+                val options = PartialOptions()
                 options.minPeers = 1
                 options.disableValidation = false
                 return options
@@ -73,16 +96,16 @@ open class UpstreamsConfig {
         }
     }
 
-    class DefaultOptions : Options() {
+    class DefaultOptions : PartialOptions() {
         var chains: List<String>? = null
-        var options: Options? = null
+        var options: PartialOptions? = null
     }
 
     class Upstream<T : UpstreamConnection> {
         var id: String? = null
         var nodeId: Int? = null
         var chain: String? = null
-        var options: Options? = null
+        var options: PartialOptions? = null
         var isEnabled = true
         var connection: T? = null
         val labels = Labels()

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -303,7 +303,7 @@ class UpstreamsConfigReader(
         }
     }
 
-    internal fun tryReadOptions(upNode: MappingNode): UpstreamsConfig.Options? {
+    internal fun tryReadOptions(upNode: MappingNode): UpstreamsConfig.PartialOptions? {
         return if (hasAny(upNode, "options")) {
             return getMapping(upNode, "options")?.let { values ->
                 readOptions(values)
@@ -348,8 +348,8 @@ class UpstreamsConfigReader(
         }
     }
 
-    internal fun readOptions(values: MappingNode): UpstreamsConfig.Options {
+    internal fun readOptions(values: MappingNode): UpstreamsConfig.PartialOptions {
-        val options = UpstreamsConfig.Options()
+        val options = UpstreamsConfig.PartialOptions()
         getValueAsBool(values, "validate-peers")?.let {
             options.validatePeers = it
         }

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -94,16 +94,17 @@ open class ConfiguredUpstreams(
             }
             log.debug("Start upstream ${up.id}")
             if (up.connection is UpstreamsConfig.GrpcConnection) {
-                val options = up.options ?: UpstreamsConfig.Options()
+                val options = up.options ?: UpstreamsConfig.PartialOptions()
-                buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options, compressionConfig.grpc.clientEnabled)
+                buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options.buildOptions(), compressionConfig.grpc.clientEnabled)
             } else {
                 val chain = Global.chainById(up.chain)
                 if (chain == Chain.UNSPECIFIED) {
                     log.error("Chain is unknown: ${up.chain}")
                     return@forEach
                 }
-                val options = (defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults())
+                val options = (defaultOptions[chain] ?: UpstreamsConfig.PartialOptions.getDefaults())
-                    .merge(up.options ?: UpstreamsConfig.Options())
+                    .merge(up.options ?: UpstreamsConfig.PartialOptions())
+                    .buildOptions()
                 val upstream = when (BlockchainType.from(chain)) {
                     BlockchainType.EVM_POS -> {
                         buildEthereumPosUpstream(
@@ -140,8 +141,8 @@ open class ConfiguredUpstreams(
         }
     }
 
-    private fun buildDefaultOptions(config: UpstreamsConfig): HashMap<Chain, UpstreamsConfig.Options> {
+    private fun buildDefaultOptions(config: UpstreamsConfig): HashMap<Chain, UpstreamsConfig.PartialOptions> {
-        val defaultOptions = HashMap<Chain, UpstreamsConfig.Options>()
+        val defaultOptions = HashMap<Chain, UpstreamsConfig.PartialOptions>()
         config.defaultOptions.forEach { defaultsConfig ->
             defaultsConfig.chains?.forEach { chainName ->
                 Global.chainById(chainName).let { chain ->

src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt

@@ -87,7 +87,7 @@ abstract class DefaultUpstream(
     }
 
     private fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability {
-        if (options.disableValidation == true) {
+        if (options.disableValidation) {
             // if we specifically told that this upstream should be _always valid_ then skip
             // the status calculation and trust the proposed value as is
             return proposed

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -242,7 +242,7 @@ abstract class Multistream(
 
     // TODO options for multistream are useless
     override fun getOptions(): UpstreamsConfig.Options {
-        return UpstreamsConfig.Options()
+        throw IllegalStateException("Options are not supported for multistream")
     }
 
     // TODO roles for multistream are useless

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt

@@ -102,7 +102,7 @@ open class BitcoinRpcUpstream(
 
         validatorSubscription?.dispose()
 
-        if (getOptions().disableValidation != null && getOptions().disableValidation!!) {
+        if (getOptions().disableValidation) {
             this.setLag(0)
             this.setStatus(UpstreamAvailability.OK)
         } else {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt

@@ -57,7 +57,7 @@ open class EthereumRpcUpstream(
     override fun start() {
         log.info("Configured for ${chain.chainName}")
         connector.start()
-        if (getOptions().disableValidation != null && getOptions().disableValidation!!) {
+        if (getOptions().disableValidation) {
             log.warn("Disable validation for upstream ${this.getId()}")
             this.setLag(0)
             this.setStatus(UpstreamAvailability.OK)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt

@@ -57,7 +57,7 @@ open class EthereumPosRpcUpstream(
     override fun start() {
         log.info("Configured for ${chain.chainName}")
         connector.start()
-        if (getOptions().disableValidation != null && getOptions().disableValidation!!) {
+        if (getOptions().disableValidation) {
             log.warn("Disable validation for upstream ${this.getId()}")
             this.setLag(0)
             this.setStatus(UpstreamAvailability.OK)

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt

@@ -57,7 +57,7 @@ class BitcoinGrpcUpstream(
 ) : BitcoinUpstream(
     "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
     chain,
-    UpstreamsConfig.Options.getDefaults(),
+    UpstreamsConfig.PartialOptions.getDefaults().buildOptions(),
     role,
     chainConfig
 ),

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt

@@ -63,7 +63,7 @@ open class EthereumGrpcUpstream(
 ) : EthereumUpstream(
     "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
     hash,
-    UpstreamsConfig.Options.getDefaults(),
+    UpstreamsConfig.PartialOptions.getDefaults().buildOptions(),
     role,
     null,
     null,

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt

@@ -57,7 +57,7 @@ open class EthereumPosGrpcUpstream(
 ) : EthereumPosUpstream(
     "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
     hash,
-    UpstreamsConfig.Options.getDefaults(),
+    UpstreamsConfig.PartialOptions.getDefaults().buildOptions(),
     role,
     null,
     null,

src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy

@@ -16,9 +16,12 @@
  */
 package io.emeraldpay.dshackle.config
 
+
 import io.emeraldpay.dshackle.test.TestingCommons
 import spock.lang.Specification
 
+import java.time.Duration
+
 class UpstreamsConfigReaderSpec extends Specification {
 
     UpstreamsConfigReader reader = new UpstreamsConfigReader(TestingCommons.fileResolver())
@@ -481,4 +484,156 @@ class UpstreamsConfigReaderSpec extends Specification {
             }
         }
     }
+
+    def "Merge options for disableValidation"() {
+        expect:
+        def a = new UpstreamsConfig.PartialOptions().tap { disableValidation = base }
+        def b = new UpstreamsConfig.PartialOptions().tap { disableValidation = overwrite }
+        def result = a.merge(b).buildOptions()
+        result.disableValidation == exp
+
+        where:
+        base        | overwrite     | exp
+        true        | true          | true
+        true        | false         | false
+        true        | null          | true
+
+        false       | true          | true
+        false       | false         | false
+        false       | null          | false
+
+        null        | true          | true
+        null        | false         | false
+        null        | null          | false
+    }
+
+    def "Merge options for providesBalance"() {
+        expect:
+        def a = new UpstreamsConfig.PartialOptions().tap { providesBalance = base }
+        def b = new UpstreamsConfig.PartialOptions().tap { providesBalance = overwrite }
+        def result = a.merge(b).buildOptions()
+        result.providesBalance == exp
+
+        where:
+        base        | overwrite     | exp
+        true        | true          | true
+        true        | false         | false
+        true        | null          | true
+
+        false       | true          | true
+        false       | false         | false
+        false       | null          | false
+
+        null        | true          | true
+        null        | false         | false
+        null        | null          | null
+    }
+
+    def "Merge options for validatePeers"() {
+        expect:
+        def a = new UpstreamsConfig.PartialOptions().tap { validatePeers = base }
+        def b = new UpstreamsConfig.PartialOptions().tap { validatePeers = overwrite }
+        def result = a.merge(b).buildOptions()
+        result.validatePeers == exp
+
+        where:
+        base        | overwrite     | exp
+        true        | true          | true
+        true        | false         | false
+        true        | null          | true
+
+        false       | true          | true
+        false       | false         | false
+        false       | null          | false
+
+        null        | true          | true
+        null        | false         | false
+        null        | null          | true
+    }
+
+    def "Merge options for validateSyncing"() {
+        expect:
+        def a = new UpstreamsConfig.PartialOptions().tap { validateSyncing = base }
+        def b = new UpstreamsConfig.PartialOptions().tap { validateSyncing = overwrite }
+        def result = a.merge(b).buildOptions()
+        result.validateSyncing == exp
+
+        where:
+        base        | overwrite     | exp
+        true        | true          | true
+        true        | false         | false
+        true        | null          | true
+
+        false       | true          | true
+        false       | false         | false
+        false       | null          | false
+
+        null        | true          | true
+        null        | false         | false
+        null        | null          | true
+    }
+
+    def "Merge options for timeout"() {
+        expect:
+        def a = new UpstreamsConfig.PartialOptions().tap {
+            timeout = base == null ? null : Duration.ofSeconds(base)
+        }
+        def b = new UpstreamsConfig.PartialOptions().tap {
+            timeout = overwrite == null ? null : Duration.ofSeconds(overwrite)
+        }
+        def result = a.merge(b).buildOptions()
+        def expValue = exp == null ? Duration.ofSeconds(60) : Duration.ofSeconds(exp)
+        result.timeout == expValue
+
+        where:
+        base     | overwrite  | exp
+        1        | 2          | 2
+        3        | 4          | 4
+        5        | null       | 5
+        null     | 6          | 6
+        null     | null       | null
+    }
+
+    def "Merge options for minPeers"() {
+        expect:
+        def a = new UpstreamsConfig.PartialOptions().tap { minPeers = base }
+        def b = new UpstreamsConfig.PartialOptions().tap { minPeers = overwrite }
+        def result = a.merge(b).buildOptions()
+        result.minPeers == exp
+
+        where:
+        base     | overwrite  | exp
+        1        | 2          | 2
+        3        | 4          | 4
+        5        | null       | 5
+        null     | 6          | 6
+        null     | null       | 1
+    }
+
+    def "Merge options for validationInterval"() {
+        expect:
+        def a = new UpstreamsConfig.PartialOptions().tap { validationInterval = base }
+        def b = new UpstreamsConfig.PartialOptions().tap { validationInterval = overwrite }
+        def result = a.merge(b).buildOptions()
+        result.validationInterval == exp
+
+        where:
+        base     | overwrite  | exp
+        1        | 2          | 2
+        3        | 4          | 4
+        5        | null       | 5
+        null     | 6          | 6
+        null     | null       | 30
+    }
+
+    def "Options with default values"() {
+        setup:
+        def partialOptions = new UpstreamsConfig.PartialOptions()
+        when:
+        def options = partialOptions.buildOptions()
+        then:
+        options == new UpstreamsConfig.Options(
+                false, 30, Duration.ofSeconds(60), null, true, 1, true
+        )
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy

@@ -83,9 +83,9 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
     }
 
     static Options getOpts() {
-        def opt = UpstreamsConfig.Options.getDefaults()
+        def opt = UpstreamsConfig.PartialOptions.getDefaults()
         opt.setDisableValidation(true)
-        return opt
+        return opt.buildOptions()
     }
 
     void nextBlock(BlockContainer block) {

src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy

@@ -60,7 +60,7 @@ class EthereumRpcUpstreamMock extends EthereumRpcUpstream {
 
     EthereumRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
         super(id, id.hashCode().byteValue(), chain,
-                UpstreamsConfig.Options.getDefaults(),
+                UpstreamsConfig.PartialOptions.getDefaults().buildOptions(),
                 UpstreamsConfig.UpstreamRole.PRIMARY,
                 methods,
                 new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()),

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy

@@ -54,7 +54,7 @@ class FilteredApisSpec extends Specification {
                     "test",
                     (byte)123,
                     Chain.ETHEREUM,
-                    new UpstreamsConfig.Options(),
+                    new UpstreamsConfig.PartialOptions().buildOptions(),
                     UpstreamsConfig.UpstreamRole.PRIMARY,
                     ethereumTargets,
                     new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)),

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy

@@ -30,7 +30,7 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Resolve to final availability"() {
         setup:
-        def validator = new EthereumUpstreamValidator(Stub(EthereumUpstream), UpstreamsConfig.Options.getDefaults())
+        def validator = new EthereumUpstreamValidator(Stub(EthereumUpstream), UpstreamsConfig.PartialOptions.getDefaults().buildOptions())
         expect:
         validator.resolve(Tuples.of(sync, peers)) == exp
         where:
@@ -48,9 +48,9 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Doesnt check eth_syncing when disabled"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validateSyncing = false
-        }
+        }.buildOptions()
         def up = Mock(EthereumUpstream)
         def validator = new EthereumUpstreamValidator(up, options)
 
@@ -63,9 +63,9 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Syncing is OK when false returned from upstream"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validateSyncing = true
-        }
+        }.buildOptions()
         def up = TestingCommons.upstream(
                 new ApiReaderMock().tap {
                     answer("eth_syncing", [], false)
@@ -81,9 +81,9 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Syncing is SYNCING when state returned from upstream"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validateSyncing = true
-        }
+        }.buildOptions()
         def up = TestingCommons.upstream(
                 new ApiReaderMock().tap {
                     answer("eth_syncing", [], [startingBlock: 100, currentBlock: 50])
@@ -99,9 +99,9 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Syncing is UNAVAILABLE when error returned from upstream"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validateSyncing = true
-        }
+        }.buildOptions()
         def up = TestingCommons.upstream(
                 new ApiReaderMock().tap {
                     answer("eth_syncing", [], new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unavailable"))
@@ -117,10 +117,10 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Doesnt validate peers when disabled"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validatePeers = false
             it.minPeers = 10
-        }
+        }.buildOptions()
         def up = Mock(EthereumUpstream)
         def validator = new EthereumUpstreamValidator(up, options)
 
@@ -133,10 +133,10 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Doesnt validate peers when zero peers is expected"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validatePeers = true
             it.minPeers = 0
-        }
+        }.buildOptions()
         def up = Mock(EthereumUpstream)
         def validator = new EthereumUpstreamValidator(up, options)
 
@@ -149,10 +149,10 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Peers is IMMATURE when state returned too few peers"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validatePeers = true
             it.minPeers = 10
-        }
+        }.buildOptions()
         def up = TestingCommons.upstream(
                 new ApiReaderMock().tap {
                     answer("net_peerCount", [], "0x5")
@@ -168,10 +168,10 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Peers is OK when state returned exactly min peers"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validatePeers = true
             it.minPeers = 10
-        }
+        }.buildOptions()
         def up = TestingCommons.upstream(
                 new ApiReaderMock().tap {
                     answer("net_peerCount", [], "0xa")
@@ -187,10 +187,10 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Peers is OK when state returned more than enough peers"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validatePeers = true
             it.minPeers = 10
-        }
+        }.buildOptions()
         def up = TestingCommons.upstream(
                 new ApiReaderMock().tap {
                     answer("net_peerCount", [], "0xff")
@@ -206,10 +206,10 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Peers is UNAVAILABLE when state returned error"() {
         setup:
-        def options = UpstreamsConfig.Options.getDefaults().tap {
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validatePeers = true
             it.minPeers = 10
-        }
+        }.buildOptions()
         def up = TestingCommons.upstream(
                 new ApiReaderMock().tap {
                     answer("net_peerCount", [], new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unavailable"))