build: bump version to 0.7.11

Cargo.lock

@@ -1096,7 +1096,7 @@ checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"
 
 [[package]]
 name = "nostr-rs-relay"
-version = "0.7.9"
+version = "0.7.11"
 dependencies = [
  "anyhow",
  "bitcoin_hashes",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nostr-rs-relay"
-version = "0.7.9"
+version = "0.7.11"
 edition = "2021"
 authors = ["Greg Heartsfield <scsibug@imap.cc>"]
 description = "A relay implementation for the Nostr protocol"

config.toml

@@ -78,8 +78,8 @@ reject_future_seconds = 1800
 # defaults to unlimited (subject to subscription limits).
 #db_conns_per_client = 0
 
-# Limit blocking threads used for database connections.  Defaults to 64.
-#max_blocking_threads = 64
+# Limit blocking threads used for database connections.  Defaults to 16.
+#max_blocking_threads = 16
 
 # Limit the maximum size of an EVENT message.  Defaults to 128 KB.
 # Set to 0 for unlimited.

src/config.rs

@@ -219,7 +219,7 @@ impl Default for Settings {
                 messages_per_sec: None,
                 subscriptions_per_min: None,
                 db_conns_per_client: None,
-                max_blocking_threads: 64,
+                max_blocking_threads: 16,
                 max_event_bytes: Some(2 << 17),      // 128K
                 max_ws_message_bytes: Some(2 << 17), // 128K
                 max_ws_frame_bytes: Some(2 << 17),   // 128K

src/db.rs

@@ -636,58 +636,62 @@ pub async fn db_query(
     query_tx: tokio::sync::mpsc::Sender<QueryResult>,
     mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
 ) {
+    let start = Instant::now();
     task::spawn_blocking(move || {
-        let mut row_count: usize = 0;
+        debug!("moved DB query to thread in {:?}", start.elapsed());
         let start = Instant::now();
+        let mut row_count: usize = 0;
         // generate SQL query
         let (q, p) = query_from_sub(&sub);
         debug!("SQL generated in {:?}", start.elapsed());
         // show pool stats
         log_pool_stats(&pool);
         // cutoff for displaying slow queries
-        let slow_cutoff = Duration::from_millis(1000);
+        let slow_cutoff = Duration::from_millis(2000);
         let start = Instant::now();
+        let mut slow_first_event;
         if let Ok(conn) = pool.get() {
             // execute the query. Don't cache, since queries vary so much.
             let mut stmt = conn.prepare(&q)?;
             let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?;
             let mut first_result = true;
             while let Some(row) = event_rows.next()? {
+                let first_event_elapsed = start.elapsed();
+                slow_first_event = first_event_elapsed >= slow_cutoff;
                 if first_result {
-                    let first_result_elapsed = start.elapsed();
-                    // logging for slow queries; show sub and SQL
-                    if first_result_elapsed >= slow_cutoff {
-                        info!(
-                            "query req (slow): {:?} (cid: {}, sub: {:?})",
-                            sub, client_id, sub.id
-                        );
-                        info!(
-                            "query string (slow): {} (cid: {}, sub: {:?})",
-                            q, client_id, sub.id
-                        );
-                    } else {
-                        trace!(
-                            "query req: {:?} (cid: {}, sub: {:?})",
-                            sub,
-                            client_id,
-                            sub.id
-                        );
-                        trace!(
-                            "query string: {} (cid: {}, sub: {:?})",
-                            q,
-                            client_id,
-                            sub.id
-                        );
-                    }
                     debug!(
                         "first result in {:?} (cid: {}, sub: {:?})",
-                        first_result_elapsed, client_id, sub.id
+                        first_event_elapsed, client_id, sub.id
                     );
                     first_result = false;
                 }
-                // check if this is still active
-                // TODO:  check every N rows
-                if abandon_query_rx.try_recv().is_ok() {
+                // logging for slow queries; show sub and SQL.
+                // to reduce logging; only show 1/16th of clients (leading 0)
+                if slow_first_event && client_id.starts_with('0') {
+                    info!(
+                        "query req (slow): {:?} (cid: {}, sub: {:?})",
+                        sub, client_id, sub.id
+                    );
+                    info!(
+                        "query string (slow): {} (cid: {}, sub: {:?})",
+                        q, client_id, sub.id
+                    );
+                } else {
+                    trace!(
+                        "query req: {:?} (cid: {}, sub: {:?})",
+                        sub,
+                        client_id,
+                        sub.id
+                    );
+                    trace!(
+                        "query string: {} (cid: {}, sub: {:?})",
+                        q,
+                        client_id,
+                        sub.id
+                    );
+                }
+                // check if this is still active; every 100 rows
+                if row_count % 100 == 0 && abandon_query_rx.try_recv().is_ok() {
                     debug!("query aborted (cid: {}, sub: {:?})", client_id, sub.id);
                     return Ok(());
                 }

src/server.rs

@@ -464,9 +464,11 @@ async fn nostr_server(
     let cid = conn.get_client_prefix();
     // Create a channel for receiving query results from the database.
     // we will send out the tx handle to any query we generate.
-    let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(256);
+    // this has capacity for some of the larger requests we see, which
+    // should allow the DB thread to release the handle earlier.
+    let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(20000);
     // Create channel for receiving NOTICEs
-    let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(32);
+    let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(128);
 
     // last time this client sent data (message, ping, etc.)
     let mut last_message_time = Instant::now();