diff --git a/examples/statusphere.rs b/examples/statusphere.rs index 31a9993..445b7bb 100644 --- a/examples/statusphere.rs +++ b/examples/statusphere.rs @@ -125,8 +125,11 @@ async fn handle_stream(index: Arc, repos: ReposControl, mut stream: let did = rec.did.as_str().to_owned(); match rec.action.as_str() { "create" | "update" => { - let Some(record_raw) = rec.record else { continue }; - let Ok(record) = serde_json::from_str::(record_raw.get()) else { + let Some(record_raw) = rec.record else { + continue; + }; + let Ok(record) = serde_json::from_str::(record_raw.get()) + else { continue; }; let Some(emoji) = record diff --git a/src/api/crawler.rs b/src/api/crawler.rs index 88e6e2d..45248d2 100644 --- a/src/api/crawler.rs +++ b/src/api/crawler.rs @@ -33,7 +33,13 @@ pub struct AddSourceRequest { fn is_private_ip(ip: IpAddr) -> bool { match ip { - IpAddr::V4(ip) => ip.is_loopback() || ip.is_private() || ip.is_link_local() || ip.is_multicast() || ip.is_unspecified(), + IpAddr::V4(ip) => { + ip.is_loopback() + || ip.is_private() + || ip.is_link_local() + || ip.is_multicast() + || ip.is_unspecified() + } IpAddr::V6(ip) => { ip.is_loopback() || ip.is_multicast() || ip.is_unspecified() || { let octets = ip.octets(); @@ -47,7 +53,9 @@ async fn validate_source_url(url: &Url) -> Result<(), String> { use std::net::IpAddr; let scheme = url.scheme(); if !matches!(scheme, "ws" | "wss" | "http" | "https") { - return Err(format!("invalid scheme `{scheme}`: only ws, wss, http, https are allowed")); + return Err(format!( + "invalid scheme `{scheme}`: only ws, wss, http, https are allowed" + )); } let Some(host) = url.host_str() else { return Err("missing host in URL".to_string()); @@ -82,17 +90,23 @@ async fn validate_source_url(url: &Url) -> Result<(), String> { async fn validate_reachability(url: &Url) -> Result<(), String> { let mut base = url.clone(); match base.scheme() { - "wss" => { let _ = base.set_scheme("https"); } - "ws" => { let _ = base.set_scheme("http"); } + "wss" => { + let _ = base.set_scheme("https"); + } + "ws" => { + let _ = base.set_scheme("http"); + } _ => {} } - let check_url = base.join("/xrpc/com.atproto.sync.listRepos?limit=1").map_err(|e| e.to_string())?; - + let check_url = base + .join("/xrpc/com.atproto.sync.listRepos?limit=1") + .map_err(|e| e.to_string())?; + let client = reqwest::Client::builder() .timeout(Duration::from_secs(5)) .build() .map_err(|e| e.to_string())?; - + match client.get(check_url).send().await { Ok(_) => Ok(()), Err(e) => Err(format!("host is unreachable: {e}")), @@ -105,7 +119,10 @@ pub async fn add_source( ) -> Result { let sources = hydrant.crawler.list_sources().await; if sources.len() >= 100 { - return Err((StatusCode::BAD_REQUEST, "Too many crawler sources configured (limit 100)".to_string())); + return Err(( + StatusCode::BAD_REQUEST, + "Too many crawler sources configured (limit 100)".to_string(), + )); } if let Err(e) = validate_source_url(&body.url).await { diff --git a/src/api/jetstream.rs b/src/api/jetstream.rs index 5af3afb..a8fb258 100644 --- a/src/api/jetstream.rs +++ b/src/api/jetstream.rs @@ -269,9 +269,9 @@ fn frame_for_event( } if compress { - let compressed = ZSTD_COMPRESSOR.with(|c| { - c.borrow_mut().compress(json) - }).map_err(|e| e.to_string())?; + let compressed = ZSTD_COMPRESSOR + .with(|c| c.borrow_mut().compress(json)) + .map_err(|e| e.to_string())?; if exceeds_max(compressed.len(), options) { return Ok(None); @@ -347,10 +347,9 @@ mod tests { fn test_frame_for_event_size_limits() { let json = b"{\"hello\": \"world\"}"; // options with max size 5 - let options = JetstreamFilter::new( - JetstreamSubscriberOptions::parse(&[], &[], 5, &[]).unwrap() - ); - + let options = + JetstreamFilter::new(JetstreamSubscriberOptions::parse(&[], &[], 5, &[]).unwrap()); + // Uncompressed exceeds limit let res = frame_for_event(json, false, &options).unwrap(); assert!(res.is_none()); @@ -360,13 +359,16 @@ mod tests { assert!(res.is_none()); // Fits within limit - let options_large = JetstreamFilter::new( - JetstreamSubscriberOptions::parse(&[], &[], 100, &[]).unwrap() - ); - let res_uncompressed = frame_for_event(json, false, &options_large).unwrap().unwrap(); + let options_large = + JetstreamFilter::new(JetstreamSubscriberOptions::parse(&[], &[], 100, &[]).unwrap()); + let res_uncompressed = frame_for_event(json, false, &options_large) + .unwrap() + .unwrap(); assert!(res_uncompressed.is_text()); - let res_compressed = frame_for_event(json, true, &options_large).unwrap().unwrap(); + let res_compressed = frame_for_event(json, true, &options_large) + .unwrap() + .unwrap(); assert!(res_compressed.is_binary()); } } diff --git a/src/api/xrpc/describe_repo.rs b/src/api/xrpc/describe_repo.rs index dd165cd..b34d659 100644 --- a/src/api/xrpc/describe_repo.rs +++ b/src/api/xrpc/describe_repo.rs @@ -57,7 +57,10 @@ pub async fn handle( let nsid = DescribeRepoResponse::NSID; let doc = super::resolve_mini_doc::resolve_mini_doc(&hydrant, &req.identifier, nsid).await?; let repo = hydrant.repos.get(&doc.did); - let collections = repo.collections().map_err(|e| internal_error(nsid, e)).await?; + let collections = repo + .collections() + .map_err(|e| internal_error(nsid, e)) + .await?; Ok(Json(DescribeRepoOutput { did: doc.did, diff --git a/src/config.rs b/src/config.rs index 16238b7..b880c2e 100644 --- a/src/config.rs +++ b/src/config.rs @@ -428,11 +428,7 @@ impl fmt::Display for Config { "db records memtable", format_args!("{} mb", self.db_records_memtable_size_mb) )?; - config_line!( - f, - "db records bloom filters", - self.db_records_bloom_filters - )?; + config_line!(f, "db records bloom filters", self.db_records_bloom_filters)?; let replay_chunk = if self.stream_replay_chunk_size == 0 { "auto".to_owned() } else { diff --git a/src/control/filter.rs b/src/control/filter.rs index d1aa619..32cd6c0 100644 --- a/src/control/filter.rs +++ b/src/control/filter.rs @@ -217,7 +217,10 @@ impl FilterPatch { mut self, collections: impl IntoIterator>, ) -> Self { - merge_set_update_append(&mut self.collections, collections.into_iter().map(Into::into)); + merge_set_update_append( + &mut self.collections, + collections.into_iter().map(Into::into), + ); self } diff --git a/src/control/hosts.rs b/src/control/hosts.rs index 5df5f87..39c878b 100644 --- a/src/control/hosts.rs +++ b/src/control/hosts.rs @@ -314,9 +314,21 @@ mod host_listing_tests { { let state = hydrant.state.clone(); tokio::task::spawn_blocking(move || -> Result<()> { - state.db.cursors.insert(keys::firehose_cursor_key("host2"), 2_i64.to_be_bytes()).into_diagnostic()?; - state.db.cursors.insert(keys::firehose_cursor_key("host4"), 4_i64.to_be_bytes()).into_diagnostic()?; - state.db.cursors.insert(keys::firehose_cursor_key("host6"), 6_i64.to_be_bytes()).into_diagnostic()?; + state + .db + .cursors + .insert(keys::firehose_cursor_key("host2"), 2_i64.to_be_bytes()) + .into_diagnostic()?; + state + .db + .cursors + .insert(keys::firehose_cursor_key("host4"), 4_i64.to_be_bytes()) + .into_diagnostic()?; + state + .db + .cursors + .insert(keys::firehose_cursor_key("host6"), 6_i64.to_be_bytes()) + .into_diagnostic()?; state.db.persist() }) .await diff --git a/src/control/seed.rs b/src/control/seed.rs index 9c8d82c..ba7b949 100644 --- a/src/control/seed.rs +++ b/src/control/seed.rs @@ -451,13 +451,18 @@ mod tests { } } -async fn read_limited_response(resp: reqwest::Response, limit: usize) -> miette::Result { +async fn read_limited_response( + resp: reqwest::Response, + limit: usize, +) -> miette::Result { use bytes::BytesMut; use futures::StreamExt as _; let mut stream = resp.bytes_stream(); let mut buf = BytesMut::new(); while let Some(chunk) = stream.next().await { - let chunk = chunk.into_diagnostic().context("failed to read response chunk")?; + let chunk = chunk + .into_diagnostic() + .context("failed to read response chunk")?; if buf.len() + chunk.len() > limit { miette::bail!("response body too large (exceeds {limit} bytes)"); } @@ -468,7 +473,13 @@ async fn read_limited_response(resp: reqwest::Response, limit: usize) -> miette: fn is_private_ip(ip: std::net::IpAddr) -> bool { match ip { - std::net::IpAddr::V4(ip) => ip.is_loopback() || ip.is_private() || ip.is_link_local() || ip.is_multicast() || ip.is_unspecified(), + std::net::IpAddr::V4(ip) => { + ip.is_loopback() + || ip.is_private() + || ip.is_link_local() + || ip.is_multicast() + || ip.is_unspecified() + } std::net::IpAddr::V6(ip) => { ip.is_loopback() || ip.is_multicast() || ip.is_unspecified() || { let octets = ip.octets(); @@ -490,4 +501,3 @@ fn is_safe_seed_host(host: &str) -> bool { } true } - diff --git a/src/crawler/by_collection.rs b/src/crawler/by_collection.rs index 2f86420..f5a16a5 100644 --- a/src/crawler/by_collection.rs +++ b/src/crawler/by_collection.rs @@ -114,7 +114,9 @@ impl ByCollectionProducer { } Err(RetryOutcome::Failed(e)) => { error!(err = %e, %collection, "by-collection fetch failed"); - return Err(e).into_diagnostic().wrap_err("by-collection fetch failed"); + return Err(e) + .into_diagnostic() + .wrap_err("by-collection fetch failed"); } }; @@ -122,7 +124,9 @@ impl ByCollectionProducer { Ok(b) => b, Err(e) => { error!(err = %e, "can't read listReposByCollection response"); - return Err(e).into_diagnostic().wrap_err("can't read listReposByCollection response"); + return Err(e) + .into_diagnostic() + .wrap_err("can't read listReposByCollection response"); } }; diff --git a/src/crawler/list_repos/producer.rs b/src/crawler/list_repos/producer.rs index 72e9b6e..18fc316 100644 --- a/src/crawler/list_repos/producer.rs +++ b/src/crawler/list_repos/producer.rs @@ -84,7 +84,9 @@ impl ListReposProducer { } Err(RetryOutcome::Failed(e)) => { error!(err = %e, "crawler failed to fetch listRepos"); - return Err(e).into_diagnostic().wrap_err("crawler failed to fetch listRepos"); + return Err(e) + .into_diagnostic() + .wrap_err("crawler failed to fetch listRepos"); } }; @@ -92,7 +94,9 @@ impl ListReposProducer { Ok(b) => b, Err(e) => { error!(err = %e, "cant read listRepos response"); - return Err(e).into_diagnostic().wrap_err("cant read listRepos response"); + return Err(e) + .into_diagnostic() + .wrap_err("cant read listRepos response"); } }; diff --git a/src/db/migration/v9.rs b/src/db/migration/v9.rs index 76a575b..2d4461c 100644 --- a/src/db/migration/v9.rs +++ b/src/db/migration/v9.rs @@ -4,11 +4,7 @@ use miette::Result; use crate::db::Db; #[cfg(feature = "indexer")] -use { - crate::db::types::TrimmedDid, - jacquard_common::types::did::Did, - miette::IntoDiagnostic, -}; +use {crate::db::types::TrimmedDid, jacquard_common::types::did::Did, miette::IntoDiagnostic}; #[cfg(feature = "indexer")] pub(crate) fn migrate_v9(db: &Db, batch: &mut OwnedWriteBatch) -> Result<()> { @@ -67,8 +63,8 @@ pub(crate) fn migrate_v9(_db: &Db, _batch: &mut OwnedWriteBatch) -> Result<()> { #[cfg(all(test, feature = "indexer"))] mod tests { use super::*; - use tempfile::tempdir; use crate::config::Config; + use tempfile::tempdir; fn test_config(path: &std::path::Path) -> Config { Config { @@ -96,12 +92,20 @@ mod tests { // Legacy PDS status & tier let legacy_status_key = "example.com|status"; let status = crate::pds_meta::HostStatus::Offline; - batch.insert(&db.filter, legacy_status_key.as_bytes(), rmp_serde::to_vec(&status).into_diagnostic()?); + batch.insert( + &db.filter, + legacy_status_key.as_bytes(), + rmp_serde::to_vec(&status).into_diagnostic()?, + ); let legacy_tier_key = "example.com|tier"; batch.insert(&db.filter, legacy_tier_key.as_bytes(), b"tier1"); - batch.insert(&db.counts, crate::db::keys::VERSIONING_KEY, 8_u64.to_be_bytes()); + batch.insert( + &db.counts, + crate::db::keys::VERSIONING_KEY, + 8_u64.to_be_bytes(), + ); batch.commit().into_diagnostic()?; db.persist()?; } @@ -122,14 +126,26 @@ mod tests { // Verify excludes migrated let legacy_exclude_key = format!("x|{}", did); - assert!(!db.filter.contains_key(legacy_exclude_key.as_bytes()).into_diagnostic()?); + assert!( + !db.filter + .contains_key(legacy_exclude_key.as_bytes()) + .into_diagnostic()? + ); let new_exclude_key = crate::db::filter::exclude_key(did)?; assert!(db.filter.contains_key(&new_exclude_key).into_diagnostic()?); // Verify PDS status and tier migrated - assert!(!db.filter.contains_key(b"example.com|status").into_diagnostic()?); - assert!(!db.filter.contains_key(b"example.com|tier").into_diagnostic()?); + assert!( + !db.filter + .contains_key(b"example.com|status") + .into_diagnostic()? + ); + assert!( + !db.filter + .contains_key(b"example.com|tier") + .into_diagnostic()? + ); let new_status_key = crate::db::pds_meta::pds_status_key("example.com"); assert!(db.filter.contains_key(&new_status_key).into_diagnostic()?); diff --git a/src/ingest/indexer/shard.rs b/src/ingest/indexer/shard.rs index f2a3b9f..812da76 100644 --- a/src/ingest/indexer/shard.rs +++ b/src/ingest/indexer/shard.rs @@ -304,8 +304,12 @@ impl FirehoseWorker { match e { IngestError::Commit(_) | IngestError::NoSigningKey(_) => false, IngestError::Generic(report) => { - if report.downcast_ref::().is_some() - || report.downcast_ref::().is_some() + if report + .downcast_ref::() + .is_some() + || report + .downcast_ref::() + .is_some() { false } else { diff --git a/src/ingest/validation.rs b/src/ingest/validation.rs index 253ad41..90524c0 100644 --- a/src/ingest/validation.rs +++ b/src/ingest/validation.rs @@ -1,10 +1,10 @@ use jacquard_common::IntoStatic; use jacquard_common::types::crypto::PublicKey; -use jacquard_repo::{BlockStore, MemoryBlockStore}; use jacquard_repo::Mst; use jacquard_repo::car::reader::{ParsedCar, parse_car_bytes}; use jacquard_repo::commit::Commit as AtpCommit; use jacquard_repo::mst::VerifiedWriteOp; +use jacquard_repo::{BlockStore, MemoryBlockStore}; use miette::IntoDiagnostic; use smol_str::ToSmolStr; use std::sync::Arc; diff --git a/src/ops.rs b/src/ops.rs index 8a5bc15..d9f535e 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -461,13 +461,13 @@ pub fn parse_path(path: &str) -> Result<(&str, &str)> { let mut parts = path.splitn(2, '/'); let collection = parts.next().wrap_err("missing collection")?; let rkey = parts.next().wrap_err("missing rkey")?; - + if collection.is_empty() || jacquard_common::types::nsid::Nsid::new(collection).is_err() { miette::bail!("invalid collection NSID: {collection}"); } if rkey.is_empty() || jacquard_common::types::string::Rkey::new(rkey).is_err() { miette::bail!("invalid record key (rkey): {rkey}"); } - + Ok((collection, rkey)) } diff --git a/src/state.rs b/src/state.rs index 5530370..ca9dd40 100644 --- a/src/state.rs +++ b/src/state.rs @@ -7,7 +7,7 @@ use miette::Result; use smol_str::SmolStr; #[cfg(feature = "indexer")] use tokio::sync::Notify; -use tokio::sync::{watch, Semaphore}; +use tokio::sync::{Semaphore, watch}; use url::Url; #[cfg(feature = "firehose-diagnostics")] diff --git a/src/util/mod.rs b/src/util/mod.rs index 0408cf7..ffff333 100644 --- a/src/util/mod.rs +++ b/src/util/mod.rs @@ -291,7 +291,10 @@ pub fn hash(val: &T) -> u64 { hasher.finish() } -pub async fn read_limited_bytes(resp: reqwest::Response, limit: usize) -> miette::Result { +pub async fn read_limited_bytes( + resp: reqwest::Response, + limit: usize, +) -> miette::Result { use bytes::BytesMut; use futures::StreamExt as _; use miette::Context as _; @@ -299,7 +302,9 @@ pub async fn read_limited_bytes(resp: reqwest::Response, limit: usize) -> miette let mut stream = resp.bytes_stream(); let mut buf = BytesMut::new(); while let Some(chunk) = stream.next().await { - let chunk = chunk.into_diagnostic().context("failed to read response chunk")?; + let chunk = chunk + .into_diagnostic() + .context("failed to read response chunk")?; if buf.len() + chunk.len() > limit { miette::bail!("response body too large (exceeds {limit} bytes)"); }