build: bump version to 0.7.12

Cargo.lock

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

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nostr-rs-relay"
-version = "0.7.9"
+version = "0.7.12"
 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

@@ -116,7 +116,7 @@ pub async fn db_writer(
             &settings,
             OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE,
             1,
-            4,
+            2,
             false,
         );
         if settings.database.in_memory {
@@ -636,69 +636,105 @@ pub async fn db_query(
     query_tx: tokio::sync::mpsc::Sender<QueryResult>,
     mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
 ) {
+    let pre_spawn_start = Instant::now();
     task::spawn_blocking(move || {
-        let mut row_count: usize = 0;
+        let db_queue_time = pre_spawn_start.elapsed();
+        // report queuing time if it is slow
+        if db_queue_time > Duration::from_secs(1) {
+            debug!(
+                "(slow) DB query queued for {:?} (cid: {}, sub: {:?})",
+                db_queue_time, client_id, sub.id
+            );
+        }
         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);
+        // any client that doesn't cause us to generate new rows in 5
+        // seconds gets dropped.
+        let abort_cutoff = Duration::from_secs(5);
         let start = Instant::now();
+        let mut slow_first_event;
+        let mut last_successful_send = Instant::now();
         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("00") {
+                    debug!(
+                        "query req (slow): {:?} (cid: {}, sub: {:?})",
+                        sub, client_id, sub.id
+                    );
+                    debug!(
+                        "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(());
                 }
                 row_count += 1;
                 let event_json = row.get(0)?;
+                loop {
+                    if query_tx.capacity() != 0 {
+                        // we have capacity to add another item
+                        break;
+                    } else {
+                        // the queue is full
+                        trace!("db reader thread is stalled");
+                        if last_successful_send + abort_cutoff < Instant::now() {
+                            // the queue has been full for too long, abort
+                            info!("aborting database query due to slow client");
+                            let ok: Result<()> = Ok(());
+                            return ok;
+                        }
+                        // give the queue a chance to clear before trying again
+                        thread::sleep(Duration::from_millis(100));
+                    }
+                }
+                // TODO: we could use try_send, but we'd have to juggle
+                // getting the query result back as part of the error
+                // result.
                 query_tx
                     .blocking_send(QueryResult {
                         sub_id: sub.get_id(),
                         event: event_json,
                     })
                     .ok();
+                last_successful_send = Instant::now();
             }
             query_tx
                 .blocking_send(QueryResult {
@@ -707,10 +743,11 @@ pub async fn db_query(
                 })
                 .ok();
             debug!(
-                "query completed in {:?} (cid: {}, sub: {:?}, rows: {})",
-                start.elapsed(),
+                "query completed in {:?} (cid: {}, sub: {:?}, db_time: {:?}, rows: {})",
+                pre_spawn_start.elapsed(),
                 client_id,
                 sub.id,
+                start.elapsed(),
                 row_count
             );
         } else {

src/schema.rs

@@ -16,7 +16,7 @@ pub const STARTUP_SQL: &str = r##"
 PRAGMA main.synchronous=NORMAL;
 PRAGMA foreign_keys = ON;
 PRAGMA journal_size_limit=32768;
-pragma mmap_size = 1073741824; -- 1024MB of mmap
+pragma mmap_size = 17179869184; -- cap mmap at 16GB
 "##;
 
 /// Latest database version

src/server.rs

@@ -339,8 +339,7 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
         let pool = db::build_pool(
             "client query",
             &settings,
-            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY
-                | rusqlite::OpenFlags::SQLITE_OPEN_SHARED_CACHE,
+            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
             db_min_conn,
             db_max_conn,
             true,
@@ -464,9 +463,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();