From fa5479bc1e86cb2fd35bf32bd03c79634ffaf900 Mon Sep 17 00:00:00 2001 From: phil Date: Thu, 25 Sep 2025 00:47:05 +0000 Subject: [PATCH] unwrap => expect --- src/client.rs | 5 ++++- src/mirror.rs | 4 ++-- src/plc_pg.rs | 14 +++++++------- src/poll.rs | 2 +- src/ratelimit.rs | 22 +++++++++++++++------- src/bin/allegedly.rs | 88 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------------- 6 file(s) changed, 90 insertion(s)(+), 45 deletion(s)(-) diff --git a/src/client.rs b/src/client.rs --- a/src/client.rs +++ b/src/client.rs @@ -10,7 +10,10 @@ ); pub static CLIENT: LazyLock = LazyLock::new(|| { - let inner = Client::builder().user_agent(UA).build().unwrap(); + let inner = Client::builder() + .user_agent(UA) + .build() + .expect("reqwest client to build"); let policy = ExponentialBackoff::builder().build_with_max_retries(12); diff --git a/src/mirror.rs b/src/mirror.rs --- a/src/mirror.rs +++ b/src/mirror.rs @@ -192,7 +192,7 @@ .user_agent(UA) .timeout(Duration::from_secs(10)) // fallback .build() - .unwrap(); + .expect("reqwest client to build"); let state = State { client, @@ -208,7 +208,7 @@ .with(Cors::new().allow_credentials(false)) .with(Compression::new()) .with(GovernorMiddleware::new(Quota::per_minute( - 3000.try_into().unwrap(), + 3000.try_into().expect("ratelimit middleware to build"), ))) .with(CatchPanic::new()) .with(Tracing); diff --git a/src/plc_pg.rs b/src/plc_pg.rs --- a/src/plc_pg.rs +++ b/src/plc_pg.rs @@ -13,12 +13,12 @@ }; fn get_tls(cert: PathBuf) -> MakeTlsConnector { - let cert = std::fs::read(cert).unwrap(); - let cert = Certificate::from_pem(&cert).unwrap(); + let cert = std::fs::read(cert).expect("to read cert file"); + let cert = Certificate::from_pem(&cert).expect("to build cert"); let connector = TlsConnector::builder() .add_root_certificate(cert) .build() - .unwrap(); + .expect("to build tls connector"); MakeTlsConnector::new(connector) } @@ -46,7 +46,7 @@ connection .await .inspect_err(|e| log::error!("connection ended with error: {e}")) - .unwrap(); + .expect("pg validation connection not to blow up"); }); (client, task) } else { @@ -55,7 +55,7 @@ connection .await .inspect_err(|e| log::error!("connection ended with error: {e}")) - .unwrap(); + .expect("pg validation connection not to blow up"); }); (client, task) }; @@ -97,7 +97,7 @@ connection .await .inspect_err(|e| log::error!("connection ended with error: {e}")) - .unwrap(); + .expect("pg connection not to blow up"); }); client } else { @@ -109,7 +109,7 @@ connection .await .inspect_err(|e| log::error!("connection ended with error: {e}")) - .unwrap(); + .expect("pg connection not to blow up"); }); client }; diff --git a/src/poll.rs b/src/poll.rs --- a/src/poll.rs +++ b/src/poll.rs @@ -276,7 +276,7 @@ let page = ExportPage { ops: vec![valid_op().to_string()], }; - PageBoundaryState::new(&page).unwrap() + PageBoundaryState::new(&page).expect("to have a base page boundary state") } #[test] diff --git a/src/ratelimit.rs b/src/ratelimit.rs --- a/src/ratelimit.rs +++ b/src/ratelimit.rs @@ -24,8 +24,8 @@ let period = quota.replenish_interval() / factor; let burst = quota .burst_size() - .checked_mul(factor.try_into().unwrap()) - .unwrap(); + .checked_mul(factor.try_into().expect("factor to be non-zero")) + .expect("burst to be able to multiply"); Quota::with_period(period).map(|q| q.allow_burst(burst)) } @@ -40,8 +40,8 @@ pub fn new(quota: Quota) -> Self { Self { per_ip: RateLimiter::keyed(quota), - ip6_56: RateLimiter::keyed(scale_quota(quota, 8).unwrap()), - ip6_48: RateLimiter::keyed(scale_quota(quota, 256).unwrap()), + ip6_56: RateLimiter::keyed(scale_quota(quota, 8).expect("to scale quota")), + ip6_48: RateLimiter::keyed(scale_quota(quota, 256).expect("to scale quota")), } } pub fn check_key(&self, ip: IpAddr) -> Result<(), Duration> { @@ -56,11 +56,19 @@ .map_err(asdf); let check_56 = self .ip6_56 - .check_key(a.octets()[..7].try_into().unwrap()) + .check_key( + a.octets()[..7] + .try_into() + .expect("to check ip6 /56 limiter"), + ) .map_err(asdf); let check_48 = self .ip6_48 - .check_key(a.octets()[..6].try_into().unwrap()) + .check_key( + a.octets()[..6] + .try_into() + .expect("to check ip6 /48 limiter"), + ) .map_err(asdf); check_ip.and(check_56).and(check_48) } @@ -135,7 +143,7 @@ let remote = req .remote_addr() .as_socket_addr() - .unwrap_or_else(|| panic!("failed to get request's remote addr")) // TODO + .expect("failed to get request's remote addr") // TODO .ip(); log::trace!("remote: {remote}"); diff --git a/src/bin/allegedly.rs b/src/bin/allegedly.rs --- a/src/bin/allegedly.rs +++ b/src/bin/allegedly.rs @@ -36,9 +36,10 @@ /// Bulk load into did-method-plc-compatible postgres instead of stdout /// /// Pass a postgres connection url like "postgresql://localhost:5432" - #[arg(long)] + #[arg(long, env = "ALLEGEDLY_TO_POSTGRES")] to_postgres: Option, /// Cert for postgres (if needed) + #[arg(long)] postgres_cert: Option, /// Delete all operations from the postgres db before starting /// @@ -82,6 +83,7 @@ #[arg(long, env = "ALLEGEDLY_WRAP_PG")] wrap_pg: Url, /// path to tls cert for the wrapped postgres db, if needed + #[arg(long, env = "ALLEGEDLY_WRAP_PG_CERT")] wrap_pg_cert: Option, /// wrapping server listen address #[arg(short, long, env = "ALLEGEDLY_BIND")] @@ -150,7 +152,7 @@ while let Some(page) = rx.recv().await && page.ops.len() > 900 { - tx.send(page).await.unwrap(); + tx.send(page).await.expect("to be able to forward a page"); } }); fwd @@ -181,12 +183,12 @@ log::info!("Reading weekly bundles from local folder {dir:?}"); backfill(FolderSource(dir), tx, source_workers.unwrap_or(1), until) .await - .unwrap(); + .expect("to source bundles from a folder"); } else { log::info!("Fetching weekly bundles from from {http}"); backfill(HttpSource(http), tx, source_workers.unwrap_or(4), until) .await - .unwrap(); + .expect("to source bundles from http"); } }); @@ -203,12 +205,16 @@ let pg_cert = postgres_cert.clone(); let bulk_out_write = tokio::task::spawn(async move { if let Some(ref url) = to_postgres_url_bulk { - let db = Db::new(url.as_str(), pg_cert).await.unwrap(); + let db = Db::new(url.as_str(), pg_cert) + .await + .expect("to get db for bulk out write"); backfill_to_pg(db, postgres_reset, rx, notify_last_at) .await - .unwrap(); + .expect("to backfill to pg"); } else { - pages_to_stdout(rx, notify_last_at).await.unwrap(); + pages_to_stdout(rx, notify_last_at) + .await + .expect("to backfill to stdout"); } }); @@ -216,20 +222,28 @@ let mut upstream = args.upstream; upstream.set_path("/export"); // wait until the time for `after` is known - let last_at = rx_last.await.unwrap(); + let last_at = rx_last.await.expect("to get the last log's createdAt"); log::info!("beginning catch-up from {last_at:?} while the writer finalizes stuff"); let (tx, rx) = mpsc::channel(256); // these are small pages - tokio::task::spawn( - async move { poll_upstream(last_at, upstream, tx).await.unwrap() }, - ); - bulk_out_write.await.unwrap(); + tokio::task::spawn(async move { + poll_upstream(last_at, upstream, tx) + .await + .expect("polling upstream to work") + }); + bulk_out_write.await.expect("to wait for bulk_out_write"); log::info!("writing catch-up pages"); let full_pages = full_pages(rx); if let Some(url) = to_postgres { - let db = Db::new(url.as_str(), postgres_cert).await.unwrap(); - pages_to_pg(db, full_pages).await.unwrap(); + let db = Db::new(url.as_str(), postgres_cert) + .await + .expect("to connect pg for catchup"); + pages_to_pg(db, full_pages) + .await + .expect("to write catch-up pages to pg"); } else { - pages_to_stdout(full_pages, None).await.unwrap(); + pages_to_stdout(full_pages, None) + .await + .expect("to write catch-up pages to stdout"); } } } @@ -241,10 +255,16 @@ let mut url = args.upstream; url.set_path("/export"); let (tx, rx) = mpsc::channel(32); // read ahead if gzip stalls for some reason - tokio::task::spawn(async move { poll_upstream(Some(after), url, tx).await.unwrap() }); + tokio::task::spawn(async move { + poll_upstream(Some(after), url, tx) + .await + .expect("to poll upstream") + }); log::trace!("ensuring output directory exists"); - std::fs::create_dir_all(&dest).unwrap(); - pages_to_weeks(rx, dest, clobber).await.unwrap(); + std::fs::create_dir_all(&dest).expect("to ensure output dir exists"); + pages_to_weeks(rx, dest, clobber) + .await + .expect("to write bundles to output files"); } Commands::Mirror { wrap, @@ -255,11 +275,13 @@ acme_cache_path, acme_directory_url, } => { - let db = Db::new(wrap_pg.as_str(), wrap_pg_cert).await.unwrap(); + let db = Db::new(wrap_pg.as_str(), wrap_pg_cert) + .await + .expect("to connect to pg for mirroring"); let latest = db .get_latest() .await - .unwrap() + .expect("to query for last createdAt") .expect("there to be at least one op in the db. did you backfill?"); let (tx, rx) = mpsc::channel(2); @@ -268,15 +290,19 @@ tokio::task::spawn(async move { log::info!("starting poll reader..."); url.set_path("/export"); - tokio::task::spawn( - async move { poll_upstream(Some(latest), url, tx).await.unwrap() }, - ); + tokio::task::spawn(async move { + poll_upstream(Some(latest), url, tx) + .await + .expect("to poll upstream for mirror sync") + }); }); // db writer let poll_db = db.clone(); tokio::task::spawn(async move { log::info!("starting db writer..."); - pages_to_pg(poll_db, rx).await.unwrap(); + pages_to_pg(poll_db, rx) + .await + .expect("to write to pg for mirror"); }); let listen_conf = match (bind, acme_domain.is_empty(), acme_cache_path) { @@ -289,15 +315,23 @@ (_, _, _) => unreachable!(), }; - serve(&args.upstream, wrap, listen_conf).await.unwrap(); + serve(&args.upstream, wrap, listen_conf) + .await + .expect("to be able to serve the mirror proxy app"); } Commands::Tail { after } => { let mut url = args.upstream; url.set_path("/export"); let start_at = after.or_else(|| Some(chrono::Utc::now())); let (tx, rx) = mpsc::channel(1); - tokio::task::spawn(async move { poll_upstream(start_at, url, tx).await.unwrap() }); - pages_to_stdout(rx, None).await.unwrap(); + tokio::task::spawn(async move { + poll_upstream(start_at, url, tx) + .await + .expect("to poll upstream") + }); + pages_to_stdout(rx, None) + .await + .expect("to write pages to stdout"); } } log::info!("whew, {:?}. goodbye!", t0.elapsed()); -- tangled.sh