diff --git a/src/bin/mst_dump.rs b/src/bin/mst_dump.rs index bba0885..eafcaee 100644 --- a/src/bin/mst_dump.rs +++ b/src/bin/mst_dump.rs @@ -33,7 +33,7 @@ async fn main() -> Result<()> { // Init resolver let plc_url = Url::parse("https://plc.directory").into_diagnostic()?; - let resolver = Resolver::new(plc_url, 100); + let resolver = Resolver::new(vec![plc_url], 100); // Resolve identity info!("Resolving {}...", identifier_str); diff --git a/src/config.rs b/src/config.rs index e0a2dc0..23fb799 100644 --- a/src/config.rs +++ b/src/config.rs @@ -39,7 +39,7 @@ impl fmt::Display for SignatureVerification { pub struct Config { pub database_path: PathBuf, pub relay_host: Url, - pub plc_url: Url, + pub plc_urls: Vec, pub full_network: bool, pub cursor_save_interval: Duration, pub repo_fetch_timeout: Duration, @@ -84,7 +84,15 @@ impl Config { "RELAY_HOST", Url::parse("wss://relay.fire.hose.cam").unwrap() ); - let plc_url = cfg!("PLC_URL", Url::parse("https://plc.wtf").unwrap()); + let plc_urls: Vec = std::env::var("HYDRANT_PLC_URL") + .ok() + .map(|s| { + s.split(',') + .map(|s| Url::parse(s.trim())) + .collect::, _>>() + .map_err(|e| miette::miette!("invalid PLC URL: {}", e)) + }) + .unwrap_or_else(|| Ok(vec![Url::parse("https://plc.wtf").unwrap()]))?; let full_network = cfg!("FULL_NETWORK", false); let backfill_concurrency_limit = cfg!("BACKFILL_CONCURRENCY_LIMIT", 32usize); @@ -107,7 +115,7 @@ impl Config { Ok(Self { database_path, relay_host, - plc_url, + plc_urls, full_network, cursor_save_interval, repo_fetch_timeout, @@ -132,7 +140,7 @@ impl fmt::Display for Config { writeln!(f, "hydrant configuration:")?; writeln!(f, " log level: {}", self.log_level)?; writeln!(f, " relay host: {}", self.relay_host)?; - writeln!(f, " plc url: {}", self.plc_url)?; + writeln!(f, " plc urls: {:?}", self.plc_urls)?; writeln!(f, " full network indexing: {}", self.full_network)?; writeln!(f, " verify signatures: {}", self.verify_signatures)?; writeln!( diff --git a/src/resolver.rs b/src/resolver.rs index f034542..05e7f37 100644 --- a/src/resolver.rs +++ b/src/resolver.rs @@ -1,5 +1,6 @@ use std::ops::Not; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use jacquard::IntoStatic; @@ -46,7 +47,8 @@ impl From for ResolverError { } struct ResolverInner { - jacquard: JacquardResolver, + jacquards: Vec, + next_idx: AtomicUsize, key_cache: HashCache, PublicKey<'static>>, } @@ -56,18 +58,27 @@ pub struct Resolver { } impl Resolver { - pub fn new(plc_url: Url, identity_cache_size: u64) -> Self { + pub fn new(plc_urls: Vec, identity_cache_size: u64) -> Self { let http = reqwest::Client::new(); - let mut opts = ResolverOptions::default(); - opts.plc_source = PlcSource::PlcDirectory { base: plc_url }; - opts.request_timeout = Some(Duration::from_secs(3)); + let mut jacquards = Vec::with_capacity(plc_urls.len()); - // no jacquard cache - we manage our own - let jacquard = JacquardResolver::new(http, opts); + for url in plc_urls { + let mut opts = ResolverOptions::default(); + opts.plc_source = PlcSource::PlcDirectory { base: url }; + opts.request_timeout = Some(Duration::from_secs(3)); + + // no jacquard cache - we manage our own + jacquards.push(JacquardResolver::new(http.clone(), opts)); + } + + if jacquards.is_empty() { + panic!("at least one PLC URL must be provided"); + } Self { inner: Arc::new(ResolverInner { - jacquard, + jacquards, + next_idx: AtomicUsize::new(0), key_cache: HashCache::with_capacity( std::cmp::min(1000, (identity_cache_size / 100) as usize), identity_cache_size as usize, @@ -76,6 +87,11 @@ impl Resolver { } } + fn get_jacquard(&self) -> &JacquardResolver { + let idx = self.inner.next_idx.fetch_add(1, Ordering::Relaxed) % self.inner.jacquards.len(); + &self.inner.jacquards[idx] + } + pub async fn resolve_did( &self, identifier: &AtIdentifier<'_>, @@ -83,7 +99,7 @@ impl Resolver { match identifier { AtIdentifier::Did(did) => Ok(did.clone().into_static()), AtIdentifier::Handle(handle) => { - let did = self.inner.jacquard.resolve_handle(handle).await?; + let did = self.get_jacquard().resolve_handle(handle).await?; Ok(did.into_static()) } } @@ -93,7 +109,7 @@ impl Resolver { &self, did: &Did<'_>, ) -> Result<(Url, Option>), ResolverError> { - let doc_resp = self.inner.jacquard.resolve_did_doc(did).await?; + let doc_resp = self.get_jacquard().resolve_did_doc(did).await?; let doc = doc_resp.parse()?; let pds = doc @@ -115,7 +131,7 @@ impl Resolver { return Ok(entry.get().clone()); } - let doc_resp = self.inner.jacquard.resolve_did_doc(&did).await?; + let doc_resp = self.get_jacquard().resolve_did_doc(&did).await?; let doc = doc_resp.parse()?; let key = doc diff --git a/src/state.rs b/src/state.rs index 1614466..1a98705 100644 --- a/src/state.rs +++ b/src/state.rs @@ -19,7 +19,7 @@ impl AppState { config.cache_size, config.disable_lz4_compression, )?; - let resolver = Resolver::new(config.plc_url.clone(), config.identity_cache_size); + let resolver = Resolver::new(config.plc_urls.clone(), config.identity_cache_size); Ok(Self { db,