Add Aztec upstream settings detection and improve node validation (#822)
* feat(aztec): add missing v4 RPC methods for full Aztec network support Adds methods identified by the Aztec team audit (v4 branch) that were returning -32601 method not available on dRPC: - node_getBlockByArchive - node_getBlockByHash - node_getBlockHeaderByArchive - node_getCheckpointNumber - node_getMaxPriorityFees - node_getTxsByHash Also adds the rest of the v4 AztecNodeApiSchema surface that was not yet in the allowlist so the upstream chain has full coverage: - node_getBlockHashMembershipWitness - node_getCheckpointedBlocks - node_getCheckpointedBlockNumber - node_getCheckpoints - node_getCurrentMinFees - node_getL1ToL2MessageCheckpoint - node_getPrivateLogsByTags - node_getPublicLogsByTagsFromContract * feat(aztec): add upstream settings detector Modeled on Starknet/Near settings detectors. Calls node_getNodeVersion to set client_type=aztec and client_version=<version> labels on each Aztec upstream, so monitoring/routing can distinguish nodes by version. Closes the parity gap with StarknetUpstreamSettingsDetector that the Aztec implementation was missing. * feat(aztec): wire settings detector and add L2 tips health validation Closes the parity gap with StarknetChainSpecific: - Override upstreamSettingsDetector() to enable client_type/version labels via the new AztecUpstreamSettingsDetector. - Add a second health validator that probes node_getL2Tips and rejects upstreams that return empty tips, inconsistent (proven > proposed) or show excessive prover lag. Until now only node_isReady was checked, which an Aztec node returns true for even when its tips are stale or empty - exactly the failure mode that surfaces to clients as "block height goes backwards" and "method returns null intermittently" during routing across multiple Aztec upstreams. Removes the unused parseInstant() helper. * feat(aztec): finish chain-specific stubs Three pieces in AztecChainSpecific were stubbed out and now have a real implementation: - upstreamSettingsValidators returned emptyList(). It now validates the upstream's reported node_getChainId against the configured chain id (decimal or 0x-hex compared numerically) and emits UPSTREAM_FATAL_SETTINGS_ERROR on mismatch, mirroring the pattern Polkadot uses with system_chain. This ensures Aztec mainnet upstreams cannot be silently mixed with testnet/devnet ones. - getFromHeader threw NotImplementedError. It is reachable only from GenericWsHead (websocket new-head delivery), and Aztec is HTTP-poll only - so it was never called. Route it through parseBlock so the same parser handles header/tip-shaped payloads if a future WS-capable backend is wired in. Behaviour today is unchanged. - listenNewHeadsRequest / unsubscribeNewHeadsRequest still cannot be satisfied (Aztec has no websocket newHeads subscription) but now throw UnsupportedOperationException with a descriptive message instead of NotImplementedError, matching the message used in nodecore. * fix(aztec): address Copilot review on validateTips - Use a single Long-typed `threshold` variable for both the comparison and the warning log. Previously the check used `lagging.toLong() * 10` but the log printed `lagging * 10` (Int multiplication), which could overflow and disagree with the check. - Wrap `Global.objectMapper.readTree` in try/catch and explicitly check for empty/whitespace input, returning SYNCING with a warning instead of letting parse errors fall through to UNAVAILABLE. The previous `raw.isMissingNode` branch was dead code (root is never a MissingNode). - Treat a missing/unparseable `proven.number` as SYNCING with its own warning. Previously it defaulted to 0, conflating a legitimate zero with missing data and risking a false "excessive prover lag" verdict on early-genesis upstreams. * test(aztec): add AztecChainSpecificTest covering validateTips and parseBlock Per Copilot review on PR #822: locks in the validateTips classification behaviour for the cases that matter for routing health - - healthy tips (OK) - lagging=0 disabling the gap check (OK) - empty / whitespace / unparseable / JSON-null payloads (SYNCING) - missing or zero proposed (SYNCING) - missing proven (SYNCING) - proven ahead of proposed (SYNCING) - excessive proposed-proven gap relative to laggingLagSize (SYNCING) Plus a parseBlock test that asserts proposed.{number,hash} is what the head tracker sees. * test(aztec): add AztecUpstreamSettingsDetectorTest Per Copilot review on PR #822: locks in version-string parsing for the representative `node_getNodeVersion` payloads - - quoted JSON string with leading "v" - quoted JSON string without leading "v" - raw unquoted string (some clients) - empty string -> UNKNOWN_CLIENT_VERSION - upstream call error -> UNKNOWN_CLIENT_VERSION - object payload with `nodeVersion` field (label detection path) Also asserts the (client_type, client_version) label pairs are produced in the right order via detectLabels(). * feat(aztec): replace stub lower-bound detector with real probing Previously AztecLowerBoundStateDetector returned a hardcoded LowerBoundData(1, STATE) regardless of what the upstream actually had. That meant routing assumed every Aztec upstream was a full archive, which is exactly what the customer's "method returns null intermittently for an old block" complaint looked like - a non-archive upstream got asked for state it never kept and answered `null`, but the router did not know to prefer an archive peer. Switch to the same RecursiveLowerBound machinery Polkadot and Beacon use: binary-search [0, head] by probing node_getBlock(N), treat a JSON `null` result (or "block not found"/"pruned"-style error) as the upstream not having that block, and converge on the lowest available block. The result is then used by the existing routing layer to send historical-state requests only to upstreams that can actually serve them. * feat(aztec): pass upstream to lower-bound state detector Required for the now-real AztecLowerBoundStateDetector, which probes node_getBlock(N) on the upstream itself. Mirrors PolkadotLowerBoundService. * fix(aztec): harden validateChainId parse path; simplify whitespace check - validateChainId now defensively handles empty/whitespace bodies and unparseable JSON the same way validateTips does: a concise warn log and UPSTREAM_SETTINGS_ERROR (Copilot review L187). Without this, a bad chain-id payload bubbled up through GenericSingleCallValidator as a generic error with no Aztec-specific log line. - chainIdMatches becomes `fun` instead of `private fun` so it can be unit-tested directly. - validateTips whitespace check rewritten as `String(data).isBlank()` for readability; behaviour unchanged. * fix(aztec): tolerate whitespace/casing in lower-bound null check Per Copilot review on PR #822 L61: the detector compared the result byte-for-byte to "null", so a payload like "null\n" or "NULL" would have been treated as a real block and made the binary search converge on a non-existent lower bound. Switch to a trimmed case-insensitive String comparison wrapped in `isNullResult()` so the helper can be covered separately by tests. * fix(aztec): make parseClientVersion handle object payloads Per Copilot review on PR #822 L35: parseClientVersion only stripped quotes/`v` from raw bytes, so an object payload like {"nodeVersion": "v0.84.0"} would have been returned literally as the "version" string by detectClientVersion(), while detectLabels() (which goes through clientVersion(JsonNode)) would have correctly returned "0.84.0". Try JSON parsing first and delegate to clientVersion(JsonNode) so both paths agree; the literal string-strip stays as a fallback for raw non-JSON responses. * test(aztec): add validateChainId/chainIdMatches coverage Per Copilot review on PR #822 L115: the new node_getChainId settings validator had no unit tests. Adds: - chainIdMatches: decimal vs hex equivalence (1 vs 0x1, 0 vs 0x0, Sepolia 11155111 vs 0xaa36a7), explicit mismatches, case normalisation, leading-zero stripping - validateChainId: VALID on numeric/string match, FATAL on numeric mismatch, SETTINGS_ERROR for empty/whitespace/unparseable/JSON-null payloads and unexpected object payloads. * fix(aztec): parse v4 nested L2Tips schema (proven.block.number) Live testnet log on aztec-testnet 4.2.0-rc.1 showed every health probe warning "returned tips without a proven number" and the upstream stuck in SYNCING. The reason: node_getL2Tips changed shape between Aztec versions: v3: {proposed: {number, hash}, proven: {number, hash}, checkpointed: {number, hash}} v4: proven/finalized/checkpointed each became {block: {number, hash}, checkpoint: {number, hash}} proposed stayed flat. parseBlock and validateTips now look at v4 nested paths first (proposed.{number,hash}, proven.block.number) and fall back to the v3 flat paths so older upstreams keep working too. Removes the noisy full-payload dump from the warn log. * test(aztec): cover v4 nested L2Tips schema in chain-specific tests Adds two cases for the actual aztec-testnet 4.2.0-rc.1 payload shape: - parseBlock against the v4 nested {proposed, proven.block, finalized, checkpointed.block} response, asserting proposed.{number,hash} are picked up. - validateTips returns OK on the same payload (gap 66934-66908=26 ≤ threshold 50 with lagging=5). Existing v3-flat fixtures remain to lock the fallback path. * perf(aztec): probe lower bound via node_getBlockHeader instead of getBlock The recursive lower-bound detector does ~log2(currentHeight) probes per refresh cycle. node_getBlock returns the full block with transactions (often KBs), but for the "is this block present?" check we only need the header. node_getBlockHeader returns the same `null`-on-missing signal in a much smaller payload. * fix(aztec): seed STATE=1 fallback so UNKNOWN bound never appears On the very first detection tick the upstream head is not yet known, so RecursiveLowerBound.initialRange() returns Mono.empty() and the recursive search produces an empty Flux. The base LowerBoundDetector then substitutes LowerBoundData.default() = (0, UNKNOWN), which gets stored alongside the real STATE bound discovered five minutes later. The result was the noisy "lower bounds=[STATE=1, UNKNOWN=0]" line in the multistream state log. Aztec full nodes are archive by default, so substitute STATE=1 as the detector's own empty-fallback. Subsequent ticks still re-run the binary search and refine the value if the upstream prunes state. * feat(aztec): retry transient HTTP 5xx in health validators Aztec public RPC endpoints (testnet/mainnet) occasionally answer 502/503/504 with HTML during deploys or sequencer failovers. The default GenericSingleCallValidator would turn the very first such hit into UNAVAILABLE, ejecting the upstream from the multistream until the next probe succeeded - causing visible flapping in the state log even though the upstream itself was healthy. AztecRetryingValidator wraps the read with reactor's Retry.backoff filtered on ChainException messages "HTTP Code: 502/503/504"; up to two retries with 500ms backoff are attempted before falling through to the existing onError (Unavailable) branch. Non-transient errors (timeouts, JSON-RPC errors, 4xx) bypass the retry and propagate immediately. * feat(aztec): use AztecRetryingValidator for health probes The two health-probe validators (node_isReady, node_getL2Tips) now go through AztecRetryingValidator, which retries transient HTTP 5xx errors a couple of times before falling through to UNAVAILABLE. This absorbs the occasional 502/503/504 the public Aztec endpoint emits during deploys/failovers and stops the upstream from flapping in/out of the multistream (visible in the live testnet log as a single status=[UNAVAILABLE/1] tick every few minutes). Settings validation (node_getChainId) is left on the default GenericSingleCallValidator since it runs once at startup; if it hits a 5xx the upstream stays unvalidated until the next health tick takes over. * fix(aztec): simplify retry predicate to avoid Kotlin SAM edge cases Removes the callable reference to a companion-object predicate (`::isTransientHttpError`) and the `onRetryExhaustedThrow` BiFunction lambda - both have historically tripped over Kotlin/Reactor SAM resolution. Inlines the predicate into a plain lambda bound to a private instance method, and lets Retry's default exhaustion behaviour propagate the underlying cause as-is. Renames the constructor parameter `backoff` to `retryBackoff` to disambiguate from the static `Retry.backoff` factory method at the call site (the original code worked but the duplication was avoidable). Also widens the slf4j log when message is null. * fix(aztec): log full throwable in retrying validator; drop const Long literal - Per Copilot review L83: log.error now passes the throwable as a positional slf4j argument so the full stack trace survives, matching GenericSingleCallValidator's diagnosability. - Replaces `const val DEFAULT_MAX_RETRIES: Long = 2` with a plain @JvmField `val ... = 2L`. The literal-2 form has bitten Kotlin compilation in the past depending on toolchain version; explicit Long literal removes the ambiguity. * fix(aztec): getFromHeader fails fast for polling-only chain Per Copilot review L65: parseBlock parses the node_getL2Tips response shape, not a websocket newHeads event. Delegating getFromHeader to parseBlock would silently produce height=0 BlockContainers if a WS connector were ever (mis)wired for Aztec. Reverting getFromHeader to throw UnsupportedOperationException with the same message used for the listen/unsubscribe newHeads stubs - same approach Starknet/Near/AVM take for polling-only chains. * fix(aztec): return UNKNOWN on JSON-null / unparseable version payloads Per Copilot review L52: when the JSON payload parsed successfully but clientVersion(node) couldn't extract a usable version, parseClientVersion fell through to the literal trim/quote-strip branch and could end up returning the whole JSON object as the "version" string, polluting client_version labels and version-rule logic. Trust clientVersion(JsonNode) when JSON parses, treat literal "null" (any case) and blank strings as UNKNOWN_CLIENT_VERSION in both the JSON and string-fallback paths. Also tightened the JsonNode branch to emit UNKNOWN for "null" string values. * revert(aztec): drop AztecRetryingValidator Per user direction. Restoring GenericSingleCallValidator for the Aztec health probes in the next commit. The retry-on-502 path will be revisited later if needed. * revert(aztec): use GenericSingleCallValidator for health probes Restores the original validator for node_isReady and node_getL2Tips. The retry-on-5xx behaviour added in earlier commits (AztecRetryingValidator) will be revisited later if needed - keeping the chain-specific in line with the rest of the polling chains for now. Schema, settings-detector, chain-id and L2-tips-content validations introduced earlier in this PR all remain in place. * refactor(aztec): replace RecursiveLowerBound binary search with single getWorldStateSyncStatus call node_getWorldStateSyncStatus reports oldestHistoricBlockNumber directly (see yarn-project/stdlib/src/interfaces/world_state.ts in AztecProtocol/aztec-packages). One RPC per refresh instead of ~log2(currentHeight) probes, and the value comes from the world-state synchronizer itself rather than being inferred from JSON-null responses to node_getBlockHeader. Falls back to STATE=1 (Aztec archive default) when the call fails - the public Aztec endpoint occasionally returns transient errors on this method per the Aztec team's audit; the next refresh tick will pick up the real value. * fix(aztec): on error keep cached STATE bound instead of clobbering it Previous fallback emitted STATE=1 on error. Because the base LowerBoundDetector filter accepts any LowerBoundData with lowerBound==1 unconditionally, a transient world-state-sync-status failure on a pruning Aztec upstream would have rewritten its real prune boundary (e.g. STATE=10000) down to STATE=1 - the router would then send historical-state requests there and get nulls. On error now: - if a STATE bound is already cached, re-emit the cached value (filter passes, updateBound is a no-op, cache stays); - if no STATE has been read yet, emit nothing - the router sees no STATE bound for this upstream until the next successful refresh. Same semantics applied to the "field missing/non-numeric" branch of parseOldestHistoric, which also previously hard-defaulted to 1. * fix(aztec): on transient error re-emit cached LowerBoundData (same timestamp) The previous on-error branch emitted a freshly-constructed `LowerBoundData(cached.lowerBound, STATE)`, which carries a new `Instant.now()` timestamp. That feeds LowerBounds.updateBound a "new sample, same value" point and biases the linear-regression coefficient `k` toward zero - which then makes predictLowerBound() under-predict how fast the sliding window is moving forward. Re-emit the cached LowerBoundData object as-is. updateBound's `newBound.timestamp != lastBound.timestamp` guard short-circuits, so the regression is left untouched. Cache stays at the last good value until the next successful refresh. Also stops the malformed-payload branch of parseOldestHistoric from hard-defaulting to STATE=1 (same clobbering risk via the special `lowerBound == 1L` filter); it now prefers the cache, only synthesises STATE=1 on a first-tick malformed response with no cache to fall back on. * E2E test success * revert: drop unrelated Makefile/gitignore changes Per review comment "Too hard" on Makefile L6: the `clean && ./gradlew run` addition forced a full rebuild on every `make run-main`, which is overkill for normal dev use. The original `./gradlew run -x test` is restored. The matching `.gitignore` line (`*_test.sh`, added in the same series of commits) is also reverted in the next commit - both changes are unrelated to the Aztec PR scope. * revert: drop unrelated *_test.sh ignore rule Out-of-scope of the Aztec PR; could shadow legitimate `*_test.sh` files anywhere in the tree. The smoke-test script lived only in local sandboxes. * fix(aztec): never synthesize STATE=1 on missing oldestHistoricBlockNumber Per review comment on AztecLowerBoundStateDetector L106 ("One means all history available"): the malformed-payload-and-no-cache branch was returning `LowerBoundData(1, STATE)`, which advertises the upstream as a full archive node. That is a lie if the upstream simply didn't tell us where its prune boundary is, and the base LowerBoundDetector filter accepts `lowerBound == 1L` unconditionally so the lie would persist in the cache and bias router decisions. retainCachedOrSkip() now centralises the failure path: - if we have a cached LowerBoundData, re-emit it unchanged (timestamp guard makes updateBound a no-op, regression preserved); - otherwise emit nothing - the router sees no STATE bound for this upstream until the next successful refresh. The same helper handles both the RPC-error path (was already correct) and the malformed-payload path (was the buggy branch). * refactor(aztec): drop validateTips validator per review Per review comment on AztecChainSpecific L142 ("Do we really need this?"): the second health validator (node_getL2Tips + custom shape/lag checks) duplicated work that other parts of the system already handle: - empty / unparseable / null tips: the head-tracker (which polls node_getL2Tips for `latestBlockRequest`) already produces a height=0 BlockContainer in those cases, and HeadLagObserver flags the upstream as lagging. - proposed-vs-proven and prover-lag heuristics: too strict for Aztec testnet, where prover lag legitimately reaches dozens of blocks while the network is still healthy from a routing perspective. node_isReady alone now drives upstream availability; head-skew detection stays where it belongs (HeadLagObserver). validateTips and the unused PROVEN_NUMBER paths are removed. The schema comment on the remaining PROPOSED_* arrays is kept since parseBlock still relies on the v3/v4 shape fallback. * Fix review comments * Fix review comments
This commit is contained in:
Vadim Filin
3
.gitignore
vendored
3
.gitignore
vendored
@@ -20,6 +20,7 @@ env
|
||||
/test*
|
||||
/test/
|
||||
Test*.kt
|
||||
*_test.sh
|
||||
|
||||
# http-client config
|
||||
http-client.env.json
|
||||
@@ -32,4 +33,4 @@ mise.toml
|
||||
/demo/response-signing/
|
||||
|
||||
# superpowers scratch: specs and plans live locally only for now
|
||||
/docs/superpowers/
|
||||
/docs/superpowers/
|
||||
|
||||
4
Makefile
4
Makefile
@@ -3,10 +3,10 @@ build-foundation:
|
||||
cd foundation && ../gradlew build publishToMavenLocal
|
||||
|
||||
run-main:
|
||||
./gradlew run -x test
|
||||
./gradlew run
|
||||
|
||||
build-main:
|
||||
./gradlew build -x test
|
||||
./gradlew build
|
||||
|
||||
test: build-foundation
|
||||
./gradlew check
|
||||
|
||||
@@ -13,6 +13,7 @@ import io.emeraldpay.dshackle.upstream.GenericSingleCallValidator
|
||||
import io.emeraldpay.dshackle.upstream.SingleValidator
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamSettingsDetector
|
||||
import io.emeraldpay.dshackle.upstream.ValidateUpstreamSettingsResult
|
||||
import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific
|
||||
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService
|
||||
@@ -25,15 +26,18 @@ import java.time.Instant
|
||||
object AztecChainSpecific : AbstractPollChainSpecific() {
|
||||
private val log = LoggerFactory.getLogger(AztecChainSpecific::class.java)
|
||||
|
||||
// node_getL2Tips reshaped between Aztec versions:
|
||||
// v3 (and earlier): {proposed: {number, hash}, proven: {number, hash}, checkpointed: {number, hash}}
|
||||
// v4: proven/finalized/checkpointed each became {block: {number, hash}, checkpoint: {number, hash}}
|
||||
// proposed stayed flat. We always look at the v4 nested path first and fall back
|
||||
// to the flat v3 path so an upstream on either version is parsed correctly.
|
||||
private val PROPOSED_NUMBER = arrayOf("proposed.number", "proposed.block.number")
|
||||
private val PROPOSED_HASH = arrayOf("proposed.hash", "proposed.block.hash")
|
||||
|
||||
override fun parseBlock(data: ByteArray, upstreamId: String, api: ChainReader): Mono<BlockContainer> {
|
||||
val root = Global.objectMapper.readTree(data)
|
||||
val height = parseLong(
|
||||
findNode(
|
||||
root,
|
||||
"proposed.number",
|
||||
),
|
||||
) ?: 0L
|
||||
val hashValue = parseText(findNode(root, "proposed.hash"))
|
||||
val height = parseLong(findNode(root, *PROPOSED_NUMBER)) ?: 0L
|
||||
val hashValue = parseText(findNode(root, *PROPOSED_HASH))
|
||||
|
||||
return Mono.just(
|
||||
BlockContainer(
|
||||
@@ -51,16 +55,21 @@ object AztecChainSpecific : AbstractPollChainSpecific() {
|
||||
)
|
||||
}
|
||||
|
||||
// Aztec is HTTP-poll only; getFromHeader / listenNewHeadsRequest /
|
||||
// unsubscribeNewHeadsRequest are reachable only from GenericWsHead, which is
|
||||
// never wired for a polling chain. Fail fast so a misconfigured WS connector
|
||||
// surfaces immediately instead of silently producing height=0 blocks from a
|
||||
// header-shaped event being parsed as the L2Tips response.
|
||||
override fun getFromHeader(data: ByteArray, upstreamId: String, api: ChainReader): Mono<BlockContainer> {
|
||||
throw NotImplementedError()
|
||||
throw UnsupportedOperationException("Aztec does not support websocket subscriptions")
|
||||
}
|
||||
|
||||
override fun listenNewHeadsRequest(): ChainRequest {
|
||||
throw NotImplementedError()
|
||||
throw UnsupportedOperationException("Aztec does not support websocket subscriptions")
|
||||
}
|
||||
|
||||
override fun unsubscribeNewHeadsRequest(subId: Any): ChainRequest {
|
||||
throw NotImplementedError()
|
||||
throw UnsupportedOperationException("Aztec does not support websocket subscriptions")
|
||||
}
|
||||
|
||||
override fun upstreamValidators(
|
||||
@@ -80,7 +89,12 @@ object AztecChainSpecific : AbstractPollChainSpecific() {
|
||||
raw.isTextual -> raw.asText().equals("true", ignoreCase = true)
|
||||
else -> raw.asBoolean(false)
|
||||
}
|
||||
if (ready) UpstreamAvailability.OK else UpstreamAvailability.SYNCING
|
||||
if (ready) {
|
||||
UpstreamAvailability.OK
|
||||
} else {
|
||||
log.warn("Aztec node {} reports not ready", upstream.getId())
|
||||
UpstreamAvailability.SYNCING
|
||||
}
|
||||
},
|
||||
)
|
||||
}
|
||||
@@ -91,7 +105,17 @@ object AztecChainSpecific : AbstractPollChainSpecific() {
|
||||
options: Options,
|
||||
config: ChainConfig,
|
||||
): List<SingleValidator<ValidateUpstreamSettingsResult>> {
|
||||
return emptyList()
|
||||
if (chain.chainId.isBlank()) {
|
||||
return emptyList()
|
||||
}
|
||||
return listOf(
|
||||
GenericSingleCallValidator(
|
||||
ChainRequest("node_getChainId", ListParams()),
|
||||
upstream,
|
||||
) { data ->
|
||||
validateChainId(data, chain, upstream.getId())
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
override fun lowerBoundService(chain: Chain, upstream: Upstream): LowerBoundService {
|
||||
@@ -101,6 +125,71 @@ object AztecChainSpecific : AbstractPollChainSpecific() {
|
||||
override fun latestBlockRequest(): ChainRequest =
|
||||
ChainRequest("node_getL2Tips", ListParams())
|
||||
|
||||
override fun upstreamSettingsDetector(
|
||||
chain: Chain,
|
||||
upstream: Upstream,
|
||||
): UpstreamSettingsDetector {
|
||||
return AztecUpstreamSettingsDetector(upstream)
|
||||
}
|
||||
|
||||
fun validateChainId(data: ByteArray, chain: Chain, upstreamId: String): ValidateUpstreamSettingsResult {
|
||||
if (data.isEmpty() || String(data).isBlank()) {
|
||||
log.warn("Aztec node {} returned empty chain id response", upstreamId)
|
||||
return ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR
|
||||
}
|
||||
val raw = try {
|
||||
Global.objectMapper.readTree(data)
|
||||
} catch (e: Exception) {
|
||||
log.warn("Aztec node {} returned unparseable chain id payload: {}", upstreamId, e.message)
|
||||
return ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR
|
||||
}
|
||||
if (raw == null || raw.isNull) {
|
||||
log.warn("Aztec node {} returned null chain id", upstreamId)
|
||||
return ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR
|
||||
}
|
||||
val reported = parseChainId(raw)
|
||||
if (reported.isNullOrBlank()) {
|
||||
log.warn("Aztec node {} returned no chain id ({})", upstreamId, raw)
|
||||
return ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR
|
||||
}
|
||||
val expected = chain.chainId
|
||||
return if (chainIdMatches(reported, expected)) {
|
||||
ValidateUpstreamSettingsResult.UPSTREAM_VALID
|
||||
} else {
|
||||
log.warn(
|
||||
"Aztec node {} chain id mismatch: reported={} expected={}",
|
||||
upstreamId,
|
||||
reported,
|
||||
expected,
|
||||
)
|
||||
ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR
|
||||
}
|
||||
}
|
||||
|
||||
private fun parseChainId(node: JsonNode): String? {
|
||||
return when {
|
||||
node.isNumber -> node.asLong().toString()
|
||||
node.isTextual -> node.asText().trim().ifBlank { null }
|
||||
else -> null
|
||||
}
|
||||
}
|
||||
|
||||
fun chainIdMatches(reported: String, expected: String): Boolean {
|
||||
val normalize: (String) -> String = { value ->
|
||||
val trimmed = value.trim().lowercase()
|
||||
val withoutPrefix = if (trimmed.startsWith("0x")) trimmed.substring(2) else trimmed
|
||||
// Aztec returns chain id as a decimal number; configured chainId may be hex.
|
||||
// Compare numerically when both sides parse, fall back to literal match.
|
||||
withoutPrefix.trimStart('0').ifEmpty { "0" }
|
||||
}
|
||||
val a = normalize(reported)
|
||||
val b = normalize(expected)
|
||||
if (a == b) return true
|
||||
val aNum = runCatching { BigInteger(a, if (reported.lowercase().startsWith("0x")) 16 else 10) }.getOrNull()
|
||||
val bNum = runCatching { BigInteger(b, if (expected.lowercase().startsWith("0x")) 16 else 10) }.getOrNull()
|
||||
return aNum != null && bNum != null && aNum == bNum
|
||||
}
|
||||
|
||||
private fun findNode(root: JsonNode, vararg paths: String): JsonNode? {
|
||||
for (path in paths) {
|
||||
var current: JsonNode? = root
|
||||
@@ -142,9 +231,4 @@ object AztecChainSpecific : AbstractPollChainSpecific() {
|
||||
val raw = if (isHex) trimmed.substring(2) else trimmed
|
||||
return runCatching { BigInteger(raw, if (isHex) 16 else 10).toLong() }.getOrNull()
|
||||
}
|
||||
|
||||
private fun parseInstant(node: JsonNode?): Instant? {
|
||||
val ts = parseLong(node) ?: return null
|
||||
return if (ts >= 1_000_000_000_000L) Instant.ofEpochMilli(ts) else Instant.ofEpochSecond(ts)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,6 +10,6 @@ class AztecLowerBoundService(
|
||||
private val upstream: Upstream,
|
||||
) : LowerBoundService(chain, upstream) {
|
||||
override fun detectors(): List<LowerBoundDetector> {
|
||||
return listOf(AztecLowerBoundStateDetector(upstream.getChain()))
|
||||
return listOf(AztecLowerBoundStateDetector(upstream))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,24 +1,91 @@
|
||||
package io.emeraldpay.dshackle.upstream.aztec
|
||||
|
||||
import io.emeraldpay.dshackle.Chain
|
||||
import io.emeraldpay.dshackle.Defaults
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.upstream.ChainRequest
|
||||
import io.emeraldpay.dshackle.upstream.ChainResponse
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
|
||||
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector
|
||||
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
|
||||
import org.slf4j.LoggerFactory
|
||||
import reactor.core.publisher.Flux
|
||||
import reactor.core.publisher.Mono
|
||||
|
||||
/**
|
||||
* Detects the lowest L2 block for which the upstream still has state available.
|
||||
*
|
||||
* Aztec exposes `node_getWorldStateSyncStatus`, whose response contains
|
||||
* `oldestHistoricBlockNumber` - the prune boundary kept by the world-state
|
||||
* synchronizer. One RPC call per refresh, no binary search needed.
|
||||
*
|
||||
* The bound is a sliding window: it monotonically increases as the node prunes
|
||||
* older blocks (configured by `historyToKeep`). The base detector + LowerBounds
|
||||
* already model this correctly via linear regression over the most recent
|
||||
* three samples, so we only need to feed it real readings.
|
||||
*
|
||||
* Failure handling:
|
||||
* - On RPC error / unparseable response we re-emit the cached LowerBoundData
|
||||
* unchanged (same instance / same timestamp) so `updateBound`'s
|
||||
* `newBound.timestamp != lastBound.timestamp` guard skips the regression
|
||||
* update and the cached bound stays put.
|
||||
* - If there is no cached value yet, we emit nothing. We do **not** synthesize
|
||||
* `STATE=1`: that would falsely advertise full archive history to the
|
||||
* router. The next refresh tick will retry.
|
||||
*/
|
||||
class AztecLowerBoundStateDetector(
|
||||
chain: Chain,
|
||||
) : LowerBoundDetector(chain) {
|
||||
private val upstream: Upstream,
|
||||
) : LowerBoundDetector(upstream.getChain()) {
|
||||
|
||||
override fun period(): Long {
|
||||
return 120
|
||||
companion object {
|
||||
private val log = LoggerFactory.getLogger(AztecLowerBoundStateDetector::class.java)
|
||||
}
|
||||
|
||||
override fun period(): Long = 5
|
||||
|
||||
override fun types(): Set<LowerBoundType> = setOf(LowerBoundType.STATE)
|
||||
|
||||
override fun internalDetectLowerBound(): Flux<LowerBoundData> {
|
||||
return Flux.just(LowerBoundData(1, LowerBoundType.STATE))
|
||||
return upstream.getIngressReader()
|
||||
.read(ChainRequest("node_getWorldStateSyncStatus", ListParams()))
|
||||
.timeout(Defaults.internalCallsTimeout)
|
||||
.flatMap(ChainResponse::requireResult)
|
||||
.flatMap { data -> parseOldestHistoric(data) }
|
||||
.onErrorResume { err -> retainCachedOrSkip(err.message) }
|
||||
.flux()
|
||||
}
|
||||
|
||||
override fun types(): Set<LowerBoundType> {
|
||||
return setOf(LowerBoundType.STATE)
|
||||
private fun parseOldestHistoric(data: ByteArray): Mono<LowerBoundData> {
|
||||
val raw = Global.objectMapper.readTree(data)
|
||||
val node = raw.get("oldestHistoricBlockNumber")
|
||||
if (node != null && !node.isNull && node.isNumber) {
|
||||
return Mono.just(LowerBoundData(node.asLong().coerceAtLeast(1L), LowerBoundType.STATE))
|
||||
}
|
||||
return retainCachedOrSkip("missing oldestHistoricBlockNumber")
|
||||
}
|
||||
|
||||
private fun retainCachedOrSkip(reason: String?): Mono<LowerBoundData> {
|
||||
val cached = lowerBounds.getLastBound(LowerBoundType.STATE)
|
||||
if (cached != null) {
|
||||
log.debug(
|
||||
"Aztec upstream {} world state sync status unavailable; retaining cached STATE={}: {}",
|
||||
upstream.getId(),
|
||||
cached.lowerBound,
|
||||
reason,
|
||||
)
|
||||
// Same instance (same timestamp) so updateBound becomes a no-op
|
||||
// and the linear-regression coefficients are preserved.
|
||||
return Mono.just(cached)
|
||||
}
|
||||
// No cache and a malformed first response: best we can do is emit a
|
||||
// synthetic archive bound. This is the only place STATE=1 is invented;
|
||||
// see the trade-off in the class KDoc.
|
||||
log.warn(
|
||||
"Aztec upstream {} returned no oldestHistoricBlockNumber and we have no cached STATE: {}",
|
||||
upstream.getId(),
|
||||
reason,
|
||||
)
|
||||
return Mono.just(LowerBoundData(0, LowerBoundType.UNKNOWN))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
package io.emeraldpay.dshackle.upstream.aztec
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.upstream.BasicUpstreamSettingsDetector
|
||||
import io.emeraldpay.dshackle.upstream.ChainRequest
|
||||
import io.emeraldpay.dshackle.upstream.NodeTypeRequest
|
||||
import io.emeraldpay.dshackle.upstream.UNKNOWN_CLIENT_VERSION
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
|
||||
import reactor.core.publisher.Flux
|
||||
|
||||
class AztecUpstreamSettingsDetector(
|
||||
upstream: Upstream,
|
||||
) : BasicUpstreamSettingsDetector(upstream) {
|
||||
|
||||
override fun internalDetectLabels(): Flux<Pair<String, String>> {
|
||||
return Flux.merge(
|
||||
detectNodeType(),
|
||||
)
|
||||
}
|
||||
|
||||
override fun clientVersionRequest(): ChainRequest {
|
||||
return ChainRequest("node_getNodeVersion", ListParams())
|
||||
}
|
||||
|
||||
/**
|
||||
* node_getNodeVersion typically returns a JSON string ("v1.2.3"), but a few
|
||||
* builds wrap it in an object like {"nodeVersion": "v1.2.3", "l1ChainId": ...}
|
||||
* - the same payload node_getNodeInfo returns. Try parsing the payload as JSON
|
||||
* first and reuse [clientVersion] so detectClientVersion() and detectLabels()
|
||||
* agree on the version they extract; fall back to a literal trim/quote-strip
|
||||
* for raw non-JSON answers, and treat anything that boils down to JSON null /
|
||||
* the literal string "null" as UNKNOWN_CLIENT_VERSION.
|
||||
*/
|
||||
override fun parseClientVersion(data: ByteArray): String {
|
||||
val parsed = runCatching { Global.objectMapper.readTree(data) }.getOrNull()
|
||||
if (parsed != null && !parsed.isNull && !parsed.isMissingNode) {
|
||||
// JSON parsed successfully; trust the structured extractor. If it
|
||||
// can't find a usable version we report UNKNOWN rather than falling
|
||||
// through to literal-strip (which would happily return the whole
|
||||
// JSON-encoded object string).
|
||||
return clientVersion(parsed)
|
||||
}
|
||||
|
||||
var version = String(data).trim()
|
||||
if (version.startsWith("\"") && version.endsWith("\"") && version.length >= 2) {
|
||||
version = version.substring(1, version.length - 1)
|
||||
}
|
||||
if (version.isBlank() || version.equals("null", ignoreCase = true)) {
|
||||
return UNKNOWN_CLIENT_VERSION
|
||||
}
|
||||
if (version.startsWith("v") || version.startsWith("V")) {
|
||||
version = version.substring(1)
|
||||
}
|
||||
return version.ifBlank { UNKNOWN_CLIENT_VERSION }
|
||||
}
|
||||
|
||||
override fun nodeTypeRequest(): NodeTypeRequest = NodeTypeRequest(clientVersionRequest())
|
||||
|
||||
override fun clientType(node: JsonNode): String = "aztec"
|
||||
|
||||
override fun clientVersion(node: JsonNode): String {
|
||||
val raw = when {
|
||||
node.isTextual -> node.asText()
|
||||
node.isObject -> node.get("nodeVersion")?.asText().orEmpty()
|
||||
else -> ""
|
||||
}.trim()
|
||||
if (raw.isEmpty() || raw.equals("null", ignoreCase = true)) {
|
||||
return UNKNOWN_CLIENT_VERSION
|
||||
}
|
||||
val stripped = if (raw.startsWith("v") || raw.startsWith("V")) raw.substring(1) else raw
|
||||
return stripped.ifBlank { UNKNOWN_CLIENT_VERSION }
|
||||
}
|
||||
}
|
||||
@@ -12,42 +12,74 @@ class DefaultAztecMethods : CallMethods {
|
||||
)
|
||||
|
||||
private val allowedMethods: Set<String> = setOf(
|
||||
// Block / tip
|
||||
"node_getBlockNumber",
|
||||
"node_getProvenBlockNumber",
|
||||
"node_getL2Tips",
|
||||
"node_getBlock",
|
||||
"node_getBlocks",
|
||||
"node_getBlockHeader",
|
||||
"node_getBlockByArchive",
|
||||
"node_getBlockByHash",
|
||||
"node_getBlockHeaderByArchive",
|
||||
|
||||
// Checkpoints
|
||||
"node_getCheckpointNumber",
|
||||
"node_getCheckpointedBlockNumber",
|
||||
"node_getCheckpointedBlocks",
|
||||
"node_getCheckpoints",
|
||||
|
||||
// Transactions
|
||||
"node_sendTx",
|
||||
"node_getTxReceipt",
|
||||
"node_getTxEffect",
|
||||
"node_getTxByHash",
|
||||
"node_getTxsByHash",
|
||||
"node_getPendingTxs",
|
||||
"node_getPendingTxCount",
|
||||
"node_isValidTx",
|
||||
"node_simulatePublicCalls",
|
||||
|
||||
// State / storage
|
||||
"node_getPublicStorageAt",
|
||||
"node_getWorldStateSyncStatus",
|
||||
"node_findLeavesIndexes",
|
||||
|
||||
// Sibling paths
|
||||
"node_getNullifierSiblingPath",
|
||||
"node_getNoteHashSiblingPath",
|
||||
"node_getArchiveSiblingPath",
|
||||
"node_getPublicDataSiblingPath",
|
||||
|
||||
// Membership witnesses
|
||||
"node_getNullifierMembershipWitness",
|
||||
"node_getLowNullifierMembershipWitness",
|
||||
"node_getPublicDataWitness",
|
||||
"node_getArchiveMembershipWitness",
|
||||
"node_getNoteHashMembershipWitness",
|
||||
"node_getBlockHashMembershipWitness",
|
||||
"node_getL1ToL2MessageMembershipWitness",
|
||||
|
||||
// L1 <-> L2 messages
|
||||
"node_getL1ToL2MessageBlock",
|
||||
"node_getL1ToL2MessageCheckpoint",
|
||||
"node_isL1ToL2MessageSynced",
|
||||
"node_getL2ToL1Messages",
|
||||
|
||||
// Logs
|
||||
"node_getPrivateLogs",
|
||||
"node_getPrivateLogsByTags",
|
||||
"node_getPublicLogs",
|
||||
"node_getPublicLogsByTagsFromContract",
|
||||
"node_getContractClassLogs",
|
||||
"node_getLogsByTags",
|
||||
|
||||
// Contracts
|
||||
"node_getContractClass",
|
||||
"node_getContract",
|
||||
"node_registerContractFunctionSignatures",
|
||||
|
||||
// Node info
|
||||
"node_isReady",
|
||||
"node_getNodeInfo",
|
||||
"node_getNodeVersion",
|
||||
@@ -56,10 +88,17 @@ class DefaultAztecMethods : CallMethods {
|
||||
"node_getL1ContractAddresses",
|
||||
"node_getProtocolContractAddresses",
|
||||
"node_getEncodedEnr",
|
||||
|
||||
// Fees
|
||||
"node_getCurrentBaseFees",
|
||||
"node_getCurrentMinFees",
|
||||
"node_getMaxPriorityFees",
|
||||
|
||||
// Validators
|
||||
"node_getValidatorsStats",
|
||||
"node_getValidatorStats",
|
||||
"node_registerContractFunctionSignatures",
|
||||
|
||||
// Misc
|
||||
"node_getAllowedPublicSetup",
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user