Compare commits
5 Commits
master
...
reload-rem
| Author | SHA1 | Date | |
|---|---|---|---|
| bcc71c6ca0 | |||
| 7f2cc0e739 | |||
| 89b331bdf3 | |||
| 0cbe8ba2e8 | |||
| a287e74bd7 |
1
.upstream-base
Normal file
1
.upstream-base
Normal file
@@ -0,0 +1 @@
|
|||||||
|
v0.79.11
|
||||||
150
FINDINGS.md
Normal file
150
FINDINGS.md
Normal file
@@ -0,0 +1,150 @@
|
|||||||
|
# dshackle v0.79.10 — SIGHUP reload removal investigation
|
||||||
|
|
||||||
|
**Base tag:** `v0.79.10` (`f2c1bd06`, "Update deps (#877)")
|
||||||
|
**Branch:** `reload-removal`
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## (a) Hot-reload codepath and the `Key UNSPECIFIED missing` failure
|
||||||
|
|
||||||
|
### SIGHUP entry
|
||||||
|
|
||||||
|
| Step | Location |
|
||||||
|
|------|----------|
|
||||||
|
| OS signal registration | `ReloadConfigSetup.kt:21-22` — `Signal.handle(signalHup, this)` |
|
||||||
|
| Handler / lock | `ReloadConfigSetup.kt:25-58` — catches exceptions, logs `Config is not reloaded, cause - ${e.message}` |
|
||||||
|
| Upstream processor | `UpstreamConfigReloadConfigProcessor` in `ReloadConfigProcessor.kt:20-46` |
|
||||||
|
|
||||||
|
### Diff logic
|
||||||
|
|
||||||
|
1. **Full-config equality short-circuit** — `ReloadConfigProcessor.kt:24-26` compares deserialized YAML to `mainConfig.initialConfig` (`ReloadConfigService.kt:27`, `MainConfig.kt:28-38`). Any change (including `methods.disabled`) marks the upstream entry as modified.
|
||||||
|
|
||||||
|
2. **Per-upstream diff** — `ReloadConfigProcessor.kt:53-79`:
|
||||||
|
- Missing in new config → `removed`
|
||||||
|
- Present but not equal (data class) → `reloaded` (treated as remove **and** re-add)
|
||||||
|
- New keys → `added`
|
||||||
|
|
||||||
|
3. **Default-options diff** — `ReloadConfigProcessor.kt:82-108`: changed/removed chain names in `defaultOptions` populate `chainsToReload`, which triggers **bulk removal of all upstreams on that chain** before re-adding (`ReloadConfigUpstreamService.kt:50-56` pre-patch).
|
||||||
|
|
||||||
|
4. **Removal execution (v0.79.10 bugs)** — `ReloadConfigUpstreamService.kt:44-68` (original):
|
||||||
|
- Calls `multistreamHolder.getUpstream(chain)` for every chain in `chainsToReload` and every `upstreamsToRemove` pair.
|
||||||
|
- `CurrentMultistreamHolder.getUpstream()` uses `chainMapping.getValue(chain)` (`CurrentMultistreamHolder.kt:33-34`).
|
||||||
|
- Multistreams are created only for `Chain.entries.filterNot { UNSPECIFIED }` (`MultistreamsConfig.kt:30-31`).
|
||||||
|
- **`Chain.UNSPECIFIED` is not in the map** → Kotlin throws `NoSuchElementException: Key UNSPECIFIED missing`.
|
||||||
|
|
||||||
|
**When `UNSPECIFIED` enters the reload set:**
|
||||||
|
- `Global.chainById(unknownChainName)` returns `UNSPECIFIED` (`Global.kt:56-62`).
|
||||||
|
- `analyzeDefaultOptions` / `analyzeUpstreams` used `chainById` without filtering (`ReloadConfigProcessor.kt:62-63, 100, 106` pre-patch).
|
||||||
|
- `buildDefaultOptions` in `ConfiguredUpstreams.kt:97-105` (pre-patch) could store options under `UNSPECIFIED` when a default-options chain name was unknown.
|
||||||
|
|
||||||
|
5. **Additional v0.79.10 removal gaps (fixed in patch):**
|
||||||
|
- **Async removal:** `Multistream.processUpstreamsEvents()` emits to a Reactor sink processed on a scheduler (`Multistream.kt:153-157, 441-444`). Reload called `addUpstreams()` immediately after firing REMOVED events, so add/remove could race.
|
||||||
|
- **gRPC upstream IDs:** Config `id: foo` registers per-chain upstreams as `foo_ethereum`, `foo_polygon`, etc. (`GenericGrpcUpstream.kt:69`). Removal looked up exact config id (`ReloadConfigUpstreamService.kt:61-64` pre-patch) → **no match, upstream never removed**.
|
||||||
|
- **gRPC client lifecycle:** `ConfiguredUpstreams` started `GrpcUpstreams.start()` with no registry/stop on config removal (`ConfiguredUpstreams.kt:77-88` pre-patch) — outbound describe loop kept running after YAML entry deleted.
|
||||||
|
|
||||||
|
### `methods.disabled` on reload
|
||||||
|
|
||||||
|
- Parsed into `ManagedCallMethods` at upstream creation (`UpstreamCreator.kt:97-112`).
|
||||||
|
- A `methods.disabled` change makes the upstream data class unequal → `reloaded` set → remove + re-add path (`ReloadConfigProcessor.kt:69-70`).
|
||||||
|
- **Does apply on reload** (not startup-only), provided the YAML change is detected.
|
||||||
|
- Aggregated method list on the multistream is recalculated in `MultistreamState.updateMethods()` (`MultistreamState.kt:81-97`) via `onUpstreamsUpdated()` after REMOVED/ADDED events.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## (b) How method availability is announced to gRPC clients
|
||||||
|
|
||||||
|
| Mechanism | RPC | Method list? | Live updates? |
|
||||||
|
|-----------|-----|--------------|---------------|
|
||||||
|
| **Describe** | `BlockchainRpc.describe` → `Describe.kt:33-79` | Yes — `DescribeChain.supportedMethods` from `multistream.getMethods().getSupportedMethods()` (`Describe.kt:42-46`) | **No** — unary snapshot at call time |
|
||||||
|
| **SubscribeChainStatus** | `BlockchainRpc.subscribeChainStatus` → `SubscribeChainStatus.kt:27-47` | Yes — `SupportedMethodsEvent` in `ChainEvent` (`SubscribeChainStatus.kt:71`, `ChainEventMapper.kt:124-125`) | **Yes** — `multistream.stateEvents()` pushes `MethodsEvent` when methods change (`MultistreamStateHandler.kt:11-12`, `MultistreamState.kt:74-76`) |
|
||||||
|
| **SubscribeStatus** | `SubscribeStatus.kt:34-48` | **No** — availability + quorum only | Yes for status |
|
||||||
|
| **NativeCall** | — | Enforces at call time via `upstream.getMethods().isAvailable(method)` (`NativeCall.kt:313-327`) | N/A |
|
||||||
|
|
||||||
|
**Implication for dRPC edge gateways**
|
||||||
|
|
||||||
|
- Edges on an active **`SubscribeChainStatus`** stream receive updated method lists without reconnect (comment at `SubscribeChainStatus.kt:28-29` explicitly mentions hot reload).
|
||||||
|
- Edges that only call **`Describe` at connect** will **not** see removals until reconnect.
|
||||||
|
- **`Describe` is not streamed.**
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## (c) NativeCall typed error for disabled / unknown methods
|
||||||
|
|
||||||
|
Proto field: `NativeCallReplyItem.item_error_code` (int), set in `NativeCall.buildResponse()` (`NativeCall.kt:193-196`).
|
||||||
|
|
||||||
|
| Condition | Behaviour (v0.79.10) |
|
||||||
|
|-----------|------------------------|
|
||||||
|
| Method not in aggregated list | `InvalidCallContext` with `ChainCallError(RpcResponseError.CODE_METHOD_NOT_EXIST, …)` (`NativeCall.kt:314-327`) |
|
||||||
|
| Method not callable on remote path | `RpcException(CODE_METHOD_NOT_EXIST, …)` (`NativeCall.kt:445-446`) |
|
||||||
|
| JSON-RPC standard code | `RpcResponseError.CODE_METHOD_NOT_EXIST = -32601` (`RpcResponseError.java:28`) — "method does not exist / is not available" |
|
||||||
|
|
||||||
|
**Bug (pre-patch):** `InvalidCallContext` stored the **request item id** in `CallError.id`, and `buildResponse` emitted that as `item_error_code` instead of `-32601`. Upstream failures via `ChainException` already used the RPC code as `CallError.id` (`NativeCall.kt:768`).
|
||||||
|
|
||||||
|
**Patch:** `CallError.itemErrorCode()` returns `upstreamError?.code ?: id` (`NativeCall.kt`).
|
||||||
|
|
||||||
|
Distinct codes exist: `-32601` (method missing) vs `-32603` (internal) vs `-32000..` (upstream) per `RpcResponseError.java`.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## (d) Graceful client disconnect mechanism (v0.79.10)
|
||||||
|
|
||||||
|
| Mechanism | Present? | Location |
|
||||||
|
|-----------|----------|----------|
|
||||||
|
| Server `@PreDestroy shutdown()` | Yes — full server stop only | `GrpcServer.kt:124-129` |
|
||||||
|
| `maxConnectionIdle(3600s)` | Yes — idle timeout, not reload-triggered | `GrpcServer.kt:81`, `Defaults.kt:33` |
|
||||||
|
| Per-connection GOAWAY / drain API | **No** | — |
|
||||||
|
| HTTP/2 GOAWAY on demand | **No** public API in grpc-java server used here | — |
|
||||||
|
|
||||||
|
**Conclusion:** v0.79.10 had **no** reload-triggered client refresh. Edges holding stale `Describe` data need either live `SubscribeChainStatus` or a forced reconnect.
|
||||||
|
|
||||||
|
### Patch approach (`reload-removal` branch)
|
||||||
|
|
||||||
|
1. **`GrpcClientRefreshService`** — server interceptor tracking active `ServerCall`s; `disconnectClients()` closes them with `Status.UNAVAILABLE` (`GrpcClientRefreshService.kt`).
|
||||||
|
2. **Config flag** — `reload.disconnect-clients-on-removal` (default `true`) in `ReloadConfig.kt` / `MainConfigReader.kt`.
|
||||||
|
3. **Triggered after** upstream/method removal in `UpstreamConfigReloadConfigProcessor.reload()` (`ReloadConfigProcessor.kt`).
|
||||||
|
|
||||||
|
This closes active RPCs (including long-lived `SubscribeChainStatus`), prompting client reconnect. It is not a raw Netty GOAWAY on the transport, but achieves the same operational outcome for grpc-java clients.
|
||||||
|
|
||||||
|
### Recommendation
|
||||||
|
|
||||||
|
| Edge behaviour | Approach |
|
||||||
|
|----------------|----------|
|
||||||
|
| Uses **SubscribeChainStatus** | Live `MethodsEvent` propagation is sufficient for method list; optional disconnect still helps for `Describe`-cached state |
|
||||||
|
| Uses **Describe-at-connect only** | **Requires reconnect** — use `disconnectClientsOnRemoval` (default on) or accept stale method lists until natural reconnect |
|
||||||
|
| Calls removed methods | **NativeCall `-32601`** — immediate typed failure even without reconnect |
|
||||||
|
|
||||||
|
**GOAWAY vs error-code:** Use **both**: error-code for in-flight calls to removed methods; disconnect (or SubscribeChainStatus stream) for updating the edge's cached capability set. Error-code alone leaves the edge believing the method still exists.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Patch summary (`reload-removal`)
|
||||||
|
|
||||||
|
1. Filter `Chain.UNSPECIFIED` from reload diffs and default-options processing.
|
||||||
|
2. Synchronous removal via `Multistream.processUpstreamsEventsSync()`.
|
||||||
|
3. Match gRPC upstreams by `{configId}_` prefix; `GrpcUpstreamsRegistry` stops outbound clients on removal.
|
||||||
|
4. Propagate removals through existing `SubscribeChainStatus` / `MultistreamState` path.
|
||||||
|
5. Optional client disconnect on removal (configurable, default on).
|
||||||
|
6. Fix `item_error_code` for disabled methods (`-32601`).
|
||||||
|
7. Tests extended in `ReloadConfigTest.kt`.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Build / test verification
|
||||||
|
|
||||||
|
```text
|
||||||
|
# Prerequisites (this workspace)
|
||||||
|
# - foundation published to mavenLocal (./foundation/gradlew publishToMavenLocal)
|
||||||
|
# - foundation/src/main/resources/public/chains.yaml (drpcorg/public submodule or raw fetch)
|
||||||
|
# - emerald-grpc/ proto submodule cloned
|
||||||
|
|
||||||
|
export JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64
|
||||||
|
./gradlew test --tests "io.emeraldpay.dshackle.config.reload.*"
|
||||||
|
# BUILD SUCCESSFUL — 8 tests (ReloadConfigTest + HealthReloadConfigProcessorTest)
|
||||||
|
|
||||||
|
./gradlew build
|
||||||
|
# compiles + ktlint pass; full suite: 1635 tests, 2 failed (pre-existing on v0.79.10):
|
||||||
|
# - GenericWsHeadSpec (WS disconnect liveness)
|
||||||
|
# - IntegrationTest (Spring context / environment)
|
||||||
|
```
|
||||||
|
|
||||||
|
Reload-specific tests all pass. Full `./gradlew build` reports two failures that also occur on unmodified `v0.79.10` in this environment.
|
||||||
61
build-image.sh
Executable file
61
build-image.sh
Executable file
@@ -0,0 +1,61 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# build-image.sh — build the StakeSquid dshackle fork image with a WIRE-CORRECT version.
|
||||||
|
#
|
||||||
|
# dRPC edges record each provider's dshackle version (Global.version, sent in every
|
||||||
|
# Describe response and SubscribeChainStatus BuildInfo) and deprioritize providers
|
||||||
|
# whose string doesn't match the expected release. gradle's getVersion() only yields
|
||||||
|
# the clean release string on an EXACT tag match; one patch commit past the tag would
|
||||||
|
# report "<next>-SNAPSHOT" to every edge. So: force-move the base release tag onto the
|
||||||
|
# patch HEAD locally before building (never push that tag), which makes our build
|
||||||
|
# report exactly what stock drpcorg/dshackle:<version> reports (verified against a
|
||||||
|
# live gateway: version.app=0.79.10).
|
||||||
|
#
|
||||||
|
# Usage: ./build-image.sh [base-tag] [suffix]
|
||||||
|
# base-tag: upstream release our branch is based on (default v0.79.10)
|
||||||
|
# suffix: local image tag suffix (default ss1) -> stakesquid/dshackle:<ver>-<suffix>
|
||||||
|
#
|
||||||
|
# Per upstream release: rebase reload-removal onto the new tag, bump the default here,
|
||||||
|
# rerun. Distribute with: docker save stakesquid/dshackle:<ver>-<suffix> | ssh <gw> docker load
|
||||||
|
set -euo pipefail
|
||||||
|
cd "$(dirname "$0")"
|
||||||
|
BASE_TAG=${1:-v0.79.10}
|
||||||
|
SUFFIX=${2:-ss1}
|
||||||
|
VER=${BASE_TAG#v}
|
||||||
|
|
||||||
|
git tag -f "$BASE_TAG" HEAD >/dev/null # local only - makes getVersion() exact-match
|
||||||
|
|
||||||
|
# Submodules are REQUIRED at build time (SSH urls -> rewrite to https; controller
|
||||||
|
# has no github key). The `public` submodule feeds foundation's resources
|
||||||
|
# (compatible-clients.yaml etc.) - foundation is consumed as a mavenLocal
|
||||||
|
# artifact, so it must be REPUBLISHED whenever the submodule moves, and the
|
||||||
|
# main build cleaned so jib can't reuse stale layers. Skipping any of this
|
||||||
|
# shipped a boot-crashing image on 2026-07-30 (ss1: Spring died on missing
|
||||||
|
# compatible-clients.yaml) - hence the hard gate below.
|
||||||
|
git config url."https://github.com/".insteadOf "git@github.com:" 2>/dev/null || true
|
||||||
|
git submodule sync -q && git submodule update --init --recursive
|
||||||
|
test -f foundation/src/main/resources/public/compatible-clients.yaml \
|
||||||
|
|| { echo "FATAL: public submodule missing compatible-clients.yaml"; exit 1; }
|
||||||
|
|
||||||
|
export JAVA_HOME=${JAVA_HOME:-/usr/lib/jvm/java-21-openjdk-amd64}
|
||||||
|
./gradlew -p foundation publishToMavenLocal -q
|
||||||
|
./gradlew clean -q
|
||||||
|
docker build -q -t drpc-dshackle . >/dev/null
|
||||||
|
./gradlew jibDockerBuild -Djib.to.image="stakesquid/dshackle:${VER}-${SUFFIX}" -x test
|
||||||
|
|
||||||
|
echo "--- resource gate (foundation jar inside the image):"
|
||||||
|
docker run --rm --entrypoint sh "stakesquid/dshackle:${VER}-${SUFFIX}" -c \
|
||||||
|
'ls /app/libs/foundation-*.jar >/dev/null' || { echo FATAL: no foundation jar; exit 1; }
|
||||||
|
python3 - "$VER" "$SUFFIX" <<'PY'
|
||||||
|
import subprocess, sys, zipfile, io
|
||||||
|
ver, suf = sys.argv[1], sys.argv[2]
|
||||||
|
jar = subprocess.run(["docker", "run", "--rm", "--entrypoint", "sh",
|
||||||
|
f"stakesquid/dshackle:{ver}-{suf}", "-c",
|
||||||
|
"cat /app/libs/foundation-*.jar"], capture_output=True).stdout
|
||||||
|
names = zipfile.ZipFile(io.BytesIO(jar)).namelist()
|
||||||
|
assert "public/compatible-clients.yaml" in names, "FATAL: compatible-clients.yaml not packaged"
|
||||||
|
print("resource gate OK: public/compatible-clients.yaml present in foundation jar")
|
||||||
|
PY
|
||||||
|
|
||||||
|
echo "--- wire version check (must equal stock ${VER}):"
|
||||||
|
docker run --rm --entrypoint cat "stakesquid/dshackle:${VER}-${SUFFIX}" \
|
||||||
|
/app/resources/version.properties
|
||||||
Submodule emerald-grpc updated: 5191731829...111ff264d4
@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle
|
|||||||
import io.emeraldpay.dshackle.auth.AuthInterceptor
|
import io.emeraldpay.dshackle.auth.AuthInterceptor
|
||||||
import io.emeraldpay.dshackle.config.MainConfig
|
import io.emeraldpay.dshackle.config.MainConfig
|
||||||
import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerGrpc
|
import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerGrpc
|
||||||
|
import io.emeraldpay.dshackle.rpc.GrpcClientRefreshService
|
||||||
import io.grpc.Codec
|
import io.grpc.Codec
|
||||||
import io.grpc.Server
|
import io.grpc.Server
|
||||||
import io.grpc.ServerCall
|
import io.grpc.ServerCall
|
||||||
@@ -46,6 +47,7 @@ open class GrpcServer(
|
|||||||
private val accessHandler: AccessHandlerGrpc,
|
private val accessHandler: AccessHandlerGrpc,
|
||||||
private val grpcServerBraveInterceptor: ServerInterceptor,
|
private val grpcServerBraveInterceptor: ServerInterceptor,
|
||||||
private val authInterceptor: AuthInterceptor,
|
private val authInterceptor: AuthInterceptor,
|
||||||
|
private val grpcClientRefreshService: GrpcClientRefreshService,
|
||||||
) {
|
) {
|
||||||
@Value("\${spring.application.max-metadata-size}")
|
@Value("\${spring.application.max-metadata-size}")
|
||||||
private var maxMetadataSize: Int = Defaults.maxMetadataSize
|
private var maxMetadataSize: Int = Defaults.maxMetadataSize
|
||||||
@@ -89,6 +91,7 @@ open class GrpcServer(
|
|||||||
}
|
}
|
||||||
|
|
||||||
serverBuilder.intercept(grpcServerBraveInterceptor)
|
serverBuilder.intercept(grpcServerBraveInterceptor)
|
||||||
|
serverBuilder.intercept(grpcClientRefreshService)
|
||||||
if (mainConfig.authorization.enabled && mainConfig.authorization.hasServerConfig()) {
|
if (mainConfig.authorization.enabled && mainConfig.authorization.hasServerConfig()) {
|
||||||
serverBuilder.intercept(authInterceptor)
|
serverBuilder.intercept(authInterceptor)
|
||||||
log.info("Token authorization is turned on")
|
log.info("Token authorization is turned on")
|
||||||
|
|||||||
@@ -44,6 +44,7 @@ class MainConfig {
|
|||||||
var compression: CompressionConfig = CompressionConfig.default()
|
var compression: CompressionConfig = CompressionConfig.default()
|
||||||
var chains: ChainsConfig = ChainsConfig.default()
|
var chains: ChainsConfig = ChainsConfig.default()
|
||||||
var authorization: AuthorizationConfig = AuthorizationConfig.default()
|
var authorization: AuthorizationConfig = AuthorizationConfig.default()
|
||||||
|
var reload: ReloadConfig = ReloadConfig()
|
||||||
|
|
||||||
var initialConfig: UpstreamsConfig? = null
|
var initialConfig: UpstreamsConfig? = null
|
||||||
private set
|
private set
|
||||||
|
|||||||
@@ -83,6 +83,18 @@ class MainConfigReader(
|
|||||||
authorizationConfigReader.read(input).let {
|
authorizationConfigReader.read(input).let {
|
||||||
config.authorization = it
|
config.authorization = it
|
||||||
}
|
}
|
||||||
|
readReloadConfig(input)?.let {
|
||||||
|
config.reload = it
|
||||||
|
}
|
||||||
return config
|
return config
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun readReloadConfig(input: MappingNode?): ReloadConfig? {
|
||||||
|
val reloadNode = getMapping(input, "reload") ?: return null
|
||||||
|
val reload = ReloadConfig()
|
||||||
|
getValueAsBool(reloadNode, "disconnect-clients-on-removal")?.let {
|
||||||
|
reload.disconnectClientsOnRemoval = it
|
||||||
|
}
|
||||||
|
return reload
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,9 @@
|
|||||||
|
package io.emeraldpay.dshackle.config
|
||||||
|
|
||||||
|
data class ReloadConfig(
|
||||||
|
/**
|
||||||
|
* When true (default), gracefully close active gRPC client RPCs after upstream or
|
||||||
|
* method removal so clients reconnect and re-fetch Describe / SubscribeChainStatus.
|
||||||
|
*/
|
||||||
|
var disconnectClientsOnRemoval: Boolean = true,
|
||||||
|
)
|
||||||
@@ -2,8 +2,11 @@ package io.emeraldpay.dshackle.config.reload
|
|||||||
|
|
||||||
import io.emeraldpay.dshackle.Chain
|
import io.emeraldpay.dshackle.Chain
|
||||||
import io.emeraldpay.dshackle.Global.Companion.chainById
|
import io.emeraldpay.dshackle.Global.Companion.chainById
|
||||||
|
import io.emeraldpay.dshackle.config.MainConfig
|
||||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||||
import io.emeraldpay.dshackle.foundation.ChainOptions
|
import io.emeraldpay.dshackle.foundation.ChainOptions
|
||||||
|
import io.emeraldpay.dshackle.rpc.GrpcClientRefreshService
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
import org.springframework.stereotype.Component
|
import org.springframework.stereotype.Component
|
||||||
import java.util.stream.Collectors
|
import java.util.stream.Collectors
|
||||||
|
|
||||||
@@ -16,7 +19,12 @@ interface ReloadConfigProcessor {
|
|||||||
class UpstreamConfigReloadConfigProcessor(
|
class UpstreamConfigReloadConfigProcessor(
|
||||||
private val reloadConfigService: ReloadConfigService,
|
private val reloadConfigService: ReloadConfigService,
|
||||||
private val reloadConfigUpstreamService: ReloadConfigUpstreamService,
|
private val reloadConfigUpstreamService: ReloadConfigUpstreamService,
|
||||||
|
private val mainConfig: MainConfig,
|
||||||
|
private val grpcClientRefreshService: GrpcClientRefreshService,
|
||||||
) : ReloadConfigProcessor {
|
) : ReloadConfigProcessor {
|
||||||
|
|
||||||
|
private val log = LoggerFactory.getLogger(UpstreamConfigReloadConfigProcessor::class.java)
|
||||||
|
|
||||||
override fun reload(): Boolean {
|
override fun reload(): Boolean {
|
||||||
val newUpstreamsConfig = reloadConfigService.readUpstreamsConfig()
|
val newUpstreamsConfig = reloadConfigService.readUpstreamsConfig()
|
||||||
val currentUpstreamsConfig = reloadConfigService.currentUpstreamsConfig()
|
val currentUpstreamsConfig = reloadConfigService.currentUpstreamsConfig()
|
||||||
@@ -35,13 +43,29 @@ class UpstreamConfigReloadConfigProcessor(
|
|||||||
)
|
)
|
||||||
|
|
||||||
val upstreamsToRemove = upstreamsAnalyzeData.removed
|
val upstreamsToRemove = upstreamsAnalyzeData.removed
|
||||||
|
.filter { it.second != Chain.UNSPECIFIED }
|
||||||
.filterNot { chainsToReload.contains(it.second) }
|
.filterNot { chainsToReload.contains(it.second) }
|
||||||
.toSet()
|
.toSet()
|
||||||
val upstreamsToAdd = upstreamsAnalyzeData.added
|
val upstreamsToAdd = upstreamsAnalyzeData.added
|
||||||
|
.filter { it.second != Chain.UNSPECIFIED }
|
||||||
|
.toSet()
|
||||||
|
|
||||||
reloadConfigService.updateUpstreamsConfig(newUpstreamsConfig)
|
reloadConfigService.updateUpstreamsConfig(newUpstreamsConfig)
|
||||||
|
|
||||||
reloadConfigUpstreamService.reloadUpstreams(chainsToReload, upstreamsToRemove, upstreamsToAdd, newUpstreamsConfig)
|
val reloadResult = reloadConfigUpstreamService.reloadUpstreams(
|
||||||
|
chainsToReload,
|
||||||
|
upstreamsToRemove,
|
||||||
|
upstreamsToAdd,
|
||||||
|
newUpstreamsConfig,
|
||||||
|
)
|
||||||
|
|
||||||
|
if (reloadResult.hadRemovals && mainConfig.reload.disconnectClientsOnRemoval) {
|
||||||
|
log.info(
|
||||||
|
"Disconnecting gRPC clients after upstream/method removal on chains {}",
|
||||||
|
reloadResult.affectedChains.map { it.chainCode },
|
||||||
|
)
|
||||||
|
grpcClientRefreshService.disconnectClients("upstream or method removed via config reload")
|
||||||
|
}
|
||||||
|
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
@@ -59,8 +83,8 @@ class UpstreamConfigReloadConfigProcessor(
|
|||||||
}
|
}
|
||||||
val reloaded = mutableSetOf<Pair<String, Chain>>()
|
val reloaded = mutableSetOf<Pair<String, Chain>>()
|
||||||
val removed = mutableSetOf<Pair<String, Chain>>()
|
val removed = mutableSetOf<Pair<String, Chain>>()
|
||||||
val currentUpstreamsMap = currentUpstreams.associateBy { it.id!! to chainById(it.chain) }
|
val currentUpstreamsMap = upstreamKeyMap(currentUpstreams)
|
||||||
val newUpstreamsMap = newUpstreams.associateBy { it.id!! to chainById(it.chain) }
|
val newUpstreamsMap = upstreamKeyMap(newUpstreams)
|
||||||
|
|
||||||
currentUpstreamsMap.forEach {
|
currentUpstreamsMap.forEach {
|
||||||
val newUpstream = newUpstreamsMap[it.key]
|
val newUpstream = newUpstreamsMap[it.key]
|
||||||
@@ -79,6 +103,19 @@ class UpstreamConfigReloadConfigProcessor(
|
|||||||
return UpstreamAnalyzeData(added, removed.plus(reloaded))
|
return UpstreamAnalyzeData(added, removed.plus(reloaded))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun upstreamKeyMap(
|
||||||
|
upstreams: List<UpstreamsConfig.Upstream<*>>,
|
||||||
|
): Map<Pair<String, Chain>, UpstreamsConfig.Upstream<*>> {
|
||||||
|
return upstreams.mapNotNull { up ->
|
||||||
|
val chain = chainById(up.chain)
|
||||||
|
if (chain == Chain.UNSPECIFIED) {
|
||||||
|
null
|
||||||
|
} else {
|
||||||
|
(up.id!! to chain) to up
|
||||||
|
}
|
||||||
|
}.toMap()
|
||||||
|
}
|
||||||
|
|
||||||
private fun analyzeDefaultOptions(
|
private fun analyzeDefaultOptions(
|
||||||
currentDefaultOptions: List<ChainOptions.DefaultOptions>,
|
currentDefaultOptions: List<ChainOptions.DefaultOptions>,
|
||||||
newDefaultOptions: List<ChainOptions.DefaultOptions>,
|
newDefaultOptions: List<ChainOptions.DefaultOptions>,
|
||||||
@@ -97,13 +134,21 @@ class UpstreamConfigReloadConfigProcessor(
|
|||||||
currentOptions.forEach {
|
currentOptions.forEach {
|
||||||
val newChainOption = newOptions[it.key]
|
val newChainOption = newOptions[it.key]
|
||||||
if (newChainOption == null) {
|
if (newChainOption == null) {
|
||||||
removed.add(chainById(it.key))
|
val chain = chainById(it.key)
|
||||||
|
if (chain != Chain.UNSPECIFIED) {
|
||||||
|
removed.add(chain)
|
||||||
|
}
|
||||||
} else if (newChainOption != it.value) {
|
} else if (newChainOption != it.value) {
|
||||||
chainsToReload.add(chainById(it.key))
|
val chain = chainById(it.key)
|
||||||
|
if (chain != Chain.UNSPECIFIED) {
|
||||||
|
chainsToReload.add(chain)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
val added = newOptions.minus(currentOptions.keys).map { chainById(it.key) }
|
val added = newOptions.minus(currentOptions.keys).mapNotNull {
|
||||||
|
chainById(it.key).takeIf { chain -> chain != Chain.UNSPECIFIED }
|
||||||
|
}
|
||||||
|
|
||||||
return chainsToReload.plus(added).plus(removed)
|
return chainsToReload.plus(added).plus(removed)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,21 +7,38 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
|
|||||||
import io.emeraldpay.dshackle.startup.ConfiguredUpstreams
|
import io.emeraldpay.dshackle.startup.ConfiguredUpstreams
|
||||||
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
|
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
|
||||||
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
|
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
|
||||||
|
import io.emeraldpay.dshackle.upstream.Upstream
|
||||||
|
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreamsRegistry
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
import org.springframework.stereotype.Component
|
import org.springframework.stereotype.Component
|
||||||
import java.util.stream.Collectors
|
import java.util.stream.Collectors
|
||||||
|
|
||||||
|
data class UpstreamReloadResult(
|
||||||
|
val hadRemovals: Boolean,
|
||||||
|
val affectedChains: Set<Chain>,
|
||||||
|
)
|
||||||
|
|
||||||
@Component
|
@Component
|
||||||
open class ReloadConfigUpstreamService(
|
open class ReloadConfigUpstreamService(
|
||||||
private val multistreamHolder: CurrentMultistreamHolder,
|
private val multistreamHolder: CurrentMultistreamHolder,
|
||||||
private val configuredUpstreams: ConfiguredUpstreams,
|
private val configuredUpstreams: ConfiguredUpstreams,
|
||||||
|
private val grpcUpstreamsRegistry: GrpcUpstreamsRegistry,
|
||||||
) {
|
) {
|
||||||
|
|
||||||
|
private val log = LoggerFactory.getLogger(ReloadConfigUpstreamService::class.java)
|
||||||
|
|
||||||
fun reloadUpstreams(
|
fun reloadUpstreams(
|
||||||
chainsToReload: Set<Chain>,
|
chainsToReload: Set<Chain>,
|
||||||
upstreamsToRemove: Set<Pair<String, Chain>>,
|
upstreamsToRemove: Set<Pair<String, Chain>>,
|
||||||
upstreamsToAdd: Set<Pair<String, Chain>>,
|
upstreamsToAdd: Set<Pair<String, Chain>>,
|
||||||
newUpstreamsConfig: UpstreamsConfig,
|
newUpstreamsConfig: UpstreamsConfig,
|
||||||
) {
|
): UpstreamReloadResult {
|
||||||
|
val safeChainsToReload = chainsToReload.filterValidChains()
|
||||||
|
val safeUpstreamsToRemove = upstreamsToRemove.filterValidPairs()
|
||||||
|
val grpcIdsToStop = safeUpstreamsToRemove.map { it.first }
|
||||||
|
.filter { grpcUpstreamsRegistry.isRegistered(it) }
|
||||||
|
.toSet()
|
||||||
|
|
||||||
val newUpstreamsCount = newUpstreamsConfig.upstreams.stream()
|
val newUpstreamsCount = newUpstreamsConfig.upstreams.stream()
|
||||||
.collect(
|
.collect(
|
||||||
Collectors.groupingBy(
|
Collectors.groupingBy(
|
||||||
@@ -30,38 +47,61 @@ open class ReloadConfigUpstreamService(
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
val usedChains = removeUpstreams(chainsToReload, upstreamsToRemove)
|
val usedChains = removeUpstreams(safeChainsToReload, safeUpstreamsToRemove, grpcIdsToStop)
|
||||||
|
|
||||||
addUpstreams(newUpstreamsConfig, chainsToReload, upstreamsToAdd.map { it.first }.toSet())
|
addUpstreams(newUpstreamsConfig, safeChainsToReload, upstreamsToAdd.map { it.first }.toSet())
|
||||||
|
|
||||||
usedChains.forEach { chain ->
|
usedChains.filterValidChains().forEach { chain ->
|
||||||
if (newUpstreamsCount[chain] == null) {
|
if (newUpstreamsCount[chain] == null || newUpstreamsCount[chain] == 0L) {
|
||||||
multistreamHolder.getUpstream(chain).stop()
|
multistreamHolder.getUpstream(chain).stop()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
val hadRemovals = safeUpstreamsToRemove.isNotEmpty() ||
|
||||||
|
safeChainsToReload.isNotEmpty() ||
|
||||||
|
grpcIdsToStop.isNotEmpty()
|
||||||
|
return UpstreamReloadResult(
|
||||||
|
hadRemovals = hadRemovals,
|
||||||
|
affectedChains = usedChains.filterValidChains().toSet(),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun removeUpstreams(
|
private fun removeUpstreams(
|
||||||
chainsToReload: Set<Chain>,
|
chainsToReload: Set<Chain>,
|
||||||
upstreamsToRemove: Set<Pair<String, Chain>>,
|
upstreamsToRemove: Set<Pair<String, Chain>>,
|
||||||
|
grpcIdsToStop: Set<String>,
|
||||||
): Set<Chain> {
|
): Set<Chain> {
|
||||||
val usedChains = mutableSetOf<Chain>()
|
val usedChains = mutableSetOf<Chain>()
|
||||||
|
|
||||||
chainsToReload.forEach {
|
grpcIdsToStop.forEach { id ->
|
||||||
usedChains.add(it)
|
grpcUpstreamsRegistry.stop(id).forEach { event ->
|
||||||
val ms = multistreamHolder.getUpstream(it)
|
usedChains.add(event.chain)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
chainsToReload.forEach { chain ->
|
||||||
|
usedChains.add(chain)
|
||||||
|
val ms = multistreamHolder.getUpstream(chain)
|
||||||
ms.getAll()
|
ms.getAll()
|
||||||
|
.toList()
|
||||||
.forEach { up ->
|
.forEach { up ->
|
||||||
ms.processUpstreamsEvents(UpstreamChangeEvent(it, up, UpstreamChangeEvent.ChangeType.REMOVED))
|
ms.processUpstreamsEventsSync(
|
||||||
|
UpstreamChangeEvent(chain, up, UpstreamChangeEvent.ChangeType.REMOVED),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
upstreamsToRemove.forEach { pair ->
|
upstreamsToRemove.forEach { pair ->
|
||||||
|
if (grpcUpstreamsRegistry.isRegistered(pair.first)) {
|
||||||
|
return@forEach
|
||||||
|
}
|
||||||
usedChains.add(pair.second)
|
usedChains.add(pair.second)
|
||||||
val ms = multistreamHolder.getUpstream(pair.second)
|
val ms = multistreamHolder.getUpstream(pair.second)
|
||||||
ms.getAll()
|
findUpstreamsToRemove(ms.getAll(), pair.first)
|
||||||
.find { pair.first == it.getId() }
|
.forEach { up ->
|
||||||
?.let { up ->
|
ms.processUpstreamsEventsSync(
|
||||||
ms.processUpstreamsEvents(UpstreamChangeEvent(pair.second, up, UpstreamChangeEvent.ChangeType.REMOVED))
|
UpstreamChangeEvent(pair.second, up, UpstreamChangeEvent.ChangeType.REMOVED),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -81,4 +121,22 @@ open class ReloadConfigUpstreamService(
|
|||||||
)
|
)
|
||||||
configuredUpstreams.processUpstreams(configToReload)
|
configuredUpstreams.processUpstreams(configToReload)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
companion object {
|
||||||
|
internal fun findUpstreamsToRemove(upstreams: List<Upstream>, configId: String): List<Upstream> {
|
||||||
|
val prefix = "${configId}_"
|
||||||
|
return upstreams.filter { up ->
|
||||||
|
up.getId() == configId || up.getId().startsWith(prefix)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun Set<Chain>.filterValidChains(): Set<Chain> =
|
||||||
|
filter { it != Chain.UNSPECIFIED }.toSet()
|
||||||
|
|
||||||
|
private fun Set<Pair<String, Chain>>.filterValidPairs(): Set<Pair<String, Chain>> =
|
||||||
|
filter { it.second != Chain.UNSPECIFIED }.toSet()
|
||||||
|
|
||||||
|
private fun Collection<Chain>.filterValidChains(): List<Chain> =
|
||||||
|
filter { it != Chain.UNSPECIFIED }
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,55 @@
|
|||||||
|
package io.emeraldpay.dshackle.rpc
|
||||||
|
|
||||||
|
import io.grpc.ForwardingServerCall
|
||||||
|
import io.grpc.Metadata
|
||||||
|
import io.grpc.ServerCall
|
||||||
|
import io.grpc.ServerCallHandler
|
||||||
|
import io.grpc.ServerInterceptor
|
||||||
|
import io.grpc.Status
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
|
import org.springframework.stereotype.Component
|
||||||
|
import java.util.concurrent.ConcurrentHashMap
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Tracks active inbound gRPC server calls and can close them on config reload so clients
|
||||||
|
* reconnect and pick up updated Describe / SubscribeChainStatus announcements.
|
||||||
|
*/
|
||||||
|
@Component
|
||||||
|
class GrpcClientRefreshService : ServerInterceptor {
|
||||||
|
|
||||||
|
private val log = LoggerFactory.getLogger(GrpcClientRefreshService::class.java)
|
||||||
|
|
||||||
|
private val activeCalls = ConcurrentHashMap.newKeySet<ServerCall<*, *>>()
|
||||||
|
|
||||||
|
override fun <ReqT : Any, RespT : Any> interceptCall(
|
||||||
|
call: ServerCall<ReqT, RespT>,
|
||||||
|
headers: Metadata,
|
||||||
|
next: ServerCallHandler<ReqT, RespT>,
|
||||||
|
): ServerCall.Listener<ReqT> {
|
||||||
|
activeCalls.add(call)
|
||||||
|
val trackedCall = object : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
|
||||||
|
override fun close(status: Status, trailers: Metadata) {
|
||||||
|
activeCalls.remove(call)
|
||||||
|
super.close(status, trailers)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return next.startCall(trackedCall, headers)
|
||||||
|
}
|
||||||
|
|
||||||
|
fun disconnectClients(reason: String = "config reload") {
|
||||||
|
val status = Status.UNAVAILABLE.withDescription(reason)
|
||||||
|
val snapshot = activeCalls.toList()
|
||||||
|
if (snapshot.isEmpty()) {
|
||||||
|
log.info("No active gRPC client calls to close for reload propagation")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
log.info("Closing {} active gRPC client call(s) after reload: {}", snapshot.size, reason)
|
||||||
|
snapshot.forEach { call ->
|
||||||
|
try {
|
||||||
|
call.close(status, Metadata())
|
||||||
|
} catch (e: Exception) {
|
||||||
|
log.debug("Failed to close gRPC client call during reload", e)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -193,7 +193,7 @@ open class NativeCall(
|
|||||||
if (it.isError()) {
|
if (it.isError()) {
|
||||||
it.error?.let { error ->
|
it.error?.let { error ->
|
||||||
result.setErrorMessage(error.message)
|
result.setErrorMessage(error.message)
|
||||||
.setItemErrorCode(error.id)
|
.setItemErrorCode(error.itemErrorCode())
|
||||||
|
|
||||||
error.data?.let { data ->
|
error.data?.let { data ->
|
||||||
result.setErrorData(data)
|
result.setErrorData(data)
|
||||||
@@ -750,6 +750,8 @@ open class NativeCall(
|
|||||||
val errorAsIs: ByteArray? = null,
|
val errorAsIs: ByteArray? = null,
|
||||||
) {
|
) {
|
||||||
|
|
||||||
|
fun itemErrorCode(): Int = upstreamError?.code ?: id
|
||||||
|
|
||||||
companion object {
|
companion object {
|
||||||
|
|
||||||
private val log = LoggerFactory.getLogger(CallError::class.java)
|
private val log = LoggerFactory.getLogger(CallError::class.java)
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.foundation.ChainOptions
|
|||||||
import io.emeraldpay.dshackle.startup.configure.UpstreamCreationData
|
import io.emeraldpay.dshackle.startup.configure.UpstreamCreationData
|
||||||
import io.emeraldpay.dshackle.startup.configure.UpstreamFactory
|
import io.emeraldpay.dshackle.startup.configure.UpstreamFactory
|
||||||
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
|
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
|
||||||
|
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreamsRegistry
|
||||||
import org.slf4j.LoggerFactory
|
import org.slf4j.LoggerFactory
|
||||||
import org.springframework.boot.ApplicationArguments
|
import org.springframework.boot.ApplicationArguments
|
||||||
import org.springframework.boot.ApplicationRunner
|
import org.springframework.boot.ApplicationRunner
|
||||||
@@ -35,6 +36,7 @@ open class ConfiguredUpstreams(
|
|||||||
private val config: UpstreamsConfig,
|
private val config: UpstreamsConfig,
|
||||||
private val multistreamHolder: CurrentMultistreamHolder,
|
private val multistreamHolder: CurrentMultistreamHolder,
|
||||||
private val chainsConfig: ChainsConfig,
|
private val chainsConfig: ChainsConfig,
|
||||||
|
private val grpcUpstreamsRegistry: GrpcUpstreamsRegistry,
|
||||||
) : ApplicationRunner {
|
) : ApplicationRunner {
|
||||||
private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
|
private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
|
||||||
|
|
||||||
@@ -74,11 +76,11 @@ open class ConfiguredUpstreams(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
upstreamFactory.createGrpcUpstream(
|
val grpcUpstreams = upstreamFactory.createGrpcUpstream(
|
||||||
up as UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
|
up as UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
|
||||||
chainsConfig,
|
chainsConfig,
|
||||||
)
|
)
|
||||||
.start()
|
val subscription = grpcUpstreams.start()
|
||||||
.doOnNext {
|
.doOnNext {
|
||||||
log.info("Chain ${it.chain} ${it.type} through gRPC at ${up.connection?.host}:${up.connection?.port}. With caps: ${it.upstream.getCapabilities()}")
|
log.info("Chain ${it.chain} ${it.type} through gRPC at ${up.connection?.host}:${up.connection?.port}. With caps: ${it.upstream.getCapabilities()}")
|
||||||
}
|
}
|
||||||
@@ -86,6 +88,7 @@ open class ConfiguredUpstreams(
|
|||||||
multistreamHolder.getUpstream(it.chain)
|
multistreamHolder.getUpstream(it.chain)
|
||||||
.processUpstreamsEvents(it)
|
.processUpstreamsEvents(it)
|
||||||
}
|
}
|
||||||
|
grpcUpstreamsRegistry.register(up.id!!, grpcUpstreams, subscription)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -95,6 +98,10 @@ open class ConfiguredUpstreams(
|
|||||||
config.defaultOptions.forEach { defaultsConfig ->
|
config.defaultOptions.forEach { defaultsConfig ->
|
||||||
defaultsConfig.chains?.forEach { chainName ->
|
defaultsConfig.chains?.forEach { chainName ->
|
||||||
Global.chainById(chainName).let { chain ->
|
Global.chainById(chainName).let { chain ->
|
||||||
|
if (chain == Chain.UNSPECIFIED) {
|
||||||
|
log.warn("Skipping unknown chain in default options: $chainName")
|
||||||
|
return@forEach
|
||||||
|
}
|
||||||
defaultsConfig.options?.let { options ->
|
defaultsConfig.options?.let { options ->
|
||||||
if (!defaultOptions.containsKey(chain)) {
|
if (!defaultOptions.containsKey(chain)) {
|
||||||
defaultOptions[chain] = options
|
defaultOptions[chain] = options
|
||||||
|
|||||||
@@ -444,6 +444,14 @@ abstract class Multistream(
|
|||||||
) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Applies an upstream change immediately on the calling thread. Used during config reload
|
||||||
|
* so removals complete before new upstreams are registered.
|
||||||
|
*/
|
||||||
|
fun processUpstreamsEventsSync(event: UpstreamChangeEvent) {
|
||||||
|
onUpstreamChange(event)
|
||||||
|
}
|
||||||
|
|
||||||
private fun onUpstreamChange(event: UpstreamChangeEvent) {
|
private fun onUpstreamChange(event: UpstreamChangeEvent) {
|
||||||
val chain = event.chain
|
val chain = event.chain
|
||||||
if (this.chain == chain) {
|
if (this.chain == chain) {
|
||||||
|
|||||||
@@ -43,6 +43,7 @@ import io.emeraldpay.dshackle.upstream.grpc.auth.GrpcUpstreamsAuth
|
|||||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
|
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
|
||||||
import io.grpc.ClientInterceptor
|
import io.grpc.ClientInterceptor
|
||||||
import io.grpc.Codec
|
import io.grpc.Codec
|
||||||
|
import io.grpc.ManagedChannel
|
||||||
import io.grpc.Status
|
import io.grpc.Status
|
||||||
import io.grpc.StatusRuntimeException
|
import io.grpc.StatusRuntimeException
|
||||||
import io.grpc.netty.NettyChannelBuilder
|
import io.grpc.netty.NettyChannelBuilder
|
||||||
@@ -64,6 +65,7 @@ import reactor.core.scheduler.Scheduler
|
|||||||
import java.io.IOException
|
import java.io.IOException
|
||||||
import java.time.Duration
|
import java.time.Duration
|
||||||
import java.util.concurrent.Executor
|
import java.util.concurrent.Executor
|
||||||
|
import java.util.concurrent.TimeUnit
|
||||||
import java.util.concurrent.locks.ReentrantLock
|
import java.util.concurrent.locks.ReentrantLock
|
||||||
import kotlin.concurrent.withLock
|
import kotlin.concurrent.withLock
|
||||||
|
|
||||||
@@ -94,8 +96,10 @@ class GrpcUpstreams(
|
|||||||
var timeout = Defaults.timeout
|
var timeout = Defaults.timeout
|
||||||
|
|
||||||
lateinit var client: ReactorBlockchainGrpc.ReactorBlockchainStub
|
lateinit var client: ReactorBlockchainGrpc.ReactorBlockchainStub
|
||||||
|
private var channel: ManagedChannel? = null
|
||||||
private val known = HashMap<Chain, DefaultUpstream>()
|
private val known = HashMap<Chain, DefaultUpstream>()
|
||||||
private val lock = ReentrantLock()
|
private val lock = ReentrantLock()
|
||||||
|
private val statusSubscriptions = mutableMapOf<Chain, Disposable>()
|
||||||
|
|
||||||
fun start(): Flux<UpstreamChangeEvent> {
|
fun start(): Flux<UpstreamChangeEvent> {
|
||||||
val chanelBuilder = NettyChannelBuilder.forAddress(host, port)
|
val chanelBuilder = NettyChannelBuilder.forAddress(host, port)
|
||||||
@@ -122,8 +126,9 @@ class GrpcUpstreams(
|
|||||||
chanelBuilder.usePlaintext()
|
chanelBuilder.usePlaintext()
|
||||||
}
|
}
|
||||||
|
|
||||||
val channel = chanelBuilder.build()
|
val builtChannel = chanelBuilder.build()
|
||||||
var client = ReactorBlockchainGrpc.newReactorStub(channel)
|
this.channel = builtChannel
|
||||||
|
var client = ReactorBlockchainGrpc.newReactorStub(builtChannel)
|
||||||
if (compression) {
|
if (compression) {
|
||||||
client = client.withCompression(Codec.Gzip().messageEncoding)
|
client = client.withCompression(Codec.Gzip().messageEncoding)
|
||||||
}
|
}
|
||||||
@@ -132,7 +137,7 @@ class GrpcUpstreams(
|
|||||||
val grpcUpstreamsAuth =
|
val grpcUpstreamsAuth =
|
||||||
if (tokenAuth != null && authorizationConfig.enabled) {
|
if (tokenAuth != null && authorizationConfig.enabled) {
|
||||||
GrpcUpstreamsAuth(
|
GrpcUpstreamsAuth(
|
||||||
ReactorAuthGrpc.newReactorStub(channel),
|
ReactorAuthGrpc.newReactorStub(builtChannel),
|
||||||
authorizationConfig,
|
authorizationConfig,
|
||||||
grpcAuthContext,
|
grpcAuthContext,
|
||||||
tokenAuth.publicKeyPath!!,
|
tokenAuth.publicKeyPath!!,
|
||||||
@@ -141,8 +146,6 @@ class GrpcUpstreams(
|
|||||||
null
|
null
|
||||||
}
|
}
|
||||||
|
|
||||||
val statusSubscriptions = mutableMapOf<Chain, Disposable>()
|
|
||||||
|
|
||||||
return Flux.interval(Duration.ZERO, Duration.ofSeconds(20))
|
return Flux.interval(Duration.ZERO, Duration.ofSeconds(20))
|
||||||
.flatMap {
|
.flatMap {
|
||||||
authAndDescribe(grpcUpstreamsAuth)
|
authAndDescribe(grpcUpstreamsAuth)
|
||||||
@@ -356,4 +359,31 @@ class GrpcUpstreams(
|
|||||||
}
|
}
|
||||||
|
|
||||||
private fun describe() = this.client.describe(DescribeRequest.newBuilder().build())
|
private fun describe() = this.client.describe(DescribeRequest.newBuilder().build())
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Stops the outbound gRPC client, disposes status subscriptions, and returns REMOVED events
|
||||||
|
* for all chains previously served through this connection.
|
||||||
|
*/
|
||||||
|
fun stopAndRemoveAll(): List<UpstreamChangeEvent> {
|
||||||
|
lock.withLock {
|
||||||
|
statusSubscriptions.values.forEach { it.dispose() }
|
||||||
|
statusSubscriptions.clear()
|
||||||
|
channel?.let { ch ->
|
||||||
|
ch.shutdown()
|
||||||
|
try {
|
||||||
|
ch.awaitTermination(5, TimeUnit.SECONDS)
|
||||||
|
} catch (_: InterruptedException) {
|
||||||
|
Thread.currentThread().interrupt()
|
||||||
|
ch.shutdownNow()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
channel = null
|
||||||
|
return known.map { (chain, upstream) ->
|
||||||
|
upstream.stop()
|
||||||
|
UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.REMOVED)
|
||||||
|
}.also {
|
||||||
|
known.clear()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,55 @@
|
|||||||
|
package io.emeraldpay.dshackle.upstream.grpc
|
||||||
|
|
||||||
|
import io.emeraldpay.dshackle.Chain
|
||||||
|
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
|
||||||
|
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
|
||||||
|
import org.slf4j.LoggerFactory
|
||||||
|
import org.springframework.stereotype.Component
|
||||||
|
import reactor.core.Disposable
|
||||||
|
import java.util.concurrent.ConcurrentHashMap
|
||||||
|
|
||||||
|
@Component
|
||||||
|
class GrpcUpstreamsRegistry(
|
||||||
|
private val multistreamHolder: CurrentMultistreamHolder,
|
||||||
|
) {
|
||||||
|
private val log = LoggerFactory.getLogger(GrpcUpstreamsRegistry::class.java)
|
||||||
|
|
||||||
|
private data class Entry(
|
||||||
|
val upstreams: GrpcUpstreams,
|
||||||
|
val subscription: Disposable,
|
||||||
|
)
|
||||||
|
|
||||||
|
private val active = ConcurrentHashMap<String, Entry>()
|
||||||
|
|
||||||
|
fun register(id: String, upstreams: GrpcUpstreams, subscription: Disposable) {
|
||||||
|
active[id]?.let { previous ->
|
||||||
|
log.warn("Replacing existing gRPC upstream registration for id=$id")
|
||||||
|
dispose(previous)
|
||||||
|
}
|
||||||
|
active[id] = Entry(upstreams, subscription)
|
||||||
|
}
|
||||||
|
|
||||||
|
fun stop(id: String): Collection<UpstreamChangeEvent> {
|
||||||
|
val entry = active.remove(id) ?: return emptyList()
|
||||||
|
dispose(entry)
|
||||||
|
return entry.upstreams.stopAndRemoveAll()
|
||||||
|
.also { events ->
|
||||||
|
events.forEach { event ->
|
||||||
|
if (event.chain != Chain.UNSPECIFIED) {
|
||||||
|
multistreamHolder.getUpstream(event.chain)
|
||||||
|
.processUpstreamsEventsSync(event)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fun stopAll(ids: Collection<String>): Boolean {
|
||||||
|
return ids.map { stop(it) }.any { it.isNotEmpty() } || ids.any { active.containsKey(it) }
|
||||||
|
}
|
||||||
|
|
||||||
|
fun isRegistered(id: String): Boolean = active.containsKey(id)
|
||||||
|
|
||||||
|
private fun dispose(entry: Entry) {
|
||||||
|
entry.subscription.dispose()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -10,6 +10,7 @@ import io.emeraldpay.dshackle.config.MainConfig
|
|||||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||||
import io.emeraldpay.dshackle.config.UpstreamsConfigReader
|
import io.emeraldpay.dshackle.config.UpstreamsConfigReader
|
||||||
import io.emeraldpay.dshackle.foundation.ChainOptionsReader
|
import io.emeraldpay.dshackle.foundation.ChainOptionsReader
|
||||||
|
import io.emeraldpay.dshackle.rpc.GrpcClientRefreshService
|
||||||
import io.emeraldpay.dshackle.startup.ConfiguredUpstreams
|
import io.emeraldpay.dshackle.startup.ConfiguredUpstreams
|
||||||
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
|
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
|
||||||
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
|
import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
|
||||||
@@ -18,6 +19,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
|
|||||||
import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry
|
import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry
|
||||||
import io.emeraldpay.dshackle.upstream.generic.GenericMultistream
|
import io.emeraldpay.dshackle.upstream.generic.GenericMultistream
|
||||||
import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
|
import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
|
||||||
|
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreamsRegistry
|
||||||
import org.junit.jupiter.api.Assertions.assertEquals
|
import org.junit.jupiter.api.Assertions.assertEquals
|
||||||
import org.junit.jupiter.api.Assertions.assertFalse
|
import org.junit.jupiter.api.Assertions.assertFalse
|
||||||
import org.junit.jupiter.api.Assertions.assertTrue
|
import org.junit.jupiter.api.Assertions.assertTrue
|
||||||
@@ -46,6 +48,8 @@ class ReloadConfigTest {
|
|||||||
private val config = mock<Config>()
|
private val config = mock<Config>()
|
||||||
private val reloadConfigService = ReloadConfigService(config, fileResolver, mainConfig)
|
private val reloadConfigService = ReloadConfigService(config, fileResolver, mainConfig)
|
||||||
private val configuredUpstreams = mock<ConfiguredUpstreams>()
|
private val configuredUpstreams = mock<ConfiguredUpstreams>()
|
||||||
|
private val grpcUpstreamsRegistry = mock<GrpcUpstreamsRegistry>()
|
||||||
|
private val grpcClientRefreshService = mock<GrpcClientRefreshService>()
|
||||||
|
|
||||||
@BeforeEach
|
@BeforeEach
|
||||||
fun setupTests() {
|
fun setupTests() {
|
||||||
@@ -88,8 +92,14 @@ class ReloadConfigTest {
|
|||||||
val reloadConfigUpstreamService = ReloadConfigUpstreamService(
|
val reloadConfigUpstreamService = ReloadConfigUpstreamService(
|
||||||
currentMultistreamHolder,
|
currentMultistreamHolder,
|
||||||
configuredUpstreams,
|
configuredUpstreams,
|
||||||
|
grpcUpstreamsRegistry,
|
||||||
|
)
|
||||||
|
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(
|
||||||
|
reloadConfigService,
|
||||||
|
reloadConfigUpstreamService,
|
||||||
|
mainConfig,
|
||||||
|
grpcClientRefreshService,
|
||||||
)
|
)
|
||||||
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(reloadConfigService, reloadConfigUpstreamService)
|
|
||||||
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
|
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
|
||||||
|
|
||||||
val initialConfigIs = ResourceUtils.getFile("classpath:configs/upstreams-initial.yaml").inputStream()
|
val initialConfigIs = ResourceUtils.getFile("classpath:configs/upstreams-initial.yaml").inputStream()
|
||||||
@@ -99,8 +109,8 @@ class ReloadConfigTest {
|
|||||||
|
|
||||||
reloadConfig.handle(Signal("HUP"))
|
reloadConfig.handle(Signal("HUP"))
|
||||||
|
|
||||||
verify(msEth).processUpstreamsEvents(UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED))
|
verify(msEth).processUpstreamsEventsSync(UpstreamChangeEvent(ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.REMOVED))
|
||||||
verify(msPoly).processUpstreamsEvents(UpstreamChangeEvent(POLYGON__MAINNET, up3, UpstreamChangeEvent.ChangeType.REMOVED))
|
verify(msPoly).processUpstreamsEventsSync(UpstreamChangeEvent(POLYGON__MAINNET, up3, UpstreamChangeEvent.ChangeType.REMOVED))
|
||||||
verify(configuredUpstreams).processUpstreams(
|
verify(configuredUpstreams).processUpstreams(
|
||||||
UpstreamsConfig(
|
UpstreamsConfig(
|
||||||
newConfig.defaultOptions,
|
newConfig.defaultOptions,
|
||||||
@@ -140,8 +150,14 @@ class ReloadConfigTest {
|
|||||||
val reloadConfigUpstreamService = ReloadConfigUpstreamService(
|
val reloadConfigUpstreamService = ReloadConfigUpstreamService(
|
||||||
currentMultistreamHolder,
|
currentMultistreamHolder,
|
||||||
configuredUpstreams,
|
configuredUpstreams,
|
||||||
|
grpcUpstreamsRegistry,
|
||||||
|
)
|
||||||
|
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(
|
||||||
|
reloadConfigService,
|
||||||
|
reloadConfigUpstreamService,
|
||||||
|
mainConfig,
|
||||||
|
grpcClientRefreshService,
|
||||||
)
|
)
|
||||||
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(reloadConfigService, reloadConfigUpstreamService)
|
|
||||||
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
|
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
|
||||||
val initialConfigIs = ResourceUtils.getFile("classpath:configs/upstreams-initial.yaml").inputStream()
|
val initialConfigIs = ResourceUtils.getFile("classpath:configs/upstreams-initial.yaml").inputStream()
|
||||||
val initialConfig = upstreamsConfigReader.read(initialConfigIs)!!
|
val initialConfig = upstreamsConfigReader.read(initialConfigIs)!!
|
||||||
@@ -172,7 +188,12 @@ class ReloadConfigTest {
|
|||||||
|
|
||||||
val reloadConfigUpstreamService = mock<ReloadConfigUpstreamService>()
|
val reloadConfigUpstreamService = mock<ReloadConfigUpstreamService>()
|
||||||
|
|
||||||
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(reloadConfigService, reloadConfigUpstreamService)
|
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(
|
||||||
|
reloadConfigService,
|
||||||
|
reloadConfigUpstreamService,
|
||||||
|
mainConfig,
|
||||||
|
grpcClientRefreshService,
|
||||||
|
)
|
||||||
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
|
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
|
||||||
|
|
||||||
whenever(config.getConfigPath()).thenReturn(initialConfigFile)
|
whenever(config.getConfigPath()).thenReturn(initialConfigFile)
|
||||||
@@ -184,6 +205,78 @@ class ReloadConfigTest {
|
|||||||
assertEquals(initialConfig, mainConfig.upstreams)
|
assertEquals(initialConfig, mainConfig.upstreams)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `reload upstream removal disconnects grpc clients by default`() {
|
||||||
|
val initialConfigFile = ResourceUtils.getFile("classpath:configs/upstreams-initial.yaml")
|
||||||
|
val newConfigFile = ResourceUtils.getFile("classpath:configs/upstreams-changed-upstreams-removed.yaml")
|
||||||
|
whenever(config.getConfigPath()).thenReturn(newConfigFile)
|
||||||
|
|
||||||
|
val reloadConfigUpstreamService = mock<ReloadConfigUpstreamService>()
|
||||||
|
whenever(
|
||||||
|
reloadConfigUpstreamService.reloadUpstreams(any(), any(), any(), any()),
|
||||||
|
).thenReturn(UpstreamReloadResult(hadRemovals = true, affectedChains = setOf(ETHEREUM__MAINNET)))
|
||||||
|
|
||||||
|
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(
|
||||||
|
reloadConfigService,
|
||||||
|
reloadConfigUpstreamService,
|
||||||
|
mainConfig,
|
||||||
|
grpcClientRefreshService,
|
||||||
|
)
|
||||||
|
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
|
||||||
|
|
||||||
|
val initialConfig = upstreamsConfigReader.read(initialConfigFile.inputStream())!!
|
||||||
|
mainConfig.upstreams = initialConfig
|
||||||
|
|
||||||
|
reloadConfig.handle(Signal("HUP"))
|
||||||
|
|
||||||
|
verify(grpcClientRefreshService).disconnectClients("upstream or method removed via config reload")
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `findUpstreamsToRemove matches grpc upstream ids`() {
|
||||||
|
val ethUpstream = upstream("provider_ethereum")
|
||||||
|
val polyUpstream = upstream("other_polygon")
|
||||||
|
val found = ReloadConfigUpstreamService.findUpstreamsToRemove(
|
||||||
|
listOf(ethUpstream, polyUpstream),
|
||||||
|
"provider",
|
||||||
|
)
|
||||||
|
assertEquals(listOf(ethUpstream), found)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `reload methods disabled change triggers upstream reload`() {
|
||||||
|
val initialConfigFile = ResourceUtils.getFile("classpath:configs/upstreams-methods-reload-initial.yaml")
|
||||||
|
val changedConfigFile = ResourceUtils.getFile("classpath:configs/upstreams-methods-reload-changed.yaml")
|
||||||
|
val initialConfig = upstreamsConfigReader.read(initialConfigFile.inputStream())!!
|
||||||
|
mainConfig.upstreams = initialConfig
|
||||||
|
|
||||||
|
val msEth = mock<Multistream> {
|
||||||
|
on { getAll() } doReturn emptyList()
|
||||||
|
}
|
||||||
|
val currentMultistreamHolder = mock<CurrentMultistreamHolder> {
|
||||||
|
on { getUpstream(ETHEREUM__MAINNET) } doReturn msEth
|
||||||
|
}
|
||||||
|
val reloadConfigUpstreamService = ReloadConfigUpstreamService(
|
||||||
|
currentMultistreamHolder,
|
||||||
|
configuredUpstreams,
|
||||||
|
grpcUpstreamsRegistry,
|
||||||
|
)
|
||||||
|
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(
|
||||||
|
reloadConfigService,
|
||||||
|
reloadConfigUpstreamService,
|
||||||
|
mainConfig,
|
||||||
|
grpcClientRefreshService,
|
||||||
|
)
|
||||||
|
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
|
||||||
|
|
||||||
|
whenever(config.getConfigPath()).thenReturn(changedConfigFile)
|
||||||
|
|
||||||
|
reloadConfig.handle(Signal("HUP"))
|
||||||
|
|
||||||
|
verify(configuredUpstreams).processUpstreams(any())
|
||||||
|
verify(grpcClientRefreshService).disconnectClients("upstream or method removed via config reload")
|
||||||
|
}
|
||||||
|
|
||||||
private fun multistream(chain: Chain): Multistream {
|
private fun multistream(chain: Chain): Multistream {
|
||||||
val cs = ChainSpecificRegistry.resolve(chain)
|
val cs = ChainSpecificRegistry.resolve(chain)
|
||||||
return GenericMultistream(
|
return GenericMultistream(
|
||||||
|
|||||||
@@ -0,0 +1,12 @@
|
|||||||
|
cluster:
|
||||||
|
upstreams:
|
||||||
|
- id: local1
|
||||||
|
chain: ethereum
|
||||||
|
methods:
|
||||||
|
disabled:
|
||||||
|
- name: eth_call
|
||||||
|
connection:
|
||||||
|
ethereum-pos:
|
||||||
|
execution:
|
||||||
|
rpc:
|
||||||
|
url: "http://localhost"
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
cluster:
|
||||||
|
upstreams:
|
||||||
|
- id: local1
|
||||||
|
chain: ethereum
|
||||||
|
connection:
|
||||||
|
ethereum-pos:
|
||||||
|
execution:
|
||||||
|
rpc:
|
||||||
|
url: "http://localhost"
|
||||||
Reference in New Issue
Block a user