Support Ripple blockchain (#630)

* Support Ripple blockchain
2025-02-13 15:13:31 +01:00
parent 8567182088
commit eca16ba30d
13 changed files with 463 additions and 16 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
@@ -16,6 +16,7 @@
 */
 package io.emeraldpay.dshackle.rpc

+import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
@@ -481,17 +482,24 @@ open class NativeCall(
            ReaderData(ctx.upstream, ctx.upstreamFilter, ctx.callQuorum, signer, tracer),
        )
        val counter = reader.attempts()
+        val isRipple = ctx.upstream.getChain() in listOf(Chain.RIPPLE__MAINNET, Chain.RIPPLE__TESTNET)
+        var streamRequest = ctx.streamRequest
+        if (isRipple) {
+            streamRequest = false
+        }

        return SpannedReader(reader, tracer, RPC_READER)
-            .read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, ctx.streamRequest))
+            .read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, streamRequest))
            .map {
                val resolvedUpstreamData = it.resolvedUpstreamData.ifEmpty {
                    ctx.upstream.getUpstreamSettingsData()?.run { listOf(this) } ?: emptyList()
                }
                if (it.stream == null) {
-                    val bytes = ctx.resultDecorator.processResult(it)
-                    validateResult(bytes, "remote", ctx)
-                    CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, resolvedUpstreamData, ctx)
+                    if (isRipple) {
+                        callRippleResult(ctx, it, resolvedUpstreamData)
+                    } else {
+                        callResult(ctx, it, resolvedUpstreamData)
+                    }
                } else {
                    CallResult.ok(ctx.id, ctx.nonce, ByteArray(0), it.signature, resolvedUpstreamData, ctx, it.stream)
                }
@@ -515,6 +523,67 @@ open class NativeCall(
            )
    }

+    private fun callResult(
+        ctx: ValidCallContext<ParsedCallDetails>,
+        it: RequestReader.Result,
+        resolvedUpstreamData: List<Upstream.UpstreamSettingsData>,
+    ): CallResult {
+        val bytes = ctx.resultDecorator.processResult(it)
+        validateResult(bytes, "remote", ctx)
+        return CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, resolvedUpstreamData, ctx)
+    }
+
+    private fun callRippleResult(
+        ctx: ValidCallContext<ParsedCallDetails>,
+        it: RequestReader.Result,
+        resolvedUpstreamData: List<Upstream.UpstreamSettingsData>,
+    ): CallResult {
+        return try {
+            val responseJson = Global.objectMapper.readTree(it.value)
+            if (isRippleErrorResponse(responseJson)) {
+                return createRippleErrorResult(ctx, it, responseJson)
+            }
+            callResult(ctx, it, resolvedUpstreamData)
+        } catch (e: Exception) {
+            log.warn("Failed to parse Ripple response: ${e.message}", e)
+            callResult(ctx, it, resolvedUpstreamData)
+        }
+    }
+
+    private fun isRippleErrorResponse(responseJson: JsonNode): Boolean =
+        !responseJson.has("status") || responseJson.get("status").asText() == "error"
+
+    private fun createRippleErrorResult(
+        ctx: ValidCallContext<ParsedCallDetails>,
+        result: RequestReader.Result,
+        responseJson: JsonNode,
+    ): CallResult {
+        val error = responseJson.get("error")?.asText().orEmpty()
+        val errorMessage = responseJson.get("error_message")?.asText().orEmpty()
+        val fullErrorMessage = if (error.isNotEmpty() && errorMessage.isNotEmpty()) {
+            "$error: $errorMessage"
+        } else {
+            error.ifEmpty { errorMessage.ifEmpty { "Ripple request failed" } }
+        }
+
+        val errorCode = responseJson.get("error_code")?.asInt() ?: RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE
+
+        return CallResult.fail(
+            ctx.id,
+            ctx.nonce,
+            CallError(
+                ctx.id,
+                fullErrorMessage,
+                ChainCallError(
+                    errorCode,
+                    fullErrorMessage,
+                ),
+                String(result.value),
+            ),
+            ctx,
+        )
+    }
+
    private fun validateResult(bytes: ByteArray, origin: String, ctx: ValidCallContext<ParsedCallDetails>) {
        if (bytes.isEmpty() || nullValue.contentEquals(bytes)) {
            log.warn("Empty result from origin $origin, method ${ctx.payload.method}, params ${ctx.payload.params}")
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CallTargetsHolder.kt
@@ -6,6 +6,7 @@ import io.emeraldpay.dshackle.BlockchainType.ETHEREUM
 import io.emeraldpay.dshackle.BlockchainType.ETHEREUM_BEACON_CHAIN
 import io.emeraldpay.dshackle.BlockchainType.NEAR
 import io.emeraldpay.dshackle.BlockchainType.POLKADOT
+import io.emeraldpay.dshackle.BlockchainType.RIPPLE
 import io.emeraldpay.dshackle.BlockchainType.SOLANA
 import io.emeraldpay.dshackle.BlockchainType.STARKNET
 import io.emeraldpay.dshackle.BlockchainType.TON
@@ -20,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.calls.DefaultCosmosMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultNearMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultPolkadotMethods
+import io.emeraldpay.dshackle.upstream.calls.DefaultRippleMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultStarknetMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultTonHttpMethods
 import org.springframework.stereotype.Component
@@ -51,6 +53,7 @@ class CallTargetsHolder {
            ETHEREUM_BEACON_CHAIN -> DefaultBeaconChainMethods()
            COSMOS -> DefaultCosmosMethods()
            TON -> DefaultTonHttpMethods(connection)
+            RIPPLE -> DefaultRippleMethods()
            UNKNOWN -> throw IllegalArgumentException("unknown chain")
        }
        callTargets[chain] = created
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultRippleMethods.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultRippleMethods.kt
@@ -0,0 +1,95 @@
+package io.emeraldpay.dshackle.upstream.calls
+
+import io.emeraldpay.dshackle.quorum.AlwaysQuorum
+import io.emeraldpay.dshackle.quorum.BroadcastQuorum
+import io.emeraldpay.dshackle.quorum.CallQuorum
+import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
+
+class DefaultRippleMethods : CallMethods {
+
+    private val all = setOf(
+        "account_channels",
+        "account_currencies",
+        "account_info",
+        "account_lines",
+        "account_nfts",
+        "account_objects",
+        "account_offers",
+        "account_tx",
+        "gateway_balances",
+        "noripple_check",
+        "ledger",
+        "ledger_closed",
+        "ledger_current",
+        "ledger_data",
+        "ledger_entry",
+        "transaction_entry",
+        "tx",
+        "tx_history",
+        "book_offers",
+        "deposit_authorized",
+        "nft_buy_offers",
+        "nft_sell_offers",
+        "path_find",
+        "ripple_path_find",
+        "channel_authorize",
+        "channel_verify",
+        "subscribe",
+        "unsubscribe",
+        "fee",
+        "manifest",
+        "server_info",
+        "server_state",
+        "ledger_index",
+        "nft_history",
+        "nft_info",
+        "nfts_by_issuer",
+        "ping",
+        "random",
+//        "amm_info",
+//        "book_changes",
+//        "get_aggregate_price",
+//        "server_definitions",
+//        "version",
+//        "server_info",
+//        "ledger",
+//        "mpt_holders",
+//        "version",
+    )
+
+    private val add = setOf(
+        "submit",
+        "submit_multisigned",
+    )
+
+    private val allowedMethods: Set<String> = all + add
+
+    override fun createQuorumFor(method: String): CallQuorum {
+        if (add.contains(method)) {
+            return BroadcastQuorum()
+        }
+        return AlwaysQuorum()
+    }
+
+    override fun isCallable(method: String): Boolean {
+        return allowedMethods.contains(method)
+    }
+
+    override fun isHardcoded(method: String): Boolean {
+        return false
+    }
+
+    override fun executeHardcoded(method: String): ByteArray {
+        throw RpcException(-32601, "Method not found")
+    }
+
+    override fun getGroupMethods(groupName: String): Set<String> =
+        when (groupName) {
+            "default" -> getSupportedMethods()
+            else -> emptyList()
+        }.toSet()
+
+    override fun getSupportedMethods(): Set<String> {
+        return allowedMethods.toSortedSet()
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt
@@ -6,6 +6,7 @@ import io.emeraldpay.dshackle.BlockchainType.ETHEREUM
 import io.emeraldpay.dshackle.BlockchainType.ETHEREUM_BEACON_CHAIN
 import io.emeraldpay.dshackle.BlockchainType.NEAR
 import io.emeraldpay.dshackle.BlockchainType.POLKADOT
+import io.emeraldpay.dshackle.BlockchainType.RIPPLE
 import io.emeraldpay.dshackle.BlockchainType.SOLANA
 import io.emeraldpay.dshackle.BlockchainType.STARKNET
 import io.emeraldpay.dshackle.BlockchainType.TON
@@ -40,6 +41,7 @@ import io.emeraldpay.dshackle.upstream.finalization.FinalizationDetector
 import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService
 import io.emeraldpay.dshackle.upstream.near.NearChainSpecific
 import io.emeraldpay.dshackle.upstream.polkadot.PolkadotChainSpecific
+import io.emeraldpay.dshackle.upstream.ripple.RippleChainSpecific
 import io.emeraldpay.dshackle.upstream.solana.SolanaChainSpecific
 import io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific
 import io.emeraldpay.dshackle.upstream.ton.TonHttpSpecific
@@ -112,6 +114,7 @@ object ChainSpecificRegistry {
            ETHEREUM_BEACON_CHAIN -> BeaconChainSpecific
            TON -> TonHttpSpecific
            COSMOS -> CosmosChainSpecific
+            RIPPLE -> RippleChainSpecific
            BITCOIN -> throw IllegalArgumentException("bitcoin should use custom streams implementation")
            UNKNOWN -> throw IllegalArgumentException("unknown chain")
        }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt
@@ -0,0 +1,235 @@
+package io.emeraldpay.dshackle.upstream.ripple
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties
+import com.fasterxml.jackson.annotation.JsonProperty
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.foundation.ChainOptions.Options
+import io.emeraldpay.dshackle.reader.ChainReader
+import io.emeraldpay.dshackle.upstream.ChainRequest
+import io.emeraldpay.dshackle.upstream.GenericSingleCallValidator
+import io.emeraldpay.dshackle.upstream.SingleValidator
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
+import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService
+import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import java.math.BigInteger
+import java.time.Instant
+
+object RippleChainSpecific : AbstractPollChainSpecific() {
+
+    private val log = LoggerFactory.getLogger(RippleChainSpecific::class.java)
+
+    override fun parseBlock(data: ByteArray, upstreamId: String, api: ChainReader): Mono<BlockContainer> {
+        val result = Global.objectMapper.readValue(data, RippleBlock::class.java)
+        val block = result.closed.ledger
+        var height: Long = 0
+        try {
+            height = block.ledgerIndex.toLong()
+        } catch (e: NumberFormatException) {
+            log.error("Invalid ledgerIndex $block.ledgerIndex, upstreamId:$upstreamId ", result)
+        }
+
+        return Mono.just(
+            BlockContainer(
+                height = height,
+                hash = BlockId.from(block.ledgerHash ?: ""),
+                difficulty = BigInteger.ZERO,
+                timestamp = Instant.EPOCH,
+                full = false,
+                json = data,
+                parsed = result,
+                transactions = emptyList(),
+                upstreamId = upstreamId,
+                parentHash = BlockId.from(block.parentHash),
+            ),
+        )
+    }
+
+    override fun getFromHeader(data: ByteArray, upstreamId: String, api: ChainReader): Mono<BlockContainer> {
+        throw NotImplementedError()
+    }
+
+    override fun listenNewHeadsRequest(): ChainRequest {
+        throw NotImplementedError()
+    }
+
+    override fun unsubscribeNewHeadsRequest(subId: String): ChainRequest {
+        throw NotImplementedError()
+    }
+
+    override fun upstreamValidators(
+        chain: Chain,
+        upstream: Upstream,
+        options: Options,
+        config: ChainConfig,
+    ): List<SingleValidator<UpstreamAvailability>> {
+        return listOf(
+            GenericSingleCallValidator(
+                ChainRequest("server_state", ListParams()),
+                upstream,
+            ) { data -> validate(data) },
+        )
+    }
+
+    override fun upstreamSettingsValidators(
+        chain: Chain,
+        upstream: Upstream,
+        options: Options,
+        config: ChainConfig,
+    ): List<SingleValidator<ValidateUpstreamSettingsResult>> {
+        return listOf(
+            GenericSingleCallValidator(
+                ChainRequest("server_state", ListParams()),
+                upstream,
+            ) { data ->
+                validateSettings(data, chain)
+            },
+        )
+    }
+
+    override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService {
+        return RippleLowerBoundService(chain, upstream)
+    }
+
+    fun validate(data: ByteArray): UpstreamAvailability {
+        val resp = Global.objectMapper.readValue(data, RippleState::class.java)
+        return when (resp.state.serverState) {
+            "full", "proposing" -> UpstreamAvailability.OK
+            "connected" -> UpstreamAvailability.SYNCING
+            else -> UpstreamAvailability.UNAVAILABLE
+        }
+    }
+
+    fun validateSettings(data: ByteArray, chain: Chain): ValidateUpstreamSettingsResult {
+        val resp = Global.objectMapper.readValue(data, RippleState::class.java)
+        return if (chain.chainId.isNotEmpty() && resp.state.networkId.toString() != chain.chainId.lowercase()) {
+            ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR
+        } else {
+            ValidateUpstreamSettingsResult.UPSTREAM_VALID
+        }
+    }
+
+    override fun latestBlockRequest(): ChainRequest =
+        ChainRequest("ledger", ListParams())
+}
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class RippleBlock(
+    @JsonProperty("closed") var closed: RippleClosed,
+    @JsonProperty("open") var open: RippleOpen?,
+    @JsonProperty("status ") var status: String?,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class RippleClosed(
+    @JsonProperty("ledger") var ledger: RippleClosedLedger,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class RippleOpen(
+    @JsonProperty("ledger") var ledger: RippleOpenLedger,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class RippleClosedLedger(
+    @JsonProperty("account_hash") var accountHash: String,
+    @JsonProperty("close_flags") var closeFlags: Short,
+    @JsonProperty("close_time") var closeTime: Long,
+    @JsonProperty("close_time_human") var closeTimeHuman: String,
+    @JsonProperty("close_time_iso") var closeTimeIso: String,
+    @JsonProperty("close_time_resolution") var closeTimeResolution: Short,
+    @JsonProperty("closed") var closed: Boolean,
+    @JsonProperty("ledger_hash") var ledgerHash: String,
+    @JsonProperty("ledger_index") var ledgerIndex: String,
+    @JsonProperty("parent_close_time") var parentCloseTime: Long,
+    @JsonProperty("parent_hash") var parentHash: String,
+    @JsonProperty("total_coins") var totalCoins: String,
+    @JsonProperty("transaction_hash") var transactionHash: String,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class RippleOpenLedger(
+    @JsonProperty("closed") var closed: Boolean,
+    @JsonProperty("ledger_index") var ledgerIndex: String,
+    @JsonProperty("parent_hash") var parentHash: String,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class RippleState(
+    @JsonProperty("state") var state: RippleServerState,
+    @JsonProperty("status") var status: String? = null,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class RippleServerState(
+    @JsonProperty("build_version") val buildVersion: String,
+    @JsonProperty("complete_ledgers") val completeLedgers: String,
+    @JsonProperty("initial_sync_duration_us") val initialSyncDurationUs: String,
+    @JsonProperty("io_latency_ms") val ioLatencyMs: Int,
+    @JsonProperty("jq_trans_overflow") val jqTransOverflow: String,
+    @JsonProperty("last_close") val lastClose: LastClose,
+    @JsonProperty("load_base") val loadBase: Int,
+    @JsonProperty("load_factor") val loadFactor: Int,
+    @JsonProperty("load_factor_fee_escalation") val loadFactorFeeEscalation: Int,
+    @JsonProperty("load_factor_fee_queue") val loadFactorFeeQueue: Int,
+    @JsonProperty("load_factor_fee_reference") val loadFactorFeeReference: Int,
+    @JsonProperty("load_factor_server") val loadFactorServer: Int,
+    @JsonProperty("network_id") val networkId: Int,
+    @JsonProperty("peer_disconnects") val peerDisconnects: String,
+    @JsonProperty("peer_disconnects_resources") val peerDisconnectsResources: String,
+    @JsonProperty("peers") val peers: Int,
+    @JsonProperty("ports") val ports: List<Port>,
+    @JsonProperty("pubkey_node") val pubkeyNode: String,
+    @JsonProperty("server_state") val serverState: String,
+    @JsonProperty("server_state_duration_us") val serverStateDurationUs: String,
+    @JsonProperty("state_accounting") val stateAccounting: StateAccounting,
+    @JsonProperty("time") val time: String,
+    @JsonProperty("uptime") val uptime: Long,
+    @JsonProperty("validated_ledger") val validatedLedger: ValidatedLedger,
+    @JsonProperty("validation_quorum") val validationQuorum: Int,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class LastClose(
+    @JsonProperty("converge_time") val convergeTime: Int,
+    @JsonProperty("proposers") val proposers: Int,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class Port(
+    @JsonProperty("port") val port: String,
+    @JsonProperty("protocol") val protocol: List<String>,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class StateAccounting(
+    @JsonProperty("connected") val connected: StateDuration,
+    @JsonProperty("disconnected") val disconnected: StateDuration,
+    @JsonProperty("full") val full: StateDuration,
+    @JsonProperty("syncing") val syncing: StateDuration,
+    @JsonProperty("tracking") val tracking: StateDuration,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class StateDuration(
+    @JsonProperty("duration_us") val durationUs: String,
+    @JsonProperty("transitions") val transitions: String,
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class ValidatedLedger(
+    @JsonProperty("base_fee") val baseFee: Int,
+    @JsonProperty("close_time") val closeTime: Long,
+    @JsonProperty("hash") val hash: String,
+    @JsonProperty("reserve_base") val reserveBase: Long,
+    @JsonProperty("reserve_inc") val reserveInc: Long,
+    @JsonProperty("seq") val seq: Long,
+)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundService.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundService.kt
@@ -0,0 +1,15 @@
+package io.emeraldpay.dshackle.upstream.ripple
+
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService
+
+class RippleLowerBoundService(
+    chain: Chain,
+    private val upstream: Upstream,
+) : LowerBoundService(chain, upstream) {
+    override fun detectors(): List<LowerBoundDetector> {
+        return listOf(RippleLowerBoundStateDetector(upstream))
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundStateDetector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleLowerBoundStateDetector.kt
@@ -0,0 +1,24 @@
+package io.emeraldpay.dshackle.upstream.ripple
+
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
+import reactor.core.publisher.Flux
+
+class RippleLowerBoundStateDetector(
+    private val upstream: Upstream,
+) : LowerBoundDetector(upstream.getChain()) {
+
+    override fun period(): Long {
+        return 120
+    }
+
+    override fun internalDetectLowerBound(): Flux<LowerBoundData> {
+        return Flux.just(LowerBoundData(1, LowerBoundType.STATE))
+    }
+
+    override fun types(): Set<LowerBoundType> {
+        return setOf(LowerBoundType.STATE)
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt
@@ -218,7 +218,8 @@ abstract class ResponseParser<T> {

        private val isResultSet = result != null || nullResult

-        val isRpcReady: Boolean = id != null &&
+        // Ripple Response doesn't have `id` field
+        val isRpcReady: Boolean = // id != null &&
            (error != null || isResultSet)

        val isSubReady: Boolean = subId != null &&
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt
@@ -30,14 +30,6 @@ class ResponseWSParser : ResponseParser<ResponseWSParser.WsResponse>() {
    }

    override fun build(state: Preparsed): WsResponse {
-        if (state.isRpcReady) {
-            return WsResponse(
-                Type.RPC,
-                state.id!!,
-                if (state.nullResult) NULL_RESULT else state.result,
-                state.error,
-            )
-        }
        if (state.isSubReady) {
            return WsResponse(
                Type.SUBSCRIPTION,
@@ -46,6 +38,14 @@ class ResponseWSParser : ResponseParser<ResponseWSParser.WsResponse>() {
                state.error,
            )
        }
+        if (state.isRpcReady) {
+            return WsResponse(
+                Type.RPC,
+                state.id!!,
+                if (state.nullResult) NULL_RESULT else state.result,
+                state.error,
+            )
+        }
        throw IllegalStateException("State is not ready")
    }