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.
+# Limit blocking threads used for database connections.  Defaults to 16.
-#max_blocking_threads = 64
+#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,27 +636,38 @@ pub async fn db_query(
     query_tx: tokio::sync::mpsc::Sender<QueryResult>,
     mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
 ) {
-    task::spawn_blocking(move || {
-        let mut row_count: usize = 0;
     let start = Instant::now();
+    task::spawn_blocking(move || {
+        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();
+                    debug!(
-                    // logging for slow queries; show sub and SQL
+                        "first result in {:?} (cid: {}, sub: {:?})",
-                    if first_result_elapsed >= slow_cutoff {
+                        first_event_elapsed, client_id, sub.id
+                    );
+                    first_result = false;
+                }
+                // 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
@@ -679,15 +690,8 @@ pub async fn db_query(
                         sub.id
                     );
                 }
-                    debug!(
+                // check if this is still active; every 100 rows
-                        "first result in {:?} (cid: {}, sub: {:?})",
+                if row_count % 100 == 0 && abandon_query_rx.try_recv().is_ok() {
-                        first_result_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() {
                     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();