From b80f14d56bff0f535668a9eb6be2169820ab3502 Mon Sep 17 00:00:00 2001 From: Mike Dilger Date: Sat, 16 Mar 2024 14:44:42 +1300 Subject: [PATCH] Use parking lot RwLock, move away from async RwLock, honor config changes to allow_scraping --- Cargo.lock | 1 + chorus-bin/Cargo.toml | 1 + chorus-bin/src/bin/dump.rs | 2 +- chorus-bin/src/globals.rs | 2 +- chorus-bin/src/main.rs | 19 +++------- chorus-bin/src/nostr.rs | 71 ++++++++++++++++--------------------- chorus-bin/src/web.rs | 2 +- chorus-lib/src/store/mod.rs | 16 ++++----- 8 files changed, 46 insertions(+), 68 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5fd4e10..9aaf960 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -153,6 +153,7 @@ dependencies = [ "hyper-tungstenite", "lazy_static", "log", + "parking_lot", "rustls-pemfile", "serde", "tempfile", diff --git a/chorus-bin/Cargo.toml b/chorus-bin/Cargo.toml index aaf9ce5..e3e0b6a 100644 --- a/chorus-bin/Cargo.toml +++ b/chorus-bin/Cargo.toml @@ -16,6 +16,7 @@ hyper = { version = "0.14", features = [ "http1", "server", "runtime", "stream" hyper-tungstenite = "0.11" lazy_static = "1.4" log = "0.4" +parking_lot = "0.12" rustls-pemfile = "1.0" serde = { version = "1.0", features = ["derive"] } textnonce = "1" diff --git a/chorus-bin/src/bin/dump.rs b/chorus-bin/src/bin/dump.rs index 0b4bfcb..32c7a76 100644 --- a/chorus-bin/src/bin/dump.rs +++ b/chorus-bin/src/bin/dump.rs @@ -43,7 +43,7 @@ fn main() -> Result<(), Error> { let (_incount, _outcount, filter) = Filter::from_json(b"{}", &mut buffer)?; let screen = |_: &Event<'_>| -> bool { true }; - let mut events = store.find_events(filter, screen)?; + let mut events = store.find_events(filter, screen, &config)?; for event in events.drain(..) { let bytes = event.as_json()?; let s = unsafe { std::str::from_utf8_unchecked(&bytes) }; diff --git a/chorus-bin/src/globals.rs b/chorus-bin/src/globals.rs index f3e3fb9..22b22b2 100644 --- a/chorus-bin/src/globals.rs +++ b/chorus-bin/src/globals.rs @@ -2,11 +2,11 @@ use chorus_lib::config::Config; use chorus_lib::store::Store; use hyper::server::conn::Http; use lazy_static::lazy_static; +use parking_lot::RwLock; use std::sync::atomic::AtomicUsize; use std::sync::OnceLock; use tokio::sync::broadcast::Sender as BroadcastSender; use tokio::sync::watch::Sender as WatchSender; -use tokio::sync::RwLock; pub struct Globals { pub config: RwLock, diff --git a/chorus-bin/src/main.rs b/chorus-bin/src/main.rs index 18c7551..1bd0e2b 100644 --- a/chorus-bin/src/main.rs +++ b/chorus-bin/src/main.rs @@ -80,7 +80,7 @@ async fn main() -> Result<(), Error> { log::info!(target: "Server", "Running on {}:{}", config.ip_address, config.port); // Store config into GLOBALS - *GLOBALS.config.write().await = config; + *GLOBALS.config.write() = config; let mut interrupt_signal = signal(SignalKind::interrupt())?; let mut quit_signal = signal(SignalKind::quit())?; @@ -114,7 +114,7 @@ async fn main() -> Result<(), Error> { let friendly_config: FriendlyConfig = toml::from_str(&contents)?; let config: Config = friendly_config.into_config()?; - *GLOBALS.config.write().await = config; + *GLOBALS.config.write() = config; }, // Accepts network connections and spawn a task to serve each one @@ -475,22 +475,13 @@ impl WebSocketService { .unwrap() .get_event_by_offset(new_event_offset)? { - let config = &*GLOBALS.config.read().await; - let event_flags = { - let user_keys_ref = &config.user_keys; - nostr::event_flags(&event, &self.user, user_keys_ref) - }; - let authorized_user = nostr::authorized_user(&self.user).await; + let event_flags = nostr::event_flags(&event, &self.user); + let authorized_user = nostr::authorized_user(&self.user); 'subs: for (subid, filters) in self.subscriptions.iter() { for filter in filters.iter() { if filter.as_filter()?.event_matches(&event)? - && nostr::screen_outgoing_event( - &event, - &event_flags, - authorized_user, - config, - ) + && nostr::screen_outgoing_event(&event, &event_flags, authorized_user) { let message = NostrReply::Event(subid, event); self.websocket diff --git a/chorus-bin/src/nostr.rs b/chorus-bin/src/nostr.rs index 9816f4b..97bc44a 100644 --- a/chorus-bin/src/nostr.rs +++ b/chorus-bin/src/nostr.rs @@ -1,6 +1,5 @@ use crate::globals::GLOBALS; use crate::WebSocketService; -use chorus_lib::config::Config; use chorus_lib::error::{ChorusError, Error}; use chorus_lib::reply::{NostrReply, NostrReplyPrefix}; use chorus_lib::types::parse::json_escape::json_unescape; @@ -83,7 +82,7 @@ impl WebSocketService { if let Err(e) = self.req_inner(&subid, filters).await { let reply = match e.inner { ChorusError::TooManySubscriptions => { - let max_subscriptions = GLOBALS.config.read().await.max_subscriptions; + let max_subscriptions = GLOBALS.config.read().max_subscriptions; NostrReply::Closed( &subid, NostrReplyPrefix::Blocked, @@ -106,13 +105,13 @@ impl WebSocketService { } async fn req_inner(&mut self, subid: &String, filters: Vec) -> Result<(), Error> { - let max_subscriptions = GLOBALS.config.read().await.max_subscriptions; + let max_subscriptions = GLOBALS.config.read().max_subscriptions; if self.subscriptions.len() >= max_subscriptions { return Err(ChorusError::TooManySubscriptions.into()); } let user = self.user; - let authorized_user = authorized_user(&user).await; + let authorized_user = authorized_user(&user); // NOTE on private events (DMs, GiftWraps) // Most relays check if you are seeking them, and of which pubkey, and if you are @@ -127,19 +126,19 @@ impl WebSocketService { { let mut events: Vec = Vec::new(); - let config = GLOBALS.config.read().await; - let user_keys_ref = &config.user_keys; - for filter in filters.iter() { let screen = |event: &Event| { - let event_flags = event_flags(event, &user, user_keys_ref); - screen_outgoing_event(event, &event_flags, authorized_user, &config) + let event_flags = event_flags(event, &user); + screen_outgoing_event(event, &event_flags, authorized_user) + }; + let filter_events = { + let config = &*GLOBALS.config.read(); + GLOBALS + .store + .get() + .unwrap() + .find_events(filter.as_filter()?, screen, config)? }; - let filter_events = GLOBALS - .store - .get() - .unwrap() - .find_events(filter.as_filter()?, screen)?; events.extend(filter_events); } @@ -231,17 +230,14 @@ impl WebSocketService { async fn event_inner(&mut self) -> Result<(), Error> { let user = self.user; - let authorized_user = authorized_user(&user).await; + let authorized_user = authorized_user(&user); // Delineate the event back out of the session buffer let event = Event::delineate(&self.buffer)?; - let event_flags = { - let user_keys_ref = &*GLOBALS.config.read().await.user_keys; - event_flags(&event, &user, user_keys_ref) - }; + let event_flags = event_flags(&event, &user); - if !event_flags.author_is_an_authorized_user || GLOBALS.config.read().await.verify_events { + if !event_flags.author_is_an_authorized_user || GLOBALS.config.read().verify_events { // Verify the event is valid (id is hash, signature is valid) if let Err(e) = event.verify() { return Err(ChorusError::EventIsInvalid(format!("{}", e)).into()); @@ -352,7 +348,7 @@ impl WebSocketService { }; if let Some(h) = url.host() { let theirhost = h.to_owned(); - if theirhost == GLOBALS.config.read().await.hostname { + if theirhost == GLOBALS.config.read().hostname { relay_ok = true; } } @@ -409,7 +405,7 @@ async fn screen_incoming_event( } // Accept if an open relay - if GLOBALS.config.read().await.open_relay { + if GLOBALS.config.read().open_relay { return Ok(true); } @@ -419,23 +415,17 @@ async fn screen_incoming_event( } // Accept relay lists from anybody - if event.kind() == Kind(10002) && GLOBALS.config.read().await.serve_relay_lists { + if event.kind() == Kind(10002) && GLOBALS.config.read().serve_relay_lists { return Ok(true); } // Allow if event kind ephemeral - if event.kind().is_ephemeral() && GLOBALS.config.read().await.serve_ephemeral { + if event.kind().is_ephemeral() && GLOBALS.config.read().serve_ephemeral { return Ok(true); } // If the author is one of our users, always accept it - if GLOBALS - .config - .read() - .await - .user_keys - .contains(&event.pubkey()) - { + if GLOBALS.config.read().user_keys.contains(&event.pubkey()) { return Ok(true); } @@ -443,7 +433,7 @@ async fn screen_incoming_event( for mut tag in event.tags()?.iter() { if tag.next() == Some(b"p") { if let Some(value) = tag.next() { - for ukhex in &GLOBALS.config.read().await.user_hex_keys { + for ukhex in &GLOBALS.config.read().user_hex_keys { if value == ukhex.as_bytes() { return Ok(true); } @@ -459,7 +449,6 @@ pub fn screen_outgoing_event( event: &Event<'_>, event_flags: &EventFlags, authorized_user: bool, - config: &Config, ) -> bool { // Forbid if it is a private event (DM or GiftWrap) and theey are neither the recipient // nor the author @@ -468,17 +457,17 @@ pub fn screen_outgoing_event( } // Allow if an open relay - if config.open_relay { + if GLOBALS.config.read().open_relay { return true; } // Allow Relay Lists - if event.kind() == Kind(10002) && config.serve_relay_lists { + if event.kind() == Kind(10002) && GLOBALS.config.read().serve_relay_lists { return true; } // Allow if event kind ephemeral - if event.kind().is_ephemeral() && config.serve_ephemeral { + if event.kind().is_ephemeral() && GLOBALS.config.read().serve_ephemeral { return true; } @@ -496,10 +485,10 @@ pub fn screen_outgoing_event( false } -pub async fn authorized_user(user: &Option) -> bool { +pub fn authorized_user(user: &Option) -> bool { match user { None => false, - Some(pk) => GLOBALS.config.read().await.user_keys.contains(pk), + Some(pk) => GLOBALS.config.read().user_keys.contains(pk), } } @@ -510,8 +499,8 @@ pub struct EventFlags { pub tags_current_user: bool, } -pub fn event_flags(event: &Event<'_>, user: &Option, user_keys: &[Pubkey]) -> EventFlags { - let author_is_an_authorized_user = user_keys.contains(&event.pubkey()); +pub fn event_flags(event: &Event<'_>, user: &Option) -> EventFlags { + let author_is_an_authorized_user = GLOBALS.config.read().user_keys.contains(&event.pubkey()); let author_is_current_user = match user { None => false, @@ -532,7 +521,7 @@ pub fn event_flags(event: &Event<'_>, user: &Option, user_keys: &[Pubkey } } - if user_keys.contains(&tagged_pk) { + if GLOBALS.config.read().user_keys.contains(&tagged_pk) { tags_an_authorized_user = true; } } diff --git a/chorus-bin/src/web.rs b/chorus-bin/src/web.rs index da50ff7..06d2ae1 100644 --- a/chorus-bin/src/web.rs +++ b/chorus-bin/src/web.rs @@ -18,7 +18,7 @@ pub async fn serve_http(peer: SocketAddr, request: Request) -> Result Result, Error> { log::debug!(target: "Client", "{}: sent NIP-11", peer); let rid = { - let config = &*GLOBALS.config.read().await; + let config = &*GLOBALS.config.read(); GLOBALS.rid.get_or_init(|| build_rid(config)) }; diff --git a/chorus-lib/src/store/mod.rs b/chorus-lib/src/store/mod.rs index 7348c8f..d01b455 100644 --- a/chorus-lib/src/store/mod.rs +++ b/chorus-lib/src/store/mod.rs @@ -31,9 +31,6 @@ pub struct Store { deleted_offsets: Database, Unit>, deleted_events: Database, Unit>, ip_data: Database, UnalignedSlice>, - allow_scraping: bool, - allow_scrape_if_limited_to: u32, - allow_scrape_if_max_seconds: u64, } impl Store { @@ -187,9 +184,6 @@ impl Store { deleted_offsets, deleted_events, ip_data, - allow_scraping: config.allow_scraping, - allow_scrape_if_limited_to: config.allow_scrape_if_limited_to, - allow_scrape_if_max_seconds: config.allow_scrape_if_max_seconds, }; // This is in migrations.rs @@ -303,7 +297,8 @@ impl Store { } /// Find all events that match the filter - pub fn find_events(&self, filter: Filter, screen: F) -> Result, Error> + pub fn find_events(&self, filter: Filter, screen: F, config: &Config) + -> Result, Error> where F: Fn(&Event) -> bool, { @@ -626,9 +621,10 @@ impl Store { } else { // SCRAPE: let maxtime = filter.until().0.min(Time::now().0); - let allow = self.allow_scraping - || filter.limit() <= self.allow_scrape_if_limited_to - || (maxtime - filter.since().0) < self.allow_scrape_if_max_seconds; + + let allow = config.allow_scraping + || filter.limit() <= config.allow_scrape_if_limited_to + || (maxtime - filter.since().0) < config.allow_scrape_if_max_seconds; if !allow { return Err(ChorusError::Scraper.into()); }