StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Forward HTTP response headers from upstream to gRPC client for Beacon Chain (#759)

Browse Source 
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.
This commit is contained in:
Anton
2025-12-18 11:30:18 +03:00
committed by GitHub
parent 77ae38e138
commit 757febc0b5
14 changed files with 98 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt
View File

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

1
src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt
View File

@@ -64,6 +64,7 @@ class BroadcastReader(
upstreams.size,
upsData,
null,
quorum.getResponse()!!.responseHeaders,
)
Mono.just(res)
} else {

3
src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt
View File

@@ -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<Upstream.UpstreamSettingsData>,
val stream: Flux<Chunk>?,
val responseHeaders: Map<String, String> = emptyMap(),
)
}

44
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -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<String, String> = 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<ParsedCallDetails>?,
val stream: Flux<Chunk>? = null,
val finalization: FinalizationData? = null,
val responseHeaders: Map<String, String> = 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<Upstream.UpstreamSettingsData>, ctx: ValidCallContext<ParsedCallDetails>?): CallResult {
return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx)
fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: List<Upstream.UpstreamSettingsData>, ctx: ValidCallContext<ParsedCallDetails>?, responseHeaders: Map<String, String> = 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<Upstream.UpstreamSettingsData>, ctx: ValidCallContext<ParsedCallDetails>?, 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<Upstream.UpstreamSettingsData>, ctx: ValidCallContext<ParsedCallDetails>?, final: FinalizationData?, responseHeaders: Map<String, String> = 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<Upstream.UpstreamSettingsData>, ctx: ValidCallContext<ParsedCallDetails>?, stream: Flux<Chunk>?): CallResult {
return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, stream)
fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamSettingsData: List<Upstream.UpstreamSettingsData>, ctx: ValidCallContext<ParsedCallDetails>?, stream: Flux<Chunk>?, responseHeaders: Map<String, String> = emptyMap()): CallResult {
return CallResult(id, nonce, result, null, signature, upstreamSettingsData, ctx, stream, null, responseHeaders)
}
fun fail(id: Int, nonce: Long?, error: CallError, ctx: ValidCallContext<ParsedCallDetails>?): CallResult {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt
View File

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

16
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainResponse.kt
View File

@@ -35,17 +35,25 @@ class ChainResponse @JvmOverloads constructor(
val providedSignature: ResponseSigner.Signature? = null,
val resolvedUpstreamData: List<Upstream.UpstreamSettingsData> = emptyList(),
val finalization: FinalizationData? = null,
val responseHeaders: Map<String, String> = emptyMap(),
) {
constructor(stream: Flux<Chunk>, 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<Chunk>, id: Int, responseHeaders: Map<String, String>) :
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<String, String>) :
this(result, error, NumberId(0), null, null, emptyList(), null, responseHeaders)
constructor(result: ByteArray?, error: ChainCallError?, resolvedUpstreamData: List<Upstream.UpstreamSettingsData>) :
this(result, error, NumberId(0), null, null, resolvedUpstreamData, null)
this(result, error, NumberId(0), null, null, resolvedUpstreamData, null, emptyMap())
constructor(result: ByteArray?, resolvedUpstreamData: List<Upstream.UpstreamSettingsData>, 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 {

9
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt
View File

@@ -121,6 +121,15 @@ object BeaconChainSpecific : AbstractPollChainSpecific() {
override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService {
return BeaconChainLowerBoundService(chain, upstream)
}
override fun getResponseHeadersToForward(): List<String> = 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(

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBeaconChainMethods.kt
View File

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

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt
View File

@@ -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<String> = emptyList()
}
object ChainSpecificRegistry {

27
src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt
View File

@@ -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<String, String> {
return headersToForward
.mapNotNull { name -> header.responseHeaders().get(name)?.let { name to it } }
.toMap()
}
override fun internalRead(key: ChainRequest): Mono<ChainResponse> {
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,
),
)
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/stream/Responses.kt
View File

@@ -35,11 +35,13 @@ data class SingleResponse(
data class StreamResponse(
val stream: Flux<Chunk>,
val headers: Map<String, String> = emptyMap(),
) : Response()
data class AggregateResponse(
val response: ByteArray,
val code: Int,
val headers: Map<String, String> = emptyMap(),
) : Response() {
override fun equals(other: Any?): Boolean {
if (this === other) return true