From 8d602d151b4fe0bf79e5bb56e051f1da7d75dee6 Mon Sep 17 00:00:00 2001 From: Miguel de Elias Date: Mon, 5 Oct 2026 09:32:37 -0300 Subject: [PATCH 1/2] feat: support multiple signers per oracle instance ORACLE_SIGNING_KEY and ORACLE_INDEX now accept comma separated lists, paired by position, so a single instance can vote on the SubgraphAvailabilityManager with several registered oracle addresses. A single key and index behave as before. - Static config errors (unparsable keys or indexes, count mismatch, duplicates, invalid contract mode) fail at startup before any transaction - Each signer posts its own config to DataEdge; a failing signer is logged and doesn't stop the others - Signers vote concurrently; a run fails only if every signer fails - A voteMany revert during gas estimation is reported as an error instead of being counted as success - Signers share one RPC provider and their logs carry oracle_index - New signer_votes_total and signer_vote_failures_total metrics labeled by oracle_index; denied_subgraphs_total still counts each change once - RewardsManager mode rejects more than one signing key --- README.md | 19 +- availability-oracle/src/contract.rs | 317 +++++++++++++++++---- availability-oracle/src/data_edge.rs | 39 +-- availability-oracle/src/main.rs | 399 ++++++++++++++++++++++----- crates/common/src/prelude.rs | 2 +- 5 files changed, 625 insertions(+), 151 deletions(-) diff --git a/README.md b/README.md index 885f5a8..6ac5517 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ OPTIONS: Minimum signal for a subgraph to be checked [env: ORACLE_MIN_SIGNAL=] [default: 100] --oracle-index - Assigned index for the oracle, to be used when voting on SubgraphAvailabilityManager [env: ORACLE_INDEX=] + Comma separated oracle indexes, to be used when voting on SubgraphAvailabilityManager. Paired by position with --signing-key [env: ORACLE_INDEX=] --period How often the oracle should check the subgraphs. With the default value of 0, the oracle will run once and terminate [env: ORACLE_PERIOD_SECS=] [default: 0] @@ -48,7 +48,7 @@ OPTIONS: The address of the rewards manager contract [env: REWARDS_MANAGER_CONTRACT=] --signing-key - The secret key of the oracle for signing transactions [env: ORACLE_SIGNING_KEY=] + Comma separated secret keys of the oracle signers, paired by position with --oracle-index [env: ORACLE_SIGNING_KEY=] --subgraph Graphql endpoint to the network subgraph [env: ORACLE_SUBGRAPH=] @@ -92,6 +92,21 @@ cargo run -p availability-oracle -- \ --graph-monitoring-subgraph ``` +### Running with multiple signers + +An oracle can vote on `SubgraphAvailabilityManager` with several signers from a single instance. Each signer must be registered in the contract at its own oracle index. Pass the keys and indexes as comma separated lists, paired by position: + +``` +ORACLE_SIGNING_KEY=, +ORACLE_INDEX=, +``` + +Spaces around commas are ignored. Duplicate keys or indexes are rejected at startup, and `--dry-run` validates them when keys are given. + +Every signer posts its own configuration to the DataEdge contract and submits the same votes, concurrently. Configuration errors fail at startup, before any transaction is sent. Runtime failures of a single signer (DataEdge post, reverted vote, RPC, insufficient funds) are logged with its `oracle_index` and do not stop the other signers; a run fails only if every signer fails. `RewardsManager` mode supports a single signer. + +Per-signer metrics: `signer_votes_total{oracle_index}` and `signer_vote_failures_total{oracle_index}`. `denied_subgraphs_total` counts each change once, regardless of the number of signers. + ### Example command to run `RewardsManager` configuration: ``` diff --git a/availability-oracle/src/contract.rs b/availability-oracle/src/contract.rs index 4f1dfe0..7082c5f 100644 --- a/availability-oracle/src/contract.rs +++ b/availability-oracle/src/contract.rs @@ -57,36 +57,43 @@ impl RewardsManagerContract { pub struct SubgraphAvailabilityManagerContract { contract: SubgraphAvailabilityManagerABI, LocalWallet>>, oracle_index: u64, - logger: Logger, +} + +pub fn http_provider(url: Url) -> Provider { + let http_client = reqwest::ClientBuilder::new() + .tcp_nodelay(true) + .timeout(Duration::from_secs(30)) + .build() + .unwrap(); + Provider::new(Http::new_with_client(url, http_client)) +} + +pub fn signer_middleware( + provider: Provider, + signing_key: &SecretKey, + chain_id: u64, +) -> Arc, LocalWallet>> { + let wallet = LocalWallet::from_bytes(signing_key.as_ref()) + .unwrap() + .with_chain_id(chain_id); + Arc::new(SignerMiddleware::new(provider, wallet)) } impl SubgraphAvailabilityManagerContract { - pub async fn new( + pub fn new( + provider: Provider, + chain_id: u64, signing_key: &SecretKey, - url: Url, subgraph_availability_manager_contract: Address, oracle_index: u64, - logger: Logger, ) -> Self { - let http_client = reqwest::ClientBuilder::new() - .tcp_nodelay(true) - .timeout(Duration::from_secs(10)) - .build() - .unwrap(); - let provider = Provider::new(Http::new_with_client(url, http_client)); - let chain_id = provider.get_chainid().await.unwrap().as_u64(); - let wallet = LocalWallet::from_bytes(signing_key.as_ref()) - .unwrap() - .with_chain_id(chain_id); - let provider = Arc::new(SignerMiddleware::new(provider, wallet)); let contract = SubgraphAvailabilityManagerABI::new( subgraph_availability_manager_contract, - provider.clone(), + signer_middleware(provider, signing_key, chain_id), ); Self { contract, oracle_index, - logger, } } } @@ -138,52 +145,134 @@ impl StateManager for RewardsManagerContract { } #[async_trait] -impl StateManager for SubgraphAvailabilityManagerContract { - async fn deny_many(&self, denied_status: Vec<([u8; 32], bool)>) -> Result<(), Error> { +pub trait OracleSigner { + fn oracle_index(&self) -> u64; + // Number of subgraphs confirmed on chain, and the error that stopped voting, if any. + async fn vote_many(&self, denied_status: &[([u8; 32], bool)]) -> (u64, Result<(), Error>); +} + +#[async_trait] +impl OracleSigner for SubgraphAvailabilityManagerContract { + fn oracle_index(&self) -> u64 { + self.oracle_index + } + + async fn vote_many(&self, denied_status: &[([u8; 32], bool)]) -> (u64, Result<(), Error>) { + let mut confirmed = 0; // 100 is considered as a good chunk size. for chunk in denied_status.chunks(100) { - let ids: Vec<[u8; 32usize]> = chunk.iter().map(|s| s.0).collect(); - let statuses: Vec = chunk.iter().map(|s| s.1).collect(); - let num_subgraphs = ids.len() as u64; - let oracle_index = U256::from(self.oracle_index); - let tx = self.contract.vote_many(ids, statuses, oracle_index); + if let Err(e) = self.vote_chunk(chunk).await { + return (confirmed, Err(e)); + } + confirmed += chunk.len() as u64; + } + (confirmed, Ok(())) + } +} - // Calculate estimated gas - let estimated_gas_tx = tx.estimate_gas().await; +impl SubgraphAvailabilityManagerContract { + async fn vote_chunk(&self, chunk: &[([u8; 32], bool)]) -> Result<(), Error> { + let ids: Vec<[u8; 32usize]> = chunk.iter().map(|s| s.0).collect(); + let statuses: Vec = chunk.iter().map(|s| s.1).collect(); + let oracle_index = U256::from(self.oracle_index); + let tx = self.contract.vote_many(ids, statuses, oracle_index); - let estimated_gas = match estimated_gas_tx { - Ok(estimate) => estimate, - Err(err) => { - let message = err.decode_revert::().unwrap_or(err.to_string()); - error!(self.logger, "Transaction failed"; - "message" => message, - ); - // Return `Ok()` to avoid double error logging - return Ok(()); - } - }; + // Calculate estimated gas + let estimated_gas_tx = tx.estimate_gas().await; - // Increase the estimated gas by 20% - let increased_estimate = estimated_gas * U256::from(120) / U256::from(100); + let estimated_gas = match estimated_gas_tx { + Ok(estimate) => estimate, + Err(err) => { + let message = err.decode_revert::().unwrap_or(err.to_string()); + return Err(anyhow!("voteMany reverted: {}", message)); + } + }; - // Set a legacy gas price explicitly. Left unset, ethers fills the - // EIP-1559 fields from its default estimator, which hardcodes a - // 3 gwei priority fee - roughly 150x the Arbitrum base fee. - let gas_price = self.contract.client().get_gas_price().await?; - let gas_price_with_buffer = gas_price * U256::from(120) / U256::from(100); + // Increase the estimated gas by 20% + let increased_estimate = estimated_gas * U256::from(120) / U256::from(100); - tx.gas(increased_estimate) - .gas_price(gas_price_with_buffer) - .send() - .await? - .await?; - METRICS.denied_subgraphs_total.inc_by(num_subgraphs); - } + // Set a legacy gas price explicitly. Left unset, ethers fills the + // EIP-1559 fields from its default estimator, which hardcodes a + // 3 gwei priority fee - roughly 150x the Arbitrum base fee. + let gas_price = self.contract.client().get_gas_price().await?; + let gas_price_with_buffer = gas_price * U256::from(120) / U256::from(100); + + tx.gas(increased_estimate) + .gas_price(gas_price_with_buffer) + .send() + .await? + .await?; Ok(()) } } +/// Submits the same deny status through every signer concurrently. Succeeds if +/// at least one signer succeeds; failures of the others are logged. +pub struct MultiStateManager { + signers: Vec>, + logger: Logger, +} + +impl MultiStateManager { + pub fn new(signers: Vec>, logger: Logger) -> Self { + Self { signers, logger } + } +} + +#[async_trait] +impl StateManager for MultiStateManager { + async fn deny_many(&self, denied_status: Vec<([u8; 32], bool)>) -> Result<(), Error> { + let results = futures::future::join_all( + self.signers + .iter() + .map(|signer| signer.vote_many(&denied_status)), + ) + .await; + + let confirmed = results.iter().map(|(n, _)| *n).max().unwrap_or(0); + METRICS.denied_subgraphs_total.inc_by(confirmed); + + let mut errors = Vec::new(); + for (signer, (n, result)) in self.signers.iter().zip(results) { + let index = signer.oracle_index().to_string(); + METRICS + .signer_votes_total + .with_label_values(&[&index]) + .inc_by(n); + if let Err(e) = result { + METRICS + .signer_vote_failures_total + .with_label_values(&[&index]) + .inc(); + errors.push((signer.oracle_index(), e)); + } + } + + if errors.len() < self.signers.len() { + for (oracle_index, e) in errors { + error!(self.logger, "Signer failed to submit deny status"; + "oracle_index" => oracle_index, + "error" => format!("{:#}", e) + ); + } + return Ok(()); + } + if errors.len() == 1 { + return Err(errors.pop().unwrap().1); + } + let causes: Vec = errors + .iter() + .map(|(oracle_index, e)| format!("oracle_index {}: {:#}", oracle_index, e)) + .collect(); + Err(anyhow!( + "all {} signers failed: {}", + errors.len(), + causes.join("; ") + )) + } +} + pub struct StateManagerDryRun { logger: Logger, } @@ -209,6 +298,8 @@ impl StateManager for StateManagerDryRun { struct Metrics { denied_subgraphs_total: prometheus::IntCounter, + signer_votes_total: prometheus::IntCounterVec, + signer_vote_failures_total: prometheus::IntCounterVec, } lazy_static! { @@ -223,6 +314,130 @@ impl Metrics { "Total denied subgraphs" ) .unwrap(), + signer_votes_total: prometheus::register_int_counter_vec!( + "signer_votes_total", + "Subgraph statuses confirmed on chain per signer", + &["oracle_index"] + ) + .unwrap(), + signer_vote_failures_total: prometheus::register_int_counter_vec!( + "signer_vote_failures_total", + "Failed deny status submissions per signer", + &["oracle_index"] + ) + .unwrap(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + struct MockSigner { + index: u64, + fail: bool, + barrier: Option>, + calls: Arc>>>, + } + + #[async_trait] + impl OracleSigner for MockSigner { + fn oracle_index(&self) -> u64 { + self.index + } + + async fn vote_many(&self, denied_status: &[([u8; 32], bool)]) -> (u64, Result<(), Error>) { + if let Some(barrier) = &self.barrier { + barrier.wait().await; + } + self.calls.lock().unwrap().push(denied_status.to_vec()); + if self.fail { + return (0, Err(anyhow!("mock signer {} failure", self.index))); + } + (denied_status.len() as u64, Ok(())) + } + } + + fn signers( + fail: &[bool], + barrier: Option>, + ) -> ( + MultiStateManager, + Vec>>>>, + ) { + let calls: Vec<_> = fail + .iter() + .map(|_| Arc::new(Mutex::new(Vec::new()))) + .collect(); + let signers = fail + .iter() + .zip(&calls) + .enumerate() + .map(|(index, (fail, calls))| { + Box::new(MockSigner { + index: index as u64, + fail: *fail, + barrier: barrier.clone(), + calls: calls.clone(), + }) as Box + }) + .collect(); + ( + MultiStateManager::new(signers, common::logging::create_logger()), + calls, + ) + } + + #[tokio::test] + async fn every_signer_submits_the_same_status() { + let (multi, calls) = signers(&[false, false], None); + let status = vec![([1u8; 32], true), ([2u8; 32], false)]; + + multi.deny_many(status.clone()).await.unwrap(); + + for calls in calls { + assert_eq!(*calls.lock().unwrap(), vec![status.clone()]); } } + + #[tokio::test] + async fn partial_failure_is_ok_and_others_still_vote() { + let (multi, calls) = signers(&[true, false], None); + let status = vec![([1u8; 32], true)]; + multi.deny_many(status.clone()).await.unwrap(); + assert_eq!(calls[0].lock().unwrap().len(), 1); + assert_eq!(*calls[1].lock().unwrap(), vec![status]); + } + + #[tokio::test] + async fn single_signer_error_passes_through_unchanged() { + let (multi, _) = signers(&[true], None); + let err = multi.deny_many(vec![([1u8; 32], true)]).await.unwrap_err(); + assert_eq!(format!("{:#}", err), "mock signer 0 failure"); + } + + #[tokio::test] + async fn all_signers_failing_reports_every_cause() { + let (multi, _) = signers(&[true, true], None); + let err = multi.deny_many(vec![([1u8; 32], true)]).await.unwrap_err(); + let message = format!("{:#}", err); + assert!(message.contains("all 2 signers failed")); + assert!(message.contains("oracle_index 0: mock signer 0 failure")); + assert!(message.contains("oracle_index 1: mock signer 1 failure")); + } + + #[tokio::test] + async fn signers_vote_concurrently() { + let barrier = Arc::new(tokio::sync::Barrier::new(2)); + let (multi, _) = signers(&[false, false], Some(barrier)); + tokio::time::timeout( + Duration::from_secs(1), + multi.deny_many(vec![([1u8; 32], true)]), + ) + .await + .expect("signers ran sequentially") + .unwrap(); + } } diff --git a/availability-oracle/src/data_edge.rs b/availability-oracle/src/data_edge.rs index b80d39b..99193bc 100644 --- a/availability-oracle/src/data_edge.rs +++ b/availability-oracle/src/data_edge.rs @@ -16,7 +16,7 @@ pub enum ConfigStatus { use ethers::core::types::U256; use ethers::middleware::SignerMiddleware; use ethers::providers::{Http, Middleware, Provider}; -use ethers::signers::{LocalWallet, Signer}; +use ethers::signers::LocalWallet; use ethers::types::TransactionRequest; use secp256k1::SecretKey; use std::sync::Arc; @@ -45,6 +45,7 @@ pub fn extract_deployment_id_from_url(url: &str) -> Result { } /// Configuration needed to build an OracleConfig from CLI parameters. +#[derive(Clone, Copy)] pub struct OracleConfigParams<'a> { pub ipfs_concurrency: usize, pub ipfs_timeout: Duration, @@ -113,29 +114,19 @@ pub struct DataEdgeContract { } impl DataEdgeContract { - pub async fn new( + pub fn new( + provider: Provider, + chain_id: u64, signing_key: &SecretKey, - rpc_url: Url, contract_address: Address, logger: Logger, - ) -> Result { - let http_client = reqwest::ClientBuilder::new() - .tcp_nodelay(true) - .timeout(Duration::from_secs(30)) - .build() - .unwrap(); - let provider = Provider::new(Http::new_with_client(rpc_url, http_client)); - let chain_id = provider.get_chainid().await?.as_u64(); - let wallet = LocalWallet::from_bytes(signing_key.as_ref()) - .unwrap() - .with_chain_id(chain_id); - let provider = Arc::new(SignerMiddleware::new(provider, wallet)); - - Ok(Self { + ) -> Self { + let provider = crate::contract::signer_middleware(provider, signing_key, chain_id); + Self { provider, contract_address, logger, - }) + } } /// Posts the oracle configuration to the DataEdge contract if it has changed. @@ -148,25 +139,22 @@ impl DataEdgeContract { ) -> Result { match check_config_status(local_config, monitoring_subgraph, oracle_index).await { ConfigStatus::Unchanged => { - info!(self.logger, "Config unchanged, skipping DataEdge post"; - "oracle_index" => oracle_index - ); + info!(self.logger, "Config unchanged, skipping DataEdge post"); return Ok(false); } ConfigStatus::Changed(changed_fields) => { info!(self.logger, "Config changed, will post to DataEdge"; - "oracle_index" => oracle_index, "changed_fields" => changed_fields.join(",") ); } ConfigStatus::NotFound => { - info!(self.logger, "Oracle not found in subgraph, posting initial config"; - "oracle_index" => oracle_index + info!( + self.logger, + "Oracle not found in subgraph, posting initial config" ); } ConfigStatus::FetchError(e) => { warn!(self.logger, "Failed to fetch current oracle config from subgraph, will post anyway"; - "oracle_index" => oracle_index, "error" => format!("{:#}", e) ); } @@ -265,6 +253,7 @@ pub async fn log_dry_run_config( } ConfigStatus::FetchError(e) => { warn!(logger, "Failed to fetch current config (dry-run)"; + "oracle_index" => oracle_index, "error" => format!("{:#}", e) ); } diff --git a/availability-oracle/src/main.rs b/availability-oracle/src/main.rs index 6c46e71..cc618bf 100644 --- a/availability-oracle/src/main.rs +++ b/availability-oracle/src/main.rs @@ -14,6 +14,7 @@ use contract::*; use data_edge::{build_oracle_config, log_dry_run_config, DataEdgeContract, OracleConfigParams}; use epoch_block_oracle_subgraph::{EpochBlockOracleSubgraph, EpochBlockOracleSubgraphImpl}; use ethers::abi::Address; +use ethers::providers::{Http, Middleware, Provider}; use ethers::signers::LocalWallet; use ethers::signers::Signer; use graph_monitoring_subgraph::GraphMonitoringSubgraphImpl; @@ -98,7 +99,8 @@ struct Config { long, env = "ORACLE_SIGNING_KEY", required_unless("dry-run"), - help = "The secret key of the oracle for signing transactions" + hide_env_values = true, + help = "Comma separated secret keys of the oracle signers, paired by position with --oracle-index" )] signing_key: Option, @@ -148,9 +150,10 @@ struct Config { #[structopt( long, env = "ORACLE_INDEX", - help = "Assigned index for the oracle, to be used when voting on SubgraphAvailabilityManager" + help = "Comma separated oracle indexes, to be used when voting on SubgraphAvailabilityManager. \ + Paired by position with --signing-key" )] - pub oracle_index: Option, + pub oracle_index: Option, #[structopt( long, @@ -187,19 +190,30 @@ async fn run(logger: Logger, config: Config) -> Result<()> { network_subgraph_url: &config.subgraph, epoch_block_oracle_subgraph_url: &config.epoch_block_oracle_subgraph, subgraph_availability_manager_contract: config.subgraph_availability_manager_contract, - oracle_index: config.oracle_index, + oracle_index: None, }; - let signing_key: Option = if config.dry_run { + let signing_keys = split_list(&config.signing_key); + let oracle_indexes = parse_indexes(&config.oracle_index)?; + let signers: Vec<(SecretKey, u64)> = if config.dry_run && signing_keys.is_empty() { + Vec::new() + } else { + parse_signers(&signing_keys, &oracle_indexes)? + }; + if !config.dry_run { + validate_mode( + signers.len(), + config.rewards_manager_contract, + config.subgraph_availability_manager_contract, + )?; + } + + let rpc = if config.dry_run { None } else { - Some( - config - .signing_key - .as_ref() - .expect("signing_key is required unless dry-run") - .parse()?, - ) + let provider = http_provider(config.url.clone()); + let chain_id = provider.get_chainid().await?.as_u64(); + Some((provider, chain_id)) }; if config.dry_run { @@ -208,30 +222,30 @@ async fn run(logger: Logger, config: Config) -> Result<()> { "Running in dry mode: no transactions will be submitted on chain!" ); // In dry-run mode, build local config and check against subgraph if available - if let Ok(local_config) = build_oracle_config(&config_params) { - let monitoring_subgraph = config - .graph_monitoring_subgraph - .as_ref() - .map(|endpoint| GraphMonitoringSubgraphImpl::new(endpoint.clone())); - log_dry_run_config( - &logger, - &local_config, - monitoring_subgraph.as_ref(), - config.oracle_index, - ) - .await; + let monitoring_subgraph = config + .graph_monitoring_subgraph + .as_ref() + .map(|endpoint| GraphMonitoringSubgraphImpl::new(endpoint.clone())); + let dry_run_indexes: Vec> = if oracle_indexes.is_empty() { + vec![None] + } else { + oracle_indexes.iter().copied().map(Some).collect() + }; + for oracle_index in dry_run_indexes { + if let Ok(local_config) = build_oracle_config(&OracleConfigParams { + oracle_index, + ..config_params + }) { + log_dry_run_config( + &logger, + &local_config, + monitoring_subgraph.as_ref(), + oracle_index, + ) + .await; + } } } else { - let signing_key = signing_key.as_ref().unwrap(); - let wallet = LocalWallet::from_bytes(signing_key.as_ref()).unwrap(); - info!(logger, "Signing account {}", wallet.address().to_string()); - - // Build local config and post to DataEdge if changed - let local_config = build_oracle_config(&config_params)?; - let oracle_index = config - .oracle_index - .ok_or_else(|| anyhow!("oracle_index is required for DataEdge posting"))?; - let monitoring_subgraph = GraphMonitoringSubgraphImpl::new( config .graph_monitoring_subgraph @@ -239,20 +253,45 @@ async fn run(logger: Logger, config: Config) -> Result<()> { .expect("graph_monitoring_subgraph is required unless dry-run") .clone(), ); + let data_edge_contract = config + .data_edge_contract + .expect("data_edge_contract is required unless dry-run"); - let data_edge = DataEdgeContract::new( - &signing_key, - config.url.clone(), - config - .data_edge_contract - .expect("data_edge_contract is required unless dry-run"), - logger.clone(), - ) - .await?; + let (provider, chain_id) = rpc.clone().expect("rpc is set unless dry-run"); - data_edge - .post_config_if_changed(&local_config, &monitoring_subgraph, oracle_index) - .await?; + // Each signer posts its own config, the monitoring subgraph only accepts + // a config from the address registered for its oracle index + for (signing_key, oracle_index) in &signers { + let signer_logger = logger.new(o!("oracle_index" => *oracle_index)); + let wallet = LocalWallet::from_bytes(signing_key.as_ref()).unwrap(); + info!( + signer_logger, + "Signing account {}", + wallet.address().to_string() + ); + + let local_config = build_oracle_config(&OracleConfigParams { + oracle_index: Some(*oracle_index), + ..config_params + })?; + + let data_edge = DataEdgeContract::new( + provider.clone(), + chain_id, + signing_key, + data_edge_contract, + signer_logger.clone(), + ); + + if let Err(e) = data_edge + .post_config_if_changed(&local_config, &monitoring_subgraph, *oracle_index) + .await + { + error!(signer_logger, "Failed to post oracle config to DataEdge"; + "error" => format!("{:#}", e) + ); + } + } } let ipfs = IpfsImpl::new( @@ -265,19 +304,19 @@ async fn run(logger: Logger, config: Config) -> Result<()> { logger.clone(), config.epoch_block_oracle_subgraph.clone(), ); - let contract: Box = if config.dry_run { - Box::new(StateManagerDryRun::new(logger.clone())) - } else { + let contract: Box = if let Some((provider, chain_id)) = rpc { state_manager( + provider, + chain_id, config.url, - signing_key.as_ref().unwrap(), + &signers, config.rewards_manager_contract, config.subgraph_availability_manager_contract, - config.oracle_index, logger.clone(), ) - .await - .expect("Configuration error: either [`REWARDS_MANAGER_CONTRACT`] or [`SUBGRAPH_AVAILABILITY_MANAGER_CONTRACT` and `ORACLE_INDEX`] must be provided.") + .await? + } else { + Box::new(StateManagerDryRun::new(logger.clone())) }; let grace_period = Duration::from_secs(config.grace_period); @@ -354,38 +393,113 @@ async fn run(logger: Logger, config: Config) -> Result<()> { } } +fn split_list(value: &Option) -> Vec { + match value.as_deref().map(str::trim) { + None | Some("") => Vec::new(), + Some(value) => value + .split(',') + .map(|part| part.trim().to_string()) + .collect(), + } +} + +fn parse_indexes(value: &Option) -> Result> { + split_list(value) + .iter() + .enumerate() + .map(|(position, index)| { + u64::from_str(index) + .with_context(|| format!("invalid oracle index at position {}", position)) + }) + .collect() +} + +fn parse_signers(signing_keys: &[String], oracle_indexes: &[u64]) -> Result> { + ensure!( + !signing_keys.is_empty(), + "at least one signing key is required unless dry-run" + ); + ensure!( + signing_keys.len() == oracle_indexes.len(), + "expected one oracle index per signing key, got {} signing keys and {} oracle indexes", + signing_keys.len(), + oracle_indexes.len() + ); + let mut signers: Vec<(SecretKey, u64)> = Vec::new(); + for (position, (signing_key, oracle_index)) in + signing_keys.iter().zip(oracle_indexes).enumerate() + { + let signing_key: SecretKey = signing_key + .trim() + .parse() + .with_context(|| format!("invalid signing key at position {}", position))?; + ensure!( + !signers.iter().any(|(_, i)| i == oracle_index), + "duplicate oracle index {}", + oracle_index + ); + ensure!( + !signers.iter().any(|(k, _)| *k == signing_key), + "duplicate signing key at position {}", + position + ); + signers.push((signing_key, *oracle_index)); + } + Ok(signers) +} + +fn validate_mode( + signers: usize, + rewards_manager_contract: Option
, + subgraph_availability_manager_contract: Option
, +) -> Result<()> { + if subgraph_availability_manager_contract.is_some() { + return Ok(()); + } + ensure!( + rewards_manager_contract.is_some(), + "Configuration error: either `REWARDS_MANAGER_CONTRACT` or `SUBGRAPH_AVAILABILITY_MANAGER_CONTRACT` must be provided" + ); + ensure!( + signers == 1, + "RewardsManager mode supports a single signing key, got {}", + signers + ); + Ok(()) +} + // This function is used to create a state manager based on the configuration. -// If subgraph_availability_manager_contract and oracle_index are provided, it will create a SubgraphAvailabilityManagerContract. -// If rewards_manager_contract is provided, it will create a RewardsManagerContract. -// If none of the above are provided, it will return None. +// If subgraph_availability_manager_contract is provided, it will create a SubgraphAvailabilityManagerContract per signer. +// If rewards_manager_contract is provided, it will create a RewardsManagerContract, which supports a single signer. async fn state_manager( + provider: Provider, + chain_id: u64, rpc_url: Url, - signing_key: &SecretKey, + signers: &[(SecretKey, u64)], rewards_manager_contract: Option
, subgraph_availability_manager_contract: Option
, - oracle_index: Option, logger: Logger, -) -> Option> { +) -> Result> { if let Some(contract_address) = subgraph_availability_manager_contract { - if let Some(oracle_index) = oracle_index { + let mut signers_out: Vec> = Vec::new(); + for (signing_key, oracle_index) in signers { let contract = SubgraphAvailabilityManagerContract::new( + provider.clone(), + chain_id, signing_key, - rpc_url, contract_address, - oracle_index, - logger.clone(), - ) - .await; - return Some(Box::new(contract)); + *oracle_index, + ); + signers_out.push(Box::new(contract)); } - } else if let Some(contract_address) = rewards_manager_contract { - let contract = - RewardsManagerContract::new(signing_key, rpc_url, contract_address, logger.clone()) - .await; - return Some(Box::new(contract)); + return Ok(Box::new(MultiStateManager::new(signers_out, logger))); } - None + let contract_address = rewards_manager_contract.expect("validated by validate_mode"); + let (signing_key, _) = &signers[0]; + let contract = + RewardsManagerContract::new(signing_key, rpc_url, contract_address, logger.clone()).await; + Ok(Box::new(contract)) } /// Does the thing that the availablitiy oracle does, namely: @@ -724,3 +838,144 @@ impl Metrics { } } } + +#[cfg(test)] +mod signer_tests { + use super::*; + + const KEY_A: &str = "ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80"; + const KEY_B: &str = "59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d"; + + fn cli(args: &[&str]) -> Config { + let base = [ + "availability-oracle", + "--ipfs", + "http://ipfs", + "--subgraph", + "http://subgraph", + "--epoch-block-oracle-subgraph", + "http://ebo", + "--url", + "http://rpc", + "--data-edge-contract", + "0x0000000000000000000000000000000000000001", + "--graph-monitoring-subgraph", + "http://monitoring", + ]; + Config::from_iter_safe(base.iter().chain(args.iter())).unwrap() + } + + #[test] + fn single_signer_config_is_unchanged() { + let config = cli(&["--signing-key", KEY_A, "--oracle-index", "2"]); + let keys = split_list(&config.signing_key); + let indexes = parse_indexes(&config.oracle_index).unwrap(); + assert_eq!(keys, vec![KEY_A.to_string()]); + assert_eq!(indexes, vec![2]); + + let signers = parse_signers(&keys, &indexes).unwrap(); + assert_eq!(signers.len(), 1); + assert_eq!(signers[0].1, 2); + } + + #[test] + fn multiple_signers_are_paired_by_position() { + let keys = format!("{},{}", KEY_A, KEY_B); + let config = cli(&["--signing-key", &keys, "--oracle-index", "0,3"]); + + let signers = parse_signers( + &split_list(&config.signing_key), + &parse_indexes(&config.oracle_index).unwrap(), + ) + .unwrap(); + assert_eq!(signers.len(), 2); + assert_eq!(signers[0], (KEY_A.parse().unwrap(), 0)); + assert_eq!(signers[1], (KEY_B.parse().unwrap(), 3)); + } + + #[test] + fn cli_flags_override_env() { + std::env::set_var("ORACLE_SIGNING_KEY", format!("{},{}", KEY_B, KEY_B)); + std::env::set_var("ORACLE_INDEX", "7,8"); + let config = cli(&["--signing-key", KEY_A, "--oracle-index", "2"]); + std::env::remove_var("ORACLE_SIGNING_KEY"); + std::env::remove_var("ORACLE_INDEX"); + + assert_eq!(split_list(&config.signing_key), vec![KEY_A.to_string()]); + assert_eq!(parse_indexes(&config.oracle_index).unwrap(), vec![2]); + } + + #[test] + fn mismatched_signers_and_indexes_fail() { + let keys = vec![KEY_A.to_string(), KEY_B.to_string()]; + let err = parse_signers(&keys, &[0]).unwrap_err(); + assert!(err + .to_string() + .contains("2 signing keys and 1 oracle indexes")); + + let err = parse_signers(&keys[..1], &[]).unwrap_err(); + assert!(err + .to_string() + .contains("1 signing keys and 0 oracle indexes")); + } + + #[test] + fn invalid_signing_key_error_does_not_leak_key() { + let bad_key = "not-a-key-but-could-be-secret"; + let keys = vec![KEY_A.to_string(), bad_key.to_string()]; + let err = parse_signers(&keys, &[0, 3]).unwrap_err(); + let message = format!("{:#}", err); + assert!(message.contains("invalid signing key at position 1")); + assert!(!message.contains(bad_key)); + assert!(!message.contains(KEY_A)); + } + + #[test] + fn whitespace_around_values_is_trimmed() { + let keys = format!(" {} , {} ", KEY_A, KEY_B); + let config = cli(&["--signing-key", &keys, "--oracle-index", " 0 , 3 "]); + let indexes = parse_indexes(&config.oracle_index).unwrap(); + assert_eq!(indexes, vec![0, 3]); + let signers = parse_signers(&split_list(&config.signing_key), &indexes).unwrap(); + assert_eq!(signers[0], (KEY_A.parse().unwrap(), 0)); + assert_eq!(signers[1], (KEY_B.parse().unwrap(), 3)); + } + + #[test] + fn duplicate_oracle_indexes_fail() { + let keys = vec![KEY_A.to_string(), KEY_B.to_string()]; + let err = parse_signers(&keys, &[1, 1]).unwrap_err(); + assert!(err.to_string().contains("duplicate oracle index 1")); + } + + #[test] + fn duplicate_signing_keys_fail_without_leaking_key() { + let keys = vec![KEY_A.to_string(), KEY_A.to_string()]; + let err = parse_signers(&keys, &[0, 1]).unwrap_err(); + let message = format!("{:#}", err); + assert!(message.contains("duplicate signing key at position 1")); + assert!(!message.contains(KEY_A)); + } + + #[test] + fn rewards_manager_rejects_multiple_signers() { + let err = validate_mode(2, Some(Address::zero()), None).unwrap_err(); + assert!(err + .to_string() + .contains("RewardsManager mode supports a single signing key, got 2")); + } + + #[test] + fn missing_contract_fails() { + let err = validate_mode(1, None, None).unwrap_err(); + assert_eq!( + err.to_string(), + "Configuration error: either `REWARDS_MANAGER_CONTRACT` or `SUBGRAPH_AVAILABILITY_MANAGER_CONTRACT` must be provided" + ); + } + + #[test] + fn availability_manager_accepts_multiple_signers() { + validate_mode(2, None, Some(Address::zero())).unwrap(); + } +} diff --git a/crates/common/src/prelude.rs b/crates/common/src/prelude.rs index 612e7af..8df9f00 100644 --- a/crates/common/src/prelude.rs +++ b/crates/common/src/prelude.rs @@ -25,7 +25,7 @@ pub use futures::{ }; pub use never::Never; pub use prometheus::Counter; -pub use slog::{error, info, trace, warn, Logger}; +pub use slog::{error, info, o, trace, warn, Logger}; pub use std::convert::{TryFrom, TryInto}; pub type Bytes32 = [u8; 32]; pub use lazy_static::lazy_static; From 15cb7a6548e9254d0b0b2efee115ebf65e28eae6 Mon Sep 17 00:00:00 2001 From: Miguel de Elias Date: Mon, 5 Oct 2026 10:26:04 -0300 Subject: [PATCH 2/2] chore: bump version to 1.6.0 --- Cargo.lock | 2 +- availability-oracle/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e0e2b60..de1d2a9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -146,7 +146,7 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" [[package]] name = "availability-oracle" -version = "1.5.0" +version = "1.6.0" dependencies = [ "async-trait", "bytes", diff --git a/availability-oracle/Cargo.toml b/availability-oracle/Cargo.toml index ad76ae3..0882240 100644 --- a/availability-oracle/Cargo.toml +++ b/availability-oracle/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "availability-oracle" -version = "1.5.0" +version = "1.6.0" authors = ["Leonardo Yvens "] edition = "2018"