diff --git a/Cargo.lock b/Cargo.lock index b80db7e91d..7ea13ea62a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1069,7 +1069,7 @@ checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" [[package]] name = "api-types" version = "0.1.0" -source = "git+https://github.com/Galxe/gravity-aptos?rev=b1f68dc85781ef0d28a568d9d64604b153be9d9e#b1f68dc85781ef0d28a568d9d64604b153be9d9e" +source = "git+https://github.com/Galxe/gravity-aptos?rev=a64f8adc274bf2681df796766ef9a5b195fee44b#a64f8adc274bf2681df796766ef9a5b195fee44b" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 2b23b9d6bc..ce9d88cdc4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -366,7 +366,7 @@ incremental = false [workspace.dependencies] # reth -gravity-api-types = { package = "api-types", git = "https://github.com/Galxe/gravity-aptos", rev = "b1f68dc85781ef0d28a568d9d64604b153be9d9e" } +gravity-api-types = { package = "api-types", git = "https://github.com/Galxe/gravity-aptos", rev = "a64f8adc274bf2681df796766ef9a5b195fee44b" } reth = { path = "bin/reth" } reth-storage-rpc-provider = { path = "crates/storage/rpc-provider" } reth-basic-payload-builder = { path = "crates/payload/basic" } diff --git a/crates/pipe-exec-layer-ext-v2/execute/src/lib.rs b/crates/pipe-exec-layer-ext-v2/execute/src/lib.rs index 3d6e5b6730..1b56e1c108 100644 --- a/crates/pipe-exec-layer-ext-v2/execute/src/lib.rs +++ b/crates/pipe-exec-layer-ext-v2/execute/src/lib.rs @@ -77,7 +77,7 @@ use crate::{ construct_metadata_txn, construct_validator_txn_from_extra_data, dkg::{convert_dkg_start_event_to_api, DKGStartEvent}, system_txns_into_executed_ordered_block_result, - types::DataRecorded, + types::{DataRecorded, OracleDelivered}, SystemTxnResult, DKG_ADDR, NATIVE_MINT_PRECOMPILE_ADDR, NATIVE_ORACLE_ADDR, RANDOMNESS_BY_HEIGHT_PRECOMPILE_ADDR, SYSTEM_CALLER, }, @@ -105,26 +105,33 @@ fn extract_gravity_events_from_system_receipts( "extract gravity events from receipt" ); for log in &receipt.logs { - // Parse DataRecorded events only from NativeOracle. + // Parse both historical and current delivery events only from NativeOracle. if log.address == NATIVE_ORACLE_ADDR { - if let Ok(event) = DataRecorded::decode_log(&log) { + let delivery = DataRecorded::decode_log(&log) + .map(|event| (event.sourceType, event.sourceId, event.nonce)) + .or_else(|_| { + OracleDelivered::decode_log(&log) + .map(|event| (event.sourceType, event.sourceId, event.nonce)) + }); + + if let Ok((source_type, source_id, nonce)) = delivery { info!(target: "execute_ordered_block", number=?block_number, - source_type=?event.sourceType, - source_id=?event.sourceId, - nonce=?event.nonce, - "data recorded event" + source_type=?source_type, + source_id=?source_id, + nonce=?nonce, + "oracle delivery event" ); // Keep only the latest nonce for each (sourceType, sourceId) - let key = (event.sourceType, event.sourceId); + let key = (source_type, source_id); data_records .entry(key) .and_modify(|existing_nonce| { - if event.nonce > *existing_nonce { - *existing_nonce = event.nonce; + if nonce > *existing_nonce { + *existing_nonce = nonce; } }) - .or_insert(event.nonce); + .or_insert(nonce); } } @@ -142,7 +149,7 @@ fn extract_gravity_events_from_system_receipts( } } - // Convert collected DataRecorded events to ProviderJWKs + // Convert collected delivery events to ProviderJWKs. if !data_records.is_empty() { let api_jwks: Vec = data_records .into_iter() @@ -168,7 +175,7 @@ fn extract_gravity_events_from_system_receipts( number=?block_number, epoch=?epoch, provider_count=?api_jwks.len(), - "constructed ProviderJWKs from DataRecorded events" + "constructed ProviderJWKs from oracle delivery events" ); gravity_events.push(GravityEvent::ObservedJWKsUpdated(epoch, api_jwks)); @@ -222,6 +229,34 @@ mod tests { } } + #[test] + fn extract_gravity_events_accepts_oracle_delivered_from_native_oracle_only() { + let event = OracleDelivered { + sourceType: 3, + sourceId: U256::from(4), + nonce: 5, + sourcePosition: 6, + payloadHash: B256::from([0x77; 32]), + }; + let forged_receipts = + vec![receipt_with_log(Address::from([0x42; 20]), event.encode_log_data())]; + + let events = extract_gravity_events_from_system_receipts(&forged_receipts, 10, 7); + assert!(events.is_empty(), "forged OracleDelivered emitter must not produce GravityEvent"); + + let valid_receipts = vec![receipt_with_log(NATIVE_ORACLE_ADDR, event.encode_log_data())]; + let events = extract_gravity_events_from_system_receipts(&valid_receipts, 10, 7); + assert_eq!(events.len(), 1); + match &events[0] { + GravityEvent::ObservedJWKsUpdated(epoch, jwks) => { + assert_eq!(*epoch, 7); + assert_eq!(jwks.len(), 1); + assert_eq!(jwks[0].version, 5); + } + other => panic!("expected ObservedJWKsUpdated, got {other:?}"), + } + } + #[test] fn extract_gravity_events_ignores_dkg_start_from_wrong_emitter() { let event = DKGStartEvent { diff --git a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/errors.rs b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/errors.rs index 2538c0795a..2b69901df7 100644 --- a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/errors.rs +++ b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/errors.rs @@ -56,10 +56,21 @@ sol! { /// @notice Nonce must be strictly increasing for each source error NonceNotIncreasing(uint32 sourceType, uint256 sourceId, uint128 currentNonce, uint128 providedNonce); + /// @notice Nonce must be exactly the next value for each source + error NonceNotSequential(uint32 sourceType, uint256 sourceId, uint128 expectedNonce, uint128 providedNonce); + /// @notice Batch arrays have mismatched lengths - error OracleBatchArrayLengthMismatch(uint256 noncesLength, uint256 payloadsLength, uint256 gasLimitsLength); + error OracleBatchArrayLengthMismatch( + uint256 noncesLength, + uint256 blockNumbersLength, + uint256 payloadsLength, + uint256 gasLimitsLength + ); + + /// @notice Oracle source position exceeds the contract's uint128 range + error OracleSourcePositionOverflow(uint256 sourcePosition); - // -------------------- JWKManager Errors (for reference, callback failures don't revert main tx) -------------------- + // -------------------- JWKManager Errors (for reference) -------------------- /// @notice JWK version must be strictly increasing error JWKVersionNotIncreasing(bytes issuer, uint64 currentVersion, uint64 providedVersion); } @@ -182,13 +193,25 @@ pub fn decode_revert_error(output: &Bytes) -> Option { Some(SystemTxnError { name: "OracleBatchArrayLengthMismatch".into(), details: format!( - "Array length mismatch: nonces={}, payloads={}, gasLimits={}", - err.noncesLength, err.payloadsLength, err.gasLimitsLength + "Array length mismatch: nonces={}, positions={}, payloads={}, gasLimits={}", + err.noncesLength, + err.blockNumbersLength, + err.payloadsLength, + err.gasLimitsLength ), severity: ErrorSeverity::Fatal, }) } + s if s == OracleSourcePositionOverflow::SELECTOR => { + let err = OracleSourcePositionOverflow::abi_decode(output).ok()?; + Some(SystemTxnError { + name: "OracleSourcePositionOverflow".into(), + details: format!("Oracle source position exceeds uint128: {}", err.sourcePosition), + severity: ErrorSeverity::Fatal, + }) + } + // -------------------- Recoverable Errors -------------------- s if s == ReconfigurationNotInProgress::SELECTOR => Some(SystemTxnError { name: "ReconfigurationNotInProgress".into(), @@ -226,6 +249,18 @@ pub fn decode_revert_error(output: &Bytes) -> Option { }) } + s if s == NonceNotSequential::SELECTOR => { + let err = NonceNotSequential::abi_decode(output).ok()?; + Some(SystemTxnError { + name: "NonceNotSequential".into(), + details: format!( + "Oracle nonce not sequential: sourceType={}, sourceId={}, expected={}, provided={}", + err.sourceType, err.sourceId, err.expectedNonce, err.providedNonce + ), + severity: ErrorSeverity::Recoverable, + }) + } + // Unknown selector _ => None, } @@ -334,6 +369,37 @@ mod tests { assert_eq!(err.severity, ErrorSeverity::Recoverable); } + #[test] + fn test_decode_nonce_not_sequential() { + let error = NonceNotSequential { + sourceType: 3, + sourceId: alloy_primitives::U256::from(42), + expectedNonce: 11, + providedNonce: 13, + }; + let result = decode_revert_error(&error.abi_encode().into()).unwrap(); + + assert_eq!(result.name, "NonceNotSequential"); + assert_eq!(result.severity, ErrorSeverity::Recoverable); + assert!(result.details.contains("expected=11")); + assert!(result.details.contains("provided=13")); + } + + #[test] + fn test_decode_current_batch_length_mismatch() { + let error = OracleBatchArrayLengthMismatch { + noncesLength: alloy_primitives::U256::from(1), + blockNumbersLength: alloy_primitives::U256::from(2), + payloadsLength: alloy_primitives::U256::from(3), + gasLimitsLength: alloy_primitives::U256::from(4), + }; + let result = decode_revert_error(&error.abi_encode().into()).unwrap(); + + assert_eq!(result.name, "OracleBatchArrayLengthMismatch"); + assert_eq!(result.severity, ErrorSeverity::Fatal); + assert!(result.details.contains("positions=2")); + } + #[test] fn test_decode_unknown_error() { // Random bytes that don't match any known error diff --git a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/jwk_oracle.rs b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/jwk_oracle.rs index bcff797cb5..a8d61da6b1 100644 --- a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/jwk_oracle.rs +++ b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/jwk_oracle.rs @@ -1,29 +1,22 @@ -//! JWK Oracle module for writing oracle updates via NativeOracle.record() +//! JWK Oracle write path for consensus-approved relayer payloads. //! -//! This module handles the WRITE path for ALL oracle updates in the new Oracle architecture: -//! - RSA JWKs: NativeOracle.record(sourceType=1, sourceId=keccak256(issuer)) -//! - UnsupportedJWK (blockchain events): NativeOracle.recordBatch() for multiple logs -//! -//! For blockchain events, the payload from relayer is ABI-encoded and passed through unchanged. -//! This ensures byte-exact match between relayer, on-chain storage, and read-back for comparison. +//! UnsupportedJWK entries carry a canonical ABI wrapper containing +//! `(nonce, source_position, callback_payload)`. The NativeOracle ABI retains +//! the legacy `blockNumber` name, but the value is a source-defined restart +//! position and payload history is no longer stored by NativeOracle. use super::{new_system_call_txn, NATIVE_ORACLE_ADDR}; use alloy_primitives::{Bytes, U256}; use alloy_sol_macro::sol; -use alloy_sol_types::SolCall; +use alloy_sol_types::{SolCall, SolValue}; use gravity_api_types::on_chain_config::jwks::{JWKStruct, ProviderJWKs}; use reth_ethereum_primitives::TransactionSigned; +use reth_pipe_exec_layer_relayer::{parse_oracle_uri, source_types}; use tracing::{debug, info, warn}; -/// Default callback gas limit for oracle updates const CALLBACK_GAS_LIMIT: u64 = 500_000; -// ============================================================================= -// Solidity Types (NativeOracle function signatures) -// ============================================================================= - sol! { - /// NativeOracle.record() function signature function record( uint32 sourceType, uint256 sourceId, @@ -33,7 +26,6 @@ sol! { uint256 callbackGasLimit ) external; - /// NativeOracle.recordBatch() function signature for multiple events function recordBatch( uint32 sourceType, uint256 sourceId, @@ -44,98 +36,40 @@ sol! { ) external; } -// ============================================================================= -// Helper Functions -// ============================================================================= - -/// Check if a JWKStruct is an RSA JWK fn is_rsa_jwk(jwk: &JWKStruct) -> bool { jwk.type_name == "0x1::jwks::RSA_JWK" } -/// Check if a JWKStruct is an UnsupportedJWK (blockchain/other oracle data) fn is_unsupported_jwk(jwk: &JWKStruct) -> bool { jwk.type_name == "0x1::jwks::Unsupported_JWK" } -/// Parse chain_id from issuer URI -/// Format: gravity://{source_type}/{chain_id}/{task_type}?... -fn parse_chain_id_from_issuer(issuer: &[u8]) -> Option { - let issuer_str = String::from_utf8_lossy(issuer); - if issuer_str.starts_with("gravity://") { - let after_protocol = &issuer_str[10..]; - // Skip source_type (first segment), get chain_id (second segment) - // Format: {source_type}/{chain_id}/{task_type}?... - let mut parts = after_protocol.split('/'); - let _source_type = parts.next()?; // Skip source_type - let chain_id_str = parts.next()?; - return chain_id_str.parse().ok(); - } - None +fn parse_source_from_issuer(issuer: &[u8]) -> Option<(u32, u64)> { + let issuer = std::str::from_utf8(issuer).ok()?; + let task = parse_oracle_uri(issuer).ok()?; + Some((task.source_type, task.source_id)) } -/// Extract nonce, block_number, and inner payload from ABI-encoded event data -/// Payload format: alloy's abi_encode(&(u128, U256, &[u8])) -/// -/// alloy encodes (u128, U256, &[u8]) as a dynamic tuple with structure: -/// - bytes 0-31: offset to tuple data (always 32 = 0x20) -/// - bytes 32-63: nonce (uint128, right-aligned, so nonce is at bytes 48-63) -/// - bytes 64-95: block_number (uint256) -/// - bytes 96-127: offset to bytes data (relative to tuple start at byte 32) -/// - bytes 128-159: payload length -/// - bytes 160+: payload data -/// -/// Returns (nonce, block_number, inner_payload) -fn extract_nonce_block_and_payload(data: &[u8]) -> Option<(u128, U256, Vec)> { - if data.len() < 160 { - warn!( - target: "gravity::onchain_config::jwk_oracle", - data_len = data.len(), - "Data too short for ABI decoding" - ); - return None; +fn callback_gas_limit(source_type: u32) -> Result { + match source_type { + source_types::BLOCKCHAIN => Ok(CALLBACK_GAS_LIMIT), + _ => Err(format!("Unsupported oracle source type: {source_type}")), } +} - // nonce is at bytes 32-63, right-aligned u128 so actual value is at bytes 48-63 - let nonce_bytes = &data[48..64]; - let nonce = u128::from_be_bytes(nonce_bytes.try_into().ok()?); - - // block_number is at bytes 64-95 (full U256) - let block_number = U256::from_be_slice(&data[64..96]); - - // Payload length is at bytes 128-159 (right-aligned u256) - let length_bytes = &data[128..160]; - let payload_len = u64::from_be_bytes(length_bytes[24..32].try_into().ok()?) as usize; - - // Check we have enough data for the payload - if data.len() < 160 + payload_len { - warn!( - target: "gravity::onchain_config::jwk_oracle", - data_len = data.len(), - payload_len = payload_len, - "Not enough data for payload" - ); - return None; +fn extract_canonical_wrapper(data: &[u8]) -> Result<(u128, U256, Vec), String> { + let decoded = <(u128, U256, Bytes)>::abi_decode(data) + .map_err(|error| format!("Failed to decode oracle payload wrapper: {error}"))?; + if decoded.abi_encode() != data { + return Err("Oracle payload wrapper is not canonically encoded".to_string()); + } + if decoded.1 > U256::from(u128::MAX) { + return Err("Oracle source position exceeds NativeOracle uint128 range".to_string()); } - // Extract the inner payload starting at byte 160 - let inner_payload = data[160..160 + payload_len].to_vec(); - - Some((nonce, block_number, inner_payload)) + Ok((decoded.0, decoded.1, decoded.2.to_vec())) } -// ============================================================================= -// Public API -// ============================================================================= - -/// Construct transaction for oracle update via NativeOracle.record() -/// -/// This is the unified entry point for ALL oracle updates. It routes based on JWK type: -/// - RSA_JWK → sourceType=1 (JWK), payload=ABI(issuer, version, jwks[]) -/// - UnsupportedJWK → Uses recordBatch for ALL logs (payload passed through unchanged) -/// -/// Note: All JWKs in provider_jwks.jwks are guaranteed to be of the same type -/// (either all RSA or all unsupported), so we only check the first element. pub fn construct_oracle_record_transaction( provider_jwks: ProviderJWKs, nonce: u64, @@ -144,46 +78,30 @@ pub fn construct_oracle_record_transaction( let issuer = &provider_jwks.issuer; let issuer_str = String::from_utf8_lossy(issuer); - // All JWKs are homogeneous, check the first one to determine the type let first_jwk = provider_jwks .jwks .first() - .ok_or_else(|| format!("No JWKs found for issuer: {}", issuer_str))?; + .ok_or_else(|| format!("No JWKs found for issuer: {issuer_str}"))?; if is_rsa_jwk(first_jwk) { - // RSA JWK path is not exercised in production today — all JWK data flows through the - // UnsupportedJWK (blockchain event) path, and the RSA record construction has never - // been audited/exercised. Fail CLOSED: return a recoverable `Err` rather than run - // unverified construction logic. It must NEVER panic — this runs on the deterministic - // execute_ordered_block system-tx path (over consensus `extra_data`), whose `Err` the - // caller logs + skips (lib.rs), so a panic here would deterministically halt every - // validator on this ordered block (gravity-audit#822 class). warn!( target: "gravity::onchain_config::jwk_oracle", issuer = %issuer_str, jwk_count = provider_jwks.jwks.len(), - "RSA JWK path entered unexpectedly — rejecting (unsupported in production)" + "RSA JWK path entered unexpectedly; rejecting unsupported execution path" ); Err(format!( - "RSA JWK oracle record path is not supported: issuer={}, jwk_count={}", - issuer_str, + "RSA JWK oracle record path is not supported: issuer={issuer_str}, jwk_count={}", provider_jwks.jwks.len() )) } else if is_unsupported_jwk(first_jwk) { - // Blockchain/oracle events - use recordBatch for ALL logs - construct_blockchain_batch_transaction(provider_jwks, nonce, gas_price) + construct_unsupported_oracle_batch_transaction(provider_jwks, nonce, gas_price) } else { - warn!(target: "gravity::onchain_config::jwk_oracle", "Unknown JWK type '{}' for issuer: {}", first_jwk.type_name, issuer_str); - Err(format!("Unknown JWK type '{}' for issuer: {}", first_jwk.type_name, issuer_str)) + Err(format!("Unknown JWK type '{}' for issuer: {issuer_str}", first_jwk.type_name)) } } -/// Construct transaction for blockchain events using recordBatch() -/// -/// This handles ALL UnsupportedJWK entries (each represents one event). -/// The payload is passed through UNCHANGED from relayer - this ensures -/// byte-exact match between what relayer sends and what gets stored on-chain. -fn construct_blockchain_batch_transaction( +fn construct_unsupported_oracle_batch_transaction( provider_jwks: ProviderJWKs, nonce: u64, gas_price: u128, @@ -191,80 +109,180 @@ fn construct_blockchain_batch_transaction( let issuer = &provider_jwks.issuer; let jwks = &provider_jwks.jwks; - // Parse chain_id from issuer - let chain_id = parse_chain_id_from_issuer(issuer) - .ok_or_else(|| format!("Failed to parse chain_id from issuer: {:?}", issuer))?; - info!(target: "gravity::onchain_config::jwk_oracle", "jwk chain_id: {}, len {:?}", chain_id, jwks.len()); + let (source_type, source_id) = parse_source_from_issuer(issuer) + .ok_or_else(|| format!("Failed to parse source coordinates from issuer: {issuer:?}"))?; + let callback_gas_limit = callback_gas_limit(source_type)?; - // All JWKs are guaranteed to be unsupported type when entering this function if jwks.is_empty() { - return Err("No blockchain event JWKs found".to_string()); + return Err("No oracle entries found".to_string()); } - // Parse sourceType from first JWK's type_name (all have same type) - let source_type: u32 = 0; - - // Build batch arrays - let mut nonces: Vec = Vec::with_capacity(jwks.len()); - let mut block_numbers: Vec = Vec::with_capacity(jwks.len()); - let mut payloads: Vec = Vec::with_capacity(jwks.len()); - let mut gas_limits: Vec = Vec::with_capacity(jwks.len()); - - for (idx, jwk) in jwks.iter().enumerate() { - let (event_nonce, block_number, inner_payload) = - match extract_nonce_block_and_payload(&jwk.data) { - Some((nonce, block_num, payload)) => (nonce, block_num, payload), - None => { - warn!( - target: "gravity::onchain_config::jwk_oracle", - idx = idx, - payload_len = jwk.data.len(), - payload_hex = %hex::encode(&jwk.data), - "Failed to extract nonce, block_number, and payload" - ); - return Err(format!( - "Failed to extract nonce, block_number, and payload at index {}", - idx - )); - } - }; + let mut nonces = Vec::with_capacity(jwks.len()); + let mut source_positions = Vec::with_capacity(jwks.len()); + let mut payloads = Vec::with_capacity(jwks.len()); + let mut gas_limits = Vec::with_capacity(jwks.len()); + let mut previous_nonce: Option = None; + + for (index, jwk) in jwks.iter().enumerate() { + if !is_unsupported_jwk(jwk) { + return Err(format!("Mixed JWK types in unsupported oracle batch at index {index}")); + } + + let (event_nonce, source_position, callback_payload) = extract_canonical_wrapper(&jwk.data) + .map_err(|error| { + warn!( + target: "gravity::onchain_config::jwk_oracle", + index, + payload_len = jwk.data.len(), + %error, + "Rejected oracle payload wrapper" + ); + format!("Invalid oracle payload wrapper at index {index}: {error}") + })?; + + if let Some(previous) = previous_nonce { + let expected = + previous.checked_add(1).ok_or_else(|| "Oracle batch nonce overflow".to_string())?; + if event_nonce != expected { + return Err(format!( + "Oracle batch nonces are not sequential at index {index}: expected {expected}, got {event_nonce}" + )); + } + } + previous_nonce = Some(event_nonce); nonces.push(event_nonce); - block_numbers.push(block_number); - // Use the inner payload (the original MessageSent.payload) - // This is what the user put in and what gets passed to the callback - payloads.push(inner_payload.into()); - gas_limits.push(U256::from(CALLBACK_GAS_LIMIT)); + source_positions.push(source_position); + payloads.push(Bytes::from(callback_payload)); + gas_limits.push(U256::from(callback_gas_limit)); debug!( - idx = idx, - event_nonce = event_nonce, - ?block_number, - inner_payload_len = payloads.last().map(|p: &Bytes| p.len()).unwrap_or(0), - "Added event to batch" + target: "gravity::onchain_config::jwk_oracle", + index, + event_nonce, + ?source_position, + "Added canonical oracle entry to batch" ); } info!( + target: "gravity::onchain_config::jwk_oracle", issuer = %String::from_utf8_lossy(issuer), - chain_id = chain_id, - source_type = source_type, - event_count = nonces.len(), - "Constructing blockchain recordBatch transaction (pass-through payload)" + source_type, + source_id, + item_count = nonces.len(), + "Constructing NativeOracle recordBatch transaction" ); - // Use recordBatch for multiple events let call = recordBatchCall { sourceType: source_type, - sourceId: U256::from(chain_id), + sourceId: U256::from(source_id), nonces, - blockNumbers: block_numbers, + blockNumbers: source_positions, payloads, callbackGasLimits: gas_limits, }; - let input: Bytes = call.abi_encode().into(); - Ok(new_system_call_txn(NATIVE_ORACLE_ADDR, nonce, gas_price, input)) + Ok(new_system_call_txn(NATIVE_ORACLE_ADDR, nonce, gas_price, call.abi_encode().into())) } -// convert_oracle_rsa_to_api_jwk is now provided by super::types +#[cfg(test)] +mod tests { + use super::*; + use alloy_consensus::Transaction; + + fn wrapped_jwk(nonce: u128, position: U256, payload: &[u8]) -> JWKStruct { + JWKStruct { + type_name: "0x1::jwks::Unsupported_JWK".to_string(), + data: (nonce, position, payload).abi_encode(), + } + } + + fn provider(uri: &[u8], jwks: Vec) -> ProviderJWKs { + ProviderJWKs { issuer: uri.to_vec(), version: 1, jwks } + } + + #[test] + fn parses_source_coordinates_from_issuer() { + let issuer = b"gravity://0/1/events?fromBlock=22000000"; + assert_eq!(parse_source_from_issuer(issuer), Some((0, 1))); + } + + #[test] + fn extracts_canonical_wrapper() { + let encoded = (7u128, U256::from(3020), b"oracle-payload".as_slice()).abi_encode(); + let (nonce, position, payload) = extract_canonical_wrapper(&encoded).unwrap(); + assert_eq!(nonce, 7); + assert_eq!(position, U256::from(3020)); + assert_eq!(payload, b"oracle-payload"); + } + + #[test] + fn rejects_noncanonical_wrapper() { + let mut encoded = (7u128, U256::from(3020), b"oracle-payload".as_slice()).abi_encode(); + encoded.push(0); + assert!(extract_canonical_wrapper(&encoded).is_err()); + } + + #[test] + fn preserves_source_zero_coordinates_and_callback_gas() { + let payload = b"bridge-event-payload"; + let provider = provider( + b"gravity://0/1/events?fromBlock=22000000", + vec![wrapped_jwk(7, U256::from(22_000_123u64), payload)], + ); + + let tx = construct_oracle_record_transaction(provider, 0, 0).unwrap(); + let call = recordBatchCall::abi_decode(tx.input()).unwrap(); + assert_eq!(call.sourceType, source_types::BLOCKCHAIN); + assert_eq!(call.sourceId, U256::from(1)); + assert_eq!(call.nonces, vec![7]); + assert_eq!(call.blockNumbers, vec![U256::from(22_000_123u64)]); + assert_eq!(call.payloads, vec![Bytes::copy_from_slice(payload)]); + assert_eq!(call.callbackGasLimits, vec![U256::from(CALLBACK_GAS_LIMIT)]); + } + + #[test] + fn rejects_provider_source_types_not_implemented_by_core() { + let provider = provider( + b"gravity://3/1001/price_feed", + vec![wrapped_jwk(1, U256::from(60_000), b"price")], + ); + let error = construct_oracle_record_transaction(provider, 0, 0).unwrap_err(); + assert_eq!(error, "Unsupported oracle source type: 3"); + } + + #[test] + fn rejects_nonsequential_batch_before_execution() { + let provider = provider( + b"gravity://0/1/events", + vec![ + wrapped_jwk(7, U256::from(100), b"first"), + wrapped_jwk(9, U256::from(101), b"gap"), + ], + ); + let error = construct_oracle_record_transaction(provider, 0, 0).unwrap_err(); + assert!(error.contains("expected 8, got 9")); + } + + #[test] + fn rejects_mixed_jwk_types() { + let mut mixed = wrapped_jwk(8, U256::from(101), b"mixed"); + mixed.type_name = "0x1::jwks::RSA_JWK".to_string(); + let provider = provider( + b"gravity://0/1/events", + vec![wrapped_jwk(7, U256::from(100), b"first"), mixed], + ); + let error = construct_oracle_record_transaction(provider, 0, 0).unwrap_err(); + assert!(error.contains("Mixed JWK types")); + } + + #[test] + fn rejects_source_position_outside_contract_range() { + let position = U256::from(u128::MAX) + U256::from(1); + let provider = + provider(b"gravity://0/1/events", vec![wrapped_jwk(1, position, b"payload")]); + let error = construct_oracle_record_transaction(provider, 0, 0).unwrap_err(); + assert!(error.contains("uint128 range")); + } +} diff --git a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/oracle_state.rs b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/oracle_state.rs index e222eaa3a4..827506b21d 100644 --- a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/oracle_state.rs +++ b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/oracle_state.rs @@ -1,12 +1,11 @@ -//! Oracle State Fetcher +//! Oracle source progress fetcher. //! -//! This module provides functionality to fetch the latest DataRecord from NativeOracle -//! for registered oracle tasks. It builds on top of oracle_task_helpers to get source -//! information and then fetches the corresponding data records. +//! New NativeOracle deployments expose one fixed-size progress checkpoint per +//! source. The legacy record fallback keeps pre-hardfork blocks readable. use super::{ base::OnchainConfigFetcher, - oracle_task_helpers::{OracleTaskClient, SOURCE_TYPE_BLOCKCHAIN}, + oracle_task_helpers::{OracleTaskClient, RELAYER_BACKED_SOURCE_TYPES}, NATIVE_ORACLE_ADDR, SYSTEM_CALLER, }; use alloy_eips::BlockId; @@ -14,26 +13,28 @@ use alloy_primitives::{Bytes, U256}; use alloy_rpc_types_eth::TransactionRequest; use alloy_sol_macro::sol; use alloy_sol_types::SolCall; -use gravity_api_types::on_chain_config::oracle_state::{LatestDataRecord, OracleSourceState}; +use gravity_api_types::on_chain_config::oracle_state::OracleSourceState; use reth_rpc_eth_api::{helpers::EthCall, RpcTypes}; -use tracing::info; - -// ============================================================================= -// ABI Definitions -// ============================================================================= +use tracing::{info, warn}; sol! { - /// DataRecord struct matching INativeOracle.DataRecord + struct SourceProgress { + uint128 latestNonce; + uint128 latestPosition; + } + + function getSourceProgress( + uint32 sourceType, + uint256 sourceId + ) external view returns (SourceProgress memory progress); + + /// Legacy record retained for pre-hardfork state fallback. struct DataRecord { - /// Timestamp when this was recorded (0 = not exists) uint64 recordedAt; - /// Block number when this was created uint256 blockNumber; - /// Stored payload data bytes data; } - /// Get a record by its key tuple function getRecord( uint32 sourceType, uint256 sourceId, @@ -41,11 +42,6 @@ sol! { ) external view returns (DataRecord memory record); } -// ============================================================================= -// Oracle State Client -// ============================================================================= - -/// Client for fetching oracle state (latest records) from NativeOracle #[derive(Debug)] pub struct OracleStateFetcher<'a, EthApi> { base_fetcher: &'a OnchainConfigFetcher, @@ -60,8 +56,20 @@ where Self { base_fetcher } } - /// Call NativeOracle.getRecord() to fetch a specific record - pub fn call_get_record( + fn call_get_source_progress( + &self, + source_type: u32, + source_id: U256, + block_id: BlockId, + ) -> Option { + let call = getSourceProgressCall { sourceType: source_type, sourceId: source_id }; + let input: Bytes = call.abi_encode().into(); + let result = + self.base_fetcher.eth_call(SYSTEM_CALLER, NATIVE_ORACLE_ADDR, input, block_id).ok()?; + getSourceProgressCall::abi_decode_returns(&result).ok() + } + + fn call_get_record( &self, source_type: u32, source_id: U256, @@ -70,109 +78,218 @@ where ) -> Option { let call = getRecordCall { sourceType: source_type, sourceId: source_id, nonce }; let input: Bytes = call.abi_encode().into(); - let result = self.base_fetcher.eth_call(SYSTEM_CALLER, NATIVE_ORACLE_ADDR, input, block_id).ok()?; - getRecordCall::abi_decode_returns(&result).ok() } - /// Fetch the latest DataRecord for a source - /// - /// Uses the latest nonce to fetch the corresponding record. - /// Returns None if nonce is 0 (no records exist). - pub fn fetch_latest_record( + /// Read the fixed-size getter, falling back to legacy nonce and record state. + fn try_fetch_source_progress( &self, + task_client: &OracleTaskClient<'_, EthApi>, source_type: u32, source_id: U256, - latest_nonce: u128, block_id: BlockId, - ) -> Option { + ) -> Option<(u128, u128)> { + if let Some(progress) = self.call_get_source_progress(source_type, source_id, block_id) { + return Some((progress.latestNonce, progress.latestPosition)); + } + + let latest_nonce = task_client.call_get_latest_nonce(source_type, source_id, block_id)?; if latest_nonce == 0 { - return None; + return Some((0, 0)); } let record = self.call_get_record(source_type, source_id, latest_nonce, block_id)?; + legacy_progress(latest_nonce, &record) + } - // Check if record exists (recordedAt > 0) - if record.recordedAt == 0 { - return None; - } + /// Fetch all currently supported relayer source states as a BCS snapshot. + pub fn fetch(&self, block_id: BlockId) -> Option { + let task_client = OracleTaskClient::new(self.base_fetcher); + let results = collect_source_states( + RELAYER_BACKED_SOURCE_TYPES, + |source_type| { + let source_ids = task_client.fetch_registered_source_ids(source_type, block_id); + if source_ids.is_none() { + warn!( + target: "oracle_state", + source_type, + "Failed to fetch or decode registered oracle source ids" + ); + } + source_ids + }, + |source_type, source_id| { + let progress = + self.try_fetch_source_progress(&task_client, source_type, source_id, block_id); + if progress.is_none() { + warn!( + target: "oracle_state", + source_type, + source_id = source_id.to_string(), + "Failed to fetch or decode oracle source progress" + ); + } + progress + }, + )?; - Some(LatestDataRecord { - recorded_at: record.recordedAt, - block_number: record.blockNumber.try_into().unwrap_or(0), - data: record.data.to_vec(), - }) + bcs::to_bytes(&results) + .map(Bytes::from) + .map_err(|error| { + warn!( + target: "oracle_state", + %error, + "Failed to BCS serialize OracleSourceStates" + ); + }) + .ok() } - /// Fetch all oracle source states for registered blockchain tasks - /// - /// Returns BCS-serialized OracleSourceStates for registered sources. - pub fn fetch(&self, block_id: BlockId) -> Option { + pub fn fetch_source_state( + &self, + source_type: u32, + source_id: U256, + block_id: BlockId, + ) -> Option { let task_client = OracleTaskClient::new(self.base_fetcher); - let mut results = Vec::new(); + let (latest_nonce, latest_position) = + self.try_fetch_source_progress(&task_client, source_type, source_id, block_id)?; + + Some(OracleSourceState { + source_type, + source_id: source_id.try_into().ok()?, + latest_nonce, + latest_position, + }) + } +} + +fn legacy_progress(latest_nonce: u128, record: &DataRecord) -> Option<(u128, u128)> { + if record.recordedAt == 0 { + return Some((latest_nonce, 0)); + } - // Get all registered blockchain source IDs - let source_ids = task_client - .fetch_registered_source_ids(SOURCE_TYPE_BLOCKCHAIN, block_id) - .unwrap_or_default(); + Some((latest_nonce, record.blockNumber.try_into().ok()?)) +} + +fn collect_source_states( + source_types: &[u32], + mut source_ids_for: SourceIds, + mut progress_for: Progress, +) -> Option> +where + SourceIds: FnMut(u32) -> Option>, + Progress: FnMut(u32, U256) -> Option<(u128, u128)>, +{ + let mut results = Vec::new(); + for source_type in source_types { + let source_ids = source_ids_for(*source_type)?; info!( target: "oracle_state", + source_type, source_count = source_ids.len(), "Fetching oracle source states" ); for source_id in source_ids { - // Get latest nonce for this source - let latest_nonce = task_client - .call_get_latest_nonce(SOURCE_TYPE_BLOCKCHAIN, source_id, block_id) - .unwrap_or(0); - - // Fetch the latest record if nonce > 0 - let latest_record = - self.fetch_latest_record(SOURCE_TYPE_BLOCKCHAIN, source_id, latest_nonce, block_id); - + let (latest_nonce, latest_position) = progress_for(*source_type, source_id)?; info!( target: "oracle_state", + source_type, source_id = source_id.to_string(), latest_nonce, - has_record = latest_record.is_some(), + latest_position, "Fetched oracle source state" ); results.push(OracleSourceState { - source_type: SOURCE_TYPE_BLOCKCHAIN, - source_id: source_id.try_into().unwrap_or(0), + source_type: *source_type, + source_id: source_id.try_into().ok()?, latest_nonce, - latest_record, + latest_position, }); } + } - Some(bcs::to_bytes(&results).expect("Failed to BCS serialize OracleSourceStates").into()) + Some(results) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::cell::Cell; + + #[test] + fn authoritative_empty_source_lists_are_valid() { + let states = collect_source_states( + RELAYER_BACKED_SOURCE_TYPES, + |_| Some(vec![]), + |_, _| panic!("progress fetch must not run without sources"), + ) + .expect("empty source lists are authoritative"); + assert!(states.is_empty()); } - /// Fetch a specific source's state by source ID - pub fn fetch_source_state( - &self, - source_type: u32, - source_id: U256, - block_id: BlockId, - ) -> OracleSourceState { - let task_client = OracleTaskClient::new(self.base_fetcher); + #[test] + fn source_id_read_failure_invalidates_whole_snapshot() { + let calls = Cell::new(0); + let states = collect_source_states( + &[0, 3], + |_| { + let call = calls.get(); + calls.set(call + 1); + (call != 1).then(Vec::new) + }, + |_, _| panic!("progress fetch must not run without sources"), + ); + assert!(states.is_none()); + assert_eq!(calls.get(), 2); + } + + #[test] + fn progress_read_failure_invalidates_whole_snapshot() { + let source_type = RELAYER_BACKED_SOURCE_TYPES[0]; + let source_id = U256::from(42); + let states = collect_source_states(&[source_type], |_| Some(vec![source_id]), |_, _| None); + assert!(states.is_none()); + } - let latest_nonce = - task_client.call_get_latest_nonce(source_type, source_id, block_id).unwrap_or(0); + #[test] + fn progress_is_included_without_payload_history() { + let source_type = RELAYER_BACKED_SOURCE_TYPES[0]; + let source_id = U256::from(42); + let states = collect_source_states( + &[source_type], + |_| Some(vec![source_id]), + |_, _| Some((7, 12_345)), + ) + .expect("progress is authoritative"); - let latest_record = - self.fetch_latest_record(source_type, source_id, latest_nonce, block_id); + assert_eq!(states.len(), 1); + assert_eq!(states[0].source_type, source_type); + assert_eq!(states[0].source_id, 42); + assert_eq!(states[0].latest_nonce, 7); + assert_eq!(states[0].latest_position, 12_345); + } - OracleSourceState { - source_type, - source_id: source_id.try_into().unwrap_or(0), - latest_nonce, - latest_record, - } + #[test] + fn legacy_record_without_payload_history_marks_position_unknown() { + let record = DataRecord { recordedAt: 0, blockNumber: U256::ZERO, data: Bytes::new() }; + assert_eq!(legacy_progress(7, &record), Some((7, 0))); + } + + #[test] + fn zero_progress_is_valid() { + let source_type = RELAYER_BACKED_SOURCE_TYPES[0]; + let source_id = U256::from(42); + let states = + collect_source_states(&[source_type], |_| Some(vec![source_id]), |_, _| Some((0, 0))) + .expect("zero progress is a valid empty source state"); + + assert_eq!(states[0].latest_nonce, 0); + assert_eq!(states[0].latest_position, 0); } } diff --git a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/oracle_task_helpers.rs b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/oracle_task_helpers.rs index 3076445e58..80c74c692d 100644 --- a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/oracle_task_helpers.rs +++ b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/oracle_task_helpers.rs @@ -23,9 +23,36 @@ use tracing::{info, warn}; /// Source type for blockchain events in NativeOracle pub const SOURCE_TYPE_BLOCKCHAIN: u32 = 0; +/// Source types implemented by this relayer build. +/// +/// Provider PRs extend this list together with their runtime dispatch. Keeping +/// discovery and execution support in one list prevents publishing tasks that +/// the local relayer cannot execute. +pub const RELAYER_BACKED_SOURCE_TYPES: &[u32] = &[SOURCE_TYPE_BLOCKCHAIN]; + +fn has_valid_task_cardinality(source_type: u32, task_count: usize) -> bool { + !RELAYER_BACKED_SOURCE_TYPES.contains(&source_type) || task_count == 1 +} + // Re-export SOURCE_TYPE_JWK from types for consistency pub use super::types::SOURCE_TYPE_JWK; +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn relayer_backed_sources_require_exactly_one_task() { + for source_type in RELAYER_BACKED_SOURCE_TYPES { + assert!(!has_valid_task_cardinality(*source_type, 0)); + assert!(has_valid_task_cardinality(*source_type, 1)); + assert!(!has_valid_task_cardinality(*source_type, 2)); + } + + assert!(has_valid_task_cardinality(SOURCE_TYPE_JWK, 2)); + } +} + // ============================================================================= // Shared ABI Definitions // ============================================================================= @@ -161,20 +188,33 @@ where /// Returns a vector of (URI, nonce) tuples for all configured blockchain tasks. /// Returns tasks even when nonce is 0 (no data recorded yet) to enable discovery. pub fn fetch_blockchain_task_uris(&self, block_id: BlockId) -> Vec<(String, u128)> { + self.fetch_task_uris(SOURCE_TYPE_BLOCKCHAIN, block_id) + } + + /// Fetch all task URIs currently implemented by the relayer. + pub fn fetch_relayer_task_uris(&self, block_id: BlockId) -> Vec<(String, u128)> { + RELAYER_BACKED_SOURCE_TYPES + .iter() + .flat_map(|source_type| self.fetch_task_uris(*source_type, block_id)) + .collect() + } + + /// Fetch task URIs for a source type with their latest NativeOracle nonce. + pub fn fetch_task_uris(&self, source_type: u32, block_id: BlockId) -> Vec<(String, u128)> { let mut results = Vec::new(); - // Get all registered blockchain source IDs let source_ids = - self.fetch_registered_source_ids(SOURCE_TYPE_BLOCKCHAIN, block_id).unwrap_or_default(); + self.fetch_registered_source_ids(source_type, block_id).unwrap_or_default(); info!( target: "oracle_task_helper", + source_type, length = source_ids.len(), "oracle task source ids length" ); for source_id in source_ids { - self.process_source_tasks(source_id, block_id, &mut results); + self.process_source_tasks(source_type, source_id, block_id, &mut results); } results @@ -183,24 +223,23 @@ where /// Process all tasks for a single source ID fn process_source_tasks( &self, + source_type: u32, source_id: U256, block_id: BlockId, results: &mut Vec<(String, u128)>, ) { // Fetch the latest nonce for this source (0 if no data recorded yet) - let nonce = - self.call_get_latest_nonce(SOURCE_TYPE_BLOCKCHAIN, source_id, block_id).unwrap_or(0); + let nonce = self.call_get_latest_nonce(source_type, source_id, block_id).unwrap_or(0); info!( target: "oracle_task_helper", + source_type, source_id = source_id.to_string(), nonce, "oracle task source id and nonce" ); - let Some(task_names) = - self.call_get_task_names(SOURCE_TYPE_BLOCKCHAIN, source_id, block_id) - else { + let Some(task_names) = self.call_get_task_names(source_type, source_id, block_id) else { return; }; @@ -210,14 +249,26 @@ where "oracle task task names length" ); + if !has_valid_task_cardinality(source_type, task_names.len()) { + warn!( + target: "oracle_task_helper", + source_type, + source_id = source_id.to_string(), + task_count = task_names.len(), + "Relayer-backed oracle source must have exactly one task; skipping source" + ); + return; + } + for task_name in task_names { - self.process_single_task(source_id, task_name, nonce, block_id, results); + self.process_single_task(source_type, source_id, task_name, nonce, block_id, results); } } /// Process a single task and add valid URI to results fn process_single_task( &self, + source_type: u32, source_id: U256, task_name: B256, nonce: u128, @@ -230,8 +281,7 @@ where "oracle task task name" ); - let Some(task) = self.call_get_task(SOURCE_TYPE_BLOCKCHAIN, source_id, task_name, block_id) - else { + let Some(task) = self.call_get_task(source_type, source_id, task_name, block_id) else { return; }; @@ -252,15 +302,29 @@ where "oracle task uri string" ); - // Validate URI + // Validate both URI syntax and its on-chain task identity. match reth_pipe_exec_layer_relayer::uri_parser::parse_oracle_uri(&uri_string) { - Ok(_) => { + Ok(parsed) + if parsed.source_type == source_type && + U256::from(parsed.source_id) == source_id => + { results.push((uri_string, nonce)); } + Ok(parsed) => { + warn!( + target: "oracle_task_helper", + uri_string, + expected_source_type = source_type, + expected_source_id = source_id.to_string(), + parsed_source_type = parsed.source_type, + parsed_source_id = parsed.source_id, + "Oracle task URI coordinates do not match registered source" + ); + } Err(e) => { warn!( target: "oracle_task_helper", - uri_string = uri_string, + uri_string, error = %e, "Failed to parse oracle URI" ); diff --git a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/types.rs b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/types.rs index a22784137a..10586ab1f4 100644 --- a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/types.rs +++ b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/types.rs @@ -222,8 +222,7 @@ sol! { } sol! { - /// DataRecorded event from NativeOracle contract - /// Emitted when data is recorded by the consensus engine via SYSTEM_CALLER + /// Legacy DataRecorded event from NativeOracle contract. /// @param sourceType The source type (0 = BLOCKCHAIN, 1 = JWK, etc.) /// @param sourceId The source identifier (e.g., chain ID for blockchains) /// @param nonce The nonce (block height, timestamp, etc.) @@ -234,6 +233,15 @@ sol! { uint128 nonce, uint256 dataLength ); + + /// OracleDelivered event emitted by the current NativeOracle contract. + event OracleDelivered( + uint32 indexed sourceType, + uint256 indexed sourceId, + uint128 nonce, + uint128 sourcePosition, + bytes32 payloadHash + ); } /// RSA JWK fields for BCS serialization - matches gravity-aptos struct order diff --git a/crates/pipe-exec-layer-ext-v2/relayer/README.md b/crates/pipe-exec-layer-ext-v2/relayer/README.md index c3580a5ce6..379da6deb8 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/README.md +++ b/crates/pipe-exec-layer-ext-v2/relayer/README.md @@ -1,335 +1,98 @@ -# Gravity Protocol Relayer +# Gravity Oracle Relayer Core -A URI parser and blockchain event relayer for the Gravity protocol. +This crate converts finalized external-source observations into the +UnsupportedJWK payloads used by Gravity validator consensus. This core slice +implements source type `0` (`GravityPortal.MessageSent`). Provider-specific +source types are added in separate modules and PRs. -## Features +## Task Identity -### URI Parser (UriParser) -Supports parsing Gravity URIs in the following formats: -- `gravity://mainnet/block?strategy=head` - Monitor latest block -- `gravity://mainnet/event?address=0x...&topic0=0x...` - Monitor contract events -- `gravity://mainnet/storage?account=0x...&slot=0x...` - Monitor storage slot changes -- `gravity://mainnet/account/0x.../activity?type=erc20_transfer` - Monitor account activity +Tasks use the following URI shape: -### Relayer (GravityRelayer) -- Periodically polls Ethereum nodes -- Maintains processing cursor state -- Detects data changes and generates update events -- Supports finalized block filtering -- Configurable polling intervals and block ranges - -### Relayer Manager (RelayerManager) -- Manages multiple relayers for different URIs -- Centralized lifecycle management -- Supports multiple RPC endpoints -- Provides unified interface for adding and polling URIs - -## Basic Usage - -```rust -use reth_pipe_exec_layer_relayer::{ - RelayerManager, UriParser -}; -use reth_tracing::{LayerInfo, LogFormat, RethTracer, Tracer}; -use tracing::level_filters::LevelFilter; -use tracing::info; - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - // Initialize tracing - let tracer = RethTracer::new().with_stdout(LayerInfo::new( - LogFormat::Terminal, - LevelFilter::INFO.to_string(), - "trace".to_string(), - None, - )); - tracer.init().unwrap(); - - // 1. Create URI parser - let parser = UriParser::new(); - - // 2. Parse task URI - let task = parser.parse("gravity://mainnet/event?address=0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48&topic0=0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")?; - - // 3. Create RelayerManager - let manager = RelayerManager::new(); - - // 4. Add URI to manager (no initial state required) - manager.add_uri(&task.original_uri, "https://rpc.ankr.com/eth").await?; - - // 5. Poll for updates - let current_state = manager.poll_uri(&task.original_uri).await?; - info!("Current state: {:?}", current_state); - - Ok(()) -} +```text +gravity:////? ``` -## Data Structures +The source type and source ID in the URI must match the coordinates under +which `OracleTaskConfig` registered the task. A relayer-backed +`(sourceType, sourceId)` has exactly one task because `NativeOracle` has one +nonce stream for that pair. -### ParsedTask -```rust -pub struct ParsedTask { - /// The parsed gravity task to be executed - pub task: GravityTask, - /// The original URI string that was parsed - pub original_uri: String, - /// The chain identifier (e.g., "mainnet", "testnet") - pub chain_specifier: String, -} -``` - -### GravityTask -```rust -pub enum GravityTask { - /// Monitor event task, contains a Filter object that can be directly used with Alloy - MonitorEvent(Filter), - /// Monitor block head task - MonitorBlockHead, - /// Monitor storage slot task - MonitorStorage { account: Address, slot: B256 }, - /// Monitor account activity task (abstract layer) - MonitorAccount { address: Address, activity_type: AccountActivityType }, -} -``` +Source type `0` example: -### ObserveState -```rust -pub struct ObserveState { - /// The block number at which the observation was made - pub block_number: u64, - /// The actual observed value (block, events, storage slot, or none) - pub observed_value: ObservedValue, - /// Chain timestamp to ensure consistency - pub timestamp: u64, - /// OnChain version for tracking changes - pub version: u64, -} +```text +gravity://0/1/events?portal=0x0000000000000000000000000000000000000001&fromBlock=19000000 ``` -### ObservedValue -```rust -pub enum ObservedValue { - /// Observed block information - Block { block_hash: B256, block_number: u64 }, - /// Observed event logs - Events { logs: Vec }, - /// Observed storage slot value - StorageSlot { slot: B256, value: B256 }, - /// No observation made - None, -} -``` +RPC URLs are local validator configuration. They are not stored in the task +URI or committed on-chain. -### EventLog -```rust -pub struct EventLog { - /// Contract address that emitted the event - pub address: Address, - /// Event topics (indexed parameters) - pub topics: Vec, - /// Event data (non-indexed parameters) - pub data: Vec, - /// Block number where the event occurred - pub block_number: u64, - /// Transaction hash that triggered the event - pub transaction_hash: B256, - /// Log index within the transaction - pub log_index: u64, -} -``` +## Canonical Delivery -## URI Format Examples +Each source observation becomes an `OracleData` value: -### Block Monitoring -```rust -// Monitor latest block -"gravity://mainnet/block?strategy=head" +```text +nonce strictly sequential source nonce +source_position source-defined restart position +payload callback payload ``` -### Event Monitoring -```rust -// Monitor specific contract events -"gravity://mainnet/event?address=0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48&topic0=0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" - -// Monitor events with multiple topics -"gravity://mainnet/event?address=0x...&topic0=0x...&topic1=0x..." - -// Monitor events with OR conditions -"gravity://mainnet/event?topic0=0x...,0x..." +The relayer submits the canonical ABI wrapper: -// Monitor events from a specific block number -"gravity://mainnet/event?address=0x...&topic0=0x...&fromBlock=1500" - -// Monitor events from block tags -"gravity://mainnet/event?address=0x...&topic0=0x...&fromBlock=latest" -"gravity://mainnet/event?address=0x...&topic0=0x...&fromBlock=finalized" -"gravity://mainnet/event?address=0x...&topic0=0x...&fromBlock=earliest" +```solidity +abi.encode(uint128 nonce, uint256 sourcePosition, bytes callbackPayload) ``` -### Storage Monitoring -```rust -// Monitor storage slot changes -"gravity://mainnet/storage?account=0x123456789abcdef123456789abcdef1234567890&slot=0x0000000000000000000000000000000000000000000000000000000000000001" -``` +After quorum, the execution layer decodes the wrapper and calls the unchanged +`NativeOracle.recordBatch` ABI. Its `blockNumbers` argument carries source +positions. NativeOracle invokes the configured callback atomically and stores +only the latest `(nonce, sourcePosition)` progress checkpoint. -### Account Activity Monitoring -```rust -// Monitor ERC20 transfers for specific address -"gravity://mainnet/account/0x123456789abcdef123456789abcdef1234567890/activity?type=erc20_transfer" +The execution adapter rejects: -// Monitor all transactions for specific address -"gravity://mainnet/account/0x123456789abcdef123456789abcdef1234567890/activity?type=all_transactions" -``` - -## Event Filter Parameters +- non-canonical ABI wrappers; +- mixed JWK variants; +- non-sequential batch nonces; +- source positions outside the NativeOracle `uint128` range; +- source types whose runtime provider is not compiled into the current core. -### fromBlock Parameter -The `fromBlock` parameter allows you to specify the starting block for event monitoring: +## Restart And Replay -```rust -// Start from a specific block number -"gravity://mainnet/event?address=0x...&topic0=0x...&fromBlock=1500" +The relayer persists three independent values per full task URI: -// Start from block tags -"gravity://mainnet/event?address=0x...&topic0=0x...&fromBlock=latest" // Latest block -"gravity://mainnet/event?address=0x...&topic0=0x...&fromBlock=finalized" // Finalized block -"gravity://mainnet/event?address=0x...&topic0=0x...&fromBlock=earliest" // Genesis block -``` +- the last locally returned nonce; +- the source position associated with that nonce; +- the latest scan cursor, including empty finalized scans. -**Supported fromBlock values:** -- **Block number**: Any positive integer (e.g., `1500`, `1000000`) -- **latest**: Start from the latest block -- **finalized**: Start from the latest finalized block (recommended for production) -- **earliest**: Start from the genesis block (block 0) +Startup reconciles that local checkpoint with authoritative NativeOracle +progress. Local state ahead of the chain is rolled back. Local state behind a +known on-chain position is fast-forwarded. -**Note**: If `fromBlock` is not specified, the relayer will start monitoring from the current finalized block by default. +Legacy NativeOracle state can contain `latestNonce > 0` with +`latestPosition == 0`. Zero means the old source position is unknown, not that +the source starts at block zero. Recovery uses the following watermarks: -## Advanced Usage +| Local checkpoint | Recovery cursor | +|---|---| +| Same nonce | Persisted scan cursor | +| Behind with a known local event position | That local event position | +| Missing, empty, or ahead | Task `fromBlock` | -### Multiple URI Management -```rust -use reth_pipe_exec_layer_relayer::RelayerManager; -use reth_tracing::{LayerInfo, LogFormat, RethTracer, Tracer}; -use tracing::level_filters::LevelFilter; -use tracing::info; +In every unknown-position case, the source starts with the authoritative +on-chain nonce and filters historical observations at or below it. The first +successful post-upgrade delivery establishes a known position. -#[tokio::main] -async fn main() -> anyhow::Result<()> { - // Initialize tracing - let tracer = RethTracer::new().with_stdout(LayerInfo::new( - LogFormat::Terminal, - LevelFilter::INFO.to_string(), - "trace".to_string(), - None, - )); - tracer.init().unwrap(); +Polls for the same URI are serialized. Different URIs can still poll in +parallel. This prevents concurrent observers from emitting the same local +scan range twice. - let manager = RelayerManager::new(); - - let uris = vec![ - "gravity://mainnet/block?strategy=head", - "gravity://mainnet/event?address=0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48&topic0=0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef", - "gravity://mainnet/storage?account=0x123456789abcdef123456789abcdef1234567890&slot=0x0", - ]; - - // Add multiple URIs (no initial state required) - for uri in &uris { - match manager.add_uri(uri, "https://rpc.ankr.com/eth").await { - Ok(()) => info!("Successfully added URI: {}", uri), - Err(e) => info!("Failed to add URI {}: {}", uri, e), - } - } - - // Poll all URIs - for uri in &uris { - match manager.poll_uri(uri).await { - Ok(state) => info!("URI {}: {:?}", uri, state), - Err(e) => info!("Error polling {}: {}", uri, e), - } - } - - Ok(()) -} -``` - -### Batch URI Parsing -```rust -use reth_pipe_exec_layer_relayer::UriParser; -use tracing::info; - -let parser = UriParser::new(); -let uris = vec![ - "gravity://mainnet/block?strategy=head".to_string(), - "gravity://mainnet/event?address=0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48".to_string(), -]; - -for uri in uris { - match parser.parse(&uri) { - Ok(task) => { - info!("Parsed URI: {} -> Chain: {}, Task: {:?}", - uri, task.chain_specifier, task.task); - } - Err(e) => { - info!("Failed to parse URI {}: {}", uri, e); - } - } -} -``` - -## Error Handling - -The library uses `anyhow::Result` for error handling. Common error scenarios include: - -- Invalid URI format -- Unsupported chain specifiers -- Missing required parameters -- Invalid Ethereum addresses or topics -- RPC connection failures -- Network timeouts - -## Performance Considerations - -1. **RPC Rate Limiting**: The library includes built-in retry logic with exponential backoff -2. **Finalized Blocks**: By default, only finalized blocks are processed to ensure consistency -3. **Cursor Management**: Efficient cursor tracking prevents reprocessing of already seen data -4. **Batch Operations**: Support for batch URI parsing and management - -## Dependencies - -- `alloy-primitives`: Ethereum primitives and types -- `alloy-rpc-types`: RPC types and filters -- `tokio`: Async runtime -- `anyhow`: Error handling -- `tracing`: Logging and debugging -- `serde`: Serialization/deserialization -- `reth-tracing`: Reth tracing utilities - -## Running Examples +## Validation ```bash -# Run the basic usage example -cargo run --example new_usage - -# Run tests -cargo test - -# Generate documentation -cargo doc --open +cargo test -p reth-pipe-exec-layer-relayer +cargo test -p reth-pipe-exec-layer-ext-v2 onchain_config +cargo clippy -p reth-pipe-exec-layer-relayer -p reth-pipe-exec-layer-ext-v2 --all-targets ``` -## Notes - -1. Ensure you provide valid Ethereum RPC endpoints -2. Storage slot monitoring requires RPC support for `eth_getStorageAt` method -3. Consider using longer polling intervals in production to avoid excessive RPC calls -4. The library automatically handles retries and backoff for failed requests -5. All operations are async and should be run in a Tokio runtime -6. Tracing initialization is required for proper logging - -## TODO - -- [ ] Add WebSocket support for real-time updates -- [ ] Support for more event filtering options -- [ ] Add metrics and monitoring capabilities -- [ ] Implement connection pooling for multiple RPC endpoints \ No newline at end of file +The ignored blockchain-source test requires an explicitly configured external +RPC and seeded events; it is not part of the offline unit suite. diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/blockchain_source.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/blockchain_source.rs index 0d4af2bd93..722cafc5ee 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/src/blockchain_source.rs +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/blockchain_source.rs @@ -7,7 +7,7 @@ use crate::{ eth_client::EthHttpCli, }; use alloy_primitives::{Address, Bytes, U256}; -use alloy_rpc_types::Filter; +use alloy_rpc_types::{Filter, Log as RpcLog}; use alloy_sol_macro::sol; use alloy_sol_types::SolEvent; use anyhow::{anyhow, Result}; @@ -17,7 +17,7 @@ use std::sync::{ Arc, }; use tokio::sync::Mutex; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; // GravityPortal.MessageSent event signature sol! { @@ -25,38 +25,11 @@ sol! { event MessageSent(uint128 indexed nonce, uint256 indexed blockNumber, bytes payload); } -/// Decode ABI-encoded `bytes` from Solidity event log data -/// -/// Solidity encodes `bytes` in events as: -/// - bytes 0-31: offset to data (typically 0x20 = 32) -/// - bytes 32-63: length of data -/// - bytes 64+: actual data -/// -/// Returns the raw bytes without the ABI encoding wrapper. -fn decode_abi_bytes(data: &[u8]) -> Option> { - // Minimum: offset (32) + length (32) = 64 bytes - if data.len() < 64 { - return None; - } - - // Read offset (last 8 bytes of first 32-byte word, as it's right-aligned) - let offset = u64::from_be_bytes(data[24..32].try_into().ok()?) as usize; - - // Offset should point to the length word - if offset + 32 > data.len() { - return None; - } - - // Read length (last 8 bytes of the length word) - let length = u64::from_be_bytes(data[offset + 24..offset + 32].try_into().ok()?) as usize; - - // Data starts after the length word - let data_start = offset + 32; - if data_start + length > data.len() { - return None; - } - - Some(data[data_start..data_start + length].to_vec()) +fn decode_message_sent(log: &RpcLog) -> Result<(u128, U256, Bytes)> { + let event = MessageSent::decode_log(&log.inner) + .map_err(|error| anyhow!("Failed to decode MessageSent event: {error}"))?; + let event = event.data; + Ok((event.nonce, event.blockNumber, event.payload)) } /// Represents the state of the last successfully processed event @@ -72,17 +45,36 @@ pub struct LastProcessedEvent { } impl LastProcessedEvent { - /// Create a new LastProcessedEvent - pub fn new(nonce: u128, block: u64) -> Self { + /// Create a new `LastProcessedEvent`. + pub const fn new(nonce: u128, block: u64) -> Self { Self { nonce, block } } /// Check if any event has been processed - pub fn is_initialized(&self) -> bool { + pub const fn is_initialized(&self) -> bool { self.nonce > 0 } } +fn canonicalize_events(mut events: Vec, last_nonce: u128) -> Result> { + events.sort_by_key(|event| event.nonce); + + let mut expected = last_nonce; + for event in &events { + expected = + expected.checked_add(1).ok_or_else(|| anyhow!("Blockchain event nonce overflow"))?; + if event.nonce != expected { + return Err(anyhow!( + "Non-sequential MessageSent events: expected nonce {}, got {}", + expected, + event.nonce + )); + } + } + + Ok(events) +} + /// Blockchain event source for monitoring GravityPortal.MessageSent events /// /// This is the primary data source for cross-chain message bridging. @@ -101,7 +93,7 @@ pub struct BlockchainEventSource { /// Ethereum RPC client rpc_client: Arc, - /// GravityPortal contract address + /// `GravityPortal` contract address portal_address: Address, /// Current block cursor for polling @@ -124,6 +116,27 @@ impl BlockchainEventSource { portal_address: Address, cursor: u64, latest_onchain_nonce: u128, + ) -> Result { + let latest_position = if latest_onchain_nonce > 0 { cursor } else { 0 }; + Self::new_with_progress( + chain_id, + rpc_url, + portal_address, + cursor, + latest_onchain_nonce, + latest_position, + ) + .await + } + + /// Create with separate local scan and confirmed source positions. + pub async fn new_with_progress( + chain_id: u64, + rpc_url: &str, + portal_address: Address, + cursor: u64, + latest_onchain_nonce: u128, + latest_position: u64, ) -> Result { let rpc_client = Arc::new(EthHttpCli::new(rpc_url)?); @@ -133,6 +146,7 @@ impl BlockchainEventSource { portal_address = ?portal_address, cursor = cursor, latest_onchain_nonce = latest_onchain_nonce, + latest_position, "Created BlockchainEventSource with persisted cursor (fast restart)" ); @@ -141,7 +155,10 @@ impl BlockchainEventSource { rpc_client, portal_address, cursor: AtomicU64::new(cursor), - last_processed: Mutex::new(LastProcessedEvent::new(latest_onchain_nonce, cursor)), + last_processed: Mutex::new(LastProcessedEvent::new( + latest_onchain_nonce, + latest_position, + )), }) } @@ -171,11 +188,7 @@ impl BlockchainEventSource { /// Get the last nonce we returned (for exactly-once tracking) pub async fn last_nonce(&self) -> Option { let state = self.last_processed.lock().await; - if state.is_initialized() { - Some(state.nonce) - } else { - None - } + state.is_initialized().then_some(state.nonce) } /// Set the last processed event (used when initializing from on-chain state) @@ -183,7 +196,7 @@ impl BlockchainEventSource { *self.last_processed.lock().await = LastProcessedEvent::new(nonce, block); } - /// Fast-forward both cursor and last_processed state + /// Fast-forward both cursor and `last_processed` state /// /// Use this when reconciling with on-chain state that is ahead of local state. /// Sets both the scanning cursor and the last processed event atomically. @@ -192,18 +205,22 @@ impl BlockchainEventSource { self.set_cursor(block); } + /// Reconcile confirmed progress without rewinding an already advanced scan cursor. + pub async fn reconcile_progress(&self, nonce: u128, position: u64) { + self.set_last_processed(nonce, position).await; + if position > 0 { + self.cursor.fetch_max(position, Ordering::Relaxed); + } + } + /// Get the block number where last event was emitted pub async fn last_nonce_block(&self) -> Option { let state = self.last_processed.lock().await; - if state.is_initialized() { - Some(state.block) - } else { - None - } + state.is_initialized().then_some(state.block) } /// Get the chain ID - pub fn chain_id(&self) -> u64 { + pub const fn chain_id(&self) -> u64 { self.chain_id } @@ -226,7 +243,8 @@ impl OracleDataSource for BlockchainEventSource { async fn poll(&self) -> Result> { let cursor = self.cursor.load(Ordering::Relaxed); let finalized_block = self.rpc_client.get_finalized_block_number().await?; - let to_block = std::cmp::min(cursor + Self::MAX_BLOCKS_PER_POLL, finalized_block); + let to_block = + std::cmp::min(cursor.saturating_add(Self::MAX_BLOCKS_PER_POLL), finalized_block); if to_block <= cursor { return Ok(vec![]); @@ -253,73 +271,50 @@ impl OracleDataSource for BlockchainEventSource { let last_nonce = self.last_processed.lock().await.nonce; for log in logs { - let nonce = if let Some(nonce_topic) = log.topics().get(1) { - let nonce_bytes = &nonce_topic.as_slice()[16..32]; - u128::from_be_bytes(nonce_bytes.try_into().unwrap_or_default()) - } else { - continue; - }; - - let block_number = if let Some(block_number_topic) = log.topics().get(2) { - U256::from_be_slice(block_number_topic.as_slice()) - } else { - continue; - }; + // The RPC filter already selects this event signature. Treat a malformed + // matching log as an error so the scan cursor cannot advance past it. + let (nonce, block_number, raw_payload) = decode_message_sent(&log)?; + let source_position: u64 = block_number + .try_into() + .map_err(|_| anyhow!("MessageSent block number exceeds u64"))?; // Strictly monotonic check: ignore events we've already processed if nonce <= last_nonce { continue; } - let log_data = log.data().data.clone(); - - // log.data is ABI-encoded `bytes payload` from Solidity event - // Format: offset (32 bytes) || length (32 bytes) || data (variable) - // We need to extract the raw PortalMessage bytes before re-encoding - let raw_payload = match decode_abi_bytes(&log_data) { - Some(payload) => payload, - None => { - warn!( - target: "blockchain_source", - chain_id = self.chain_id, - nonce = nonce, - log_data_len = log_data.len(), - "Failed to decode ABI bytes from log.data, skipping" - ); - continue; - } - }; - // ABI encode (nonce, raw_payload) together // This preserves the nonce when passing through JWKStruct // Format: abi.encode(uint128 nonce, bytes payload) // Now raw_payload is the actual PortalMessage (sender || messageNonce || message) - let encoded_payload = alloy_sol_types::SolValue::abi_encode(&( - nonce, - block_number, - raw_payload.as_slice(), - )); + let encoded_payload = + alloy_sol_types::SolValue::abi_encode(&(nonce, block_number, raw_payload.as_ref())); debug!( target: "blockchain_source", chain_id = self.chain_id, nonce = nonce, - log_data_len = log_data.len(), raw_payload_len = raw_payload.len(), encoded_payload_len = encoded_payload.len(), "Found new MessageSent event - decoded and re-encoded" ); - results.push(OracleData { nonce, payload: Bytes::from(encoded_payload) }); + results.push(OracleData { + nonce, + source_position, + payload: Bytes::from(encoded_payload), + }); } + let results = canonicalize_events(results, last_nonce)?; + // Update cursor self.cursor.store(to_block, Ordering::Relaxed); // Track max nonce for exactly-once semantics (atomic update of nonce + block) - if !results.is_empty() { - let max_nonce = results.iter().map(|d| d.nonce).max().unwrap(); - *self.last_processed.lock().await = LastProcessedEvent::new(max_nonce, to_block); + if let Some(last) = results.last() { + *self.last_processed.lock().await = + LastProcessedEvent::new(last.nonce, last.source_position); } let current_last_nonce = self.last_nonce().await; @@ -359,11 +354,11 @@ mod tests { // --nocapture // ========================================================================= - /// GravityPortal address on local Anvil (deterministic, nonce 1) + /// `GravityPortal` address on local Anvil (deterministic, nonce 1) const ANVIL_PORTAL_ADDRESS: &str = "0x0f761B1B3c1aC9232C9015A7276692560aD6a05F"; - /// GBridgeSender address on local Anvil (deterministic, nonce 2) - const ANVIL_SENDER_ADDRESS: &str = "0x3fc870008B1cc26f3614F14a726F8077227CA2c3"; + /// `GBridgeSender` address on local Anvil (deterministic, nonce 2) + const _ANVIL_SENDER_ADDRESS: &str = "0x3fc870008B1cc26f3614F14a726F8077227CA2c3"; /// Anvil RPC URL const ANVIL_RPC_URL: &str = "https://sepolia.drpc.org"; @@ -371,17 +366,18 @@ mod tests { /// Local Anvil chain ID const ANVIL_CHAIN_ID: u64 = 10201262; - /// PortalMessage format decoder for relayer output + /// `PortalMessage` format decoder for relayer output /// - /// Relayer output format: abi.encode((uint128 nonce, bytes raw_portal_message)) - /// Where raw_portal_message is: sender (20 bytes) || messageNonce (16 bytes) || message + /// Relayer output format: + /// abi.encode(uint128 nonce, uint256 sourcePosition, bytes `raw_portal_message`) + /// Where `raw_portal_message` is: sender (20 bytes) || messageNonce (16 bytes) || message /// /// ABI encoding structure: - /// - bytes 0-31: tuple offset (0x20) - /// - bytes 32-63: nonce (uint128, right-aligned, actual value at bytes 48-63) - /// - bytes 64-95: offset to bytes data (relative to tuple start) + /// - bytes 0-31: nonce (uint128, right-aligned) + /// - bytes 32-63: source position + /// - bytes 64-95: offset to bytes data /// - bytes 96-127: length of bytes data - /// - bytes 128+: raw PortalMessage data + /// - bytes 128+: raw `PortalMessage` data fn decode_portal_message(payload: &[u8]) -> Option<(Address, u128, Vec)> { // Minimum: tuple offset (32) + nonce (32) + bytes offset (32) + length (32) + min data (36) // = 164 @@ -389,7 +385,7 @@ mod tests { return None; } - // Extract the raw PortalMessage from ABI-encoded (nonce, bytes) tuple + // Extract the raw PortalMessage from the canonical relayer wrapper. // Length is at bytes 96-127 (right-aligned) let length = u64::from_be_bytes(payload[120..128].try_into().ok()?) as usize; @@ -424,7 +420,39 @@ mod tests { Some((amount, recipient)) } + fn oracle_data(nonce: u128, source_position: u64) -> OracleData { + OracleData { nonce, source_position, payload: Bytes::new() } + } + + #[test] + fn canonicalize_events_sorts_by_nonce() { + let events = vec![oracle_data(8, 102), oracle_data(7, 101)]; + let canonical = canonicalize_events(events, 6).unwrap(); + assert_eq!(canonical.iter().map(|event| event.nonce).collect::>(), vec![7, 8]); + assert_eq!(canonical.last().unwrap().source_position, 102); + } + + #[test] + fn canonicalize_events_rejects_gaps_without_advancing() { + let error = canonicalize_events(vec![oracle_data(8, 102)], 6).unwrap_err(); + assert!(error.to_string().contains("expected nonce 7, got 8")); + } + + #[test] + fn canonicalize_events_rejects_duplicates() { + let error = + canonicalize_events(vec![oracle_data(7, 101), oracle_data(7, 101)], 6).unwrap_err(); + assert!(error.to_string().contains("expected nonce 8, got 7")); + } + + #[test] + fn malformed_matching_log_is_rejected() { + let error = decode_message_sent(&RpcLog::default()).unwrap_err(); + assert!(error.to_string().contains("Failed to decode MessageSent event")); + } + #[tokio::test] + #[ignore = "requires a configured external RPC endpoint and seeded events"] async fn test_poll_anvil_events() { use crate::eth_client::EthHttpCli; @@ -501,15 +529,15 @@ mod tests { } // Verify we got at least one event if bridge_test.sh was run - if !events.is_empty() { + if events.is_empty() { + println!("No events found. Make sure to run:"); + println!(" 1. ./scripts/start_anvil.sh"); + println!(" 2. ./scripts/bridge_test.sh"); + } else { println!("✓ Successfully polled and decoded events!"); let first_event = &events[0]; assert!(!first_event.payload.is_empty(), "Payload should not be empty"); - } else { - println!("No events found. Make sure to run:"); - println!(" 1. ./scripts/start_anvil.sh"); - println!(" 2. ./scripts/bridge_test.sh"); } } } diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/data_source.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/data_source.rs index 1cc9fdd709..12a052b502 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/src/data_source.rs +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/data_source.rs @@ -11,14 +11,17 @@ use crate::blockchain_source::BlockchainEventSource; /// Data returned by oracle data sources /// -/// This is the unified format for all data that flows into NativeOracle. +/// This is the unified format for all data that flows into `NativeOracle`. #[derive(Debug, Clone)] pub struct OracleData { /// Strictly increasing nonce for this (sourceType, sourceId) pair /// - For Blockchain: MessageSent.nonce pub nonce: u128, - /// ABI-encoded payload to be stored in NativeOracle + /// Source-defined restart position committed with this delivery + pub source_position: u64, + + /// ABI-encoded consensus payload delivered to `NativeOracle` pub payload: Bytes, } @@ -38,12 +41,12 @@ pub trait OracleDataSource: Send + Sync { /// Poll for new data /// - /// Returns a list of (nonce, payload) pairs that should be recorded in NativeOracle. + /// Returns a list of (nonce, payload) pairs that should be recorded in `NativeOracle`. /// The implementation should track its own cursor to avoid returning duplicates. async fn poll(&self) -> Result>; } -/// Source type constants (matching NativeOracle) +/// Source type constants (matching `NativeOracle`) pub mod source_types { /// Blockchain cross-chain events (e.g., GravityPortal.MessageSent) pub const BLOCKCHAIN: u32 = 0; @@ -54,30 +57,62 @@ pub mod source_types { /// New source types can be added by: /// 1. Adding a new variant here /// 2. Implementing the source struct -/// 3. Adding a case in DataSourceFactory +/// 3. Adding a case in `DataSourceFactory` #[derive(Debug)] pub enum DataSourceKind { /// Blockchain cross-chain events (sourceType=0) Blockchain(BlockchainEventSource), } +impl DataSourceKind { + pub(crate) async fn last_nonce(&self) -> Option { + match self { + Self::Blockchain(source) => source.last_nonce().await, + } + } + + pub(crate) async fn last_nonce_position(&self) -> Option { + match self { + Self::Blockchain(source) => source.last_nonce_block().await, + } + } + + pub(crate) async fn reconcile_progress(&self, nonce: u128, position: u64) { + match self { + Self::Blockchain(source) => source.reconcile_progress(nonce, position).await, + } + } + + pub(crate) fn cursor(&self) -> u64 { + match self { + Self::Blockchain(source) => source.cursor(), + } + } + + pub(crate) const fn source_id_u64(&self) -> u64 { + match self { + Self::Blockchain(source) => source.chain_id(), + } + } +} + #[async_trait] impl OracleDataSource for DataSourceKind { fn source_type(&self) -> u32 { match self { - DataSourceKind::Blockchain(_) => source_types::BLOCKCHAIN, + Self::Blockchain(_) => source_types::BLOCKCHAIN, } } fn source_id(&self) -> U256 { match self { - DataSourceKind::Blockchain(s) => s.source_id(), + Self::Blockchain(s) => s.source_id(), } } async fn poll(&self) -> Result> { match self { - DataSourceKind::Blockchain(s) => s.poll().await, + Self::Blockchain(s) => s.poll().await, } } } diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/oracle_manager.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/oracle_manager.rs index 3d95de9877..0894759622 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/src/oracle_manager.rs +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/oracle_manager.rs @@ -1,6 +1,6 @@ //! Oracle Relayer Manager //! -//! Manages oracle data sources keyed by URI, matching gaptos JWKObserver interface. +//! Manages oracle data sources keyed by their full task URI. use crate::{ blockchain_source::BlockchainEventSource, @@ -10,183 +10,281 @@ use crate::{ }; use anyhow::{anyhow, Result}; use std::{collections::HashMap, path::PathBuf, sync::Arc}; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, RwLock}; use tracing::{debug, info, warn}; -// Re-export types from gravity-api-types for external use pub use gravity_api_types::{on_chain_config::jwks::JWKStruct, relayer::PollResult}; -/// Startup scenario for determining initial source state -/// -/// When a data source is added, we need to determine where to start scanning. -/// This enum captures the 4 possible scenarios based on persisted and on-chain state. #[derive(Debug)] enum StartupScenario { - /// Persisted state exists but is stale - fast-forward to on-chain state - FastForward { onchain_nonce: u128, onchain_block: u64, persisted_nonce: u128 }, - /// Persisted state is valid - use it for fast restart - Restore { cursor: u64, nonce: u128 }, - /// No persisted state, but on-chain has data - sync from on-chain - ColdStartWithSync { onchain_nonce: u128, onchain_block: u64 }, - /// No persisted state, no on-chain data - start from config default - ColdStart { from_block: u64 }, + FastForward { + onchain_nonce: u128, + onchain_position: u64, + persisted_nonce: u128, + }, + RollbackToOnChain { + onchain_nonce: u128, + onchain_position: u64, + persisted_nonce: u128, + persisted_cursor: u64, + restart_cursor: u64, + }, + Restore { + cursor: u64, + nonce: u128, + position: u64, + }, + /// A legacy callback may have advanced `NativeOracle` nonce without storing + /// the source position. Rescan from a safe watermark while filtering all + /// observations at or below the authoritative on-chain nonce. + RecoverUnknownPosition { + cursor: u64, + onchain_nonce: u128, + persisted_nonce: Option, + }, + ColdStartWithSync { + onchain_nonce: u128, + onchain_position: u64, + }, + ColdStart { + from_block: u64, + }, } impl StartupScenario { - /// Determine which startup scenario applies based on persisted and on-chain state fn determine( persisted: Option<&SourceState>, onchain_nonce: u128, - onchain_block: u64, + onchain_position: u64, default_from_block: u64, - ) -> Self { - match persisted { - Some(state) if onchain_nonce > state.last_nonce as u128 => Self::FastForward { + ) -> Result { + if onchain_nonce == 0 && onchain_position != 0 { + return Err(anyhow!( + "Invalid NativeOracle progress: zero nonce has nonzero source position" + )); + } + + if onchain_nonce == 0 { + return Ok(match persisted { + Some(state) if state.last_nonce > 0 => Self::RollbackToOnChain { + onchain_nonce, + onchain_position, + persisted_nonce: state.last_nonce, + persisted_cursor: state.cursor_block, + restart_cursor: default_from_block, + }, + Some(state) => Self::Restore { cursor: state.cursor_block, nonce: 0, position: 0 }, + None => Self::ColdStart { from_block: default_from_block }, + }); + } + + if onchain_position == 0 { + return Ok(match persisted { + Some(state) if state.last_nonce == onchain_nonce => Self::Restore { + cursor: state.cursor_block, + nonce: onchain_nonce, + position: state.last_nonce_block, + }, + Some(state) if state.last_nonce < onchain_nonce => { + let cursor = if state.last_nonce > 0 && state.last_nonce_block > 0 { + state.last_nonce_block + } else { + default_from_block + }; + Self::RecoverUnknownPosition { + cursor, + onchain_nonce, + persisted_nonce: Some(state.last_nonce), + } + } + Some(state) => Self::RecoverUnknownPosition { + cursor: default_from_block, + onchain_nonce, + persisted_nonce: Some(state.last_nonce), + }, + None => Self::RecoverUnknownPosition { + cursor: default_from_block, + onchain_nonce, + persisted_nonce: None, + }, + }); + } + + Ok(match persisted { + Some(state) if onchain_nonce > state.last_nonce => Self::FastForward { onchain_nonce, - onchain_block, - persisted_nonce: state.last_nonce as u128, + onchain_position, + persisted_nonce: state.last_nonce, }, - Some(state) => { - Self::Restore { cursor: state.cursor_block, nonce: state.last_nonce as u128 } - } - None if onchain_nonce > 0 => Self::ColdStartWithSync { onchain_nonce, onchain_block }, - None => Self::ColdStart { from_block: default_from_block }, - } + Some(state) if state.last_nonce > onchain_nonce => Self::RollbackToOnChain { + onchain_nonce, + onchain_position, + persisted_nonce: state.last_nonce, + persisted_cursor: state.cursor_block, + restart_cursor: onchain_position, + }, + Some(state) => Self::Restore { + cursor: state.cursor_block, + nonce: state.last_nonce, + position: onchain_position, + }, + None => Self::ColdStartWithSync { onchain_nonce, onchain_position }, + }) } - /// Get (cursor, nonce) for source initialization - fn into_init_params(self) -> (u64, u128) { + const fn into_init_params(self) -> (u64, u128, u64) { match self { - Self::FastForward { onchain_nonce, onchain_block, .. } => { - (onchain_block, onchain_nonce) + Self::FastForward { onchain_nonce, onchain_position, .. } | + Self::ColdStartWithSync { onchain_nonce, onchain_position } => { + (onchain_position, onchain_nonce, onchain_position) + } + Self::RollbackToOnChain { onchain_nonce, onchain_position, restart_cursor, .. } => { + (restart_cursor, onchain_nonce, onchain_position) } - Self::Restore { cursor, nonce } => (cursor, nonce), - Self::ColdStartWithSync { onchain_nonce, onchain_block } => { - (onchain_block, onchain_nonce) + Self::Restore { cursor, nonce, position } => (cursor, nonce, position), + Self::RecoverUnknownPosition { cursor, onchain_nonce, .. } => { + (cursor, onchain_nonce, 0) } - Self::ColdStart { from_block } => (from_block, 0), + Self::ColdStart { from_block } => (from_block, 0, 0), } } - /// Log the startup scenario fn log(&self, uri: &str) { match self { - Self::FastForward { onchain_nonce, onchain_block, persisted_nonce } => { - warn!( - target: "oracle_manager", - uri, - persisted_nonce, - onchain_nonce, - onchain_block, - "Persisted state is stale, fast-forwarding to on-chain state" - ); - } - Self::Restore { cursor, nonce } => { - info!( - target: "oracle_manager", - uri, - persisted_nonce = nonce, - cursor_block = cursor, - "Using persisted state for fast restart" - ); - } - Self::ColdStartWithSync { onchain_nonce, onchain_block } => { - info!( - target: "oracle_manager", - uri, - onchain_nonce, - onchain_block, - "Cold start with on-chain state" - ); - } - Self::ColdStart { .. } => { - info!(target: "oracle_manager", uri, "Cold start from config (nonce=0)"); - } + Self::FastForward { onchain_nonce, onchain_position, persisted_nonce } => warn!( + target: "oracle_manager", + uri, + persisted_nonce, + onchain_nonce, + onchain_position, + "Persisted state is stale; fast-forwarding to confirmed on-chain progress" + ), + Self::RollbackToOnChain { + onchain_nonce, + onchain_position, + persisted_nonce, + persisted_cursor, + restart_cursor, + } => warn!( + target: "oracle_manager", + uri, + persisted_nonce, + persisted_cursor, + onchain_nonce, + onchain_position, + restart_cursor, + "Persisted state is ahead of NativeOracle; rolling back to confirmed progress" + ), + Self::Restore { cursor, nonce, position } => info!( + target: "oracle_manager", + uri, + persisted_nonce = nonce, + cursor_block = cursor, + source_position = position, + "Using persisted state for restart" + ), + Self::RecoverUnknownPosition { cursor, onchain_nonce, persisted_nonce } => warn!( + target: "oracle_manager", + uri, + cursor, + onchain_nonce, + ?persisted_nonce, + "NativeOracle source position is unknown; replaying from a safe watermark" + ), + Self::ColdStartWithSync { onchain_nonce, onchain_position } => info!( + target: "oracle_manager", + uri, + onchain_nonce, + onchain_position, + "Cold start from confirmed on-chain progress" + ), + Self::ColdStart { from_block } => info!( + target: "oracle_manager", + uri, + from_block, + "Cold start from task configuration" + ), } } } -/// Oracle Relayer Manager -/// -/// Manages data sources keyed by URI for per-observer polling. -/// Supports optional persistence for fast restart. #[derive(Debug)] +struct SourceEntry { + source: Arc, + /// JWK observers may call one provider concurrently. Serialize a URI's + /// cursor mutation so a scan range can only be emitted once locally. + poll_lock: Mutex<()>, +} + +#[derive(Debug)] +/// Owns configured relayer sources and their durable polling checkpoints. pub struct OracleRelayerManager { - /// Data sources keyed by URI - sources: RwLock>>, + sources: RwLock>>, datadir: PathBuf, - /// In-memory state for persistence state: RwLock, } impl OracleRelayerManager { - /// Create a new OracleRelayerManager with optional persistence - /// - /// # Arguments - /// * `datadir` - Optional path to data directory for state persistence + /// Opens a manager rooted at `datadir`, restoring persisted source checkpoints when present. pub fn new(datadir: PathBuf) -> Self { - let state = load_state_if_exists(&datadir).unwrap_or_else(RelayerState::new); - + let state = match load_state_if_exists(&datadir) { + Some(state) => state, + None => RelayerState::new(), + }; Self { sources: RwLock::new(HashMap::new()), datadir, state: RwLock::new(state) } } - /// Add a source by URI with on-chain state for warm-start - /// - /// If persistence is enabled and state exists for this URI, validates - /// the persisted state against on-chain state. If on-chain is ahead, - /// fast-forwards to on-chain state. Otherwise uses persisted state. - /// - /// # Arguments - /// * `uri` - The oracle task URI - /// * `rpc_url` - RPC endpoint URL - /// * `onchain_nonce` - Latest nonce from NativeOracle - /// * `onchain_block_number` - Block number where onchain_nonce was recorded + /// Registers a task URI and reconciles its local cursor with `NativeOracle` progress. pub async fn add_uri( &self, uri: &str, rpc_url: &str, onchain_nonce: u128, - onchain_block_number: u64, + onchain_position: u128, ) -> Result<()> { - { - let sources = self.sources.read().await; - if sources.contains_key(uri) { - info!(target: "oracle_manager", uri = uri, "Source already exists, skipping"); - return Ok(()); - } + if self.sources.read().await.contains_key(uri) { + info!(target: "oracle_manager", uri, "Source already exists; skipping"); + return Ok(()); } let task = parse_oracle_uri(uri)?; + let onchain_position = u64::try_from(onchain_position) + .map_err(|_| anyhow!("On-chain source position exceeds relayer u64 range"))?; - // Determine startup scenario based on persisted and on-chain state let scenario = { let state = self.state.read().await; + let persisted = matching_persisted_state(&state, uri, &task)?; StartupScenario::determine( - state.get(uri), + persisted, onchain_nonce, - onchain_block_number, + onchain_position, task.from_block(), - ) + )? }; scenario.log(uri); - let (start_cursor, start_nonce) = scenario.into_init_params(); + let (start_cursor, start_nonce, start_position) = scenario.into_init_params(); + + let source = self + .create_source_from_task(&task, rpc_url, start_nonce, start_position, start_cursor) + .await?; + let entry = Arc::new(SourceEntry { source: Arc::new(source), poll_lock: Mutex::new(()) }); - // Create source with reconciled state - let source = - self.create_source_from_task(&task, rpc_url, start_nonce, Some(start_cursor)).await?; + let mut sources = self.sources.write().await; + if sources.contains_key(uri) { + return Ok(()); + } + sources.insert(uri.to_string(), entry); info!( target: "oracle_manager", - uri = uri, + uri, source_type = task.source_type, source_id = task.source_id, - start_nonce = start_nonce, - start_cursor = start_cursor, + start_nonce, + start_cursor, + start_position, "Added data source" ); - - let mut sources = self.sources.write().await; - sources.insert(uri.to_string(), Arc::new(source)); Ok(()) } @@ -195,168 +293,253 @@ impl OracleRelayerManager { task: &ParsedOracleTask, rpc_url: &str, latest_onchain_nonce: u128, - persisted_cursor: Option, + latest_onchain_position: u64, + cursor: u64, ) -> Result { match task.source_type { source_types::BLOCKCHAIN => { - let portal_address = task.portal_address()?; - let config_start_block = task.from_block(); - let source = BlockchainEventSource::new_with_cursor( + let source = BlockchainEventSource::new_with_progress( task.source_id, rpc_url, - portal_address, - persisted_cursor.unwrap_or(config_start_block), + task.portal_address()?, + cursor, latest_onchain_nonce, + latest_onchain_position, ) .await?; - Ok(DataSourceKind::Blockchain(source)) } _ => Err(anyhow!("Unknown source type: {}", task.source_type)), } } - /// Poll a source by URI with optional on-chain state for reconciliation - /// - /// If on-chain state is provided and is ahead of local state, fast-forwards - /// local state before polling. After polling, persists the updated state. - /// - /// # Arguments - /// * `uri` - The oracle task URI to poll - /// * `onchain_nonce` - Optional current on-chain nonce for reconciliation - /// * `onchain_block_number` - Optional on-chain block for reconciliation + /// Polls one task URI and returns canonical payloads for JWK observation. pub async fn poll_uri( &self, uri: &str, onchain_nonce: Option, - onchain_block_number: Option, + onchain_position: Option, ) -> Result { - let sources = self.sources.read().await; - let source = sources.get(uri).ok_or_else(|| anyhow!("Source not found: {}", uri))?; - - // Reconcile with on-chain state before polling - if let (Some(onchain_nonce), Some(onchain_block)) = (onchain_nonce, onchain_block_number) { - match source.as_ref() { - DataSourceKind::Blockchain(s) => { - let current_nonce = s.last_nonce().await.unwrap_or(0); - if onchain_nonce > current_nonce { - info!( - target: "oracle_manager", - uri = uri, - local_nonce = current_nonce, - onchain_nonce = onchain_nonce, - onchain_block = onchain_block, - "On-chain ahead of local, fast-forwarding" - ); - s.fast_forward(onchain_nonce, onchain_block).await; - } - } + let entry = self + .sources + .read() + .await + .get(uri) + .cloned() + .ok_or_else(|| anyhow!("Source not found: {uri}"))?; + let _poll_guard = entry.poll_lock.lock().await; + let source = &entry.source; + + if let (Some(onchain_nonce), Some(onchain_position)) = (onchain_nonce, onchain_position) { + let onchain_position = u64::try_from(onchain_position) + .map_err(|_| anyhow!("On-chain source position exceeds relayer u64 range"))?; + let current_nonce = source.last_nonce().await.unwrap_or(0); + let current_position = source.last_nonce_position().await.unwrap_or(0); + if onchain_nonce > current_nonce || + (onchain_nonce == current_nonce && onchain_position > 0 && current_position == 0) + { + info!( + target: "oracle_manager", + uri, + current_nonce, + onchain_nonce, + onchain_position, + "Reconciling local source with confirmed on-chain progress" + ); + source.reconcile_progress(onchain_nonce, onchain_position).await; } } let data = source.poll().await?; + let nonce = source.last_nonce().await; + let last_nonce_position = source.last_nonce_position().await; + let cursor = source.cursor(); + let source_type = source.source_type(); + let source_id = source.source_id_u64(); - // Get nonce, cursor, and source info - let (nonce, last_nonce_block, max_block_number, source_type, source_id) = - match source.as_ref() { - DataSourceKind::Blockchain(s) => ( - s.last_nonce().await, - s.last_nonce_block().await, - s.cursor(), - source_types::BLOCKCHAIN, - s.chain_id(), - ), - }; - - let jwk_structs: Vec = data + let jwk_structs = data .iter() - .map(|d| JWKStruct { + .map(|data| JWKStruct { + // This becomes UnsupportedJWK.id during observation. The SDK + // execution adapter later restores the canonical Move type name. type_name: source.source_type().to_string(), - data: d.payload.to_vec(), + data: data.payload.to_vec(), }) - .collect(); - + .collect::>(); let updated = !data.is_empty(); debug!( target: "oracle_manager", - uri = uri, + uri, num_items = data.len(), - max_block = max_block_number, - nonce = ?nonce, - updated = updated, + cursor, + ?nonce, + updated, "Poll completed" ); - if let Some(n) = nonce { - self.update_and_save_state( - uri, - source_type, - source_id, - n, - last_nonce_block.unwrap_or(0), - max_block_number, - ) - .await; - } - - Ok(PollResult { jwk_structs, max_block_number, nonce, updated }) + // Persist empty scans too. Otherwise a source with no events repeats the + // same finalized range after every restart. + self.update_and_save_state( + uri, + source_type, + source_id, + nonce.unwrap_or(0), + last_nonce_position.unwrap_or(0), + cursor, + ) + .await; + + Ok(PollResult { jwk_structs, max_block_number: cursor, nonce, updated }) } - /// Update in-memory state and persist to disk. - /// - /// Writes to disk first (via a temporary clone) so that on the success - /// path the on-disk state is never behind in-memory state. If the disk - /// write fails, in-memory state is still advanced to avoid duplicate - /// delivery in the running process — only a subsequent crash would - /// replay events from the stale checkpoint. async fn update_and_save_state( &self, uri: &str, source_type: u32, source_id: u64, last_nonce: u128, - last_nonce_block: u64, - cursor_block: u64, + last_nonce_position: u64, + cursor: u64, ) { let mut state = self.state.write().await; - - // Build a candidate state with the update applied, without mutating - // the live state yet. let mut candidate = state.clone(); - candidate.update(uri, source_type, source_id, last_nonce, last_nonce_block, cursor_block); + candidate.update(uri, source_type, source_id, last_nonce, last_nonce_position, cursor); - let path = state_file_path(&self.datadir); - if let Err(e) = candidate.save(&path) { + if let Err(error) = candidate.save(&state_file_path(&self.datadir)) { warn!( target: "oracle_manager", - error = ?e, - path = ?path, - "Failed to persist relayer state; a crash may replay events from the last checkpoint" + ?error, + "Failed to persist relayer state; a crash may replay the last checkpoint" ); } - // Always commit to memory so the running process does not re-deliver. + // Keep the running process monotonic even if disk persistence failed. + // Startup reconciliation treats this checkpoint as unconfirmed until + // NativeOracle reports the same or a later nonce. *state = candidate; } - /// Remove a source by URI + /// Removes a configured task URI. pub async fn remove_uri(&self, uri: &str) -> Option> { - self.sources.write().await.remove(uri) + self.sources.write().await.remove(uri).map(|entry| entry.source.clone()) } - /// Get the number of registered sources + /// Returns the number of configured source URIs. pub async fn source_count(&self) -> usize { self.sources.read().await.len() } - /// Check if a source exists by URI + /// Returns whether a source URI is configured. pub async fn has_uri(&self, uri: &str) -> bool { self.sources.read().await.contains_key(uri) } - /// List all registered URIs + /// Lists configured source URIs. pub async fn list_uris(&self) -> Vec { self.sources.read().await.keys().cloned().collect() } } + +fn matching_persisted_state<'a>( + state: &'a RelayerState, + uri: &str, + task: &ParsedOracleTask, +) -> Result> { + let Some(persisted) = state.get(uri) else { + return Ok(None); + }; + if persisted.source_type != task.source_type || persisted.source_id != task.source_id { + return Err(anyhow!("Persisted oracle source identity does not match configured URI")); + } + Ok(Some(persisted)) +} + +#[cfg(test)] +mod tests { + use super::*; + + const URI: &str = + "gravity://0/1/events?portal=0x0000000000000000000000000000000000000001&fromBlock=100"; + + fn state(last_nonce: u128, last_position: u64, cursor: u64) -> RelayerState { + let mut state = RelayerState::new(); + state.update(URI, source_types::BLOCKCHAIN, 1, last_nonce, last_position, cursor); + state + } + + fn params(state: Option<&SourceState>, nonce: u128, position: u64) -> (u64, u128, u64) { + StartupScenario::determine(state, nonce, position, 100).unwrap().into_init_params() + } + + #[test] + fn known_position_fast_forwards_stale_persistence() { + let state = state(3, 130, 140); + assert_eq!(params(state.get(URI), 5, 160), (160, 5, 160)); + } + + #[test] + fn known_position_rolls_back_unconfirmed_local_data() { + let state = state(7, 180, 200); + assert_eq!(params(state.get(URI), 5, 160), (160, 5, 160)); + } + + #[test] + fn empty_onchain_state_rolls_back_to_task_start() { + let state = state(2, 120, 150); + assert_eq!(params(state.get(URI), 0, 0), (100, 0, 0)); + } + + #[test] + fn matching_checkpoint_preserves_scan_watermark() { + let state = state(5, 160, 220); + assert_eq!(params(state.get(URI), 5, 160), (220, 5, 160)); + } + + #[test] + fn unknown_position_without_persistence_replays_from_task_start() { + assert_eq!(params(None, 5, 0), (100, 5, 0)); + } + + #[test] + fn unknown_position_uses_matching_local_checkpoint() { + let state = state(5, 160, 220); + assert_eq!(params(state.get(URI), 5, 0), (220, 5, 160)); + } + + #[test] + fn unknown_position_rescans_from_last_locally_known_event_when_behind() { + let state = state(3, 130, 220); + assert_eq!(params(state.get(URI), 5, 0), (130, 5, 0)); + } + + #[test] + fn unknown_position_with_empty_or_ahead_state_uses_task_start() { + let empty = state(0, 0, 220); + assert_eq!(params(empty.get(URI), 5, 0), (100, 5, 0)); + + let ahead = state(7, 180, 220); + assert_eq!(params(ahead.get(URI), 5, 0), (100, 5, 0)); + } + + #[test] + fn rejects_inconsistent_zero_nonce_progress() { + let error = StartupScenario::determine(None, 0, 1, 100).unwrap_err(); + assert!(error.to_string().contains("zero nonce")); + } + + #[test] + fn rejects_persisted_identity_mismatch() { + let mut state = state(1, 110, 120); + state.sources.get_mut(URI).unwrap().source_id = 2; + let task = parse_oracle_uri(URI).unwrap(); + let error = matching_persisted_state(&state, URI, &task).unwrap_err(); + assert!(error.to_string().contains("identity")); + } + + #[test] + fn rejects_source_position_outside_runtime_range() { + let position = u128::from(u64::MAX) + 1; + assert!(u64::try_from(position).is_err()); + } +} diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/persistence.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/persistence.rs index f8a12e122d..be37547fb8 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/src/persistence.rs +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/persistence.rs @@ -26,7 +26,9 @@ pub struct SourceState { pub source_id: u64, /// Last nonce we fetched and returned pub last_nonce: u128, - /// Block number where last_nonce was emitted + /// Source-defined position associated with `last_nonce`. + /// + /// The field name is retained for backward-compatible JSON persistence. pub last_nonce_block: u64, /// Current polling cursor (block number) pub cursor_block: u64, @@ -51,11 +53,10 @@ impl RelayerState { /// Load state from a file pub fn load(path: &Path) -> Result { - let content = fs::read_to_string(path) - .with_context(|| format!("Failed to read state file: {}", path.display()))?; + let content = fs::read_to_string(path).context("Failed to read relayer state file")?; - let state: Self = serde_json::from_str(&content) - .with_context(|| format!("Failed to parse state file: {}", path.display()))?; + let state: Self = + serde_json::from_str(&content).context("Failed to parse relayer state file")?; if state.version != SCHEMA_VERSION { // Return a fresh state instead of loading incompatible data. @@ -68,7 +69,7 @@ impl RelayerState { return Ok(Self::new()); } - debug!("Loaded relayer state with {} sources from {}", state.sources.len(), path.display()); + debug!("Loaded relayer state with {} sources", state.sources.len()); Ok(state) } @@ -77,22 +78,18 @@ impl RelayerState { pub fn save(&self, path: &Path) -> Result<()> { // Ensure parent directory exists if let Some(parent) = path.parent() { - fs::create_dir_all(parent) - .with_context(|| format!("Failed to create directory: {}", parent.display()))?; + fs::create_dir_all(parent).context("Failed to create relayer state directory")?; } let content = serde_json::to_string_pretty(self).context("Failed to serialize state")?; // Write to temp file first, then rename for atomicity let temp_path = path.with_extension("json.tmp"); - fs::write(&temp_path, &content) - .with_context(|| format!("Failed to write temp file: {}", temp_path.display()))?; + fs::write(&temp_path, &content).context("Failed to write temporary relayer state file")?; - fs::rename(&temp_path, path).with_context(|| { - format!("Failed to rename {} to {}", temp_path.display(), path.display()) - })?; + fs::rename(&temp_path, path).context("Failed to commit relayer state file")?; - debug!("Saved relayer state to {}", path.display()); + debug!("Saved relayer state"); Ok(()) } @@ -135,7 +132,7 @@ pub fn state_file_path(datadir: &Path) -> PathBuf { pub fn load_state_if_exists(datadir: &Path) -> Option { let path = state_file_path(datadir); if !path.exists() { - info!("No existing relayer state at {}", path.display()); + info!("No existing relayer state"); return None; }