diff --git a/.gitignore b/.gitignore index 5c031af2..fa52543b 100644 --- a/.gitignore +++ b/.gitignore @@ -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/ \ No newline at end of file +/docs/superpowers/ diff --git a/Makefile b/Makefile index dc37beb4..bc1bac9b 100644 --- a/Makefile +++ b/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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecChainSpecific.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecChainSpecific.kt index af4ca028..0400c2ab 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecChainSpecific.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecChainSpecific.kt @@ -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 { 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 { - 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> { - 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) - } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecLowerBoundService.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecLowerBoundService.kt index 69f327aa..874ec882 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecLowerBoundService.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecLowerBoundService.kt @@ -10,6 +10,6 @@ class AztecLowerBoundService( private val upstream: Upstream, ) : LowerBoundService(chain, upstream) { override fun detectors(): List { - return listOf(AztecLowerBoundStateDetector(upstream.getChain())) + return listOf(AztecLowerBoundStateDetector(upstream)) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecLowerBoundStateDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecLowerBoundStateDetector.kt index fc29f1ab..8f6bbe6d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecLowerBoundStateDetector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecLowerBoundStateDetector.kt @@ -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 = setOf(LowerBoundType.STATE) + override fun internalDetectLowerBound(): Flux { - 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 { - return setOf(LowerBoundType.STATE) + private fun parseOldestHistoric(data: ByteArray): Mono { + 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 { + 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)) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecUpstreamSettingsDetector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecUpstreamSettingsDetector.kt new file mode 100644 index 00000000..bb279f62 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/aztec/AztecUpstreamSettingsDetector.kt @@ -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> { + 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 } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultAztecMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultAztecMethods.kt index ede7c902..a7be791c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultAztecMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultAztecMethods.kt @@ -12,42 +12,74 @@ class DefaultAztecMethods : CallMethods { ) private val allowedMethods: Set = 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", )