From 757febc0b539f38a2d5d3354050e2341d378ac91 Mon Sep 17 00:00:00 2001 From: Anton Date: Thu, 18 Dec 2025 11:30:18 +0300 Subject: [PATCH] Forward HTTP response headers from upstream to gRPC client for Beacon Chain (#759) Added response_headers field to NativeCallReplyItem proto. Implemented chain-specific header filtering via getResponseHeadersToForward(). BeaconChain forwards: Eth-Consensus-Version, Eth-Consensus-Finalized, Eth-Execution-Optimistic, Eth-Execution-Payload-Blinded/Value, Eth-Consensus-Block-Value. --- .gitignore | 3 +- emerald-grpc | 2 +- gradle/libs.versions.toml | 2 +- .../dshackle/quorum/QuorumRequestReader.kt | 4 +- .../dshackle/reader/BroadcastReader.kt | 1 + .../dshackle/reader/RequestReaderFactory.kt | 3 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 44 +++++++++++++------ .../dshackle/upstream/BasicHttpFactory.kt | 2 +- .../dshackle/upstream/ChainResponse.kt | 16 +++++-- .../beaconchain/BeaconChainSpecific.kt | 9 ++++ .../calls/DefaultBeaconChainMethods.kt | 8 +++- .../upstream/generic/ChainSpecific.kt | 6 +++ .../upstream/restclient/RestHttpReader.kt | 27 +++++++++--- .../dshackle/upstream/stream/Responses.kt | 2 + 14 files changed, 98 insertions(+), 31 deletions(-) diff --git a/.gitignore b/.gitignore index 5bf3ce9c..25ca0b16 100644 --- a/.gitignore +++ b/.gitignore @@ -25,4 +25,5 @@ Test*.kt http-client.env.json .DS_Store -mise.toml \ No newline at end of file +mise.toml +.claude/ \ No newline at end of file diff --git a/emerald-grpc b/emerald-grpc index 1698b6f9..f5c4aac8 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit 1698b6f9e332548fe263098f877cd8e08c9963f8 +Subproject commit f5c4aac8196db84b708149ed3c5f92b103682a0a diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 8ca6212b..928227fc 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -1,7 +1,7 @@ [versions] detekt = "1.23.1" groovy = "4.0.15" -protoc = "4.29.2" +protoc = "4.33.2" jackson = "2.11.0" grpc = "1.57.0" reactive-grpc = "1.2.0" diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt index d64a9f1c..e5aa5360 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt @@ -135,7 +135,7 @@ class QuorumRequestReader( .map { quorum -> val response = quorum.getResponse()!! // TODO find actual quorum number - Result(response.getResult(), quorum.getSignature(), 1, resolvedBy(), response.stream) + Result(response.getResult(), quorum.getSignature(), 1, resolvedBy(), response.stream, response.responseHeaders) } .switchIfEmpty(defaultResult) } @@ -236,7 +236,7 @@ class QuorumRequestReader( val cause = getCause(method) ?: return Mono.error(RpcException(1, "No response for method $method", getFullCause())) if (cause.shouldReturnNull) { Mono.just( - Result(Global.nullValue, null, 1, emptyList(), null), + Result(Global.nullValue, null, 1, emptyList(), null, emptyMap()), ) } else { Mono.error(RpcException(1, "No response for method $method. Cause - ${cause.cause}")) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt index 490952d3..72302139 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt @@ -64,6 +64,7 @@ class BroadcastReader( upstreams.size, upsData, null, + quorum.getResponse()!!.responseHeaders, ) Mono.just(res) } else { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt index bdb11fc5..23492e08 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt @@ -49,12 +49,13 @@ abstract class RequestReader( null } - class Result( + class Result @JvmOverloads constructor( val value: ByteArray, val signature: ResponseSigner.Signature?, val quorum: Int, val resolvedUpstreamData: List, val stream: Flux?, + val responseHeaders: Map = emptyMap(), ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index fd9e35ca..24bf55cb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -143,7 +143,7 @@ open class NativeCall( Flux.concat( Mono.just(firstChunk) .map { - val result = buildStreamResult(it, callResult.id) + val result = buildStreamResult(it, callResult.id, callResult.responseHeaders) if (callResult.upstreamSettingsData.isNotEmpty()) { getUpstreamIdsAndVersions(callResult.upstreamSettingsData) .let { idsAndVersions -> @@ -160,13 +160,22 @@ open class NativeCall( } } - private fun buildStreamResult(chunk: Chunk, id: Int): BlockchainOuterClass.NativeCallReplyItem.Builder { - return BlockchainOuterClass.NativeCallReplyItem.newBuilder() + private fun buildStreamResult(chunk: Chunk, id: Int, headers: Map = emptyMap()): BlockchainOuterClass.NativeCallReplyItem.Builder { + val builder = BlockchainOuterClass.NativeCallReplyItem.newBuilder() .setSucceed(true) .setFinalChunk(chunk.finalChunk) .setChunked(true) .setPayload(ByteString.copyFrom(chunk.chunkData)) .setId(id) + headers.forEach { (key, value) -> + builder.addResponseHeaders( + BlockchainOuterClass.KeyValue.newBuilder() + .setKey(key) + .setValue(value) + .build(), + ) + } + return builder } private fun completeSpan(callResult: CallResult, requestCount: Int) { @@ -263,6 +272,14 @@ open class NativeCall( .setType(it.type.toProtoFinalizationType()) .build() } + it.responseHeaders.forEach { (key, value) -> + result.addResponseHeaders( + BlockchainOuterClass.KeyValue.newBuilder() + .setKey(key) + .setValue(value) + .build(), + ) + } return result.build() } @@ -460,9 +477,9 @@ open class NativeCall( } else { ctx.upstream.getId() } - CallResult.ok(ctx.id, ctx.nonce, result, signer.sign(ctx.nonce, result, source), resolvedUpstreamData, ctx, it.finalization) + CallResult.ok(ctx.id, ctx.nonce, result, signer.sign(ctx.nonce, result, source), resolvedUpstreamData, ctx, it.finalization, it.responseHeaders) } else { - CallResult.ok(ctx.id, null, result, null, resolvedUpstreamData, ctx, it.finalization) + CallResult.ok(ctx.id, null, result, null, resolvedUpstreamData, ctx, it.finalization, it.responseHeaders) } } }.switchIfEmpty( @@ -505,7 +522,7 @@ open class NativeCall( callResult(ctx, it, resolvedUpstreamData) } } else { - CallResult.ok(ctx.id, ctx.nonce, ByteArray(0), it.signature, resolvedUpstreamData, ctx, it.stream) + CallResult.ok(ctx.id, ctx.nonce, ByteArray(0), it.signature, resolvedUpstreamData, ctx, it.stream, it.responseHeaders) } } .onErrorResume { t -> @@ -534,7 +551,7 @@ open class NativeCall( ): CallResult { val bytes = ctx.resultDecorator.processResult(it) validateResult(bytes, "remote", ctx) - return CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, resolvedUpstreamData, ctx) + return CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, resolvedUpstreamData, ctx, it.responseHeaders) } private fun callRippleResult( @@ -828,6 +845,7 @@ open class NativeCall( val ctx: ValidCallContext?, val stream: Flux? = null, val finalization: FinalizationData? = null, + val responseHeaders: Map = emptyMap(), ) { constructor( @@ -840,16 +858,16 @@ open class NativeCall( ) : this(id, nonce, result, callError, signature, callError?.upstreamSettingsData ?: emptyList(), ctx) companion object { - fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: List, ctx: ValidCallContext?): CallResult { - return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx) + fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: List, ctx: ValidCallContext?, responseHeaders: Map = emptyMap()): CallResult { + return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, null, null, responseHeaders) } - fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: List, ctx: ValidCallContext?, final: FinalizationData?): CallResult { - return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, null, final) + fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: List, ctx: ValidCallContext?, final: FinalizationData?, responseHeaders: Map = emptyMap()): CallResult { + return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, null, final, responseHeaders) } - fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: List, ctx: ValidCallContext?, stream: Flux?): CallResult { - return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, stream) + fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: List, ctx: ValidCallContext?, stream: Flux?, responseHeaders: Map = emptyMap()): CallResult { + return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, stream, null, responseHeaders) } fun fail(id: Int, nonce: Long?, error: CallError, ctx: ValidCallContext?): CallResult { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt index 5a9b90b6..e9ccad09 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt @@ -46,7 +46,7 @@ class BasicHttpFactory( ) if (chain.type.apiType == ApiType.REST) { - return RestHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, basicAuth, tls) + return RestHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, chain, basicAuth, tls) } return JsonRpcHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, basicAuth, tls) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt index d73bc865..6e49836b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt @@ -35,17 +35,25 @@ class ChainResponse @JvmOverloads constructor( val providedSignature: ResponseSigner.Signature? = null, val resolvedUpstreamData: List = emptyList(), val finalization: FinalizationData? = null, + val responseHeaders: Map = emptyMap(), ) { constructor(stream: Flux, id: Int) : - this(null, null, NumberId(id.toLong()), stream, null, emptyList(), null) + this(null, null, NumberId(id.toLong()), stream, null, emptyList(), null, emptyMap()) + + constructor(stream: Flux, id: Int, responseHeaders: Map) : + this(null, null, NumberId(id.toLong()), stream, null, emptyList(), null, responseHeaders) constructor(result: ByteArray?, error: ChainCallError?) : this(result, error, NumberId(0), null, null) + constructor(result: ByteArray?, error: ChainCallError?, responseHeaders: Map) : + this(result, error, NumberId(0), null, null, emptyList(), null, responseHeaders) + constructor(result: ByteArray?, error: ChainCallError?, resolvedUpstreamData: List) : - this(result, error, NumberId(0), null, null, resolvedUpstreamData, null) + this(result, error, NumberId(0), null, null, resolvedUpstreamData, null, emptyMap()) + constructor(result: ByteArray?, resolvedUpstreamData: List, finalization: FinalizationData) : - this(result, null, NumberId(0), null, null, resolvedUpstreamData, finalization) + this(result, null, NumberId(0), null, null, resolvedUpstreamData, finalization, emptyMap()) companion object { private val NULL_VALUE = "null".toByteArray() @@ -139,7 +147,7 @@ class ChainResponse @JvmOverloads constructor( } fun copyWithId(id: Id): ChainResponse { - return ChainResponse(result, error, id, stream, providedSignature, resolvedUpstreamData) + return ChainResponse(result, error, id, stream, providedSignature, resolvedUpstreamData, finalization, responseHeaders) } override fun equals(other: Any?): Boolean { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt index b7f33524..319f9307 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt @@ -121,6 +121,15 @@ object BeaconChainSpecific : AbstractPollChainSpecific() { override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService { return BeaconChainLowerBoundService(chain, upstream) } + + override fun getResponseHeadersToForward(): List = listOf( + "Eth-Consensus-Version", + "Eth-Consensus-Finalized", + "Eth-Execution-Optimistic", + "Eth-Execution-Payload-Blinded", + "Eth-Execution-Payload-Value", + "Eth-Consensus-Block-Value", + ) } data class BeaconChainBlockHeader( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBeaconChainMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBeaconChainMethods.kt index 8391f3d6..3e0d137b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBeaconChainMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBeaconChainMethods.kt @@ -117,7 +117,13 @@ class DefaultBeaconChainMethods : CallMethods { } override fun isCallable(method: String): Boolean { - return allowedMethods.contains(method) + if (allowedMethods.contains(method)) { + return true + } + // Check wildcard patterns (e.g., GET#/eth/v1/beacon/headers/* matches GET#/eth/v1/beacon/headers/head) + return allowedMethods.any { pattern -> + pattern.contains("*") && method.matches(pattern.replace("*", "[^/]+").toRegex()) + } } override fun getSupportedMethods(): Set { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt index fed279b8..d211f4fc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt @@ -101,6 +101,12 @@ interface ChainSpecific { fun callSelector(caches: Caches): CallSelector? fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService + + /** + * List of HTTP response header names to forward from upstream to client. + * Override in chain-specific implementations to specify relevant headers. + */ + fun getResponseHeadersToForward(): List = emptyList() } object ChainSpecificRegistry { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt index 78276960..b78ab8ec 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt @@ -1,11 +1,13 @@ package io.emeraldpay.dshackle.upstream.restclient +import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.upstream.ChainRequest import io.emeraldpay.dshackle.upstream.ChainResponse import io.emeraldpay.dshackle.upstream.HttpReader import io.emeraldpay.dshackle.upstream.RequestMetrics +import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry import io.emeraldpay.dshackle.upstream.rpcclient.ResponseRpcParser import io.emeraldpay.dshackle.upstream.rpcclient.RestParams import io.emeraldpay.dshackle.upstream.stream.AggregateResponse @@ -19,6 +21,7 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.scheduler.Scheduler import reactor.kotlin.core.publisher.switchIfEmpty +import reactor.netty.http.client.HttpClientResponse import java.util.concurrent.TimeUnit class RestHttpReader( @@ -27,12 +30,20 @@ class RestHttpReader( queueSize: Int, metrics: RequestMetrics, private val httpScheduler: Scheduler, + private val chain: Chain, basicAuth: AuthConfig.ClientBasicAuth? = null, tlsCAAuth: ByteArray? = null, ) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth) { private val parser = ResponseRpcParser() private val requestParser = RestRequestParser + private val headersToForward = ChainSpecificRegistry.resolve(chain).getResponseHeadersToForward() + + private fun extractResponseHeaders(header: HttpClientResponse): Map { + return headersToForward + .mapNotNull { name -> header.responseHeaders().get(name)?.let { name to it } } + .toMap() + } override fun internalRead(key: ChainRequest): Mono { val startTime = StopWatch() @@ -50,13 +61,13 @@ class RestHttpReader( } .handle { it, sink -> when (it) { - is StreamResponse -> sink.next(ChainResponse(it.stream, key.id)) + is StreamResponse -> sink.next(ChainResponse(it.stream, key.id, it.headers)) is AggregateResponse -> { if (it.code != 200) { val error = parser.readError(Global.objectMapper.createParser(it.response)) - sink.next(ChainResponse(null, error)) + sink.next(ChainResponse(null, error, it.headers)) } else { - sink.next(ChainResponse(it.response, null)) + sink.next(ChainResponse(it.response, null, it.headers)) } } else -> sink.error(IllegalStateException("Wrong response type")) @@ -87,18 +98,21 @@ class RestHttpReader( return if (!key.isStreamed) { response.response { header, bytes -> val statusCode = header.status().code() + val responseHeaders = extractResponseHeaders(header) bytes.aggregate().asByteArray().publishOn(httpScheduler).map { - AggregateResponse(it, statusCode) + AggregateResponse(it, statusCode, responseHeaders) }.switchIfEmpty { - Mono.just(AggregateResponse(ByteArray(0), statusCode)) + Mono.just(AggregateResponse(ByteArray(0), statusCode, responseHeaders)) } }.single() } else { response.responseConnection { t, u -> + val responseHeaders = extractResponseHeaders(t) + if (t.status().code() != 200) { u.inbound().receive().aggregate().asByteArray().publishOn(httpScheduler) - .map { AggregateResponse(it, t.status().code()) } + .map { AggregateResponse(it, t.status().code(), responseHeaders) } } else { Mono.just( StreamResponse( @@ -107,6 +121,7 @@ class RestHttpReader( .map { Chunk(it, false) }, Mono.just(Chunk(ByteArray(0), true)), ), + responseHeaders, ), ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/stream/Responses.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/stream/Responses.kt index 1b2e7a4f..621c3619 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/stream/Responses.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/stream/Responses.kt @@ -35,11 +35,13 @@ data class SingleResponse( data class StreamResponse( val stream: Flux, + val headers: Map = emptyMap(), ) : Response() data class AggregateResponse( val response: ByteArray, val code: Int, + val headers: Map = emptyMap(), ) : Response() { override fun equals(other: Any?): Boolean { if (this === other) return true